//! `ConnectorRegistryBackend` impl for `JsonlBackend`.

use async_trait::async_trait;
use uuid::Uuid;

use crate::backend::ConnectorRegistryBackend;
use crate::backends::jsonl::backend::JsonlBackend;
use crate::error::{ArchivistError, Result};
use crate::storage::{
    read_connector_index, read_json, write_connector_index, write_json,
};
use crate::types::{ConnectorIndexRow, ConnectorRecord};

#[async_trait]
impl ConnectorRegistryBackend for JsonlBackend {
    async fn put_connector(&self, record: ConnectorRecord) -> Result<()> {
        // Write connector.json
        let connector_dir = self.paths.connector_dir(record.connector_uid);
        tokio::fs::create_dir_all(&connector_dir).await?;
        write_json(&connector_dir.join("connector.json"), &record).await?;

        // Append row to index.tsv (read-modify-write).
        let index_path = self.paths.connector_index_tsv();
        let mut rows = read_connector_index(&index_path).await?;
        rows.push(ConnectorIndexRow {
            connector_uid: record.connector_uid,
            r#type: record.r#type.clone(),
            title: record.title.clone(),
            client_native_id: record.client_native_id.clone(),
            alias_of: record.alias_of,
            created_at: record.created_at,
            fingerprint: record.fingerprint.clone(),
        });
        write_connector_index(&index_path, &rows).await?;

        // Update cache
        self.connector_cache
            .write()
            .await
            .insert(record.connector_uid, record);

        Ok(())
    }

    async fn get_connector(&self, connector_uid: Uuid) -> Result<Option<ConnectorRecord>> {
        // Fast path: consult the in-memory cache.
        {
            let cache = self.connector_cache.read().await;
            if let Some(record) = cache.get(&connector_uid) {
                return Ok(Some(record.clone()));
            }
        }

        // Disk fallback.
        let connector_json = self
            .paths
            .connector_dir(connector_uid)
            .join("connector.json");
        match read_json::<ConnectorRecord>(&connector_json).await {
            Ok(record) => Ok(Some(record)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    async fn list_connectors(&self) -> Result<Vec<ConnectorRecord>> {
        let index_path = self.paths.connector_index_tsv();
        let rows = read_connector_index(&index_path).await?;
        let mut connectors = Vec::new();
        for row in rows {
            if row.alias_of.is_some() {
                continue;
            }
            let connector_json = self
                .paths
                .connector_dir(row.connector_uid)
                .join("connector.json");
            match read_json::<ConnectorRecord>(&connector_json).await {
                Ok(record) => connectors.push(record),
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
                Err(e) => return Err(e.into()),
            }
        }
        Ok(connectors)
    }

    async fn resolve_connector_uid(
        &self,
        client_native_id: &str,
    ) -> Result<Option<Uuid>> {
        // First, try parsing client_native_id as a UUID directly
        // This handles the common case where the connector_id IS the UUID
        if let Ok(uuid) = Uuid::parse_str(client_native_id) {
            // Check if this UUID is a registered connector_uid in cache
            let cache = self.connector_cache.read().await;
            if cache.contains_key(&uuid) {
                return Ok(Some(uuid));
            }
            drop(cache);

            // Check on disk if not in cache
            let connector_json = self.paths.connector_dir(uuid).join("connector.json");
            if connector_json.exists() {
                return Ok(Some(uuid));
            }
        }

        // Not a UUID or not registered as a connector_uid - search by client_native_id
        // Load connector index and find by client_native_id
        let index_path = self.paths.connector_index_tsv();
        let connectors = read_connector_index(&index_path).await?;

        if let Some(connector) = connectors
            .iter()
            .find(|c| c.client_native_id == client_native_id)
        {
            return Ok(Some(connector.connector_uid));
        }

        // Not found - return Ok(None). Error wrapping is a coordinator concern.
        tracing::warn!(
            "Failed to resolve connector_uid for client_native_id '{}'. \
             This connector may not be registered with the archivist.",
            client_native_id
        );

        Ok(None)
    }

    async fn update_connector_fingerprint(
        &self,
        connector_uid: Uuid,
        fingerprint: String,
    ) -> Result<()> {
        // 1. Read and update connector.json
        let connector_dir = self.paths.connector_dir(connector_uid);
        let connector_json = connector_dir.join("connector.json");
        let mut record: ConnectorRecord = read_json(&connector_json)
            .await
            .map_err(|_| ArchivistError::ConnectorUnknown(connector_uid))?;

        record.fingerprint = Some(fingerprint.clone());
        write_json(&connector_json, &record).await?;

        // 2. Update in-memory cache
        self.connector_cache
            .write()
            .await
            .insert(connector_uid, record);

        // 3. Update index.tsv
        let index_path = self.paths.connector_index_tsv();
        let mut rows = read_connector_index(&index_path).await?;
        if let Some(row) = rows.iter_mut().find(|r| r.connector_uid == connector_uid) {
            row.fingerprint = Some(fingerprint);
        }
        write_connector_index(&index_path, &rows).await?;

        Ok(())
    }
}
