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

fix: pass transaction to expand_addressing too

Browse source
This commit is contained in:
əlemi 2024-06-06 19:04:39 +02:00
parent f6d30b3bec
commit 1bb8df0ac5
Signed by: alemi
GPG key ID: A4895B84D311642C
4 changed files with 15 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
upub/core/src/model/relation.rs
View file

@ -63,7 +63,7 @@ impl ActiveModelBehavior for ActiveModel {}
impl Entity { impl Entity {
// TODO this is 2 queries!!! can it be optimized down to 1? // TODO this is 2 queries!!! can it be optimized down to 1?
pub async fn followers(uid: &str, db: &DatabaseConnection) -> Result<Option<Vec<String>>, DbErr> { pub async fn followers(uid: &str, db: &impl ConnectionTrait) -> Result<Option<Vec<String>>, DbErr> {
let Some(internal_id) = super::actor::Entity::ap_to_internal(uid, db).await? let Some(internal_id) = super::actor::Entity::ap_to_internal(uid, db).await?
else { else {
return Ok(None); return Ok(None);
@ -88,7 +88,7 @@ impl Entity {
} }
// TODO this is 2 queries!!! can it be optimized down to 1? // TODO this is 2 queries!!! can it be optimized down to 1?
pub async fn following(uid: &str, db: &DatabaseConnection) -> Result<Option<Vec<String>>, DbErr> { pub async fn following(uid: &str, db: &impl ConnectionTrait) -> Result<Option<Vec<String>>, DbErr> {
let Some(internal_id) = super::actor::Entity::ap_to_internal(uid, db).await? let Some(internal_id) = super::actor::Entity::ap_to_internal(uid, db).await?
else { else {
return Ok(None); return Ok(None);

6
upub/core/src/traits/address.rs
View file

@ -4,19 +4,19 @@ use crate::traits::fetch::Fetcher;
#[async_trait::async_trait] #[async_trait::async_trait]
pub trait Addresser { pub trait Addresser {
async fn expand_addressing(&self, targets: Vec<String>) -> Result<Vec<String>, DbErr>; async fn expand_addressing(&self, targets: Vec<String>, tx: &impl ConnectionTrait) -> Result<Vec<String>, DbErr>;
async fn address_to(&self, aid: Option<i64>, oid: Option<i64>, targets: &[String], tx: &impl ConnectionTrait) -> Result<(), DbErr>; async fn address_to(&self, aid: Option<i64>, oid: Option<i64>, targets: &[String], tx: &impl ConnectionTrait) -> Result<(), DbErr>;
async fn deliver_to(&self, aid: &str, from: &str, targets: &[String], tx: &impl ConnectionTrait) -> Result<(), DbErr>; async fn deliver_to(&self, aid: &str, from: &str, targets: &[String], tx: &impl ConnectionTrait) -> Result<(), DbErr>;
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl Addresser for crate::Context { impl Addresser for crate::Context {
async fn expand_addressing(&self, targets: Vec<String>) -> Result<Vec<String>, DbErr> { async fn expand_addressing(&self, targets: Vec<String>, tx: &impl ConnectionTrait) -> Result<Vec<String>, DbErr> {
let mut out = Vec::new(); let mut out = Vec::new();
for target in targets { for target in targets {
if target.ends_with("/followers") { if target.ends_with("/followers") {
let target_id = target.replace("/followers", ""); let target_id = target.replace("/followers", "");
let mut followers = crate::model::relation::Entity::followers(&target_id, self.db()) let mut followers = crate::model::relation::Entity::followers(&target_id, tx)
.await? .await?
.unwrap_or_else(Vec::new); .unwrap_or_else(Vec::new);
if followers.is_empty() { // stuff with zero addressing will never be seen again!!! TODO if followers.is_empty() { // stuff with zero addressing will never be seen again!!! TODO

4
upub/core/src/traits/fetch.rs
View file

@ -358,7 +358,7 @@ impl Fetcher for crate::Context {
let activity_model = self.insert_activity(activity, tx).await?; let activity_model = self.insert_activity(activity, tx).await?;
let addressed = activity_model.addressed(); let addressed = activity_model.addressed();
let expanded_addresses = self.expand_addressing(addressed).await?; let expanded_addresses = self.expand_addressing(addressed, tx).await?;
self.address_to(Some(activity_model.internal), None, &expanded_addresses, tx).await?; self.address_to(Some(activity_model.internal), None, &expanded_addresses, tx).await?;
Ok(activity_model) Ok(activity_model)
@ -418,7 +418,7 @@ async fn resolve_object_r(ctx: &crate::Context, object: serde_json::Value, depth
let object_model = ctx.insert_object(object, tx).await?; let object_model = ctx.insert_object(object, tx).await?;
let expanded_addresses = ctx.expand_addressing(addressed).await?; let expanded_addresses = ctx.expand_addressing(addressed, tx).await?;
ctx.address_to(None, Some(object_model.internal), &expanded_addresses, tx).await?; ctx.address_to(None, Some(object_model.internal), &expanded_addresses, tx).await?;
Ok(object_model) Ok(object_model)

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

@ -75,7 +75,7 @@ pub async fn create(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
let object_model = ctx.insert_object(object_node, tx).await?; let object_model = ctx.insert_object(object_node, tx).await?;
object_model.internal object_model.internal
}; };
let expanded_addressing = ctx.expand_addressing(addressed).await?; let expanded_addressing = ctx.expand_addressing(addressed, tx).await?;
ctx.address_to(Some(activity_model.internal), Some(internal_oid), &expanded_addressing, tx).await?; ctx.address_to(Some(activity_model.internal), Some(internal_oid), &expanded_addressing, tx).await?;
tracing::info!("{} posted {}", activity_model.actor, oid); tracing::info!("{} posted {}", activity_model.actor, oid);
Ok(()) Ok(())
@ -110,7 +110,7 @@ pub async fn like(ctx: &crate::Context, activity: impl apb::Activity, tx: &Datab
.exec(tx) .exec(tx)
.await?; .await?;
let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed()).await?; let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed(), tx).await?;
if expanded_addressing.is_empty() { // WHY MASTODON!!!!!!! if expanded_addressing.is_empty() { // WHY MASTODON!!!!!!!
expanded_addressing.push( expanded_addressing.push(
crate::model::object::Entity::find_by_id(obj.internal) crate::model::object::Entity::find_by_id(obj.internal)
@ -144,7 +144,7 @@ pub async fn follow(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
}; };
crate::model::relation::Entity::insert(relation_model) crate::model::relation::Entity::insert(relation_model)
.exec(tx).await?; .exec(tx).await?;
let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed()).await?; let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed(), tx).await?;
if !expanded_addressing.contains(&target_actor) { if !expanded_addressing.contains(&target_actor) {
expanded_addressing.push(target_actor); expanded_addressing.push(target_actor);
} }
@ -192,7 +192,7 @@ pub async fn accept(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
tracing::info!("{} accepted follow request by {}", target_actor, follow_activity.actor); tracing::info!("{} accepted follow request by {}", target_actor, follow_activity.actor);
let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed()).await?; let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed(), tx).await?;
if !expanded_addressing.contains(&follow_activity.actor) { if !expanded_addressing.contains(&follow_activity.actor) {
expanded_addressing.push(follow_activity.actor); expanded_addressing.push(follow_activity.actor);
} }
@ -222,7 +222,7 @@ pub async fn reject(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
tracing::info!("{} rejected follow request by {}", uid, follow_activity.actor); tracing::info!("{} rejected follow request by {}", uid, follow_activity.actor);
let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed()).await?; let mut expanded_addressing = ctx.expand_addressing(activity_model.addressed(), tx).await?;
if !expanded_addressing.contains(&follow_activity.actor) { if !expanded_addressing.contains(&follow_activity.actor) {
expanded_addressing.push(follow_activity.actor); expanded_addressing.push(follow_activity.actor);
} }
@ -276,7 +276,7 @@ pub async fn update(ctx: &crate::Context, activity: impl apb::Activity, tx: &Dat
} }
tracing::info!("{} updated {}", uid, oid); tracing::info!("{} updated {}", uid, oid);
let expanded_addressing = ctx.expand_addressing(activity_model.addressed()).await?; let expanded_addressing = ctx.expand_addressing(activity_model.addressed(), tx).await?;
ctx.address_to(Some(activity_model.internal), None, &expanded_addressing, tx).await?; ctx.address_to(Some(activity_model.internal), None, &expanded_addressing, tx).await?;
Ok(()) Ok(())
} }
@ -298,7 +298,7 @@ pub async fn undo(ctx: &crate::Context, activity: impl apb::Activity, tx: &Datab
.ok_or(ProcessorError::Incomplete)?; .ok_or(ProcessorError::Incomplete)?;
let activity_type = activity.activity_type()?; let activity_type = activity.activity_type()?;
let targets = ctx.expand_addressing(activity.addressed()).await?; let targets = ctx.expand_addressing(activity.addressed(), tx).await?;
let activity_model = ctx.insert_activity(activity, tx).await?; let activity_model = ctx.insert_activity(activity, tx).await?;
ctx.address_to(Some(activity_model.internal), None, &targets, tx).await?; ctx.address_to(Some(activity_model.internal), None, &targets, tx).await?;
@ -400,7 +400,7 @@ pub async fn announce(ctx: &crate::Context, activity: impl apb::Activity, tx: &D
published: Set(published), published: Set(published),
}; };
let expanded_addressing = ctx.expand_addressing(addressed).await?; let expanded_addressing = ctx.expand_addressing(addressed, tx).await?;
ctx.address_to(Some(activity_model.internal), None, &expanded_addressing, tx).await?; ctx.address_to(Some(activity_model.internal), None, &expanded_addressing, tx).await?;
crate::model::announce::Entity::insert(share) crate::model::announce::Entity::insert(share)
.exec(tx).await?; .exec(tx).await?;