- payload: Payload,
- context: web::Data<LemmyContext>,
-) -> Result<HttpResponse, LemmyError> {
- let unparsed = payload_to_string(payload).await?;
- info!("Received shared inbox activity {}", unparsed);
- let activity_data: ActivityCommonFields = serde_json::from_str(&unparsed)?;
- let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
- match activity.inner() {
- SharedInboxActivities::GroupInboxActivities(g) => {
- receive_group_inbox(*g, activity_data, request, &context).await
- }
- SharedInboxActivities::PersonInboxActivities(p) => {
- receive_person_inbox(*p, activity_data, request, &context).await
- }
- }
-}
-
-async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
- let mut bytes = BytesMut::new();
- while let Some(item) = payload.next().await {
- bytes.extend_from_slice(&item?);
- }
- let mut unparsed = String::new();
- Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
- Ok(unparsed)
-}
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub(crate) struct ActivityCommonFields {
- pub(crate) id: Url,
- pub(crate) actor: Url,
-}
-
-// TODO: move most of this code to library
-#[tracing::instrument(skip_all)]
-async fn receive_activity<'a, T>(
- request: HttpRequest,
- activity: T,
- activity_data: ActivityCommonFields,
- context: &LemmyContext,
-) -> Result<HttpResponse, LemmyError>
-where
- T: ActivityHandler<DataType = LemmyContext>
- + Clone
- + Deserialize<'a>
- + Serialize
- + std::fmt::Debug
- + Send
- + 'static,
-{
- check_is_apub_id_valid(&activity_data.actor, false, &context.settings())?;
- let request_counter = &mut 0;
- let actor = ObjectId::<UserOrCommunity>::new(activity_data.actor)
- .dereference(context, context.client(), request_counter)
- .await?;
- verify_signature(&request, &actor.public_key())?;
-
- info!("Verifying activity {}", activity_data.id.to_string());
- activity
- .verify(&Data::new(context.clone()), request_counter)
- .await?;
- assert_activity_not_local(&activity_data.id, &context.settings().hostname)?;
-
- // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
- // if we receive the same activity twice in very quick succession.
- let object_value = serde_json::to_value(&activity)?;
- let insert =
- insert_activity(&activity_data.id, object_value, false, true, context.pool()).await?;
- if !insert {
- debug!(
- "Received duplicate activity {}",
- activity_data.id.to_string()
- );
- return Ok(HttpResponse::BadRequest().finish());
- }
-
- info!("Receiving activity {}", activity_data.id.to_string());
- activity
- .receive(&Data::new(context.clone()), request_counter)
- .await?;
- Ok(HttpResponse::Ok().finish())