use super::{ actor::{self, ID}, conversation_context, db, federate, gen_converation_id, types::{ArticleExt, Conversation}, util::{accept, rewrite_srcs}, DOMAIN, }; use crate::generator::{HtmlContent, Post}; use activitystreams::{ activity::Create, context, iri_string::types::IriString, object::{Article, ObjectExt}, prelude::{BaseExt, ExtendsExt}, public, time::OffsetDateTime, }; use axum::{body::Body, http::Request, response::IntoResponse}; use log::{error, info}; use mime::Mime; use once_cell::sync::Lazy; use sqlx::SqlitePool; use url::Url; pub async fn insert_posts(posts: &[Post], pool: &SqlitePool) { let existing = sqlx::query!("select id from articles") .fetch_all(pool) .await .unwrap(); for post in posts.iter() { let id = format!("https://{}{}", *DOMAIN, post.comments_permalink()); if existing.iter().any(|e| e.id == id) { continue; } let mut article = ArticleExt::new(Article::new(), Conversation::new()); article.add_context(context()); article.add_context(conversation_context()); article.set_id(id.parse::().unwrap()); article.set_attributed_to(ID.clone()); article.set_published( OffsetDateTime::from_unix_timestamp(post.metadata.date.timestamp()) .expect("convert chrono::DateTime to time::OffsetDateTime"), ); article.ext_one.conversation = Some(gen_converation_id()); article.add_to(public()); article.add_cc(actor::FOLLOWERS.clone()); article.set_name(post.metadata.title.clone()); let content = post.content.html(); let content_with_absolute_srcs = rewrite_srcs::rewrite(&content, &Url::parse(&id).unwrap()); article.set_content(content_with_absolute_srcs); let conversation = article.ext_one.conversation.as_ref().unwrap(); let article_json: String = serde_json::to_string_pretty(&article).unwrap(); sqlx::query!("insert into articles (id, conversation, has_federated, article_object) values (?1, ?2, false, ?3)", id, conversation, article_json) .execute(pool) .await.unwrap(); } } static AP_JSON: Lazy = Lazy::new(|| "application/activity+json".parse::().unwrap()); static LD_JSON: Lazy = Lazy::new(|| "application/ld+json".parse::().unwrap()); pub async fn handle(request: &Request) -> Option { let mimes = [&*AP_JSON, &*LD_JSON, &mime::APPLICATION_JSON]; let best = accept::best_match(request.headers(), &mimes); match best { Some(t) if t == &*AP_JSON || t == &*LD_JSON || *t == mime::APPLICATION_JSON => { let pool = request.extensions().get::().unwrap(); if let Some(article_json) = db::get_article_for_path(request.uri().path(), pool).await { let headers = [("Content-Type", best.unwrap().to_string())]; Some((headers, article_json).into_response()) } else { None } } _ => None, } } pub async fn federate_outgoing(pool: &SqlitePool) -> anyhow::Result<()> { let articles_objs = db::get_unfederated_articles(pool).await?; let followers = db::get_followers(pool).await?; let mut inboxes = vec![]; for actor in followers { let inbox = actor .shared_inbox .clone() .unwrap_or_else(|| actor.inbox.clone()); if !inboxes.contains(&inbox) { inboxes.push(inbox); } } for (article_id, article_obj) in articles_objs { info!("Federating out article {}", article_id); let article = serde_json::from_str::
(&article_obj)?; let to = article.to().unwrap().clone().many().unwrap(); let cc = article.cc().unwrap().clone().many().unwrap(); let mut create = Create::new(ID.as_str(), article.into_any_base()?); create.add_context(context()); create.add_context(conversation_context()); create.set_many_tos(to); create.set_many_ccs(cc); for inbox in &inboxes { match federate::sign_and_send(&create, inbox).await { Ok(()) => (), Err(e) => error!( "federating outbound article {} to {}: {:?}", article_id, inbox, e ), } } db::set_has_federated(&article_id, true, pool).await?; } Ok(()) }