importer feature

crates/application/src/commands.rs

@@ -1,5 +1,6 @@
 use chrono::NaiveDateTime;
 use domain::models::{ExportFormat, UserRole};
+use importer::FieldMapping;
 use uuid::Uuid;
 
 pub struct LogReviewCommand {
@@ -42,3 +43,44 @@ pub struct ExportCommand {
     pub user_id: Uuid,
     pub format: ExportFormat,
 }
+
+pub enum FileFormat {
+    Csv,
+    Json,
+    Xlsx,
+}
+
+pub struct CreateImportSessionCommand {
+    pub user_id: Uuid,
+    pub bytes: Vec<u8>,
+    pub format: FileFormat,
+}
+
+pub struct ApplyImportMappingCommand {
+    pub user_id: Uuid,
+    pub session_id: Uuid,
+    pub mappings: Vec<FieldMapping>,
+}
+
+pub struct ExecuteImportCommand {
+    pub user_id: Uuid,
+    pub session_id: Uuid,
+    pub confirmed_indices: Vec<usize>,
+}
+
+pub struct SaveImportProfileCommand {
+    pub user_id: Uuid,
+    pub session_id: Uuid,
+    pub name: String,
+}
+
+pub struct ApplyImportProfileCommand {
+    pub user_id: Uuid,
+    pub session_id: Uuid,
+    pub profile_id: Uuid,
+}
+
+pub struct DeleteImportProfileCommand {
+    pub user_id: Uuid,
+    pub profile_id: Uuid,
+}

crates/application/src/context.rs

@@ -1,9 +1,10 @@
 use std::sync::Arc;
 
 use domain::ports::{
-    AuthService, DiaryExporter, DiaryRepository, EventPublisher, MetadataClient, MovieRepository,
-    PasswordHasher, PosterFetcherClient, PosterStorage, ReviewRepository, StatsRepository,
-    UserRepository,
+    AuthService, DiaryExporter, DiaryRepository, EventPublisher,
+    ImportProfileRepository, ImportSessionRepository,
+    MetadataClient, MovieRepository, PasswordHasher, PosterFetcherClient,
+    PosterStorage, ReviewRepository, StatsRepository, UserRepository,
 };
 
 use crate::config::AppConfig;
@@ -22,5 +23,7 @@ pub struct AppContext {
     pub auth_service: Arc<dyn AuthService>,
     pub password_hasher: Arc<dyn PasswordHasher>,
     pub user_repository: Arc<dyn UserRepository>,
+    pub import_session_repository: Arc<dyn ImportSessionRepository>,
+    pub import_profile_repository: Arc<dyn ImportProfileRepository>,
     pub config: AppConfig,
 }

crates/application/src/event_handlers.rs

@@ -1,62 +0,0 @@
-use std::time::Duration;
-
-use async_trait::async_trait;
-use domain::ports::EventHandler;
-use domain::{errors::DomainError, events::DomainEvent};
-
-use crate::{commands::SyncPosterCommand, context::AppContext, use_cases::sync_poster};
-
-pub struct PosterSyncHandler {
-    ctx: AppContext,
-    max_retries: u32,
-}
-
-impl PosterSyncHandler {
-    pub fn new(ctx: AppContext, max_retries: u32) -> Self {
-        Self { ctx, max_retries }
-    }
-}
-
-#[async_trait]
-impl EventHandler for PosterSyncHandler {
-    async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
-        let (movie_id, external_metadata_id) = match event {
-            DomainEvent::MovieDiscovered {
-                movie_id,
-                external_metadata_id,
-            } => (movie_id.value(), external_metadata_id.value().to_owned()),
-            _ => return Ok(()),
-        };
-
-        let mut last_err: Option<DomainError> = None;
-        for attempt in 0..=self.max_retries {
-            let cmd = SyncPosterCommand {
-                movie_id,
-                external_metadata_id: external_metadata_id.clone(),
-            };
-            match sync_poster::execute(&self.ctx, cmd).await {
-                Ok(()) => return Ok(()),
-                Err(e) => {
-                    if attempt < self.max_retries {
-                        let delay = Duration::from_secs(2u64.pow(attempt));
-                        tracing::warn!(
-                            attempt = attempt + 1,
-                            max_attempts = self.max_retries + 1,
-                            delay_secs = delay.as_secs(),
-                            "poster sync failed, retrying: {e}"
-                        );
-                        tokio::time::sleep(delay).await;
-                    }
-                    last_err = Some(e);
-                }
-            }
-        }
-
-        let err = last_err.expect("loop runs at least once and always sets last_err on Err");
-        tracing::error!(
-            attempts = self.max_retries + 1,
-            "poster sync failed after all attempts: {err}"
-        );
-        Err(err)
-    }
-}

crates/application/src/lib.rs

@@ -1,5 +1,4 @@
 pub mod commands;
-pub mod event_handlers;
 pub mod worker;
 pub mod config;
 pub mod context;

crates/application/src/ports.rs

@@ -95,6 +95,45 @@ pub struct FollowersPageData {
     pub error: Option<String>,
 }
 
+pub struct ImportUploadPageData {
+    pub ctx: HtmlPageContext,
+    pub profiles: Vec<ImportProfileView>,
+    pub error: Option<String>,
+}
+
+pub struct ImportProfileView {
+    pub id: String,
+    pub name: String,
+}
+
+pub struct ImportMappingPageData {
+    pub ctx: HtmlPageContext,
+    pub session_id: String,
+    pub columns: Vec<String>,
+    pub sample_rows: Vec<Vec<String>>,
+    pub domain_fields: Vec<(&'static str, &'static str)>,
+    pub error: Option<String>,
+}
+
+pub struct ImportPreviewRow {
+    pub index: usize,
+    pub status: ImportRowStatus,
+    pub cells: Vec<String>,
+}
+
+pub enum ImportRowStatus {
+    Valid,
+    Duplicate,
+    Invalid(String),
+}
+
+pub struct ImportPreviewPageData {
+    pub ctx: HtmlPageContext,
+    pub session_id: String,
+    pub columns: Vec<String>,
+    pub rows: Vec<ImportPreviewRow>,
+}
+
 pub trait HtmlRenderer: Send + Sync {
     fn render_diary_page(
         &self,
@@ -109,6 +148,9 @@ pub trait HtmlRenderer: Send + Sync {
     fn render_profile_page(&self, data: ProfilePageData) -> Result<String, String>;
     fn render_following_page(&self, data: FollowingPageData) -> Result<String, String>;
     fn render_followers_page(&self, data: FollowersPageData) -> Result<String, String>;
+    fn render_import_upload_page(&self, data: ImportUploadPageData) -> Result<String, String>;
+    fn render_import_mapping_page(&self, data: ImportMappingPageData) -> Result<String, String>;
+    fn render_import_preview_page(&self, data: ImportPreviewPageData) -> Result<String, String>;
 }
 
 pub trait RssFeedRenderer: Send + Sync {

crates/application/src/use_cases/apply_import_mapping.rs

@@ -0,0 +1,62 @@
+use domain::{
+    errors::DomainError,
+    value_objects::{ExternalMetadataId, ImportSessionId, MovieTitle, ReleaseYear, UserId},
+};
+use importer::{AnnotatedRow, ParsedFile, apply_mapping};
+
+use crate::{commands::ApplyImportMappingCommand, context::AppContext};
+
+pub async fn execute(ctx: &AppContext, cmd: ApplyImportMappingCommand) -> Result<Vec<AnnotatedRow>, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let session_id = ImportSessionId::from_uuid(cmd.session_id);
+    let mappings = cmd.mappings;
+    let mut session = ctx.import_session_repository
+        .get(&session_id, &user_id)
+        .await?
+        .ok_or_else(|| DomainError::NotFound("import session".into()))?;
+
+    let parsed: ParsedFile = serde_json::from_str(&session.parsed_data)
+        .map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
+
+    let mut annotated = apply_mapping(&parsed, &mappings);
+
+    for row in annotated.iter_mut() {
+        if let importer::RowResult::Valid(ref import_row) = row.result {
+            row.is_duplicate = check_duplicate(ctx, import_row).await?;
+        }
+    }
+
+    session.field_mappings = Some(
+        serde_json::to_string(&mappings)
+            .map_err(|e| DomainError::InfrastructureError(e.to_string()))?
+    );
+    session.row_results = Some(
+        serde_json::to_string(&annotated)
+            .map_err(|e| DomainError::InfrastructureError(e.to_string()))?
+    );
+
+    ctx.import_session_repository.update(&session).await?;
+
+    Ok(annotated)
+}
+
+async fn check_duplicate(ctx: &AppContext, row: &importer::ImportRow) -> Result<bool, DomainError> {
+    if let Some(ext_id) = &row.external_metadata_id {
+        if let Ok(eid) = ExternalMetadataId::new(ext_id.clone()) {
+            if ctx.movie_repository.get_movie_by_external_id(&eid).await?.is_some() {
+                return Ok(true);
+            }
+        }
+    }
+    if let (Some(title), Some(year_str)) = (&row.title, &row.release_year) {
+        let title_vo = MovieTitle::new(title.clone());
+        let year_vo = year_str.parse::<u16>().ok().and_then(|y| ReleaseYear::new(y).ok());
+        if let (Ok(t), Some(y)) = (title_vo, year_vo) {
+            let matches = ctx.movie_repository.get_movies_by_title_and_year(&t, &y).await?;
+            if !matches.is_empty() {
+                return Ok(true);
+            }
+        }
+    }
+    Ok(false)
+}

crates/application/src/use_cases/apply_import_profile.rs

@@ -0,0 +1,20 @@
+use domain::{errors::DomainError, value_objects::{ImportProfileId, ImportSessionId, UserId}};
+use crate::{commands::ApplyImportProfileCommand, context::AppContext};
+
+/// Copies the profile's field_mappings onto the session. Caller must then invoke
+/// apply_import_mapping to regenerate row_results with the new mappings.
+pub async fn execute(ctx: &AppContext, cmd: ApplyImportProfileCommand) -> Result<(), DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let session_id = ImportSessionId::from_uuid(cmd.session_id);
+    let profile_id = ImportProfileId::from_uuid(cmd.profile_id);
+
+    let profile = ctx.import_profile_repository
+        .get(&profile_id, &user_id).await?
+        .ok_or_else(|| DomainError::NotFound("import profile".into()))?;
+    let mut session = ctx.import_session_repository
+        .get(&session_id, &user_id).await?
+        .ok_or_else(|| DomainError::NotFound("import session".into()))?;
+    session.field_mappings = Some(profile.field_mappings);
+    session.row_results = None;
+    ctx.import_session_repository.update(&session).await
+}

crates/application/src/use_cases/cleanup_expired_import_sessions.rs

@@ -0,0 +1,6 @@
+use domain::errors::DomainError;
+use crate::context::AppContext;
+
+pub async fn execute(ctx: &AppContext) -> Result<u64, DomainError> {
+    ctx.import_session_repository.delete_expired().await
+}

crates/application/src/use_cases/create_import_session.rs

@@ -0,0 +1,44 @@
+use chrono::Utc;
+use domain::{errors::DomainError, models::ImportSession, value_objects::{ImportSessionId, UserId}};
+use importer::{ImportError, ParsedFile};
+
+use crate::{commands::{CreateImportSessionCommand, FileFormat}, context::AppContext};
+
+pub struct CreateSessionResult {
+    pub session_id: ImportSessionId,
+    pub columns: Vec<String>,
+    pub sample_rows: Vec<Vec<String>>,
+}
+
+pub async fn execute(ctx: &AppContext, cmd: CreateImportSessionCommand) -> Result<CreateSessionResult, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    ctx.import_session_repository.delete_expired_for_user(&user_id).await?;
+
+    let parsed = parse(cmd.bytes, cmd.format).map_err(|e| DomainError::ValidationError(e.to_string()))?;
+    let sample_rows = parsed.rows.iter().take(5).cloned().collect();
+    let columns = parsed.columns.clone();
+
+    let parsed_data = serde_json::to_string(&parsed)
+        .map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
+
+    let now = Utc::now().naive_utc();
+    let session = ImportSession::new(ImportSessionId::generate(), user_id, parsed_data, now);
+    let session_id = session.id.clone();
+
+    ctx.import_session_repository.create(&session).await?;
+
+    Ok(CreateSessionResult { session_id, columns, sample_rows })
+}
+
+fn parse(bytes: Vec<u8>, format: FileFormat) -> Result<ParsedFile, ImportError> {
+    match format {
+        FileFormat::Csv => importer::parse_csv(&bytes),
+        FileFormat::Json => importer::parse_json(&bytes),
+        FileFormat::Xlsx => {
+            #[cfg(feature = "xlsx")]
+            { importer::parse_xlsx(&bytes) }
+            #[cfg(not(feature = "xlsx"))]
+            { Err(ImportError::Xlsx("XLSX support not compiled in".into())) }
+        }
+    }
+}

crates/application/src/use_cases/delete_import_profile.rs

@@ -0,0 +1,12 @@
+use domain::{errors::DomainError, value_objects::{ImportProfileId, UserId}};
+use crate::{commands::DeleteImportProfileCommand, context::AppContext};
+
+pub async fn execute(ctx: &AppContext, cmd: DeleteImportProfileCommand) -> Result<(), DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let profile_id = ImportProfileId::from_uuid(cmd.profile_id);
+
+    ctx.import_profile_repository
+        .get(&profile_id, &user_id).await?
+        .ok_or_else(|| DomainError::NotFound("import profile".into()))?;
+    ctx.import_profile_repository.delete(&profile_id).await
+}

crates/application/src/use_cases/execute_import.rs

@@ -0,0 +1,84 @@
+use chrono::NaiveDateTime;
+use domain::{errors::DomainError, value_objects::{ImportSessionId, UserId}};
+use importer::{AnnotatedRow, ImportRow, RowResult};
+use uuid::Uuid;
+
+use crate::{commands::{ExecuteImportCommand, LogReviewCommand}, context::AppContext, use_cases::log_review};
+
+pub struct ImportSummary {
+    pub imported: usize,
+    pub skipped_duplicates: usize,
+    pub failed: Vec<(usize, String)>,
+}
+
+pub async fn execute(ctx: &AppContext, cmd: ExecuteImportCommand) -> Result<ImportSummary, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let session_id = ImportSessionId::from_uuid(cmd.session_id);
+    let confirmed_indices = cmd.confirmed_indices;
+    let session = ctx.import_session_repository
+        .get(&session_id, &user_id)
+        .await?
+        .ok_or_else(|| DomainError::NotFound("import session".into()))?;
+
+    let row_results: Vec<AnnotatedRow> = session.row_results
+        .as_deref()
+        .and_then(|s| serde_json::from_str(s).ok())
+        .unwrap_or_default();
+
+    let confirmed_set: std::collections::HashSet<usize> = confirmed_indices.into_iter().collect();
+
+    let mut imported = 0;
+    let mut skipped_duplicates = 0;
+    let mut failed = Vec::new();
+
+    for (idx, annotated) in row_results.into_iter().enumerate() {
+        if !confirmed_set.contains(&idx) {
+            skipped_duplicates += 1;
+            continue;
+        }
+        match annotated.result {
+            RowResult::Valid(row) => {
+                match row_to_command(&row, user_id.value()) {
+                    Ok(cmd) => {
+                        match log_review::execute(ctx, cmd).await {
+                            Ok(_) => imported += 1,
+                            Err(e) => failed.push((idx, e.to_string())),
+                        }
+                    }
+                    Err(e) => failed.push((idx, e)),
+                }
+            }
+            RowResult::Invalid { errors, .. } => {
+                failed.push((idx, errors.join("; ")));
+            }
+        }
+    }
+
+    ctx.import_session_repository.delete(&session_id).await?;
+
+    Ok(ImportSummary { imported, skipped_duplicates, failed })
+}
+
+fn row_to_command(row: &ImportRow, user_id: Uuid) -> Result<LogReviewCommand, String> {
+    let rating = row.rating.as_deref()
+        .ok_or("missing rating")?
+        .parse::<u8>()
+        .map_err(|_| "rating is not a valid u8".to_string())?;
+
+    let watched_at_str = row.watched_at.as_deref().ok_or("missing watched_at")?;
+    let watched_at = NaiveDateTime::parse_from_str(&format!("{} 00:00:00", watched_at_str), "%Y-%m-%d %H:%M:%S")
+        .or_else(|_| NaiveDateTime::parse_from_str(watched_at_str, "%Y-%m-%d %H:%M:%S"))
+        .or_else(|_| NaiveDateTime::parse_from_str(watched_at_str, "%Y-%m-%dT%H:%M:%S"))
+        .map_err(|_| format!("cannot parse watched_at: '{}'", watched_at_str))?;
+
+    Ok(LogReviewCommand {
+        external_metadata_id: row.external_metadata_id.clone(),
+        manual_title: row.title.clone(),
+        manual_release_year: row.release_year.as_deref().and_then(|s| s.parse().ok()),
+        manual_director: row.director.clone(),
+        user_id,
+        rating,
+        comment: row.comment.clone(),
+        watched_at,
+    })
+}

crates/application/src/use_cases/list_import_profiles.rs

@@ -0,0 +1,6 @@
+use domain::{errors::DomainError, models::ImportProfile, value_objects::UserId};
+use crate::context::AppContext;
+
+pub async fn execute(ctx: &AppContext, user_id: &UserId) -> Result<Vec<ImportProfile>, DomainError> {
+    ctx.import_profile_repository.list_for_user(user_id).await
+}

crates/application/src/use_cases/mod.rs

@@ -1,4 +1,12 @@
+pub mod apply_import_mapping;
+pub mod apply_import_profile;
+pub mod cleanup_expired_import_sessions;
+pub mod create_import_session;
+pub mod delete_import_profile;
 pub mod delete_review;
+pub mod execute_import;
+pub mod list_import_profiles;
+pub mod save_import_profile;
 pub mod export_diary;
 pub mod get_activity_feed;
 pub mod get_diary;

crates/application/src/use_cases/save_import_profile.rs

@@ -0,0 +1,18 @@
+use chrono::Utc;
+use domain::{errors::DomainError, models::ImportProfile, value_objects::{ImportProfileId, ImportSessionId, UserId}};
+use crate::{commands::SaveImportProfileCommand, context::AppContext};
+
+pub async fn execute(ctx: &AppContext, cmd: SaveImportProfileCommand) -> Result<ImportProfileId, DomainError> {
+    let user_id = UserId::from_uuid(cmd.user_id);
+    let session_id = ImportSessionId::from_uuid(cmd.session_id);
+
+    let session = ctx.import_session_repository
+        .get(&session_id, &user_id).await?
+        .ok_or_else(|| DomainError::NotFound("import session".into()))?;
+    let mappings = session.field_mappings
+        .ok_or_else(|| DomainError::ValidationError("no mapping applied to this session yet".into()))?;
+    let profile = ImportProfile::new(ImportProfileId::generate(), user_id, cmd.name, mappings, Utc::now().naive_utc());
+    let id = profile.id.clone();
+    ctx.import_profile_repository.save(&profile).await?;
+    Ok(id)
+}