alemi/upub
alemi/upub
1
0
Fork 1
Code Issues 4 Pull requests Projects Releases Packages 1 Wiki Activity Actions

fix: if we get Creates for objects already fetched

Browse source
This commit is contained in:
əlemi 2024-06-06 16:32:52 +02:00
parent 677cab1871
commit b17060df3d
Signed by: alemi
GPG key ID: A4895B84D311642C
1 changed files with 17 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
upub/core/src/traits/process.rs
View file

@ -60,16 +60,24 @@ pub async fn create(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
tracing::error!("refusing to process activity without embedded object"); tracing::error!("refusing to process activity without embedded object");
return Err(ProcessorError::Unprocessable); return Err(ProcessorError::Unprocessable);
}; };
if let Ok(reply) = object_node.in_reply_to().id() { let oid = object_node.id()?.to_string();
if let Err(e) = ctx.fetch_object(reply, tx).await { let addressed = object_node.addressed();
tracing::warn!("failed fetching replies for received object: {e}");
}
}
let activity_model = ctx.insert_activity(activity, tx).await?; let activity_model = ctx.insert_activity(activity, tx).await?;
let object_model = ctx.insert_object(object_node, tx).await?; let internal_oid = if let Some(internal) = model::object::Entity::ap_to_internal(&oid, tx).await? {
let expanded_addressing = ctx.expand_addressing(object_model.addressed()).await?; tracing::debug!("skipping insertion of already known object #{internal}");
ctx.address_to(Some(activity_model.internal), Some(object_model.internal), &expanded_addressing, tx).await?; internal
tracing::info!("{} posted {}", activity_model.actor, object_model.id); } else {
if let Ok(reply) = object_node.in_reply_to().id() {
if let Err(e) = ctx.fetch_object(reply, tx).await {
tracing::warn!("failed fetching replies for received object: {e}");
}
}
let object_model = ctx.insert_object(object_node, tx).await?;
object_model.internal
};
let expanded_addressing = ctx.expand_addressing(addressed).await?;
ctx.address_to(Some(activity_model.internal), Some(internal_oid), &expanded_addressing, tx).await?;
tracing::info!("{} posted {}", activity_model.actor, oid);
Ok(()) Ok(())
} }