mirror of https://github.com/LemmyNet/lemmy
implement ActivitySender actor (#89)
Merge pull request 'Adding unique ap_ids. Fixes #1100' (#90) from unique_ap_ids into activity-sender Reviewed-on: https://yerbamate.dev/LemmyNet/lemmy/pulls/90 Adding back in on_conflict. Trying to add back in the on_conflict_do_nothing. Trying to reduce delay time. Removing createFakes. Removing some unit tests. Adding comment jest timeout. Fixing tests again. Fixing tests again. Merge branch 'activity-sender' into unique_ap_ids_2 Replace actix client with reqwest to speed up federation tests Trying to fix tests again. Fixing unit tests. Fixing some broken unit tests, not done yet. Adding uniques. Adding unique ap_ids. Fixes #1100 use proper sql functionality for upsert added logging in fetcher, replace post/comment::create with upsert no need to do an actual update in post/comment::upsert Merge branch 'main' into activity-sender implement upsert for user/community reuse http client got it working attempt to use background-jobs crate rewrite with proper error handling and less boilerplate remove do_send, dont return errors from activity_sender WIP: implement ActivitySender actor Co-authored-by: dessalines <dessalines@noreply.yerbamate.dev> Co-authored-by: Dessalines <tyhou13@gmx.com> Co-authored-by: Felix Ableitner <me@nutomic.com> Reviewed-on: https://yerbamate.dev/LemmyNet/lemmy/pulls/89pull/1111/head
parent
18002bc837
commit
d4dccd17ae
@ -0,0 +1,27 @@
|
||||
-- Drop the uniques
|
||||
alter table private_message drop constraint idx_private_message_ap_id;
|
||||
alter table post drop constraint idx_post_ap_id;
|
||||
alter table comment drop constraint idx_comment_ap_id;
|
||||
alter table user_ drop constraint idx_user_actor_id;
|
||||
alter table community drop constraint idx_community_actor_id;
|
||||
|
||||
alter table private_message alter column ap_id set not null;
|
||||
alter table private_message alter column ap_id set default 'http://fake.com';
|
||||
|
||||
alter table post alter column ap_id set not null;
|
||||
alter table post alter column ap_id set default 'http://fake.com';
|
||||
|
||||
alter table comment alter column ap_id set not null;
|
||||
alter table comment alter column ap_id set default 'http://fake.com';
|
||||
|
||||
update private_message
|
||||
set ap_id = 'http://fake.com'
|
||||
where ap_id like 'changeme_%';
|
||||
|
||||
update post
|
||||
set ap_id = 'http://fake.com'
|
||||
where ap_id like 'changeme_%';
|
||||
|
||||
update comment
|
||||
set ap_id = 'http://fake.com'
|
||||
where ap_id like 'changeme_%';
|
@ -0,0 +1,56 @@
|
||||
-- Add unique ap_id for private_message, comment, and post
|
||||
|
||||
-- Need to delete the possible dupes for ones that don't start with the fake one
|
||||
delete from private_message a using (
|
||||
select min(id) as id, ap_id
|
||||
from private_message
|
||||
group by ap_id having count(*) > 1
|
||||
) b
|
||||
where a.ap_id = b.ap_id
|
||||
and a.id <> b.id;
|
||||
|
||||
delete from post a using (
|
||||
select min(id) as id, ap_id
|
||||
from post
|
||||
group by ap_id having count(*) > 1
|
||||
) b
|
||||
where a.ap_id = b.ap_id
|
||||
and a.id <> b.id;
|
||||
|
||||
delete from comment a using (
|
||||
select min(id) as id, ap_id
|
||||
from comment
|
||||
group by ap_id having count(*) > 1
|
||||
) b
|
||||
where a.ap_id = b.ap_id
|
||||
and a.id <> b.id;
|
||||
|
||||
-- Replacing the current default on the columns, to the unique one
|
||||
update private_message
|
||||
set ap_id = generate_unique_changeme()
|
||||
where ap_id = 'http://fake.com';
|
||||
|
||||
update post
|
||||
set ap_id = generate_unique_changeme()
|
||||
where ap_id = 'http://fake.com';
|
||||
|
||||
update comment
|
||||
set ap_id = generate_unique_changeme()
|
||||
where ap_id = 'http://fake.com';
|
||||
|
||||
-- Add the unique indexes
|
||||
alter table private_message alter column ap_id set not null;
|
||||
alter table private_message alter column ap_id set default generate_unique_changeme();
|
||||
|
||||
alter table post alter column ap_id set not null;
|
||||
alter table post alter column ap_id set default generate_unique_changeme();
|
||||
|
||||
alter table comment alter column ap_id set not null;
|
||||
alter table comment alter column ap_id set default generate_unique_changeme();
|
||||
|
||||
-- Add the uniques, for user_ and community too
|
||||
alter table private_message add constraint idx_private_message_ap_id unique (ap_id);
|
||||
alter table post add constraint idx_post_ap_id unique (ap_id);
|
||||
alter table comment add constraint idx_comment_ap_id unique (ap_id);
|
||||
alter table user_ add constraint idx_user_actor_id unique (actor_id);
|
||||
alter table community add constraint idx_community_actor_id unique (actor_id);
|
@ -0,0 +1,133 @@
|
||||
use crate::{
|
||||
apub::{check_is_apub_id_valid, extensions::signatures::sign, ActorType},
|
||||
LemmyError,
|
||||
};
|
||||
use activitystreams::{
|
||||
base::{Extends, ExtendsExt},
|
||||
object::AsObject,
|
||||
};
|
||||
use anyhow::{anyhow, Context, Error};
|
||||
use awc::Client;
|
||||
use background_jobs::{
|
||||
create_server,
|
||||
memory_storage::Storage,
|
||||
ActixJob,
|
||||
Backoff,
|
||||
MaxRetries,
|
||||
QueueHandle,
|
||||
WorkerConfig,
|
||||
};
|
||||
use lemmy_utils::{location_info, settings::Settings};
|
||||
use log::warn;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{future::Future, pin::Pin};
|
||||
use url::Url;
|
||||
|
||||
pub fn send_activity<T, Kind>(
|
||||
activity_sender: &QueueHandle,
|
||||
activity: T,
|
||||
actor: &dyn ActorType,
|
||||
to: Vec<Url>,
|
||||
) -> Result<(), LemmyError>
|
||||
where
|
||||
T: AsObject<Kind>,
|
||||
T: Extends<Kind>,
|
||||
Kind: Serialize,
|
||||
<T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
|
||||
{
|
||||
if !Settings::get().federation.enabled {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let activity = activity.into_any_base()?;
|
||||
let serialised_activity = serde_json::to_string(&activity)?;
|
||||
|
||||
for to_url in &to {
|
||||
check_is_apub_id_valid(&to_url)?;
|
||||
}
|
||||
|
||||
// TODO: it would make sense to create a separate task for each destination server
|
||||
let message = SendActivityTask {
|
||||
activity: serialised_activity,
|
||||
to,
|
||||
actor_id: actor.actor_id()?,
|
||||
private_key: actor.private_key().context(location_info!())?,
|
||||
};
|
||||
activity_sender.queue::<SendActivityTask>(message)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
struct SendActivityTask {
|
||||
activity: String,
|
||||
to: Vec<Url>,
|
||||
actor_id: Url,
|
||||
private_key: String,
|
||||
}
|
||||
|
||||
impl ActixJob for SendActivityTask {
|
||||
type State = MyState;
|
||||
type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
|
||||
const NAME: &'static str = "SendActivityTask";
|
||||
|
||||
const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
|
||||
const BACKOFF: Backoff = Backoff::Exponential(2);
|
||||
|
||||
fn run(self, state: Self::State) -> Self::Future {
|
||||
Box::pin(async move {
|
||||
for to_url in &self.to {
|
||||
let request = state
|
||||
.client
|
||||
.post(to_url.as_str())
|
||||
.header("Content-Type", "application/json");
|
||||
|
||||
// TODO: i believe we have to do the signing in here because it is only valid for a few seconds
|
||||
let signed = sign(
|
||||
request,
|
||||
self.activity.clone(),
|
||||
&self.actor_id,
|
||||
self.private_key.to_owned(),
|
||||
)
|
||||
.await;
|
||||
let signed = match signed {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
warn!("{}", e);
|
||||
// dont return an error because retrying would probably not fix the signing
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
if let Err(e) = signed.send().await {
|
||||
warn!("{}", e);
|
||||
return Err(anyhow!(
|
||||
"Failed to send activity {} to {}",
|
||||
&self.activity,
|
||||
to_url
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub fn create_activity_queue() -> QueueHandle {
|
||||
// Start the application server. This guards access to to the jobs store
|
||||
let queue_handle = create_server(Storage::new());
|
||||
|
||||
// Configure and start our workers
|
||||
WorkerConfig::new(|| MyState {
|
||||
client: Client::default(),
|
||||
})
|
||||
.register::<SendActivityTask>()
|
||||
.start(queue_handle.clone());
|
||||
|
||||
queue_handle
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct MyState {
|
||||
pub client: Client,
|
||||
}
|
Loading…
Reference in New Issue