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

feat: fetch threads recursively up to 16

Browse source
This commit is contained in:
əlemi 2024-04-22 00:43:17 +02:00
parent d575ef0bef
commit e9a19b3cb4
Signed by: alemi
GPG key ID: A4895B84D311642C
2 changed files with 39 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -41,6 +41,7 @@ sea-orm-migration = { version = "0.12", optional = true }
# mastodon # mastodon
mastodon-async-entities = { version = "1.1.0", optional = true } mastodon-async-entities = { version = "1.1.0", optional = true }
time = { version = "0.3", features = ["serde"], optional = true } time = { version = "0.3", features = ["serde"], optional = true }
async-recursion = "1.1"
[features] [features]
default = ["faker", "migrations", "mastodon"] default = ["faker", "migrations", "mastodon"]

29
src/server/fetcher.rs
View file

@ -120,32 +120,45 @@ impl Fetcher for Context {
} }
async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model> { async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model> {
if let Some(x) = model::object::Entity::find_by_id(id).one(self.db()).await? { fetch_object_inner(self, id, 0).await
}
}
#[async_recursion::async_recursion]
async fn fetch_object_inner(ctx: &Context, id: &str, depth: usize) -> crate::Result<model::object::Model> {
if let Some(x) = model::object::Entity::find_by_id(id).one(ctx.db()).await? {
return Ok(x); // already in db, easy return Ok(x); // already in db, easy
} }
let object = Self::request( let object = Context::request(
Method::GET, id, None, &format!("https://{}", self.domain()), &self.app().private_key, self.domain(), Method::GET, id, None, &format!("https://{}", ctx.domain()), &ctx.app().private_key, ctx.domain(),
).await?.json::<serde_json::Value>().await?; ).await?.json::<serde_json::Value>().await?;
let addressed = object.addressed(); let addressed = object.addressed();
let object_model = model::object::Model::new(&object)?; let object_model = model::object::Model::new(&object)?;
if let Some(reply) = &object_model.in_reply_to {
if depth <= 16 {
fetch_object_inner(ctx, reply, depth + 1).await?;
} else {
tracing::warn!("thread deeper than 16, giving up fetching more replies");
}
}
for attachment in object.attachment() { for attachment in object.attachment() {
let attachment_model = model::attachment::ActiveModel::new(&attachment, object_model.id.clone())?; let attachment_model = model::attachment::ActiveModel::new(&attachment, object_model.id.clone())?;
model::attachment::Entity::insert(attachment_model) model::attachment::Entity::insert(attachment_model)
.exec(self.db()) .exec(ctx.db())
.await?; .await?;
} }
let expanded_addresses = self.expand_addressing(addressed).await?; let expanded_addresses = ctx.expand_addressing(addressed).await?;
self.address_to(None, Some(&object_model.id), &expanded_addresses).await?; ctx.address_to(None, Some(&object_model.id), &expanded_addresses).await?;
model::object::Entity::insert(object_model.clone().into_active_model()) model::object::Entity::insert(object_model.clone().into_active_model())
.exec(self.db()).await?; .exec(ctx.db()).await?;
Ok(object_model) Ok(object_model)
}
} }
#[axum::async_trait] #[axum::async_trait]