2021-11-11 19:49:15 +00:00
|
|
|
use crate::{signatures::sign_and_send, traits::ActorType};
|
2020-08-31 13:48:02 +00:00
|
|
|
use anyhow::{anyhow, Context, Error};
|
|
|
|
use background_jobs::{
|
|
|
|
create_server,
|
|
|
|
memory_storage::Storage,
|
|
|
|
ActixJob,
|
|
|
|
Backoff,
|
|
|
|
MaxRetries,
|
|
|
|
QueueHandle,
|
|
|
|
WorkerConfig,
|
|
|
|
};
|
2021-09-22 15:57:09 +00:00
|
|
|
use lemmy_utils::{location_info, LemmyError};
|
2021-11-16 17:03:09 +00:00
|
|
|
use log::{info, warn};
|
2020-09-29 13:10:55 +00:00
|
|
|
use reqwest::Client;
|
2021-02-01 20:56:37 +00:00
|
|
|
use serde::{Deserialize, Serialize};
|
2021-11-11 19:49:15 +00:00
|
|
|
use std::{env, fmt::Debug, future::Future, pin::Pin};
|
2020-08-31 13:48:02 +00:00
|
|
|
use url::Url;
|
|
|
|
|
2021-10-06 20:20:05 +00:00
|
|
|
pub async fn send_activity(
|
2021-11-16 17:03:09 +00:00
|
|
|
activity_id: &Url,
|
2021-07-27 22:18:50 +00:00
|
|
|
actor: &dyn ActorType,
|
2021-10-06 20:20:05 +00:00
|
|
|
inboxes: Vec<&Url>,
|
2021-11-16 17:03:09 +00:00
|
|
|
activity: String,
|
2021-10-06 20:20:05 +00:00
|
|
|
client: &Client,
|
|
|
|
activity_queue: &QueueHandle,
|
2021-07-27 22:18:50 +00:00
|
|
|
) -> Result<(), LemmyError> {
|
2020-10-12 16:02:28 +00:00
|
|
|
for i in inboxes {
|
2020-10-06 17:19:53 +00:00
|
|
|
let message = SendActivityTask {
|
2021-11-16 17:03:09 +00:00
|
|
|
activity_id: activity_id.clone(),
|
2021-01-15 01:18:18 +00:00
|
|
|
inbox: i.to_owned(),
|
2021-01-27 16:42:23 +00:00
|
|
|
actor_id: actor.actor_id(),
|
2021-11-16 17:03:09 +00:00
|
|
|
activity: activity.clone(),
|
2020-10-06 17:19:53 +00:00
|
|
|
private_key: actor.private_key().context(location_info!())?,
|
|
|
|
};
|
2021-10-06 20:20:05 +00:00
|
|
|
if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
|
|
|
|
do_send(message, client).await?;
|
2020-12-14 16:44:27 +00:00
|
|
|
} else {
|
2021-10-06 20:20:05 +00:00
|
|
|
activity_queue.queue::<SendActivityTask>(message)?;
|
2020-12-14 16:44:27 +00:00
|
|
|
}
|
2020-10-06 17:19:53 +00:00
|
|
|
}
|
2020-08-31 13:48:02 +00:00
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Clone, Debug, Deserialize, Serialize)]
|
|
|
|
struct SendActivityTask {
|
2021-11-16 17:03:09 +00:00
|
|
|
activity_id: Url,
|
2020-10-12 16:02:28 +00:00
|
|
|
inbox: Url,
|
2020-08-31 13:48:02 +00:00
|
|
|
actor_id: Url,
|
2021-11-16 17:03:09 +00:00
|
|
|
activity: String,
|
2020-08-31 13:48:02 +00:00
|
|
|
private_key: String,
|
|
|
|
}
|
|
|
|
|
2020-10-19 14:29:35 +00:00
|
|
|
/// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
|
|
|
|
/// if the delivery failed.
|
2020-08-31 13:48:02 +00:00
|
|
|
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 {
|
2020-12-14 17:01:12 +00:00
|
|
|
Box::pin(async move { do_send(self, &state.client).await })
|
2020-12-14 16:44:27 +00:00
|
|
|
}
|
|
|
|
}
|
2020-09-29 13:10:55 +00:00
|
|
|
|
2020-12-14 17:01:12 +00:00
|
|
|
async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
|
2021-11-16 17:03:09 +00:00
|
|
|
info!("Sending {} to {}", task.activity_id, task.inbox);
|
2020-12-14 16:44:27 +00:00
|
|
|
let result = sign_and_send(
|
|
|
|
client,
|
|
|
|
&task.inbox,
|
|
|
|
task.activity.clone(),
|
|
|
|
&task.actor_id,
|
|
|
|
task.private_key.to_owned(),
|
|
|
|
)
|
2020-12-14 17:01:12 +00:00
|
|
|
.await;
|
2020-12-14 16:44:27 +00:00
|
|
|
|
2021-11-16 17:03:09 +00:00
|
|
|
match result {
|
|
|
|
Ok(o) => {
|
|
|
|
if !o.status().is_success() {
|
|
|
|
warn!(
|
|
|
|
"Send {} to {} failed with status {}: {}",
|
|
|
|
task.activity_id,
|
|
|
|
task.inbox,
|
|
|
|
o.status(),
|
|
|
|
o.text().await?
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Err(e) => {
|
|
|
|
return Err(anyhow!(
|
|
|
|
"Failed to send activity {} to {}: {}",
|
|
|
|
&task.activity_id,
|
|
|
|
task.inbox,
|
|
|
|
e
|
|
|
|
));
|
|
|
|
}
|
2020-08-31 13:48:02 +00:00
|
|
|
}
|
2020-12-14 16:44:27 +00:00
|
|
|
Ok(())
|
2020-08-31 13:48:02 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
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());
|
2021-10-08 14:36:53 +00:00
|
|
|
let arbiter = actix_web::rt::Arbiter::new();
|
2020-08-31 13:48:02 +00:00
|
|
|
|
|
|
|
// Configure and start our workers
|
|
|
|
WorkerConfig::new(|| MyState {
|
|
|
|
client: Client::default(),
|
|
|
|
})
|
|
|
|
.register::<SendActivityTask>()
|
2021-10-08 14:36:53 +00:00
|
|
|
.start_in_arbiter(&arbiter, queue_handle.clone());
|
2020-08-31 13:48:02 +00:00
|
|
|
|
|
|
|
queue_handle
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Clone)]
|
|
|
|
struct MyState {
|
|
|
|
pub client: Client,
|
|
|
|
}
|