//! Backfill functionality for importing existing sessions from connectors.
//!
//! This module provides utilities to import sessions and messages from connectors
//! that support listing operations (like OpenCode connectors) into the Archivist.

use futures::future::BoxFuture;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::coordinator::Archivist;
use crate::error::{ArchivistError, Result};
use crate::types::{MessageRecord, RegisterSessionRequest, RegisterStatus};
use dirigent_protocol::{Message, Session};

/// Statistics collected during a backfill operation.
///
/// This provides a summary of what was imported and any errors encountered.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct BackfillStats {
    /// Total number of sessions found in the connector
    pub sessions_found: usize,
    /// Number of sessions successfully imported (new registrations)
    pub sessions_imported: usize,
    /// Number of sessions skipped (already archived)
    pub sessions_skipped: usize,
    /// Total number of messages imported across all sessions
    pub messages_imported: usize,
    /// Error messages for sessions that failed to import
    pub errors: Vec<String>,
}

impl BackfillStats {
    /// Create a new BackfillStats with all counts at zero
    pub fn new() -> Self {
        Self {
            sessions_found: 0,
            sessions_imported: 0,
            sessions_skipped: 0,
            messages_imported: 0,
            errors: Vec::new(),
        }
    }
}

impl Default for BackfillStats {
    fn default() -> Self {
        Self::new()
    }
}

/// Backfill sessions from a connector into the archive.
///
/// This function imports existing sessions from a connector by:
/// 1. Attempting to register each session with the archivist
/// 2. For newly registered sessions, fetching messages via the provided closure
/// 3. Appending fetched messages to the archive
/// 4. Collecting statistics on successes, failures, and skips
///
/// # Arguments
///
/// * `archivist` - The archivist to backfill into
/// * `connector_uid` - The UID of the connector being backfilled
/// * `sessions` - List of sessions to import (from connector's list_sessions())
/// * `fetch_messages` - Async closure to fetch messages for a given native session ID
///
/// # Returns
///
/// Statistics about the backfill operation including counts and errors
///
/// # Error Handling
///
/// This function continues processing all sessions even if individual sessions fail.
/// Errors are collected in `BackfillStats.errors` rather than aborting the operation.
///
/// # Example
///
/// ```no_run
/// use dirigent_archivist::{Archivist, backfill_from_sessions};
/// use dirigent_protocol::{Session, Message};
/// use uuid::Uuid;
///
/// # async fn example(archivist: &Archivist, sessions: Vec<Session>) {
/// let connector_uid = Uuid::now_v7();
///
/// let stats = backfill_from_sessions(
///     archivist,
///     connector_uid,
///     sessions,
///     |session_id| {
///         Box::pin(async move {
///             // Fetch messages from connector
///             // Return Vec<Message>
///             Ok(vec![])
///         })
///     }
/// ).await.unwrap();
///
/// println!("Imported {} sessions, {} messages",
///     stats.sessions_imported,
///     stats.messages_imported);
/// # }
/// ```
pub async fn backfill_from_sessions<F>(
    archivist: &Archivist,
    connector_uid: Uuid,
    sessions: Vec<Session>,
    fetch_messages: F,
) -> Result<BackfillStats>
where
    F: Fn(&str) -> BoxFuture<'static, Result<Vec<Message>>> + Send + Sync,
{
    let mut stats = BackfillStats::new();
    stats.sessions_found = sessions.len();

    for session in sessions {
        let native_session_id = session.id.clone();

        // Try to resolve the session - if it exists, skip it
        match archivist
            .resolve_session(connector_uid, &native_session_id, None)
            .await
        {
            Ok(_scroll_id) => {
                // Session already archived, skip
                stats.sessions_skipped += 1;
                continue;
            }
            Err(ArchivistError::SessionUnknown(_)) => {
                // Session not found, proceed with import
            }
            Err(e) => {
                // Unexpected error during resolution
                stats.errors.push(format!(
                    "Failed to resolve session {}: {}",
                    native_session_id, e
                ));
                continue;
            }
        }

        // Register the session
        let register_req = RegisterSessionRequest {
            connector_uid,
            native_session_id: native_session_id.clone(),
            title: Some(session.title.clone()),
            custom_scroll_id: None, // Let archivist generate
            metadata: serde_json::to_value(&session.metadata)
                .unwrap_or_else(|_| serde_json::json!({})),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let scroll_id = match archivist.register_session(register_req, None).await {
            Ok(response) => {
                match response.status {
                    RegisterStatus::Accepted => {
                        stats.sessions_imported += 1;
                        response.scroll_id
                    }
                    RegisterStatus::Aliased => {
                        // Already exists (shouldn't happen since we checked, but handle gracefully)
                        stats.sessions_skipped += 1;
                        continue;
                    }
                    RegisterStatus::Rejected => {
                        // Registration rejected (collision inconsistency)
                        stats.errors.push(format!(
                            "Session registration rejected for {}: UID collision",
                            native_session_id
                        ));
                        continue;
                    }
                }
            }
            Err(e) => {
                stats.errors.push(format!(
                    "Failed to register session {}: {}",
                    native_session_id, e
                ));
                continue;
            }
        };

        // Fetch messages for this session
        let messages = match fetch_messages(&native_session_id).await {
            Ok(msgs) => msgs,
            Err(e) => {
                stats.errors.push(format!(
                    "Failed to fetch messages for session {}: {}",
                    native_session_id, e
                ));
                continue;
            }
        };

        // Convert protocol messages to message records
        let message_records: Vec<MessageRecord> = messages
            .into_iter()
            .map(|msg| convert_message_to_record(msg, scroll_id))
            .collect();

        let message_count = message_records.len();

        // Append messages to the archive
        if let Err(e) = archivist
            .append_messages(scroll_id, message_records, None)
            .await
        {
            stats.errors.push(format!(
                "Failed to append messages for session {}: {}",
                native_session_id, e
            ));
            continue;
        }

        stats.messages_imported += message_count;
    }

    Ok(stats)
}

/// Convert a dirigent_protocol::Message to a MessageRecord for archival.
///
/// This function translates the protocol message format into the archivist's
/// internal storage format, extracting markdown content and metadata.
pub fn convert_message_to_record(msg: Message, scroll_id: Uuid) -> MessageRecord {
    // Extract text content from message parts and convert to markdown
    let mut md_parts = Vec::new();
    for part in &msg.content {
        match part {
            dirigent_protocol::MessagePart::Text { text } => {
                md_parts.push(text.clone());
            }
            dirigent_protocol::MessagePart::Thinking { text } => {
                md_parts.push(format!("<thinking>\n{}\n</thinking>", text));
            }
            dirigent_protocol::MessagePart::Code { language, code } => {
                md_parts.push(format!("```{}\n{}\n```", language, code));
            }
            dirigent_protocol::MessagePart::Tool {
                tool,
                tool_call_id: _,
                input,
                output,
            } => {
                let mut tool_text =
                    format!("**Tool: {}**\n\nInput:\n```json\n{}\n```", tool, input);
                if let Some(out) = output {
                    tool_text.push_str(&format!("\n\nOutput:\n```json\n{}\n```", out));
                }
                md_parts.push(tool_text);
            }
            dirigent_protocol::MessagePart::File { path, content } => {
                md_parts.push(format!("**File: {}**\n\n```\n{}\n```", path, content));
            }
        }
    }
    let content_md = md_parts.join("\n\n");

    // Serialize original content parts for rich UI rendering
    let content_parts = serde_json::to_value(&msg.content).ok();

    // Convert role
    let role = match msg.role {
        dirigent_protocol::MessageRole::User => "user",
        dirigent_protocol::MessageRole::Assistant => "assistant",
    }
    .to_string();

    // Generate message ID from the protocol message ID or create new one
    let message_id = Uuid::now_v7();

    MessageRecord {
        version: 1,
        message_id,
        session: scroll_id,
        parent_id: None,
        ts: msg.created_at,
        role,
        author: None, // Protocol messages don't have author field
        content_md,
        content_parts,
        attachments: Vec::new(), // Would need to extract from message parts if supported
        metadata: msg
            .metadata
            .and_then(|m| serde_json::to_value(m).ok())
            .unwrap_or_else(|| serde_json::json!({})),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::coordinator::Archivist;
    use chrono::Utc;
    use dirigent_protocol::{MessageRole, MessageStatus, SessionMetadata};
    use tempfile::TempDir;

    async fn setup_test_archivist() -> (Archivist, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        // Use `from_single_backend` so each test is isolated (no shared
        // registry file in the tempdir's parent racing against siblings).
        let backend = std::sync::Arc::new(
            crate::backends::JsonlBackend::new(temp_dir.path().to_path_buf())
                .await
                .unwrap(),
        );
        let archivist = Archivist::from_single_backend("main".into(), backend)
            .await
            .unwrap();
        (archivist, temp_dir)
    }

    fn create_test_session(id: &str, title: &str) -> Session {
        Session {
            id: id.to_string(),
            title: title.to_string(),
            created_at: Utc::now(),
            updated_at: Utc::now(),
            metadata: SessionMetadata {
                project_path: "/test".to_string(),
                model: Some("test-model".to_string()),
                total_messages: 0,
                system_message: None,
                current_mode_id: None,
                _meta: None,
                project_id: None,
            },
            cwd: None,
            config_options: None,
            acp_client_id: None,
            models: None,
            modes: None,
        }
    }

    fn create_test_message(id: &str, session_id: &str, role: MessageRole, text: &str) -> Message {
        Message {
            id: id.to_string(),
            session_id: session_id.to_string(),
            role,
            created_at: Utc::now(),
            content: vec![dirigent_protocol::MessagePart::Text {
                text: text.to_string(),
            }],
            status: MessageStatus::Completed,
            metadata: None,
        }
    }

    #[tokio::test]
    async fn test_backfill_new_sessions() {
        let (archivist, _temp) = setup_test_archivist().await;

        // Register connector first
        let connector_uid = Uuid::now_v7();
        let connector_req = crate::types::RegisterConnectorRequest {
            custom_uid: Some(connector_uid),
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "test-connector".to_string(),
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        archivist
            .register_connector(connector_req, None)
            .await
            .unwrap();

        // Create test sessions
        let sessions = vec![
            create_test_session("session-1", "Session 1"),
            create_test_session("session-2", "Session 2"),
        ];

        // Mock message fetcher
        let fetch_messages = |session_id: &str| {
            let sid = session_id.to_string();
            Box::pin(async move {
                Ok(vec![
                    create_test_message("msg-1", &sid, MessageRole::User, "Hello"),
                    create_test_message("msg-2", &sid, MessageRole::Assistant, "Hi there"),
                ])
            }) as BoxFuture<'static, Result<Vec<Message>>>
        };

        // Backfill
        let stats = backfill_from_sessions(&archivist, connector_uid, sessions, fetch_messages)
            .await
            .unwrap();

        // Verify stats
        assert_eq!(stats.sessions_found, 2);
        assert_eq!(stats.sessions_imported, 2);
        assert_eq!(stats.sessions_skipped, 0);
        assert_eq!(stats.messages_imported, 4); // 2 messages per session
        assert_eq!(stats.errors.len(), 0);
    }

    #[tokio::test]
    async fn test_backfill_skips_existing_sessions() {
        let (archivist, _temp) = setup_test_archivist().await;

        // Register connector first
        let connector_uid = Uuid::now_v7();
        let connector_req = crate::types::RegisterConnectorRequest {
            custom_uid: Some(connector_uid),
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "test-connector".to_string(),
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        archivist
            .register_connector(connector_req, None)
            .await
            .unwrap();

        // Pre-register one session
        let session1 = create_test_session("session-1", "Session 1");
        let req = RegisterSessionRequest {
            connector_uid,
            native_session_id: session1.id.clone(),
            title: Some(session1.title.clone()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };
        archivist.register_session(req, None).await.unwrap();

        // Create sessions including the pre-registered one
        let sessions = vec![session1, create_test_session("session-2", "Session 2")];

        // Mock message fetcher
        let fetch_messages = |session_id: &str| {
            let sid = session_id.to_string();
            Box::pin(async move {
                Ok(vec![create_test_message(
                    "msg-1",
                    &sid,
                    MessageRole::User,
                    "Test",
                )])
            }) as BoxFuture<'static, Result<Vec<Message>>>
        };

        // Backfill
        let stats = backfill_from_sessions(&archivist, connector_uid, sessions, fetch_messages)
            .await
            .unwrap();

        // Verify stats - session-1 should be skipped
        assert_eq!(stats.sessions_found, 2);
        assert_eq!(stats.sessions_imported, 1); // Only session-2
        assert_eq!(stats.sessions_skipped, 1); // session-1 already exists
        assert_eq!(stats.messages_imported, 1); // Only messages from session-2
        assert_eq!(stats.errors.len(), 0);
    }

    #[tokio::test]
    async fn test_backfill_handles_fetch_errors() {
        let (archivist, _temp) = setup_test_archivist().await;

        // Register connector first
        let connector_uid = Uuid::now_v7();
        let connector_req = crate::types::RegisterConnectorRequest {
            custom_uid: Some(connector_uid),
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "test-connector".to_string(),
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        archivist
            .register_connector(connector_req, None)
            .await
            .unwrap();

        let sessions = vec![create_test_session("session-1", "Session 1")];

        // Mock message fetcher that fails
        let fetch_messages = |_session_id: &str| {
            Box::pin(async move {
                Err(ArchivistError::InvalidRequest(
                    "Failed to fetch messages".to_string(),
                ))
            }) as BoxFuture<'static, Result<Vec<Message>>>
        };

        // Backfill
        let stats = backfill_from_sessions(&archivist, connector_uid, sessions, fetch_messages)
            .await
            .unwrap();

        // Verify stats - session registered but messages failed
        assert_eq!(stats.sessions_found, 1);
        assert_eq!(stats.sessions_imported, 1); // Session was registered
        assert_eq!(stats.messages_imported, 0); // But no messages imported
        assert_eq!(stats.errors.len(), 1); // Error recorded
        assert!(stats.errors[0].contains("Failed to fetch messages"));
    }

    #[test]
    fn test_backfill_stats_default() {
        let stats = BackfillStats::default();
        assert_eq!(stats.sessions_found, 0);
        assert_eq!(stats.sessions_imported, 0);
        assert_eq!(stats.sessions_skipped, 0);
        assert_eq!(stats.messages_imported, 0);
        assert_eq!(stats.errors.len(), 0);
    }

    #[test]
    fn test_convert_message_to_record() {
        let scroll_id = Uuid::now_v7();
        let msg = create_test_message("msg-1", "session-1", MessageRole::User, "Hello world");

        let record = convert_message_to_record(msg, scroll_id);

        assert_eq!(record.session, scroll_id);
        assert_eq!(record.role, "user");
        assert_eq!(record.content_md, "Hello world");
        assert_eq!(record.version, 1);
    }

    #[test]
    fn test_convert_message_with_thinking() {
        let scroll_id = Uuid::now_v7();
        let msg = Message {
            id: "msg-1".to_string(),
            session_id: "session-1".to_string(),
            role: MessageRole::Assistant,
            created_at: Utc::now(),
            content: vec![dirigent_protocol::MessagePart::Thinking {
                text: "Let me think...".to_string(),
            }],
            status: MessageStatus::Completed,
            metadata: None,
        };

        let record = convert_message_to_record(msg, scroll_id);

        assert!(record.content_md.contains("<thinking>"));
        assert!(record.content_md.contains("Let me think..."));
        assert!(record.content_md.contains("</thinking>"));
    }
}
