fix: align follow event with accepted state; redact internal error details

crates/adapters/auth/Cargo.toml

@@ -10,5 +10,7 @@ thiserror    = { workspace = true }
 uuid         = { workspace = true }
 chrono       = { workspace = true }
 tokio        = { workspace = true }
+serde        = { workspace = true }
 jsonwebtoken = "9"
 argon2       = "0.5"
+rand          = "0.8"

crates/adapters/auth/src/lib.rs

@@ -0,0 +1,116 @@
+use async_trait::async_trait;
+use chrono::{Duration, Utc};
+use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
+use serde::{Deserialize, Serialize};
+use domain::{
+    errors::DomainError,
+    ports::{AuthService, GeneratedToken, PasswordHasher},
+    value_objects::{PasswordHash, UserId},
+};
+
+#[derive(Serialize, Deserialize)]
+struct Claims {
+    sub: String,
+    exp: usize,
+}
+
+pub struct JwtAuthService {
+    secret: String,
+    ttl_seconds: i64,
+}
+
+impl JwtAuthService {
+    pub fn new(secret: String, ttl_seconds: i64) -> Self {
+        Self { secret, ttl_seconds }
+    }
+}
+
+impl AuthService for JwtAuthService {
+    fn generate_token(&self, user_id: &UserId) -> Result<GeneratedToken, DomainError> {
+        let exp = (Utc::now() + Duration::seconds(self.ttl_seconds)).timestamp() as usize;
+        let claims = Claims {
+            sub: user_id.as_uuid().to_string(),
+            exp,
+        };
+        let token = encode(
+            &Header::default(),
+            &claims,
+            &EncodingKey::from_secret(self.secret.as_bytes()),
+        )
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(GeneratedToken {
+            token,
+            user_id: user_id.clone(),
+        })
+    }
+
+    fn validate_token(&self, token: &str) -> Result<UserId, DomainError> {
+        let data = decode::<Claims>(
+            token,
+            &DecodingKey::from_secret(self.secret.as_bytes()),
+            &Validation::default(),
+        )
+        .map_err(|_| DomainError::Unauthorized)?;
+        let uuid = uuid::Uuid::parse_str(&data.claims.sub)
+            .map_err(|_| DomainError::Unauthorized)?;
+        Ok(UserId::from_uuid(uuid))
+    }
+}
+
+pub struct Argon2PasswordHasher;
+
+#[async_trait]
+impl PasswordHasher for Argon2PasswordHasher {
+    async fn hash(&self, plain: &str) -> Result<PasswordHash, DomainError> {
+        use argon2::{
+            password_hash::SaltString,
+            Argon2, PasswordHasher as _,
+        };
+        use rand::rngs::OsRng;
+        let salt = SaltString::generate(OsRng);
+        let hash = Argon2::default()
+            .hash_password(plain.as_bytes(), &salt)
+            .map_err(|e| DomainError::Internal(e.to_string()))?
+            .to_string();
+        Ok(PasswordHash(hash))
+    }
+
+    async fn verify(&self, plain: &str, hash: &PasswordHash) -> Result<bool, DomainError> {
+        use argon2::{password_hash::PasswordHash as ArgonHash, Argon2, PasswordVerifier};
+        let parsed = ArgonHash::new(&hash.0)
+            .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(Argon2::default()
+            .verify_password(plain.as_bytes(), &parsed)
+            .is_ok())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::ports::AuthService;
+
+    #[test]
+    fn generate_and_validate_token() {
+        let svc = JwtAuthService::new("secret".into(), 3600);
+        let id = UserId::new();
+        let tok = svc.generate_token(&id).unwrap();
+        let parsed = svc.validate_token(&tok.token).unwrap();
+        assert_eq!(parsed.as_uuid(), id.as_uuid());
+    }
+
+    #[test]
+    fn invalid_token_returns_unauthorized() {
+        let svc = JwtAuthService::new("secret".into(), 3600);
+        let err = svc.validate_token("not.a.token").unwrap_err();
+        assert!(matches!(err, DomainError::Unauthorized));
+    }
+
+    #[tokio::test]
+    async fn hash_and_verify() {
+        let hasher = Argon2PasswordHasher;
+        let hash = hasher.hash("mypassword").await.unwrap();
+        assert!(hasher.verify("mypassword", &hash).await.unwrap());
+        assert!(!hasher.verify("wrongpassword", &hash).await.unwrap());
+    }
+}

crates/adapters/postgres/src/api_key.rs

@@ -1,2 +1,73 @@
-pub struct PgApiKeyRepository { _pool: sqlx::PgPool }
-impl PgApiKeyRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::api_key::ApiKey, ports::ApiKeyRepository, value_objects::{ApiKeyId, UserId}};
+
+pub struct PgApiKeyRepository { pool: PgPool }
+impl PgApiKeyRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl ApiKeyRepository for PgApiKeyRepository {
+    async fn save(&self, k: &ApiKey) -> Result<(), DomainError> {
+        sqlx::query("INSERT INTO api_keys(id,user_id,key_hash,name,created_at) VALUES($1,$2,$3,$4,$5)")
+            .bind(k.id.as_uuid()).bind(k.user_id.as_uuid()).bind(&k.key_hash).bind(&k.name).bind(k.created_at)
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn find_by_hash(&self, hash: &str) -> Result<Option<ApiKey>, DomainError> {
+        #[derive(sqlx::FromRow)] struct Row { id: uuid::Uuid, user_id: uuid::Uuid, key_hash: String, name: String, created_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>("SELECT id,user_id,key_hash,name,created_at FROM api_keys WHERE key_hash=$1")
+            .bind(hash).fetch_optional(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|o| o.map(|r| ApiKey { id: ApiKeyId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), key_hash: r.key_hash, name: r.name, created_at: r.created_at }))
+    }
+
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<ApiKey>, DomainError> {
+        #[derive(sqlx::FromRow)] struct Row { id: uuid::Uuid, user_id: uuid::Uuid, key_hash: String, name: String, created_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>("SELECT id,user_id,key_hash,name,created_at FROM api_keys WHERE user_id=$1 ORDER BY created_at DESC")
+            .bind(user_id.as_uuid()).fetch_all(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|rows| rows.into_iter().map(|r| ApiKey { id: ApiKeyId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), key_hash: r.key_hash, name: r.name, created_at: r.created_at }).collect())
+    }
+
+    async fn delete(&self, id: &ApiKeyId, user_id: &UserId) -> Result<(), DomainError> {
+        sqlx::query("DELETE FROM api_keys WHERE id=$1 AND user_id=$2")
+            .bind(id.as_uuid()).bind(user_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::user::User, value_objects::*};
+    use crate::user::PgUserRepository;
+    use domain::ports::UserRepository;
+
+    async fn seed_user(pool: &sqlx::PgPool) -> User {
+        let repo = PgUserRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()));
+        repo.save(&u).await.unwrap(); u
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn save_and_find_by_hash(pool: sqlx::PgPool) {
+        let user = seed_user(&pool).await;
+        let repo = PgApiKeyRepository::new(pool);
+        let key = ApiKey { id: ApiKeyId::new(), user_id: user.id.clone(), key_hash: "abc123".into(), name: "test".into(), created_at: Utc::now() };
+        repo.save(&key).await.unwrap();
+        let found = repo.find_by_hash("abc123").await.unwrap().unwrap();
+        assert_eq!(found.name, "test");
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn delete_key(pool: sqlx::PgPool) {
+        let user = seed_user(&pool).await;
+        let repo = PgApiKeyRepository::new(pool);
+        let key = ApiKey { id: ApiKeyId::new(), user_id: user.id.clone(), key_hash: "def456".into(), name: "key2".into(), created_at: Utc::now() };
+        repo.save(&key).await.unwrap();
+        repo.delete(&key.id, &user.id).await.unwrap();
+        assert!(repo.find_by_hash("def456").await.unwrap().is_none());
+    }
+}

crates/adapters/postgres/src/block.rs

@@ -1,2 +1,81 @@
-pub struct PgBlockRepository { _pool: sqlx::PgPool }
-impl PgBlockRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::social::Block, ports::BlockRepository, value_objects::UserId};
+
+pub struct PgBlockRepository { pool: PgPool }
+impl PgBlockRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl BlockRepository for PgBlockRepository {
+    async fn save(&self, b: &Block) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO blocks(blocker_id,blocked_id,created_at) VALUES($1,$2,$3) ON CONFLICT DO NOTHING"
+        )
+        .bind(b.blocker_id.as_uuid())
+        .bind(b.blocked_id.as_uuid())
+        .bind(b.created_at)
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))
+        .map(|_| ())
+    }
+
+    async fn delete(&self, blocker_id: &UserId, blocked_id: &UserId) -> Result<(), DomainError> {
+        sqlx::query("DELETE FROM blocks WHERE blocker_id=$1 AND blocked_id=$2")
+            .bind(blocker_id.as_uuid())
+            .bind(blocked_id.as_uuid())
+            .execute(&self.pool)
+            .await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|_| ())
+    }
+
+    async fn exists(&self, blocker_id: &UserId, blocked_id: &UserId) -> Result<bool, DomainError> {
+        let count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM blocks WHERE blocker_id=$1 AND blocked_id=$2"
+        )
+        .bind(blocker_id.as_uuid())
+        .bind(blocked_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(count > 0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::user::User, value_objects::*};
+    use crate::user::PgUserRepository;
+    use domain::ports::UserRepository;
+
+    async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
+        let repo = PgUserRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new(username).unwrap(), Email::new(email).unwrap(), PasswordHash("h".into()));
+        repo.save(&u).await.unwrap(); u
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn block_exists(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgBlockRepository::new(pool);
+        let block = Block { blocker_id: alice.id.clone(), blocked_id: bob.id.clone(), created_at: Utc::now() };
+        repo.save(&block).await.unwrap();
+        assert!(repo.exists(&alice.id, &bob.id).await.unwrap());
+        assert!(!repo.exists(&bob.id, &alice.id).await.unwrap());
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn unblock(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgBlockRepository::new(pool);
+        let block = Block { blocker_id: alice.id.clone(), blocked_id: bob.id.clone(), created_at: Utc::now() };
+        repo.save(&block).await.unwrap();
+        repo.delete(&alice.id, &bob.id).await.unwrap();
+        assert!(!repo.exists(&alice.id, &bob.id).await.unwrap());
+    }
+}

crates/adapters/postgres/src/boost.rs

@@ -1,2 +1,80 @@
-pub struct PgBoostRepository { _pool: sqlx::PgPool }
-impl PgBoostRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::social::Boost, ports::BoostRepository, value_objects::{BoostId, ThoughtId, UserId}};
+
+pub struct PgBoostRepository { pool: PgPool }
+impl PgBoostRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl BoostRepository for PgBoostRepository {
+    async fn save(&self, b: &Boost) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO boosts(id,user_id,thought_id,ap_id,created_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT(user_id,thought_id) DO NOTHING"
+        )
+        .bind(b.id.as_uuid()).bind(b.user_id.as_uuid()).bind(b.thought_id.as_uuid()).bind(&b.ap_id).bind(b.created_at)
+        .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn delete(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+        let r = sqlx::query("DELETE FROM boosts WHERE user_id=$1 AND thought_id=$2")
+            .bind(user_id.as_uuid()).bind(thought_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        if r.rows_affected() == 0 { return Err(DomainError::NotFound); }
+        Ok(())
+    }
+
+    async fn find(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<Option<Boost>, DomainError> {
+        #[derive(sqlx::FromRow)]
+        struct Row { id: uuid::Uuid, user_id: uuid::Uuid, thought_id: uuid::Uuid, ap_id: Option<String>, created_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>("SELECT id,user_id,thought_id,ap_id,created_at FROM boosts WHERE user_id=$1 AND thought_id=$2")
+            .bind(user_id.as_uuid()).bind(thought_id.as_uuid())
+            .fetch_optional(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|o| o.map(|r| Boost { id: BoostId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), thought_id: ThoughtId::from_uuid(r.thought_id), ap_id: r.ap_id, created_at: r.created_at }))
+    }
+
+    async fn count_for_thought(&self, thought_id: &ThoughtId) -> Result<i64, DomainError> {
+        sqlx::query_scalar("SELECT COUNT(*) FROM boosts WHERE thought_id=$1")
+            .bind(thought_id.as_uuid()).fetch_one(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::{thought::{Thought, Visibility}, user::User}, value_objects::*};
+    use crate::{thought::PgThoughtRepository, user::PgUserRepository};
+    use domain::ports::{ThoughtRepository, UserRepository};
+
+    async fn seed(pool: &sqlx::PgPool) -> (User, Thought) {
+        let urepo = PgUserRepository::new(pool.clone());
+        let trepo = PgThoughtRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()));
+        urepo.save(&u).await.unwrap();
+        let t = Thought::new_local(ThoughtId::new(), u.id.clone(), Content::new_local("hi").unwrap(), None, Visibility::Public, None, false);
+        trepo.save(&t).await.unwrap();
+        (u, t)
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn boost_and_count(pool: sqlx::PgPool) {
+        let (user, thought) = seed(&pool).await;
+        let repo = PgBoostRepository::new(pool);
+        let boost = Boost { id: BoostId::new(), user_id: user.id.clone(), thought_id: thought.id.clone(), ap_id: None, created_at: Utc::now() };
+        repo.save(&boost).await.unwrap();
+        assert_eq!(repo.count_for_thought(&thought.id).await.unwrap(), 1);
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn unboost(pool: sqlx::PgPool) {
+        let (user, thought) = seed(&pool).await;
+        let repo = PgBoostRepository::new(pool);
+        let boost = Boost { id: BoostId::new(), user_id: user.id.clone(), thought_id: thought.id.clone(), ap_id: None, created_at: Utc::now() };
+        repo.save(&boost).await.unwrap();
+        repo.delete(&user.id, &thought.id).await.unwrap();
+        assert_eq!(repo.count_for_thought(&thought.id).await.unwrap(), 0);
+    }
+}

crates/adapters/postgres/src/feed.rs

@@ -1,2 +1,173 @@
-pub struct PgFeedRepository { _pool: sqlx::PgPool }
-impl PgFeedRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{
+    errors::DomainError,
+    models::{feed::{FeedEntry, PageParams, Paginated}, thought::Thought, user::User},
+    ports::FeedRepository,
+    value_objects::{Content, Email, PasswordHash, ThoughtId, UserId, Username},
+};
+use domain::models::thought::Visibility;
+
+pub struct PgFeedRepository { pool: PgPool }
+impl PgFeedRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[derive(sqlx::FromRow)]
+struct FeedRow {
+    thought_id: uuid::Uuid,
+    t_user_id: uuid::Uuid,
+    content: String,
+    in_reply_to_id: Option<uuid::Uuid>,
+    in_reply_to_url: Option<String>,
+    t_ap_id: Option<String>,
+    visibility: String,
+    content_warning: Option<String>,
+    sensitive: bool,
+    t_local: bool,
+    thought_created_at: DateTime<Utc>,
+    updated_at: Option<DateTime<Utc>>,
+    author_id: uuid::Uuid,
+    username: String,
+    email: String,
+    password_hash: String,
+    display_name: Option<String>,
+    bio: Option<String>,
+    avatar_url: Option<String>,
+    header_url: Option<String>,
+    custom_css: Option<String>,
+    author_local: bool,
+    u_ap_id: Option<String>,
+    inbox_url: Option<String>,
+    public_key: Option<String>,
+    private_key: Option<String>,
+    author_created_at: DateTime<Utc>,
+    author_updated_at: DateTime<Utc>,
+    like_count: i64,
+    boost_count: i64,
+    reply_count: i64,
+}
+
+const FEED_SELECT: &str = "
+    SELECT
+        t.id AS thought_id, t.user_id AS t_user_id, t.content,
+        t.in_reply_to_id, t.in_reply_to_url, t.ap_id AS t_ap_id,
+        t.visibility, t.content_warning, t.sensitive, t.local AS t_local,
+        t.created_at AS thought_created_at, t.updated_at,
+        u.id AS author_id, u.username, u.email, u.password_hash,
+        u.display_name, u.bio, u.avatar_url, u.header_url, u.custom_css,
+        u.local AS author_local, u.ap_id AS u_ap_id, u.inbox_url,
+        u.public_key, u.private_key,
+        u.created_at AS author_created_at, u.updated_at AS author_updated_at,
+        (SELECT COUNT(*) FROM likes l WHERE l.thought_id=t.id) AS like_count,
+        (SELECT COUNT(*) FROM boosts b WHERE b.thought_id=t.id) AS boost_count,
+        (SELECT COUNT(*) FROM thoughts r WHERE r.in_reply_to_id=t.id) AS reply_count
+    FROM thoughts t JOIN users u ON u.id=t.user_id";
+
+fn row_to_entry(r: FeedRow) -> FeedEntry {
+    let thought = Thought {
+        id: ThoughtId::from_uuid(r.thought_id),
+        user_id: UserId::from_uuid(r.t_user_id),
+        content: Content::new_remote(r.content),
+        in_reply_to_id: r.in_reply_to_id.map(ThoughtId::from_uuid),
+        in_reply_to_url: r.in_reply_to_url,
+        ap_id: r.t_ap_id,
+        visibility: Visibility::from_str(&r.visibility),
+        content_warning: r.content_warning,
+        sensitive: r.sensitive,
+        local: r.t_local,
+        created_at: r.thought_created_at,
+        updated_at: r.updated_at,
+    };
+    let author = User {
+        id: UserId::from_uuid(r.author_id),
+        username: Username::from_trusted(r.username),
+        email: Email::from_trusted(r.email),
+        password_hash: PasswordHash(r.password_hash),
+        display_name: r.display_name, bio: r.bio,
+        avatar_url: r.avatar_url, header_url: r.header_url, custom_css: r.custom_css,
+        local: r.author_local, ap_id: r.u_ap_id, inbox_url: r.inbox_url,
+        public_key: r.public_key, private_key: r.private_key,
+        created_at: r.author_created_at, updated_at: r.author_updated_at,
+    };
+    FeedEntry { thought, author, like_count: r.like_count, boost_count: r.boost_count, reply_count: r.reply_count, liked_by_viewer: false, boosted_by_viewer: false }
+}
+
+#[async_trait]
+impl FeedRepository for PgFeedRepository {
+    async fn home_feed(&self, following_ids: &[UserId], page: &PageParams, _viewer_id: Option<&UserId>) -> Result<Paginated<FeedEntry>, DomainError> {
+        let ids: Vec<uuid::Uuid> = following_ids.iter().map(|id| id.as_uuid()).collect();
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM thoughts t WHERE t.user_id=ANY($1) AND t.visibility='public'"
+        ).bind(&ids).fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let sql = format!("{FEED_SELECT} WHERE t.user_id=ANY($1) AND t.visibility='public' ORDER BY t.created_at DESC LIMIT $2 OFFSET $3");
+        let rows = sqlx::query_as::<_, FeedRow>(&sql)
+            .bind(&ids).bind(page.limit()).bind(page.offset())
+            .fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated { items: rows.into_iter().map(row_to_entry).collect(), total, page: page.page, per_page: page.per_page })
+    }
+
+    async fn public_feed(&self, page: &PageParams, _viewer_id: Option<&UserId>) -> Result<Paginated<FeedEntry>, DomainError> {
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM thoughts t WHERE t.local=true AND t.visibility='public'"
+        ).fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let sql = format!("{FEED_SELECT} WHERE t.local=true AND t.visibility='public' ORDER BY t.created_at DESC LIMIT $1 OFFSET $2");
+        let rows = sqlx::query_as::<_, FeedRow>(&sql)
+            .bind(page.limit()).bind(page.offset())
+            .fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated { items: rows.into_iter().map(row_to_entry).collect(), total, page: page.page, per_page: page.per_page })
+    }
+
+    async fn search(&self, query: &str, page: &PageParams, _viewer_id: Option<&UserId>) -> Result<Paginated<FeedEntry>, DomainError> {
+        let pattern = format!("%{}%", query.replace('%', "\\%").replace('_', "\\_"));
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM thoughts t WHERE t.content ILIKE $1 AND t.visibility='public'"
+        ).bind(&pattern).fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let sql = format!("{FEED_SELECT} WHERE t.content ILIKE $1 AND t.visibility='public' ORDER BY t.created_at DESC LIMIT $2 OFFSET $3");
+        let rows = sqlx::query_as::<_, FeedRow>(&sql)
+            .bind(&pattern).bind(page.limit()).bind(page.offset())
+            .fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated { items: rows.into_iter().map(row_to_entry).collect(), total, page: page.page, per_page: page.per_page })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{models::{thought::{Thought, Visibility}, user::User}, ports::{ThoughtRepository, UserRepository}, value_objects::*};
+    use crate::{thought::PgThoughtRepository, user::PgUserRepository};
+
+    async fn seed(pool: &sqlx::PgPool, username: &str, content: &str) -> (User, Thought) {
+        let urepo = PgUserRepository::new(pool.clone());
+        let trepo = PgThoughtRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new(username).unwrap(), Email::new(format!("{username}@ex.com")).unwrap(), PasswordHash("h".into()));
+        urepo.save(&u).await.unwrap();
+        let t = Thought::new_local(ThoughtId::new(), u.id.clone(), Content::new_local(content).unwrap(), None, Visibility::Public, None, false);
+        trepo.save(&t).await.unwrap();
+        (u, t)
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn public_feed_returns_local_thoughts(pool: sqlx::PgPool) {
+        let (_, _) = seed(&pool, "alice", "hello").await;
+        let repo = PgFeedRepository::new(pool);
+        let result = repo.public_feed(&PageParams { page: 1, per_page: 20 }, None).await.unwrap();
+        assert_eq!(result.total, 1);
+        assert_eq!(result.items[0].thought.content.as_str(), "hello");
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn search_returns_matching_thoughts(pool: sqlx::PgPool) {
+        let (_, _) = seed(&pool, "alice", "hello world").await;
+        let (_, _) = seed(&pool, "bob", "goodbye world").await;
+        let repo = PgFeedRepository::new(pool);
+        let result = repo.search("hello", &PageParams { page: 1, per_page: 20 }, None).await.unwrap();
+        assert_eq!(result.total, 1);
+        assert_eq!(result.items[0].thought.content.as_str(), "hello world");
+    }
+}

crates/adapters/postgres/src/follow.rs

@@ -1,2 +1,194 @@
-pub struct PgFollowRepository { _pool: sqlx::PgPool }
-impl PgFollowRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{
+    errors::DomainError,
+    models::{feed::{PageParams, Paginated}, social::{Follow, FollowState}, user::User},
+    ports::FollowRepository,
+    value_objects::UserId,
+};
+
+pub struct PgFollowRepository { pool: PgPool }
+impl PgFollowRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl FollowRepository for PgFollowRepository {
+    async fn save(&self, f: &Follow) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO follows(follower_id,following_id,state,ap_id,created_at)
+             VALUES($1,$2,$3,$4,$5)
+             ON CONFLICT(follower_id,following_id) DO UPDATE SET state=EXCLUDED.state,ap_id=EXCLUDED.ap_id"
+        )
+        .bind(f.follower_id.as_uuid())
+        .bind(f.following_id.as_uuid())
+        .bind(f.state.as_str())
+        .bind(&f.ap_id)
+        .bind(f.created_at)
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))
+        .map(|_| ())
+    }
+
+    async fn delete(&self, follower_id: &UserId, following_id: &UserId) -> Result<(), DomainError> {
+        let r = sqlx::query("DELETE FROM follows WHERE follower_id=$1 AND following_id=$2")
+            .bind(follower_id.as_uuid())
+            .bind(following_id.as_uuid())
+            .execute(&self.pool)
+            .await
+            .map_err(|e| DomainError::Internal(e.to_string()))?;
+        if r.rows_affected() == 0 { return Err(DomainError::NotFound); }
+        Ok(())
+    }
+
+    async fn find(&self, follower_id: &UserId, following_id: &UserId) -> Result<Option<Follow>, DomainError> {
+        #[derive(sqlx::FromRow)]
+        struct Row { follower_id: uuid::Uuid, following_id: uuid::Uuid, state: String, ap_id: Option<String>, created_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>(
+            "SELECT follower_id,following_id,state,ap_id,created_at FROM follows WHERE follower_id=$1 AND following_id=$2"
+        )
+        .bind(follower_id.as_uuid())
+        .bind(following_id.as_uuid())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))
+        .map(|o| o.map(|r| Follow {
+            follower_id: UserId::from_uuid(r.follower_id),
+            following_id: UserId::from_uuid(r.following_id),
+            state: FollowState::from_str(&r.state),
+            ap_id: r.ap_id,
+            created_at: r.created_at,
+        }))
+    }
+
+    async fn update_state(&self, follower_id: &UserId, following_id: &UserId, state: &FollowState) -> Result<(), DomainError> {
+        sqlx::query("UPDATE follows SET state=$3 WHERE follower_id=$1 AND following_id=$2")
+            .bind(follower_id.as_uuid())
+            .bind(following_id.as_uuid())
+            .bind(state.as_str())
+            .execute(&self.pool)
+            .await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|_| ())
+    }
+
+    async fn list_followers(&self, user_id: &UserId, page: &PageParams) -> Result<Paginated<User>, DomainError> {
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM follows WHERE following_id=$1 AND state='accepted'"
+        )
+        .bind(user_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let rows = sqlx::query_as::<_, crate::user::UserRow>(
+            "SELECT u.id,u.username,u.email,u.password_hash,u.display_name,u.bio,u.avatar_url,u.header_url,u.custom_css,u.local,u.ap_id,u.inbox_url,u.public_key,u.private_key,u.created_at,u.updated_at
+             FROM users u JOIN follows f ON f.follower_id=u.id
+             WHERE f.following_id=$1 AND f.state='accepted'
+             ORDER BY f.created_at DESC LIMIT $2 OFFSET $3"
+        )
+        .bind(user_id.as_uuid())
+        .bind(page.limit())
+        .bind(page.offset())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated {
+            items: rows.into_iter().map(User::from).collect(),
+            total,
+            page: page.page,
+            per_page: page.per_page,
+        })
+    }
+
+    async fn list_following(&self, user_id: &UserId, page: &PageParams) -> Result<Paginated<User>, DomainError> {
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM follows WHERE follower_id=$1 AND state='accepted'"
+        )
+        .bind(user_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let rows = sqlx::query_as::<_, crate::user::UserRow>(
+            "SELECT u.id,u.username,u.email,u.password_hash,u.display_name,u.bio,u.avatar_url,u.header_url,u.custom_css,u.local,u.ap_id,u.inbox_url,u.public_key,u.private_key,u.created_at,u.updated_at
+             FROM users u JOIN follows f ON f.following_id=u.id
+             WHERE f.follower_id=$1 AND f.state='accepted'
+             ORDER BY f.created_at DESC LIMIT $2 OFFSET $3"
+        )
+        .bind(user_id.as_uuid())
+        .bind(page.limit())
+        .bind(page.offset())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated {
+            items: rows.into_iter().map(User::from).collect(),
+            total,
+            page: page.page,
+            per_page: page.per_page,
+        })
+    }
+
+    async fn get_accepted_following_ids(&self, user_id: &UserId) -> Result<Vec<UserId>, DomainError> {
+        let ids: Vec<uuid::Uuid> = sqlx::query_scalar(
+            "SELECT following_id FROM follows WHERE follower_id=$1 AND state='accepted'"
+        )
+        .bind(user_id.as_uuid())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(ids.into_iter().map(UserId::from_uuid).collect())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::user::User, value_objects::*};
+    use crate::user::PgUserRepository;
+    use domain::ports::UserRepository;
+
+    async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
+        let repo = PgUserRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new(username).unwrap(), Email::new(email).unwrap(), PasswordHash("h".into()));
+        repo.save(&u).await.unwrap(); u
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn save_and_find_follow(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgFollowRepository::new(pool);
+        let follow = Follow { follower_id: alice.id.clone(), following_id: bob.id.clone(), state: FollowState::Accepted, ap_id: None, created_at: Utc::now() };
+        repo.save(&follow).await.unwrap();
+        let found = repo.find(&alice.id, &bob.id).await.unwrap().unwrap();
+        assert_eq!(found.state, FollowState::Accepted);
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn update_state(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgFollowRepository::new(pool);
+        let follow = Follow { follower_id: alice.id.clone(), following_id: bob.id.clone(), state: FollowState::Pending, ap_id: None, created_at: Utc::now() };
+        repo.save(&follow).await.unwrap();
+        repo.update_state(&alice.id, &bob.id, &FollowState::Accepted).await.unwrap();
+        let found = repo.find(&alice.id, &bob.id).await.unwrap().unwrap();
+        assert_eq!(found.state, FollowState::Accepted);
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn get_accepted_following_ids(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgFollowRepository::new(pool);
+        let follow = Follow { follower_id: alice.id.clone(), following_id: bob.id.clone(), state: FollowState::Accepted, ap_id: None, created_at: Utc::now() };
+        repo.save(&follow).await.unwrap();
+        let ids = repo.get_accepted_following_ids(&alice.id).await.unwrap();
+        assert_eq!(ids, vec![bob.id]);
+    }
+}

crates/adapters/postgres/src/like.rs

@@ -1,2 +1,80 @@
-pub struct PgLikeRepository { _pool: sqlx::PgPool }
-impl PgLikeRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::social::Like, ports::LikeRepository, value_objects::{LikeId, ThoughtId, UserId}};
+
+pub struct PgLikeRepository { pool: PgPool }
+impl PgLikeRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl LikeRepository for PgLikeRepository {
+    async fn save(&self, l: &Like) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO likes(id,user_id,thought_id,ap_id,created_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT(user_id,thought_id) DO NOTHING"
+        )
+        .bind(l.id.as_uuid()).bind(l.user_id.as_uuid()).bind(l.thought_id.as_uuid()).bind(&l.ap_id).bind(l.created_at)
+        .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn delete(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+        let r = sqlx::query("DELETE FROM likes WHERE user_id=$1 AND thought_id=$2")
+            .bind(user_id.as_uuid()).bind(thought_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        if r.rows_affected() == 0 { return Err(DomainError::NotFound); }
+        Ok(())
+    }
+
+    async fn find(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<Option<Like>, DomainError> {
+        #[derive(sqlx::FromRow)]
+        struct Row { id: uuid::Uuid, user_id: uuid::Uuid, thought_id: uuid::Uuid, ap_id: Option<String>, created_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>("SELECT id,user_id,thought_id,ap_id,created_at FROM likes WHERE user_id=$1 AND thought_id=$2")
+            .bind(user_id.as_uuid()).bind(thought_id.as_uuid())
+            .fetch_optional(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+            .map(|o| o.map(|r| Like { id: LikeId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), thought_id: ThoughtId::from_uuid(r.thought_id), ap_id: r.ap_id, created_at: r.created_at }))
+    }
+
+    async fn count_for_thought(&self, thought_id: &ThoughtId) -> Result<i64, DomainError> {
+        sqlx::query_scalar("SELECT COUNT(*) FROM likes WHERE thought_id=$1")
+            .bind(thought_id.as_uuid()).fetch_one(&self.pool).await
+            .map_err(|e| DomainError::Internal(e.to_string()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::{thought::{Thought, Visibility}, user::User}, value_objects::*};
+    use crate::{thought::PgThoughtRepository, user::PgUserRepository};
+    use domain::ports::{ThoughtRepository, UserRepository};
+
+    async fn seed(pool: &sqlx::PgPool) -> (User, Thought) {
+        let urepo = PgUserRepository::new(pool.clone());
+        let trepo = PgThoughtRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()));
+        urepo.save(&u).await.unwrap();
+        let t = Thought::new_local(ThoughtId::new(), u.id.clone(), Content::new_local("hi").unwrap(), None, Visibility::Public, None, false);
+        trepo.save(&t).await.unwrap();
+        (u, t)
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn like_and_count(pool: sqlx::PgPool) {
+        let (user, thought) = seed(&pool).await;
+        let repo = PgLikeRepository::new(pool);
+        let like = Like { id: LikeId::new(), user_id: user.id.clone(), thought_id: thought.id.clone(), ap_id: None, created_at: Utc::now() };
+        repo.save(&like).await.unwrap();
+        assert_eq!(repo.count_for_thought(&thought.id).await.unwrap(), 1);
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn unlike(pool: sqlx::PgPool) {
+        let (user, thought) = seed(&pool).await;
+        let repo = PgLikeRepository::new(pool);
+        let like = Like { id: LikeId::new(), user_id: user.id.clone(), thought_id: thought.id.clone(), ap_id: None, created_at: Utc::now() };
+        repo.save(&like).await.unwrap();
+        repo.delete(&user.id, &thought.id).await.unwrap();
+        assert_eq!(repo.count_for_thought(&thought.id).await.unwrap(), 0);
+    }
+}

crates/adapters/postgres/src/notification.rs

@@ -1,2 +1,91 @@
-pub struct PgNotificationRepository { _pool: sqlx::PgPool }
-impl PgNotificationRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::{feed::{PageParams, Paginated}, notification::{Notification, NotificationType}}, ports::NotificationRepository, value_objects::{NotificationId, ThoughtId, UserId}};
+
+pub struct PgNotificationRepository { pool: PgPool }
+impl PgNotificationRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl NotificationRepository for PgNotificationRepository {
+    async fn save(&self, n: &Notification) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO notifications(id,user_id,type,from_user_id,thought_id,read,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)"
+        )
+        .bind(n.id.as_uuid()).bind(n.user_id.as_uuid()).bind(n.notification_type.as_str())
+        .bind(n.from_user_id.as_ref().map(|u| u.as_uuid()))
+        .bind(n.thought_id.as_ref().map(|t| t.as_uuid()))
+        .bind(n.read).bind(n.created_at)
+        .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn list_for_user(&self, user_id: &UserId, page: &PageParams) -> Result<Paginated<Notification>, DomainError> {
+        let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM notifications WHERE user_id=$1")
+            .bind(user_id.as_uuid()).fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        #[derive(sqlx::FromRow)]
+        struct Row { id: uuid::Uuid, user_id: uuid::Uuid, r#type: String, from_user_id: Option<uuid::Uuid>, thought_id: Option<uuid::Uuid>, read: bool, created_at: DateTime<Utc> }
+        let rows = sqlx::query_as::<_, Row>(
+            "SELECT id,user_id,type,from_user_id,thought_id,read,created_at FROM notifications WHERE user_id=$1 ORDER BY created_at DESC LIMIT $2 OFFSET $3"
+        ).bind(user_id.as_uuid()).bind(page.limit()).bind(page.offset())
+        .fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        let items = rows.into_iter().map(|r| Notification {
+            id: NotificationId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id),
+            notification_type: NotificationType::from_str(&r.r#type),
+            from_user_id: r.from_user_id.map(UserId::from_uuid),
+            thought_id: r.thought_id.map(ThoughtId::from_uuid),
+            read: r.read, created_at: r.created_at,
+        }).collect();
+        Ok(Paginated { items, total, page: page.page, per_page: page.per_page })
+    }
+
+    async fn mark_read(&self, id: &NotificationId, user_id: &UserId) -> Result<(), DomainError> {
+        sqlx::query("UPDATE notifications SET read=true WHERE id=$1 AND user_id=$2")
+            .bind(id.as_uuid()).bind(user_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn mark_all_read(&self, user_id: &UserId) -> Result<(), DomainError> {
+        sqlx::query("UPDATE notifications SET read=true WHERE user_id=$1")
+            .bind(user_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use domain::{models::{notification::NotificationType, user::User}, value_objects::*};
+    use crate::user::PgUserRepository;
+    use domain::ports::UserRepository;
+
+    async fn seed_user(pool: &sqlx::PgPool) -> User {
+        let repo = PgUserRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()));
+        repo.save(&u).await.unwrap(); u
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn save_and_list(pool: sqlx::PgPool) {
+        let user = seed_user(&pool).await;
+        let repo = PgNotificationRepository::new(pool);
+        use domain::models::feed::PageParams;
+        let n = Notification { id: NotificationId::new(), user_id: user.id.clone(), notification_type: NotificationType::Like, from_user_id: None, thought_id: None, read: false, created_at: Utc::now() };
+        repo.save(&n).await.unwrap();
+        let page = repo.list_for_user(&user.id, &PageParams { page: 1, per_page: 20 }).await.unwrap();
+        assert_eq!(page.total, 1);
+        assert!(!page.items[0].read);
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn mark_all_read(pool: sqlx::PgPool) {
+        let user = seed_user(&pool).await;
+        let repo = PgNotificationRepository::new(pool);
+        use domain::models::feed::PageParams;
+        let n = Notification { id: NotificationId::new(), user_id: user.id.clone(), notification_type: NotificationType::Follow, from_user_id: None, thought_id: None, read: false, created_at: Utc::now() };
+        repo.save(&n).await.unwrap();
+        repo.mark_all_read(&user.id).await.unwrap();
+        let page = repo.list_for_user(&user.id, &PageParams { page: 1, per_page: 20 }).await.unwrap();
+        assert!(page.items[0].read);
+    }
+}

crates/adapters/postgres/src/remote_actor.rs

@@ -1,2 +1,33 @@
-pub struct PgRemoteActorRepository { _pool: sqlx::PgPool }
-impl PgRemoteActorRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::remote_actor::RemoteActor, ports::RemoteActorRepository};
+
+pub struct PgRemoteActorRepository { pool: PgPool }
+impl PgRemoteActorRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl RemoteActorRepository for PgRemoteActorRepository {
+    async fn upsert(&self, a: &RemoteActor) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO remote_actors(url,handle,display_name,inbox_url,shared_inbox_url,public_key,last_fetched_at)
+             VALUES($1,$2,$3,$4,$5,$6,$7)
+             ON CONFLICT(url) DO UPDATE SET handle=EXCLUDED.handle,display_name=EXCLUDED.display_name,
+             inbox_url=EXCLUDED.inbox_url,shared_inbox_url=EXCLUDED.shared_inbox_url,
+             public_key=EXCLUDED.public_key,last_fetched_at=EXCLUDED.last_fetched_at"
+        )
+        .bind(&a.url).bind(&a.handle).bind(&a.display_name).bind(&a.inbox_url)
+        .bind(&a.shared_inbox_url).bind(&a.public_key).bind(a.last_fetched_at)
+        .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn find_by_url(&self, url: &str) -> Result<Option<RemoteActor>, DomainError> {
+        #[derive(sqlx::FromRow)]
+        struct Row { url: String, handle: String, display_name: Option<String>, inbox_url: String, shared_inbox_url: Option<String>, public_key: String, last_fetched_at: DateTime<Utc> }
+        sqlx::query_as::<_, Row>(
+            "SELECT url,handle,display_name,inbox_url,shared_inbox_url,public_key,last_fetched_at FROM remote_actors WHERE url=$1"
+        ).bind(url).fetch_optional(&self.pool).await
+        .map_err(|e| DomainError::Internal(e.to_string()))
+        .map(|o| o.map(|r| RemoteActor { url: r.url, handle: r.handle, display_name: r.display_name, inbox_url: r.inbox_url, shared_inbox_url: r.shared_inbox_url, public_key: r.public_key, last_fetched_at: r.last_fetched_at }))
+    }
+}

crates/adapters/postgres/src/tag.rs

@@ -1,2 +1,87 @@
-pub struct PgTagRepository { _pool: sqlx::PgPool }
-impl PgTagRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::{feed::{PageParams, Paginated}, tag::Tag, thought::Thought}, ports::TagRepository, value_objects::ThoughtId};
+
+pub struct PgTagRepository { pool: PgPool }
+impl PgTagRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl TagRepository for PgTagRepository {
+    async fn find_or_create(&self, name: &str) -> Result<Tag, DomainError> {
+        let name = name.to_lowercase();
+        sqlx::query("INSERT INTO tags(name) VALUES($1) ON CONFLICT(name) DO NOTHING")
+            .bind(&name).execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        #[derive(sqlx::FromRow)] struct Row { id: i32, name: String }
+        let row = sqlx::query_as::<_, Row>("SELECT id,name FROM tags WHERE name=$1").bind(&name)
+            .fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(Tag { id: row.id, name: row.name })
+    }
+
+    async fn attach_to_thought(&self, thought_id: &ThoughtId, tag_id: i32) -> Result<(), DomainError> {
+        sqlx::query("INSERT INTO thought_tags(thought_id,tag_id) VALUES($1,$2) ON CONFLICT DO NOTHING")
+            .bind(thought_id.as_uuid()).bind(tag_id)
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn detach_from_thought(&self, thought_id: &ThoughtId) -> Result<(), DomainError> {
+        sqlx::query("DELETE FROM thought_tags WHERE thought_id=$1").bind(thought_id.as_uuid())
+            .execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
+    }
+
+    async fn list_for_thought(&self, thought_id: &ThoughtId) -> Result<Vec<Tag>, DomainError> {
+        #[derive(sqlx::FromRow)] struct Row { id: i32, name: String }
+        sqlx::query_as::<_, Row>(
+            "SELECT t.id,t.name FROM tags t JOIN thought_tags tt ON tt.tag_id=t.id WHERE tt.thought_id=$1"
+        ).bind(thought_id.as_uuid()).fetch_all(&self.pool).await
+        .map_err(|e| DomainError::Internal(e.to_string()))
+        .map(|rows| rows.into_iter().map(|r| Tag { id: r.id, name: r.name }).collect())
+    }
+
+    async fn list_thoughts_by_tag(&self, tag_name: &str, page: &PageParams) -> Result<Paginated<Thought>, DomainError> {
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM thought_tags tt JOIN tags t ON t.id=tt.tag_id WHERE t.name=$1"
+        ).bind(tag_name).fetch_one(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        let rows = sqlx::query_as::<_, crate::thought::ThoughtRow>(
+            "SELECT th.id,th.user_id,th.content,th.in_reply_to_id,th.in_reply_to_url,th.ap_id,th.visibility,th.content_warning,th.sensitive,th.local,th.created_at,th.updated_at
+             FROM thoughts th JOIN thought_tags tt ON tt.thought_id=th.id JOIN tags t ON t.id=tt.tag_id
+             WHERE t.name=$1 ORDER BY th.created_at DESC LIMIT $2 OFFSET $3"
+        ).bind(tag_name).bind(page.limit()).bind(page.offset())
+        .fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(Paginated { items: rows.into_iter().map(Thought::from).collect(), total, page: page.page, per_page: page.per_page })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{models::{thought::{Thought, Visibility}, user::User}, value_objects::*};
+    use crate::{thought::PgThoughtRepository, user::PgUserRepository};
+    use domain::ports::{ThoughtRepository, UserRepository};
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn find_or_create_tag(pool: sqlx::PgPool) {
+        let repo = PgTagRepository::new(pool);
+        let t1 = repo.find_or_create("rust").await.unwrap();
+        let t2 = repo.find_or_create("rust").await.unwrap();
+        assert_eq!(t1.id, t2.id);
+        assert_eq!(t1.name, "rust");
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn attach_and_list(pool: sqlx::PgPool) {
+        let urepo = PgUserRepository::new(pool.clone());
+        let trepo = PgThoughtRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()));
+        urepo.save(&u).await.unwrap();
+        let t = Thought::new_local(ThoughtId::new(), u.id.clone(), Content::new_local("hi").unwrap(), None, Visibility::Public, None, false);
+        trepo.save(&t).await.unwrap();
+        let repo = PgTagRepository::new(pool);
+        let tag = repo.find_or_create("greetings").await.unwrap();
+        repo.attach_to_thought(&t.id, tag.id).await.unwrap();
+        let tags = repo.list_for_thought(&t.id).await.unwrap();
+        assert_eq!(tags.len(), 1);
+        assert_eq!(tags[0].name, "greetings");
+    }
+}

crates/adapters/postgres/src/top_friend.rs

@@ -1,2 +1,95 @@
-pub struct PgTopFriendRepository { _pool: sqlx::PgPool }
-impl PgTopFriendRepository { pub fn new(pool: sqlx::PgPool) -> Self { Self { _pool: pool } } }
+use async_trait::async_trait;
+use sqlx::PgPool;
+use domain::{errors::DomainError, models::{top_friend::TopFriend, user::User}, ports::TopFriendRepository, value_objects::UserId};
+
+pub struct PgTopFriendRepository { pool: PgPool }
+impl PgTopFriendRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+
+#[async_trait]
+impl TopFriendRepository for PgTopFriendRepository {
+    async fn set_top_friends(&self, user_id: &UserId, friends: Vec<(UserId, i16)>) -> Result<(), DomainError> {
+        let mut tx = self.pool.begin().await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        sqlx::query("DELETE FROM top_friends WHERE user_id=$1")
+            .bind(user_id.as_uuid()).execute(&mut *tx).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        for (friend_id, pos) in friends {
+            sqlx::query("INSERT INTO top_friends(user_id,friend_id,position) VALUES($1,$2,$3)")
+                .bind(user_id.as_uuid()).bind(friend_id.as_uuid()).bind(pos)
+                .execute(&mut *tx).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+        }
+        tx.commit().await.map_err(|e| DomainError::Internal(e.to_string()))
+    }
+
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<(TopFriend, User)>, DomainError> {
+        #[derive(sqlx::FromRow)]
+        struct Row {
+            tf_user_id: uuid::Uuid, friend_id: uuid::Uuid, position: i16,
+            id: uuid::Uuid, username: String, email: String, password_hash: String,
+            display_name: Option<String>, bio: Option<String>, avatar_url: Option<String>,
+            header_url: Option<String>, custom_css: Option<String>, local: bool,
+            ap_id: Option<String>, inbox_url: Option<String>, public_key: Option<String>,
+            private_key: Option<String>,
+            created_at: chrono::DateTime<chrono::Utc>, updated_at: chrono::DateTime<chrono::Utc>,
+        }
+        let rows = sqlx::query_as::<_, Row>(
+            "SELECT tf.user_id AS tf_user_id, tf.friend_id, tf.position,
+             u.id, u.username, u.email, u.password_hash, u.display_name, u.bio,
+             u.avatar_url, u.header_url, u.custom_css, u.local, u.ap_id, u.inbox_url,
+             u.public_key, u.private_key, u.created_at, u.updated_at
+             FROM top_friends tf JOIN users u ON u.id=tf.friend_id
+             WHERE tf.user_id=$1 ORDER BY tf.position"
+        ).bind(user_id.as_uuid()).fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(rows.into_iter().map(|r| {
+            use domain::value_objects::{Email, PasswordHash, Username};
+            let tf = TopFriend { user_id: UserId::from_uuid(r.tf_user_id), friend_id: UserId::from_uuid(r.friend_id), position: r.position };
+            let u = User {
+                id: UserId::from_uuid(r.id), username: Username::from_trusted(r.username),
+                email: Email::from_trusted(r.email), password_hash: PasswordHash(r.password_hash),
+                display_name: r.display_name, bio: r.bio, avatar_url: r.avatar_url,
+                header_url: r.header_url, custom_css: r.custom_css, local: r.local,
+                ap_id: r.ap_id, inbox_url: r.inbox_url, public_key: r.public_key,
+                private_key: r.private_key, created_at: r.created_at, updated_at: r.updated_at,
+            };
+            (tf, u)
+        }).collect())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{models::user::User, value_objects::*};
+    use crate::user::PgUserRepository;
+    use domain::ports::UserRepository;
+
+    async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
+        let repo = PgUserRepository::new(pool.clone());
+        let u = User::new_local(UserId::new(), Username::new(username).unwrap(), Email::new(email).unwrap(), PasswordHash("h".into()));
+        repo.save(&u).await.unwrap(); u
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn set_and_list_top_friends(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let repo = PgTopFriendRepository::new(pool);
+        repo.set_top_friends(&alice.id, vec![(bob.id.clone(), 1)]).await.unwrap();
+        let friends = repo.list_for_user(&alice.id).await.unwrap();
+        assert_eq!(friends.len(), 1);
+        assert_eq!(friends[0].0.position, 1);
+        assert_eq!(friends[0].1.username.as_str(), "bob");
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn replace_top_friends(pool: sqlx::PgPool) {
+        let alice = seed_user(&pool, "alice", "alice@ex.com").await;
+        let bob   = seed_user(&pool, "bob",   "bob@ex.com").await;
+        let carol = seed_user(&pool, "carol", "carol@ex.com").await;
+        let repo = PgTopFriendRepository::new(pool);
+        repo.set_top_friends(&alice.id, vec![(bob.id.clone(), 1)]).await.unwrap();
+        repo.set_top_friends(&alice.id, vec![(carol.id.clone(), 1)]).await.unwrap();
+        let friends = repo.list_for_user(&alice.id).await.unwrap();
+        assert_eq!(friends.len(), 1);
+        assert_eq!(friends[0].1.username.as_str(), "carol");
+    }
+}

crates/api-types/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod requests;
+pub mod responses;

crates/api-types/src/requests.rs

@@ -0,0 +1,71 @@
+use serde::Deserialize;
+use uuid::Uuid;
+
+#[derive(Deserialize)]
+pub struct RegisterRequest {
+    pub username: String,
+    pub email: String,
+    pub password: String,
+}
+
+#[derive(Deserialize)]
+pub struct LoginRequest {
+    pub email: String,
+    pub password: String,
+}
+
+#[derive(Deserialize)]
+pub struct CreateThoughtRequest {
+    pub content: String,
+    pub in_reply_to_id: Option<Uuid>,
+    pub visibility: Option<String>,
+    pub content_warning: Option<String>,
+    pub sensitive: Option<bool>,
+}
+
+#[derive(Deserialize)]
+pub struct EditThoughtRequest {
+    pub content: String,
+}
+
+#[derive(Deserialize)]
+pub struct UpdateProfileRequest {
+    pub display_name: Option<String>,
+    pub bio: Option<String>,
+    pub avatar_url: Option<String>,
+    pub header_url: Option<String>,
+    pub custom_css: Option<String>,
+}
+
+#[derive(Deserialize)]
+pub struct SetTopFriendsRequest {
+    pub friend_ids: Vec<Uuid>,
+}
+
+#[derive(Deserialize)]
+pub struct CreateApiKeyRequest {
+    pub name: String,
+}
+
+#[derive(Deserialize)]
+pub struct PaginationQuery {
+    pub page: Option<u64>,
+    pub per_page: Option<u64>,
+}
+
+impl PaginationQuery {
+    pub fn page(&self) -> u64 {
+        self.page.unwrap_or(1).max(1)
+    }
+
+    pub fn per_page(&self) -> u64 {
+        self.per_page.unwrap_or(20).min(100)
+    }
+}
+
+#[derive(Deserialize)]
+pub struct SearchQuery {
+    pub q: String,
+    pub page: Option<u64>,
+    pub per_page: Option<u64>,
+}

crates/api-types/src/responses.rs

@@ -0,0 +1,69 @@
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use uuid::Uuid;
+
+#[derive(Serialize)]
+pub struct AuthResponse {
+    pub token: String,
+    pub user: UserResponse,
+}
+
+#[derive(Serialize, Clone)]
+pub struct UserResponse {
+    pub id: Uuid,
+    pub username: String,
+    pub display_name: Option<String>,
+    pub bio: Option<String>,
+    pub avatar_url: Option<String>,
+    pub header_url: Option<String>,
+    pub local: bool,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Serialize, Clone)]
+pub struct ThoughtResponse {
+    pub id: Uuid,
+    pub content: String,
+    pub author: UserResponse,
+    pub in_reply_to_id: Option<Uuid>,
+    pub visibility: String,
+    pub content_warning: Option<String>,
+    pub sensitive: bool,
+    pub like_count: i64,
+    pub boost_count: i64,
+    pub reply_count: i64,
+    pub liked_by_viewer: bool,
+    pub boosted_by_viewer: bool,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Serialize)]
+pub struct PagedResponse<T: Serialize> {
+    pub items: Vec<T>,
+    pub total: i64,
+    pub page: u64,
+    pub per_page: u64,
+}
+
+#[derive(Serialize)]
+pub struct ApiKeyResponse {
+    pub id: Uuid,
+    pub name: String,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Serialize)]
+pub struct NotificationResponse {
+    pub id: Uuid,
+    pub notification_type: String,
+    pub from_user: Option<UserResponse>,
+    pub thought_id: Option<Uuid>,
+    pub read: bool,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Serialize)]
+pub struct ErrorResponse {
+    pub error: String,
+}

crates/application/Cargo.toml

@@ -9,6 +9,8 @@ async-trait = { workspace = true }
 thiserror   = { workspace = true }
 uuid        = { workspace = true }
 chrono      = { workspace = true }
+sha2        = "0.10"
+hex         = "0.4"
 
 [dev-dependencies]
 tokio  = { workspace = true, features = ["full"] }

crates/application/src/lib.rs

@@ -0,0 +1 @@
+pub mod use_cases;

crates/application/src/use_cases/api_keys.rs

@@ -0,0 +1,29 @@
+use chrono::Utc;
+use domain::{
+    errors::DomainError,
+    models::api_key::ApiKey,
+    ports::ApiKeyRepository,
+    value_objects::{ApiKeyId, UserId},
+};
+
+pub async fn list_api_keys(keys: &dyn ApiKeyRepository, user_id: &UserId) -> Result<Vec<ApiKey>, DomainError> {
+    keys.list_for_user(user_id).await
+}
+
+pub async fn create_api_key(keys: &dyn ApiKeyRepository, user_id: &UserId, name: String) -> Result<(ApiKey, String), DomainError> {
+    let raw_key = uuid::Uuid::new_v4().to_string().replace('-', "");
+    let key_hash = sha256_hex(&raw_key);
+    let key = ApiKey { id: ApiKeyId::new(), user_id: user_id.clone(), key_hash, name, created_at: Utc::now() };
+    keys.save(&key).await?;
+    Ok((key, raw_key))
+}
+
+pub async fn delete_api_key(keys: &dyn ApiKeyRepository, user_id: &UserId, key_id: &ApiKeyId) -> Result<(), DomainError> {
+    keys.delete(key_id, user_id).await
+}
+
+fn sha256_hex(s: &str) -> String {
+    use sha2::{Digest, Sha256};
+    let hash = Sha256::digest(s.as_bytes());
+    hex::encode(hash)
+}

crates/application/src/use_cases/auth.rs

@@ -0,0 +1,115 @@
+use domain::{
+    errors::DomainError,
+    models::user::User,
+    ports::{AuthService, EventPublisher, PasswordHasher, UserRepository},
+    value_objects::{Email, UserId, Username},
+};
+
+pub struct RegisterInput { pub username: String, pub email: String, pub password: String }
+#[derive(Debug)]
+pub struct RegisterOutput { pub user: User, pub token: String }
+
+pub async fn register(
+    users: &dyn UserRepository,
+    hasher: &dyn PasswordHasher,
+    auth: &dyn AuthService,
+    _events: &dyn EventPublisher,
+    input: RegisterInput,
+) -> Result<RegisterOutput, DomainError> {
+    let username = Username::new(input.username)?;
+    let email = Email::new(input.email)?;
+    if users.find_by_username(&username).await?.is_some() {
+        return Err(DomainError::Conflict("username taken".into()));
+    }
+    if users.find_by_email(&email).await?.is_some() {
+        return Err(DomainError::Conflict("email taken".into()));
+    }
+    let hash = hasher.hash(&input.password).await?;
+    let user = User::new_local(UserId::new(), username, email, hash);
+    users.save(&user).await?;
+    let token = auth.generate_token(&user.id)?;
+    Ok(RegisterOutput { user, token: token.token })
+}
+
+pub struct LoginInput { pub email: String, pub password: String }
+#[derive(Debug)]
+pub struct LoginOutput { pub user: User, pub token: String }
+
+pub async fn login(
+    users: &dyn UserRepository,
+    hasher: &dyn PasswordHasher,
+    auth: &dyn AuthService,
+    input: LoginInput,
+) -> Result<LoginOutput, DomainError> {
+    let email = Email::new(input.email)?;
+    let user = users.find_by_email(&email).await?.ok_or(DomainError::Unauthorized)?;
+    if !hasher.verify(&input.password, &user.password_hash).await? {
+        return Err(DomainError::Unauthorized);
+    }
+    let token = auth.generate_token(&user.id)?;
+    Ok(LoginOutput { user, token: token.token })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use async_trait::async_trait;
+    use domain::{
+        errors::DomainError,
+        ports::{AuthService, GeneratedToken, PasswordHasher},
+        testing::{NoOpEventPublisher, TestStore},
+        value_objects::{PasswordHash, UserId},
+    };
+
+    struct FakeHasher;
+    #[async_trait] impl PasswordHasher for FakeHasher {
+        async fn hash(&self, plain: &str) -> Result<PasswordHash, DomainError> { Ok(PasswordHash(plain.to_string())) }
+        async fn verify(&self, plain: &str, hash: &PasswordHash) -> Result<bool, DomainError> { Ok(plain == hash.0) }
+    }
+
+    struct FakeAuth;
+    impl AuthService for FakeAuth {
+        fn generate_token(&self, uid: &UserId) -> Result<GeneratedToken, DomainError> {
+            Ok(GeneratedToken { token: uid.to_string(), user_id: uid.clone() })
+        }
+        fn validate_token(&self, token: &str) -> Result<UserId, DomainError> {
+            Ok(UserId::from_uuid(uuid::Uuid::parse_str(token).map_err(|_| DomainError::Unauthorized)?))
+        }
+    }
+
+    fn input() -> RegisterInput {
+        RegisterInput { username: "alice".into(), email: "alice@ex.com".into(), password: "pw".into() }
+    }
+
+    #[tokio::test]
+    async fn register_creates_user() {
+        let store = TestStore::default();
+        let out = register(&store, &FakeHasher, &FakeAuth, &NoOpEventPublisher, input()).await.unwrap();
+        assert_eq!(out.user.username.as_str(), "alice");
+        assert!(!out.token.is_empty());
+    }
+
+    #[tokio::test]
+    async fn register_rejects_duplicate_username() {
+        let store = TestStore::default();
+        register(&store, &FakeHasher, &FakeAuth, &NoOpEventPublisher, input()).await.unwrap();
+        let err = register(&store, &FakeHasher, &FakeAuth, &NoOpEventPublisher, input()).await.unwrap_err();
+        assert!(matches!(err, DomainError::Conflict(_)));
+    }
+
+    #[tokio::test]
+    async fn login_succeeds_with_correct_password() {
+        let store = TestStore::default();
+        register(&store, &FakeHasher, &FakeAuth, &NoOpEventPublisher, input()).await.unwrap();
+        let out = login(&store, &FakeHasher, &FakeAuth, LoginInput { email: "alice@ex.com".into(), password: "pw".into() }).await.unwrap();
+        assert!(!out.token.is_empty());
+    }
+
+    #[tokio::test]
+    async fn login_fails_wrong_password() {
+        let store = TestStore::default();
+        register(&store, &FakeHasher, &FakeAuth, &NoOpEventPublisher, input()).await.unwrap();
+        let err = login(&store, &FakeHasher, &FakeAuth, LoginInput { email: "alice@ex.com".into(), password: "wrong".into() }).await.unwrap_err();
+        assert!(matches!(err, DomainError::Unauthorized));
+    }
+}

crates/application/src/use_cases/feed.rs

@@ -0,0 +1,43 @@
+use domain::{
+    errors::DomainError,
+    models::{
+        feed::{FeedEntry, PageParams, Paginated, UserSummary},
+        thought::Thought,
+        user::User,
+    },
+    ports::{FeedRepository, FollowRepository, TagRepository, ThoughtRepository, UserRepository},
+    value_objects::UserId,
+};
+
+pub async fn get_home_feed(feed: &dyn FeedRepository, follows: &dyn FollowRepository, user_id: &UserId, page: PageParams) -> Result<Paginated<FeedEntry>, DomainError> {
+    let following_ids = follows.get_accepted_following_ids(user_id).await?;
+    feed.home_feed(&following_ids, &page, Some(user_id)).await
+}
+
+pub async fn get_public_feed(feed: &dyn FeedRepository, viewer_id: Option<&UserId>, page: PageParams) -> Result<Paginated<FeedEntry>, DomainError> {
+    feed.public_feed(&page, viewer_id).await
+}
+
+pub async fn get_user_feed(thoughts: &dyn ThoughtRepository, user_id: &UserId, page: PageParams) -> Result<Paginated<FeedEntry>, DomainError> {
+    thoughts.list_by_user(user_id, &page).await
+}
+
+pub async fn get_followers(follows: &dyn FollowRepository, user_id: &UserId, page: PageParams) -> Result<Paginated<User>, DomainError> {
+    follows.list_followers(user_id, &page).await
+}
+
+pub async fn get_following(follows: &dyn FollowRepository, user_id: &UserId, page: PageParams) -> Result<Paginated<User>, DomainError> {
+    follows.list_following(user_id, &page).await
+}
+
+pub async fn get_by_tag(tags: &dyn TagRepository, tag_name: &str, page: PageParams) -> Result<Paginated<Thought>, DomainError> {
+    tags.list_thoughts_by_tag(tag_name, &page).await
+}
+
+pub async fn search(feed: &dyn FeedRepository, query: &str, page: PageParams, viewer_id: Option<&UserId>) -> Result<Paginated<FeedEntry>, DomainError> {
+    feed.search(query, &page, viewer_id).await
+}
+
+pub async fn list_users(users: &dyn UserRepository) -> Result<Vec<UserSummary>, DomainError> {
+    users.list_with_stats().await
+}

crates/application/src/use_cases/mod.rs

@@ -0,0 +1,6 @@
+pub mod api_keys;
+pub mod auth;
+pub mod feed;
+pub mod profile;
+pub mod social;
+pub mod thoughts;

crates/application/src/use_cases/profile.rs

@@ -0,0 +1,37 @@
+use domain::{
+    errors::DomainError,
+    models::{top_friend::TopFriend, user::User},
+    ports::{TopFriendRepository, UserRepository},
+    value_objects::{UserId, Username},
+};
+
+pub async fn get_user(users: &dyn UserRepository, user_id: &UserId) -> Result<User, DomainError> {
+    users.find_by_id(user_id).await?.ok_or(DomainError::NotFound)
+}
+
+pub async fn get_user_by_username(users: &dyn UserRepository, username: &str) -> Result<User, DomainError> {
+    let username = Username::from_trusted(username.to_string());
+    users.find_by_username(&username).await?.ok_or(DomainError::NotFound)
+}
+
+pub async fn update_profile(
+    users: &dyn UserRepository,
+    user_id: &UserId,
+    display_name: Option<String>,
+    bio: Option<String>,
+    avatar_url: Option<String>,
+    header_url: Option<String>,
+    custom_css: Option<String>,
+) -> Result<(), DomainError> {
+    users.update_profile(user_id, display_name, bio, avatar_url, header_url, custom_css).await
+}
+
+pub async fn get_top_friends(top_friends: &dyn TopFriendRepository, user_id: &UserId) -> Result<Vec<(TopFriend, User)>, DomainError> {
+    top_friends.list_for_user(user_id).await
+}
+
+pub async fn set_top_friends(top_friends: &dyn TopFriendRepository, user_id: &UserId, friend_ids: Vec<UserId>) -> Result<(), DomainError> {
+    if friend_ids.len() > 8 { return Err(DomainError::InvalidInput("top friends: max 8".into())); }
+    let friends: Vec<(UserId, i16)> = friend_ids.into_iter().enumerate().map(|(i, id)| (id, (i + 1) as i16)).collect();
+    top_friends.set_top_friends(user_id, friends).await
+}

crates/application/src/use_cases/social.rs

@@ -0,0 +1,117 @@
+use chrono::Utc;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    models::social::{Block, Boost, Follow, FollowState, Like},
+    ports::{BlockRepository, BoostRepository, EventPublisher, FollowRepository, LikeRepository},
+    value_objects::{BoostId, LikeId, ThoughtId, UserId},
+};
+
+pub async fn like_thought(likes: &dyn LikeRepository, events: &dyn EventPublisher, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+    let like = Like { id: LikeId::new(), user_id: user_id.clone(), thought_id: thought_id.clone(), ap_id: None, created_at: Utc::now() };
+    likes.save(&like).await?;
+    events.publish(&DomainEvent::LikeAdded { like_id: like.id, user_id: user_id.clone(), thought_id: thought_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn unlike_thought(likes: &dyn LikeRepository, events: &dyn EventPublisher, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+    likes.delete(user_id, thought_id).await?;
+    events.publish(&DomainEvent::LikeRemoved { user_id: user_id.clone(), thought_id: thought_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn boost_thought(boosts: &dyn BoostRepository, events: &dyn EventPublisher, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+    let boost = Boost { id: BoostId::new(), user_id: user_id.clone(), thought_id: thought_id.clone(), ap_id: None, created_at: Utc::now() };
+    boosts.save(&boost).await?;
+    events.publish(&DomainEvent::BoostAdded { boost_id: boost.id, user_id: user_id.clone(), thought_id: thought_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn unboost_thought(boosts: &dyn BoostRepository, events: &dyn EventPublisher, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
+    boosts.delete(user_id, thought_id).await?;
+    events.publish(&DomainEvent::BoostRemoved { user_id: user_id.clone(), thought_id: thought_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn follow_user(follows: &dyn FollowRepository, events: &dyn EventPublisher, follower_id: &UserId, following_id: &UserId) -> Result<(), DomainError> {
+    if follower_id == following_id { return Err(DomainError::InvalidInput("cannot follow yourself".into())); }
+    let follow = Follow { follower_id: follower_id.clone(), following_id: following_id.clone(), state: FollowState::Accepted, ap_id: None, created_at: Utc::now() };
+    follows.save(&follow).await?;
+    events.publish(&DomainEvent::FollowAccepted { follower_id: follower_id.clone(), following_id: following_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn unfollow_user(follows: &dyn FollowRepository, events: &dyn EventPublisher, follower_id: &UserId, following_id: &UserId) -> Result<(), DomainError> {
+    follows.delete(follower_id, following_id).await?;
+    events.publish(&DomainEvent::Unfollowed { follower_id: follower_id.clone(), following_id: following_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn accept_follow(follows: &dyn FollowRepository, events: &dyn EventPublisher, follower_id: &UserId, following_id: &UserId) -> Result<(), DomainError> {
+    follows.update_state(follower_id, following_id, &FollowState::Accepted).await?;
+    events.publish(&DomainEvent::FollowAccepted { follower_id: follower_id.clone(), following_id: following_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn reject_follow(follows: &dyn FollowRepository, events: &dyn EventPublisher, follower_id: &UserId, following_id: &UserId) -> Result<(), DomainError> {
+    follows.update_state(follower_id, following_id, &FollowState::Rejected).await?;
+    events.publish(&DomainEvent::FollowRejected { follower_id: follower_id.clone(), following_id: following_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn block_user(blocks: &dyn BlockRepository, events: &dyn EventPublisher, blocker_id: &UserId, blocked_id: &UserId) -> Result<(), DomainError> {
+    if blocker_id == blocked_id { return Err(DomainError::InvalidInput("cannot block yourself".into())); }
+    let block = Block { blocker_id: blocker_id.clone(), blocked_id: blocked_id.clone(), created_at: Utc::now() };
+    blocks.save(&block).await?;
+    events.publish(&DomainEvent::UserBlocked { blocker_id: blocker_id.clone(), blocked_id: blocked_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn unblock_user(blocks: &dyn BlockRepository, blocker_id: &UserId, blocked_id: &UserId) -> Result<(), DomainError> {
+    blocks.delete(blocker_id, blocked_id).await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{
+        models::{thought::{Thought, Visibility}, user::User},
+        testing::TestStore,
+        value_objects::*,
+    };
+
+    fn user(name: &str) -> User {
+        User::new_local(UserId::new(), Username::new(name).unwrap(), Email::new(format!("{name}@ex.com")).unwrap(), PasswordHash("h".into()))
+    }
+
+    #[tokio::test]
+    async fn like_and_unlike() {
+        let store = TestStore::default();
+        let alice = user("alice");
+        let tid = ThoughtId::new();
+        store.thoughts.lock().unwrap().push(Thought::new_local(tid.clone(), alice.id.clone(), Content::new_local("hi").unwrap(), None, Visibility::Public, None, false));
+        like_thought(&store, &store, &alice.id, &tid).await.unwrap();
+        assert_eq!(store.likes.lock().unwrap().len(), 1);
+        unlike_thought(&store, &store, &alice.id, &tid).await.unwrap();
+        assert!(store.likes.lock().unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn follow_and_unfollow() {
+        let store = TestStore::default();
+        let alice = user("alice"); let bob = user("bob");
+        follow_user(&store, &store, &alice.id, &bob.id).await.unwrap();
+        assert_eq!(store.follows.lock().unwrap().len(), 1);
+        unfollow_user(&store, &store, &alice.id, &bob.id).await.unwrap();
+        assert!(store.follows.lock().unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn cannot_follow_self() {
+        let store = TestStore::default();
+        let alice = user("alice");
+        let err = follow_user(&store, &store, &alice.id, &alice.id).await.unwrap_err();
+        assert!(matches!(err, DomainError::InvalidInput(_)));
+    }
+}

crates/application/src/use_cases/thoughts.rs

@@ -0,0 +1,122 @@
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    models::thought::{Thought, Visibility},
+    ports::{EventPublisher, ThoughtRepository, UserRepository},
+    value_objects::{Content, ThoughtId, UserId},
+};
+
+pub struct CreateThoughtInput {
+    pub user_id: UserId,
+    pub content: String,
+    pub in_reply_to_id: Option<ThoughtId>,
+    pub visibility: Option<String>,
+    pub content_warning: Option<String>,
+    pub sensitive: bool,
+}
+pub struct CreateThoughtOutput { pub thought: Thought }
+
+pub async fn create_thought(
+    thoughts: &dyn ThoughtRepository,
+    _users: &dyn UserRepository,
+    events: &dyn EventPublisher,
+    input: CreateThoughtInput,
+) -> Result<CreateThoughtOutput, DomainError> {
+    let content = Content::new_local(input.content)?;
+    let visibility = input.visibility.as_deref().map(Visibility::from_str).unwrap_or(Visibility::Public);
+    let thought = Thought::new_local(
+        ThoughtId::new(), input.user_id,
+        content, input.in_reply_to_id.clone(),
+        visibility, input.content_warning, input.sensitive,
+    );
+    thoughts.save(&thought).await?;
+    events.publish(&DomainEvent::ThoughtCreated {
+        thought_id: thought.id.clone(),
+        user_id: thought.user_id.clone(),
+        in_reply_to_id: input.in_reply_to_id,
+    }).await?;
+    Ok(CreateThoughtOutput { thought })
+}
+
+pub async fn delete_thought(
+    thoughts: &dyn ThoughtRepository,
+    events: &dyn EventPublisher,
+    id: &ThoughtId,
+    user_id: &UserId,
+) -> Result<(), DomainError> {
+    let thought = thoughts.find_by_id(id).await?.ok_or(DomainError::NotFound)?;
+    if thought.user_id != *user_id { return Err(DomainError::NotFound); }
+    thoughts.delete(id, user_id).await?;
+    events.publish(&DomainEvent::ThoughtDeleted { thought_id: id.clone(), user_id: user_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn edit_thought(
+    thoughts: &dyn ThoughtRepository,
+    events: &dyn EventPublisher,
+    id: &ThoughtId,
+    user_id: &UserId,
+    new_content: String,
+) -> Result<(), DomainError> {
+    let thought = thoughts.find_by_id(id).await?.ok_or(DomainError::NotFound)?;
+    if thought.user_id != *user_id { return Err(DomainError::NotFound); }
+    let content = Content::new_local(new_content)?;
+    thoughts.update_content(id, &content).await?;
+    events.publish(&DomainEvent::ThoughtUpdated { thought_id: id.clone(), user_id: user_id.clone() }).await?;
+    Ok(())
+}
+
+pub async fn get_thought(thoughts: &dyn ThoughtRepository, id: &ThoughtId) -> Result<Thought, DomainError> {
+    thoughts.find_by_id(id).await?.ok_or(DomainError::NotFound)
+}
+
+pub async fn get_thread(thoughts: &dyn ThoughtRepository, id: &ThoughtId) -> Result<Vec<Thought>, DomainError> {
+    thoughts.get_thread(id).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{
+        models::user::User,
+        testing::{NoOpEventPublisher, TestStore},
+        value_objects::*,
+    };
+
+    fn user() -> User {
+        User::new_local(UserId::new(), Username::new("alice").unwrap(), Email::new("alice@ex.com").unwrap(), PasswordHash("h".into()))
+    }
+
+    fn input(uid: UserId) -> CreateThoughtInput {
+        CreateThoughtInput { user_id: uid, content: "hello".into(), in_reply_to_id: None, visibility: None, content_warning: None, sensitive: false }
+    }
+
+    #[tokio::test]
+    async fn create_thought_saves_and_emits_event() {
+        let store = TestStore::default();
+        let u = user(); store.users.lock().unwrap().push(u.clone());
+        let out = create_thought(&store, &store, &store, input(u.id.clone())).await.unwrap();
+        assert_eq!(out.thought.content.as_str(), "hello");
+        assert_eq!(store.events.lock().unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn delete_own_thought_succeeds() {
+        let store = TestStore::default();
+        let u = user(); store.users.lock().unwrap().push(u.clone());
+        let out = create_thought(&store, &store, &NoOpEventPublisher, input(u.id.clone())).await.unwrap();
+        delete_thought(&store, &NoOpEventPublisher, &out.thought.id, &u.id).await.unwrap();
+        assert!(store.thoughts.lock().unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn delete_other_thought_returns_not_found() {
+        let store = TestStore::default();
+        let alice = user();
+        let bob = User::new_local(UserId::new(), Username::new("bob").unwrap(), Email::new("bob@ex.com").unwrap(), PasswordHash("h".into()));
+        store.users.lock().unwrap().extend([alice.clone(), bob.clone()]);
+        let out = create_thought(&store, &store, &NoOpEventPublisher, input(alice.id.clone())).await.unwrap();
+        let err = delete_thought(&store, &NoOpEventPublisher, &out.thought.id, &bob.id).await.unwrap_err();
+        assert!(matches!(err, DomainError::NotFound));
+    }
+}

crates/presentation/Cargo.toml

@@ -14,6 +14,7 @@ api-types   = { workspace = true }
 postgres    = { workspace = true }
 auth        = { workspace = true }
 axum        = { workspace = true }
+sqlx        = { workspace = true }
 tower-http  = { workspace = true }
 tokio       = { workspace = true, features = ["full"] }
 serde       = { workspace = true }

crates/presentation/src/errors.rs

@@ -0,0 +1,29 @@
+use axum::{http::StatusCode, response::{IntoResponse, Response}, Json};
+use domain::errors::DomainError;
+use api_types::responses::ErrorResponse;
+
+pub enum ApiError {
+    Domain(DomainError),
+    Unauthorized,
+    BadRequest(String),
+}
+
+impl From<DomainError> for ApiError {
+    fn from(e: DomainError) -> Self { Self::Domain(e) }
+}
+
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        let (status, msg) = match self {
+            Self::Domain(DomainError::NotFound)          => (StatusCode::NOT_FOUND, "not found".into()),
+            Self::Domain(DomainError::Unauthorized)      => (StatusCode::UNAUTHORIZED, "unauthorized".into()),
+            Self::Domain(DomainError::Forbidden)         => (StatusCode::FORBIDDEN, "forbidden".into()),
+            Self::Domain(DomainError::Conflict(m))       => (StatusCode::CONFLICT, m),
+            Self::Domain(DomainError::InvalidInput(m))   => (StatusCode::UNPROCESSABLE_ENTITY, m),
+            Self::Domain(DomainError::Internal(_))       => (StatusCode::INTERNAL_SERVER_ERROR, "internal server error".into()),
+            Self::Unauthorized                           => (StatusCode::UNAUTHORIZED, "unauthorized".into()),
+            Self::BadRequest(m)                          => (StatusCode::BAD_REQUEST, m),
+        };
+        (status, Json(ErrorResponse { error: msg })).into_response()
+    }
+}

crates/presentation/src/extractors.rs

@@ -0,0 +1,47 @@
+use axum::{extract::FromRequestParts, http::request::Parts};
+use domain::value_objects::UserId;
+use crate::{errors::ApiError, state::AppState};
+
+pub struct AuthUser(pub UserId);
+pub struct OptionalAuthUser(pub Option<UserId>);
+
+impl FromRequestParts<AppState> for AuthUser {
+    type Rejection = ApiError;
+    async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, ApiError> {
+        extract_user_id(parts, state).await?
+            .ok_or(ApiError::Unauthorized)
+            .map(AuthUser)
+    }
+}
+
+impl FromRequestParts<AppState> for OptionalAuthUser {
+    type Rejection = ApiError;
+    async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, ApiError> {
+        Ok(OptionalAuthUser(extract_user_id(parts, state).await?))
+    }
+}
+
+async fn extract_user_id(parts: &mut Parts, state: &AppState) -> Result<Option<UserId>, ApiError> {
+    if let Some(auth_header) = parts.headers.get("Authorization") {
+        if let Ok(s) = auth_header.to_str() {
+            if let Some(token) = s.strip_prefix("Bearer ") {
+                return state.auth.validate_token(token).map(Some).map_err(|_| ApiError::Unauthorized);
+            }
+        }
+    }
+    if let Some(key_header) = parts.headers.get("X-Api-Key") {
+        if let Ok(raw) = key_header.to_str() {
+            let hash = sha256_hex(raw);
+            if let Ok(Some(key)) = state.api_keys.find_by_hash(&hash).await {
+                return Ok(Some(key.user_id));
+            }
+        }
+    }
+    Ok(None)
+}
+
+fn sha256_hex(s: &str) -> String {
+    use sha2::{Digest, Sha256};
+    let hash = Sha256::digest(s.as_bytes());
+    hex::encode(hash)
+}

crates/presentation/src/handlers/api_keys.rs

@@ -0,0 +1,19 @@
+use axum::{extract::{Path, State}, http::StatusCode, Json};
+use uuid::Uuid;
+use api_types::{requests::CreateApiKeyRequest, responses::ApiKeyResponse};
+use application::use_cases::api_keys::{create_api_key, delete_api_key, list_api_keys};
+use domain::value_objects::ApiKeyId;
+use crate::{errors::ApiError, extractors::AuthUser, state::AppState};
+
+pub async fn get_api_keys(State(s): State<AppState>, AuthUser(uid): AuthUser) -> Result<Json<Vec<ApiKeyResponse>>, ApiError> {
+    let keys = list_api_keys(&*s.api_keys, &uid).await?;
+    Ok(Json(keys.into_iter().map(|k| ApiKeyResponse { id: k.id.as_uuid(), name: k.name, created_at: k.created_at }).collect()))
+}
+pub async fn post_api_key(State(s): State<AppState>, AuthUser(uid): AuthUser, Json(body): Json<CreateApiKeyRequest>) -> Result<Json<serde_json::Value>, ApiError> {
+    let (key, raw) = create_api_key(&*s.api_keys, &uid, body.name).await?;
+    Ok(Json(serde_json::json!({ "id": key.id.as_uuid(), "name": key.name, "key": raw })))
+}
+pub async fn delete_api_key_handler(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    delete_api_key(&*s.api_keys, &uid, &ApiKeyId::from_uuid(id)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}

crates/presentation/src/handlers/auth.rs

@@ -0,0 +1,35 @@
+use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
+use api_types::{requests::{LoginRequest, RegisterRequest}, responses::{AuthResponse, UserResponse}};
+use application::use_cases::auth::{login, register, LoginInput, RegisterInput};
+use crate::{errors::ApiError, state::AppState};
+
+pub fn to_user_response(u: &domain::models::user::User) -> UserResponse {
+    UserResponse {
+        id: u.id.as_uuid(),
+        username: u.username.to_string(),
+        display_name: u.display_name.clone(),
+        bio: u.bio.clone(),
+        avatar_url: u.avatar_url.clone(),
+        header_url: u.header_url.clone(),
+        local: u.local,
+        created_at: u.created_at,
+    }
+}
+
+pub async fn post_register(State(s): State<AppState>, Json(body): Json<RegisterRequest>) -> Result<impl IntoResponse, ApiError> {
+    let out = register(&*s.users, &*s.hasher, &*s.auth, &*s.events, RegisterInput {
+        username: body.username,
+        email: body.email,
+        password: body.password,
+    }).await?;
+    let resp = AuthResponse { token: out.token, user: to_user_response(&out.user) };
+    Ok((StatusCode::CREATED, Json(resp)))
+}
+
+pub async fn post_login(State(s): State<AppState>, Json(body): Json<LoginRequest>) -> Result<impl IntoResponse, ApiError> {
+    let out = login(&*s.users, &*s.hasher, &*s.auth, LoginInput {
+        email: body.email,
+        password: body.password,
+    }).await?;
+    Ok(Json(AuthResponse { token: out.token, user: to_user_response(&out.user) }))
+}

crates/presentation/src/handlers/feed.rs

@@ -0,0 +1,38 @@
+use axum::{extract::{Path, Query, State}, Json};
+use api_types::requests::{PaginationQuery, SearchQuery};
+use application::use_cases::feed::{get_home_feed, get_public_feed, get_followers, get_following, search};
+use domain::models::feed::PageParams;
+use crate::{errors::ApiError, extractors::{AuthUser, OptionalAuthUser}, handlers::auth::to_user_response, state::AppState};
+use application::use_cases::profile::get_user_by_username;
+
+pub async fn home_feed(State(s): State<AppState>, AuthUser(uid): AuthUser, Query(q): Query<PaginationQuery>) -> Result<Json<serde_json::Value>, ApiError> {
+    let page = PageParams { page: q.page(), per_page: q.per_page() };
+    let result = get_home_feed(&*s.feed, &*s.follows, &uid, page).await?;
+    Ok(Json(serde_json::json!({ "items": result.items.iter().map(|e| e.thought.id.as_uuid()).collect::<Vec<_>>(), "total": result.total, "page": result.page })))
+}
+
+pub async fn public_feed(State(s): State<AppState>, OptionalAuthUser(viewer): OptionalAuthUser, Query(q): Query<PaginationQuery>) -> Result<Json<serde_json::Value>, ApiError> {
+    let page = PageParams { page: q.page(), per_page: q.per_page() };
+    let result = get_public_feed(&*s.feed, viewer.as_ref(), page).await?;
+    Ok(Json(serde_json::json!({ "items": result.items.iter().map(|e| e.thought.id.as_uuid()).collect::<Vec<_>>(), "total": result.total, "page": result.page })))
+}
+
+pub async fn search_handler(State(s): State<AppState>, OptionalAuthUser(viewer): OptionalAuthUser, Query(q): Query<SearchQuery>) -> Result<Json<serde_json::Value>, ApiError> {
+    let page = PageParams { page: q.page.unwrap_or(1), per_page: q.per_page.unwrap_or(20) };
+    let result = search(&*s.feed, &q.q, page, viewer.as_ref()).await?;
+    Ok(Json(serde_json::json!({ "items": result.items.iter().map(|e| e.thought.id.as_uuid()).collect::<Vec<_>>(), "total": result.total })))
+}
+
+pub async fn get_following_handler(State(s): State<AppState>, Path(username): Path<String>, Query(q): Query<PaginationQuery>) -> Result<Json<serde_json::Value>, ApiError> {
+    let user = get_user_by_username(&*s.users, &username).await?;
+    let page = PageParams { page: q.page(), per_page: q.per_page() };
+    let result = get_following(&*s.follows, &user.id, page).await?;
+    Ok(Json(serde_json::json!({ "total": result.total, "items": result.items.iter().map(to_user_response).collect::<Vec<_>>() })))
+}
+
+pub async fn get_followers_handler(State(s): State<AppState>, Path(username): Path<String>, Query(q): Query<PaginationQuery>) -> Result<Json<serde_json::Value>, ApiError> {
+    let user = get_user_by_username(&*s.users, &username).await?;
+    let page = PageParams { page: q.page(), per_page: q.per_page() };
+    let result = get_followers(&*s.follows, &user.id, page).await?;
+    Ok(Json(serde_json::json!({ "total": result.total, "items": result.items.iter().map(to_user_response).collect::<Vec<_>>() })))
+}

crates/presentation/src/handlers/mod.rs

@@ -0,0 +1,7 @@
+pub mod api_keys;
+pub mod auth;
+pub mod feed;
+pub mod notifications;
+pub mod social;
+pub mod thoughts;
+pub mod users;

crates/presentation/src/handlers/notifications.rs

@@ -0,0 +1,18 @@
+use axum::{extract::{Path, State}, http::StatusCode, Json};
+use uuid::Uuid;
+use domain::{models::feed::PageParams, value_objects::NotificationId};
+use crate::{errors::ApiError, extractors::AuthUser, state::AppState};
+
+pub async fn list_notifications(State(s): State<AppState>, AuthUser(uid): AuthUser) -> Result<Json<serde_json::Value>, ApiError> {
+    let page = PageParams { page: 1, per_page: 20 };
+    let result = s.notifications.list_for_user(&uid, &page).await?;
+    Ok(Json(serde_json::json!({ "total": result.total, "unread": result.items.iter().filter(|n| !n.read).count() })))
+}
+pub async fn mark_notification_read(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    s.notifications.mark_read(&NotificationId::from_uuid(id), &uid).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn mark_all_read(State(s): State<AppState>, AuthUser(uid): AuthUser) -> Result<StatusCode, ApiError> {
+    s.notifications.mark_all_read(&uid).await?;
+    Ok(StatusCode::NO_CONTENT)
+}

crates/presentation/src/handlers/social.rs

@@ -0,0 +1,51 @@
+use axum::{extract::{Path, State}, http::StatusCode, Json};
+use uuid::Uuid;
+use api_types::requests::SetTopFriendsRequest;
+use application::use_cases::social::*;
+use application::use_cases::profile::{get_top_friends, set_top_friends, get_user_by_username};
+use domain::value_objects::{ThoughtId, UserId};
+use crate::{errors::ApiError, extractors::AuthUser, state::AppState};
+
+pub async fn post_like(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    like_thought(&*s.likes, &*s.events, &uid, &ThoughtId::from_uuid(id)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn delete_like(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    unlike_thought(&*s.likes, &*s.events, &uid, &ThoughtId::from_uuid(id)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn post_boost(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    boost_thought(&*s.boosts, &*s.events, &uid, &ThoughtId::from_uuid(id)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn delete_boost(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    unboost_thought(&*s.boosts, &*s.events, &uid, &ThoughtId::from_uuid(id)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn post_follow(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(target): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    follow_user(&*s.follows, &*s.events, &uid, &UserId::from_uuid(target)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn delete_follow(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(target): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    unfollow_user(&*s.follows, &*s.events, &uid, &UserId::from_uuid(target)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn post_block(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(target): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    block_user(&*s.blocks, &*s.events, &uid, &UserId::from_uuid(target)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn delete_block(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(target): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    unblock_user(&*s.blocks, &uid, &UserId::from_uuid(target)).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn put_top_friends(State(s): State<AppState>, AuthUser(uid): AuthUser, Json(body): Json<SetTopFriendsRequest>) -> Result<StatusCode, ApiError> {
+    let ids: Vec<UserId> = body.friend_ids.into_iter().map(UserId::from_uuid).collect();
+    set_top_friends(&*s.top_friends, &uid, ids).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+pub async fn get_top_friends_handler(State(s): State<AppState>, Path(username): Path<String>) -> Result<Json<serde_json::Value>, ApiError> {
+    let user = get_user_by_username(&*s.users, &username).await?;
+    let friends = get_top_friends(&*s.top_friends, &user.id).await?;
+    let ids: Vec<Uuid> = friends.iter().map(|(tf, _)| tf.friend_id.as_uuid()).collect();
+    Ok(Json(serde_json::json!({ "top_friends": ids })))
+}

crates/presentation/src/handlers/thoughts.rs

@@ -0,0 +1,64 @@
+use axum::{extract::{Path, State}, http::StatusCode, response::IntoResponse, Json};
+use uuid::Uuid;
+use api_types::requests::{CreateThoughtRequest, EditThoughtRequest};
+use application::use_cases::thoughts::{create_thought, delete_thought, edit_thought, get_thought, get_thread, CreateThoughtInput};
+use domain::value_objects::ThoughtId;
+use crate::{errors::ApiError, extractors::{AuthUser, OptionalAuthUser}, handlers::auth::to_user_response, state::AppState};
+
+fn thought_to_json(t: &domain::models::thought::Thought, author: &domain::models::user::User, like_count: i64, boost_count: i64, reply_count: i64) -> serde_json::Value {
+    serde_json::json!({
+        "id": t.id.as_uuid(),
+        "content": t.content.as_str(),
+        "author": to_user_response(author),
+        "in_reply_to_id": t.in_reply_to_id.as_ref().map(|x| x.as_uuid()),
+        "visibility": t.visibility.as_str(),
+        "content_warning": t.content_warning,
+        "sensitive": t.sensitive,
+        "like_count": like_count,
+        "boost_count": boost_count,
+        "reply_count": reply_count,
+        "created_at": t.created_at,
+        "updated_at": t.updated_at,
+    })
+}
+
+pub async fn post_thought(State(s): State<AppState>, AuthUser(uid): AuthUser, Json(body): Json<CreateThoughtRequest>) -> Result<impl IntoResponse, ApiError> {
+    let in_reply_to = body.in_reply_to_id.map(ThoughtId::from_uuid);
+    let out = create_thought(&*s.thoughts, &*s.users, &*s.events, CreateThoughtInput {
+        user_id: uid.clone(),
+        content: body.content,
+        in_reply_to_id: in_reply_to,
+        visibility: body.visibility,
+        content_warning: body.content_warning,
+        sensitive: body.sensitive.unwrap_or(false),
+    }).await?;
+    let author = s.users.find_by_id(&uid).await?.ok_or(domain::errors::DomainError::NotFound)?;
+    Ok((StatusCode::CREATED, Json(thought_to_json(&out.thought, &author, 0, 0, 0))))
+}
+
+pub async fn get_thought_handler(State(s): State<AppState>, Path(id): Path<Uuid>, OptionalAuthUser(_viewer): OptionalAuthUser) -> Result<Json<serde_json::Value>, ApiError> {
+    let thought = get_thought(&*s.thoughts, &ThoughtId::from_uuid(id)).await?;
+    let author = s.users.find_by_id(&thought.user_id).await?.ok_or(domain::errors::DomainError::NotFound)?;
+    Ok(Json(thought_to_json(&thought, &author, 0, 0, 0)))
+}
+
+pub async fn delete_thought_handler(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>) -> Result<StatusCode, ApiError> {
+    delete_thought(&*s.thoughts, &*s.events, &ThoughtId::from_uuid(id), &uid).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+pub async fn patch_thought(State(s): State<AppState>, AuthUser(uid): AuthUser, Path(id): Path<Uuid>, Json(body): Json<EditThoughtRequest>) -> Result<StatusCode, ApiError> {
+    edit_thought(&*s.thoughts, &*s.events, &ThoughtId::from_uuid(id), &uid, body.content).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+pub async fn get_thread_handler(State(s): State<AppState>, Path(id): Path<Uuid>) -> Result<Json<Vec<serde_json::Value>>, ApiError> {
+    let thoughts = get_thread(&*s.thoughts, &ThoughtId::from_uuid(id)).await?;
+    let mut items = Vec::new();
+    for t in &thoughts {
+        if let Ok(Some(author)) = s.users.find_by_id(&t.user_id).await {
+            items.push(thought_to_json(t, &author, 0, 0, 0));
+        }
+    }
+    Ok(Json(items))
+}

crates/presentation/src/handlers/users.rs

@@ -0,0 +1,15 @@
+use axum::{extract::{Path, State}, Json};
+use api_types::{requests::UpdateProfileRequest, responses::UserResponse};
+use application::use_cases::profile::{get_user_by_username, update_profile};
+use crate::{errors::ApiError, extractors::AuthUser, handlers::auth::to_user_response, state::AppState};
+
+pub async fn get_user(State(s): State<AppState>, Path(username): Path<String>) -> Result<Json<UserResponse>, ApiError> {
+    let user = get_user_by_username(&*s.users, &username).await?;
+    Ok(Json(to_user_response(&user)))
+}
+
+pub async fn patch_profile(State(s): State<AppState>, AuthUser(uid): AuthUser, Json(body): Json<UpdateProfileRequest>) -> Result<Json<UserResponse>, ApiError> {
+    update_profile(&*s.users, &uid, body.display_name, body.bio, body.avatar_url, body.header_url, body.custom_css).await?;
+    let user = s.users.find_by_id(&uid).await?.ok_or(domain::errors::DomainError::NotFound)?;
+    Ok(Json(to_user_response(&user)))
+}

crates/presentation/src/lib.rs

@@ -0,0 +1,41 @@
+pub mod errors;
+pub mod extractors;
+pub mod handlers;
+pub mod routes;
+pub mod state;
+
+use std::sync::Arc;
+use sqlx::PgPool;
+use state::AppState;
+
+use async_trait::async_trait;
+use domain::{errors::DomainError, events::DomainEvent, ports::EventPublisher};
+
+struct NoOpEventPublisher;
+
+#[async_trait]
+impl EventPublisher for NoOpEventPublisher {
+    async fn publish(&self, _e: &DomainEvent) -> Result<(), DomainError> {
+        Ok(())
+    }
+}
+
+pub fn build_state(pool: PgPool, jwt_secret: String) -> AppState {
+    AppState {
+        users:         Arc::new(postgres::user::PgUserRepository::new(pool.clone())),
+        thoughts:      Arc::new(postgres::thought::PgThoughtRepository::new(pool.clone())),
+        likes:         Arc::new(postgres::like::PgLikeRepository::new(pool.clone())),
+        boosts:        Arc::new(postgres::boost::PgBoostRepository::new(pool.clone())),
+        follows:       Arc::new(postgres::follow::PgFollowRepository::new(pool.clone())),
+        blocks:        Arc::new(postgres::block::PgBlockRepository::new(pool.clone())),
+        tags:          Arc::new(postgres::tag::PgTagRepository::new(pool.clone())),
+        api_keys:      Arc::new(postgres::api_key::PgApiKeyRepository::new(pool.clone())),
+        top_friends:   Arc::new(postgres::top_friend::PgTopFriendRepository::new(pool.clone())),
+        notifications: Arc::new(postgres::notification::PgNotificationRepository::new(pool.clone())),
+        remote_actors: Arc::new(postgres::remote_actor::PgRemoteActorRepository::new(pool.clone())),
+        feed:          Arc::new(postgres::feed::PgFeedRepository::new(pool.clone())),
+        auth:          Arc::new(auth::JwtAuthService::new(jwt_secret, 86400 * 30)),
+        hasher:        Arc::new(auth::Argon2PasswordHasher),
+        events:        Arc::new(NoOpEventPublisher),
+    }
+}

crates/presentation/src/main.rs

@@ -1 +1,28 @@
-fn main() {}
+use sqlx::PgPool;
+use tower_http::cors::CorsLayer;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() {
+    dotenvy::dotenv().ok();
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .init();
+
+    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL required");
+    let jwt_secret   = std::env::var("JWT_SECRET").expect("JWT_SECRET required");
+    let port         = std::env::var("PORT").unwrap_or_else(|_| "3000".into());
+
+    let pool = PgPool::connect(&database_url).await.expect("DB connect failed");
+    sqlx::migrate!("../adapters/postgres/migrations").run(&pool).await.expect("Migrations failed");
+
+    let state = presentation::build_state(pool, jwt_secret);
+    let app = presentation::routes::router()
+        .with_state(state)
+        .layer(CorsLayer::permissive());
+
+    let addr = format!("0.0.0.0:{port}");
+    tracing::info!("Listening on {addr}");
+    let listener = tokio::net::TcpListener::bind(&addr).await.unwrap();
+    axum::serve(listener, app).await.unwrap();
+}

crates/presentation/src/routes.rs

@@ -0,0 +1,60 @@
+use axum::{
+    routing::{delete, get, patch, post, put},
+    Router,
+};
+use crate::{handlers::*, state::AppState};
+
+pub fn router() -> Router<AppState> {
+    Router::new()
+        // auth
+        .route("/auth/register", post(auth::post_register))
+        .route("/auth/login", post(auth::post_login))
+        // users — static paths before parameterised
+        .route("/users/me", patch(users::patch_profile))
+        .route("/users/me/top-friends", put(social::put_top_friends))
+        .route("/users/{username}", get(users::get_user))
+        .route("/users/{username}/following", get(feed::get_following_handler))
+        .route("/users/{username}/followers", get(feed::get_followers_handler))
+        .route("/users/{username}/top-friends", get(social::get_top_friends_handler))
+        // follows & blocks (use {id} param)
+        .route(
+            "/users/{id}/follow",
+            post(social::post_follow).delete(social::delete_follow),
+        )
+        .route(
+            "/users/{id}/block",
+            post(social::post_block).delete(social::delete_block),
+        )
+        // thoughts
+        .route("/thoughts", post(thoughts::post_thought))
+        .route(
+            "/thoughts/{id}",
+            get(thoughts::get_thought_handler)
+                .patch(thoughts::patch_thought)
+                .delete(thoughts::delete_thought_handler),
+        )
+        .route("/thoughts/{id}/thread", get(thoughts::get_thread_handler))
+        // likes & boosts
+        .route(
+            "/thoughts/{id}/like",
+            post(social::post_like).delete(social::delete_like),
+        )
+        .route(
+            "/thoughts/{id}/boost",
+            post(social::post_boost).delete(social::delete_boost),
+        )
+        // feeds
+        .route("/feed", get(feed::home_feed))
+        .route("/feed/public", get(feed::public_feed))
+        .route("/search", get(feed::search_handler))
+        // notifications
+        .route("/notifications", get(notifications::list_notifications))
+        .route("/notifications/read-all", post(notifications::mark_all_read))
+        .route("/notifications/{id}/read", post(notifications::mark_notification_read))
+        // api keys
+        .route(
+            "/api-keys",
+            get(api_keys::get_api_keys).post(api_keys::post_api_key),
+        )
+        .route("/api-keys/{id}", delete(api_keys::delete_api_key_handler))
+}

crates/presentation/src/state.rs

@@ -0,0 +1,21 @@
+use std::sync::Arc;
+use domain::ports::*;
+
+#[derive(Clone)]
+pub struct AppState {
+    pub users:         Arc<dyn UserRepository>,
+    pub thoughts:      Arc<dyn ThoughtRepository>,
+    pub likes:         Arc<dyn LikeRepository>,
+    pub boosts:        Arc<dyn BoostRepository>,
+    pub follows:       Arc<dyn FollowRepository>,
+    pub blocks:        Arc<dyn BlockRepository>,
+    pub tags:          Arc<dyn TagRepository>,
+    pub api_keys:      Arc<dyn ApiKeyRepository>,
+    pub top_friends:   Arc<dyn TopFriendRepository>,
+    pub notifications: Arc<dyn NotificationRepository>,
+    pub remote_actors: Arc<dyn RemoteActorRepository>,
+    pub feed:          Arc<dyn FeedRepository>,
+    pub auth:          Arc<dyn AuthService>,
+    pub hasher:        Arc<dyn PasswordHasher>,
+    pub events:        Arc<dyn EventPublisher>,
+}