activity-pub implementation

crates/adapters/sqlite/Cargo.toml

@@ -12,6 +12,8 @@ sqlx = { version = "0.8.6", features = [
 ] }
 
 domain = { workspace = true }
+activitypub = { workspace = true }
+anyhow = { workspace = true }
 uuid = { workspace = true }
 chrono = { workspace = true }
 tracing = { workspace = true }

crates/adapters/sqlite/migrations/0003_activitypub.sql

@@ -0,0 +1,28 @@
+ALTER TABLE reviews ADD COLUMN remote_actor_url TEXT;
+CREATE TABLE ap_followers (
+    local_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    remote_actor_url TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'pending',
+    created_at TEXT NOT NULL,
+    PRIMARY KEY (local_user_id, remote_actor_url)
+);
+CREATE TABLE ap_following (
+    local_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    remote_actor_url TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    PRIMARY KEY (local_user_id, remote_actor_url)
+);
+CREATE TABLE ap_remote_actors (
+    url TEXT PRIMARY KEY,
+    handle TEXT NOT NULL,
+    inbox_url TEXT NOT NULL,
+    shared_inbox_url TEXT,
+    display_name TEXT,
+    fetched_at TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS ap_local_actors (
+    user_id TEXT PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+    public_key TEXT NOT NULL,
+    private_key TEXT NOT NULL,
+    created_at TEXT NOT NULL
+);

crates/adapters/sqlite/src/federation.rs

@@ -0,0 +1,332 @@
+use anyhow::{anyhow, Result};
+use async_trait::async_trait;
+use chrono::Utc;
+use sqlx::{Row, SqlitePool};
+
+use activitypub::repository::{FederationRepository, Follower, FollowerStatus, RemoteActor};
+use domain::models::{Review, ReviewSource};
+use domain::value_objects::UserId;
+
+use crate::models::datetime_to_str;
+
+pub struct SqliteFederationRepository {
+    pool: SqlitePool,
+}
+
+impl SqliteFederationRepository {
+    pub fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+}
+
+fn status_to_str(status: &FollowerStatus) -> &'static str {
+    match status {
+        FollowerStatus::Pending => "pending",
+        FollowerStatus::Accepted => "accepted",
+        FollowerStatus::Rejected => "rejected",
+    }
+}
+
+fn str_to_status(s: &str) -> FollowerStatus {
+    match s {
+        "accepted" => FollowerStatus::Accepted,
+        "rejected" => FollowerStatus::Rejected,
+        _ => FollowerStatus::Pending,
+    }
+}
+
+#[async_trait]
+impl FederationRepository for SqliteFederationRepository {
+    async fn add_follower(
+        &self,
+        local_user_id: UserId,
+        remote_actor_url: &str,
+        status: FollowerStatus,
+    ) -> Result<()> {
+        let uid = local_user_id.value().to_string();
+        let status_str = status_to_str(&status);
+        let now = Utc::now().naive_utc();
+        let created_at = datetime_to_str(&now);
+
+        sqlx::query(
+            "INSERT INTO ap_followers (local_user_id, remote_actor_url, status, created_at)
+             VALUES (?1, ?2, ?3, ?4)
+             ON CONFLICT(local_user_id, remote_actor_url) DO UPDATE SET status = excluded.status",
+        )
+        .bind(&uid)
+        .bind(remote_actor_url)
+        .bind(status_str)
+        .bind(&created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn remove_follower(&self, local_user_id: UserId, remote_actor_url: &str) -> Result<()> {
+        let uid = local_user_id.value().to_string();
+
+        sqlx::query(
+            "DELETE FROM ap_followers WHERE local_user_id = ? AND remote_actor_url = ?",
+        )
+        .bind(&uid)
+        .bind(remote_actor_url)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn get_followers(&self, local_user_id: UserId) -> Result<Vec<Follower>> {
+        let uid = local_user_id.value().to_string();
+
+        let rows = sqlx::query(
+            "SELECT f.remote_actor_url, f.status,
+                    a.handle, a.inbox_url, a.shared_inbox_url, a.display_name
+             FROM ap_followers f
+             LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url
+             WHERE f.local_user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let followers = rows
+            .into_iter()
+            .map(|row| {
+                let url: String = row.get("remote_actor_url");
+                let status_str: String = row.get("status");
+                let handle: String = row.try_get("handle").unwrap_or_default();
+                let inbox_url: String = row.try_get("inbox_url").unwrap_or_default();
+                let shared_inbox_url: Option<String> = row.try_get("shared_inbox_url").ok().flatten();
+                let display_name: Option<String> = row.try_get("display_name").ok().flatten();
+
+                Follower {
+                    actor: RemoteActor {
+                        url,
+                        handle,
+                        inbox_url,
+                        shared_inbox_url,
+                        display_name,
+                    },
+                    status: str_to_status(&status_str),
+                }
+            })
+            .collect();
+
+        Ok(followers)
+    }
+
+    async fn update_follower_status(
+        &self,
+        local_user_id: UserId,
+        remote_actor_url: &str,
+        status: FollowerStatus,
+    ) -> Result<()> {
+        let uid = local_user_id.value().to_string();
+        let status_str = status_to_str(&status);
+
+        let result = sqlx::query(
+            "UPDATE ap_followers SET status = ? WHERE local_user_id = ? AND remote_actor_url = ?",
+        )
+        .bind(status_str)
+        .bind(&uid)
+        .bind(remote_actor_url)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() == 0 {
+            tracing::warn!(
+                local_user_id = %local_user_id.value(),
+                remote_actor_url = remote_actor_url,
+                "update_follower_status: no row found"
+            );
+        }
+
+        Ok(())
+    }
+
+    async fn add_following(&self, local_user_id: UserId, actor: RemoteActor) -> Result<()> {
+        let uid = local_user_id.value().to_string();
+        let now = Utc::now().naive_utc();
+        let created_at = datetime_to_str(&now);
+
+        self.upsert_remote_actor(actor.clone()).await?;
+
+        sqlx::query(
+            "INSERT OR IGNORE INTO ap_following (local_user_id, remote_actor_url, created_at)
+             VALUES (?, ?, ?)",
+        )
+        .bind(&uid)
+        .bind(&actor.url)
+        .bind(&created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn remove_following(&self, local_user_id: UserId, actor_url: &str) -> Result<()> {
+        let uid = local_user_id.value().to_string();
+
+        sqlx::query(
+            "DELETE FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ?",
+        )
+        .bind(&uid)
+        .bind(actor_url)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn get_following(&self, local_user_id: UserId) -> Result<Vec<RemoteActor>> {
+        let uid = local_user_id.value().to_string();
+
+        let rows = sqlx::query(
+            "SELECT a.url, a.handle, a.inbox_url, a.shared_inbox_url, a.display_name
+             FROM ap_following f
+             INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
+             WHERE f.local_user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let actors = rows
+            .into_iter()
+            .map(|row| RemoteActor {
+                url: row.get("url"),
+                handle: row.get("handle"),
+                inbox_url: row.get("inbox_url"),
+                shared_inbox_url: row.try_get("shared_inbox_url").ok().flatten(),
+                display_name: row.try_get("display_name").ok().flatten(),
+            })
+            .collect();
+
+        Ok(actors)
+    }
+
+    async fn count_following(&self, local_user_id: UserId) -> Result<usize> {
+        let uid = local_user_id.value().to_string();
+        let count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM ap_following WHERE local_user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(count as usize)
+    }
+
+    async fn upsert_remote_actor(&self, actor: RemoteActor) -> Result<()> {
+        let now = Utc::now().naive_utc();
+        let fetched_at = datetime_to_str(&now);
+
+        sqlx::query(
+            "INSERT INTO ap_remote_actors (url, handle, inbox_url, shared_inbox_url, display_name, fetched_at)
+             VALUES (?, ?, ?, ?, ?, ?)
+             ON CONFLICT(url) DO UPDATE SET
+                 handle           = excluded.handle,
+                 inbox_url        = excluded.inbox_url,
+                 shared_inbox_url = excluded.shared_inbox_url,
+                 display_name     = excluded.display_name,
+                 fetched_at       = excluded.fetched_at",
+        )
+        .bind(&actor.url)
+        .bind(&actor.handle)
+        .bind(&actor.inbox_url)
+        .bind(&actor.shared_inbox_url)
+        .bind(&actor.display_name)
+        .bind(&fetched_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn get_remote_actor(&self, actor_url: &str) -> Result<Option<RemoteActor>> {
+        let row = sqlx::query(
+            "SELECT url, handle, inbox_url, shared_inbox_url, display_name
+             FROM ap_remote_actors WHERE url = ?",
+        )
+        .bind(actor_url)
+        .fetch_optional(&self.pool)
+        .await?;
+
+        Ok(row.map(|row| RemoteActor {
+            url: row.get("url"),
+            handle: row.get("handle"),
+            inbox_url: row.get("inbox_url"),
+            shared_inbox_url: row.try_get("shared_inbox_url").ok().flatten(),
+            display_name: row.try_get("display_name").ok().flatten(),
+        }))
+    }
+
+    async fn get_local_actor_keypair(&self, user_id: UserId) -> Result<Option<(String, String)>> {
+        let uid = user_id.value().to_string();
+        let row = sqlx::query(
+            "SELECT public_key, private_key FROM ap_local_actors WHERE user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_optional(&self.pool)
+        .await?;
+
+        Ok(row.map(|r| (r.get("public_key"), r.get("private_key"))))
+    }
+
+    async fn save_local_actor_keypair(&self, user_id: UserId, public_key: String, private_key: String) -> Result<()> {
+        let uid = user_id.value().to_string();
+        let now = Utc::now().naive_utc();
+        let created_at = datetime_to_str(&now);
+
+        sqlx::query(
+            "INSERT INTO ap_local_actors (user_id, public_key, private_key, created_at)
+             VALUES (?, ?, ?, ?)
+             ON CONFLICT(user_id) DO UPDATE SET
+                 public_key  = excluded.public_key,
+                 private_key = excluded.private_key",
+        )
+        .bind(&uid)
+        .bind(&public_key)
+        .bind(&private_key)
+        .bind(&created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn save_remote_review(&self, review: &Review) -> Result<()> {
+        let actor_url = match review.source() {
+            ReviewSource::Remote { actor_url } => actor_url.clone(),
+            ReviewSource::Local => {
+                return Err(anyhow!("save_remote_review called with a local review"));
+            }
+        };
+
+        let id = review.id().value().to_string();
+        let movie_id = review.movie_id().value().to_string();
+        let user_id = review.user_id().value().to_string();
+        let rating = review.rating().value() as i64;
+        let comment = review.comment().map(|c| c.value().to_string());
+        let watched_at = datetime_to_str(review.watched_at());
+        let created_at = datetime_to_str(review.created_at());
+
+        sqlx::query(
+            "INSERT OR IGNORE INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url)
+             VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        )
+        .bind(&id)
+        .bind(&movie_id)
+        .bind(&user_id)
+        .bind(rating)
+        .bind(&comment)
+        .bind(&watched_at)
+        .bind(&created_at)
+        .bind(&actor_url)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+}

crates/adapters/sqlite/src/lib.rs

@@ -4,7 +4,7 @@ use domain::{
     events::DomainEvent,
     models::{
         DiaryEntry, DiaryFilter, DirectorStat, FeedEntry, Movie, MonthlyRating,
-        Review, ReviewHistory, SortDirection, UserStats, UserTrends,
+        Review, ReviewHistory, ReviewSource, SortDirection, UserStats, UserTrends,
         collections::{PageParams, Paginated},
     },
     ports::MovieRepository,
@@ -12,6 +12,7 @@ use domain::{
 };
 use sqlx::SqlitePool;
 
+mod federation;
 mod migrations;
 mod models;
 mod users;
@@ -21,6 +22,7 @@ use models::{
     UserTotalsRow, datetime_to_str,
 };
 
+pub use federation::SqliteFederationRepository;
 pub use users::SqliteUserRepository;
 
 fn format_year_month(ym: &str) -> String {
@@ -80,7 +82,7 @@ impl SqliteMovieRepository {
             SortDirection::Descending | SortDirection::ByRatingDesc => 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.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
                  ORDER BY r.watched_at DESC
@@ -95,7 +97,7 @@ impl SqliteMovieRepository {
             SortDirection::Ascending => 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.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
                  ORDER BY r.watched_at ASC
@@ -121,7 +123,7 @@ impl SqliteMovieRepository {
             SortDirection::Descending | SortDirection::ByRatingDesc => 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.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 = ?
@@ -138,7 +140,7 @@ impl SqliteMovieRepository {
             SortDirection::Ascending => 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.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 = ?
@@ -173,7 +175,7 @@ impl SqliteMovieRepository {
         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.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 = ?
@@ -195,7 +197,7 @@ impl SqliteMovieRepository {
         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.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 = ?
@@ -223,7 +225,7 @@ impl SqliteMovieRepository {
         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.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url,
                     u.email AS user_email
              FROM reviews r
              INNER JOIN movies m ON m.id = r.movie_id
@@ -386,17 +388,22 @@ impl MovieRepository for SqliteMovieRepository {
         let comment = review.comment().map(|c| c.value().to_string());
         let watched_at = datetime_to_str(review.watched_at());
         let created_at = datetime_to_str(review.created_at());
+        let remote_actor_url = match review.source() {
+            ReviewSource::Local => None,
+            ReviewSource::Remote { actor_url } => Some(actor_url.clone()),
+        };
 
         sqlx::query!(
-            "INSERT INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at)
-             VALUES (?, ?, ?, ?, ?, ?, ?)",
+            "INSERT INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url)
+             VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
             id,
             movie_id,
             user_id,
             rating,
             comment,
             watched_at,
-            created_at
+            created_at,
+            remote_actor_url
         )
         .execute(&self.pool)
         .await
@@ -464,7 +471,7 @@ impl MovieRepository for SqliteMovieRepository {
         let id = review_id.value().to_string();
         sqlx::query_as!(
             ReviewRow,
-            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at
+            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url
              FROM reviews WHERE id = ?",
             id
         )
@@ -510,7 +517,7 @@ impl MovieRepository for SqliteMovieRepository {
 
         let viewings = sqlx::query_as!(
             ReviewRow,
-            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at
+            "SELECT id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url
              FROM reviews WHERE movie_id = ? ORDER BY watched_at ASC",
             id_str
         )
@@ -573,7 +580,7 @@ impl MovieRepository for SqliteMovieRepository {
         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.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 = ?

crates/adapters/sqlite/src/models.rs

@@ -1,7 +1,7 @@
 use chrono::NaiveDateTime;
 use domain::{
     errors::DomainError,
-    models::{DiaryEntry, FeedEntry, Movie, Review, UserSummary},
+    models::{DiaryEntry, FeedEntry, Movie, Review, ReviewSource, UserSummary},
     value_objects::{
         Comment, ExternalMetadataId, MovieId, MovieTitle, PosterPath, Rating, ReleaseYear,
         ReviewId, UserId,
@@ -49,6 +49,7 @@ pub(crate) struct ReviewRow {
     pub comment: Option<String>,
     pub watched_at: String,
     pub created_at: String,
+    pub remote_actor_url: Option<String>,
 }
 
 impl ReviewRow {
@@ -60,8 +61,12 @@ impl ReviewRow {
         let comment = self.comment.map(Comment::new).transpose()?;
         let watched_at = parse_datetime(&self.watched_at)?;
         let created_at = parse_datetime(&self.created_at)?;
+        let source = match self.remote_actor_url {
+            None => ReviewSource::Local,
+            Some(url) => ReviewSource::Remote { actor_url: url },
+        };
         Ok(Review::from_persistence(
-            id, movie_id, user_id, rating, comment, watched_at, created_at,
+            id, movie_id, user_id, rating, comment, watched_at, created_at, source,
         ))
     }
 }
@@ -82,6 +87,7 @@ pub(crate) struct DiaryRow {
     pub comment: Option<String>,
     pub watched_at: String,
     pub created_at: String,
+    pub remote_actor_url: Option<String>,
 }
 
 impl DiaryRow {
@@ -104,6 +110,7 @@ impl DiaryRow {
             comment: self.comment,
             watched_at: self.watched_at,
             created_at: self.created_at,
+            remote_actor_url: self.remote_actor_url,
         }
         .to_domain()?;
 
@@ -127,6 +134,7 @@ pub(crate) struct FeedRow {
     pub comment: Option<String>,
     pub watched_at: String,
     pub created_at: String,
+    pub remote_actor_url: Option<String>,
     pub user_email: String,
 }
 
@@ -146,6 +154,7 @@ impl FeedRow {
             comment: self.comment,
             watched_at: self.watched_at,
             created_at: self.created_at,
+            remote_actor_url: self.remote_actor_url,
         }
         .to_domain()?;
         Ok(FeedEntry::new(diary, self.user_email))