feat: implement unread notification count and enhance user listing with pagination

crates/adapters/postgres/src/notification.rs

@@ -100,6 +100,17 @@ impl NotificationRepository for PgNotificationRepository {
         })
     }
 
+    async fn count_unread(&self, user_id: &UserId) -> Result<u64, DomainError> {
+        let count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM notifications WHERE user_id=$1 AND read=false",
+        )
+        .bind(user_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(count as u64)
+    }
+
     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())

crates/application/src/use_cases/federation_management.rs

@@ -1,8 +1,20 @@
 use domain::{
-    errors::DomainError, models::remote_actor::RemoteActor, ports::FederationActionPort,
+    errors::DomainError,
+    events::DomainEvent,
+    models::{
+        actor_connection_summary::ActorConnectionSummary,
+        feed::{FeedEntry, PageParams, Paginated},
+        remote_actor::RemoteActor,
+    },
+    ports::{
+        ActivityPubRepository, EventPublisher, FederationActionPort, FeedRepository,
+        FollowRepository, RemoteActorConnectionRepository, UserRepository,
+    },
     value_objects::UserId,
 };
 
+use super::social;
+
 pub async fn list_pending_requests(
     federation: &dyn FederationActionPort,
     user_id: &UserId,
@@ -48,6 +60,87 @@ pub async fn list_remote_following(
     federation.get_remote_following(user_id).await
 }
 
+pub async fn remove_remote_following(
+    follows: &dyn FollowRepository,
+    users: &dyn UserRepository,
+    federation: &dyn FederationActionPort,
+    events: &dyn EventPublisher,
+    user_id: &UserId,
+    handle: &str,
+) -> Result<(), DomainError> {
+    social::unfollow_actor(follows, users, federation, events, user_id, handle).await
+}
+
+pub async fn get_remote_actor_posts(
+    federation: &dyn FederationActionPort,
+    ap_repo: &dyn ActivityPubRepository,
+    feed: &dyn FeedRepository,
+    events: &dyn EventPublisher,
+    handle: &str,
+    page: PageParams,
+    viewer_id: Option<&UserId>,
+) -> Result<Paginated<FeedEntry>, DomainError> {
+    let actor = federation.lookup_actor(handle).await?;
+    let ap_url = url::Url::parse(&actor.url).map_err(|e| DomainError::Internal(e.to_string()))?;
+    let author_id = match ap_repo.find_remote_actor_id(&ap_url).await? {
+        Some(id) => id,
+        None => ap_repo.intern_remote_actor(&ap_url).await?,
+    };
+    let result = feed.user_feed(&author_id, &page, viewer_id).await?;
+    if let Some(outbox_url) = actor.outbox_url {
+        let _ = events
+            .publish(&DomainEvent::FetchRemoteActorPosts {
+                actor_ap_url: actor.url,
+                outbox_url,
+            })
+            .await;
+    }
+    Ok(result)
+}
+
+const ACTOR_CONNECTIONS_CACHE_TTL_SECS: i64 = 3600;
+
+pub async fn get_actor_connections_page(
+    federation: &dyn FederationActionPort,
+    connections: &dyn RemoteActorConnectionRepository,
+    events: &dyn EventPublisher,
+    handle: &str,
+    connection_type: &str,
+    page: u32,
+) -> Result<(Vec<ActorConnectionSummary>, bool), DomainError> {
+    const PAGE_SIZE: usize = 20;
+    let actor = federation.lookup_actor(handle).await?;
+    let collection_url = match connection_type {
+        "followers" => actor.followers_url.ok_or(DomainError::NotFound)?,
+        _ => actor.following_url.ok_or(DomainError::NotFound)?,
+    };
+    let items = connections
+        .list_connections(&actor.url, connection_type, page)
+        .await?;
+    let stale = match connections
+        .connection_page_age(&actor.url, connection_type, page)
+        .await?
+    {
+        None => true,
+        Some(age) => {
+            chrono::Utc::now().signed_duration_since(age).num_seconds()
+                > ACTOR_CONNECTIONS_CACHE_TTL_SECS
+        }
+    };
+    if stale {
+        let _ = events
+            .publish(&DomainEvent::FetchActorConnections {
+                actor_ap_url: actor.url,
+                collection_url,
+                connection_type: connection_type.to_string(),
+                page,
+            })
+            .await;
+    }
+    let has_more = items.len() >= PAGE_SIZE;
+    Ok((items, has_more))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/application/src/use_cases/feed.rs

@@ -74,6 +74,26 @@ pub async fn list_users(users: &dyn UserRepository) -> Result<Vec<UserSummary>,
     users.list_with_stats().await
 }
 
+pub async fn list_users_paginated(
+    users: &dyn UserRepository,
+    page: PageParams,
+) -> Result<Paginated<UserSummary>, DomainError> {
+    let all = users.list_with_stats().await?;
+    let total = all.len() as i64;
+    let start = ((page.page.saturating_sub(1)) * page.per_page) as usize;
+    let items: Vec<UserSummary> = all
+        .into_iter()
+        .skip(start)
+        .take(page.per_page as usize)
+        .collect();
+    Ok(Paginated {
+        items,
+        total,
+        page: page.page,
+        per_page: page.per_page,
+    })
+}
+
 pub async fn get_popular_tags(
     tags: &dyn TagRepository,
     limit: usize,

crates/application/src/use_cases/notifications.rs

@@ -14,17 +14,34 @@ pub async fn list_notifications(
     repo.list_for_user(user_id, &page).await
 }
 
+pub async fn count_unread_notifications(
+    repo: &dyn NotificationRepository,
+    user_id: &UserId,
+) -> Result<u64, DomainError> {
+    repo.count_unread(user_id).await
+}
+
 pub async fn mark_notification_read(
     repo: &dyn NotificationRepository,
     id: &NotificationId,
     user_id: &UserId,
+    is_read: bool,
 ) -> Result<(), DomainError> {
-    repo.mark_read(id, user_id).await
+    if is_read {
+        repo.mark_read(id, user_id).await
+    } else {
+        Ok(())
+    }
 }
 
 pub async fn mark_all_notifications_read(
     repo: &dyn NotificationRepository,
     user_id: &UserId,
+    is_read: bool,
 ) -> Result<(), DomainError> {
-    repo.mark_all_read(user_id).await
+    if is_read {
+        repo.mark_all_read(user_id).await
+    } else {
+        Ok(())
+    }
 }

crates/application/src/use_cases/social.rs

@@ -210,6 +210,36 @@ pub async fn reject_follow(
     Ok(())
 }
 
+pub async fn block_by_username(
+    blocks: &dyn BlockRepository,
+    users: &dyn UserRepository,
+    events: &dyn EventPublisher,
+    blocker_id: &UserId,
+    username: &str,
+) -> Result<(), DomainError> {
+    let uname = Username::new(username).map_err(|_| DomainError::NotFound)?;
+    let target = users
+        .find_by_username(&uname)
+        .await?
+        .ok_or(DomainError::NotFound)?;
+    block_user(blocks, events, blocker_id, &target.id).await
+}
+
+pub async fn unblock_by_username(
+    blocks: &dyn BlockRepository,
+    users: &dyn UserRepository,
+    events: &dyn EventPublisher,
+    blocker_id: &UserId,
+    username: &str,
+) -> Result<(), DomainError> {
+    let uname = Username::new(username).map_err(|_| DomainError::NotFound)?;
+    let target = users
+        .find_by_username(&uname)
+        .await?
+        .ok_or(DomainError::NotFound)?;
+    unblock_user(blocks, events, blocker_id, &target.id).await
+}
+
 pub async fn block_user(
     blocks: &dyn BlockRepository,
     events: &dyn EventPublisher,

crates/domain/src/models/thought.rs

@@ -25,6 +25,17 @@ pub struct Thought {
     pub updated_at: Option<DateTime<Utc>>,
 }
 
+impl Visibility {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Visibility::Public => "public",
+            Visibility::Followers => "followers",
+            Visibility::Unlisted => "unlisted",
+            Visibility::Direct => "direct",
+        }
+    }
+}
+
 impl Thought {
     pub fn new_local(
         id: ThoughtId,

crates/domain/src/ports.rs

@@ -184,6 +184,7 @@ pub trait NotificationRepository: Send + Sync {
         user_id: &UserId,
         page: &PageParams,
     ) -> Result<Paginated<Notification>, DomainError>;
+    async fn count_unread(&self, user_id: &UserId) -> Result<u64, DomainError>;
     async fn mark_read(&self, id: &NotificationId, user_id: &UserId) -> Result<(), DomainError>;
     async fn mark_all_read(&self, user_id: &UserId) -> Result<(), DomainError>;
 }

crates/domain/src/testing.rs

@@ -498,6 +498,15 @@ impl NotificationRepository for TestStore {
             per_page: 20,
         })
     }
+    async fn count_unread(&self, uid: &UserId) -> Result<u64, DomainError> {
+        Ok(self
+            .notifications
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|n| &n.user_id == uid && !n.read)
+            .count() as u64)
+    }
     async fn mark_read(&self, id: &NotificationId, _uid: &UserId) -> Result<(), DomainError> {
         if let Some(n) = self
             .notifications

crates/presentation/src/handlers/federation_actors.rs

@@ -6,12 +6,14 @@ use api_types::{
     requests::PaginationQuery,
     responses::{ActorConnectionPageResponse, ActorConnectionResponse},
 };
-use application::use_cases::feed::get_user_feed;
+use application::use_cases::federation_management::{
+    get_actor_connections_page, get_remote_actor_posts,
+};
 use axum::{
     extract::{Path, Query, State},
     Json,
 };
-use domain::{events::DomainEvent, models::feed::PageParams};
+use domain::models::feed::PageParams;
 
 pub async fn remote_actor_posts_handler(
     State(s): State<AppState>,
@@ -19,53 +21,20 @@ pub async fn remote_actor_posts_handler(
     Query(q): Query<PaginationQuery>,
     OptionalAuthUser(viewer): OptionalAuthUser,
 ) -> Result<Json<serde_json::Value>, ApiError> {
-    tracing::info!(%handle, "remote_actor_posts: looking up actor");
-    let actor = s.federation.lookup_actor(&handle).await?;
-    tracing::info!(actor_url = %actor.url, has_outbox = actor.outbox_url.is_some(), "remote_actor_posts: actor found");
-
-    let ap_url = url::Url::parse(&actor.url).map_err(|e| ApiError::BadRequest(e.to_string()))?;
-
-    let author_id = match s.ap_repo.find_remote_actor_id(&ap_url).await? {
-        Some(id) => {
-            tracing::info!(?id, "remote_actor_posts: actor already interned");
-            id
-        }
-        None => {
-            tracing::info!("remote_actor_posts: interning actor");
-            let id = s.ap_repo.intern_remote_actor(&ap_url).await?;
-            tracing::info!(?id, "remote_actor_posts: actor interned");
-            id
-        }
-    };
-
     let page = PageParams {
         page: q.page(),
         per_page: q.per_page(),
     };
-    let result = get_user_feed(&*s.feed, &author_id, page, viewer.as_ref()).await?;
-    tracing::info!(
-        post_count = result.items.len(),
-        "remote_actor_posts: cached posts fetched"
-    );
-
-    match &actor.outbox_url {
-        Some(outbox_url) => {
-            tracing::info!(%outbox_url, "remote_actor_posts: publishing FetchRemoteActorPosts");
-            match s
-                .events
-                .publish(&DomainEvent::FetchRemoteActorPosts {
-                    actor_ap_url: actor.url.clone(),
-                    outbox_url: outbox_url.clone(),
-                })
-                .await
-            {
-                Ok(_) => tracing::info!("remote_actor_posts: event published"),
-                Err(e) => tracing::warn!("remote_actor_posts: event publish failed: {e}"),
-            }
-        }
-        None => tracing::warn!("remote_actor_posts: actor has no outbox_url, skipping fetch"),
-    }
-
+    let result = get_remote_actor_posts(
+        &*s.federation,
+        &*s.ap_repo,
+        &*s.feed,
+        &*s.events,
+        &handle,
+        page,
+        viewer.as_ref(),
+    )
+    .await?;
     Ok(Json(serde_json::json!({
         "total": result.total,
         "page": result.page,
@@ -74,8 +43,6 @@ pub async fn remote_actor_posts_handler(
     })))
 }
 
-const CACHE_TTL_SECS: i64 = 3600;
-
 pub async fn actor_followers_handler(
     State(s): State<AppState>,
     Path(handle): Path<String>,
@@ -98,46 +65,15 @@ async fn actor_connections_handler(
     connection_type: &str,
     page: u32,
 ) -> Result<Json<ActorConnectionPageResponse>, ApiError> {
-    const PAGE_SIZE: usize = 20;
-
-    let actor = s.federation.lookup_actor(&handle).await?;
-
-    let collection_url = match connection_type {
-        "followers" => actor
-            .followers_url
-            .ok_or_else(|| ApiError::BadRequest("actor has no followers URL".into()))?,
-        _ => actor
-            .following_url
-            .ok_or_else(|| ApiError::BadRequest("actor has no following URL".into()))?,
-    };
-
-    let items = s
-        .remote_actor_connections
-        .list_connections(&actor.url, connection_type, page)
-        .await?;
-
-    let stale = match s
-        .remote_actor_connections
-        .connection_page_age(&actor.url, connection_type, page)
-        .await?
-    {
-        None => true,
-        Some(age) => chrono::Utc::now().signed_duration_since(age).num_seconds() > CACHE_TTL_SECS,
-    };
-
-    if stale {
-        let _ = s
-            .events
-            .publish(&DomainEvent::FetchActorConnections {
-                actor_ap_url: actor.url.clone(),
-                collection_url,
-                connection_type: connection_type.to_string(),
-                page,
-            })
-            .await;
-    }
-
-    let has_more = items.len() >= PAGE_SIZE;
+    let (items, has_more) = get_actor_connections_page(
+        &*s.federation,
+        &*s.remote_actor_connections,
+        &*s.events,
+        &handle,
+        connection_type,
+        page,
+    )
+    .await?;
     Ok(Json(ActorConnectionPageResponse {
         items: items
             .into_iter()

crates/presentation/src/handlers/federation_management.rs

@@ -1,8 +1,8 @@
 use crate::{errors::ApiError, extractors::AuthUser, state::AppState};
-use api_types::responses::RemoteActorResponse;
+use api_types::responses::{ProfileField, RemoteActorResponse};
 use application::use_cases::federation_management::{
     accept_follow_request, list_pending_requests, list_remote_followers, list_remote_following,
-    reject_follow_request,
+    reject_follow_request, remove_remote_following,
 };
 use axum::{extract::State, http::StatusCode, Json};
 use serde::Deserialize;
@@ -29,7 +29,11 @@ fn to_response(a: domain::models::remote_actor::RemoteActor) -> RemoteActorRespo
         outbox_url: a.outbox_url,
         followers_url: a.followers_url,
         following_url: a.following_url,
-        attachment: vec![],
+        attachment: a
+            .attachment
+            .into_iter()
+            .map(|(name, value)| ProfileField { name, value })
+            .collect(),
     }
 }
 
@@ -80,7 +84,7 @@ pub async fn delete_following(
     AuthUser(uid): AuthUser,
     Json(body): Json<HandleBody>,
 ) -> Result<StatusCode, ApiError> {
-    application::use_cases::social::unfollow_actor(
+    remove_remote_following(
         &*s.follows,
         &*s.users,
         &*s.federation,

crates/presentation/src/handlers/feed.rs

@@ -10,7 +10,7 @@ use application::use_cases::feed::{
     get_by_tag, get_followers, get_following, get_home_feed,
     get_popular_tags as uc_get_popular_tags, get_public_feed, get_user_feed,
 };
-use application::use_cases::profile::get_user_by_username;
+use application::use_cases::profile::{get_user_by_id_or_username, get_user_by_username};
 use application::use_cases::search::{search_thoughts, search_users};
 use axum::{
     extract::{Path, Query, State},
@@ -19,17 +19,6 @@ use axum::{
     Json,
 };
 use domain::models::feed::PageParams;
-use domain::value_objects::UserId;
-
-fn visibility_as_str(v: &domain::models::thought::Visibility) -> &'static str {
-    use domain::models::thought::Visibility;
-    match v {
-        Visibility::Public => "public",
-        Visibility::Followers => "followers",
-        Visibility::Unlisted => "unlisted",
-        Visibility::Direct => "direct",
-    }
-}
 
 pub fn to_thought_response(e: &domain::models::feed::FeedEntry) -> ThoughtResponse {
     ThoughtResponse {
@@ -38,7 +27,7 @@ pub fn to_thought_response(e: &domain::models::feed::FeedEntry) -> ThoughtRespon
         author: to_user_response(&e.author),
         in_reply_to_id: e.thought.in_reply_to_id.as_ref().map(|id| id.as_uuid()),
         in_reply_to_url: e.thought.in_reply_to_url.clone(),
-        visibility: visibility_as_str(&e.thought.visibility).to_string(),
+        visibility: e.thought.visibility.as_str().to_string(),
         content_warning: e.thought.content_warning.clone(),
         sensitive: e.thought.sensitive,
         like_count: e.like_count,
@@ -175,7 +164,8 @@ pub async fn get_following_handler(
         .unwrap_or("");
 
     if accept.contains("application/activity+json") {
-        let user_id = resolve_user_id(&s, &param).await?;
+        let user = get_user_by_id_or_username(&*s.users, &param).await?;
+        let user_id = user.id;
         let page = q.page().try_into().ok();
         let json = s
             .federation
@@ -209,7 +199,8 @@ pub async fn get_followers_handler(
         .unwrap_or("");
 
     if accept.contains("application/activity+json") {
-        let user_id = resolve_user_id(&s, &param).await?;
+        let user = get_user_by_id_or_username(&*s.users, &param).await?;
+        let user_id = user.id;
         let page = q.page().try_into().ok();
         let json = s
             .federation
@@ -231,18 +222,6 @@ pub async fn get_followers_handler(
     .into_response())
 }
 
-async fn resolve_user_id(s: &AppState, param: &str) -> Result<UserId, ApiError> {
-    if let Ok(uuid) = uuid::Uuid::parse_str(param) {
-        s.users
-            .find_by_id(&UserId::from_uuid(uuid))
-            .await?
-            .map(|u| u.id)
-            .ok_or_else(|| ApiError::from(domain::errors::DomainError::NotFound))
-    } else {
-        Ok(get_user_by_username(&*s.users, param).await?.id)
-    }
-}
-
 #[utoipa::path(
     get, path = "/users/{username}/thoughts",
     params(

crates/presentation/src/handlers/notifications.rs

@@ -1,8 +1,8 @@
 use crate::{errors::ApiError, extractors::AuthUser, state::AppState};
 use api_types::requests::NotificationUpdateRequest;
 use application::use_cases::notifications::{
-    list_notifications as uc_list_notifications, mark_all_notifications_read,
-    mark_notification_read as uc_mark_notification_read,
+    count_unread_notifications, list_notifications as uc_list_notifications,
+    mark_all_notifications_read, mark_notification_read as uc_mark_notification_read,
 };
 use axum::{
     extract::{Path, State},
@@ -22,9 +22,10 @@ pub async fn list_notifications(
         per_page: 20,
     };
     let result = uc_list_notifications(&*s.notifications, &uid, page).await?;
+    let unread = count_unread_notifications(&*s.notifications, &uid).await?;
     Ok(Json(serde_json::json!({
         "total": result.total,
-        "unread": result.items.iter().filter(|n| !n.read).count()
+        "unread": unread
     })))
 }
 
@@ -35,9 +36,13 @@ pub async fn mark_notification_read(
     Path(id): Path<Uuid>,
     Json(body): Json<NotificationUpdateRequest>,
 ) -> Result<StatusCode, ApiError> {
-    if body.read {
-        uc_mark_notification_read(&*s.notifications, &NotificationId::from_uuid(id), &uid).await?;
-    }
+    uc_mark_notification_read(
+        &*s.notifications,
+        &NotificationId::from_uuid(id),
+        &uid,
+        body.read,
+    )
+    .await?;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -47,9 +52,7 @@ pub async fn mark_all_read(
     AuthUser(uid): AuthUser,
     Json(body): Json<NotificationUpdateRequest>,
 ) -> Result<StatusCode, ApiError> {
-    if body.read {
-        mark_all_notifications_read(&*s.notifications, &uid).await?;
-    }
+    mark_all_notifications_read(&*s.notifications, &uid, body.read).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 

crates/presentation/src/handlers/social.rs

@@ -96,8 +96,7 @@ pub async fn post_block(
     AuthUser(uid): AuthUser,
     Path(username): Path<String>,
 ) -> Result<StatusCode, ApiError> {
-    let target = get_user_by_username(&*s.users, &username).await?;
-    block_user(&*s.blocks, &*s.events, &uid, &target.id).await?;
+    block_by_username(&*s.blocks, &*s.users, &*s.events, &uid, &username).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 #[utoipa::path(delete, path = "/users/{username}/block", params(("username" = String, Path, description = "Username")), responses((status = 204, description = "Unblocked")), security(("bearer_auth" = [])))]
@@ -106,8 +105,7 @@ pub async fn delete_block(
     AuthUser(uid): AuthUser,
     Path(username): Path<String>,
 ) -> Result<StatusCode, ApiError> {
-    let target = get_user_by_username(&*s.users, &username).await?;
-    unblock_user(&*s.blocks, &*s.events, &uid, &target.id).await?;
+    unblock_by_username(&*s.blocks, &*s.users, &*s.events, &uid, &username).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 #[utoipa::path(put, path = "/users/me/top-friends", request_body = SetTopFriendsRequest, responses((status = 204, description = "Top friends updated")), security(("bearer_auth" = [])))]

crates/presentation/src/handlers/thoughts.rs

@@ -20,16 +20,6 @@ use axum::{
 use domain::value_objects::ThoughtId;
 use uuid::Uuid;
 
-fn visibility_as_str(v: &domain::models::thought::Visibility) -> &'static str {
-    use domain::models::thought::Visibility;
-    match v {
-        Visibility::Public => "public",
-        Visibility::Followers => "followers",
-        Visibility::Unlisted => "unlisted",
-        Visibility::Direct => "direct",
-    }
-}
-
 fn thought_to_json(
     t: &domain::models::thought::Thought,
     author: &domain::models::user::User,
@@ -42,7 +32,7 @@ fn thought_to_json(
         "content": t.content.as_str(),
         "author": to_user_response(author),
         "replyToId": t.in_reply_to_id.as_ref().map(|x| x.as_uuid()),
-        "visibility": visibility_as_str(&t.visibility),
+        "visibility": t.visibility.as_str(),
         "contentWarning": t.content_warning,
         "sensitive": t.sensitive,
         "likeCount": like_count,

crates/presentation/src/handlers/users.rs

@@ -8,7 +8,7 @@ use api_types::{
     requests::{PaginationQuery, UpdateProfileRequest},
     responses::{ErrorResponse, ProfileField, RemoteActorResponse, UserResponse},
 };
-use application::use_cases::feed::list_users;
+use application::use_cases::feed::list_users_paginated;
 use application::use_cases::profile::{
     get_user as fetch_user, get_user_by_id_or_username, update_profile,
 };
@@ -146,13 +146,10 @@ pub async fn get_users(
         })));
     }
 
-    let all = list_users(&*s.users).await?;
-    let total = all.len() as i64;
-    let start = ((page - 1) * per_page) as usize;
-    let items: Vec<_> = all
-        .into_iter()
-        .skip(start)
-        .take(per_page as usize)
+    let result = list_users_paginated(&*s.users, page_params).await?;
+    let items: Vec<_> = result
+        .items
+        .iter()
         .map(|u| {
             serde_json::json!({
                 "id": u.id.as_uuid(),
@@ -169,7 +166,7 @@ pub async fn get_users(
         })
         .collect();
     Ok(Json(serde_json::json!({
-        "items": items, "total": total, "page": page, "per_page": per_page
+        "items": items, "total": result.total, "page": result.page, "per_page": result.per_page
     })))
 }