feat: http wrapper, start type rewrite, add nextest

Cargo.lock

@@ -2162,7 +2162,10 @@ dependencies = [
  "reqwest",
  "rocket",
  "rocket_db_pools",
+ "serde",
+ "serde_json",
  "sqlx",
+ "thiserror 2.0.12",
  "tracing",
  "tracing-subscriber",
  "url",

Cargo.toml

@@ -11,6 +11,7 @@ uuid = { version = "1.16.0", features = ["v4"] }
 chrono = "0.4.40"
 rand = "0.8"
 thiserror = "2.0.12"
+serde_json = "1.0.140"
 
 tracing = "0.1.40"
 tracing-appender = "0.2.3"

ferri-main/Cargo.toml

@@ -12,8 +12,8 @@ uuid = { workspace = true }
 rand = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
+serde_json = { workspace = true }
 
 base64 = "0.22.1"
 rsa = { version = "0.9.8", features = ["sha2"] }
 url = "2.5.4"
-serde_json = "1.0.140"

ferri-main/src/ap/mod.rs

@@ -12,6 +12,9 @@ pub use user::*;
 mod post;
 pub use post::*;
 
+mod request_queue;
+pub use request_queue::*;
+
 pub const AS_CONTEXT: &str = "https://www.w3.org/ns/activitystreams";
 
 pub fn new_id() -> String {

ferri-main/src/ap/post.rs

@@ -3,7 +3,7 @@ use chrono::{DateTime, Utc};
 use serde::Serialize;
 use sqlx::Sqlite;
 
-const POST_TYPE: &str = "Post";
+const POST_TYPE: &str = "Note";
 
 #[derive(Clone)]
 pub struct Post {

ferri-main/src/ap/request_queue.rs

@@ -0,0 +1,59 @@
+use std::sync::mpsc;
+use std::thread;
+use tracing::{info, span, Level};
+
+#[derive(Debug)]
+pub enum QueueMessage {
+    Heartbeat
+}
+
+pub struct RequestQueue {
+    name: &'static str,
+    send: mpsc::Sender<QueueMessage>,
+    recv: mpsc::Receiver<QueueMessage>
+}
+
+#[derive(Clone)]
+pub struct QueueHandle {
+    send: mpsc::Sender<QueueMessage>
+}
+
+impl QueueHandle {
+    pub fn send(&self, msg: QueueMessage) {
+        self.send.send(msg).unwrap();
+    }
+}
+
+impl RequestQueue {
+    pub fn new(name: &'static str) -> Self {
+        let (send, recv) = mpsc::channel();
+        Self {
+            name,
+            send,
+            recv
+        }
+    }
+
+    pub fn spawn(self) -> QueueHandle {
+        info!("starting up queue '{}'", self.name);
+        
+        thread::spawn(move || {
+            info!("queue '{}' up", self.name);
+            let recv = self.recv;
+            
+            while let Ok(req) = recv.recv() {
+                // FIXME: When we make this do async things we will need to add tokio and
+                //        use proper async handled spans as the enter/drop won't work.
+                //        See inbox.rs for how we handle that.
+                let s = span!(Level::INFO, "queue", queue_name = self.name);
+                let _enter = s.enter();
+                
+                info!(?req, "got a message into the queue");
+                
+                drop(_enter);
+            }
+        });
+        
+        QueueHandle { send: self.send }
+    }
+}

ferri-main/src/ap/user.rs

@@ -106,9 +106,10 @@ impl User {
             "#,
             username
         )
-        .fetch_one(conn)
+            .fetch_one(conn)
-        .await
+            .await
-        .unwrap();
+            .unwrap();
+        
         User {
             id: user.id,
             username: user.username,

ferri-main/src/config/mod.rs

@@ -9,3 +9,41 @@ pub struct ServerConfig {
 pub struct Config {
     pub server: ServerConfig,
 }
+
+impl Config {
+    pub fn host(&self) -> &str {
+        &self.server.host
+    }
+    
+    pub fn user_url(&self, user_uuid: &str) -> String {
+        format!("{}/users/{}", self.host(), user_uuid)
+    }
+
+    pub fn user_web_url(&self, user_name: &str) -> String {
+        format!("{}/{}", self.host(), user_name)
+    }
+
+    pub fn followers_url(&self, user_uuid: &str) -> String {
+        format!("{}/followers", self.user_url(user_uuid))
+    }
+    
+    pub fn following_url(&self, user_uuid: &str) -> String {
+        format!("{}/following", self.user_url(user_uuid))
+    }
+
+    pub fn inbox_url(&self, user_uuid: &str) -> String {
+        format!("{}/inbox", self.user_url(user_uuid))
+    }
+
+    pub fn outbox_url(&self, user_uuid: &str) -> String {
+        format!("{}/outbox", self.user_url(user_uuid))
+    }
+
+    pub fn post_url(&self, poster_uuid: &str, post_uuid: &str) -> String {
+        format!("{}/{}", self.user_url(poster_uuid), post_uuid)
+    }
+
+    pub fn activity_url(&self, activity_uuid: &str) -> String {
+        format!("{}/activities/{}", self.host(), activity_uuid)
+    }
+}

ferri-main/src/lib.rs

@@ -1,5 +1,7 @@
 pub mod ap;
 pub mod config;
+mod types_rewrite;
+
 use rand::{Rng, distributions::Alphanumeric};
 
 pub fn gen_token(len: usize) -> String {

ferri-main/src/types_rewrite/convert.rs

@@ -0,0 +1,28 @@
+use crate::types_rewrite::api;
+use crate::types_rewrite::ap;
+use crate::types_rewrite::db;
+
+use crate::types_rewrite::{Object, as_context};
+
+impl From<db::Actor> for ap::Actor {
+    fn from(val: db::Actor) -> ap::Actor {
+        ap::Actor {
+            obj: Object {
+                context: as_context(),
+                id: val.id
+            },
+            inbox: val.inbox,
+            outbox: val.outbox
+        }
+    }
+}
+
+impl From<ap::Actor> for db::Actor {
+    fn from(val: ap::Actor) -> db::Actor {
+        db::Actor {
+            id: val.obj.id,
+            inbox: val.inbox,
+            outbox: val.outbox
+        }
+    }
+}

ferri-main/src/types_rewrite/mod.rs

@@ -0,0 +1,87 @@
+use serde::{Serialize, Deserialize};
+
+mod convert;
+pub use convert::*;
+
+pub const AS_CONTEXT_RAW: &'static str = "https://www.w3.org/ns/activitystreams";
+pub fn as_context() -> ObjectContext {
+    ObjectContext::Str(AS_CONTEXT_RAW.to_string())
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+#[serde(untagged)]
+pub enum ObjectContext {
+    Str(String),
+    Vec(Vec<serde_json::Value>),
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct ObjectUri(String);
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct Object {
+    #[serde(rename = "@context")]
+    context: ObjectContext,
+    id: ObjectUri,
+}
+
+pub mod db {
+    use serde::{Serialize, Deserialize};
+    use super::*;
+    
+    #[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+    pub struct Actor {
+        pub id: ObjectUri,
+        pub inbox: String,
+        pub outbox: String,
+    }    
+}
+
+pub mod ap {
+    use serde::{Serialize, Deserialize};
+    use super::*;
+    
+    #[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+    pub struct Actor {
+        #[serde(flatten)]
+        pub obj: Object,
+        
+        pub inbox: String,
+        pub outbox: String,
+    }
+}
+
+pub mod api {
+    use serde::{Serialize, Deserialize};
+    use super::*;
+    
+    // API will not really use actors so treat them as DB actors
+    // until we require specificity
+    pub type Actor = db::Actor;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    
+    #[test]
+    fn ap_actor_to_db() {
+        let domain = "https://example.com";
+        let ap = ap::Actor {
+            obj: Object {
+                context: as_context(),
+                id: ObjectUri(format!("{}/users/sample", domain)),
+            },
+            inbox: format!("{}/users/sample/inbox", domain),
+            outbox: format!("{}/users/sample/outbox", domain),
+        };
+
+        let db: db::Actor = ap.into();
+        
+        assert_eq!(db, db::Actor {
+            id: ObjectUri("https://example.com/users/sample".to_string()),
+            inbox: "https://example.com/users/sample/inbox".to_string(),
+            outbox: "https://example.com/users/sample/outbox".to_string(),
+        });
+    }
+}

ferri-server/Cargo.toml

@@ -15,4 +15,7 @@ uuid = { workspace = true }
 chrono = { workspace = true }
 rand = { workspace = true }
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true }
+tracing-subscriber = { workspace = true }
+thiserror = { workspace = true }
+serde_json = { workspace = true }
+serde = { workspace = true }

ferri-server/src/endpoints/api/instance.rs

@@ -1,4 +1,6 @@
-use rocket::{get, serde::json::Json};
+use rocket::{get, serde::json::Json, State};
+
+use crate::Config;
 
 use crate::types::instance::{
     Accounts, Configuration, Contact, Instance, MediaAttachments, Polls, Registrations, Statuses,
@@ -6,9 +8,9 @@ use crate::types::instance::{
 };
 
 #[get("/instance")]
-pub async fn instance() -> Json<Instance> {
+pub async fn instance(config: &State<Config>) -> Json<Instance> {
     Json(Instance {
-        domain: "ferri.amy.mov".to_string(),
+        domain: config.host().to_string(),
         title: "Ferri".to_string(),
         version: "0.0.1".to_string(),
         source_url: "https://forge.amy.mov/amy/ferri".to_string(),

ferri-server/src/endpoints/api/status.rs

@@ -26,11 +26,11 @@ pub struct StatusContext {
     descendants: Vec<Status>
 }
 
-#[get("/statuses/<status>/context")]
+#[get("/statuses/<_status>/context")]
 pub async fn status_context(
-    status: &str,
+    _status: &str,
-    user: AuthenticatedUser,
+    _user: AuthenticatedUser,
-    mut db: Connection<Db>
+    _db: Connection<Db>
 ) -> Json<StatusContext> {
     Json(StatusContext {
         ancestors: vec![],
@@ -56,44 +56,37 @@ async fn create_status(
 
     post.save(&mut **db).await;
 
-    let actors = sqlx::query!("SELECT * FROM actor")
+    let actor = sqlx::query!("SELECT * FROM actor WHERE id = ?1", "https://fedi.amy.mov/users/9zkygethkdw60001")
-        .fetch_all(&mut **db)
+        .fetch_one(&mut **db)
         .await
         .unwrap();
 
-    for record in actors {
+    let create_id = format!("https://ferri.amy.mov/activities/{}", Uuid::new_v4());
-        // Don't send to ourselves
-        if record.id == user.actor_id() {
-            continue;
-        }
 
-        let create_id = format!("https://ferri.amy.mov/activities/{}", Uuid::new_v4());
+    let activity = ap::Activity {
+        id: create_id,
+        ty: ap::ActivityType::Create,
+        object: post.clone().to_ap(),
+        to: vec![format!("{}/followers", user.uri())],
+        published: now,
+        cc: vec!["https://www.w3.org/ns/activitystreams#Public".to_string()],
+        ..Default::default()
+    };
 
-        let activity = ap::Activity {
+    let actor = ap::Actor::from_raw(
-            id: create_id,
+        actor.id.clone(),
-            ty: ap::ActivityType::Create,
+        actor.inbox.clone(),
-            object: post.clone().to_ap(),
+        actor.outbox.clone(),
-            to: vec![format!("{}/followers", user.uri())],
+    );
-            published: now,
-            cc: vec!["https://www.w3.org/ns/activitystreams#Public".to_string()],
-            ..Default::default()
-        };
 
-        let actor = ap::Actor::from_raw(
+    let req = ap::OutgoingActivity {
-            record.id.clone(),
+        req: activity,
-            record.inbox.clone(),
+        signed_by: format!("{}#main-key", user.uri()),
-            record.outbox.clone(),
+        to: actor,
-        );
+    };
 
-        let req = ap::OutgoingActivity {
+    req.save(&mut **db).await;
-            req: activity,
+    outbox.post(req).await;
-            signed_by: format!("{}#main-key", user.uri()),
-            to: actor,
-        };
-
-        req.save(&mut **db).await;
-        outbox.post(req).await;
-    }
 
     TimelineStatus {
         id: post.id().to_string(),
@@ -111,6 +104,7 @@ async fn create_status(
         favourites_count: 0,
         favourited: false,
         reblogged: false,
+        reblog: None,
         muted: false,
         bookmarked: false,
         media_attachments: vec![],

ferri-server/src/endpoints/api/timeline.rs

@@ -1,5 +1,6 @@
-use crate::{AuthenticatedUser, Db, endpoints::api::user::CredentialAcount};
+use crate::{AuthenticatedUser, Db, endpoints::api::user::CredentialAcount, Config};
 use rocket::{
+    State,
     get,
     serde::{Deserialize, Serialize, json::Json},
 };
@@ -27,20 +28,21 @@ pub struct TimelineStatus {
     pub reblogged: bool,
     pub muted: bool,
     pub bookmarked: bool,
+    pub reblog: Option<Box<TimelineStatus>>,
     pub media_attachments: Vec<()>,
     pub account: TimelineAccount,
 }
 
-#[get("/timelines/home?<limit>")]
+#[get("/timelines/home")]
 pub async fn home(
     mut db: Connection<Db>,
-    limit: i64,
+    config: &State<Config>,
-    user: AuthenticatedUser,
+    _user: AuthenticatedUser,
 ) -> Json<Vec<TimelineStatus>> {
     let posts = sqlx::query!(
         r#"
             SELECT p.id as "post_id", u.id as "user_id", p.content, p.uri as "post_uri", 
-                u.username, u.display_name, u.actor_id, p.created_at 
+                u.username, u.display_name, u.actor_id, p.created_at, p.boosted_post_id
             FROM post p
             INNER JOIN user u on p.user_id = u.id
         "#
@@ -51,7 +53,65 @@ pub async fn home(
 
     let mut out = Vec::<TimelineStatus>::new();
     for record in posts {
-        let user_uri = format!("https://ferri.amy.mov/users/{}", record.username);
+        let mut boost: Option<Box<TimelineStatus>> = None;
+        if let Some(boosted_id) = record.boosted_post_id {
+            let record = sqlx::query!(
+               r#"
+                 SELECT p.id as "post_id", u.id as "user_id", p.content, p.uri as "post_uri", 
+                   u.username, u.display_name, u.actor_id, p.created_at, p.boosted_post_id
+                 FROM post p
+                 INNER JOIN user u on p.user_id = u.id
+                 WHERE p.id = ?1
+               "#, boosted_id)
+                .fetch_one(&mut **db)
+                .await
+                .unwrap();
+            
+            let user_uri = config.user_url(&record.user_id);
+            boost = Some(Box::new(TimelineStatus {
+                id: record.post_id.clone(),
+                created_at: record.created_at.clone(),
+                in_reply_to_id: None,
+                in_reply_to_account_id: None,
+                content: record.content.clone(),
+                visibility: "public".to_string(),
+                spoiler_text: "".to_string(),
+                sensitive: false,
+                uri: record.post_uri.clone(),
+                url: record.post_uri.clone(),
+                replies_count: 0,
+                reblogs_count: 0,
+                favourites_count: 0,
+                favourited: false,
+                reblogged: false,
+                reblog: boost,
+                muted: false,
+                bookmarked: false,
+                media_attachments: vec![],
+                account: CredentialAcount {
+                    id: record.user_id.clone(),
+                    username: record.username.clone(),
+                    acct: record.username.clone(),
+                    display_name: record.display_name.clone(),
+                    locked: false,
+                    bot: false,
+                    created_at: "2025-04-10T22:12:09Z".to_string(),
+                    attribution_domains: vec![],
+                    note: "".to_string(),
+                    url: user_uri,
+                    avatar: "https://ferri.amy.mov/assets/pfp.png".to_string(),
+                    avatar_static: "https://ferri.amy.mov/assets/pfp.png".to_string(),
+                    header: "https://ferri.amy.mov/assets/pfp.png".to_string(),
+                    header_static: "https://ferri.amy.mov/assets/pfp.png".to_string(),
+                    followers_count: 1,
+                    following_count: 1,
+                    statuses_count: 1,
+                    last_status_at: "2025-04-10T22:14:34Z".to_string(),
+                },                
+            }))
+        }
+        
+        let user_uri = config.user_web_url(&record.username);
         out.push(TimelineStatus {
             id: record.post_id.clone(),
             created_at: record.created_at.clone(),
@@ -68,6 +128,7 @@ pub async fn home(
             favourites_count: 0,
             favourited: false,
             reblogged: false,
+            reblog: boost,
             muted: false,
             bookmarked: false,
             media_attachments: vec![],

ferri-server/src/endpoints/api/user.rs

@@ -95,7 +95,7 @@ pub async fn new_follow(
 pub async fn account(
     mut db: Connection<Db>,
     uuid: &str,
-    user: AuthenticatedUser,
+    _user: AuthenticatedUser,
 ) -> Result<Json<TimelineAccount>, NotFound<String>> {
     let user = ap::User::from_id(uuid, &mut **db)
         .await
@@ -123,12 +123,12 @@ pub async fn account(
     }))
 }
 
-#[get("/accounts/<uuid>/statuses?<limit>")]
+#[get("/accounts/<uuid>/statuses?<_limit>")]
 pub async fn statuses(
     mut db: Connection<Db>,
     uuid: &str,
-    limit: Option<i64>,
+    _limit: Option<i64>,
-    user: AuthenticatedUser,
+    _user: AuthenticatedUser,
 ) -> Result<Json<Vec<TimelineStatus>>, NotFound<String>> {
     let user = ap::User::from_id(uuid, &mut **db)
         .await
@@ -165,6 +165,7 @@ pub async fn statuses(
             favourites_count: 0,
             favourited: false,
             reblogged: false,
+            reblog: None,
             muted: false,
             bookmarked: false,
             media_attachments: vec![],

ferri-server/src/endpoints/custom.rs

@@ -1,6 +1,8 @@
 use main::ap::http::HttpClient;
 use rocket::{State, get, response::status};
 use rocket_db_pools::Connection;
+use main::ap;
+use crate::OutboundQueue;
 
 use uuid::Uuid;
 
@@ -84,7 +86,9 @@ pub async fn resolve_user(acct: &str, host: &str) -> types::Person {
 }
 
 #[get("/test")]
-pub async fn test(http: &State<HttpClient>) -> &'static str {
+pub async fn test(http: &State<HttpClient>, outbound: &State<OutboundQueue>) -> &'static str {
+    outbound.0.send(ap::QueueMessage::Heartbeat);
+    
     let user = resolve_user("amy@fedi.amy.mov", "fedi.amy.mov").await;
 
     let post = activity::CreateActivity {
@@ -99,6 +103,7 @@ pub async fn test(http: &State<HttpClient>) -> &'static str {
             ts: "2025-04-10T10:48:11Z".to_string(),
             to: vec!["https://ferri.amy.mov/users/amy/followers".to_string()],
             cc: vec!["https://www.w3.org/ns/activitystreams#Public".to_string()],
+            attributed_to: None
         },
         ts: "2025-04-10T10:48:11Z".to_string(),
         to: vec!["https://ferri.amy.mov/users/amy/followers".to_string()],
@@ -118,5 +123,7 @@ pub async fn test(http: &State<HttpClient>) -> &'static str {
         .await
         .unwrap();
 
+    dbg!(follow);
+
     "Hello, world!"
 }

ferri-server/src/endpoints/inbox.rs

@@ -1,4 +1,5 @@
 use chrono::Local;
+use tracing::Instrument;
 use main::ap;
 use rocket::serde::json::serde_json;
 use rocket::{State, post};
@@ -6,12 +7,13 @@ use rocket_db_pools::Connection;
 use sqlx::Sqlite;
 use url::Url;
 use uuid::Uuid;
-use tracing::{event, span, Level, debug, warn, info};
+use tracing::{event, span, Level, debug, warn, info, error};
+use crate::http_wrapper::HttpWrapper;
 
 use crate::{
     Db,
     http::HttpClient,
-    types::{Person, activity},
+    types::{Person, content::Post, activity},
 };
 
 fn handle_delete_activity(activity: activity::DeleteActivity) {
@@ -46,7 +48,9 @@ async fn create_user(
     // HACK: Allow us to formulate a `user@host` username by assuming the actor is on the same host as the user
     let url = Url::parse(&actor).unwrap();
     let host = url.host_str().unwrap();
-    let username = format!("{}@{}", user.name, host);
+    info!("creating user '{}'@'{}' ({:#?})", user.preferred_username, host, user);
+    
+    let username = format!("{}@{}", user.preferred_username, host);
 
     let uuid = Uuid::new_v4().to_string();
     sqlx::query!(
@@ -58,7 +62,7 @@ async fn create_user(
         uuid,
         username,
         actor,
-        user.preferred_username
+        user.name
     )
     .execute(conn)
     .await
@@ -84,29 +88,27 @@ async fn create_follow(
     .unwrap();
 }
 
-async fn handle_follow_activity(
+async fn handle_follow_activity<'a>(
     followed_account: &str,
     activity: activity::FollowActivity,
-    http: &HttpClient,
+    http: HttpWrapper<'a>,
     mut db: Connection<Db>,
 ) {
-    let user = http
+    let user = http.get_person(&activity.actor).await;
-        .get(&activity.actor)
+    if let Err(e) = user {
-        .activity()
+        error!("could not load user {}: {}", activity.actor, e.to_string());
-        .send()
+        return
-        .await
+    }
-        .unwrap()
+    
-        .json::<Person>()
+    let user = user.unwrap();
-        .await
-        .unwrap();
 
     create_actor(&user, &activity.actor, &mut **db).await;
     create_user(&user, &activity.actor, &mut **db).await;
     create_follow(&activity, &mut **db).await;
 
     let follower = ap::User::from_actor_id(&activity.actor, &mut **db).await;
-    let followed = ap::User::from_username(&followed_account, &mut **db).await;
+    let followed = ap::User::from_id(&followed_account, &mut **db).await.unwrap();
-    let outbox = ap::Outbox::for_user(followed.clone(), http);
+    let outbox = ap::Outbox::for_user(followed.clone(), http.client());
 
     let activity = ap::Activity {
         id: format!("https://ferri.amy.mov/activities/{}", Uuid::new_v4()),
@@ -142,23 +144,109 @@ async fn handle_like_activity(activity: activity::LikeActivity, mut db: Connecti
     }
 }
 
-async fn handle_create_activity(
+async fn handle_boost_activity<'a>(
+    activity: activity::BoostActivity,
+    http: HttpWrapper<'a>,
+    mut db: Connection<Db>,
+) {
+    let key_id = "https://ferri.amy.mov/users/amy#main-key";
+    dbg!(&activity);
+    let post = http
+        .client()
+        .get(&activity.object)
+        .activity()
+        .sign(&key_id)
+        .send()
+        .await
+        .unwrap()
+        .text()
+        .await
+        .unwrap();
+    
+    info!("{}", post);
+    
+    let post = serde_json::from_str::<Post>(&post);
+    if let Err(e) = post {
+        error!(?e, "when decoding post");
+        return
+    }
+    
+    let post = post.unwrap();
+
+    info!("{:#?}", post);
+    let attribution = post.attributed_to.unwrap();
+    
+    let post_user = http.get_person(&attribution).await;
+    if let Err(e) = post_user {
+        error!("could not load post_user {}: {}", attribution, e.to_string());
+        return
+    }
+    let post_user = post_user.unwrap();
+
+    let user = http.get_person(&activity.actor).await;
+    if let Err(e) = user {
+        error!("could not load actor {}: {}", activity.actor, e.to_string());
+        return
+    }
+    let user = user.unwrap();
+
+    debug!("creating actor {}", activity.actor);
+    create_actor(&user, &activity.actor, &mut **db).await;
+
+    debug!("creating user {}", activity.actor);
+    create_user(&user, &activity.actor, &mut **db).await;
+
+    debug!("creating actor {}", attribution);
+    create_actor(&post_user, &attribution, &mut **db).await;
+
+    debug!("creating user {}", attribution);
+    create_user(&post_user, &attribution, &mut **db).await;
+    
+    let attributed_user = ap::User::from_actor_id(&attribution, &mut **db).await;
+    let actor_user = ap::User::from_actor_id(&activity.actor, &mut **db).await;
+    
+    let base_id = ap::new_id();
+    let now = ap::new_ts();
+    
+    let reblog_id = ap::new_id();
+
+    let attr_id = attributed_user.id();
+    sqlx::query!("
+       INSERT INTO post (id, uri, user_id, content, created_at)
+       VALUES (?1, ?2, ?3, ?4, ?5)
+    ", reblog_id, post.id, attr_id, post.content, post.ts)
+        .execute(&mut **db)
+        .await
+        .unwrap();
+
+    let uri = format!("https://ferri.amy.mov/users/{}/posts/{}", actor_user.id(), post.id);
+    let user_id = actor_user.id();
+    
+    sqlx::query!("
+       INSERT INTO post (id, uri, user_id, content, created_at, boosted_post_id)
+       VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+    ", base_id, uri, user_id, "", now, reblog_id)
+        .execute(&mut **db)
+        .await
+        .unwrap();
+
+}
+
+async fn handle_create_activity<'a>(
     activity: activity::CreateActivity,
-    http: &HttpClient,
+    http: HttpWrapper<'a>,
     mut db: Connection<Db>,
 ) {
     assert!(&activity.object.ty == "Note");
     debug!("resolving user {}", activity.actor);
     
-    let user = http
+    let user = http.get_person(&activity.actor).await;
-        .get(&activity.actor)
+    if let Err(e) = user {
-        .activity()
+        error!("could not load user {}: {}", activity.actor, e.to_string());
-        .send()
+        return
-        .await
+    }
-        .unwrap()
+
-        .json::<Person>()
+    let user = user.unwrap();
-        .await
-        .unwrap();
 
     debug!("creating actor {}", activity.actor);
     create_actor(&user, &activity.actor, &mut **db).await;
@@ -198,31 +286,42 @@ pub async fn inbox(db: Connection<Db>, http: &State<HttpClient>, user: &str, bod
     let min = serde_json::from_str::<activity::MinimalActivity>(&body).unwrap();
     let inbox_span = span!(Level::INFO, "inbox-post", user_id = user);
 
-    let _enter = inbox_span.enter();
+    async move {
-    event!(Level::INFO, ?min, "received an activity");
+        event!(Level::INFO, ?min, "received an activity");
-    
+        
-    match min.ty.as_str() {
+        let key_id = "https://ferri.amy.mov/users/amy#main-key";
-        "Delete" => {
+        let wrapper = HttpWrapper::new(http.inner(), key_id);
-            let activity = serde_json::from_str::<activity::DeleteActivity>(&body).unwrap();
+        
-            handle_delete_activity(activity);
+        match min.ty.as_str() {
+            "Delete" => {
+                let activity = serde_json::from_str::<activity::DeleteActivity>(&body).unwrap();
+                handle_delete_activity(activity);
+            }
+            "Follow" => {
+                let activity = serde_json::from_str::<activity::FollowActivity>(&body).unwrap();
+                handle_follow_activity(user, activity, wrapper, db).await;
+            }
+            "Create" => {
+                let activity = serde_json::from_str::<activity::CreateActivity>(&body).unwrap();
+                handle_create_activity(activity, wrapper, db).await;
+            }
+            "Like" => {
+                let activity = serde_json::from_str::<activity::LikeActivity>(&body).unwrap();
+                handle_like_activity(activity, db).await;
+            }
+            "Announce" => {
+                let activity = serde_json::from_str::<activity::BoostActivity>(&body).unwrap();
+                handle_boost_activity(activity, wrapper, db).await;
+            }
+            
+            act => {
+                warn!(act, body, "unknown activity");
+            }
         }
-        "Follow" => {
-            let activity = serde_json::from_str::<activity::FollowActivity>(&body).unwrap();
-            handle_follow_activity(user, activity, http.inner(), db).await;
-        }
-        "Create" => {
-            let activity = serde_json::from_str::<activity::CreateActivity>(&body).unwrap();
-            handle_create_activity(activity, http.inner(), db).await;
-        }
-        "Like" => {
-            let activity = serde_json::from_str::<activity::LikeActivity>(&body).unwrap();
-            handle_like_activity(activity, db).await;
-        }
-        act => {
-            warn!(act, body, "unknown activity");
-        }
-    }
 
-    debug!("body in inbox: {}", body);
+        debug!("body in inbox: {}", body);
-    drop(_enter)
+    }
+    // Allow the span to be used inside the async code
+    // https://docs.rs/tracing/latest/tracing/span/struct.EnteredSpan.html#deref-methods-Span
+    .instrument(inbox_span).await;
 }

ferri-server/src/endpoints/oauth.rs

@@ -8,12 +8,12 @@ use rocket::{
 };
 use rocket_db_pools::Connection;
 
-#[get("/oauth/authorize?<client_id>&<scope>&<redirect_uri>&<response_type>")]
+#[get("/oauth/authorize?<client_id>&<scope>&<redirect_uri>&<_response_type>")]
 pub async fn authorize(
     client_id: &str,
     scope: &str,
     redirect_uri: &str,
-    response_type: &str,
+    _response_type: &str,
     mut db: Connection<Db>,
 ) -> Redirect {
     // For now, we will always authorize the request and assign it to an admin user
@@ -68,11 +68,11 @@ pub struct Token {
 #[derive(Deserialize, Debug, FromForm)]
 #[serde(crate = "rocket::serde")]
 pub struct NewTokenRequest {
-    client_id: String,
+    // pub client_id: String,
-    redirect_uri: String,
+    // pub redirect_uri: String,
-    grant_type: String,
+    // pub grant_type: String,
-    code: String,
+    pub code: String,
-    client_secret: String,
+    // pub client_secret: String,
 }
 
 #[post("/oauth/token", data = "<req>")]

ferri-server/src/endpoints/user.rs

@@ -1,17 +1,19 @@
 use main::ap;
-use rocket::{get, http::ContentType, serde::json::Json};
+use rocket::{get, http::ContentType, serde::json::Json, State, Responder};
 use rocket_db_pools::Connection;
+use rocket::response::Redirect;
 use rocket::response::status::NotFound;
 
 use crate::{
+    Config,
     Db,
     types::{OrderedCollection, Person, UserKey, content},
 };
 
 use super::activity_type;
 
-#[get("/users/<user>/inbox")]
+#[get("/users/<_user>/inbox")]
-pub async fn inbox(user: String) -> Json<OrderedCollection> {
+pub async fn inbox(_user: String) -> Json<OrderedCollection> {
     Json(OrderedCollection {
         ty: "OrderedCollection".to_string(),
         total_items: 0,
@@ -19,8 +21,8 @@ pub async fn inbox(user: String) -> Json<OrderedCollection> {
     })
 }
 
-#[get("/users/<user>/outbox")]
+#[get("/users/<_user>/outbox")]
-pub async fn outbox(user: String) -> Json<OrderedCollection> {
+pub async fn outbox(_user: String) -> Json<OrderedCollection> {
     Json(OrderedCollection {
         ty: "OrderedCollection".to_string(),
         total_items: 0,
@@ -89,6 +91,7 @@ pub async fn following(mut db: Connection<Db>, uuid: &str) -> Result<Json<Ordere
 #[get("/users/<uuid>/posts/<post>")]
 pub async fn post(
     mut db: Connection<Db>,
+    config: &State<Config>,
     uuid: &str,
     post: String,
 ) -> (ContentType, Json<content::Post>) {
@@ -106,40 +109,67 @@ pub async fn post(
         activity_type(),
         Json(content::Post {
             context: "https://www.w3.org/ns/activitystreams".to_string(),
-            id: format!("https://ferri.amy.mov/users/{}/posts/{}", uuid, post.id),
+            id: config.post_url(uuid, &post.id),
+            attributed_to: Some(config.user_url(uuid)),
             ty: "Note".to_string(),
             content: post.content,
             ts: post.created_at,
-            to: vec!["https://ferri.amy.mov/users/amy/followers".to_string()],
+            to: vec![config.followers_url(uuid)],
             cc: vec!["https://www.w3.org/ns/activitystreams#Public".to_string()],
         }),
     )
 }
 
+#[derive(Debug, Responder)]
+pub enum UserFetchError {
+    NotFound(NotFound<String>),
+    Moved(Redirect),
+}
+
+type ActivityResponse<T> = (ContentType, T);
+fn ap_response<T>(t: T) -> ActivityResponse<T> {
+    (activity_type(), t)
+}
+
+fn ap_ok<T, E>(t: T) -> Result<ActivityResponse<T>, E> {
+    Ok(ap_response(t))
+}
+
 #[get("/users/<uuid>")]
-pub async fn user(mut db: Connection<Db>, uuid: &str) -> Result<(ContentType, Json<Person>), NotFound<String>> {
+pub async fn user(
+    mut db: Connection<Db>,
+    config: &State<Config>,
+    uuid: &str
+) -> Result<ActivityResponse<Json<Person>>, UserFetchError> {
+    if uuid == "amy" {
+        return Err(
+            UserFetchError::Moved(
+                Redirect::permanent("https://ferri.amy.mov/users/9b9d497b-2731-435f-a929-e609ca69dac9")
+            )
+        )
+    }
+    
     let user = ap::User::from_id(uuid, &mut **db)
         .await
-        .map_err(|e| NotFound(e.to_string()))?;
+        .map_err(|e| UserFetchError::NotFound(NotFound(e.to_string())))?;
 
-    Ok((
+    let person = Person {
-        activity_type(),
+        context: "https://www.w3.org/ns/activitystreams".to_string(),
-        Json(Person {
+        ty: "Person".to_string(),
-            context: "https://www.w3.org/ns/activitystreams".to_string(),
+        id: config.user_url(user.id()),
-            ty: "Person".to_string(),
+        name: user.username().to_string(),
-            id: format!("https://ferri.amy.mov/users/{}", user.id()),
+        preferred_username: user.display_name().to_string(),
-            name: user.username().to_string(),
+        followers: config.followers_url(user.id()),
-            preferred_username: user.display_name().to_string(),
+        following: config.following_url(user.id()),
-            followers: format!("https://ferri.amy.mov/users/{}/followers", uuid),
+        summary: format!("ferri {}", user.username()),
-            following: format!("https://ferri.amy.mov/users/{}/following", uuid),
+        inbox: config.inbox_url(user.id()),
-            summary: format!("ferri {}", user.username()),
+        outbox: config.outbox_url(user.id()),
-            inbox: format!("https://ferri.amy.mov/users/{}/inbox", uuid),
+        public_key: Some(UserKey {
-            outbox: format!("https://ferri.amy.mov/users/{}/outbox", uuid),
+            id: format!("https://ferri.amy.mov/users/{}#main-key", uuid),
-            public_key: Some(UserKey {
+            owner: config.user_url(user.id()),
-                id: format!("https://ferri.amy.mov/users/{}#main-key", uuid),
+            public_key: include_str!("../../../public.pem").to_string(),
-                owner: format!("https://ferri.amy.mov/users/{}", uuid),
-                public_key: include_str!("../../../public.pem").to_string(),
-            }),
         }),
-    ))
+    };
+    
+    ap_ok(Json(person))
 }

ferri-server/src/endpoints/well_known.rs

@@ -1,9 +1,10 @@
 use main::ap;
-use rocket::{get, serde::json::Json};
+use rocket::{get, serde::json::Json, State};
 use rocket_db_pools::Connection;
 use tracing::info;
 
 use crate::{
+    Config,
     Db,
     types::webfinger::{Link, WebfingerResponse},
 };
@@ -20,7 +21,7 @@ pub async fn host_meta() -> &'static str {
 
 // https://mastodon.social/.well-known/webfinger?resource=acct:gargron@mastodon.social
 #[get("/.well-known/webfinger?<resource>")]
-pub async fn webfinger(mut db: Connection<Db>, resource: &str) -> Json<WebfingerResponse> {
+pub async fn webfinger(mut db: Connection<Db>, config: &State<Config>, resource: &str) -> Json<WebfingerResponse> {
     info!(?resource, "incoming webfinger request");
 
     let acct = resource.strip_prefix("acct:").unwrap();
@@ -30,19 +31,19 @@ pub async fn webfinger(mut db: Connection<Db>, resource: &str) -> Json<Webfinger
     Json(WebfingerResponse {
         subject: resource.to_string(),
         aliases: vec![
-            format!("https://ferri.amy.mov/users/{}", user.id()),
+            config.user_url(user.id()),
-            format!("https://ferri.amy.mov/{}", user.username()),
+            config.user_web_url(user.username())
         ],
         links: vec![
             Link {
                 rel: "http://webfinger.net/rel/profile-page".to_string(),
                 ty: Some("text/html".to_string()),
-                href: Some(format!("https://ferri.amy.mov/{}", user.username())),
+                href: Some(config.user_web_url(user.username())),
             },
             Link {
                 rel: "self".to_string(),
                 ty: Some("application/activity+json".to_string()),
-                href: Some(format!("https://ferri.amy.mov/users/{}", user.id())),
+                href: Some(config.user_url(user.id())),
             },
         ],
     })

ferri-server/src/http_wrapper.rs

@@ -0,0 +1,71 @@
+use thiserror::Error;
+use tracing::{error, event, Level};
+use crate::http::HttpClient;
+use crate::types::Person;
+use std::fmt::Debug;
+
+pub struct HttpWrapper<'a> {
+    client: &'a HttpClient,
+    key_id: &'a str
+}
+
+#[derive(Error, Debug)]
+pub enum HttpError {
+    #[error("entity of type `{0}` @ URL `{1}` could not be loaded")]
+    LoadFailure(String, String),
+    #[error("entity of type `{0}` @ URL `{1}` could not be parsed ({2})")]
+    ParseFailure(String, String, String),
+}
+
+impl <'a> HttpWrapper<'a> {
+    pub fn new(client: &'a HttpClient, key_id: &'a str) -> HttpWrapper<'a> {
+        Self {
+            client,
+            key_id
+        }
+    }
+
+    pub fn client(&self) -> &'a HttpClient {
+        &self.client
+    }
+
+    async fn get<T : serde::de::DeserializeOwned + Debug>(&self, ty: &str, url: &str) -> Result<T, HttpError> {
+        let ty = ty.to_string();
+        event!(Level::INFO, url, "loading {}", ty);
+        
+        let http_result = self.client
+            .get(url)
+            .sign(self.key_id)
+            .activity()
+            .send()
+            .await;
+
+        if let Err(e) = http_result {
+            error!("could not load url {}: {:#?}", url, e);
+            return Err(HttpError::LoadFailure(ty, url.to_string()));
+        }
+
+        let raw_body = http_result.unwrap().text().await;
+        if let Err(e) = raw_body {
+            error!("could not get text for url {}: {:#?}", url, e);
+            return Err(HttpError::LoadFailure(ty, url.to_string()));
+        }
+
+        let decoded = serde_json::from_str::<T>(&raw_body.unwrap());
+        
+        if let Err(e) = decoded {
+            error!("could not parse {} for url {}: {:#?}", ty, url, e);
+            return Err(HttpError::ParseFailure(
+                ty,
+                url.to_string(),
+                e.to_string()
+            ));
+        }
+
+        Ok(decoded.unwrap())
+    }
+
+    pub async fn get_person(&self, url: &str) -> Result<Person, HttpError> {
+        self.get("Person", url).await
+    }
+}

ferri-server/src/lib.rs

@@ -3,9 +3,10 @@ use endpoints::{
     custom, inbox, oauth, user, well_known,
 };
 
-use tracing::Level;
 use tracing_subscriber::fmt;
 
+use main::ap;
+
 use main::ap::http;
 use main::config::Config;
 use rocket::{
@@ -19,23 +20,24 @@ use rocket_db_pools::{Connection, Database, sqlx};
 mod cors;
 mod endpoints;
 mod types;
+mod http_wrapper;
 
 #[derive(Database)]
 #[database("sqlite_ferri")]
 pub struct Db(sqlx::SqlitePool);
 
 #[get("/")]
-async fn user_profile(cfg: &rocket::State<Config>) -> (ContentType, &'static str) {
+async fn user_profile() -> (ContentType, &'static str) {
     (ContentType::HTML, "<p>hello</p>")
 }
 
-#[get("/activities/<activity>")]
+#[get("/activities/<_activity>")]
-async fn activity_endpoint(activity: String) {
+async fn activity_endpoint(_activity: String) {
 
 }
 
 #[derive(Debug)]
-struct AuthenticatedUser {
+pub struct AuthenticatedUser {
     pub username: String,
     pub id: String,
     pub token: String,
@@ -43,10 +45,7 @@ struct AuthenticatedUser {
 }
 
 #[derive(Debug)]
-enum LoginError {
+pub enum LoginError {
-    InvalidData,
-    UsernameDoesNotExist,
-    WrongPassword,
 }
 
 #[rocket::async_trait]
@@ -88,7 +87,8 @@ impl<'a> FromRequest<'a> for AuthenticatedUser {
     }
 }
 
-
+pub struct OutboundQueue(pub ap::QueueHandle);
+pub struct InboundQueue(pub ap::QueueHandle);
 
 pub fn launch(cfg: Config) -> Rocket<Build> {
     let format = fmt::format()
@@ -104,11 +104,19 @@ pub fn launch(cfg: Config) -> Rocket<Build> {
         .event_format(format)
         .with_writer(std::io::stdout)
         .init();
-    
+
+    let outbound = ap::RequestQueue::new("outbound");
+    let outbound_handle = outbound.spawn();
+
+    let inbound = ap::RequestQueue::new("inbound");
+    let inbound_handle = inbound.spawn();
+
     let http_client = http::HttpClient::new();
     build()
         .manage(cfg)
         .manage(http_client)
+        .manage(OutboundQueue(outbound_handle))
+        .manage(InboundQueue(inbound_handle))
         .attach(Db::init())
         .attach(cors::CORS)
         .mount("/assets", rocket::fs::FileServer::from("./assets"))

ferri-server/src/types/activity.rs

@@ -1,5 +1,4 @@
 use rocket::serde::{Deserialize, Serialize};
-
 use crate::types::content::Post;
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -60,3 +59,17 @@ pub struct AcceptActivity {
     pub object: String,
     pub actor: String,
 }
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(crate = "rocket::serde")]
+pub struct BoostActivity {
+    #[serde(rename = "type")]
+    pub ty: String,
+
+    pub id: String,
+    pub actor: String,
+    pub published: String,
+    pub to: Vec<String>,
+    pub cc: Vec<String>,
+    pub object: String,
+}

ferri-server/src/types/content.rs

@@ -15,4 +15,7 @@ pub struct Post {
     pub content: String,
     pub to: Vec<String>,
     pub cc: Vec<String>,
+    
+    #[serde(rename = "attributedTo")]
+    pub attributed_to: Option<String>
 }

ferri-server/src/types/mod.rs

@@ -34,6 +34,7 @@ pub struct Person {
     pub inbox: String,
     pub outbox: String,
     pub preferred_username: String,
+    #[serde(default)]
     pub name: String,
     pub public_key: Option<UserKey>,
 }

flake.nix

@@ -29,6 +29,7 @@
           
           packages = with pkgs; [
             sqlx-cli
+            cargo-nextest
             (rust-bin.stable.latest.default.override {
               extensions = [ "rust-src" "rust-analyzer" ];
               targets = [  ];

migrations/20250410112325_add_follow.sql

@@ -4,6 +4,7 @@ CREATE TABLE IF NOT EXISTS follow
 	id TEXT PRIMARY KEY NOT NULL,
 	follower_id TEXT NOT NULL,
 	followed_id TEXT NOT NULL,
+	
 	FOREIGN KEY(follower_id) REFERENCES actor(id),
 	FOREIGN KEY(followed_id) REFERENCES actor(id)
 );

migrations/20250410182845_add_post.sql

@@ -5,6 +5,8 @@ CREATE TABLE IF NOT EXISTS post
 	user_id TEXT NOT NULL,
 	content TEXT NOT NULL,
 	created_at TEXT NOT NULL,
+	boosted_post_id TEXT,
 
-	FOREIGN KEY(user_id) REFERENCES user(id)
+	FOREIGN KEY(user_id) REFERENCES user(id),
+	FOREIGN KEY(boosted_post_id) REFERENCES post(id)
 );