refactor: sqlite repositories per aggregate

crates/adapters/sqlite/src/repositories/diary/mod.rs

@@ -0,0 +1,174 @@
+use domain::{errors::DomainError, models::ReviewSortBy};
+use sqlx::SqlitePool;
+
+use rows::DiaryRow;
+
+mod query;
+mod rows;
+
+pub struct SqliteDiaryRepository {
+    pool: SqlitePool,
+}
+
+impl SqliteDiaryRepository {
+    pub fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    async fn count_diary_entries(&self, movie_id: Option<&str>) -> Result<i64, DomainError> {
+        match movie_id {
+            None => sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reviews")
+                .fetch_one(&self.pool)
+                .await
+                .map_err(crate::shared::map_sqlx_error),
+            Some(id) => {
+                sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reviews WHERE movie_id = ?")
+                    .bind(id)
+                    .fetch_one(&self.pool)
+                    .await
+                    .map_err(crate::shared::map_sqlx_error)
+            }
+        }
+    }
+
+    async fn fetch_all_diary_rows(
+        &self,
+        sort: &ReviewSortBy,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<DiaryRow>, DomainError> {
+        let order_clause = match sort {
+            ReviewSortBy::ByRatingDesc => "r.rating DESC, r.watched_at DESC",
+            ReviewSortBy::ByRatingAsc => "r.rating ASC, r.watched_at ASC",
+            ReviewSortBy::Ascending => "r.watched_at ASC",
+            ReviewSortBy::Descending => "r.watched_at DESC",
+        };
+        let sql = format!(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             ORDER BY {}
+             LIMIT ? OFFSET ?",
+            order_clause
+        );
+        sqlx::query_as::<_, DiaryRow>(&sql)
+            .bind(limit)
+            .bind(offset)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)
+    }
+
+    async fn fetch_movie_diary_rows(
+        &self,
+        movie_id: &str,
+        sort: &ReviewSortBy,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<DiaryRow>, DomainError> {
+        let order_clause = match sort {
+            ReviewSortBy::ByRatingDesc => "r.rating DESC, r.watched_at DESC",
+            ReviewSortBy::ByRatingAsc => "r.rating ASC, r.watched_at ASC",
+            ReviewSortBy::Ascending => "r.watched_at ASC",
+            ReviewSortBy::Descending => "r.watched_at DESC",
+        };
+        let sql = format!(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             WHERE r.movie_id = ?
+             ORDER BY {}
+             LIMIT ? OFFSET ?",
+            order_clause
+        );
+        sqlx::query_as::<_, DiaryRow>(&sql)
+            .bind(movie_id)
+            .bind(limit)
+            .bind(offset)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)
+    }
+
+    async fn count_user_diary_entries(
+        &self,
+        user_id: &str,
+        search: Option<&str>,
+        include_remote: bool,
+    ) -> Result<i64, DomainError> {
+        let has_search = search.map(|s| !s.is_empty()).unwrap_or(false);
+        let remote_clause = if include_remote {
+            ""
+        } else {
+            " AND r.remote_actor_url IS NULL"
+        };
+        let search_clause = if has_search {
+            " AND m.title LIKE '%' || ? || '%'"
+        } else {
+            ""
+        };
+        let sql = format!(
+            "SELECT COUNT(*) FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             WHERE r.user_id = ?{remote_clause}{search_clause}"
+        );
+        let mut q = sqlx::query_scalar::<_, i64>(&sql).bind(user_id);
+        if has_search {
+            q = q.bind(search.unwrap());
+        }
+        q.fetch_one(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)
+    }
+
+    async fn fetch_user_diary_rows(
+        &self,
+        user_id: &str,
+        sort: &ReviewSortBy,
+        search: Option<&str>,
+        include_remote: bool,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<DiaryRow>, DomainError> {
+        let has_search = search.map(|s| !s.is_empty()).unwrap_or(false);
+        let remote_clause = if include_remote {
+            ""
+        } else {
+            " AND r.remote_actor_url IS NULL"
+        };
+        let search_clause = if has_search {
+            " AND m.title LIKE '%' || ? || '%'"
+        } else {
+            ""
+        };
+        let order_clause = match sort {
+            ReviewSortBy::ByRatingDesc => "r.rating DESC, r.watched_at DESC",
+            ReviewSortBy::ByRatingAsc => "r.rating ASC, r.watched_at ASC",
+            ReviewSortBy::Ascending => "r.watched_at ASC",
+            ReviewSortBy::Descending => "r.watched_at DESC",
+        };
+        let sql = format!(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             WHERE r.user_id = ?{remote_clause}{search_clause}
+             ORDER BY {order_clause}
+             LIMIT ? OFFSET ?",
+        );
+        let mut q = sqlx::query_as::<_, DiaryRow>(&sql).bind(user_id);
+        if has_search {
+            q = q.bind(search.unwrap());
+        }
+        q.bind(limit)
+            .bind(offset)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/adapters/sqlite/src/repositories/diary/query.rs

@@ -0,0 +1,356 @@
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    models::{
+        DiaryEntry, DiaryFilter, FeedEntry, MovieStats, ReviewHistory,
+        collections::{PageParams, Paginated},
+    },
+    ports::DiaryQuery,
+    value_objects::{MovieId, UserId},
+};
+use futures::stream::BoxStream;
+
+use super::SqliteDiaryRepository;
+use super::rows::{DiaryRow, FeedRow, MovieStatsRow};
+use crate::repositories::movie::rows::MovieRow;
+use crate::repositories::review::rows::ReviewRow;
+
+#[async_trait]
+impl DiaryQuery for SqliteDiaryRepository {
+    async fn query_diary(
+        &self,
+        filter: &DiaryFilter,
+    ) -> Result<Paginated<DiaryEntry>, DomainError> {
+        let limit = filter.page.limit as i64;
+        let offset = filter.page.offset as i64;
+
+        let (total, rows) = match (&filter.movie_id, &filter.user_id) {
+            (None, None) => tokio::try_join!(
+                self.count_diary_entries(None),
+                self.fetch_all_diary_rows(&filter.sort_by, limit, offset)
+            )?,
+            (Some(id), None) => {
+                let id_str = id.value().to_string();
+                tokio::try_join!(
+                    self.count_diary_entries(Some(id_str.as_str())),
+                    self.fetch_movie_diary_rows(&id_str, &filter.sort_by, limit, offset)
+                )?
+            }
+            (None, Some(uid)) => {
+                let uid_str = uid.value().to_string();
+                let search = filter.search.as_deref();
+                let inc = filter.include_remote;
+                tokio::try_join!(
+                    self.count_user_diary_entries(&uid_str, search, inc),
+                    self.fetch_user_diary_rows(
+                        &uid_str,
+                        &filter.sort_by,
+                        search,
+                        inc,
+                        limit,
+                        offset
+                    )
+                )?
+            }
+            (Some(_), Some(_)) => {
+                return Err(DomainError::ValidationError(
+                    "Combined movie_id + user_id filter not supported".into(),
+                ));
+            }
+        };
+
+        let items = rows
+            .into_iter()
+            .map(DiaryRow::into_domain)
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(Paginated {
+            items,
+            total_count: total as u64,
+            limit: filter.page.limit,
+            offset: filter.page.offset,
+        })
+    }
+
+    async fn query_activity_feed(
+        &self,
+        page: &PageParams,
+    ) -> Result<Paginated<FeedEntry>, DomainError> {
+        self.query_activity_feed_filtered(page, &domain::models::FeedSortBy::Date, None, None)
+            .await
+    }
+
+    async fn query_activity_feed_filtered(
+        &self,
+        page: &PageParams,
+        sort_by: &domain::models::FeedSortBy,
+        search: Option<&str>,
+        following: Option<&domain::models::FollowingFilter>,
+    ) -> Result<Paginated<FeedEntry>, DomainError> {
+        use domain::models::FeedSortBy;
+
+        let limit = page.limit as i64;
+        let offset = page.offset as i64;
+        let has_search = search.map(|s| !s.is_empty()).unwrap_or(false);
+
+        let mut where_parts = vec!["1=1".to_string()];
+
+        if has_search {
+            where_parts.push("m.title LIKE '%' || ? || '%'".to_string());
+        }
+
+        if let Some(f) = following {
+            let local_in = if f.local_user_ids.is_empty() {
+                "SELECT NULL WHERE 0".to_string()
+            } else {
+                f.local_user_ids
+                    .iter()
+                    .map(|_| "?")
+                    .collect::<Vec<_>>()
+                    .join(",")
+            };
+            let remote_in = if f.remote_actor_urls.is_empty() {
+                "SELECT NULL WHERE 0".to_string()
+            } else {
+                f.remote_actor_urls
+                    .iter()
+                    .map(|_| "?")
+                    .collect::<Vec<_>>()
+                    .join(",")
+            };
+            where_parts.push(format!(
+                "(r.user_id IN ({}) OR r.remote_actor_url IN ({}))",
+                local_in, remote_in
+            ));
+        }
+
+        let order_clause = match sort_by {
+            FeedSortBy::Date => "r.watched_at DESC",
+            FeedSortBy::DateAsc => "r.watched_at ASC",
+            FeedSortBy::Rating => "r.rating DESC, r.watched_at DESC",
+            FeedSortBy::RatingAsc => "r.rating ASC, r.watched_at ASC",
+        };
+
+        let where_clause = where_parts.join(" AND ");
+
+        let count_sql = format!(
+            "SELECT COUNT(*) FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             WHERE {}",
+            where_clause
+        );
+
+        let select_sql = format!(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment,
+                    r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers,
+                    CASE WHEN r.remote_actor_url IS NOT NULL THEN COALESCE(a.handle, r.remote_actor_url)
+                         ELSE COALESCE(u.email, r.user_id) END AS user_email
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             LEFT JOIN users u ON u.id = r.user_id
+             LEFT JOIN ap_remote_actors a ON a.url = r.remote_actor_url
+             WHERE {}
+             ORDER BY {}
+             LIMIT ? OFFSET ?",
+            where_clause, order_clause
+        );
+
+        macro_rules! bind_filter_params {
+            ($q:expr) => {{
+                let mut q = $q;
+                if has_search {
+                    q = q.bind(search.unwrap());
+                }
+                if let Some(f) = following {
+                    for uid in &f.local_user_ids {
+                        q = q.bind(uid.to_string());
+                    }
+                    for url in &f.remote_actor_urls {
+                        q = q.bind(url.as_str());
+                    }
+                }
+                q
+            }};
+        }
+
+        let count_q = bind_filter_params!(sqlx::query_scalar::<_, i64>(&count_sql));
+        let total = count_q
+            .fetch_one(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)?;
+
+        let rows_q = bind_filter_params!(sqlx::query_as::<_, FeedRow>(&select_sql));
+        let rows = rows_q
+            .bind(limit)
+            .bind(offset)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(crate::shared::map_sqlx_error)?;
+
+        let items = rows
+            .into_iter()
+            .map(FeedRow::into_domain)
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(Paginated {
+            items,
+            total_count: total as u64,
+            limit: page.limit,
+            offset: page.offset,
+        })
+    }
+
+    async fn get_review_history(&self, movie_id: &MovieId) -> Result<ReviewHistory, DomainError> {
+        let id_str = movie_id.value().to_string();
+
+        let movie = sqlx::query_as::<_, MovieRow>(
+            "SELECT id, external_metadata_id, title, release_year, director, poster_path
+             FROM movies WHERE id = ?",
+        )
+        .bind(&id_str)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?
+        .ok_or_else(|| DomainError::NotFound(format!("Movie {}", id_str)))?
+        .into_domain()?;
+
+        let viewings = sqlx::query_as::<_, ReviewRow>(
+            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url, watch_medium, spoilers
+             FROM reviews WHERE movie_id = ? ORDER BY watched_at ASC",
+        )
+        .bind(&id_str)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?
+        .into_iter()
+        .map(ReviewRow::into_domain)
+        .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(ReviewHistory::new(movie, viewings))
+    }
+
+    async fn get_user_history(&self, user_id: &UserId) -> Result<Vec<DiaryEntry>, DomainError> {
+        let uid = user_id.value().to_string();
+        let rows = sqlx::query_as::<_, DiaryRow>(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             WHERE r.user_id = ?
+             ORDER BY r.watched_at DESC",
+        )
+        .bind(&uid)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?;
+
+        rows.into_iter().map(DiaryRow::into_domain).collect()
+    }
+
+    fn stream_user_history(
+        &self,
+        user_id: UserId,
+    ) -> BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        let pool = self.pool.clone();
+        let uid = user_id.value().to_string();
+        Box::pin(async_stream::stream! {
+            let mut rows = sqlx::query_as::<_, DiaryRow>(
+                "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                        r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers
+                 FROM reviews r
+                 INNER JOIN movies m ON m.id = r.movie_id
+                 WHERE r.user_id = ?
+                 ORDER BY r.watched_at DESC",
+            )
+            .bind(&uid)
+            .fetch(&pool);
+            while let Some(row) = futures::StreamExt::next(&mut rows).await {
+                yield match row {
+                    Ok(r) => r.into_domain(),
+                    Err(e) => Err(crate::shared::map_sqlx_error(e)),
+                };
+            }
+        })
+    }
+
+    async fn get_movie_stats(&self, movie_id: &MovieId) -> Result<MovieStats, DomainError> {
+        let id_str = movie_id.value().to_string();
+        sqlx::query_as::<_, MovieStatsRow>(
+            "SELECT
+                COUNT(*) AS total_count,
+                AVG(CAST(rating AS REAL)) AS avg_rating,
+                COUNT(CASE WHEN remote_actor_url IS NOT NULL THEN 1 END) AS federated_count,
+                COUNT(CASE WHEN rating = 1 THEN 1 END) AS rating_1,
+                COUNT(CASE WHEN rating = 2 THEN 1 END) AS rating_2,
+                COUNT(CASE WHEN rating = 3 THEN 1 END) AS rating_3,
+                COUNT(CASE WHEN rating = 4 THEN 1 END) AS rating_4,
+                COUNT(CASE WHEN rating = 5 THEN 1 END) AS rating_5
+             FROM reviews WHERE movie_id = ?",
+        )
+        .bind(id_str)
+        .fetch_one(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)
+        .map(MovieStatsRow::into_domain)
+    }
+
+    async fn get_movie_social_feed(
+        &self,
+        movie_id: &MovieId,
+        page: &PageParams,
+    ) -> Result<Paginated<FeedEntry>, DomainError> {
+        let id_str = movie_id.value().to_string();
+        let limit = page.limit as i64;
+        let offset = page.offset as i64;
+
+        let total: i64 =
+            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reviews WHERE movie_id = ?")
+                .bind(&id_str)
+                .fetch_one(&self.pool)
+                .await
+                .map_err(crate::shared::map_sqlx_error)?;
+
+        let rows = sqlx::query_as::<_, FeedRow>(
+            "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                    r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment,
+                    r.watched_at, r.created_at, r.remote_actor_url, r.watch_medium, r.spoilers,
+                    CASE WHEN r.remote_actor_url IS NOT NULL THEN COALESCE(a.handle, r.remote_actor_url)
+                         ELSE COALESCE(u.email, r.user_id) END AS user_email
+             FROM reviews r
+             INNER JOIN movies m ON m.id = r.movie_id
+             LEFT JOIN users u ON u.id = r.user_id
+             LEFT JOIN ap_remote_actors a ON a.url = r.remote_actor_url
+             WHERE r.movie_id = ?
+             ORDER BY r.watched_at DESC
+             LIMIT ? OFFSET ?",
+        )
+        .bind(&id_str)
+        .bind(limit)
+        .bind(offset)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?;
+
+        let items = rows
+            .into_iter()
+            .map(FeedRow::into_domain)
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(Paginated {
+            items,
+            total_count: total as u64,
+            limit: page.limit,
+            offset: page.offset,
+        })
+    }
+
+    async fn count_local_posts(&self) -> Result<u64, DomainError> {
+        let count: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM reviews WHERE remote_actor_url IS NULL")
+                .fetch_one(&self.pool)
+                .await
+                .map_err(crate::shared::map_sqlx_error)?;
+        Ok(count as u64)
+    }
+}

crates/adapters/sqlite/src/repositories/diary/rows.rs

@@ -0,0 +1,128 @@
+use crate::shared::{movie_row_to_domain, movie_stats_to_domain, review_row_to_domain};
+use domain::{
+    errors::DomainError,
+    models::{DiaryEntry, FeedEntry},
+};
+
+#[derive(sqlx::FromRow)]
+pub(crate) struct DiaryRow {
+    pub id: String,
+    pub external_metadata_id: Option<String>,
+    pub title: String,
+    pub release_year: i64,
+    pub director: Option<String>,
+    pub poster_path: Option<String>,
+    pub review_id: String,
+    pub movie_id: String,
+    pub user_id: String,
+    pub rating: i64,
+    pub comment: Option<String>,
+    pub watched_at: String,
+    pub created_at: String,
+    pub remote_actor_url: Option<String>,
+    pub watch_medium: Option<String>,
+    pub spoilers: bool,
+}
+
+impl DiaryRow {
+    pub fn into_domain(self) -> Result<DiaryEntry, DomainError> {
+        let movie = movie_row_to_domain(
+            self.id,
+            self.external_metadata_id,
+            self.title,
+            self.release_year,
+            self.director,
+            self.poster_path,
+        )?;
+        let review = review_row_to_domain(
+            self.review_id,
+            self.movie_id,
+            self.user_id,
+            self.rating,
+            self.comment,
+            self.watched_at,
+            self.created_at,
+            self.remote_actor_url,
+            self.watch_medium,
+            self.spoilers,
+        )?;
+        Ok(DiaryEntry::new(movie, review))
+    }
+}
+
+#[derive(sqlx::FromRow)]
+pub(crate) struct FeedRow {
+    pub id: String,
+    pub external_metadata_id: Option<String>,
+    pub title: String,
+    pub release_year: i64,
+    pub director: Option<String>,
+    pub poster_path: Option<String>,
+    pub review_id: String,
+    pub movie_id: String,
+    pub user_id: String,
+    pub rating: i64,
+    pub comment: Option<String>,
+    pub watched_at: String,
+    pub created_at: String,
+    pub remote_actor_url: Option<String>,
+    pub watch_medium: Option<String>,
+    pub spoilers: bool,
+    pub user_email: String,
+}
+
+impl FeedRow {
+    pub fn into_domain(self) -> Result<FeedEntry, DomainError> {
+        let movie = movie_row_to_domain(
+            self.id,
+            self.external_metadata_id,
+            self.title,
+            self.release_year,
+            self.director,
+            self.poster_path,
+        )?;
+        let review = review_row_to_domain(
+            self.review_id,
+            self.movie_id,
+            self.user_id,
+            self.rating,
+            self.comment,
+            self.watched_at,
+            self.created_at,
+            self.remote_actor_url,
+            self.watch_medium,
+            self.spoilers,
+        )?;
+        let diary = DiaryEntry::new(movie, review);
+        Ok(FeedEntry::new(diary, self.user_email))
+    }
+}
+
+#[derive(sqlx::FromRow)]
+pub(crate) struct MovieStatsRow {
+    pub total_count: i64,
+    pub avg_rating: Option<f64>,
+    pub federated_count: i64,
+    pub rating_1: i64,
+    pub rating_2: i64,
+    pub rating_3: i64,
+    pub rating_4: i64,
+    pub rating_5: i64,
+}
+
+impl MovieStatsRow {
+    pub fn into_domain(self) -> domain::models::MovieStats {
+        movie_stats_to_domain(
+            self.total_count,
+            self.avg_rating,
+            self.federated_count,
+            [
+                self.rating_1,
+                self.rating_2,
+                self.rating_3,
+                self.rating_4,
+                self.rating_5,
+            ],
+        )
+    }
+}

crates/adapters/sqlite/src/repositories/diary/tests.rs

@@ -0,0 +1,207 @@
+use super::*;
+use domain::{
+    models::collections::PageParams,
+    models::{FeedSortBy, FollowingFilter},
+    ports::DiaryQuery,
+};
+use sqlx::SqlitePool;
+
+async fn setup(pool: &SqlitePool) {
+    sqlx::migrate!("./migrations").run(pool).await.unwrap();
+
+    sqlx::query(
+        "INSERT INTO users (id, email, username, password_hash, created_at) VALUES
+         ('11111111-1111-1111-1111-111111111111', 'alice@example.com', 'alice', 'hash', '2024-01-01 00:00:00'),
+         ('22222222-2222-2222-2222-222222222222', 'bob@example.com', 'bob', 'hash', '2024-01-01 00:00:00'),
+         ('33333333-3333-3333-3333-333333333333', 'carol@remote.social', 'carol', 'hash', '2024-01-01 00:00:00')",
+    )
+    .execute(pool)
+    .await
+    .unwrap();
+
+    sqlx::query(
+        "INSERT INTO movies (id, title, release_year) VALUES
+         ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', 'Inception', 2010),
+         ('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 'Interstellar', 2014),
+         ('cccccccc-cccc-cccc-cccc-cccccccccccc', 'Dune', 2021)",
+    )
+    .execute(pool)
+    .await
+    .unwrap();
+
+    sqlx::query(
+        "INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at, remote_actor_url) VALUES
+         ('a1a1a1a1-a1a1-a1a1-a1a1-a1a1a1a1a1a1', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '11111111-1111-1111-1111-111111111111', 5, '2024-01-01 00:00:00', '2024-01-01 00:00:00', NULL),
+         ('b2b2b2b2-b2b2-b2b2-b2b2-b2b2b2b2b2b2', 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', '22222222-2222-2222-2222-222222222222', 3, '2024-01-02 00:00:00', '2024-01-02 00:00:00', NULL),
+         ('c3c3c3c3-c3c3-c3c3-c3c3-c3c3c3c3c3c3', 'cccccccc-cccc-cccc-cccc-cccccccccccc', '33333333-3333-3333-3333-333333333333', 4, '2024-01-03 00:00:00', '2024-01-03 00:00:00', 'https://remote.social/users/carol')",
+    )
+    .execute(pool)
+    .await
+    .unwrap();
+}
+
+#[tokio::test]
+async fn test_sort_by_rating_descending() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let page = PageParams::new(Some(10), Some(0)).unwrap();
+    let result = repo
+        .query_activity_feed_filtered(&page, &FeedSortBy::Rating, None, None)
+        .await
+        .unwrap();
+
+    let ratings: Vec<u8> = result
+        .items
+        .iter()
+        .map(|e| e.review().rating().value())
+        .collect();
+    assert_eq!(ratings, vec![5, 4, 3]);
+}
+
+#[tokio::test]
+async fn test_search_by_title() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let page = PageParams::new(Some(10), Some(0)).unwrap();
+    let result = repo
+        .query_activity_feed_filtered(&page, &FeedSortBy::Date, Some("Dune"), None)
+        .await
+        .unwrap();
+
+    assert_eq!(result.items.len(), 1);
+    assert_eq!(result.items[0].movie().title().value(), "Dune");
+}
+
+#[tokio::test]
+async fn test_following_filter() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let filter = FollowingFilter {
+        local_user_ids: vec![
+            uuid::Uuid::parse_str("11111111-1111-1111-1111-111111111111").unwrap(),
+        ],
+        remote_actor_urls: vec!["https://remote.social/users/carol".to_string()],
+    };
+    let page = PageParams::new(Some(10), Some(0)).unwrap();
+    let result = repo
+        .query_activity_feed_filtered(&page, &FeedSortBy::Date, None, Some(&filter))
+        .await
+        .unwrap();
+
+    assert_eq!(result.items.len(), 2);
+    let titles: Vec<String> = result
+        .items
+        .iter()
+        .map(|e| e.movie().title().value().to_string())
+        .collect();
+    assert!(titles.contains(&"Inception".to_string()));
+    assert!(titles.contains(&"Dune".to_string()));
+}
+
+#[tokio::test]
+async fn test_get_movie_stats_local() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let movie_id = domain::value_objects::MovieId::from_uuid(
+        uuid::Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(),
+    );
+    let stats = repo.get_movie_stats(&movie_id).await.unwrap();
+
+    assert_eq!(stats.total_count, 1);
+    assert_eq!(stats.federated_count, 0);
+    assert!((stats.avg_rating.unwrap() - 5.0).abs() < 0.001);
+    assert_eq!(stats.rating_histogram[4], 1);
+    assert_eq!(stats.rating_histogram[0], 0);
+}
+
+#[tokio::test]
+async fn test_get_movie_social_feed_returns_reviews_for_movie() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let movie_id = domain::value_objects::MovieId::from_uuid(
+        uuid::Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(),
+    );
+    let page = PageParams::new(Some(10), Some(0)).unwrap();
+    let result = repo.get_movie_social_feed(&movie_id, &page).await.unwrap();
+
+    assert_eq!(result.total_count, 1);
+    assert_eq!(result.items.len(), 1);
+    assert_eq!(result.items[0].movie().title().value(), "Inception");
+    assert_eq!(result.items[0].review().rating().value(), 5);
+    assert_eq!(result.items[0].user_display_name(), "alice");
+    assert!(!result.items[0].review().is_remote());
+}
+
+#[tokio::test]
+async fn test_get_movie_social_feed_federated_review() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let movie_id = domain::value_objects::MovieId::from_uuid(
+        uuid::Uuid::parse_str("cccccccc-cccc-cccc-cccc-cccccccccccc").unwrap(),
+    );
+    let page = PageParams::new(Some(10), Some(0)).unwrap();
+    let result = repo.get_movie_social_feed(&movie_id, &page).await.unwrap();
+
+    assert_eq!(result.total_count, 1);
+    assert_eq!(result.items.len(), 1);
+    assert!(result.items[0].review().is_remote());
+    assert_eq!(
+        result.items[0].user_email(),
+        "https://remote.social/users/carol"
+    );
+}
+
+#[tokio::test]
+async fn test_get_movie_social_feed_pagination() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let movie_id = domain::value_objects::MovieId::from_uuid(
+        uuid::Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(),
+    );
+    let page = PageParams::new(Some(10), Some(5)).unwrap();
+    let result = repo.get_movie_social_feed(&movie_id, &page).await.unwrap();
+
+    assert_eq!(result.total_count, 1);
+    assert_eq!(result.items.len(), 0);
+}
+
+#[tokio::test]
+async fn test_get_movie_stats_federated() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let movie_id = domain::value_objects::MovieId::from_uuid(
+        uuid::Uuid::parse_str("cccccccc-cccc-cccc-cccc-cccccccccccc").unwrap(),
+    );
+    let stats = repo.get_movie_stats(&movie_id).await.unwrap();
+
+    assert_eq!(stats.total_count, 1);
+    assert_eq!(stats.federated_count, 1);
+    assert_eq!(stats.rating_histogram[3], 1);
+    assert_eq!(stats.rating_histogram[4], 0);
+}
+
+#[tokio::test]
+async fn count_local_posts_excludes_remote_reviews() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    setup(&pool).await;
+    let repo = SqliteDiaryRepository::new(pool);
+
+    let count = repo.count_local_posts().await.unwrap();
+    assert_eq!(count, 2);
+}