//! `SessionMappingBackend` impl for `JsonlBackend`.

use async_trait::async_trait;
use chrono::Utc;
use uuid::Uuid;

use crate::backend::SessionMappingBackend;
use crate::backends::jsonl::backend::JsonlBackend;
use crate::error::Result;
use crate::storage::{append_ndjson, read_connector_index, read_ndjson, write_ndjson};
use crate::types::SessionMapping;

#[async_trait]
impl SessionMappingBackend for JsonlBackend {
    async fn put_mapping(
        &self,
        connector_uid: Uuid,
        native_session_id: &str,
        scroll_id: Uuid,
    ) -> Result<()> {
        // Ported from the mapping-persistence tail of
        // `FileBasedArchivist::register_session`: ensure the connector
        // directory exists, append a `SessionMapping` row to
        // `.db/connectors/{uid}/sessions.jsonl`, and prime `session_cache`.
        //
        // No alias detection — the caller has already chosen `scroll_id`.
        let now = Utc::now();

        // Ensure connector directory exists before appending.
        self.paths.ensure_connector_dir(connector_uid).await?;

        let session_mapping = SessionMapping {
            version: 1,
            connector_uid,
            native_session_id: native_session_id.to_string(),
            scroll_id,
            created_at: now,
            alias_of: None,
        };

        let sessions_write_path = self.paths.sessions_path_for_write(connector_uid);
        append_ndjson(&sessions_write_path, &session_mapping).await?;

        // Prime the in-memory cache for fast resolution.
        self.session_cache
            .write()
            .await
            .insert((connector_uid, native_session_id.to_string()), scroll_id);

        Ok(())
    }

    async fn get_mapping(
        &self,
        connector_uid: Uuid,
        native_session_id: &str,
    ) -> Result<Option<Uuid>> {
        // Ported from `FileBasedArchivist::resolve_session`. Cache-first
        // lookup; on miss, scan the connector's sessions file and populate
        // the cache on hit. Unlike the archivist trait, a miss returns
        // `Ok(None)` instead of `Err(SessionUnknown)`.

        // Check cache first
        let cache_key = (connector_uid, native_session_id.to_string());
        {
            let cache = self.session_cache.read().await;
            if let Some(&scroll_id) = cache.get(&cache_key) {
                return Ok(Some(scroll_id));
            }
        }

        // Cache miss - load from disk
        let sessions_path = self.paths.sessions_path_for_read(connector_uid);
        let mappings: Vec<SessionMapping> = read_ndjson(&sessions_path).await?;

        // Find mapping by native_session_id
        if let Some(mapping) = mappings
            .iter()
            .find(|m| m.native_session_id == native_session_id)
        {
            // Update cache
            self.session_cache
                .write()
                .await
                .insert(cache_key, mapping.scroll_id);
            Ok(Some(mapping.scroll_id))
        } else {
            Ok(None)
        }
    }

    async fn list_mappings_for_connector(
        &self,
        connector_uid: Uuid,
    ) -> Result<Vec<SessionMapping>> {
        // Read `.db/connectors/{uid}/sessions.jsonl` (with `.ndjson`
        // fallback handled by `sessions_path_for_read` + `read_ndjson`).
        let sessions_path = self.paths.sessions_path_for_read(connector_uid);
        let mappings: Vec<SessionMapping> =
            read_ndjson(&sessions_path).await.unwrap_or_default();
        Ok(mappings)
    }

    async fn find_owner(
        &self,
        native_session_id: &str,
    ) -> Result<Option<(Uuid, Uuid)>> {
        // Ported verbatim from `FileBasedArchivist::find_session_owner`.

        // Fast path: scan in-memory session_cache
        {
            let cache = self.session_cache.read().await;
            for ((connector_uid, cached_native_id), scroll_id) in cache.iter() {
                if cached_native_id == native_session_id {
                    return Ok(Some((*connector_uid, *scroll_id)));
                }
            }
        }

        // Slow path: read connector index and scan each connector's sessions file
        let index_path = self.paths.connector_index_tsv();
        let rows = read_connector_index(&index_path).await?;

        for row in &rows {
            // Skip alias connectors - only search primary connectors
            if row.alias_of.is_some() {
                continue;
            }

            let sessions_path = self.paths.sessions_path_for_read(row.connector_uid);
            let mappings: Vec<SessionMapping> = read_ndjson(&sessions_path).await?;

            if let Some(mapping) = mappings
                .iter()
                .find(|m| m.native_session_id == native_session_id)
            {
                // Cache the found mapping for future lookups
                let cache_key = (row.connector_uid, native_session_id.to_string());
                self.session_cache
                    .write()
                    .await
                    .insert(cache_key, mapping.scroll_id);

                return Ok(Some((row.connector_uid, mapping.scroll_id)));
            }
        }

        Ok(None)
    }

    async fn rewrite_connector_mappings(
        &self,
        connector_uid: Uuid,
        mappings: Vec<SessionMapping>,
    ) -> Result<()> {
        // Ensure the connector directory exists before we write.
        self.paths.ensure_connector_dir(connector_uid).await?;

        // Invalidate cache entries for this connector first, then re-prime
        // from the new mapping set. Any (connector_uid, native_id) entry not
        // present in `mappings` is dropped.
        {
            let mut cache = self.session_cache.write().await;
            cache.retain(|(cu, _), _| *cu != connector_uid);
            for m in &mappings {
                cache.insert(
                    (connector_uid, m.native_session_id.clone()),
                    m.scroll_id,
                );
            }
        }

        // Truncate + re-write the canonical `.jsonl` table.
        let write_path = self.paths.sessions_path_for_write(connector_uid);
        write_ndjson(&write_path, &mappings).await?;

        Ok(())
    }
}
