test: cascading in-memory store

crates/domain/src/testing/fakes/mod.rs

@@ -0,0 +1,26 @@
+mod insights;
+mod noops;
+mod panics;
+mod store;
+mod store_goal;
+mod store_import;
+mod store_movie;
+mod store_review;
+mod store_session;
+mod store_social;
+mod store_user;
+mod store_watch_event;
+mod store_watchlist;
+mod stubs;
+#[cfg(test)]
+mod tests;
+mod wrapup;
+
+pub use insights::*;
+pub use noops::*;
+pub use panics::*;
+pub use store::{
+    InMemoryStore, MOVIE_CASCADE_TABLES, USER_CASCADE_TABLES, USER_CASCADE_TABLES_OUTSIDE_STORE,
+};
+pub use stubs::*;
+pub use wrapup::*;

crates/domain/src/testing/fakes/store.rs

@@ -0,0 +1,171 @@
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+
+use uuid::Uuid;
+
+use crate::models::{
+    Goal, ImportProfile, ImportSession, Movie, MovieProfile, ProfileField, RefreshSession,
+    RemoteWatchlistEntry, Review, User, UserSettings, WatchEvent, WatchlistEntry, WebhookToken,
+};
+use crate::value_objects::SocialIdentity;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) enum FollowState {
+    Pending,
+    Accepted,
+}
+
+/// Tables whose rows the schema removes when a user row is deleted
+/// (`ON DELETE CASCADE` on a column referencing `users(id)`), and which
+/// `InMemoryStore` models. The sqlite adapter's cascade test asserts the same list.
+pub const USER_CASCADE_TABLES: &[&str] = &[
+    "watchlist_entries",
+    "goals",
+    "user_settings",
+    "user_profile_fields",
+    "webhook_tokens",
+    "watch_events",
+    "ap_followers",
+    "ap_following",
+    "blocked_actors",
+];
+
+/// Tables that cascade from a user but whose aggregate this store does not hold
+/// (they belong to the federation crate).
+pub const USER_CASCADE_TABLES_OUTSIDE_STORE: &[&str] = &["ap_local_actors"];
+
+/// Tables whose rows the schema removes when a movie row is deleted.
+pub const MOVIE_CASCADE_TABLES: &[&str] = &[
+    "watchlist_entries",
+    "movie_profiles",
+    "movie_genres",
+    "movie_keywords",
+    "movie_cast",
+    "movie_crew",
+];
+
+/// One in-memory implementation of every persistence port, mirroring the
+/// relational constraints of the sqlite schema (cascades, uniqueness, FK
+/// restrictions). Share one `Arc<InMemoryStore>` across all port slots.
+pub struct InMemoryStore {
+    pub(super) movies: RwLock<HashMap<Uuid, Movie>>,
+    pub(super) movie_profiles: RwLock<HashMap<Uuid, MovieProfile>>,
+    pub(super) reviews: RwLock<HashMap<Uuid, Review>>,
+    pub(super) users: RwLock<HashMap<Uuid, User>>,
+    pub(super) profile_fields: RwLock<HashMap<Uuid, Vec<ProfileField>>>,
+    pub(super) user_settings: RwLock<HashMap<Uuid, UserSettings>>,
+    pub(super) refresh_sessions: RwLock<Vec<RefreshSession>>,
+    pub(super) watchlist: RwLock<HashMap<(Uuid, Uuid), WatchlistEntry>>,
+    pub(super) remote_watchlist: RwLock<Vec<RemoteWatchlistEntry>>,
+    pub(super) goals: RwLock<HashMap<Uuid, Goal>>,
+    pub(super) watch_events: RwLock<Vec<WatchEvent>>,
+    pub(super) webhook_tokens: RwLock<Vec<WebhookToken>>,
+    pub(super) import_sessions: RwLock<HashMap<Uuid, ImportSession>>,
+    pub(super) import_profiles: RwLock<HashMap<Uuid, ImportProfile>>,
+    pub(super) follows: RwLock<Vec<(Uuid, SocialIdentity, FollowState)>>,
+    pub(super) blocked: RwLock<Vec<(Uuid, SocialIdentity)>>,
+}
+
+impl InMemoryStore {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            movies: RwLock::default(),
+            movie_profiles: RwLock::default(),
+            reviews: RwLock::default(),
+            users: RwLock::default(),
+            profile_fields: RwLock::default(),
+            user_settings: RwLock::default(),
+            refresh_sessions: RwLock::default(),
+            watchlist: RwLock::default(),
+            remote_watchlist: RwLock::default(),
+            goals: RwLock::default(),
+            watch_events: RwLock::default(),
+            webhook_tokens: RwLock::default(),
+            import_sessions: RwLock::default(),
+            import_profiles: RwLock::default(),
+            follows: RwLock::default(),
+            blocked: RwLock::default(),
+        })
+    }
+
+    pub fn with_remote_watchlist(entries: Vec<RemoteWatchlistEntry>) -> Arc<Self> {
+        let store = Self::new();
+        *store.remote_watchlist.write().unwrap() = entries;
+        store
+    }
+
+    pub fn movie_count(&self) -> usize {
+        self.movies.read().unwrap().len()
+    }
+
+    pub fn movie_profile_count(&self) -> usize {
+        self.movie_profiles.read().unwrap().len()
+    }
+
+    pub fn review_count(&self) -> usize {
+        self.reviews.read().unwrap().len()
+    }
+
+    pub fn user_count(&self) -> usize {
+        self.users.read().unwrap().len()
+    }
+
+    pub fn watchlist_count(&self) -> usize {
+        self.watchlist.read().unwrap().len()
+    }
+
+    pub fn goal_count(&self) -> usize {
+        self.goals.read().unwrap().len()
+    }
+
+    pub fn import_session_count(&self) -> usize {
+        self.import_sessions.read().unwrap().len()
+    }
+
+    pub fn refresh_session_count(&self) -> usize {
+        self.refresh_sessions.read().unwrap().len()
+    }
+
+    /// Inserts a user bypassing the uniqueness checks of `UserRepository::save`.
+    pub fn seed_user(&self, user: User) {
+        self.users.write().unwrap().insert(user.id().value(), user);
+    }
+
+    /// Row count of the sqlite table of the same name, for the tables this
+    /// store models. Panics on a table the store does not hold.
+    pub fn rows_in(&self, table: &str) -> usize {
+        match table {
+            "watchlist_entries" => self.watchlist_count(),
+            "goals" => self.goal_count(),
+            "user_settings" => self.user_settings.read().unwrap().len(),
+            "user_profile_fields" => self
+                .profile_fields
+                .read()
+                .unwrap()
+                .values()
+                .map(Vec::len)
+                .sum(),
+            "webhook_tokens" => self.webhook_tokens.read().unwrap().len(),
+            "watch_events" => self.watch_events.read().unwrap().len(),
+            "ap_following" => self.follows.read().unwrap().len(),
+            "ap_followers" => self
+                .follows
+                .read()
+                .unwrap()
+                .iter()
+                .filter(|(_, target, _)| matches!(target, SocialIdentity::Local(_)))
+                .count(),
+            "blocked_actors" => self.blocked.read().unwrap().len(),
+            "movie_profiles" => self.movie_profile_count(),
+            "movie_genres" => self.sum_profiles(|p| p.genres.len()),
+            "movie_keywords" => self.sum_profiles(|p| p.keywords.len()),
+            "movie_cast" => self.sum_profiles(|p| p.cast.len()),
+            "movie_crew" => self.sum_profiles(|p| p.crew.len()),
+            other => panic!("InMemoryStore does not model table {other}"),
+        }
+    }
+
+    fn sum_profiles(&self, f: impl Fn(&MovieProfile) -> usize) -> usize {
+        self.movie_profiles.read().unwrap().values().map(f).sum()
+    }
+}

crates/domain/src/testing/fakes/store_goal.rs

@@ -0,0 +1,70 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::Goal,
+    ports::{GoalCommand, GoalQuery},
+    value_objects::{GoalId, UserId},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl GoalCommand for InMemoryStore {
+    /// `goals` is UNIQUE(user_id, year).
+    async fn save(&self, goal: &Goal) -> Result<(), DomainError> {
+        let mut goals = self.goals.write().unwrap();
+        let duplicate = goals.values().any(|g| {
+            g.id() == goal.id() || (g.user_id() == goal.user_id() && g.year() == goal.year())
+        });
+        if duplicate {
+            return Err(DomainError::InfrastructureError(
+                "Database operation failed".into(),
+            ));
+        }
+        goals.insert(goal.id().value(), goal.clone());
+        Ok(())
+    }
+
+    async fn update(&self, goal: &Goal) -> Result<(), DomainError> {
+        let mut store = self.goals.write().unwrap();
+        match store.entry(goal.id().value()) {
+            std::collections::hash_map::Entry::Occupied(mut e) => {
+                e.insert(goal.clone());
+                Ok(())
+            }
+            std::collections::hash_map::Entry::Vacant(_) => {
+                Err(DomainError::NotFound("goal".into()))
+            }
+        }
+    }
+
+    async fn delete(&self, id: &GoalId, _user_id: &UserId) -> Result<(), DomainError> {
+        self.goals.write().unwrap().remove(&id.value());
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl GoalQuery for InMemoryStore {
+    async fn find_by_user_and_year(
+        &self,
+        user_id: &UserId,
+        year: u16,
+    ) -> Result<Option<Goal>, DomainError> {
+        let store = self.goals.read().unwrap();
+        Ok(store
+            .values()
+            .find(|g| g.user_id().value() == user_id.value() && g.year() == year)
+            .cloned())
+    }
+
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<Goal>, DomainError> {
+        let store = self.goals.read().unwrap();
+        Ok(store
+            .values()
+            .filter(|g| g.user_id().value() == user_id.value())
+            .cloned()
+            .collect())
+    }
+}

crates/domain/src/testing/fakes/store_import.rs

@@ -0,0 +1,98 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{ImportProfile, ImportSession},
+    ports::{ImportProfileRepository, ImportSessionRepository},
+    value_objects::{ImportProfileId, ImportSessionId, UserId},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl ImportSessionRepository for InMemoryStore {
+    async fn create(&self, session: &ImportSession) -> Result<(), DomainError> {
+        self.import_sessions
+            .write()
+            .unwrap()
+            .insert(session.id.value(), session.clone());
+        Ok(())
+    }
+
+    async fn get(
+        &self,
+        id: &ImportSessionId,
+        user_id: &UserId,
+    ) -> Result<Option<ImportSession>, DomainError> {
+        let store = self.import_sessions.read().unwrap();
+        Ok(store
+            .get(&id.value())
+            .filter(|s| s.user_id.value() == user_id.value())
+            .cloned())
+    }
+
+    async fn update(&self, session: &ImportSession) -> Result<(), DomainError> {
+        self.import_sessions
+            .write()
+            .unwrap()
+            .insert(session.id.value(), session.clone());
+        Ok(())
+    }
+
+    async fn delete(&self, id: &ImportSessionId) -> Result<(), DomainError> {
+        self.import_sessions.write().unwrap().remove(&id.value());
+        Ok(())
+    }
+
+    async fn delete_expired(&self) -> Result<u64, DomainError> {
+        let mut store = self.import_sessions.write().unwrap();
+        let now = chrono::Utc::now().naive_utc();
+        let before_len = store.len();
+        store.retain(|_, s| s.expires_at > now);
+        Ok((before_len - store.len()) as u64)
+    }
+
+    async fn delete_expired_for_user(&self, user_id: &UserId) -> Result<(), DomainError> {
+        let mut store = self.import_sessions.write().unwrap();
+        let now = chrono::Utc::now().naive_utc();
+        store.retain(|_, s| !(s.user_id.value() == user_id.value() && s.expires_at <= now));
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl ImportProfileRepository for InMemoryStore {
+    async fn save(&self, profile: &ImportProfile) -> Result<(), DomainError> {
+        self.import_profiles
+            .write()
+            .unwrap()
+            .insert(profile.id.value(), profile.clone());
+        Ok(())
+    }
+
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<ImportProfile>, DomainError> {
+        let store = self.import_profiles.read().unwrap();
+        Ok(store
+            .values()
+            .filter(|p| p.user_id.value() == user_id.value())
+            .cloned()
+            .collect())
+    }
+
+    async fn get(
+        &self,
+        id: &ImportProfileId,
+        user_id: &UserId,
+    ) -> Result<Option<ImportProfile>, DomainError> {
+        let store = self.import_profiles.read().unwrap();
+        Ok(store
+            .get(&id.value())
+            .filter(|p| p.user_id.value() == user_id.value())
+            .cloned())
+    }
+
+    async fn delete(&self, id: &ImportProfileId) -> Result<(), DomainError> {
+        self.import_profiles.write().unwrap().remove(&id.value());
+        Ok(())
+    }
+}

crates/domain/src/testing/fakes/store_movie.rs

@@ -0,0 +1,200 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        Movie, MovieFilter, MovieProfile, MovieSummary, PersistedWatchEvent, WatchEvent,
+        collections::{PageParams, Paginated},
+    },
+    ports::{MovieCommand, MovieProfileRepository, MovieQuery},
+    value_objects::{ExternalMetadataId, MovieId, MovieTitle, ReleaseYear},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl MovieCommand for InMemoryStore {
+    /// `movies.external_metadata_id` is UNIQUE: another movie row holding the
+    /// same external id is a constraint violation, which the sqlite adapter
+    /// surfaces as an infrastructure error.
+    async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> {
+        let mut movies = self.movies.write().unwrap();
+        if let Some(external) = movie.external_metadata_id() {
+            let taken = movies.values().any(|m| {
+                m.id() != movie.id()
+                    && m.external_metadata_id().map(|e| e.value()) == Some(external.value())
+            });
+            if taken {
+                return Err(DomainError::InfrastructureError(
+                    "Database operation failed".into(),
+                ));
+            }
+        }
+        movies.insert(movie.id().value(), movie.clone());
+        Ok(())
+    }
+
+    /// Schema: `reviews.movie_id REFERENCES movies(id)` has no cascade, so a
+    /// movie with reviews cannot be deleted. `watchlist_entries` and the
+    /// `movie_*` profile tables cascade; `watch_events.movie_id` is set to NULL.
+    async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> {
+        let id = movie_id.value();
+        if !self.movies.read().unwrap().contains_key(&id) {
+            return Ok(());
+        }
+        if self
+            .reviews
+            .read()
+            .unwrap()
+            .values()
+            .any(|r| r.movie_id().value() == id)
+        {
+            return Err(DomainError::InfrastructureError(
+                "Database operation failed".into(),
+            ));
+        }
+        self.movies.write().unwrap().remove(&id);
+        self.watchlist
+            .write()
+            .unwrap()
+            .retain(|(_, movie), _| *movie != id);
+        self.movie_profiles.write().unwrap().remove(&id);
+        let mut events = self.watch_events.write().unwrap();
+        for event in events.iter_mut() {
+            if event.movie_id().map(|m| m.value()) == Some(id) {
+                *event = detached_from_movie(event);
+            }
+        }
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl MovieQuery for InMemoryStore {
+    async fn get_movie_by_external_id(
+        &self,
+        external_metadata_id: &ExternalMetadataId,
+    ) -> Result<Option<Movie>, DomainError> {
+        let store = self.movies.read().unwrap();
+        Ok(store
+            .values()
+            .find(|m| {
+                m.external_metadata_id()
+                    .map(|e| e.value() == external_metadata_id.value())
+                    .unwrap_or(false)
+            })
+            .cloned())
+    }
+
+    async fn get_movie_by_id(&self, movie_id: &MovieId) -> Result<Option<Movie>, DomainError> {
+        Ok(self.movies.read().unwrap().get(&movie_id.value()).cloned())
+    }
+
+    async fn get_movies_by_title_and_year(
+        &self,
+        title: &MovieTitle,
+        year: &ReleaseYear,
+    ) -> Result<Vec<Movie>, DomainError> {
+        let store = self.movies.read().unwrap();
+        Ok(store
+            .values()
+            .filter(|m| m.title() == title && m.release_year() == year)
+            .cloned()
+            .collect())
+    }
+
+    async fn existing_external_ids(
+        &self,
+        ids: &[ExternalMetadataId],
+    ) -> Result<std::collections::HashSet<String>, DomainError> {
+        let store = self.movies.read().unwrap();
+        let known: std::collections::HashSet<String> = store
+            .values()
+            .filter_map(|m| m.external_metadata_id().map(|e| e.value().to_string()))
+            .collect();
+        Ok(ids
+            .iter()
+            .map(|id| id.value().to_string())
+            .filter(|v| known.contains(v))
+            .collect())
+    }
+
+    async fn existing_title_year_pairs(
+        &self,
+        pairs: &[(MovieTitle, ReleaseYear)],
+    ) -> Result<std::collections::HashSet<(String, u16)>, DomainError> {
+        let store = self.movies.read().unwrap();
+        let known: std::collections::HashSet<(String, u16)> = store
+            .values()
+            .map(|m| (m.title().value().to_string(), m.release_year().value()))
+            .collect();
+        Ok(pairs
+            .iter()
+            .map(|(t, y)| (t.value().to_string(), y.value()))
+            .filter(|p| known.contains(p))
+            .collect())
+    }
+
+    async fn list_movies(
+        &self,
+        _page: &PageParams,
+        _filter: &MovieFilter,
+    ) -> Result<Paginated<MovieSummary>, DomainError> {
+        Ok(Paginated {
+            items: vec![],
+            total_count: 0,
+            limit: 10,
+            offset: 0,
+        })
+    }
+
+    async fn list_movies_with_external_id(&self) -> Result<Vec<Movie>, DomainError> {
+        Ok(self
+            .movies
+            .read()
+            .unwrap()
+            .values()
+            .filter(|m| m.external_metadata_id().is_some())
+            .cloned()
+            .collect())
+    }
+}
+
+#[async_trait]
+impl MovieProfileRepository for InMemoryStore {
+    async fn upsert(&self, profile: &MovieProfile) -> Result<(), DomainError> {
+        self.movie_profiles
+            .write()
+            .unwrap()
+            .insert(profile.movie_id.value(), profile.clone());
+        Ok(())
+    }
+
+    async fn get_by_movie_id(&self, id: &MovieId) -> Result<Option<MovieProfile>, DomainError> {
+        Ok(self
+            .movie_profiles
+            .read()
+            .unwrap()
+            .get(&id.value())
+            .cloned())
+    }
+
+    async fn list_stale(&self) -> Result<Vec<(MovieId, String)>, DomainError> {
+        Ok(vec![])
+    }
+}
+
+fn detached_from_movie(event: &WatchEvent) -> WatchEvent {
+    WatchEvent::from_persistence(PersistedWatchEvent {
+        id: event.id().clone(),
+        user_id: event.user_id().clone(),
+        movie_id: None,
+        title: event.title().to_string(),
+        year: event.year(),
+        external_metadata_id: event.external_metadata_id().map(str::to_string),
+        source: event.source().clone(),
+        watched_at: *event.watched_at(),
+        status: event.status().clone(),
+        created_at: *event.created_at(),
+    })
+}

crates/domain/src/testing/fakes/store_review.rs

@@ -0,0 +1,52 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::Review,
+    ports::ReviewRepository,
+    value_objects::{ReviewId, UserId},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl ReviewRepository for InMemoryStore {
+    async fn save_review(&self, review: &Review) -> Result<(), DomainError> {
+        self.reviews
+            .write()
+            .unwrap()
+            .insert(review.id().value(), review.clone());
+        Ok(())
+    }
+
+    async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError> {
+        Ok(self
+            .reviews
+            .read()
+            .unwrap()
+            .get(&review_id.value())
+            .cloned())
+    }
+
+    async fn update_review(&self, review: &Review) -> Result<(), DomainError> {
+        self.reviews
+            .write()
+            .unwrap()
+            .insert(review.id().value(), review.clone());
+        Ok(())
+    }
+
+    async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError> {
+        self.reviews.write().unwrap().remove(&review_id.value());
+        Ok(())
+    }
+
+    async fn get_all_reviews_for_user(&self, user_id: &UserId) -> Result<Vec<Review>, DomainError> {
+        let store = self.reviews.read().unwrap();
+        Ok(store
+            .values()
+            .filter(|r| r.user_id() == user_id)
+            .cloned()
+            .collect())
+    }
+}

crates/domain/src/testing/fakes/store_session.rs

@@ -0,0 +1,58 @@
+use async_trait::async_trait;
+use chrono::Utc;
+
+use crate::{
+    errors::DomainError, models::RefreshSession, ports::RefreshSessionRepository,
+    value_objects::UserId,
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl RefreshSessionRepository for InMemoryStore {
+    /// `refresh_sessions.token` is UNIQUE.
+    async fn create(&self, session: &RefreshSession) -> Result<(), DomainError> {
+        let mut sessions = self.refresh_sessions.write().unwrap();
+        if sessions.iter().any(|s| s.token == session.token) {
+            return Err(DomainError::InfrastructureError(
+                "Database operation failed".into(),
+            ));
+        }
+        sessions.push(session.clone());
+        Ok(())
+    }
+
+    async fn get_by_token(&self, token: &str) -> Result<Option<RefreshSession>, DomainError> {
+        Ok(self
+            .refresh_sessions
+            .read()
+            .unwrap()
+            .iter()
+            .find(|s| s.token == token)
+            .cloned())
+    }
+
+    async fn revoke(&self, token: &str) -> Result<(), DomainError> {
+        self.refresh_sessions
+            .write()
+            .unwrap()
+            .retain(|s| s.token != token);
+        Ok(())
+    }
+
+    async fn revoke_all_for_user(&self, user_id: &UserId) -> Result<(), DomainError> {
+        self.refresh_sessions
+            .write()
+            .unwrap()
+            .retain(|s| s.user_id != *user_id);
+        Ok(())
+    }
+
+    async fn delete_expired(&self) -> Result<u64, DomainError> {
+        let mut store = self.refresh_sessions.write().unwrap();
+        let before = store.len();
+        let now = Utc::now();
+        store.retain(|s| s.expires_at >= now);
+        Ok((before - store.len()) as u64)
+    }
+}

crates/domain/src/testing/fakes/store_social.rs

@@ -0,0 +1,274 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    ports::{BlockQuery, FollowGraphQuery, SocialCommand},
+    value_objects::{FollowRelation, FollowStatus, SocialActor, SocialIdentity, UserId},
+};
+
+use super::InMemoryStore;
+use super::store::FollowState;
+
+fn identity_to_actor(identity: &SocialIdentity) -> SocialActor {
+    let handle = match identity {
+        SocialIdentity::Local(uid) => format!("user-{}", uid.value()),
+        SocialIdentity::Remote { actor_url } => actor_url.clone(),
+    };
+    SocialActor {
+        identity: identity.clone(),
+        handle,
+        display_name: None,
+        avatar_url: None,
+    }
+}
+
+#[async_trait]
+impl SocialCommand for InMemoryStore {
+    async fn follow(
+        &self,
+        follower: &UserId,
+        target: &crate::value_objects::FollowTarget,
+    ) -> Result<(), DomainError> {
+        let identity = match target {
+            crate::value_objects::FollowTarget::Identity(id) => id.clone(),
+            crate::value_objects::FollowTarget::Handle(h) => SocialIdentity::Remote {
+                actor_url: h.clone(),
+            },
+        };
+        if let SocialIdentity::Local(target_id) = &identity
+            && follower == target_id
+        {
+            return Err(DomainError::ValidationError(
+                "Cannot follow yourself".into(),
+            ));
+        }
+        let mut store = self.follows.write().unwrap();
+        let already = store
+            .iter()
+            .any(|(f, t, _)| *f == follower.value() && *t == identity);
+        if already {
+            return Err(DomainError::ValidationError("Already following".into()));
+        }
+        store.push((follower.value(), identity, FollowState::Pending));
+        Ok(())
+    }
+
+    async fn unfollow(
+        &self,
+        follower: &UserId,
+        target: &SocialIdentity,
+    ) -> Result<(), DomainError> {
+        let mut store = self.follows.write().unwrap();
+        let before = store.len();
+        store.retain(|(f, t, _)| !(*f == follower.value() && t == target));
+        if store.len() == before {
+            return Err(DomainError::NotFound(
+                "Follow relationship not found".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    async fn accept_follow(
+        &self,
+        owner: &UserId,
+        requester: &SocialIdentity,
+    ) -> Result<(), DomainError> {
+        let mut store = self.follows.write().unwrap();
+        let target_identity = SocialIdentity::Local(owner.clone());
+        for (f, t, state) in store.iter_mut() {
+            let requester_matches = match requester {
+                SocialIdentity::Local(uid) => *f == uid.value(),
+                SocialIdentity::Remote { actor_url } => {
+                    if let SocialIdentity::Remote {
+                        actor_url: stored_url,
+                    } = requester
+                    {
+                        stored_url == actor_url
+                    } else {
+                        false
+                    }
+                }
+            };
+            if requester_matches && *t == target_identity && *state == FollowState::Pending {
+                *state = FollowState::Accepted;
+                return Ok(());
+            }
+        }
+        Err(DomainError::NotFound(
+            "Pending follow request not found".into(),
+        ))
+    }
+
+    async fn reject_follow(
+        &self,
+        owner: &UserId,
+        requester: &SocialIdentity,
+    ) -> Result<(), DomainError> {
+        let mut store = self.follows.write().unwrap();
+        let target_identity = SocialIdentity::Local(owner.clone());
+        let before = store.len();
+        store.retain(|(f, t, state)| {
+            let requester_matches = match requester {
+                SocialIdentity::Local(uid) => *f == uid.value(),
+                SocialIdentity::Remote { .. } => {
+                    // For remote, match by checking the stored requester identity
+                    false // simplified: reject removes by follower uuid match
+                }
+            };
+            !(requester_matches && *t == target_identity && *state == FollowState::Pending)
+        });
+        if store.len() == before {
+            return Err(DomainError::NotFound(
+                "Pending follow request not found".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    async fn remove_follower(
+        &self,
+        owner: &UserId,
+        follower: &SocialIdentity,
+    ) -> Result<(), DomainError> {
+        let mut store = self.follows.write().unwrap();
+        let target_identity = SocialIdentity::Local(owner.clone());
+        let before = store.len();
+        store.retain(|(f, t, _)| {
+            let follower_matches = match follower {
+                SocialIdentity::Local(uid) => *f == uid.value(),
+                SocialIdentity::Remote { .. } => false,
+            };
+            !(follower_matches && *t == target_identity)
+        });
+        if store.len() == before {
+            return Err(DomainError::NotFound("Follower not found".into()));
+        }
+        Ok(())
+    }
+
+    async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
+        let mut store = self.blocked.write().unwrap();
+        store.push((blocker.value(), target.clone()));
+        // Also remove any existing follow relationships
+        let mut follows = self.follows.write().unwrap();
+        follows.retain(|(f, t, _)| !(*f == blocker.value() && t == target));
+        Ok(())
+    }
+
+    async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
+        let mut store = self.blocked.write().unwrap();
+        store.retain(|(b, t)| !(*b == blocker.value() && t == target));
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl FollowGraphQuery for InMemoryStore {
+    async fn get_following(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
+        let store = self.follows.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|(f, _, state)| *f == user.value() && *state == FollowState::Accepted)
+            .map(|(_, t, _)| identity_to_actor(t))
+            .collect())
+    }
+
+    async fn get_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
+        let store = self.follows.read().unwrap();
+        let target = SocialIdentity::Local(user.clone());
+        Ok(store
+            .iter()
+            .filter(|(_, t, state)| *t == target && *state == FollowState::Accepted)
+            .map(|(f, _, _)| {
+                let id = SocialIdentity::Local(UserId::from_uuid(*f));
+                identity_to_actor(&id)
+            })
+            .collect())
+    }
+
+    async fn get_pending_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
+        let store = self.follows.read().unwrap();
+        let target = SocialIdentity::Local(user.clone());
+        Ok(store
+            .iter()
+            .filter(|(_, t, state)| *t == target && *state == FollowState::Pending)
+            .map(|(f, _, _)| {
+                let id = SocialIdentity::Local(UserId::from_uuid(*f));
+                identity_to_actor(&id)
+            })
+            .collect())
+    }
+
+    async fn get_pending_following(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
+        let store = self.follows.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|(f, _, state)| *f == user.value() && *state == FollowState::Pending)
+            .map(|(_, t, _)| identity_to_actor(t))
+            .collect())
+    }
+
+    async fn count_following(&self, user: &UserId) -> Result<usize, DomainError> {
+        let store = self.follows.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|(f, _, state)| *f == user.value() && *state == FollowState::Accepted)
+            .count())
+    }
+
+    async fn count_followers(&self, user: &UserId) -> Result<usize, DomainError> {
+        let store = self.follows.read().unwrap();
+        let target = SocialIdentity::Local(user.clone());
+        Ok(store
+            .iter()
+            .filter(|(_, t, state)| *t == target && *state == FollowState::Accepted)
+            .count())
+    }
+
+    async fn count_pending_followers(&self, user: &UserId) -> Result<usize, DomainError> {
+        let store = self.follows.read().unwrap();
+        let target = SocialIdentity::Local(user.clone());
+        Ok(store
+            .iter()
+            .filter(|(_, t, state)| *t == target && *state == FollowState::Pending)
+            .count())
+    }
+
+    async fn get_relation(
+        &self,
+        viewer: &UserId,
+        target: &SocialIdentity,
+    ) -> Result<FollowRelation, DomainError> {
+        let store = self.follows.read().unwrap();
+        let viewer_identity = SocialIdentity::Local(viewer.clone());
+        let state_to_status = |s: &FollowState| match s {
+            FollowState::Pending => FollowStatus::Pending,
+            FollowState::Accepted => FollowStatus::Accepted,
+        };
+        Ok(FollowRelation {
+            following: store
+                .iter()
+                .find(|(f, t, _)| *f == viewer.value() && t == target)
+                .map(|(_, _, s)| state_to_status(s)),
+            followed_by: store
+                .iter()
+                .find(|(f, t, _)| {
+                    SocialIdentity::Local(UserId::from_uuid(*f)) == *target && *t == viewer_identity
+                })
+                .map(|(_, _, s)| state_to_status(s)),
+        })
+    }
+}
+
+#[async_trait]
+impl BlockQuery for InMemoryStore {
+    async fn get_blocked(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
+        let store = self.blocked.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|(b, _)| *b == user.value())
+            .map(|(_, t)| identity_to_actor(t))
+            .collect())
+    }
+}

crates/domain/src/testing/fakes/store_user.rs

@@ -0,0 +1,169 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{FederationFlags, ProfileField, User, UserSettings, UserSummary},
+    ports::{
+        UserFederationSettingsQuery, UserProfileFieldsRepository, UserRepository,
+        UserSettingsRepository,
+    },
+    value_objects::{Email, SocialIdentity, UserId, Username},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl UserRepository for InMemoryStore {
+    async fn find_by_email(&self, email: &Email) -> Result<Option<User>, DomainError> {
+        let store = self.users.read().unwrap();
+        Ok(store
+            .values()
+            .find(|u| u.email().value() == email.value())
+            .cloned())
+    }
+
+    async fn find_by_username(&self, username: &Username) -> Result<Option<User>, DomainError> {
+        let store = self.users.read().unwrap();
+        Ok(store
+            .values()
+            .find(|u| u.username().value() == username.value())
+            .cloned())
+    }
+
+    /// `users.email` and `users.username` are UNIQUE; the sqlite adapter checks
+    /// both up front and reports a validation error.
+    async fn save(&self, user: &User) -> Result<(), DomainError> {
+        let mut users = self.users.write().unwrap();
+        if users
+            .values()
+            .any(|u| u.email().value() == user.email().value())
+        {
+            return Err(DomainError::ValidationError(
+                "Email already registered".into(),
+            ));
+        }
+        if users
+            .values()
+            .any(|u| u.username().value() == user.username().value())
+        {
+            return Err(DomainError::ValidationError(
+                "Username already taken".into(),
+            ));
+        }
+        users.insert(user.id().value(), user.clone());
+        Ok(())
+    }
+
+    async fn find_by_id(&self, id: &UserId) -> Result<Option<User>, DomainError> {
+        Ok(self.users.read().unwrap().get(&id.value()).cloned())
+    }
+
+    async fn list_with_stats(&self) -> Result<Vec<UserSummary>, DomainError> {
+        Ok(vec![])
+    }
+
+    async fn update_profile(
+        &self,
+        user_id: &UserId,
+        profile: &crate::models::UserProfile,
+    ) -> Result<(), DomainError> {
+        let mut store = self.users.write().unwrap();
+        if let Some(user) = store.get_mut(&user_id.value()) {
+            user.update_profile(profile.clone());
+        }
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl UserProfileFieldsRepository for InMemoryStore {
+    async fn get_fields(&self, user_id: &UserId) -> Result<Vec<ProfileField>, DomainError> {
+        let store = self.profile_fields.read().unwrap();
+        Ok(store.get(&user_id.value()).cloned().unwrap_or_default())
+    }
+
+    async fn set_fields(
+        &self,
+        user_id: &UserId,
+        fields: Vec<ProfileField>,
+    ) -> Result<(), DomainError> {
+        self.profile_fields
+            .write()
+            .unwrap()
+            .insert(user_id.value(), fields);
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl UserSettingsRepository for InMemoryStore {
+    async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError> {
+        let store = self.user_settings.read().unwrap();
+        Ok(store
+            .get(&user_id.value())
+            .cloned()
+            .unwrap_or_else(|| UserSettings::new(user_id.clone())))
+    }
+
+    async fn save(&self, settings: &UserSettings) -> Result<(), DomainError> {
+        self.user_settings
+            .write()
+            .unwrap()
+            .insert(settings.user_id().value(), settings.clone());
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl UserFederationSettingsQuery for InMemoryStore {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
+        let store = self.user_settings.read().unwrap();
+        let settings = store
+            .get(&user_id.value())
+            .cloned()
+            .unwrap_or_else(|| UserSettings::new(user_id.clone()));
+        Ok(FederationFlags {
+            goals: settings.federate_goals(),
+            reviews: settings.federate_reviews(),
+            watchlist: settings.federate_watchlist(),
+        })
+    }
+}
+
+impl InMemoryStore {
+    /// Deletes a user the way `DELETE FROM users` does under the production
+    /// schema: rows in `USER_CASCADE_TABLES` go with it. Reviews, refresh
+    /// sessions, import sessions and import profiles carry a plain `user_id`
+    /// column with no foreign key, so they stay.
+    pub fn delete_user(&self, user_id: &UserId) {
+        let id = user_id.value();
+        self.users.write().unwrap().remove(&id);
+        self.watchlist
+            .write()
+            .unwrap()
+            .retain(|(user, _), _| *user != id);
+        self.goals
+            .write()
+            .unwrap()
+            .retain(|_, g| g.user_id().value() != id);
+        self.user_settings.write().unwrap().remove(&id);
+        self.profile_fields.write().unwrap().remove(&id);
+        self.webhook_tokens
+            .write()
+            .unwrap()
+            .retain(|t| t.user_id().value() != id);
+        self.watch_events
+            .write()
+            .unwrap()
+            .retain(|e| e.user_id().value() != id);
+        let me = SocialIdentity::Local(user_id.clone());
+        self.follows
+            .write()
+            .unwrap()
+            .retain(|(follower, target, _)| *follower != id && *target != me);
+        self.blocked
+            .write()
+            .unwrap()
+            .retain(|(blocker, _)| *blocker != id);
+    }
+}

crates/domain/src/testing/fakes/store_watch_event.rs

@@ -0,0 +1,123 @@
+use async_trait::async_trait;
+use chrono::NaiveDateTime;
+use uuid::Uuid;
+
+use crate::{
+    errors::DomainError,
+    models::{WatchEvent, WatchEventStatus, WebhookToken},
+    ports::{WatchEventCommand, WatchEventQuery, WebhookTokenRepository},
+    value_objects::{UserId, WatchEventId, WebhookTokenId},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl WatchEventCommand for InMemoryStore {
+    async fn save(&self, event: &WatchEvent) -> Result<(), DomainError> {
+        self.watch_events.write().unwrap().push(event.clone());
+        Ok(())
+    }
+
+    async fn update_status(
+        &self,
+        _id: &WatchEventId,
+        _status: WatchEventStatus,
+    ) -> Result<(), DomainError> {
+        Ok(())
+    }
+
+    async fn update_status_batch(
+        &self,
+        ids: &[WatchEventId],
+        _status: WatchEventStatus,
+    ) -> Result<u64, DomainError> {
+        Ok(ids.len() as u64)
+    }
+
+    async fn delete_non_pending_older_than(
+        &self,
+        before: NaiveDateTime,
+    ) -> Result<u64, DomainError> {
+        let mut store = self.watch_events.write().unwrap();
+        let before_len = store.len();
+        store.retain(|e| *e.status() == WatchEventStatus::Pending || *e.created_at() >= before);
+        Ok((before_len - store.len()) as u64)
+    }
+}
+
+#[async_trait]
+impl WatchEventQuery for InMemoryStore {
+    async fn list_pending(&self, user_id: &UserId) -> Result<Vec<WatchEvent>, DomainError> {
+        let store = self.watch_events.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|e| {
+                e.user_id().value() == user_id.value() && *e.status() == WatchEventStatus::Pending
+            })
+            .cloned()
+            .collect())
+    }
+
+    async fn get_by_id(&self, id: &WatchEventId) -> Result<Option<WatchEvent>, DomainError> {
+        let store = self.watch_events.read().unwrap();
+        Ok(store.iter().find(|e| e.id().value() == id.value()).cloned())
+    }
+
+    async fn get_by_ids(&self, ids: &[WatchEventId]) -> Result<Vec<WatchEvent>, DomainError> {
+        let id_vals: Vec<Uuid> = ids.iter().map(|id| id.value()).collect();
+        let store = self.watch_events.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|e| id_vals.contains(&e.id().value()))
+            .cloned()
+            .collect())
+    }
+
+    async fn find_duplicate(
+        &self,
+        user_id: &UserId,
+        external_id: &str,
+        after: NaiveDateTime,
+    ) -> Result<bool, DomainError> {
+        let store = self.watch_events.read().unwrap();
+        Ok(store.iter().any(|e| {
+            e.user_id().value() == user_id.value()
+                && e.external_metadata_id() == Some(external_id)
+                && *e.watched_at() > after
+        }))
+    }
+}
+
+#[async_trait]
+impl WebhookTokenRepository for InMemoryStore {
+    async fn save(&self, token: &WebhookToken) -> Result<(), DomainError> {
+        self.webhook_tokens.write().unwrap().push(token.clone());
+        Ok(())
+    }
+
+    async fn find_by_token_hash(&self, hash: &str) -> Result<Option<WebhookToken>, DomainError> {
+        let store = self.webhook_tokens.read().unwrap();
+        Ok(store.iter().find(|t| t.token_hash() == hash).cloned())
+    }
+
+    async fn list_by_user(&self, user_id: &UserId) -> Result<Vec<WebhookToken>, DomainError> {
+        let store = self.webhook_tokens.read().unwrap();
+        Ok(store
+            .iter()
+            .filter(|t| t.user_id().value() == user_id.value())
+            .cloned()
+            .collect())
+    }
+
+    async fn delete(&self, id: &WebhookTokenId, _user_id: &UserId) -> Result<(), DomainError> {
+        self.webhook_tokens
+            .write()
+            .unwrap()
+            .retain(|t| t.id().value() != id.value());
+        Ok(())
+    }
+
+    async fn touch_last_used(&self, _id: &WebhookTokenId) -> Result<(), DomainError> {
+        Ok(())
+    }
+}

crates/domain/src/testing/fakes/store_watchlist.rs

@@ -0,0 +1,110 @@
+use async_trait::async_trait;
+use uuid::Uuid;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        RemoteWatchlistEntry, WatchlistEntry, WatchlistWithMovie,
+        collections::{PageParams, Paginated},
+    },
+    ports::{RemoteWatchlistRepository, WatchlistRepository},
+    value_objects::{MovieId, UserId},
+};
+
+use super::InMemoryStore;
+
+#[async_trait]
+impl WatchlistRepository for InMemoryStore {
+    /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op.
+    async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError> {
+        let key = (entry.user_id.value(), entry.movie_id.value());
+        self.watchlist
+            .write()
+            .unwrap()
+            .entry(key)
+            .or_insert_with(|| entry.clone());
+        Ok(())
+    }
+
+    async fn remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError> {
+        let key = (user_id.value(), movie_id.value());
+        self.watchlist
+            .write()
+            .unwrap()
+            .remove(&key)
+            .ok_or_else(|| DomainError::NotFound("watchlist entry".into()))?;
+        Ok(())
+    }
+
+    async fn remove_if_present(
+        &self,
+        user_id: &UserId,
+        movie_id: &MovieId,
+    ) -> Result<bool, DomainError> {
+        let key = (user_id.value(), movie_id.value());
+        Ok(self.watchlist.write().unwrap().remove(&key).is_some())
+    }
+
+    async fn get_for_user(
+        &self,
+        _user_id: &UserId,
+        _page: &PageParams,
+    ) -> Result<Paginated<WatchlistWithMovie>, DomainError> {
+        Ok(Paginated {
+            items: vec![],
+            total_count: 0,
+            limit: 10,
+            offset: 0,
+        })
+    }
+
+    async fn contains(&self, user_id: &UserId, movie_id: &MovieId) -> Result<bool, DomainError> {
+        let key = (user_id.value(), movie_id.value());
+        Ok(self.watchlist.read().unwrap().contains_key(&key))
+    }
+}
+
+#[async_trait]
+impl RemoteWatchlistRepository for InMemoryStore {
+    async fn save(&self, entry: RemoteWatchlistEntry) -> Result<(), DomainError> {
+        self.remote_watchlist.write().unwrap().push(entry);
+        Ok(())
+    }
+
+    async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError> {
+        self.remote_watchlist
+            .write()
+            .unwrap()
+            .retain(|e| !(e.ap_id == ap_id && e.actor_url == actor_url));
+        Ok(())
+    }
+
+    async fn get_by_actor_url(
+        &self,
+        actor_url: &str,
+    ) -> Result<Vec<RemoteWatchlistEntry>, DomainError> {
+        Ok(self
+            .remote_watchlist
+            .read()
+            .unwrap()
+            .iter()
+            .filter(|e| e.actor_url == actor_url)
+            .cloned()
+            .collect())
+    }
+
+    async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError> {
+        self.remote_watchlist
+            .write()
+            .unwrap()
+            .retain(|e| e.actor_url != actor_url);
+        Ok(())
+    }
+
+    async fn get_by_derived_uuid(
+        &self,
+        _uuid: Uuid,
+    ) -> Result<Vec<RemoteWatchlistEntry>, DomainError> {
+        Ok(self.remote_watchlist.read().unwrap().clone())
+    }
+}

crates/domain/src/testing/fakes/tests.rs

@@ -0,0 +1,388 @@
+use std::sync::Arc;
+
+use chrono::Utc;
+use uuid::Uuid;
+
+use super::{InMemoryStore, MOVIE_CASCADE_TABLES, USER_CASCADE_TABLES};
+use crate::errors::DomainError;
+use crate::models::{
+    CastMember, CrewMember, Genre, Keyword, Movie, MovieProfile, ProfileField, RefreshSession,
+    UserSettings, WatchEvent, WatchEventSource, WatchlistEntry, WebhookToken,
+};
+use crate::ports::{
+    GoalCommand, MovieCommand, MovieProfileRepository, RefreshSessionRepository, ReviewRepository,
+    SocialCommand, UserProfileFieldsRepository, UserRepository, UserSettingsRepository,
+    WatchEventCommand, WatchEventQuery, WatchlistRepository, WebhookTokenRepository,
+};
+use crate::testing::{
+    test_goal, test_movie, test_movie_with_external_id, test_review, test_user, test_watched_at,
+};
+use crate::value_objects::{ExternalMetadataId, FollowTarget, MovieId, SocialIdentity, UserId};
+
+const REMOTE_ACTOR: &str = "https://remote.example/users/ann";
+
+fn movie_profile(movie_id: &MovieId) -> MovieProfile {
+    MovieProfile {
+        movie_id: movie_id.clone(),
+        tmdb_id: 603,
+        imdb_id: None,
+        overview: None,
+        tagline: None,
+        runtime_minutes: None,
+        budget_usd: None,
+        revenue_usd: None,
+        vote_average: None,
+        vote_count: None,
+        original_language: None,
+        collection_name: None,
+        genres: vec![Genre {
+            tmdb_id: 1,
+            name: "Sci-Fi".into(),
+        }],
+        keywords: vec![Keyword {
+            tmdb_id: 2,
+            name: "simulation".into(),
+        }],
+        cast: vec![CastMember {
+            tmdb_person_id: 3,
+            name: "Actor".into(),
+            character: "Role".into(),
+            billing_order: 0,
+            profile_path: None,
+        }],
+        crew: vec![CrewMember {
+            tmdb_person_id: 4,
+            name: "Director".into(),
+            job: "Director".into(),
+            department: "Directing".into(),
+            profile_path: None,
+        }],
+        enriched_at: Utc::now(),
+    }
+}
+
+/// Saves a user and one row in every table the schema cascades from a user,
+/// plus rows in the tables that hang off a user without a foreign key.
+async fn seed_user_with_dependents(store: &Arc<InMemoryStore>) -> (UserId, Movie) {
+    let user = test_user("alice");
+    let user_id = user.id().clone();
+    let other = test_user("bob");
+    let movie = test_movie("Heat");
+    UserRepository::save(&**store, &user).await.unwrap();
+    UserRepository::save(&**store, &other).await.unwrap();
+    store.upsert_movie(&movie).await.unwrap();
+
+    store
+        .add(&WatchlistEntry::new(user_id.clone(), movie.id().clone()))
+        .await
+        .unwrap();
+    GoalCommand::save(&**store, &test_goal(&user_id, 2024, 50))
+        .await
+        .unwrap();
+    UserSettingsRepository::save(&**store, &UserSettings::new(user_id.clone()))
+        .await
+        .unwrap();
+    store
+        .set_fields(
+            &user_id,
+            vec![ProfileField {
+                name: "Site".into(),
+                value: "https://example.com".into(),
+            }],
+        )
+        .await
+        .unwrap();
+    WebhookTokenRepository::save(
+        &**store,
+        &WebhookToken::new(
+            user_id.clone(),
+            "hash".into(),
+            WatchEventSource::Jellyfin,
+            None,
+        ),
+    )
+    .await
+    .unwrap();
+    WatchEventCommand::save(
+        &**store,
+        &WatchEvent::new(
+            user_id.clone(),
+            "Heat".into(),
+            Some(1995),
+            None,
+            WatchEventSource::Jellyfin,
+            test_watched_at(),
+            None,
+        ),
+    )
+    .await
+    .unwrap();
+    store
+        .follow(
+            &user_id,
+            &FollowTarget::Identity(SocialIdentity::Remote {
+                actor_url: REMOTE_ACTOR.into(),
+            }),
+        )
+        .await
+        .unwrap();
+    store
+        .follow(
+            other.id(),
+            &FollowTarget::Identity(SocialIdentity::Local(user_id.clone())),
+        )
+        .await
+        .unwrap();
+    store
+        .block(
+            &user_id,
+            &SocialIdentity::Remote {
+                actor_url: REMOTE_ACTOR.into(),
+            },
+        )
+        .await
+        .unwrap();
+
+    store
+        .save_review(&test_review(movie.id(), &user_id, 4))
+        .await
+        .unwrap();
+    store
+        .create(&RefreshSession {
+            id: Uuid::new_v4(),
+            user_id: user_id.clone(),
+            token: "refresh-token".into(),
+            expires_at: Utc::now() + chrono::Duration::days(1),
+            created_at: Utc::now(),
+        })
+        .await
+        .unwrap();
+
+    (user_id, movie)
+}
+
+#[tokio::test]
+async fn deleting_a_user_removes_watchlist_goals_settings_profile_fields_tokens_events_follows_and_blocks()
+ {
+    let store = InMemoryStore::new();
+    let (user_id, _) = seed_user_with_dependents(&store).await;
+    for table in USER_CASCADE_TABLES {
+        assert!(store.rows_in(table) > 0, "{table} should be seeded");
+    }
+
+    store.delete_user(&user_id);
+
+    assert!(store.find_by_id(&user_id).await.unwrap().is_none());
+    for table in USER_CASCADE_TABLES {
+        assert_eq!(store.rows_in(table), 0, "{table} should cascade");
+    }
+}
+
+#[tokio::test]
+async fn deleting_a_user_keeps_reviews_and_refresh_sessions_because_the_schema_has_no_foreign_key()
+{
+    let store = InMemoryStore::new();
+    let (user_id, _) = seed_user_with_dependents(&store).await;
+
+    store.delete_user(&user_id);
+
+    assert_eq!(store.review_count(), 1);
+    assert_eq!(store.refresh_session_count(), 1);
+}
+
+#[tokio::test]
+async fn deleting_a_user_leaves_other_users_rows_alone() {
+    let store = InMemoryStore::new();
+    let (user_id, movie) = seed_user_with_dependents(&store).await;
+    let bystander = test_user("carol");
+    UserRepository::save(&*store, &bystander).await.unwrap();
+    store
+        .add(&WatchlistEntry::new(
+            bystander.id().clone(),
+            movie.id().clone(),
+        ))
+        .await
+        .unwrap();
+
+    store.delete_user(&user_id);
+
+    assert_eq!(store.watchlist_count(), 1);
+    assert!(store.find_by_id(bystander.id()).await.unwrap().is_some());
+}
+
+async fn seed_movie_with_dependents(store: &Arc<InMemoryStore>, with_review: bool) -> Movie {
+    let user = test_user("alice");
+    UserRepository::save(&**store, &user).await.unwrap();
+    let movie = test_movie("Heat");
+    store.upsert_movie(&movie).await.unwrap();
+    store
+        .add(&WatchlistEntry::new(user.id().clone(), movie.id().clone()))
+        .await
+        .unwrap();
+    store.upsert(&movie_profile(movie.id())).await.unwrap();
+    WatchEventCommand::save(
+        &**store,
+        &WatchEvent::new(
+            user.id().clone(),
+            "Heat".into(),
+            Some(1995),
+            None,
+            WatchEventSource::Plex,
+            test_watched_at(),
+            Some(movie.id().clone()),
+        ),
+    )
+    .await
+    .unwrap();
+    if with_review {
+        store
+            .save_review(&test_review(movie.id(), user.id(), 5))
+            .await
+            .unwrap();
+    }
+    movie
+}
+
+#[tokio::test]
+async fn deleting_a_movie_removes_watchlist_entries_and_enrichment_profile() {
+    let store = InMemoryStore::new();
+    let movie = seed_movie_with_dependents(&store, false).await;
+    for table in MOVIE_CASCADE_TABLES {
+        assert!(store.rows_in(table) > 0, "{table} should be seeded");
+    }
+
+    store.delete_movie(movie.id()).await.unwrap();
+
+    assert_eq!(store.movie_count(), 0);
+    for table in MOVIE_CASCADE_TABLES {
+        assert_eq!(store.rows_in(table), 0, "{table} should cascade");
+    }
+}
+
+#[tokio::test]
+async fn deleting_a_movie_detaches_watch_events_instead_of_removing_them() {
+    let store = InMemoryStore::new();
+    let movie = seed_movie_with_dependents(&store, false).await;
+    let user_id = store.watch_events.read().unwrap()[0].user_id().clone();
+
+    store.delete_movie(movie.id()).await.unwrap();
+
+    let pending = store.list_pending(&user_id).await.unwrap();
+    assert_eq!(pending.len(), 1);
+    assert!(pending[0].movie_id().is_none());
+}
+
+#[tokio::test]
+async fn deleting_a_movie_that_has_reviews_is_refused_because_reviews_do_not_cascade() {
+    let store = InMemoryStore::new();
+    let movie = seed_movie_with_dependents(&store, true).await;
+
+    let result = store.delete_movie(movie.id()).await;
+
+    assert!(matches!(result, Err(DomainError::InfrastructureError(_))));
+    assert_eq!(store.movie_count(), 1);
+    assert_eq!(store.review_count(), 1);
+    assert_eq!(store.watchlist_count(), 1);
+}
+
+#[tokio::test]
+async fn saving_a_user_with_a_taken_username_is_a_validation_error() {
+    let store = InMemoryStore::new();
+    UserRepository::save(&*store, &test_user("alice"))
+        .await
+        .unwrap();
+    let clash = crate::models::User::new(
+        crate::value_objects::Email::new("other@example.com".into()).unwrap(),
+        crate::value_objects::Username::new("alice".into()).unwrap(),
+        crate::value_objects::PasswordHash::new("hashed".into()).unwrap(),
+        crate::models::UserRole::Standard,
+    );
+
+    let result = UserRepository::save(&*store, &clash).await;
+
+    assert!(matches!(result, Err(DomainError::ValidationError(_))));
+}
+
+#[tokio::test]
+async fn saving_a_user_with_a_taken_email_is_a_validation_error() {
+    let store = InMemoryStore::new();
+    UserRepository::save(&*store, &test_user("alice"))
+        .await
+        .unwrap();
+    let clash = crate::models::User::new(
+        crate::value_objects::Email::new("alice@example.com".into()).unwrap(),
+        crate::value_objects::Username::new("alice2".into()).unwrap(),
+        crate::value_objects::PasswordHash::new("hashed".into()).unwrap(),
+        crate::models::UserRole::Standard,
+    );
+
+    let result = UserRepository::save(&*store, &clash).await;
+
+    assert!(matches!(result, Err(DomainError::ValidationError(_))));
+}
+
+#[tokio::test]
+async fn saving_a_second_goal_for_the_same_user_and_year_is_an_infrastructure_error() {
+    let store = InMemoryStore::new();
+    let user_id = UserId::generate();
+    GoalCommand::save(&*store, &test_goal(&user_id, 2024, 10))
+        .await
+        .unwrap();
+
+    let result = GoalCommand::save(&*store, &test_goal(&user_id, 2024, 20)).await;
+
+    assert!(matches!(result, Err(DomainError::InfrastructureError(_))));
+    assert_eq!(store.goal_count(), 1);
+}
+
+#[tokio::test]
+async fn adding_the_same_movie_to_a_watchlist_twice_keeps_one_entry() {
+    let store = InMemoryStore::new();
+    let user_id = UserId::generate();
+    let movie_id = MovieId::generate();
+
+    store
+        .add(&WatchlistEntry::new(user_id.clone(), movie_id.clone()))
+        .await
+        .unwrap();
+    store
+        .add(&WatchlistEntry::new(user_id, movie_id))
+        .await
+        .unwrap();
+
+    assert_eq!(store.watchlist_count(), 1);
+}
+
+#[tokio::test]
+async fn upserting_a_different_movie_with_a_taken_external_id_is_an_infrastructure_error() {
+    let store = InMemoryStore::new();
+    let external = ExternalMetadataId::new("tt0113277".into()).unwrap();
+    store
+        .upsert_movie(&test_movie_with_external_id("Heat", 1995, external.clone()))
+        .await
+        .unwrap();
+
+    let result = store
+        .upsert_movie(&test_movie_with_external_id("Heat 2", 2030, external))
+        .await;
+
+    assert!(matches!(result, Err(DomainError::InfrastructureError(_))));
+    assert_eq!(store.movie_count(), 1);
+}
+
+#[tokio::test]
+async fn creating_a_refresh_session_with_a_taken_token_is_an_infrastructure_error() {
+    let store = InMemoryStore::new();
+    let session = |token: &str| RefreshSession {
+        id: Uuid::new_v4(),
+        user_id: UserId::generate(),
+        token: token.into(),
+        expires_at: Utc::now() + chrono::Duration::days(1),
+        created_at: Utc::now(),
+    };
+    store.create(&session("same")).await.unwrap();
+
+    let result = store.create(&session("same")).await;
+
+    assert!(matches!(result, Err(DomainError::InfrastructureError(_))));
+}

crates/domain/src/testing/mod.rs

@@ -1,15 +1,5 @@
 mod factories;
 mod fakes;
-mod in_memory;
-mod insights;
-mod noops;
-mod panics;
-mod wrapup;
 
 pub use factories::*;
 pub use fakes::*;
-pub use in_memory::*;
-pub use insights::*;
-pub use noops::*;
-pub use panics::*;
-pub use wrapup::*;