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

chore: also modularized APInbox

Browse source
This commit is contained in:
əlemi 2024-04-08 02:52:51 +02:00
parent 7ce872cfff
commit 20ca18f9e3
Signed by: alemi
GPG key ID: A4895B84D311642C
5 changed files with 205 additions and 163 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
src/activitypub/mod.rs
View file

@ -132,11 +132,23 @@ pub async fn auth(State(ctx): State<Context>, Json(login): Json<LoginForm>) -> R
#[axum::async_trait] #[axum::async_trait]
pub trait APOutbox { pub trait APOutbox {
async fn post_note(&self, uid: String, object: serde_json::Value) -> crate::Result<String>; async fn create_note(&self, uid: String, object: serde_json::Value) -> crate::Result<String>;
async fn post_activity(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>; async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>;
async fn like(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>; async fn like(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>;
async fn follow(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>; async fn follow(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>;
async fn accept(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>; async fn accept(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>;
async fn reject(&self, _uid: String, _activity: serde_json::Value) -> crate::Result<String>; async fn reject(&self, _uid: String, _activity: serde_json::Value) -> crate::Result<String>;
async fn undo(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>; async fn undo(&self, uid: String, activity: serde_json::Value) -> crate::Result<String>;
} }
#[axum::async_trait]
pub trait APInbox {
async fn create(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn like(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn follow(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn accept(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn reject(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn undo(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn delete(&self, activity: serde_json::Value) -> crate::Result<()>;
async fn update(&self, activity: serde_json::Value) -> crate::Result<()>;
}

174
src/activitypub/user/inbox.rs
View file

@ -1,8 +1,8 @@
use axum::{extract::{Path, Query, State}, http::StatusCode, Json}; use axum::{extract::{Path, Query, State}, http::StatusCode, Json};
use sea_orm::{sea_query::Expr, ColumnTrait, Condition, EntityTrait, IntoActiveModel, Order, QueryFilter, QueryOrder, QuerySelect, Set}; use sea_orm::{ColumnTrait, Condition, EntityTrait, Order, QueryFilter, QueryOrder, QuerySelect};
use apb::{Activity, ActivityType, Object, ObjectType, Base, BaseType}; use apb::{ActivityType, ObjectType, Base, BaseType};
use crate::{activitypub::{activity::ap_activity, jsonld::LD, Addressed, JsonLD, Pagination}, auth::{AuthIdentity, Identity}, errors::{LoggableError, UpubError}, model, server::Context, url}; use crate::{activitypub::{activity::ap_activity, jsonld::LD, APInbox, JsonLD, Pagination}, auth::{AuthIdentity, Identity}, errors::UpubError, model, server::Context, url};
pub async fn get( pub async fn get(
State(ctx): State<Context>, State(ctx): State<Context>,
@ -68,177 +68,49 @@ pub async fn page(
pub async fn post( pub async fn post(
State(ctx): State<Context>, State(ctx): State<Context>,
Path(_id): Path<String>, Path(_id): Path<String>,
Json(object): Json<serde_json::Value> Json(activity): Json<serde_json::Value>
) -> Result<(), UpubError> { ) -> Result<(), UpubError> {
match object.base_type() { match activity.base_type() {
None => { Err(StatusCode::BAD_REQUEST.into()) }, None => { Err(StatusCode::BAD_REQUEST.into()) },
Some(BaseType::Link(_x)) => { Some(BaseType::Link(_x)) => {
tracing::warn!("skipping remote activity: {}", serde_json::to_string_pretty(&object).unwrap()); tracing::warn!("skipping remote activity: {}", serde_json::to_string_pretty(&activity).unwrap());
Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // we could but not yet Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // we could but not yet
}, },
Some(BaseType::Object(ObjectType::Activity(ActivityType::Activity))) => { Some(BaseType::Object(ObjectType::Activity(ActivityType::Activity))) => {
tracing::warn!("skipping unprocessable base activity: {}", serde_json::to_string_pretty(&object).unwrap()); tracing::warn!("skipping unprocessable base activity: {}", serde_json::to_string_pretty(&activity).unwrap());
Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // won't ingest useless stuff Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // won't ingest useless stuff
}, },
Some(BaseType::Object(ObjectType::Activity(ActivityType::Delete))) => { Some(BaseType::Object(ObjectType::Activity(ActivityType::Delete))) =>
// TODO verify the signature before just deleting lmao Ok(ctx.delete(activity).await?),
let oid = object.object().id().ok_or(StatusCode::BAD_REQUEST)?;
// TODO maybe we should keep the tombstone?
model::user::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from users");
model::activity::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from activities");
model::object::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from objects");
Ok(())
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Follow))) => { Some(BaseType::Object(ObjectType::Activity(ActivityType::Follow))) =>
let activity_targets = object.addressed(); Ok(ctx.follow(activity).await?),
let activity_entity = model::activity::Model::new(&object)?;
let aid = activity_entity.id.clone();
tracing::info!("{} wants to follow {}", activity_entity.actor, activity_entity.object.as_deref().unwrap_or("<no-one???>"));
model::activity::Entity::insert(activity_entity.into_active_model())
.exec(ctx.db()).await?;
ctx.address_to(&aid, None, &activity_targets).await?;
Ok(())
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Accept(_)))) => { Some(BaseType::Object(ObjectType::Activity(ActivityType::Accept(_)))) =>
// TODO what about TentativeAccept Ok(ctx.accept(activity).await?),
let activity_model = model::activity::Model::new(&object)?;
let Some(follow_request_id) = activity_model.object else {
return Err(StatusCode::BAD_REQUEST.into());
};
let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id)
.one(ctx.db()).await?
else {
return Err(StatusCode::NOT_FOUND.into());
};
if follow_activity.object.unwrap_or("".into()) != activity_model.actor {
return Err(StatusCode::FORBIDDEN.into());
}
tracing::info!("{} accepted follow request by {}", activity_model.actor, follow_activity.actor); Some(BaseType::Object(ObjectType::Activity(ActivityType::Reject(_)))) =>
Ok(ctx.reject(activity).await?),
model::relation::Entity::insert( Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) =>
model::relation::ActiveModel { Ok(ctx.like(activity).await?),
follower: Set(follow_activity.actor),
following: Set(activity_model.actor),
..Default::default()
}
).exec(ctx.db()).await?;
ctx.address_to(&activity_model.id, None, &object.addressed()).await?; Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) =>
Ok(()) Ok(ctx.create(activity).await?),
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Reject(_)))) => { Some(BaseType::Object(ObjectType::Activity(ActivityType::Update))) =>
// TODO what about TentativeReject? Ok(ctx.update(activity).await?),
let activity_model = model::activity::Model::new(&object)?;
let Some(follow_request_id) = activity_model.object else {
return Err(StatusCode::BAD_REQUEST.into());
};
let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id)
.one(ctx.db()).await?
else {
return Err(StatusCode::NOT_FOUND.into());
};
if follow_activity.object.unwrap_or("".into()) != activity_model.actor {
return Err(StatusCode::FORBIDDEN.into());
}
tracing::info!("{} rejected follow request by {}", activity_model.actor, follow_activity.actor);
ctx.address_to(&activity_model.id, None, &object.addressed()).await?;
Ok(())
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) => {
let aid = object.actor().id().ok_or(StatusCode::BAD_REQUEST)?;
let oid = object.object().id().ok_or(StatusCode::BAD_REQUEST)?;
let like = model::like::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
actor: sea_orm::Set(aid.clone()),
likes: sea_orm::Set(oid.clone()),
date: sea_orm::Set(chrono::Utc::now()),
};
match model::like::Entity::insert(like).exec(ctx.db()).await {
Err(sea_orm::DbErr::RecordNotInserted) => Err(StatusCode::NOT_MODIFIED.into()),
Err(sea_orm::DbErr::Exec(_)) => Err(StatusCode::NOT_MODIFIED.into()), // bad fix for sqlite
Err(e) => {
tracing::error!("unexpected error procesing like from {aid} to {oid}: {e}");
Err(StatusCode::INTERNAL_SERVER_ERROR.into())
}
Ok(_) => {
model::object::Entity::update_many()
.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1))
.filter(model::object::Column::Id.eq(oid.clone()))
.exec(ctx.db())
.await?;
tracing::info!("{} liked {}", aid, oid);
Ok(())
},
}
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) => {
let activity_model = model::activity::Model::new(&object)?;
let activity_targets = object.addressed();
let Some(object_node) = object.object().extract() else {
// TODO we could process non-embedded activities or arrays but im lazy rn
tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&object).unwrap());
return Err(StatusCode::UNPROCESSABLE_ENTITY.into());
};
let object_model = model::object::Model::new(&object_node)?;
let aid = activity_model.id.clone();
let oid = object_model.id.clone();
model::object::Entity::insert(object_model.into_active_model()).exec(ctx.db()).await?;
model::activity::Entity::insert(activity_model.into_active_model()).exec(ctx.db()).await?;
ctx.address_to(&aid, Some(&oid), &activity_targets).await?;
tracing::info!("{} posted {}", aid, oid);
Ok(())
},
Some(BaseType::Object(ObjectType::Activity(ActivityType::Update))) => {
let activity_model = model::activity::Model::new(&object)?;
let activity_targets = object.addressed();
let Some(object_node) = object.object().extract() else {
// TODO we could process non-embedded activities or arrays but im lazy rn
tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&object).unwrap());
return Err(StatusCode::UNPROCESSABLE_ENTITY.into());
};
let aid = activity_model.id.clone();
let Some(oid) = object_node.id().map(|x| x.to_string()) else {
return Err(StatusCode::BAD_REQUEST.into());
};
model::activity::Entity::insert(activity_model.into_active_model()).exec(ctx.db()).await?;
match object_node.object_type() {
Some(ObjectType::Actor(_)) => {
// TODO oof here is an example of the weakness of this model, we have to go all the way
// back up to serde_json::Value because impl Object != impl Actor
let actor_model = model::user::Model::new(&object_node)?;
model::user::Entity::update(actor_model.into_active_model())
.exec(ctx.db()).await?;
},
Some(ObjectType::Note) => {
let object_model = model::object::Model::new(&object_node)?;
model::object::Entity::update(object_model.into_active_model())
.exec(ctx.db()).await?;
},
Some(t) => tracing::warn!("no side effects implemented for update type {t:?}"),
None => tracing::warn!("empty type on embedded updated object"),
}
ctx.address_to(&aid, Some(&oid), &activity_targets).await?;
tracing::info!("{} updated {}", aid, oid);
Ok(())
},
Some(BaseType::Object(ObjectType::Activity(_x))) => { Some(BaseType::Object(ObjectType::Activity(_x))) => {
tracing::info!("received unimplemented activity on inbox: {}", serde_json::to_string_pretty(&object).unwrap()); tracing::info!("received unimplemented activity on inbox: {}", serde_json::to_string_pretty(&activity).unwrap());
Err(StatusCode::NOT_IMPLEMENTED.into()) Err(StatusCode::NOT_IMPLEMENTED.into())
}, },
Some(_x) => { Some(_x) => {
tracing::warn!("ignoring non-activity object in inbox: {}", serde_json::to_string_pretty(&object).unwrap()); tracing::warn!("ignoring non-activity object in inbox: {}", serde_json::to_string_pretty(&activity).unwrap());
Err(StatusCode::UNPROCESSABLE_ENTITY.into()) Err(StatusCode::UNPROCESSABLE_ENTITY.into())
} }
} }

4
src/activitypub/user/outbox.rs
View file

@ -79,10 +79,10 @@ pub async fn post(
Some(BaseType::Link(_)) => Err(StatusCode::UNPROCESSABLE_ENTITY.into()), Some(BaseType::Link(_)) => Err(StatusCode::UNPROCESSABLE_ENTITY.into()),
Some(BaseType::Object(ObjectType::Note)) => Some(BaseType::Object(ObjectType::Note)) =>
Ok(CreationResult(ctx.post_note(uid, activity).await?)), Ok(CreationResult(ctx.create_note(uid, activity).await?)),
Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) => Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) =>
Ok(CreationResult(ctx.post_activity(uid, activity).await?)), Ok(CreationResult(ctx.create(uid, activity).await?)),
Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) => Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) =>
Ok(CreationResult(ctx.like(uid, activity).await?)), Ok(CreationResult(ctx.like(uid, activity).await?)),

8
src/errors.rs
View file

@ -1,5 +1,3 @@
use reqwest::StatusCode;
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum UpubError { pub enum UpubError {
#[error("database error: {0}")] #[error("database error: {0}")]
@ -20,7 +18,11 @@ pub enum UpubError {
impl UpubError { impl UpubError {
pub fn bad_request() -> Self { pub fn bad_request() -> Self {
Self::Status(StatusCode::BAD_REQUEST) Self::Status(axum::http::StatusCode::BAD_REQUEST)
}
pub fn unprocessable() -> Self {
Self::Status(axum::http::StatusCode::UNPROCESSABLE_ENTITY)
} }
} }

166
src/server.rs
View file

@ -2,10 +2,10 @@ use std::{str::Utf8Error, sync::Arc};
use openssl::rsa::Rsa; use openssl::rsa::Rsa;
use reqwest::StatusCode; use reqwest::StatusCode;
use sea_orm::{ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, IntoActiveModel, QueryFilter, QuerySelect, SelectColumns, Set}; use sea_orm::{sea_query::Expr, ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, IntoActiveModel, QueryFilter, QuerySelect, SelectColumns, Set};
use crate::{activitypub::{jsonld::LD, APOutbox, Addressed, CreationResult, PUBLIC_TARGET}, dispatcher::Dispatcher, errors::UpubError, fetcher::Fetcher, model}; use crate::{activitypub::{jsonld::LD, APInbox, APOutbox, Addressed, PUBLIC_TARGET}, dispatcher::Dispatcher, errors::{LoggableError, UpubError}, fetcher::Fetcher, model};
use apb::{Activity, ActivityMut, BaseMut, CollectionMut, CollectionPageMut, CollectionType, Node, ObjectMut}; use apb::{Activity, ActivityMut, Base, BaseMut, CollectionMut, CollectionPageMut, CollectionType, Node, Object, ObjectMut};
#[derive(Clone)] #[derive(Clone)]
pub struct Context(Arc<ContextInner>); pub struct Context(Arc<ContextInner>);
@ -240,7 +240,7 @@ impl Context {
#[axum::async_trait] #[axum::async_trait]
impl APOutbox for Context { impl APOutbox for Context {
async fn post_note(&self, uid: String, object: serde_json::Value) -> crate::Result<String> { async fn create_note(&self, uid: String, object: serde_json::Value) -> crate::Result<String> {
let oid = self.oid(uuid::Uuid::new_v4().to_string()); let oid = self.oid(uuid::Uuid::new_v4().to_string());
let aid = self.aid(uuid::Uuid::new_v4().to_string()); let aid = self.aid(uuid::Uuid::new_v4().to_string());
let activity_targets = object.addressed(); let activity_targets = object.addressed();
@ -273,7 +273,7 @@ impl APOutbox for Context {
Ok(aid) Ok(aid)
} }
async fn post_activity(&self, uid: String, activity: serde_json::Value) -> crate::Result<String> { async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result<String> {
let Some(object) = activity.object().extract() else { let Some(object) = activity.object().extract() else {
return Err(UpubError::bad_request()); return Err(UpubError::bad_request());
}; };
@ -450,3 +450,159 @@ impl APOutbox for Context {
Ok(aid) Ok(aid)
} }
} }
#[axum::async_trait]
impl APInbox for Context {
async fn create(&self, activity: serde_json::Value) -> crate::Result<()> {
let activity_model = model::activity::Model::new(&activity)?;
let activity_targets = activity.addressed();
let Some(object_node) = activity.object().extract() else {
// TODO we could process non-embedded activities or arrays but im lazy rn
tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&activity).unwrap());
return Err(StatusCode::UNPROCESSABLE_ENTITY.into());
};
let object_model = model::object::Model::new(&object_node)?;
let aid = activity_model.id.clone();
let oid = object_model.id.clone();
model::object::Entity::insert(object_model.into_active_model()).exec(self.db()).await?;
model::activity::Entity::insert(activity_model.into_active_model()).exec(self.db()).await?;
self.address_to(&aid, Some(&oid), &activity_targets).await?;
tracing::info!("{} posted {}", aid, oid);
Ok(())
}
async fn like(&self, activity: serde_json::Value) -> crate::Result<()> {
let aid = activity.actor().id().ok_or(StatusCode::BAD_REQUEST)?;
let oid = activity.object().id().ok_or(StatusCode::BAD_REQUEST)?;
let like = model::like::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
actor: sea_orm::Set(aid.clone()),
likes: sea_orm::Set(oid.clone()),
date: sea_orm::Set(chrono::Utc::now()),
};
match model::like::Entity::insert(like).exec(self.db()).await {
Err(sea_orm::DbErr::RecordNotInserted) => Err(StatusCode::NOT_MODIFIED.into()),
Err(sea_orm::DbErr::Exec(_)) => Err(StatusCode::NOT_MODIFIED.into()), // bad fix for sqlite
Err(e) => {
tracing::error!("unexpected error procesing like from {aid} to {oid}: {e}");
Err(StatusCode::INTERNAL_SERVER_ERROR.into())
}
Ok(_) => {
model::object::Entity::update_many()
.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1))
.filter(model::object::Column::Id.eq(oid.clone()))
.exec(self.db())
.await?;
tracing::info!("{} liked {}", aid, oid);
Ok(())
},
}
}
async fn follow(&self, activity: serde_json::Value) -> crate::Result<()> {
let activity_targets = activity.addressed();
let activity_model = model::activity::Model::new(&activity)?;
let aid = activity_model.id.clone();
tracing::info!("{} wants to follow {}", activity_model.actor, activity_model.object.as_deref().unwrap_or("<no-one???>"));
model::activity::Entity::insert(activity_model.into_active_model())
.exec(self.db()).await?;
self.address_to(&aid, None, &activity_targets).await?;
Ok(())
}
async fn accept(&self, activity: serde_json::Value) -> crate::Result<()> {
// TODO what about TentativeAccept
let activity_model = model::activity::Model::new(&activity)?;
let Some(follow_request_id) = activity_model.object else {
return Err(StatusCode::BAD_REQUEST.into());
};
let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id)
.one(self.db()).await?
else {
return Err(StatusCode::NOT_FOUND.into());
};
if follow_activity.object.unwrap_or("".into()) != activity_model.actor {
return Err(StatusCode::FORBIDDEN.into());
}
tracing::info!("{} accepted follow request by {}", activity_model.actor, follow_activity.actor);
model::relation::Entity::insert(
model::relation::ActiveModel {
follower: Set(follow_activity.actor),
following: Set(activity_model.actor),
..Default::default()
}
).exec(self.db()).await?;
self.address_to(&activity_model.id, None, &activity.addressed()).await?;
Ok(())
}
async fn reject(&self, activity: serde_json::Value) -> crate::Result<()> {
// TODO what about TentativeReject?
let activity_model = model::activity::Model::new(&activity)?;
let Some(follow_request_id) = activity_model.object else {
return Err(StatusCode::BAD_REQUEST.into());
};
let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id)
.one(self.db()).await?
else {
return Err(StatusCode::NOT_FOUND.into());
};
if follow_activity.object.unwrap_or("".into()) != activity_model.actor {
return Err(StatusCode::FORBIDDEN.into());
}
tracing::info!("{} rejected follow request by {}", activity_model.actor, follow_activity.actor);
self.address_to(&activity_model.id, None, &activity.addressed()).await?;
Ok(())
}
async fn delete(&self, activity: serde_json::Value) -> crate::Result<()> {
// TODO verify the signature before just deleting lmao
let oid = activity.object().id().ok_or(StatusCode::BAD_REQUEST)?;
// TODO maybe we should keep the tombstone?
model::user::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from users");
model::activity::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from activities");
model::object::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from objects");
Ok(())
}
async fn update(&self, activity: serde_json::Value) -> crate::Result<()> {
let activity_model = model::activity::Model::new(&activity)?;
let activity_targets = activity.addressed();
let Some(object_node) = activity.object().extract() else {
// TODO we could process non-embedded activities or arrays but im lazy rn
tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&activity).unwrap());
return Err(UpubError::unprocessable());
};
let aid = activity_model.id.clone();
let Some(oid) = object_node.id().map(|x| x.to_string()) else {
return Err(UpubError::bad_request());
};
model::activity::Entity::insert(activity_model.into_active_model()).exec(self.db()).await?;
match object_node.object_type() {
Some(apb::ObjectType::Actor(_)) => {
// TODO oof here is an example of the weakness of this model, we have to go all the way
// back up to serde_json::Value because impl Object != impl Actor
let actor_model = model::user::Model::new(&object_node)?;
model::user::Entity::update(actor_model.into_active_model())
.exec(self.db()).await?;
},
Some(apb::ObjectType::Note) => {
let object_model = model::object::Model::new(&object_node)?;
model::object::Entity::update(object_model.into_active_model())
.exec(self.db()).await?;
},
Some(t) => tracing::warn!("no side effects implemented for update type {t:?}"),
None => tracing::warn!("empty type on embedded updated object"),
}
self.address_to(&aid, Some(&oid), &activity_targets).await?;
tracing::info!("{} updated {}", aid, oid);
Ok(())
}
async fn undo(&self, _activity: serde_json::Value) -> crate::Result<()> {
todo!()
}
}