feat: Jellyfin/Plex auto-import via watch queue

Webhook ingestion from media servers — movies land in a pending
watch queue, user rates and confirms to create diary entries.

- domain: WatchEvent, WebhookToken models, MediaServerParser port
- adapters: jellyfin + plex parser crates, SQLite + Postgres repos
- application: ingest/confirm/dismiss/cleanup use cases, token mgmt
- presentation: webhook endpoints (bearer + query param auth),
  watch queue + integrations settings HTML pages, OpenAPI docs
- worker: WatchEventCleanupJob (daily, 30d retention)

Movie resolution deferred to confirm — single canonical path
through log_review for enrichment, poster fetch, federation.

crates/application/src/commands.rs

@@ -72,6 +72,41 @@ pub struct DeleteImportProfileCommand {
     pub profile_id: Uuid,
 }
 
+// ── Media server integration ──────────────────────────────────────────────────
+
+pub struct IngestWatchEventCommand {
+    pub token: String,
+    pub raw_payload: Vec<u8>,
+    pub source: domain::models::WatchEventSource,
+}
+
+pub struct WatchEventConfirmation {
+    pub watch_event_id: Uuid,
+    pub rating: u8,
+    pub comment: Option<String>,
+}
+
+pub struct ConfirmWatchEventsCommand {
+    pub user_id: Uuid,
+    pub confirmations: Vec<WatchEventConfirmation>,
+}
+
+pub struct DismissWatchEventsCommand {
+    pub user_id: Uuid,
+    pub event_ids: Vec<Uuid>,
+}
+
+pub struct GenerateWebhookTokenCommand {
+    pub user_id: Uuid,
+    pub provider: domain::models::WatchEventSource,
+    pub label: Option<String>,
+}
+
+pub struct RevokeWebhookTokenCommand {
+    pub user_id: Uuid,
+    pub token_id: Uuid,
+}
+
 pub struct UpdateProfileCommand {
     pub user_id: Uuid,
     pub display_name: Option<String>,

crates/application/src/context.rs

@@ -1,14 +1,14 @@
 use std::sync::Arc;
 
-#[cfg(feature = "federation")]
-use domain::ports::RemoteWatchlistRepository;
 use domain::ports::{
     AuthService, DiaryExporter, DiaryRepository, DocumentParser, EventPublisher, ImageStorage,
     ImportProfileRepository, ImportSessionRepository, MetadataClient, MovieProfileRepository,
     MovieRepository, PasswordHasher, PersonCommand, PersonQuery, PosterFetcherClient,
-    ReviewRepository, SearchCommand, SearchPort, SocialQueryPort, StatsRepository,
-    UserProfileFieldsRepository, UserRepository, WatchlistRepository,
+    ReviewRepository, SearchCommand, SearchPort, StatsRepository, UserProfileFieldsRepository,
+    UserRepository, WatchEventRepository, WatchlistRepository, WebhookTokenRepository,
 };
+#[cfg(feature = "federation")]
+use domain::ports::{RemoteWatchlistRepository, SocialQueryPort};
 
 use crate::config::AppConfig;
 
@@ -35,6 +35,8 @@ pub struct AppContext {
     pub search_port: Arc<dyn SearchPort>,
     pub search_command: Arc<dyn SearchCommand>,
     pub watchlist_repository: Arc<dyn WatchlistRepository>,
+    pub watch_event_repository: Arc<dyn WatchEventRepository>,
+    pub webhook_token_repository: Arc<dyn WebhookTokenRepository>,
     pub profile_fields_repository: Arc<dyn UserProfileFieldsRepository>,
     #[cfg(feature = "federation")]
     pub remote_watchlist_repository: Arc<dyn RemoteWatchlistRepository>,

crates/application/src/jobs.rs

@@ -28,6 +28,31 @@ impl PeriodicJob for ImportSessionCleanupJob {
     }
 }
 
+pub struct WatchEventCleanupJob {
+    ctx: AppContext,
+}
+
+impl WatchEventCleanupJob {
+    pub fn new(ctx: AppContext) -> Self {
+        Self { ctx }
+    }
+}
+
+#[async_trait]
+impl PeriodicJob for WatchEventCleanupJob {
+    fn interval(&self) -> Duration {
+        Duration::from_secs(86400)
+    }
+
+    async fn run(&self) -> Result<(), DomainError> {
+        let n = crate::use_cases::cleanup_watch_events::execute(&self.ctx).await?;
+        if n > 0 {
+            tracing::info!("watch event cleanup: removed {n} old entries");
+        }
+        Ok(())
+    }
+}
+
 pub struct EnrichmentStalenessJob {
     ctx: AppContext,
 }

crates/application/src/ports.rs

@@ -206,6 +206,36 @@ pub struct BlockedActorsPageData {
     pub actors: Vec<BlockedActorEntry>,
 }
 
+pub struct WebhookTokenView {
+    pub id: String,
+    pub provider: String,
+    pub label: Option<String>,
+    pub created_at: String,
+    pub last_used_at: Option<String>,
+}
+
+pub struct IntegrationsPageData {
+    pub ctx: HtmlPageContext,
+    pub tokens: Vec<WebhookTokenView>,
+    pub webhook_base_url: String,
+    pub new_token: Option<String>,
+}
+
+pub struct WatchQueueDisplayEntry {
+    pub id: String,
+    pub title: String,
+    pub year: Option<u16>,
+    pub source: String,
+    pub watched_at: String,
+    pub movie_url: Option<String>,
+}
+
+pub struct WatchQueuePageData {
+    pub ctx: HtmlPageContext,
+    pub entries: Vec<WatchQueueDisplayEntry>,
+    pub error: Option<String>,
+}
+
 pub trait HtmlRenderer: Send + Sync {
     fn render_diary_page(
         &self,
@@ -229,6 +259,8 @@ pub trait HtmlRenderer: Send + Sync {
     fn render_blocked_domains_page(&self, data: BlockedDomainsPageData) -> Result<String, String>;
     fn render_blocked_actors_page(&self, data: BlockedActorsPageData) -> Result<String, String>;
     fn render_watchlist_page(&self, data: WatchlistPageData) -> Result<String, String>;
+    fn render_integrations_page(&self, data: IntegrationsPageData) -> Result<String, String>;
+    fn render_watch_queue_page(&self, data: WatchQueuePageData) -> Result<String, String>;
 }
 
 pub trait RssFeedRenderer: Send + Sync {

crates/application/src/queries.rs

@@ -105,3 +105,11 @@ pub struct IsOnWatchlistQuery {
 pub struct GetCurrentProfileQuery {
     pub user_id: Uuid,
 }
+
+pub struct GetWatchQueueQuery {
+    pub user_id: Uuid,
+}
+
+pub struct GetWebhookTokensQuery {
+    pub user_id: Uuid,
+}

crates/application/src/test_helpers.rs

@@ -10,16 +10,16 @@ use domain::{
         ImportProfileRepository, ImportSessionRepository, MetadataClient, MovieProfileRepository,
         MovieRepository, PasswordHasher, PersonCommand, PersonQuery, PosterFetcherClient,
         ReviewRepository, SearchCommand, SearchPort, StatsRepository, UserProfileFieldsRepository,
-        UserRepository, WatchlistRepository,
+        UserRepository, WatchEventRepository, WatchlistRepository, WebhookTokenRepository,
     },
     testing::{
-        FakeAuthService, FakeDiaryRepository, FakeMetadataClient, FakePasswordHasher,
-        InMemoryMovieRepository, InMemoryReviewRepository, InMemoryUserRepository,
-        InMemoryWatchlistRepository, NoopEventPublisher, NoopImageStorage, PanicDiaryExporter,
-        PanicDiaryRepository, PanicDocumentParser, PanicImportProfileRepository,
-        PanicImportSessionRepository, PanicMovieProfileRepository, PanicPersonCommand,
-        PanicPersonQuery, PanicPosterFetcher, PanicProfileFieldsRepo, PanicSearchCommand,
-        PanicSearchPort, PanicStatsRepository,
+        FakeAuthService, FakeMetadataClient, FakePasswordHasher, InMemoryMovieRepository,
+        InMemoryReviewRepository, InMemoryUserRepository, InMemoryWatchlistRepository,
+        NoopEventPublisher, NoopImageStorage, PanicDiaryExporter, PanicDiaryRepository,
+        PanicDocumentParser, PanicImportProfileRepository, PanicImportSessionRepository,
+        PanicMovieProfileRepository, PanicPersonCommand, PanicPersonQuery, PanicPosterFetcher,
+        PanicProfileFieldsRepo, PanicSearchCommand, PanicSearchPort, PanicStatsRepository,
+        PanicWatchEventRepository, PanicWebhookTokenRepository,
     },
 };
 
@@ -43,6 +43,8 @@ pub struct TestContextBuilder {
     pub import_profile_repo: Arc<dyn ImportProfileRepository>,
     pub movie_profile_repo: Arc<dyn MovieProfileRepository>,
     pub watchlist_repo: Arc<dyn WatchlistRepository>,
+    pub watch_event_repo: Arc<dyn WatchEventRepository>,
+    pub webhook_token_repo: Arc<dyn WebhookTokenRepository>,
     pub profile_fields_repo: Arc<dyn UserProfileFieldsRepository>,
     pub person_command: Arc<dyn PersonCommand>,
     pub person_query: Arc<dyn PersonQuery>,
@@ -71,6 +73,8 @@ impl TestContextBuilder {
             import_profile_repo: Arc::new(PanicImportProfileRepository),
             movie_profile_repo: Arc::new(PanicMovieProfileRepository),
             watchlist_repo: InMemoryWatchlistRepository::new(),
+            watch_event_repo: Arc::new(PanicWatchEventRepository),
+            webhook_token_repo: Arc::new(PanicWebhookTokenRepository),
             profile_fields_repo: Arc::new(PanicProfileFieldsRepo),
             person_command: Arc::new(PanicPersonCommand),
             person_query: Arc::new(PanicPersonQuery),
@@ -138,6 +142,8 @@ impl TestContextBuilder {
             import_profile_repository: self.import_profile_repo,
             movie_profile_repository: self.movie_profile_repo,
             watchlist_repository: self.watchlist_repo,
+            watch_event_repository: self.watch_event_repo,
+            webhook_token_repository: self.webhook_token_repo,
             profile_fields_repository: self.profile_fields_repo,
             person_command: self.person_command,
             person_query: self.person_query,

crates/application/src/tests/worker.rs

@@ -58,6 +58,7 @@ impl EventHandler for RecordingHandler {
             DomainEvent::FollowAccepted { .. } => "follow_accepted",
             DomainEvent::BackfillFollower { .. } => "backfill_follower",
             DomainEvent::FederationDeliveryRequested { .. } => "federation_delivery",
+            DomainEvent::WatchEventIngested { .. } => "watch_event_ingested",
         };
         self.calls.lock().unwrap().push(label);
         Ok(())

crates/application/src/use_cases/cleanup_watch_events.rs

@@ -0,0 +1,11 @@
+use chrono::Duration;
+use domain::errors::DomainError;
+
+use crate::context::AppContext;
+
+pub async fn execute(ctx: &AppContext) -> Result<u64, DomainError> {
+    let cutoff = chrono::Utc::now().naive_utc() - Duration::days(30);
+    ctx.watch_event_repository
+        .delete_non_pending_older_than(cutoff)
+        .await
+}

crates/application/src/use_cases/confirm_watch_events.rs

@@ -0,0 +1,65 @@
+use domain::{
+    errors::DomainError,
+    models::WatchEventStatus,
+    value_objects::{UserId, WatchEventId},
+};
+
+use crate::{
+    commands::{ConfirmWatchEventsCommand, LogReviewCommand, MovieInput},
+    context::AppContext,
+    use_cases::log_review,
+};
+
+pub async fn execute(ctx: &AppContext, cmd: ConfirmWatchEventsCommand) -> Result<u32, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let mut confirmed = 0u32;
+
+    for c in cmd.confirmations {
+        let event_id = WatchEventId::from_uuid(c.watch_event_id);
+        let event = ctx
+            .watch_event_repository
+            .get_by_id(&event_id)
+            .await?
+            .ok_or_else(|| DomainError::NotFound(format!("WatchEvent {}", c.watch_event_id)))?;
+
+        if event.user_id() != &user_id {
+            return Err(DomainError::Unauthorized("not your watch event".into()));
+        }
+
+        let input = if let Some(movie_id) = event.movie_id() {
+            MovieInput {
+                movie_id: Some(movie_id.value()),
+                external_metadata_id: None,
+                manual_title: None,
+                manual_release_year: None,
+                manual_director: None,
+            }
+        } else {
+            MovieInput {
+                movie_id: None,
+                external_metadata_id: event.external_metadata_id().map(String::from),
+                manual_title: Some(event.title().to_string()),
+                manual_release_year: event.year(),
+                manual_director: None,
+            }
+        };
+
+        let review_cmd = LogReviewCommand {
+            user_id: cmd.user_id,
+            input,
+            rating: c.rating,
+            comment: c.comment,
+            watched_at: *event.watched_at(),
+        };
+
+        log_review::execute(ctx, review_cmd).await?;
+
+        ctx.watch_event_repository
+            .update_status(&event_id, WatchEventStatus::Confirmed)
+            .await?;
+
+        confirmed += 1;
+    }
+
+    Ok(confirmed)
+}

crates/application/src/use_cases/dismiss_watch_events.rs

@@ -0,0 +1,33 @@
+use domain::{
+    errors::DomainError,
+    models::WatchEventStatus,
+    value_objects::{UserId, WatchEventId},
+};
+
+use crate::{commands::DismissWatchEventsCommand, context::AppContext};
+
+pub async fn execute(ctx: &AppContext, cmd: DismissWatchEventsCommand) -> Result<u32, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let mut dismissed = 0u32;
+
+    for id in cmd.event_ids {
+        let event_id = WatchEventId::from_uuid(id);
+        let event = ctx
+            .watch_event_repository
+            .get_by_id(&event_id)
+            .await?
+            .ok_or_else(|| DomainError::NotFound(format!("WatchEvent {id}")))?;
+
+        if event.user_id() != &user_id {
+            return Err(DomainError::Unauthorized("not your watch event".into()));
+        }
+
+        ctx.watch_event_repository
+            .update_status(&event_id, WatchEventStatus::Dismissed)
+            .await?;
+
+        dismissed += 1;
+    }
+
+    Ok(dismissed)
+}

crates/application/src/use_cases/generate_webhook_token.rs

@@ -0,0 +1,38 @@
+use domain::{errors::DomainError, models::WebhookToken, value_objects::UserId};
+use sha2::{Digest, Sha256};
+
+use crate::{commands::GenerateWebhookTokenCommand, context::AppContext};
+
+pub struct GeneratedWebhookToken {
+    pub token_plaintext: String,
+    pub token: WebhookToken,
+}
+
+pub async fn execute(
+    ctx: &AppContext,
+    cmd: GenerateWebhookTokenCommand,
+) -> Result<GeneratedWebhookToken, DomainError> {
+    let plaintext = generate_random_token();
+    let hash = hash_token(&plaintext);
+
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let token = WebhookToken::new(user_id, hash, cmd.provider, cmd.label);
+
+    ctx.webhook_token_repository.save(&token).await?;
+
+    Ok(GeneratedWebhookToken {
+        token_plaintext: plaintext,
+        token,
+    })
+}
+
+fn generate_random_token() -> String {
+    let bytes: [u8; 32] = rand::random();
+    hex::encode(bytes)
+}
+
+pub fn hash_token(plaintext: &str) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(plaintext.as_bytes());
+    hex::encode(hasher.finalize())
+}

crates/application/src/use_cases/get_watch_queue.rs

@@ -0,0 +1,11 @@
+use domain::{errors::DomainError, models::WatchEvent, value_objects::UserId};
+
+use crate::{context::AppContext, queries::GetWatchQueueQuery};
+
+pub async fn execute(
+    ctx: &AppContext,
+    query: GetWatchQueueQuery,
+) -> Result<Vec<WatchEvent>, DomainError> {
+    let user_id = UserId::from_uuid(query.user_id);
+    ctx.watch_event_repository.list_pending(&user_id).await
+}

crates/application/src/use_cases/get_webhook_tokens.rs

@@ -0,0 +1,11 @@
+use domain::{errors::DomainError, models::WebhookToken, value_objects::UserId};
+
+use crate::{context::AppContext, queries::GetWebhookTokensQuery};
+
+pub async fn execute(
+    ctx: &AppContext,
+    query: GetWebhookTokensQuery,
+) -> Result<Vec<WebhookToken>, DomainError> {
+    let user_id = UserId::from_uuid(query.user_id);
+    ctx.webhook_token_repository.list_by_user(&user_id).await
+}

crates/application/src/use_cases/ingest_watch_event.rs

@@ -0,0 +1,69 @@
+use chrono::Duration;
+use domain::{
+    errors::DomainError, events::DomainEvent, models::WatchEvent, ports::MediaServerParser,
+};
+
+use crate::{
+    commands::IngestWatchEventCommand, context::AppContext, use_cases::generate_webhook_token,
+};
+
+pub async fn execute(
+    ctx: &AppContext,
+    cmd: IngestWatchEventCommand,
+    parser: &dyn MediaServerParser,
+) -> Result<(), DomainError> {
+    let token_hash = generate_webhook_token::hash_token(&cmd.token);
+    let webhook_token = ctx
+        .webhook_token_repository
+        .find_by_token_hash(&token_hash)
+        .await?
+        .ok_or_else(|| DomainError::Unauthorized("invalid webhook token".into()))?;
+
+    let _ = ctx
+        .webhook_token_repository
+        .touch_last_used(webhook_token.id())
+        .await;
+
+    let parsed = match parser.parse_playback_event(&cmd.raw_payload)? {
+        Some(event) => event,
+        None => return Ok(()),
+    };
+
+    let external_metadata_id = parsed.tmdb_id.or(parsed.imdb_id);
+    let user_id = webhook_token.user_id().clone();
+
+    if let Some(ref ext_id) = external_metadata_id {
+        let one_hour_ago = chrono::Utc::now().naive_utc() - Duration::hours(1);
+        if ctx
+            .watch_event_repository
+            .find_duplicate(&user_id, ext_id, one_hour_ago)
+            .await?
+        {
+            return Ok(());
+        }
+    }
+
+    let watched_at = chrono::Utc::now().naive_utc();
+    let event = WatchEvent::new(
+        user_id,
+        parsed.title,
+        parsed.year,
+        external_metadata_id,
+        cmd.source,
+        watched_at,
+        None,
+    );
+
+    ctx.watch_event_repository.save(&event).await?;
+
+    let _ = ctx
+        .event_publisher
+        .publish(&DomainEvent::WatchEventIngested {
+            user_id: event.user_id().clone(),
+            title: event.title().to_string(),
+            source: event.source().to_string(),
+        })
+        .await;
+
+    Ok(())
+}

crates/application/src/use_cases/log_review.rs

@@ -66,7 +66,7 @@ mod tests {
 
     use domain::{
         models::Movie,
-        value_objects::{MovieId, MovieTitle, ReleaseYear},
+        value_objects::{MovieTitle, ReleaseYear},
     };
 
     use domain::ports::MovieRepository;

crates/application/src/use_cases/mod.rs

@@ -2,12 +2,16 @@ pub mod add_to_watchlist;
 pub mod apply_import_mapping;
 pub mod apply_import_profile;
 pub mod cleanup_expired_import_sessions;
+pub mod cleanup_watch_events;
+pub mod confirm_watch_events;
 pub mod create_import_session;
 pub mod delete_import_profile;
 pub mod delete_review;
+pub mod dismiss_watch_events;
 pub mod enrich_movie;
 pub mod execute_import;
 pub mod export_diary;
+pub mod generate_webhook_token;
 pub mod get_activity_feed;
 pub mod get_current_profile;
 pub mod get_diary;
@@ -20,8 +24,11 @@ pub mod get_remote_watchlist;
 pub mod get_review_history;
 pub mod get_user_profile;
 pub mod get_users;
+pub mod get_watch_queue;
 pub mod get_watchlist;
 pub mod get_watchlist_page;
+pub mod get_webhook_tokens;
+pub mod ingest_watch_event;
 pub mod is_on_watchlist;
 pub mod list_import_profiles;
 pub mod log_review;
@@ -29,6 +36,7 @@ pub mod login;
 pub mod register;
 pub mod register_and_login;
 pub mod remove_from_watchlist;
+pub mod revoke_webhook_token;
 pub mod save_import_profile;
 pub mod search;
 pub mod sync_poster;

crates/application/src/use_cases/revoke_webhook_token.rs

@@ -0,0 +1,14 @@
+use domain::{
+    errors::DomainError,
+    value_objects::{UserId, WebhookTokenId},
+};
+
+use crate::{commands::RevokeWebhookTokenCommand, context::AppContext};
+
+pub async fn execute(ctx: &AppContext, cmd: RevokeWebhookTokenCommand) -> Result<(), DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let token_id = WebhookTokenId::from_uuid(cmd.token_id);
+    ctx.webhook_token_repository
+        .delete(&token_id, &user_id)
+        .await
+}