thoughts/crates/adapters/postgres/src/notification.rs

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
    errors::DomainError,
    models::{
        feed::{PageParams, Paginated},
        notification::{Notification, NotificationType},
    },
    ports::NotificationRepository,
    value_objects::{NotificationId, ThoughtId, UserId},
};
use sqlx::PgPool;

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 crate::user::PgUserRepository;
    use chrono::Utc;
    use domain::ports::UserRepository;
    use domain::{
        models::{notification::NotificationType, user::User},
        value_objects::*,
    };

    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);
    }
}