//! Connector orchestration for `Archivist`.
//!
//! Alias detection and tri-state registration logic live here; persistence is
//! delegated to each backend's `ConnectorRegistryBackend` sub-trait. Ported
//! from `FileBasedArchivist::register_connector` and
//! `MultiArchiveArchivist::resolve_connector_uid`.

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

use crate::backend::ArchiveCapability;
use crate::coordinator::Archivist;
use crate::error::{ArchivistError, Result};
use crate::types::{
    ConnectorRecord, RegisterConnectorRequest, RegisterConnectorResponse, RegisterStatus,
};

impl Archivist {
    /// Register a connector with alias detection.
    ///
    /// Ported from `FileBasedArchivist::register_connector`. Decision order:
    ///
    /// 1. If `custom_uid` collides with an existing connector:
    ///    - same `client_native_id` → `Aliased` (idempotent re-registration).
    ///    - different `client_native_id` → `CollisionInconsistent` error.
    /// 2. If the `client_native_id` is already registered under a different
    ///    UID → `Aliased` to that pre-existing UID.
    /// 3. If a `fingerprint` matches a pre-existing connector → `Aliased` to
    ///    that UID. (Identity persistence across connector re-adds.)
    /// 4. Otherwise → `Accepted`; a new `ConnectorRecord` is persisted via
    ///    `ConnectorRegistryBackend::put_connector`.
    // TODO(phase3 task 16): register_connector fanout requires replicating the
    // ConnectorRecord to secondaries. Since connectors are identity-shaped (UIDs
    // must match across backends), the tri-state alias detection must stay
    // canonical on the primary, but the accepted record should be mirrored to
    // secondaries. Deferred to a follow-up within Phase 3 — the core Task 16
    // plan covers append_messages and the session mutators which are the hot
    // paths. Current behaviour: single-primary via `resolve_backend`.
    pub async fn register_connector(
        &self,
        req: RegisterConnectorRequest,
        archive: Option<String>,
    ) -> Result<RegisterConnectorResponse> {
        let backend = self.resolve_backend(archive).await?;
        let registry = backend.as_connector_registry().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::ConnectorRegistry,
                backend: "selected".into(),
            }
        })?;

        // Generate connector UID (use custom_uid or generate new)
        let connector_uid = req.custom_uid.unwrap_or_else(Uuid::now_v7);

        // Load existing (non-alias) connectors for collision detection.
        let existing_connectors = registry.list_connectors().await?;

        // 1. Check for UID collision.
        if let Some(existing) = existing_connectors
            .iter()
            .find(|c| c.connector_uid == connector_uid)
        {
            if existing.client_native_id == req.client_native_id {
                // Same UID with same client_native_id -> ALIASED (idempotent).
                return Ok(RegisterConnectorResponse {
                    status: RegisterStatus::Aliased,
                    connector_uid,
                    alias_of: Some(connector_uid),
                    note: Some("Connector already registered with this UID".to_string()),
                });
            } else {
                // Same UID with different client_native_id -> REJECTED.
                return Err(ArchivistError::CollisionInconsistent(connector_uid));
            }
        }

        // 2. Check for existing client_native_id (different UID collision).
        if let Some(existing) = existing_connectors
            .iter()
            .find(|c| c.client_native_id == req.client_native_id)
        {
            return Ok(RegisterConnectorResponse {
                status: RegisterStatus::Aliased,
                connector_uid: existing.connector_uid,
                alias_of: Some(existing.connector_uid),
                note: Some("Connector already registered with different UID".to_string()),
            });
        }

        // 3. Check for fingerprint match (identity persistence across re-adds).
        //
        // Note: the original `FileBasedArchivist` additionally refreshed the
        // matched connector's `title`/`metadata` on disk and in cache here.
        // That refresh bypassed both the TSV index and any backend abstraction
        // (direct `read_json`/`write_json` against `connector.json`). The
        // `ConnectorRegistryBackend` trait does not yet expose an
        // "update metadata" method, and `put_connector` would append a
        // duplicate row to the index rather than mutate in place. The refresh
        // was best-effort (`let _ = write_json(...)`) and is not exercised by
        // existing tests; deliberately skipped here. Re-introduce via a
        // dedicated backend method if a consumer relies on it.
        if let Some(ref fp) = req.fingerprint {
            if let Some(existing) = existing_connectors
                .iter()
                .find(|c| c.fingerprint.as_deref() == Some(fp.as_str()))
            {
                let matched_uid = existing.connector_uid;
                return Ok(RegisterConnectorResponse {
                    status: RegisterStatus::Aliased,
                    connector_uid: matched_uid,
                    alias_of: Some(matched_uid),
                    note: Some(format!("Matched by fingerprint: {}", fp)),
                });
            }
        }

        // 4. No collision -> ACCEPTED, create and persist new connector.
        let now = Utc::now();
        let connector_record = ConnectorRecord {
            version: 1,
            connector_uid,
            r#type: req.r#type,
            title: req.title,
            client_native_id: req.client_native_id,
            alias_of: None,
            created_at: now,
            metadata: req.metadata,
            fingerprint: req.fingerprint,
        };

        registry.put_connector(connector_record).await?;

        Ok(RegisterConnectorResponse {
            status: RegisterStatus::Accepted,
            connector_uid,
            alias_of: None,
            note: None,
        })
    }

    /// Resolve a connector UID by scanning every registered backend.
    ///
    /// Ported from `MultiArchiveArchivist::resolve_connector_uid`: each
    /// backend is tried in turn; the first backend that recognises the
    /// `client_native_id` wins. As a secondary path, if `client_native_id`
    /// parses as a UUID, checks whether a backend already has a connector
    /// record at that UID. Returns `ConnectorUnknown(Uuid::nil())` if no
    /// backend can resolve it.
    pub async fn resolve_connector_uid(&self, client_native_id: &str) -> Result<Uuid> {
        // Hand-rolled walk rather than `read_walk_collection`: we want
        // "try every backend" semantics — a backend that returns `Ok(None)`
        // should NOT win the walk. `read_walk_collection` treats any `Ok(_)`
        // as a hit, so it would stop at the first backend that answered at
        // all. Health drift is still wired through `record_read_*`.
        let regs: Vec<_> = self.registrations.read().await.clone();
        for reg in regs.iter() {
            if !reg.enabled {
                continue;
            }
            let Some(registry) = reg.backend.as_connector_registry() else {
                continue;
            };
            match registry.resolve_connector_uid(client_native_id).await {
                Ok(Some(uid)) => {
                    self.record_read_success(reg).await;
                    return Ok(uid);
                }
                Ok(None) => {
                    self.record_read_success(reg).await;
                    if let Ok(parsed) = Uuid::parse_str(client_native_id) {
                        match registry.get_connector(parsed).await {
                            Ok(Some(_)) => return Ok(parsed),
                            Ok(None) => {}
                            Err(_) => {
                                self.record_read_failure(reg).await;
                            }
                        }
                    }
                }
                Err(_) => {
                    self.record_read_failure(reg).await;
                }
            }
        }
        Err(ArchivistError::ConnectorUnknown(Uuid::nil()))
    }

    /// List connectors in the selected archive (non-aliases only).
    ///
    /// When `archive` is `Some`, the explicit override still resolves directly
    /// against that named backend (returning `ArchiveNameUnknown` /
    /// `CapabilityNotSupported` as appropriate). When `None`, routing walks
    /// enabled backends in `read_priority` order and returns the first
    /// `ConnectorRegistry`-capable answer.
    pub async fn list_connectors(
        &self,
        archive: Option<String>,
    ) -> Result<Vec<ConnectorRecord>> {
        if let Some(name) = archive {
            let reg = self
                .find_registration(&name)
                .await
                .ok_or(ArchivistError::ArchiveNameUnknown(name))?;
            let registry = reg.backend.as_connector_registry().ok_or_else(|| {
                ArchivistError::CapabilityNotSupported {
                    capability: ArchiveCapability::ConnectorRegistry,
                    backend: reg.name.clone(),
                }
            })?;
            return registry.list_connectors().await;
        }
        Ok(self
            .read_walk_collection(
                |reg| reg.backend.as_connector_registry().is_some(),
                |backend| async move {
                    let cr = backend
                        .as_connector_registry()
                        .expect("predicate ensured");
                    cr.list_connectors().await
                },
            )
            .await?
            .unwrap_or_default())
    }

    /// Update the stable fingerprint of an existing connector.
    ///
    /// NOTE: read-mutate-write on the backend side; falls through to inline
    /// under `WritePolicy::Queued` (no `WriteOp` variant).
    pub async fn update_connector_fingerprint(
        &self,
        connector_uid: Uuid,
        fingerprint: String,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let primary_reg = primary.backend.as_connector_registry().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks ConnectorRegistry capability".into(),
            }
        })?;
        if let Err(e) = primary_reg
            .update_connector_fingerprint(connector_uid, fingerprint.clone())
            .await
        {
            self.record_write_failure(&primary, &format!("{e}")).await;
            return Err(e);
        }
        self.record_write_success(&primary).await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            let Some(sec_reg) = reg.backend.as_connector_registry() else {
                tracing::debug!(
                    backend = reg.name.as_str(),
                    type_name = reg.type_name,
                    op = "update_connector_fingerprint",
                    "capability_skip"
                );
                continue;
            };
            if let Err(e) = sec_reg
                .update_connector_fingerprint(connector_uid, fingerprint.clone())
                .await
            {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
            } else {
                self.record_write_success(reg).await;
            }
        }
        Ok(())
    }
}
