insights-correlations (#22)

Reviewed-on: #22

crates/adapters/postgres/src/diary.rs

@@ -276,7 +276,6 @@ impl DiaryQuery for PostgresDiaryRepository {
         let offset = page.offset as i64;
         let has_search = search.map(|s| !s.is_empty()).unwrap_or(false);
 
-        // Dynamic param counter
         let mut p: i32 = 0;
         let mut next_param = || {
             p += 1;
@@ -323,9 +322,6 @@ impl DiaryQuery for PostgresDiaryRepository {
 
         let where_clause = where_parts.join(" AND ");
 
-        // Reset counter for count query (reuse same where_clause string but re-bind)
-        // We need a separate counter for count SQL — but since where_clause is already built
-        // with the right $N references, both queries share it.
         let count_sql = format!(
             "SELECT COUNT(*) FROM reviews r
              INNER JOIN movies m ON m.id = r.movie_id
@@ -352,7 +348,6 @@ impl DiaryQuery for PostgresDiaryRepository {
             where_clause, order_clause, limit_param, offset_param
         );
 
-        // Bind helper closure — binds search + following params in order
         macro_rules! bind_filter_params {
             ($q:expr) => {{
                 let mut q = $q;

crates/adapters/postgres/src/insights.rs

@@ -0,0 +1,81 @@
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError, models::insights::WatchRow, ports::InsightsQuery, value_objects::UserId,
+};
+use sqlx::{PgPool, Row};
+
+use adapter_common::{parse_datetime, parse_uuid};
+
+use crate::wrapup::fetch_genres_pg;
+
+/// Postgres twin of `SqliteInsightsQuery` — same shape, `$n` binds, and the
+/// same thin "rows in, rows out" contract (ADR-0010).
+pub struct PostgresInsightsQuery {
+    pool: PgPool,
+}
+
+impl PostgresInsightsQuery {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl InsightsQuery for PostgresInsightsQuery {
+    async fn get_watch_rows(&self, user_id: &UserId) -> Result<Vec<WatchRow>, DomainError> {
+        let rows = sqlx::query(
+            "SELECT movie_id, rating, \
+                    to_char(watched_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS watched_at, \
+                    watch_medium \
+             FROM reviews \
+             WHERE user_id = $1 AND remote_actor_url IS NULL \
+             ORDER BY watched_at ASC",
+        )
+        .bind(user_id.value().to_string())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(adapter_common::map_sqlx_error)?;
+
+        let mut movie_ids: Vec<String> = Vec::new();
+        let mut seen = std::collections::HashSet::new();
+        for row in &rows {
+            let mid: String = row
+                .try_get("movie_id")
+                .map_err(adapter_common::map_sqlx_error)?;
+            if seen.insert(mid.clone()) {
+                movie_ids.push(mid);
+            }
+        }
+
+        let genres_map: HashMap<String, Vec<String>> =
+            fetch_genres_pg(&self.pool, &movie_ids).await?;
+
+        let mut result = Vec::with_capacity(rows.len());
+        for row in &rows {
+            let movie_id: String = row
+                .try_get("movie_id")
+                .map_err(adapter_common::map_sqlx_error)?;
+            let rating: i64 = row
+                .try_get("rating")
+                .map_err(adapter_common::map_sqlx_error)?;
+            let watched_at: String = row
+                .try_get("watched_at")
+                .map_err(adapter_common::map_sqlx_error)?;
+            let watch_medium: Option<String> = row
+                .try_get("watch_medium")
+                .map_err(adapter_common::map_sqlx_error)?;
+
+            result.push(WatchRow {
+                movie_id: parse_uuid(&movie_id)?,
+                rating: rating as u8,
+                watched_at: parse_datetime(&watched_at)?,
+                watch_medium,
+                genres: genres_map.get(&movie_id).cloned().unwrap_or_default(),
+            });
+        }
+
+        Ok(result)
+    }
+}

crates/adapters/postgres/src/lib.rs

@@ -6,6 +6,7 @@ mod goals;
 mod image_ref;
 mod import_profile;
 mod import_session;
+mod insights;
 mod models;
 mod movie;
 mod movie_dedup;
@@ -25,6 +26,7 @@ pub use diary::PostgresDiaryRepository;
 pub use image_ref::{PostgresImageRefAdapter, create_image_ref};
 pub use import_profile::PostgresImportProfileRepository;
 pub use import_session::PostgresImportSessionRepository;
+pub use insights::PostgresInsightsQuery;
 pub use movie::PostgresMovieRepository;
 pub use movie_dedup::PostgresMovieDeduplicator;
 pub use persons::{PostgresPersonAdapter, create_person_adapter};
@@ -68,6 +70,7 @@ pub struct PostgresWireOutput {
     pub ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
     pub wrapup_repo: std::sync::Arc<dyn domain::ports::WrapUpRepository>,
     pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
+    pub insights: std::sync::Arc<dyn domain::ports::InsightsQuery>,
     pub goal_command: std::sync::Arc<dyn domain::ports::GoalCommand>,
     pub goal_query: std::sync::Arc<dyn domain::ports::GoalQuery>,
     pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
@@ -111,6 +114,7 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
         ap_content: std::sync::Arc::new(PostgresApContentQuery::new(pool.clone())) as _,
         wrapup_repo: std::sync::Arc::new(PostgresWrapUpRepository::new(pool.clone())) as _,
         wrapup_stats: std::sync::Arc::new(PostgresWrapUpStatsQuery::new(pool.clone())) as _,
+        insights: std::sync::Arc::new(PostgresInsightsQuery::new(pool.clone())) as _,
         goal_command: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
         goal_query: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
         user_settings: std::sync::Arc::clone(&user_settings_repo) as _,

crates/adapters/postgres/src/movie_dedup.rs

@@ -37,7 +37,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
             .await
             .map_err(adapter_common::map_sqlx_error)?;
 
-        // 1. Upsert canonical movie record
         sqlx::query(
             "INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
              VALUES ($1, $2, $3, $4, $5, $6)
@@ -48,7 +47,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
         .bind(&new).bind(&ext_id).bind(&title).bind(year).bind(&director).bind(&poster)
         .execute(&mut *tx).await.map_err(adapter_common::map_sqlx_error)?;
 
-        // 2. Re-point simple FK tables
         let reviews = sqlx::query("UPDATE reviews SET movie_id = $1 WHERE movie_id = $2")
             .bind(&new)
             .bind(&old)
@@ -74,7 +72,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
             .map_err(adapter_common::map_sqlx_error)?
             .rows_affected();
 
-        // 3. Re-point movie_profiles (PK — move only if canonical has none)
         let profiles = sqlx::query("UPDATE movie_profiles SET movie_id = $1 WHERE movie_id = $2")
             .bind(&new)
             .bind(&old)
@@ -83,8 +80,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
             .map_err(adapter_common::map_sqlx_error)?
             .rows_affected();
 
-        // 4. Re-point enrichment tables with composite PKs (INSERT … ON CONFLICT DO NOTHING + DELETE)
-        //    Canonical's existing rows win on conflict — old duplicates are discarded.
         sqlx::query(
             "INSERT INTO movie_genres (movie_id, tmdb_id, name)
              SELECT $1, tmdb_id, name FROM movie_genres WHERE movie_id = $2
@@ -139,7 +134,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
             .await
             .map_err(adapter_common::map_sqlx_error)?;
 
-        // 5. Delete the now-empty old movie record (remaining cascades are safe: all FKs cleared above)
         sqlx::query("DELETE FROM movies WHERE id = $1")
             .bind(&old)
             .execute(&mut *tx)

crates/adapters/postgres/src/wrapup.rs

@@ -414,7 +414,8 @@ struct CastEntry {
     profile_path: Option<String>,
 }
 
-async fn fetch_genres_pg(
+/// Shared by WrapUp and Insights: movie-id → genre names.
+pub(crate) async fn fetch_genres_pg(
     pool: &PgPool,
     movie_ids: &[String],
 ) -> Result<HashMap<String, Vec<String>>, DomainError> {