making it AP complaint

crates/adapters/sqlite/Cargo.toml

@@ -12,6 +12,7 @@ sqlx = { version = "0.8.6", features = [
 ] }
 
 domain = { workspace = true }
+sqlite-federation = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 anyhow = { workspace = true }

crates/adapters/sqlite/src/ap_content.rs

@@ -1,197 +0,0 @@
-use async_trait::async_trait;
-use domain::{
-    errors::DomainError,
-    models::{DiaryEntry, Goal, Movie, Review, WatchlistWithMovie},
-    ports::LocalApContentQuery,
-    value_objects::{MovieId, ReviewId, UserId},
-};
-use sqlx::SqlitePool;
-
-use crate::models::{DiaryRow, MovieRow, ReviewRow, WatchlistRow};
-
-pub struct SqliteApContentQuery {
-    pool: SqlitePool,
-}
-
-impl SqliteApContentQuery {
-    pub fn new(pool: SqlitePool) -> Self {
-        Self { pool }
-    }
-
-    fn map_err(e: sqlx::Error) -> DomainError {
-        tracing::error!("Database error: {:?}", e);
-        DomainError::InfrastructureError("Database operation failed".into())
-    }
-}
-
-#[async_trait]
-impl LocalApContentQuery for SqliteApContentQuery {
-    async fn get_local_reviews_for_user(
-        &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
-             FROM reviews r
-             INNER JOIN movies m ON m.id = r.movie_id
-             WHERE r.user_id = ? AND r.remote_actor_url IS NULL
-             ORDER BY r.created_at DESC",
-        )
-        .bind(&uid)
-        .fetch_all(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-        rows.into_iter().map(DiaryRow::into_domain).collect()
-    }
-
-    async fn get_local_watchlist_for_user(
-        &self,
-        user_id: &UserId,
-    ) -> Result<Vec<WatchlistWithMovie>, DomainError> {
-        let uid = user_id.value().to_string();
-        let rows: Vec<WatchlistRow> = sqlx::query_as(
-            "SELECT w.id, w.user_id, w.movie_id, w.added_at,
-                    m.id AS m_id, m.external_metadata_id, m.title, m.release_year,
-                    m.director, m.poster_path
-             FROM watchlist_entries w
-             JOIN movies m ON m.id = w.movie_id
-             WHERE w.user_id = ?
-             ORDER BY w.added_at DESC",
-        )
-        .bind(&uid)
-        .fetch_all(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-        rows.into_iter().map(WatchlistRow::into_domain).collect()
-    }
-
-    async fn get_local_reviews_for_movie(
-        &self,
-        movie_id: &MovieId,
-    ) -> Result<Vec<DiaryEntry>, DomainError> {
-        let mid = movie_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
-             FROM reviews r
-             INNER JOIN movies m ON m.id = r.movie_id
-             WHERE r.movie_id = ? AND r.remote_actor_url IS NULL
-             ORDER BY r.created_at DESC",
-        )
-        .bind(&mid)
-        .fetch_all(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-        rows.into_iter().map(DiaryRow::into_domain).collect()
-    }
-
-    async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError> {
-        let id = review_id.value().to_string();
-        sqlx::query_as::<_, ReviewRow>(
-            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url
-             FROM reviews WHERE id = ?",
-        )
-        .bind(&id)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(Self::map_err)?
-        .map(ReviewRow::into_domain)
-        .transpose()
-    }
-
-    async fn get_movie_by_id(&self, movie_id: &MovieId) -> Result<Option<Movie>, DomainError> {
-        let id = movie_id.value().to_string();
-        sqlx::query_as::<_, MovieRow>(
-            "SELECT id, external_metadata_id, title, release_year, director, poster_path
-             FROM movies WHERE id = ?",
-        )
-        .bind(&id)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(Self::map_err)?
-        .map(MovieRow::into_domain)
-        .transpose()
-    }
-
-    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(Self::map_err)?;
-        Ok(count as u64)
-    }
-
-    async fn get_local_reviews_page(
-        &self,
-        user_id: &UserId,
-        before: Option<chrono::NaiveDateTime>,
-        limit: usize,
-    ) -> Result<Vec<DiaryEntry>, DomainError> {
-        let uid = user_id.value().to_string();
-        let limit_i64 = limit as i64;
-
-        let rows = if let Some(before_ts) = before {
-            let ts = before_ts.format("%Y-%m-%d %H:%M:%S").to_string();
-            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
-                 FROM reviews r
-                 INNER JOIN movies m ON m.id = r.movie_id
-                 WHERE r.user_id = ? AND r.remote_actor_url IS NULL AND r.watched_at < ?
-                 ORDER BY r.watched_at DESC
-                 LIMIT ?",
-            )
-            .bind(&uid)
-            .bind(&ts)
-            .bind(limit_i64)
-            .fetch_all(&self.pool)
-            .await
-            .map_err(Self::map_err)?
-        } else {
-            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
-                 FROM reviews r
-                 INNER JOIN movies m ON m.id = r.movie_id
-                 WHERE r.user_id = ? AND r.remote_actor_url IS NULL
-                 ORDER BY r.watched_at DESC
-                 LIMIT ?",
-            )
-            .bind(&uid)
-            .bind(limit_i64)
-            .fetch_all(&self.pool)
-            .await
-            .map_err(Self::map_err)?
-        };
-        rows.into_iter().map(DiaryRow::into_domain).collect()
-    }
-
-    async fn get_goal_with_progress(
-        &self,
-        user_id: &UserId,
-        year: u16,
-    ) -> Result<Option<(Goal, u32)>, DomainError> {
-        let uid = user_id.value().to_string();
-        let y = year as i64;
-
-        let row = sqlx::query(
-            "SELECT id, user_id, year, target_count, goal_type, created_at \
-             FROM goals WHERE user_id = ? AND year = ?",
-        )
-        .bind(&uid)
-        .bind(y)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-
-        let Some(r) = row else { return Ok(None) };
-
-        let goal = crate::goals::row_to_goal(&r)?;
-        let count = crate::goals::count_reviews_in_year(&self.pool, user_id, year).await?;
-
-        Ok(Some((goal, count)))
-    }
-}

crates/adapters/sqlite/src/lib.rs

@@ -1,6 +1,5 @@
 use sqlx::SqlitePool;
 
-mod ap_content;
 mod diary;
 mod goals;
 mod image_ref;
@@ -9,11 +8,11 @@ mod import_session;
 mod migrations;
 mod models;
 mod movie;
+mod movie_dedup;
 mod persons;
 mod profile;
 mod profile_fields;
 mod refresh_sessions;
-mod remote_goals;
 mod review;
 mod stats;
 mod user_settings;
@@ -22,17 +21,18 @@ mod watch_event;
 mod watchlist;
 mod wrapup;
 
-pub use ap_content::SqliteApContentQuery;
 pub use diary::SqliteDiaryRepository;
 pub use image_ref::{SqliteImageRefAdapter, create_image_ref};
 pub use import_profile::SqliteImportProfileRepository;
 pub use import_session::SqliteImportSessionRepository;
 pub use movie::SqliteMovieRepository;
+pub use movie_dedup::SqliteMovieDeduplicator;
 pub use persons::{SqlitePersonAdapter, create_person_adapter};
 pub use profile::SqliteMovieProfileRepository;
 pub use profile_fields::SqliteProfileFieldsRepository;
 pub use refresh_sessions::SqliteRefreshSessionAdapter;
 pub use review::SqliteReviewRepository;
+pub use sqlite_federation::SqliteApContentQuery;
 pub use stats::SqliteStatsRepository;
 pub use users::SqliteUserRepository;
 pub use watch_event::{SqliteWatchEventRepository, SqliteWebhookTokenRepository};
@@ -91,6 +91,7 @@ pub struct SqliteWireOutput {
     pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
     pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
+    pub deduplicator: std::sync::Arc<dyn domain::ports::MovieDeduplicator>,
 }
 
 pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
@@ -135,6 +136,9 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
         goal: std::sync::Arc::new(goals::SqliteGoalRepository::new(pool.clone())) as _,
         user_settings: std::sync::Arc::clone(&user_settings_repo) as _,
         federation_settings: user_settings_repo as _,
-        remote_goal: std::sync::Arc::new(remote_goals::SqliteRemoteGoalRepository::new(pool)) as _,
+        remote_goal: std::sync::Arc::new(sqlite_federation::SqliteRemoteGoalRepository::new(
+            pool.clone(),
+        )) as _,
+        deduplicator: std::sync::Arc::new(SqliteMovieDeduplicator::new(pool)) as _,
     })
 }

crates/adapters/sqlite/src/movie.rs

@@ -248,4 +248,17 @@ impl MovieRepository for SqliteMovieRepository {
             offset: page.offset,
         })
     }
+
+    async fn list_movies_with_external_id(&self) -> Result<Vec<Movie>, DomainError> {
+        sqlx::query_as::<_, MovieRow>(
+            "SELECT id, external_metadata_id, title, release_year, director, poster_path
+             FROM movies WHERE external_metadata_id IS NOT NULL",
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(Self::map_err)?
+        .into_iter()
+        .map(MovieRow::into_domain)
+        .collect()
+    }
 }

crates/adapters/sqlite/src/movie_dedup.rs

@@ -0,0 +1,149 @@
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError, models::Movie, ports::MovieDeduplicator, value_objects::MovieId,
+};
+use sqlx::SqlitePool;
+
+pub struct SqliteMovieDeduplicator {
+    pool: SqlitePool,
+}
+
+impl SqliteMovieDeduplicator {
+    pub fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    fn map_err(e: sqlx::Error) -> DomainError {
+        tracing::error!("Database error: {:?}", e);
+        DomainError::InfrastructureError("Database operation failed".into())
+    }
+}
+
+#[async_trait]
+impl MovieDeduplicator for SqliteMovieDeduplicator {
+    async fn merge_into_canonical(
+        &self,
+        old_id: &MovieId,
+        canonical: &Movie,
+    ) -> Result<u64, DomainError> {
+        let old = old_id.value().to_string();
+        let new = canonical.id().value().to_string();
+        let ext_id = canonical
+            .external_metadata_id()
+            .map(|id| id.value().to_string());
+        let title = canonical.title().value().to_string();
+        let year = canonical.release_year().value() as i64;
+        let director = canonical.director().map(str::to_string);
+        let poster = canonical.poster_path().map(|p| p.value().to_string());
+
+        let mut tx = self.pool.begin().await.map_err(Self::map_err)?;
+
+        // 1. Upsert canonical movie record
+        sqlx::query(
+            "INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
+             VALUES (?, ?, ?, ?, ?, ?)
+             ON CONFLICT(id) DO UPDATE SET
+                 external_metadata_id = COALESCE(excluded.external_metadata_id, movies.external_metadata_id),
+                 poster_path = COALESCE(excluded.poster_path, movies.poster_path)",
+        )
+        .bind(&new).bind(&ext_id).bind(&title).bind(year).bind(&director).bind(&poster)
+        .execute(&mut *tx).await.map_err(Self::map_err)?;
+
+        // 2. Re-point simple FK tables
+        let reviews = sqlx::query("UPDATE reviews SET movie_id = ? WHERE movie_id = ?")
+            .bind(&new)
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?
+            .rows_affected();
+
+        let watchlist = sqlx::query("UPDATE watchlist_entries SET movie_id = ? WHERE movie_id = ?")
+            .bind(&new)
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?
+            .rows_affected();
+
+        let watch_events = sqlx::query("UPDATE watch_events SET movie_id = ? WHERE movie_id = ?")
+            .bind(&new)
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?
+            .rows_affected();
+
+        // 3. Re-point movie_profiles (PK — move only if canonical has none)
+        let profiles = sqlx::query("UPDATE movie_profiles SET movie_id = ? WHERE movie_id = ?")
+            .bind(&new)
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?
+            .rows_affected();
+
+        // 4. Re-point enrichment tables with composite PKs (INSERT OR IGNORE + DELETE)
+        //    Canonical's existing rows win on conflict — old duplicates are discarded.
+        sqlx::query(
+            "INSERT OR IGNORE INTO movie_genres (movie_id, tmdb_id, name)
+             SELECT ?, tmdb_id, name FROM movie_genres WHERE movie_id = ?",
+        )
+        .bind(&new)
+        .bind(&old)
+        .execute(&mut *tx)
+        .await
+        .map_err(Self::map_err)?;
+        sqlx::query("DELETE FROM movie_genres WHERE movie_id = ?")
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?;
+
+        sqlx::query(
+            "INSERT OR IGNORE INTO movie_keywords (movie_id, tmdb_id, name)
+             SELECT ?, tmdb_id, name FROM movie_keywords WHERE movie_id = ?",
+        )
+        .bind(&new)
+        .bind(&old)
+        .execute(&mut *tx)
+        .await
+        .map_err(Self::map_err)?;
+        sqlx::query("DELETE FROM movie_keywords WHERE movie_id = ?")
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?;
+
+        sqlx::query(
+            "INSERT OR IGNORE INTO movie_cast (movie_id, tmdb_person_id, name, character, billing_order, profile_path)
+             SELECT ?, tmdb_person_id, name, character, billing_order, profile_path FROM movie_cast WHERE movie_id = ?",
+        ).bind(&new).bind(&old).execute(&mut *tx).await.map_err(Self::map_err)?;
+        sqlx::query("DELETE FROM movie_cast WHERE movie_id = ?")
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?;
+
+        sqlx::query(
+            "INSERT OR IGNORE INTO movie_crew (movie_id, tmdb_person_id, name, job, department, profile_path)
+             SELECT ?, tmdb_person_id, name, job, department, profile_path FROM movie_crew WHERE movie_id = ?",
+        ).bind(&new).bind(&old).execute(&mut *tx).await.map_err(Self::map_err)?;
+        sqlx::query("DELETE FROM movie_crew WHERE movie_id = ?")
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?;
+
+        // 5. Delete the now-empty old movie record (remaining cascades are safe: all FKs cleared above)
+        sqlx::query("DELETE FROM movies WHERE id = ?")
+            .bind(&old)
+            .execute(&mut *tx)
+            .await
+            .map_err(Self::map_err)?;
+
+        tx.commit().await.map_err(Self::map_err)?;
+
+        Ok(reviews + watchlist + watch_events + profiles)
+    }
+}

crates/adapters/sqlite/src/remote_goals.rs

@@ -1,114 +0,0 @@
-use async_trait::async_trait;
-use chrono::TimeZone;
-use domain::{errors::DomainError, models::RemoteGoalEntry, ports::RemoteGoalRepository};
-use sqlx::{Row, SqlitePool};
-
-pub struct SqliteRemoteGoalRepository {
-    pool: SqlitePool,
-}
-
-impl SqliteRemoteGoalRepository {
-    pub fn new(pool: SqlitePool) -> Self {
-        Self { pool }
-    }
-
-    fn map_err(e: sqlx::Error) -> DomainError {
-        tracing::error!("Database error: {:?}", e);
-        DomainError::InfrastructureError("Database operation failed".into())
-    }
-}
-
-#[async_trait]
-impl RemoteGoalRepository for SqliteRemoteGoalRepository {
-    async fn save(&self, entry: RemoteGoalEntry) -> Result<(), DomainError> {
-        let received = entry.received_at.format("%Y-%m-%d %H:%M:%S").to_string();
-
-        sqlx::query(
-            "INSERT OR REPLACE INTO remote_goals \
-             (ap_id, actor_url, year, target_count, current_count, received_at) \
-             VALUES (?, ?, ?, ?, ?, ?)",
-        )
-        .bind(&entry.ap_id)
-        .bind(&entry.actor_url)
-        .bind(entry.year as i64)
-        .bind(entry.target_count as i64)
-        .bind(entry.current_count as i64)
-        .bind(&received)
-        .execute(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-
-        Ok(())
-    }
-
-    async fn update_by_ap_id(
-        &self,
-        ap_id: &str,
-        target: u32,
-        current: u32,
-    ) -> Result<(), DomainError> {
-        sqlx::query("UPDATE remote_goals SET target_count = ?, current_count = ? WHERE ap_id = ?")
-            .bind(target as i64)
-            .bind(current as i64)
-            .bind(ap_id)
-            .execute(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
-        Ok(())
-    }
-
-    async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError> {
-        sqlx::query("DELETE FROM remote_goals WHERE ap_id = ? AND actor_url = ?")
-            .bind(ap_id)
-            .bind(actor_url)
-            .execute(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
-        Ok(())
-    }
-
-    async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError> {
-        sqlx::query("DELETE FROM remote_goals WHERE actor_url = ?")
-            .bind(actor_url)
-            .execute(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
-        Ok(())
-    }
-
-    async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError> {
-        let rows = sqlx::query(
-            "SELECT ap_id, actor_url, year, target_count, current_count, received_at \
-             FROM remote_goals WHERE actor_url = ? ORDER BY year DESC",
-        )
-        .bind(actor_url)
-        .fetch_all(&self.pool)
-        .await
-        .map_err(Self::map_err)?;
-
-        rows.iter()
-            .map(|r| {
-                let year: i64 = r.try_get("year").unwrap_or(0);
-                let target: i64 = r.try_get("target_count").unwrap_or(0);
-                let current: i64 = r.try_get("current_count").unwrap_or(0);
-                let received_str: String = r.try_get("received_at").unwrap_or_default();
-                let received_at =
-                    chrono::NaiveDateTime::parse_from_str(&received_str, "%Y-%m-%d %H:%M:%S")
-                        .map(|ndt| chrono::Utc.from_utc_datetime(&ndt))
-                        .unwrap_or_else(|_| chrono::Utc::now());
-
-                Ok(RemoteGoalEntry {
-                    ap_id: r.try_get("ap_id").unwrap_or_default(),
-                    actor_url: r.try_get("actor_url").unwrap_or_default(),
-                    year: year as u16,
-                    target_count: target as u32,
-                    current_count: current as u32,
-                    received_at,
-                })
-            })
-            .collect()
-    }
-}