restructure

crates/domain/src/ports/auth.rs

@@ -0,0 +1,61 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{GeneratedToken, RefreshSession, User, UserSettings, UserSummary},
+    value_objects::{Email, PasswordHash, UserId, Username},
+};
+
+#[async_trait]
+pub trait AuthService: Send + Sync {
+    async fn generate_token(&self, user_id: &UserId) -> Result<GeneratedToken, DomainError>;
+    async fn validate_token(&self, token: &str) -> Result<UserId, DomainError>;
+}
+
+#[async_trait]
+pub trait UserRepository: Send + Sync {
+    async fn find_by_email(&self, email: &Email) -> Result<Option<User>, DomainError>;
+    async fn find_by_username(&self, username: &Username) -> Result<Option<User>, DomainError>;
+    async fn save(&self, user: &User) -> Result<(), DomainError>;
+    async fn find_by_id(&self, id: &UserId) -> Result<Option<User>, DomainError>;
+    async fn list_with_stats(&self) -> Result<Vec<UserSummary>, DomainError>;
+    async fn update_profile(
+        &self,
+        user_id: &UserId,
+        profile: &crate::models::UserProfile,
+    ) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait UserProfileFieldsRepository: Send + Sync {
+    async fn get_fields(
+        &self,
+        user_id: &UserId,
+    ) -> Result<Vec<crate::models::ProfileField>, DomainError>;
+    async fn set_fields(
+        &self,
+        user_id: &UserId,
+        fields: Vec<crate::models::ProfileField>,
+    ) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait PasswordHasher: Send + Sync {
+    async fn hash(&self, plain_password: &str) -> Result<PasswordHash, DomainError>;
+    async fn verify(&self, plain_password: &str, hash: &PasswordHash) -> Result<bool, DomainError>;
+}
+
+#[async_trait]
+pub trait UserSettingsRepository: Send + Sync {
+    async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError>;
+    async fn save(&self, settings: &UserSettings) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait RefreshSessionRepository: Send + Sync {
+    async fn create(&self, session: &RefreshSession) -> Result<(), DomainError>;
+    async fn get_by_token(&self, token: &str) -> Result<Option<RefreshSession>, DomainError>;
+    async fn revoke(&self, token: &str) -> Result<(), DomainError>;
+    async fn revoke_all_for_user(&self, user_id: &UserId) -> Result<(), DomainError>;
+    async fn delete_expired(&self) -> Result<u64, DomainError>;
+}

crates/domain/src/ports/diary.rs

@@ -0,0 +1,64 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    events::DomainEvent,
+    models::{
+        DiaryEntry, DiaryFilter, ExportFormat, FeedEntry, FeedSortBy, FollowingFilter, MovieStats,
+        Review, ReviewHistory, UserStats, UserTrends,
+        collections::{PageParams, Paginated},
+    },
+    value_objects::{MovieId, ReviewId, UserId},
+};
+
+#[async_trait]
+pub trait DiaryRepository: Send + Sync {
+    async fn query_diary(&self, filter: &DiaryFilter)
+    -> Result<Paginated<DiaryEntry>, DomainError>;
+    async fn query_activity_feed(
+        &self,
+        page: &PageParams,
+    ) -> Result<Paginated<FeedEntry>, DomainError>;
+    async fn query_activity_feed_filtered(
+        &self,
+        page: &PageParams,
+        sort_by: &FeedSortBy,
+        search: Option<&str>,
+        following: Option<&FollowingFilter>,
+    ) -> Result<Paginated<FeedEntry>, DomainError>;
+    async fn get_review_history(&self, movie_id: &MovieId) -> Result<ReviewHistory, DomainError>;
+    async fn get_user_history(&self, user_id: &UserId) -> Result<Vec<DiaryEntry>, DomainError>;
+    fn stream_user_history(
+        &self,
+        user_id: UserId,
+    ) -> futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>>;
+    async fn get_movie_stats(&self, movie_id: &MovieId) -> Result<MovieStats, DomainError>;
+    async fn get_movie_social_feed(
+        &self,
+        movie_id: &MovieId,
+        page: &PageParams,
+    ) -> Result<Paginated<FeedEntry>, DomainError>;
+    async fn count_local_posts(&self) -> Result<u64, DomainError>;
+}
+
+#[async_trait]
+pub trait ReviewRepository: Send + Sync {
+    async fn save_review(&self, review: &Review) -> Result<DomainEvent, DomainError>;
+    async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError>;
+    async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError>;
+    async fn get_all_reviews_for_user(&self, user_id: &UserId) -> Result<Vec<Review>, DomainError>;
+}
+
+#[async_trait]
+pub trait StatsRepository: Send + Sync {
+    async fn get_user_stats(&self, user_id: &UserId) -> Result<UserStats, DomainError>;
+    async fn get_user_trends(&self, user_id: &UserId) -> Result<UserTrends, DomainError>;
+}
+
+pub trait DiaryExporter: Send + Sync {
+    fn stream_entries(
+        &self,
+        stream: futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>>,
+        format: ExportFormat,
+    ) -> futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>;
+}

crates/domain/src/ports/events.rs

@@ -0,0 +1,23 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    events::{DomainEvent, EventEnvelope},
+};
+
+#[async_trait]
+pub trait EventPublisher: Send + Sync {
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError>;
+}
+
+pub trait EventConsumer: Send + Sync {
+    /// Returns a stream of event envelopes. Each envelope carries a domain event
+    /// and an ack handle — callers ack after successful dispatch, nack on failure.
+    /// Implementations decide transport (NATS, DB queue, in-memory channel).
+    fn consume(&self) -> futures::stream::BoxStream<'_, Result<EventEnvelope, DomainError>>;
+}
+
+#[async_trait]
+pub trait EventHandler: Send + Sync {
+    async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError>;
+}

crates/domain/src/ports/goals.rs

@@ -0,0 +1,21 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::Goal,
+    value_objects::{GoalId, UserId},
+};
+
+#[async_trait]
+pub trait GoalRepository: Send + Sync {
+    async fn save(&self, goal: &Goal) -> Result<(), DomainError>;
+    async fn update(&self, goal: &Goal) -> Result<(), DomainError>;
+    async fn delete(&self, id: &GoalId, user_id: &UserId) -> Result<(), DomainError>;
+    async fn find_by_user_and_year(
+        &self,
+        user_id: &UserId,
+        year: u16,
+    ) -> Result<Option<Goal>, DomainError>;
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<Goal>, DomainError>;
+    async fn count_reviews_in_year(&self, user_id: &UserId, year: u16) -> Result<u32, DomainError>;
+}

crates/domain/src/ports/images.rs

@@ -0,0 +1,30 @@
+use async_trait::async_trait;
+
+use crate::{errors::DomainError, value_objects::PosterUrl};
+
+#[async_trait]
+pub trait ObjectStorage: Send + Sync {
+    /// Stores `image_bytes` at `key` and returns the stored key.
+    async fn store(&self, key: &str, image_bytes: &[u8]) -> Result<String, DomainError>;
+    async fn get(&self, key: &str) -> Result<Vec<u8>, DomainError>;
+    async fn get_stream(
+        &self,
+        key: &str,
+    ) -> Result<futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>, DomainError>;
+    async fn delete(&self, key: &str) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait PosterFetcherClient: Send + Sync {
+    async fn fetch_poster_bytes(&self, poster_url: &PosterUrl) -> Result<Vec<u8>, DomainError>;
+}
+
+#[async_trait]
+pub trait ImageRefCommand: Send + Sync {
+    async fn swap(&self, old_key: &str, new_key: &str) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait ImageRefQuery: Send + Sync {
+    async fn list_keys(&self) -> Result<Vec<String>, DomainError>;
+}

crates/domain/src/ports/import.rs

@@ -0,0 +1,41 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        AnnotatedRow, FieldMapping, FileFormat, ImportError, ImportProfile, ImportSession,
+        ParsedFile,
+    },
+    value_objects::{ImportProfileId, ImportSessionId, UserId},
+};
+
+pub trait DocumentParser: Send + Sync {
+    fn parse(&self, bytes: &[u8], format: FileFormat) -> Result<ParsedFile, ImportError>;
+    fn apply_mapping(&self, file: &ParsedFile, mappings: &[FieldMapping]) -> Vec<AnnotatedRow>;
+}
+
+#[async_trait]
+pub trait ImportSessionRepository: Send + Sync {
+    async fn create(&self, session: &ImportSession) -> Result<(), DomainError>;
+    async fn get(
+        &self,
+        id: &ImportSessionId,
+        user_id: &UserId,
+    ) -> Result<Option<ImportSession>, DomainError>;
+    async fn update(&self, session: &ImportSession) -> Result<(), DomainError>;
+    async fn delete(&self, id: &ImportSessionId) -> Result<(), DomainError>;
+    async fn delete_expired(&self) -> Result<u64, DomainError>;
+    async fn delete_expired_for_user(&self, user_id: &UserId) -> Result<(), DomainError>;
+}
+
+#[async_trait]
+pub trait ImportProfileRepository: Send + Sync {
+    async fn save(&self, profile: &ImportProfile) -> Result<(), DomainError>;
+    async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<ImportProfile>, DomainError>;
+    async fn get(
+        &self,
+        id: &ImportProfileId,
+        user_id: &UserId,
+    ) -> Result<Option<ImportProfile>, DomainError>;
+    async fn delete(&self, id: &ImportProfileId) -> Result<(), DomainError>;
+}

crates/domain/src/ports/jobs.rs

@@ -0,0 +1,9 @@
+use async_trait::async_trait;
+
+use crate::errors::DomainError;
+
+#[async_trait]
+pub trait PeriodicJob: Send + Sync {
+    fn interval(&self) -> std::time::Duration;
+    async fn run(&self) -> Result<(), DomainError>;
+}

crates/domain/src/ports/media_server.rs

@@ -0,0 +1,50 @@
+use async_trait::async_trait;
+use chrono::NaiveDateTime;
+
+use crate::{
+    errors::DomainError,
+    models::{ParsedPlaybackEvent, WatchEvent, WatchEventStatus, WebhookToken},
+    value_objects::{UserId, WatchEventId, WebhookTokenId},
+};
+
+pub trait MediaServerParser: Send + Sync {
+    fn parse_playback_event(&self, body: &[u8])
+    -> Result<Option<ParsedPlaybackEvent>, DomainError>;
+}
+
+#[async_trait]
+pub trait WatchEventRepository: Send + Sync {
+    async fn save(&self, event: &WatchEvent) -> Result<(), DomainError>;
+    async fn update_status(
+        &self,
+        id: &WatchEventId,
+        status: WatchEventStatus,
+    ) -> Result<(), DomainError>;
+    async fn list_pending(&self, user_id: &UserId) -> Result<Vec<WatchEvent>, DomainError>;
+    async fn get_by_id(&self, id: &WatchEventId) -> Result<Option<WatchEvent>, DomainError>;
+    async fn get_by_ids(&self, ids: &[WatchEventId]) -> Result<Vec<WatchEvent>, DomainError>;
+    async fn update_status_batch(
+        &self,
+        ids: &[WatchEventId],
+        status: WatchEventStatus,
+    ) -> Result<u64, DomainError>;
+    async fn find_duplicate(
+        &self,
+        user_id: &UserId,
+        external_id: &str,
+        after: NaiveDateTime,
+    ) -> Result<bool, DomainError>;
+    async fn delete_non_pending_older_than(
+        &self,
+        before: NaiveDateTime,
+    ) -> Result<u64, DomainError>;
+}
+
+#[async_trait]
+pub trait WebhookTokenRepository: Send + Sync {
+    async fn save(&self, token: &WebhookToken) -> Result<(), DomainError>;
+    async fn find_by_token_hash(&self, hash: &str) -> Result<Option<WebhookToken>, DomainError>;
+    async fn list_by_user(&self, user_id: &UserId) -> Result<Vec<WebhookToken>, DomainError>;
+    async fn delete(&self, id: &WebhookTokenId, user_id: &UserId) -> Result<(), DomainError>;
+    async fn touch_last_used(&self, id: &WebhookTokenId) -> Result<(), DomainError>;
+}

crates/domain/src/ports/mod.rs

@@ -0,0 +1,29 @@
+pub mod auth;
+pub mod diary;
+pub mod events;
+pub mod goals;
+pub mod images;
+pub mod import;
+pub mod jobs;
+pub mod media_server;
+pub mod movie;
+pub mod person;
+pub mod search;
+pub mod social;
+pub mod watchlist;
+pub mod wrapup;
+
+pub use auth::*;
+pub use diary::*;
+pub use events::*;
+pub use goals::*;
+pub use images::*;
+pub use import::*;
+pub use jobs::*;
+pub use media_server::*;
+pub use movie::*;
+pub use person::*;
+pub use search::*;
+pub use social::*;
+pub use watchlist::*;
+pub use wrapup::*;

crates/domain/src/ports/movie.rs

@@ -0,0 +1,70 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        MetadataSearchCriteria, Movie, MovieFilter, MovieProfile, MovieSummary,
+        collections::{PageParams, Paginated},
+    },
+    value_objects::{ExternalMetadataId, MovieId, MovieTitle, PosterUrl, ReleaseYear},
+};
+
+#[async_trait]
+pub trait MovieRepository: Send + Sync {
+    async fn get_movie_by_external_id(
+        &self,
+        external_metadata_id: &ExternalMetadataId,
+    ) -> Result<Option<Movie>, DomainError>;
+    async fn get_movie_by_id(&self, movie_id: &MovieId) -> Result<Option<Movie>, DomainError>;
+    async fn get_movies_by_title_and_year(
+        &self,
+        title: &MovieTitle,
+        year: &ReleaseYear,
+    ) -> Result<Vec<Movie>, DomainError>;
+    async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError>;
+    async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError>;
+    async fn existing_external_ids(
+        &self,
+        ids: &[ExternalMetadataId],
+    ) -> Result<std::collections::HashSet<String>, DomainError>;
+    async fn existing_title_year_pairs(
+        &self,
+        pairs: &[(MovieTitle, ReleaseYear)],
+    ) -> Result<std::collections::HashSet<(String, u16)>, DomainError>;
+    async fn list_movies(
+        &self,
+        page: &PageParams,
+        filter: &MovieFilter,
+    ) -> Result<Paginated<MovieSummary>, DomainError>;
+}
+
+#[async_trait]
+pub trait MetadataClient: Send + Sync {
+    async fn fetch_movie_metadata(
+        &self,
+        criteria: &MetadataSearchCriteria,
+    ) -> Result<Movie, DomainError>;
+    async fn get_poster_url(
+        &self,
+        external_metadata_id: &ExternalMetadataId,
+    ) -> Result<Option<PosterUrl>, DomainError>;
+}
+
+#[async_trait]
+pub trait MovieProfileRepository: Send + Sync {
+    async fn upsert(&self, profile: &MovieProfile) -> Result<(), DomainError>;
+    async fn get_by_movie_id(&self, id: &MovieId) -> Result<Option<MovieProfile>, DomainError>;
+    /// Returns (movie_id, external_metadata_id) for movies with no profile or a stale one
+    /// (enriched_at older than 30 days).
+    async fn list_stale(&self) -> Result<Vec<(MovieId, String)>, DomainError>;
+}
+
+#[async_trait]
+pub trait MovieEnrichmentClient: Send + Sync {
+    /// Resolves an external ID (TMDb or IMDb) and fetches the full movie profile.
+    async fn fetch_profile(
+        &self,
+        movie_id: MovieId,
+        external_metadata_id: &str,
+    ) -> Result<MovieProfile, DomainError>;
+}

crates/domain/src/ports/person.rs

@@ -0,0 +1,45 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{ExternalPersonId, Person, PersonCredits, PersonEnrichmentData, PersonId},
+};
+
+#[async_trait]
+pub trait PersonEnrichmentClient: Send + Sync {
+    async fn fetch_details(&self, external_id: &str) -> Result<PersonEnrichmentData, DomainError>;
+}
+
+/// Write port — mutates the persons table. No reads.
+#[async_trait]
+pub trait PersonCommand: Send + Sync {
+    /// Upsert a batch of persons. Uses INSERT OR REPLACE (SQLite) / ON CONFLICT DO UPDATE (Postgres).
+    async fn upsert_batch(&self, persons: &[Person]) -> Result<(), DomainError>;
+    /// Insert a batch of missing persons from movie_cast/movie_crew into the persons table.
+    /// Returns (inserted_count, has_more).
+    async fn backfill_from_credits_batch(
+        &self,
+        batch_size: u32,
+    ) -> Result<(u64, bool), DomainError>;
+    async fn update_enrichment(
+        &self,
+        id: &PersonId,
+        data: &PersonEnrichmentData,
+    ) -> Result<(), DomainError>;
+}
+
+/// Read port — queries persons and credits. No mutations.
+#[async_trait]
+pub trait PersonQuery: Send + Sync {
+    async fn get_by_id(&self, id: &PersonId) -> Result<Option<Person>, DomainError>;
+    async fn get_by_external_id(
+        &self,
+        id: &ExternalPersonId,
+    ) -> Result<Option<Person>, DomainError>;
+    /// Returns the person's full cast and crew credit history across all indexed movies.
+    async fn get_credits(&self, id: &PersonId) -> Result<PersonCredits, DomainError>;
+    /// Returns persons who have no remaining entries in movie_cast or movie_crew.
+    /// Called after movie deletion to find index entries that can be pruned.
+    async fn list_orphaned_persons(&self) -> Result<Vec<PersonId>, DomainError>;
+    async fn list_page(&self, limit: u32, offset: u32) -> Result<Vec<Person>, DomainError>;
+}

crates/domain/src/ports/search.rs

@@ -0,0 +1,21 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{EntityType, IndexableDocument, SearchQuery, SearchResults},
+};
+
+/// Read port — executes search queries. No mutations.
+#[async_trait]
+pub trait SearchPort: Send + Sync {
+    async fn search(&self, query: &SearchQuery) -> Result<SearchResults, DomainError>;
+}
+
+/// Write port — manages the search index. No reads.
+#[async_trait]
+pub trait SearchCommand: Send + Sync {
+    /// Add or replace a document in the search index.
+    async fn index(&self, doc: IndexableDocument) -> Result<(), DomainError>;
+    /// Remove a document from the search index by entity type and internal ID string.
+    async fn remove(&self, entity_type: EntityType, id: &str) -> Result<(), DomainError>;
+}

crates/domain/src/ports/social.rs

@@ -0,0 +1,93 @@
+use async_trait::async_trait;
+use chrono::NaiveDateTime;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        DiaryEntry, FederationFlags, Goal, Movie, PendingFollowerInfo, RemoteActorInfo,
+        RemoteGoalEntry, RemoteWatchlistEntry, Review, WatchlistWithMovie,
+    },
+    value_objects::{MovieId, ReviewId, UserId},
+};
+
+#[async_trait]
+pub trait SocialQueryPort: Send + Sync {
+    async fn get_accepted_following_urls(
+        &self,
+        user_id: uuid::Uuid,
+    ) -> Result<Vec<String>, DomainError>;
+    async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError>;
+    async fn count_following(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
+    async fn count_accepted_followers(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
+    async fn get_pending_followers(
+        &self,
+        user_id: uuid::Uuid,
+    ) -> Result<Vec<PendingFollowerInfo>, DomainError>;
+}
+
+#[async_trait]
+pub trait UserFederationSettingsQuery: Send + Sync {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError>;
+}
+
+#[async_trait]
+pub trait RemoteWatchlistRepository: Send + Sync {
+    async fn save(&self, entry: RemoteWatchlistEntry) -> Result<(), DomainError>;
+    async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError>;
+    async fn get_by_actor_url(
+        &self,
+        actor_url: &str,
+    ) -> Result<Vec<RemoteWatchlistEntry>, DomainError>;
+    async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError>;
+    /// Find entries for a remote actor whose URL hashes (v5 UUID) to the given UUID.
+    async fn get_by_derived_uuid(
+        &self,
+        uuid: uuid::Uuid,
+    ) -> Result<Vec<RemoteWatchlistEntry>, DomainError>;
+}
+
+#[async_trait]
+pub trait RemoteGoalRepository: Send + Sync {
+    async fn save(&self, entry: RemoteGoalEntry) -> Result<(), DomainError>;
+    async fn update_by_ap_id(
+        &self,
+        ap_id: &str,
+        target: u32,
+        current: u32,
+    ) -> Result<(), DomainError>;
+    async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError>;
+    async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError>;
+    async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError>;
+}
+
+/// Read-only query port used exclusively by the ActivityPub adapter.
+/// Consolidates all reads the AP adapter needs so it never touches write repositories.
+#[async_trait]
+pub trait LocalApContentQuery: Send + Sync {
+    async fn get_local_reviews_for_user(
+        &self,
+        user_id: &UserId,
+    ) -> Result<Vec<DiaryEntry>, DomainError>;
+    async fn get_local_watchlist_for_user(
+        &self,
+        user_id: &UserId,
+    ) -> Result<Vec<WatchlistWithMovie>, DomainError>;
+    async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError>;
+    async fn get_movie_by_id(&self, movie_id: &MovieId) -> Result<Option<Movie>, DomainError>;
+    async fn count_local_posts(&self) -> Result<u64, DomainError>;
+    async fn get_local_reviews_for_movie(
+        &self,
+        movie_id: &MovieId,
+    ) -> Result<Vec<DiaryEntry>, DomainError>;
+    async fn get_local_reviews_page(
+        &self,
+        user_id: &UserId,
+        before: Option<NaiveDateTime>,
+        limit: usize,
+    ) -> Result<Vec<DiaryEntry>, DomainError>;
+    async fn get_goal_with_progress(
+        &self,
+        user_id: &UserId,
+        year: u16,
+    ) -> Result<Option<(Goal, u32)>, DomainError>;
+}

crates/domain/src/ports/watchlist.rs

@@ -0,0 +1,34 @@
+use async_trait::async_trait;
+
+use crate::{
+    errors::DomainError,
+    models::{
+        WatchlistEntry, WatchlistWithMovie,
+        collections::{PageParams, Paginated},
+    },
+    value_objects::{MovieId, UserId},
+};
+
+#[async_trait]
+pub trait WatchlistRepository: Send + Sync {
+    /// Add a new entry. Silently succeeds if the entry already exists.
+    async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError>;
+
+    /// Remove an entry. Returns NotFound if the entry does not exist.
+    async fn remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError>;
+
+    /// Remove an entry if it exists. Never returns NotFound.
+    async fn remove_if_present(
+        &self,
+        user_id: &UserId,
+        movie_id: &MovieId,
+    ) -> Result<bool, DomainError>;
+
+    async fn get_for_user(
+        &self,
+        user_id: &UserId,
+        page: &PageParams,
+    ) -> Result<Paginated<WatchlistWithMovie>, DomainError>;
+
+    async fn contains(&self, user_id: &UserId, movie_id: &MovieId) -> Result<bool, DomainError>;
+}

crates/domain/src/ports/wrapup.rs

@@ -0,0 +1,43 @@
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime};
+use uuid::Uuid;
+
+use crate::{
+    errors::DomainError,
+    models::wrapup::{
+        DateRange, WrapUpMovieRow, WrapUpRecord, WrapUpReport, WrapUpScope, WrapUpStatus,
+    },
+    value_objects::WrapUpId,
+};
+
+#[async_trait]
+pub trait WrapUpRepository: Send + Sync {
+    async fn create(&self, record: &WrapUpRecord) -> Result<(), DomainError>;
+    async fn update_status(
+        &self,
+        id: &WrapUpId,
+        status: &WrapUpStatus,
+        error: Option<&str>,
+    ) -> Result<(), DomainError>;
+    async fn set_complete(&self, id: &WrapUpId, report: &WrapUpReport) -> Result<(), DomainError>;
+    async fn get_by_id(&self, id: &WrapUpId) -> Result<Option<WrapUpRecord>, DomainError>;
+    async fn list_for_user(&self, user_id: Uuid) -> Result<Vec<WrapUpRecord>, DomainError>;
+    async fn list_global(&self) -> Result<Vec<WrapUpRecord>, DomainError>;
+    async fn find_existing(
+        &self,
+        user_id: Option<Uuid>,
+        start: NaiveDate,
+        end: NaiveDate,
+    ) -> Result<Option<WrapUpRecord>, DomainError>;
+    async fn delete(&self, id: &WrapUpId) -> Result<(), DomainError>;
+    async fn delete_failed_older_than(&self, before: NaiveDateTime) -> Result<u64, DomainError>;
+}
+
+#[async_trait]
+pub trait WrapUpStatsQuery: Send + Sync {
+    async fn get_reviews_with_profiles(
+        &self,
+        scope: &WrapUpScope,
+        range: &DateRange,
+    ) -> Result<Vec<WrapUpMovieRow>, DomainError>;
+}