//! `JsonlBackend` — the Phase 2 concrete backend.

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;

use async_trait::async_trait;
use chrono::Utc;
use tokio::sync::RwLock;
use uuid::Uuid;

use crate::backend::{
    ArchiveBackend, ArchiveCapability, CapabilitySet, ConnectorRegistryBackend,
    DagBackend, HealthStatus, MetaEventsBackend, SessionMappingBackend,
};
use crate::error::{ArchivistError, Result};
use crate::storage::{
    append_ndjson, read_connector_index, read_json, read_ndjson, write_json, ArchivePaths,
};
use crate::types::{
    ConnectorRecord, MessageCursor, MessagePage, MessageRecord, SessionCompleteness,
    SessionKind, SessionListQuery, SessionMapping, SessionMetadata, SessionPage,
};

/// NDJSON/JSON/TSV file-based `ArchiveBackend`.
pub struct JsonlBackend {
    pub(crate) paths: ArchivePaths,
    pub(crate) connector_cache: RwLock<HashMap<Uuid, ConnectorRecord>>,
    pub(crate) session_cache: RwLock<HashMap<(Uuid, String), Uuid>>,
    pub(crate) capabilities: CapabilitySet,
}

impl JsonlBackend {
    /// Create a new backend rooted at `archive_root`.
    ///
    /// Creates the required directories (`.contexts`, `.db/connectors`, `.files`)
    /// and initializes empty caches. Matches `FileBasedArchivist::new`.
    pub async fn new(archive_root: PathBuf) -> Result<Self> {
        let paths = ArchivePaths::new(archive_root);

        tokio::fs::create_dir_all(paths.root().join(".contexts")).await?;
        tokio::fs::create_dir_all(paths.root().join(".db").join("connectors")).await?;
        tokio::fs::create_dir_all(paths.root().join(".files")).await?;

        let mut capabilities = HashSet::new();
        capabilities.insert(ArchiveCapability::Dag);
        capabilities.insert(ArchiveCapability::MetaEvents);
        capabilities.insert(ArchiveCapability::ConnectorRegistry);
        capabilities.insert(ArchiveCapability::SessionMapping);

        Ok(Self {
            paths,
            connector_cache: RwLock::new(HashMap::new()),
            session_cache: RwLock::new(HashMap::new()),
            capabilities,
        })
    }

    /// Filesystem path utilities for this backend.
    pub fn paths(&self) -> &ArchivePaths {
        &self.paths
    }

    /// Read and chronologically sort all messages for a session.
    ///
    /// See module docs for the append-order vs. chronological-order rationale.
    pub(crate) async fn read_messages_sorted(
        &self,
        scroll_id: Uuid,
    ) -> Result<Vec<MessageRecord>> {
        let path = self.paths.messages_path_for_read(scroll_id);
        let mut msgs: Vec<MessageRecord> =
            read_ndjson(&path).await.unwrap_or_default();
        msgs.sort_by(|a, b| {
            a.ts.cmp(&b.ts).then(a.message_id.cmp(&b.message_id))
        });
        Ok(msgs)
    }

    /// Locate the (connector_uid, native_session_id) owning `scroll_id` by
    /// scanning the session cache first, then each connector's session
    /// mapping files on disk.
    async fn find_mapping_for_scroll_id(&self, scroll_id: Uuid) -> Option<(Uuid, String)> {
        // Check cache first
        {
            let cache = self.session_cache.read().await;
            for ((connector_uid, native_id), cached_scroll_id) in cache.iter() {
                if *cached_scroll_id == scroll_id {
                    return Some((*connector_uid, native_id.clone()));
                }
            }
        }

        // Cache miss: scan connector index and each connector's sessions file
        let index_path = self.paths.connector_index_tsv();
        let rows = match read_connector_index(&index_path).await {
            Ok(rows) => rows,
            Err(_) => return None,
        };

        for row in &rows {
            let sessions_path = self.paths.sessions_path_for_read(row.connector_uid);
            let mappings: Vec<SessionMapping> = match read_ndjson(&sessions_path).await {
                Ok(m) => m,
                Err(_) => continue,
            };
            for mapping in mappings {
                if mapping.scroll_id == scroll_id {
                    return Some((row.connector_uid, mapping.native_session_id));
                }
            }
        }

        None
    }

    /// Load every session for a connector, including hidden ones. Used by
    /// `list_sessions_paged` — it applies visibility filters itself.
    async fn load_sessions_for_connector(
        &self,
        connector_uid: Uuid,
    ) -> Result<Vec<SessionMetadata>> {
        let sessions_path = self.paths.sessions_path_for_read(connector_uid);
        let mappings: Vec<SessionMapping> = read_ndjson(&sessions_path).await?;

        let mut sessions = Vec::new();
        for mapping in mappings {
            let session_json_path = self.paths.session_json(mapping.scroll_id);
            match read_json::<SessionMetadata>(&session_json_path).await {
                Ok(metadata) => sessions.push(metadata),
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                    tracing::debug!(
                        scroll_id = %mapping.scroll_id,
                        "session.json missing, surfacing as Discovered stub"
                    );
                    sessions.push(SessionMetadata {
                        version: 1,
                        scroll_id: mapping.scroll_id,
                        created_at: mapping.created_at,
                        updated_at: mapping.created_at,
                        title: None,
                        connector_uid,
                        native_session_id: Some(mapping.native_session_id.clone()),
                        agent_id: None,
                        parent_scroll_id: None,
                        continuation: None,
                        tags: Vec::new(),
                        metadata: serde_json::json!({}),
                        no_update: false,
                        kind: SessionKind::Chat,
                        acp_client_id: None,
                        is_connected: None,
                        current_session_id: None,
                        models: None,
                        modes: None,
                        config_options: None,
                        completeness: SessionCompleteness::Discovered,
                        matrix_room_id: None,
                        matrix_sharing_active: false,
                        matrix_shared_at: None,
                        is_subagent: false,
                        subagent_type: None,
                        spawning_tool_use_id: None,
                    });
                }
                Err(e) => return Err(e.into()),
            }
        }
        Ok(sessions)
    }
}

/// Returns true if `session` satisfies every filter in `query`.
///
/// `connector_uid` is already honored by the caller (it picks which connector
/// directories to scan), so we do not re-check it here.
fn matches_query(
    session: &SessionMetadata,
    query: &crate::types::SessionListQuery,
) -> bool {
    // Visibility
    if !query.include_hidden && (session.no_update || session.is_subagent) {
        return false;
    }

    // Project scope — project_ids lives in metadata.project_id
    if !query.project_ids.is_empty() {
        let session_project_id = session
            .metadata
            .get("project_id")
            .and_then(|v| v.as_str());
        match session_project_id {
            Some(pid) => {
                if !query.project_ids.iter().any(|q| q.as_str() == pid) {
                    return false;
                }
            }
            None => return false,
        }
    }

    // Project path filter — exact match on metadata.project_path
    if let Some(ref path) = query.project_path {
        let session_path = session
            .metadata
            .get("project_path")
            .and_then(|v| v.as_str());
        if session_path != Some(path.as_str()) {
            return false;
        }
    }

    // Title filter — case-insensitive substring.
    if let Some(q) = query.title_query.as_ref() {
        let needle = q.to_lowercase();
        let haystack = match session.title.as_ref() {
            Some(t) => t.to_lowercase(),
            None => return false,
        };
        if !haystack.contains(&needle) {
            return false;
        }
    }

    // Tag filter — all requested tags must be present on the session.
    if !query.tags.is_empty() {
        for required in &query.tags {
            if !session.tags.iter().any(|t| t == required) {
                return false;
            }
        }
    }

    // Model filter — case-insensitive substring on metadata.model.
    if let Some(q) = query.model_filter.as_ref() {
        let needle = q.to_lowercase();
        let haystack = session
            .metadata
            .get("model")
            .and_then(|v| v.as_str())
            .map(|s| s.to_lowercase());
        match haystack {
            Some(h) if h.contains(&needle) => {}
            _ => return false,
        }
    }

    true
}

#[async_trait]
impl ArchiveBackend for JsonlBackend {
    fn capabilities(&self) -> &CapabilitySet {
        &self.capabilities
    }

    async fn health_check(&self) -> HealthStatus {
        match tokio::fs::metadata(self.paths.root()).await {
            Ok(m) if m.is_dir() => HealthStatus::Healthy,
            Ok(_) => HealthStatus::Unavailable {
                reason: "archive root is not a directory".into(),
            },
            Err(e) => HealthStatus::Unavailable {
                reason: format!("stat archive root failed: {e}"),
            },
        }
    }

    async fn put_session(&self, meta: SessionMetadata) -> Result<()> {
        tokio::fs::create_dir_all(&self.paths.session_dir(meta.scroll_id)).await?;
        write_json(&self.paths.session_json(meta.scroll_id), &meta).await?;
        Ok(())
    }

    async fn get_session(&self, scroll_id: Uuid) -> Result<Option<SessionMetadata>> {
        // FileBasedArchivist ignores archive parameter (single-archive only)
        let session_json_path = self.paths.session_json(scroll_id);

        match read_json(&session_json_path).await {
            Ok(metadata) => Ok(Some(metadata)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    async fn list_sessions_paged(&self, query: SessionListQuery) -> Result<SessionPage> {
        use crate::types::{SessionCursor, SessionPage, MAX_PAGE_LIMIT};

        // Determine which connectors to scan.
        let connector_uids: Vec<Uuid> = if !query.connector_uids.is_empty() {
            query.connector_uids.clone()
        } else {
            // Iterate every primary (non-alias) connector.
            let index_path = self.paths.connector_index_tsv();
            let connectors = read_connector_index(&index_path).await?;
            connectors
                .into_iter()
                .filter(|c| c.alias_of.is_none())
                .map(|c| c.connector_uid)
                .collect()
        };

        // Stream matching sessions from every selected connector.
        let mut matched: Vec<SessionMetadata> = Vec::new();

        for connector_uid in connector_uids {
            let sessions = match self.load_sessions_for_connector(connector_uid).await {
                Ok(s) => s,
                Err(e) => {
                    tracing::warn!(
                        connector_uid = %connector_uid,
                        error = %e,
                        "Failed to list sessions for connector during paged scan, skipping"
                    );
                    continue;
                }
            };

            for session in sessions {
                if !matches_query(&session, &query) {
                    continue;
                }
                matched.push(session);
            }
        }

        // Sort by (updated_at DESC, scroll_id DESC).
        matched.sort_by(|a, b| {
            b.updated_at
                .cmp(&a.updated_at)
                .then_with(|| b.scroll_id.cmp(&a.scroll_id))
        });

        // Skip entries at-or-before the cursor.
        if let Some(cursor) = query.cursor.as_ref() {
            matched.retain(|s| {
                (s.updated_at, s.scroll_id) < (cursor.updated_at, cursor.scroll_id)
            });
        }

        // Capture total count before slicing.
        let total_count = matched.len();

        // Clamp limit and paginate.
        let effective_limit = query.limit.min(MAX_PAGE_LIMIT).max(1);
        let has_more = matched.len() > effective_limit;
        matched.truncate(effective_limit);

        let next_cursor = if has_more {
            matched.last().map(|s| SessionCursor {
                updated_at: s.updated_at,
                scroll_id: s.scroll_id,
            })
        } else {
            None
        };

        Ok(SessionPage {
            items: matched,
            next_cursor,
            total_count: Some(total_count),
        })
    }

    async fn delete_session(&self, scroll_id: Uuid) -> Result<()> {
        // FileBasedArchivist ignores archive parameter (single-archive only)

        // First, read session metadata to get connector_uid and native_session_id
        let session_dir = self.paths.session_dir(scroll_id);
        let session_json_path = self.paths.session_json(scroll_id);

        if !session_dir.exists() {
            return Err(ArchivistError::SessionUnknown(scroll_id));
        }

        // Read session metadata to get connector info
        let metadata: SessionMetadata = read_json(&session_json_path).await?;
        let connector_uid = metadata.connector_uid;
        let native_session_id = metadata.native_session_id.clone();

        // Delete the session directory and all its contents
        tokio::fs::remove_dir_all(&session_dir).await.map_err(|e| {
            tracing::error!("Failed to delete session directory {:?}: {}", session_dir, e);
            ArchivistError::Io(e)
        })?;

        tracing::info!(
            "Deleted session directory for scroll_id: {}",
            scroll_id
        );

        // Remove from session cache
        if let Some(native_id) = &native_session_id {
            let mut cache = self.session_cache.write().await;
            cache.remove(&(connector_uid, native_id.clone()));
        }

        // Note: We're not removing from sessions.ndjson because it's append-only.
        // The session simply won't have a directory anymore, so list_sessions will skip it.
        // A future enhancement could add a "deleted" flag or periodic compaction.

        tracing::info!(
            "Successfully deleted session {} (connector: {})",
            scroll_id,
            connector_uid
        );

        Ok(())
    }

    async fn append_messages(
        &self,
        scroll_id: Uuid,
        messages: Vec<MessageRecord>,
    ) -> Result<()> {
        // Ensure session directory exists (handles resync case where directory was deleted)
        self.paths.ensure_dirs(scroll_id).await?;

        // Append each message to messages.jsonl
        let messages_path = self.paths.messages_path_for_write(scroll_id);
        for message in &messages {
            append_ndjson(&messages_path, message).await?;
        }

        // Update session.json timestamp (or create if missing)
        let session_json_path = self.paths.session_json(scroll_id);
        let now = Utc::now();

        let session_metadata = match read_json::<SessionMetadata>(&session_json_path).await {
            Ok(mut metadata) => {
                metadata.updated_at = now;
                metadata
            }
            Err(_) => {
                // session.json doesn't exist, create minimal metadata
                // This handles resync case where directory was deleted but mapping still exists
                tracing::info!(
                    scroll_id = %scroll_id,
                    "Creating minimal session.json during append (was missing)"
                );

                // Look up the correct connector_uid and native_session_id via session mappings
                let (connector_uid, native_session_id) = match self.find_mapping_for_scroll_id(scroll_id).await {
                    Some(mapping) => mapping,
                    None => {
                        tracing::error!(
                            scroll_id = %scroll_id,
                            "Cannot reconstruct session.json: no connector mapping found. \
                             Messages written but session metadata will remain missing."
                        );
                        return Ok(());
                    }
                };

                SessionMetadata {
                    version: 1,
                    scroll_id,
                    created_at: now,
                    updated_at: now,
                    title: None,
                    connector_uid,
                    native_session_id: Some(native_session_id),
                    agent_id: None,
                    parent_scroll_id: None,
                    continuation: None,
                    tags: Vec::new(),
                    metadata: serde_json::json!({}),
                    no_update: false,
                    kind: SessionKind::Chat,
                    acp_client_id: None,
                    is_connected: None,
                    current_session_id: None,
                    models: None,
                    modes: None,
                    config_options: None,
                    completeness: SessionCompleteness::default(),
                    matrix_room_id: None,
                    matrix_sharing_active: false,
                    matrix_shared_at: None,
                    is_subagent: false,
                    subagent_type: None,
                    spawning_tool_use_id: None,
                }
            }
        };

        write_json(&session_json_path, &session_metadata).await?;

        Ok(())
    }

    async fn get_messages_paged(
        &self,
        scroll_id: Uuid,
        cursor: Option<MessageCursor>,
        limit: usize,
    ) -> Result<MessagePage> {
        use crate::types::MAX_PAGE_LIMIT;

        // Hard-clamp limit — same policy as sessions.
        let effective_limit = limit.min(MAX_PAGE_LIMIT).max(1);

        // Read NDJSON, sort, apply cursor.
        let mut all = self.read_messages_sorted(scroll_id).await?;

        if let Some(c) = cursor.as_ref() {
            // Keep strictly-after the cursor point in (ts, message_id) order.
            all.retain(|m| (m.ts, m.message_id) > (c.ts, c.message_id));
        }

        let total = all.len();
        let taken: Vec<_> = all.into_iter().take(effective_limit).collect();

        let next_cursor = if total > taken.len() {
            taken.last().map(|m| MessageCursor {
                ts: m.ts,
                message_id: m.message_id,
            })
        } else {
            None
        };

        Ok(MessagePage {
            items: taken,
            next_cursor,
        })
    }

    async fn count_messages(&self, scroll_id: Uuid) -> Result<usize> {
        let messages_path = self.paths.messages_path_for_read(scroll_id);

        // Read file and count lines (each line = one message)
        // If file doesn't exist, return 0 (empty session)
        match tokio::fs::read_to_string(&messages_path).await {
            Ok(content) => {
                // Count non-empty lines
                let count = content.lines().filter(|line| !line.trim().is_empty()).count();
                Ok(count)
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                // File doesn't exist yet - empty session
                Ok(0)
            }
            Err(e) => Err(e.into()),
        }
    }

    async fn clear_session_messages(&self, scroll_id: Uuid) -> Result<()> {
        // First, verify the session exists by reading its metadata
        let session_json_path = self.paths.session_json(scroll_id);
        let mut session_metadata: SessionMetadata = match read_json(&session_json_path).await {
            Ok(metadata) => metadata,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Err(ArchivistError::SessionUnknown(scroll_id));
            }
            Err(e) => return Err(e.into()),
        };

        // Truncate the messages file (clear all messages)
        // First try to clear .jsonl (new format), then fall back to .ndjson (legacy)
        let jsonl_path = self.paths.messages_path_for_write(scroll_id);
        #[allow(deprecated)]
        let ndjson_path = self.paths.messages_ndjson(scroll_id);

        let mut cleared = false;

        // Clear .jsonl if it exists
        if jsonl_path.exists() {
            tokio::fs::write(&jsonl_path, "").await?;
            cleared = true;
        }

        // Also clear .ndjson if it exists (in case both are present)
        if ndjson_path.exists() {
            tokio::fs::write(&ndjson_path, "").await?;
            cleared = true;
        }

        if cleared {
            tracing::info!(
                scroll_id = %scroll_id,
                "Cleared all messages from session"
            );
        }

        // Update the session's updated_at timestamp
        session_metadata.updated_at = Utc::now();
        write_json(&session_json_path, &session_metadata).await?;

        tracing::info!(
            scroll_id = %scroll_id,
            "Updated session metadata after clearing messages"
        );

        Ok(())
    }

    fn as_dag(&self) -> Option<&dyn DagBackend> {
        Some(self)
    }
    fn as_meta_events(&self) -> Option<&dyn MetaEventsBackend> {
        Some(self)
    }
    fn as_connector_registry(&self) -> Option<&dyn ConnectorRegistryBackend> {
        Some(self)
    }
    fn as_session_mapping(&self) -> Option<&dyn SessionMappingBackend> {
        Some(self)
    }
}

#[cfg(test)]
mod contract_tests {
    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn jsonl_backend_honors_all_contracts() {
        let dir = tempdir().expect("tempdir");
        let backend = JsonlBackend::new(dir.path().to_path_buf())
            .await
            .expect("new");
        crate::backend::contract::verify_all_contracts(&backend).await;
    }
}
