//! Generic import infrastructure for bringing external sessions into the archive.
//!
//! This module provides the shared types and orchestration logic that all importers
//! (Claude, ChatGPT, etc.) reuse. Each importer implements discovery and message
//! conversion, then delegates to [`import_sessions`] for the actual import.

pub mod progress;
pub mod registry;
pub mod sources;
pub mod trait_def;

/// Backwards-compatible re-export — external callers (e.g. `api`) import
/// `dirigent_archivist::import::claude::{discover_claude_import,
/// import_claude_sessions}`. Keep the path stable until those callsites
/// migrate to the `Importer` trait.
pub use sources::claude;
#[cfg(feature = "importer-claude")]
pub use sources::claude::ClaudeImporter;

pub use progress::{ImportProgressEvent, ImportProgressSink, SessionOutcome, StatsDelta};
pub use registry::ImporterRegistry;
pub use trait_def::{ConfigField, ConfigFieldKind, ImportConfig, ImportConfigShape, ImportError, ImportTarget, Importer, ImporterInfo};

use std::collections::HashMap;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::coordinator::Archivist;
use crate::error::{ArchivistError, Result};
use crate::types::{
    MessageRecord, RegisterConnectorRequest, RegisterSessionRequest, RegisterStatus,
    SessionCompleteness,
};

/// Statistics collected during an import operation.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
pub struct ImportStats {
    /// Number of sessions found by the importer's discovery phase.
    pub sessions_discovered: usize,
    /// Number of sessions successfully imported as new.
    pub sessions_imported: usize,
    /// Number of sessions skipped (already present with same or more messages).
    pub sessions_skipped: usize,
    /// Number of sessions that were updated with new messages.
    pub sessions_updated: usize,
    /// Total number of message records written to the archive.
    pub messages_written: usize,
    /// Number of messages that were already present (from existing sessions).
    pub messages_already_present: usize,
    /// Number of sessions skipped because the fingerprint matched (no source changes).
    #[serde(default)]
    pub sessions_fingerprint_skipped: usize,
    /// Errors encountered during import (non-fatal; import continues).
    pub errors: Vec<String>,
}

impl ImportStats {
    /// Total sessions processed (imported + skipped + updated + errored).
    pub fn total_sessions_processed(&self) -> usize {
        self.sessions_imported + self.sessions_skipped + self.sessions_updated + self.errors.len()
    }

    /// Whether any errors were encountered during import.
    pub fn has_errors(&self) -> bool {
        !self.errors.is_empty()
    }
}

/// Intermediate representation for a session discovered by any importer.
///
/// This is source-agnostic: each importer converts its native session format
/// into `DiscoveredSession` before handing it to [`import_sessions`].
#[derive(Debug, Clone)]
pub struct DiscoveredSession {
    /// The session ID from the original source (e.g., Claude's JSONL filename).
    pub native_session_id: String,
    /// Human-readable session title, if available.
    pub title: Option<String>,
    /// When the session was created in the source system.
    pub created_at: Option<DateTime<Utc>>,
    /// When the session was last updated in the source system.
    pub updated_at: Option<DateTime<Utc>>,
    /// Number of messages in the source session (used for skip/update decisions).
    pub message_count: usize,
    /// Arbitrary source-specific metadata preserved for provenance.
    pub metadata: serde_json::Value,
    /// Project path associated with the session, if known.
    pub project_path: Option<String>,
    /// Size of the source file in bytes, if available. Used for fingerprint-based
    /// change detection to skip unchanged sessions on re-import.
    pub file_size: Option<u64>,
}

/// Snapshot of source-side signals captured after a successful import.
///
/// Stored in the session's `metadata` JSON under the `"_import_snapshot"` key.
/// On re-import, comparing the current `DiscoveredSession` against the stored
/// snapshot lets us skip expensive full-parse when nothing has changed (O(1) gate).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ImportSnapshot {
    /// Number of messages in the source at the time of import.
    pub source_message_count: usize,
    /// Source-side `updated_at` timestamp at the time of import.
    pub source_updated_at: Option<DateTime<Utc>>,
    /// Source file size in bytes at the time of import.
    pub source_file_size: Option<u64>,
    /// When this snapshot was recorded.
    pub imported_at: DateTime<Utc>,
}

/// Key used to store [`ImportSnapshot`] in session metadata JSON.
const IMPORT_SNAPSHOT_KEY: &str = "_import_snapshot";

impl ImportSnapshot {
    /// Check whether the source signals in `discovered` match this snapshot.
    ///
    /// Returns `true` if all present signals match, meaning the session has not
    /// changed since this snapshot was taken and a full re-parse can be skipped.
    pub fn matches(&self, discovered: &DiscoveredSession) -> bool {
        if self.source_message_count != discovered.message_count {
            return false;
        }
        if self.source_updated_at != discovered.updated_at {
            return false;
        }
        // file_size: only compare when both sides have a value.
        if let (Some(snap_size), Some(disc_size)) = (self.source_file_size, discovered.file_size) {
            if snap_size != disc_size {
                return false;
            }
        }
        true
    }

    /// Build a snapshot from a discovered session (captures current source signals).
    pub fn from_discovered(discovered: &DiscoveredSession) -> Self {
        Self {
            source_message_count: discovered.message_count,
            source_updated_at: discovered.updated_at,
            source_file_size: discovered.file_size,
            imported_at: Utc::now(),
        }
    }

    /// Try to deserialize a snapshot from a session's metadata JSON.
    pub fn from_metadata(metadata: &serde_json::Value) -> Option<Self> {
        metadata
            .get(IMPORT_SNAPSHOT_KEY)
            .and_then(|v| serde_json::from_value(v.clone()).ok())
    }

    /// Serialize this snapshot into the session's metadata JSON under the
    /// `_import_snapshot` key.
    pub fn write_to_metadata(&self, metadata: &mut serde_json::Value) {
        if let Some(obj) = metadata.as_object_mut() {
            if let Ok(val) = serde_json::to_value(self) {
                obj.insert(IMPORT_SNAPSHOT_KEY.to_string(), val);
            }
        } else {
            tracing::warn!("cannot write import snapshot: metadata is not a JSON object");
        }
    }
}

/// Summary returned by the discovery phase before actual import begins.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ImportDiscovery {
    /// Human-readable name of the import source (e.g., "Claude Code").
    pub source_name: String,
    /// Filesystem path or URI that was scanned.
    pub source_path: String,
    /// Projects discovered, grouped by name.
    pub projects: Vec<ImportProject>,
    /// Total number of sessions found across all projects.
    pub total_sessions: usize,
    /// Estimated total messages across all discovered sessions.
    pub total_estimated_messages: usize,
}

/// A project grouping within an import discovery result.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ImportProject {
    /// Project name (typically derived from the directory path).
    pub name: String,
    /// Number of sessions belonging to this project.
    pub session_count: usize,
}

/// Resolves the `updated_at` timestamp for an imported session.
///
/// Prefers the source-provided timestamp from `discovered.updated_at`; falls
/// back to `Utc::now()` only when the source does not supply one.
fn resolve_updated_at(discovered: &DiscoveredSession) -> DateTime<Utc> {
    discovered.updated_at.unwrap_or_else(chrono::Utc::now)
}

/// Generic async orchestrator that imports discovered sessions into the archive.
///
/// This function handles the full import lifecycle:
/// 1. Registers the connector (idempotent via fingerprint).
/// 2. For each discovered session, checks whether it already exists in the archive.
/// 3. New sessions are registered and their messages are converted and appended.
/// 4. Existing sessions with fewer archived messages are logged and skipped (v1).
/// 5. Existing sessions with the same or more archived messages are skipped.
///
/// The `convert_messages` closure receives a `native_session_id` and returns
/// `MessageRecord`s with `Uuid::nil()` in the `session` field. This function
/// patches each record's `session` to the real `scroll_id` before appending.
///
/// # Arguments
///
/// * `archivist` - The archivist to import into.
/// * `connector_req` - Registration request for the import connector.
/// * `sessions` - Sessions discovered by the importer.
/// * `convert_messages` - Closure that converts a native session into `MessageRecord`s.
/// * `archive` - Optional archive name (`None` for default archive).
/// * `progress` - Sink for per-session progress events (use
///   [`ImportProgressSink::noop`] when progress reporting is not needed).
pub async fn import_sessions<F>(
    archivist: &Archivist,
    connector_req: RegisterConnectorRequest,
    sessions: Vec<DiscoveredSession>,
    convert_messages: F,
    archive: Option<String>,
    progress: &ImportProgressSink,
    force_deep_scan: bool,
    project_map: &HashMap<String, String>,
) -> Result<ImportStats>
where
    F: Fn(&str) -> Result<Vec<MessageRecord>> + Send + Sync,
{
    let mut stats = ImportStats::default();
    stats.sessions_discovered = sessions.len();

    // Step 1: Register the connector (idempotent).
    let connector_resp = archivist
        .register_connector(connector_req, archive.clone())
        .await?;
    let connector_uid = connector_resp.connector_uid;

    tracing::info!(
        connector_uid = %connector_uid,
        status = ?connector_resp.status,
        "Import connector registered"
    );

    // Step 2: Process each discovered session.
    let total = sessions.len();
    for (index, session) in sessions.iter().enumerate() {
        let native_id = &session.native_session_id;

        progress
            .send(ImportProgressEvent::SessionStarted {
                native_id: native_id.clone(),
                index,
                total,
            })
            .await;

        // Per-session outcome + stats delta. Updated as we go; on the early
        // `continue` paths we emit Failed/Skipped before moving on.
        let mut messages_written_delta: u64 = 0;
        let mut messages_already_present_delta: u64 = 0;
        let mut session_changed = false;

        // Helper: emit SessionFinished and fall out of the iteration.
        macro_rules! emit_finished {
            ($outcome:expr) => {{
                progress
                    .send(ImportProgressEvent::SessionFinished {
                        native_id: native_id.clone(),
                        outcome: $outcome,
                        stats_delta: StatsDelta {
                            messages_written: messages_written_delta,
                            messages_already_present: messages_already_present_delta,
                        },
                    })
                    .await;
            }};
        }

        // --- Step 1: Resolve or create scroll_id BEFORE convert_messages ---
        let (scroll_id, session_is_new) = match archivist
            .resolve_session(connector_uid, native_id, archive.clone())
            .await
        {
            Ok(id) => (id, false),
            Err(ArchivistError::SessionUnknown(_)) => {
                // Inject project_id from project_map if the session has a
                // project_path that maps to a known project.
                let mut metadata = session.metadata.clone();
                if let Some(project_path) = session.project_path.as_deref() {
                    if let Some(pid) = project_map.get(project_path) {
                        if let Some(obj) = metadata.as_object_mut() {
                            obj.insert(
                                "project_id".to_string(),
                                serde_json::Value::String(pid.clone()),
                            );
                        }
                    }
                }

                let register_req = RegisterSessionRequest {
                    connector_uid,
                    native_session_id: native_id.clone(),
                    title: session.title.clone(),
                    custom_scroll_id: None,
                    metadata,
                    completeness: SessionCompleteness::Complete,
                    parent_scroll_id: None,
                    is_subagent: false,
                    continuation: None,
                    agent_id: None,
                    subagent_type: None,
                    spawning_tool_use_id: None,
                };
                match archivist
                    .register_session(register_req, archive.clone())
                    .await
                {
                    Ok(resp) => match resp.status {
                        RegisterStatus::Accepted => (resp.scroll_id, true),
                        RegisterStatus::Aliased => {
                            stats.sessions_skipped += 1;
                            emit_finished!(SessionOutcome::Skipped);
                            continue;
                        }
                        RegisterStatus::Rejected => {
                            stats.errors.push(format!(
                                "Session registration rejected for {native_id}"
                            ));
                            emit_finished!(SessionOutcome::Failed);
                            continue;
                        }
                    },
                    Err(e) => {
                        stats.errors.push(format!(
                            "Failed to register session {native_id}: {e}"
                        ));
                        emit_finished!(SessionOutcome::Failed);
                        continue;
                    }
                }
            }
            Err(e) => {
                stats.errors.push(format!(
                    "Failed to resolve session {native_id}: {e}"
                ));
                emit_finished!(SessionOutcome::Failed);
                continue;
            }
        };

        // --- Step 2: Hoist metadata read for existing sessions ---
        // Load metadata once; reused for fingerprint check AND title/model diff.
        let existing_meta = if !session_is_new {
            match archivist
                .get_session_metadata(scroll_id, archive.clone())
                .await
            {
                Ok(m) => Some(m),
                Err(e) => {
                    stats.errors.push(format!(
                        "Failed to read session metadata for {native_id}: {e}"
                    ));
                    emit_finished!(SessionOutcome::Failed);
                    continue;
                }
            }
        } else {
            None
        };

        // --- Step 2b: Retroactive project_id linking for existing sessions ---
        // Sessions imported before project detection (or before the project was
        // created) have project_path but no project_id. Patch it now if the
        // project_map has a match — this runs even for fingerprint-skipped
        // sessions so re-import can link them without any source-side changes.
        if !session_is_new {
            if let Some(ref meta) = existing_meta {
                let has_project_path = meta
                    .metadata
                    .get("project_path")
                    .and_then(|v| v.as_str())
                    .is_some();
                let has_project_id = meta
                    .metadata
                    .get("project_id")
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                    .is_some();

                if has_project_path && !has_project_id {
                    let stored_path = meta
                        .metadata
                        .get("project_path")
                        .and_then(|v| v.as_str())
                        .unwrap();
                    if let Some(pid) = project_map.get(stored_path) {
                        if let Ok(primary) =
                            archivist.resolve_primary(archive.clone()).await
                        {
                            let mut patched = meta.clone();
                            if let Some(obj) = patched.metadata.as_object_mut() {
                                obj.insert(
                                    "project_id".to_string(),
                                    serde_json::Value::String(pid.clone()),
                                );
                            }
                            patched.updated_at = resolve_updated_at(session);
                            match primary.backend.put_session(patched).await {
                                Ok(_) => {
                                    tracing::info!(
                                        scroll_id = %scroll_id,
                                        project_id = %pid,
                                        "Retroactively linked session to project"
                                    );
                                    session_changed = true;
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        scroll_id = %scroll_id,
                                        error = %e,
                                        "Failed to retroactively link session to project"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        }

        // --- Step 3: Fingerprint gate — skip unchanged sessions ---
        if !session_is_new && !force_deep_scan {
            if let Some(ref meta) = existing_meta {
                if let Some(snapshot) = ImportSnapshot::from_metadata(&meta.metadata) {
                    if snapshot.matches(session) {
                        stats.sessions_fingerprint_skipped += 1;
                        if session_changed {
                            tracing::debug!(
                                native_id = %native_id,
                                "Fingerprint match — skipping message scan (metadata was updated)"
                            );
                            stats.sessions_updated += 1;
                            emit_finished!(SessionOutcome::Updated);
                        } else {
                            tracing::debug!(
                                native_id = %native_id,
                                "Fingerprint match — skipping unchanged session"
                            );
                            stats.sessions_skipped += 1;
                            emit_finished!(SessionOutcome::Skipped);
                        }
                        continue;
                    }
                }
            }
        }

        // --- Step 4: Convert messages (EXPENSIVE — after fingerprint gate) ---
        let source_records = match convert_messages(native_id) {
            Ok(r) => r,
            Err(e) => {
                stats.errors.push(format!(
                    "Failed to convert messages for session {native_id}: {e}"
                ));
                emit_finished!(SessionOutcome::Failed);
                continue;
            }
        };

        // Build existing_ids set — empty for brand-new sessions.
        let existing_ids: std::collections::HashSet<Uuid> = if session_is_new {
            std::collections::HashSet::new()
        } else {
            match archivist.get_messages(scroll_id, archive.clone()).await {
                Ok(msgs) => msgs.into_iter().map(|m| m.message_id).collect(),
                Err(e) => {
                    stats.errors.push(format!(
                        "Failed to read existing messages for session {native_id}: {e}"
                    ));
                    emit_finished!(SessionOutcome::Failed);
                    continue;
                }
            }
        };

        // Patch placeholder session field and partition.
        let mut new_messages: Vec<MessageRecord> = Vec::new();
        let mut already_present_count: usize = 0;
        for mut record in source_records {
            if record.session == Uuid::nil() {
                record.session = scroll_id;
            }
            if existing_ids.contains(&record.message_id) {
                already_present_count += 1;
            } else {
                new_messages.push(record);
            }
        }

        let new_count = new_messages.len();
        if new_count > 0 {
            if let Err(e) = archivist
                .append_messages(scroll_id, new_messages, archive.clone())
                .await
            {
                stats.errors.push(format!(
                    "Failed to append messages for session {native_id}: {e}"
                ));
                emit_finished!(SessionOutcome::Failed);
                continue;
            }
            stats.messages_written += new_count;
            messages_written_delta = new_count as u64;
            session_changed = true;
        }
        stats.messages_already_present += already_present_count;
        messages_already_present_delta = already_present_count as u64;

        // --- Step 5: Metadata diff (reuse hoisted metadata) ---
        if !session_is_new {
            // SAFETY: existing_meta is Some when !session_is_new (guarded above).
            let current_meta = existing_meta.unwrap();

            let new_title = session.title.as_ref();
            let title_differs = new_title.is_some() && new_title != current_meta.title.as_ref();

            let new_model = session
                .metadata
                .get("model")
                .and_then(|v| v.as_str())
                .map(String::from);
            let current_model = current_meta
                .metadata
                .get("model")
                .and_then(|v| v.as_str())
                .map(String::from);
            let model_differs = new_model.is_some() && new_model != current_model;

            if title_differs || model_differs {
                if let Err(e) = archivist
                    .update_session_metadata(
                        scroll_id,
                        if title_differs { new_title.cloned() } else { None },
                        if model_differs { new_model } else { None },
                        archive.clone(),
                    )
                    .await
                {
                    stats.errors.push(format!(
                        "Failed to update session metadata for {native_id}: {e}"
                    ));
                    emit_finished!(SessionOutcome::Failed);
                    continue;
                }
                session_changed = true;
            }

            let new_project_path = session
                .metadata
                .get("project_path")
                .and_then(|v| v.as_str())
                .map(String::from);
            let current_project_path = current_meta
                .metadata
                .get("project_path")
                .and_then(|v| v.as_str())
                .map(String::from);
            let project_path_differs =
                new_project_path.is_some() && new_project_path != current_project_path;

            if project_path_differs {
                // project_path lives in the free-form metadata JSON.
                // Re-read to pick up any title/model changes applied above.
                let mut patched_meta = archivist
                    .get_session_metadata(scroll_id, archive.clone())
                    .await
                    .unwrap_or(current_meta);
                if let Some(obj) = patched_meta.metadata.as_object_mut() {
                    let path_val = new_project_path.clone().unwrap_or_default();
                    obj.insert(
                        "project_path".to_string(),
                        serde_json::Value::String(path_val.clone()),
                    );
                    if let Some(pid) = project_map.get(&path_val) {
                        obj.insert(
                            "project_id".to_string(),
                            serde_json::Value::String(pid.clone()),
                        );
                    }
                }
                patched_meta.updated_at = resolve_updated_at(session);
                if let Ok(primary) = archivist.resolve_primary(archive.clone()).await {
                    if let Err(e) = primary.backend.put_session(patched_meta).await {
                        tracing::warn!(
                            scroll_id = %scroll_id,
                            error = %e,
                            "Failed to update project_path in session metadata"
                        );
                    }
                }
                session_changed = true;
            }
        }

        // --- Step 6: Write import snapshot after successful import/update ---
        {
            let snapshot = ImportSnapshot::from_discovered(session);
            // Re-read metadata to get the latest state (may have been updated above).
            let write_result = async {
                let mut meta = archivist
                    .get_session_metadata(scroll_id, archive.clone())
                    .await?;
                snapshot.write_to_metadata(&mut meta.metadata);
                meta.updated_at = resolve_updated_at(session);
                let primary = archivist.resolve_primary(archive.clone()).await?;
                primary.backend.put_session(meta).await.map_err(|e| {
                    ArchivistError::InvalidRequest(format!(
                        "Failed to write import snapshot: {e}"
                    ))
                })
            }
            .await;
            if let Err(e) = write_result {
                tracing::warn!(
                    scroll_id = %scroll_id,
                    error = %e,
                    "Failed to write import snapshot (session still imported)"
                );
            }
        }

        // Accounting: exactly one of {imported, updated, skipped} per session.
        let outcome = if session_is_new {
            stats.sessions_imported += 1;
            SessionOutcome::Imported
        } else if session_changed {
            stats.sessions_updated += 1;
            SessionOutcome::Updated
        } else {
            stats.sessions_skipped += 1;
            SessionOutcome::Skipped
        };

        emit_finished!(outcome);
    }

    Ok(stats)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_import_stats_default() {
        let stats = ImportStats::default();
        assert_eq!(stats.sessions_discovered, 0);
        assert_eq!(stats.sessions_imported, 0);
        assert_eq!(stats.sessions_skipped, 0);
        assert_eq!(stats.sessions_updated, 0);
        assert_eq!(stats.messages_written, 0);
        assert_eq!(stats.messages_already_present, 0);
        assert!(stats.errors.is_empty());
    }

    #[test]
    fn test_import_stats_total_sessions_processed() {
        let mut stats = ImportStats::default();
        stats.sessions_imported = 3;
        stats.sessions_skipped = 2;
        stats.sessions_updated = 1;
        stats.errors.push("oops".to_string());
        assert_eq!(stats.total_sessions_processed(), 7);
    }

    #[test]
    fn test_import_stats_has_errors() {
        let mut stats = ImportStats::default();
        assert!(!stats.has_errors());
        stats.errors.push("something went wrong".to_string());
        assert!(stats.has_errors());
    }
}

#[cfg(test)]
mod idempotency_tests {
    use super::*;
    use crate::Archivist;
    use chrono::Utc;
    use uuid::Uuid;

    async fn mk() -> (Archivist, std::path::PathBuf) {
        let tmp = std::env::temp_dir().join(format!("import_idem_{}", Uuid::now_v7()));
        // Use `from_single_backend` rather than `new_with_single_archive` so
        // each test's archive is fully self-contained (no shared `.archives.json`
        // in the parent tempdir racing against sibling tests).
        let backend = std::sync::Arc::new(
            crate::backends::JsonlBackend::new(tmp.clone()).await.unwrap(),
        );
        let a = Archivist::from_single_backend("main".into(), backend)
            .await
            .unwrap();
        (a, tmp)
    }

    fn connector() -> RegisterConnectorRequest {
        // Stable client_native_id so that re-registering within the same test
        // (which uses an isolated temp dir per test) aliases onto the same
        // connector_uid — otherwise each call would produce a fresh connector
        // and defeat idempotency.
        RegisterConnectorRequest {
            r#type: "Fake".into(),
            title: "fake".into(),
            client_native_id: "fake@local:stable".into(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        }
    }

    fn record(session: Uuid, id: Uuid, role: &str, content: &str) -> MessageRecord {
        MessageRecord {
            version: 1,
            message_id: id,
            session,
            parent_id: None,
            ts: Utc::now(),
            role: role.to_string(),
            author: None,
            content_md: content.to_string(),
            content_parts: None,
            attachments: Vec::new(),
            metadata: serde_json::json!({}),
        }
    }

    #[tokio::test]
    async fn import_skips_already_present_messages() {
        let (archivist, tmp) = mk().await;

        let a = Uuid::now_v7();
        let b = Uuid::now_v7();
        let c = Uuid::now_v7();

        let discovered = vec![DiscoveredSession {
            native_session_id: "s1".into(),
            title: Some("t".into()),
            created_at: None,
            updated_at: None,
            message_count: 3,
            metadata: serde_json::json!({}),
            project_path: None,
            file_size: None,
        }];
        let convert = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![
                record(Uuid::nil(), a, "user", "hi-a"),
                record(Uuid::nil(), b, "user", "hi-b"),
                record(Uuid::nil(), c, "user", "hi-c"),
            ])
        };
        let stats = import_sessions(&archivist, connector(), discovered.clone(), convert, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();
        assert_eq!(stats.sessions_imported, 1);
        assert_eq!(stats.messages_written, 3);

        // Re-import with IDENTICAL records — nothing should be written.
        let convert2 = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![
                record(Uuid::nil(), a, "user", "hi-a"),
                record(Uuid::nil(), b, "user", "hi-b"),
                record(Uuid::nil(), c, "user", "hi-c"),
            ])
        };
        let stats2 = import_sessions(&archivist, connector(), discovered, convert2, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();
        assert_eq!(stats2.messages_written, 0);
        assert_eq!(stats2.messages_already_present, 3);
        assert_eq!(stats2.sessions_skipped, 1);
        assert_eq!(stats2.sessions_imported, 0);
        assert_eq!(stats2.sessions_updated, 0);

        let _ = tokio::fs::remove_dir_all(tmp).await;
    }

    #[tokio::test]
    async fn import_appends_new_messages_only() {
        let (archivist, tmp) = mk().await;

        let a = Uuid::now_v7();
        let b = Uuid::now_v7();
        let c = Uuid::now_v7();
        let d = Uuid::now_v7();

        let discovered = vec![DiscoveredSession {
            native_session_id: "s1".into(),
            title: Some("t".into()),
            created_at: None,
            updated_at: None,
            message_count: 2,
            metadata: serde_json::json!({}),
            project_path: None,
            file_size: None,
        }];
        let convert1 = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![
                record(Uuid::nil(), a, "user", "hi-a"),
                record(Uuid::nil(), b, "user", "hi-b"),
            ])
        };
        let _ = import_sessions(&archivist, connector(), discovered.clone(), convert1, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();

        // Second run: source has grown to 4 messages.
        let convert2 = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![
                record(Uuid::nil(), a, "user", "hi-a"),
                record(Uuid::nil(), b, "user", "hi-b"),
                record(Uuid::nil(), c, "user", "hi-c"),
                record(Uuid::nil(), d, "user", "hi-d"),
            ])
        };
        let stats = import_sessions(&archivist, connector(), discovered, convert2, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();
        assert_eq!(stats.messages_written, 2);
        assert_eq!(stats.messages_already_present, 2);
        assert_eq!(stats.sessions_updated, 1);
        assert_eq!(stats.sessions_skipped, 0);
        assert_eq!(stats.sessions_imported, 0);

        let _ = tokio::fs::remove_dir_all(tmp).await;
    }

    #[tokio::test]
    async fn import_updates_metadata_only() {
        let (archivist, tmp) = mk().await;

        let a = Uuid::now_v7();
        let convert = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![record(Uuid::nil(), a, "user", "hi")])
        };

        let first = vec![DiscoveredSession {
            native_session_id: "s1".into(),
            title: Some("old title".into()),
            created_at: None,
            updated_at: None,
            message_count: 1,
            metadata: serde_json::json!({}),
            project_path: None,
            file_size: None,
        }];
        let _ = import_sessions(&archivist, connector(), first, convert, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();

        // Re-import with same messages but new title.
        let second = vec![DiscoveredSession {
            native_session_id: "s1".into(),
            title: Some("new title".into()),
            created_at: None,
            updated_at: None,
            message_count: 1,
            metadata: serde_json::json!({}),
            project_path: None,
            file_size: None,
        }];
        let convert2 = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![record(Uuid::nil(), a, "user", "hi")])
        };
        let stats = import_sessions(&archivist, connector(), second, convert2, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();
        assert_eq!(stats.messages_written, 0);
        assert_eq!(stats.sessions_updated, 1);
        assert_eq!(stats.sessions_skipped, 0);

        // Verify title landed on disk.
        let meta_list = archivist
            .list_sessions_paged(
                crate::types::SessionListQuery::default().with_limit(50),
            )
            .await
            .unwrap();
        assert!(meta_list.items.iter().any(|m| m.title.as_deref() == Some("new title")));

        let _ = tokio::fs::remove_dir_all(tmp).await;
    }

    #[tokio::test]
    async fn import_handles_metadata_unchanged() {
        let (archivist, tmp) = mk().await;

        let a = Uuid::now_v7();
        let discovered = vec![DiscoveredSession {
            native_session_id: "s1".into(),
            title: Some("t".into()),
            created_at: None,
            updated_at: None,
            message_count: 1,
            metadata: serde_json::json!({"model": "claude"}),
            project_path: None,
            file_size: None,
        }];
        let convert = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![record(Uuid::nil(), a, "user", "hi")])
        };
        let _ = import_sessions(&archivist, connector(), discovered.clone(), convert, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();

        let convert2 = |_: &str| -> Result<Vec<MessageRecord>> {
            Ok(vec![record(Uuid::nil(), a, "user", "hi")])
        };
        let stats = import_sessions(&archivist, connector(), discovered, convert2, None, &ImportProgressSink::noop(), true, &HashMap::new())
            .await
            .unwrap();
        assert_eq!(stats.sessions_skipped, 1);
        assert_eq!(stats.sessions_updated, 0);
        assert_eq!(stats.messages_written, 0);

        let _ = tokio::fs::remove_dir_all(tmp).await;
    }
}
