//! Meta events, DAG, and cleanup orchestration for `Archivist`.
//!
//! Ported from `FileBasedArchivist` in `archivist.rs`. Meta events and DAG
//! methods are thin delegates over `as_meta_events()` / `as_dag()`;
//! `get_session_tree` performs a recursive DAG walk; `cleanup_empty_sessions`
//! pages through all sessions and deletes those with zero messages (skipping
//! `SessionKind::AcpConnection` meta sessions, which track events rather than
//! messages).

use uuid::Uuid;

use crate::backend::ArchiveCapability;
use crate::coordinator::Archivist;
use crate::error::{ArchivistError, Result};
use crate::types::{
    DagEdge, MetaEventRecord, SessionKind, SessionListQuery, SessionMetadata, MAX_PAGE_LIMIT,
};

impl Archivist {
    // ------------------------------------------------------------------
    // Meta events
    // ------------------------------------------------------------------

    pub async fn append_meta_events(
        &self,
        scroll_id: Uuid,
        events: Vec<MetaEventRecord>,
        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();

        // Primary must have MetaEvents capability even in the queued path —
        // the writer task dispatches to `as_meta_events()`, so we'd silently
        // drop events on an incapable primary. Fail fast here.
        let _ = primary.backend.as_meta_events().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks MetaEvents capability".into(),
            }
        })?;

        match &primary.write_policy {
            crate::registry::WritePolicy::Inline => {
                let primary_meta = primary
                    .backend
                    .as_meta_events()
                    .expect("capability checked above");
                if let Err(e) = primary_meta
                    .append_meta_events(scroll_id, events.clone())
                    .await
                {
                    self.record_write_failure(&primary, &format!("{e}")).await;
                    return Err(e);
                }
                self.record_write_success(&primary).await;
            }
            crate::registry::WritePolicy::Queued { .. } => {
                let writer = primary
                    .writer
                    .as_ref()
                    .expect("queued policy implies writer handle present");
                writer
                    .enqueue(crate::registry::writer::WriteOp::AppendMetaEvents {
                        scroll_id,
                        events: events.clone(),
                    })
                    .await?;
            }
        }

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "append_meta_events",
                    "filter_skip"
                );
                continue;
            }
            if reg.backend.as_meta_events().is_none() {
                tracing::debug!(
                    backend = reg.name.as_str(),
                    type_name = reg.type_name,
                    op = "append_meta_events",
                    "capability_skip"
                );
                continue;
            }
            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    let me = reg
                        .backend
                        .as_meta_events()
                        .expect("capability checked above");
                    if let Err(e) = me.append_meta_events(scroll_id, events.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;
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    let writer = reg
                        .writer
                        .as_ref()
                        .expect("queued policy implies writer handle present");
                    if let Err(e) = writer
                        .enqueue(crate::registry::writer::WriteOp::AppendMetaEvents {
                            scroll_id,
                            events: events.clone(),
                        })
                        .await
                    {
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    pub async fn get_meta_events(
        &self,
        scroll_id: Uuid,
        _archive: Option<String>,
    ) -> Result<Vec<MetaEventRecord>> {
        // `archive` is now ignored for reads; routing picks the highest-priority
        // backend that has the session and supports `MetaEvents`.
        Ok(self
            .read_walk_per_session(
                scroll_id,
                |reg| reg.backend.as_meta_events().is_some(),
                |backend| async move {
                    let me = backend.as_meta_events().expect("predicate ensured");
                    me.get_meta_events(scroll_id).await.map(Some)
                },
            )
            .await?
            .unwrap_or_default())
    }

    /// Update the connection status of an ACP meta-session.
    ///
    /// NOTE: read-mutate-write on the backend side (the impl rewrites fields
    /// on the stored session); falls through to inline under
    /// `WritePolicy::Queued` (no `WriteOp` variant).
    pub async fn update_meta_session_status(
        &self,
        scroll_id: Uuid,
        is_connected: bool,
        current_session_id: Option<Uuid>,
        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_meta = primary.backend.as_meta_events().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks MetaEvents capability".into(),
            }
        })?;
        if let Err(e) = primary_meta
            .update_meta_session_status(scroll_id, is_connected, current_session_id)
            .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(me) = reg.backend.as_meta_events() else {
                tracing::debug!(
                    backend = reg.name.as_str(),
                    type_name = reg.type_name,
                    op = "update_meta_session_status",
                    "capability_skip"
                );
                continue;
            };
            if let Err(e) = me
                .update_meta_session_status(scroll_id, is_connected, current_session_id)
                .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(())
    }

    pub async fn list_meta_sessions(
        &self,
        _archive: Option<String>,
    ) -> Result<Vec<SessionMetadata>> {
        // Collection-shape read: first enabled/healthy backend that supports
        // `MetaEvents` wins. `archive` override is no longer honoured here —
        // routing decides.
        Ok(self
            .read_walk_collection(
                |reg| reg.backend.as_meta_events().is_some(),
                |backend| async move {
                    let me = backend.as_meta_events().expect("predicate ensured");
                    me.list_meta_sessions().await
                },
            )
            .await?
            .unwrap_or_default())
    }

    pub async fn find_meta_session_by_client(
        &self,
        client_id: &str,
        _archive: Option<String>,
    ) -> Result<Option<SessionMetadata>> {
        // Collection-shape read: first enabled/healthy backend that supports
        // `MetaEvents` wins. The inner op returns `Result<Option<_>>`, so the
        // walker's outer `Option` flattens to the inner one — "no backend
        // answered" and "backend answered None" collapse the same way.
        let client_id = client_id.to_string();
        let result = self
            .read_walk_collection(
                |reg| reg.backend.as_meta_events().is_some(),
                |backend| {
                    let client_id = client_id.clone();
                    async move {
                        let me = backend.as_meta_events().expect("predicate ensured");
                        me.find_meta_session_by_client(&client_id).await
                    }
                },
            )
            .await?;
        Ok(result.flatten())
    }

    // ------------------------------------------------------------------
    // DAG
    // ------------------------------------------------------------------

    pub async fn append_dag_edge(
        &self,
        edge: DagEdge,
        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();

        // Primary must have DAG capability — even in the queued path the
        // writer task dispatches to `as_dag()`, so silently accepting a
        // non-DAG primary would lose the edge.
        let _ = primary.backend.as_dag().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks Dag capability".into(),
            }
        })?;

        match &primary.write_policy {
            crate::registry::WritePolicy::Inline => {
                let primary_dag = primary
                    .backend
                    .as_dag()
                    .expect("capability checked above");
                if let Err(e) = primary_dag.append_dag_edge(edge.clone()).await {
                    self.record_write_failure(&primary, &format!("{e}")).await;
                    return Err(e);
                }
                self.record_write_success(&primary).await;
            }
            crate::registry::WritePolicy::Queued { .. } => {
                let writer = primary
                    .writer
                    .as_ref()
                    .expect("queued policy implies writer handle present");
                writer
                    .enqueue(crate::registry::writer::WriteOp::AppendDagEdge(edge.clone()))
                    .await?;
            }
        }

        // DAG edges are indexed under the parent scroll_id, so use that for
        // filter evaluation (the session whose DAG is being extended).
        let parent_scroll_id = edge.parent;
        let session_metadata_for_filter = self
            .load_metadata_for_filter(parent_scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %parent_scroll_id,
                    op = "append_dag_edge",
                    "filter_skip"
                );
                continue;
            }
            if reg.backend.as_dag().is_none() {
                tracing::debug!(
                    backend = reg.name.as_str(),
                    type_name = reg.type_name,
                    op = "append_dag_edge",
                    "capability_skip"
                );
                continue;
            }
            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    let d = reg.backend.as_dag().expect("capability checked above");
                    if let Err(e) = d.append_dag_edge(edge.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;
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    let writer = reg
                        .writer
                        .as_ref()
                        .expect("queued policy implies writer handle present");
                    if let Err(e) = writer
                        .enqueue(crate::registry::writer::WriteOp::AppendDagEdge(edge.clone()))
                        .await
                    {
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    pub async fn get_children(
        &self,
        scroll_id: Uuid,
        _archive: Option<String>,
    ) -> Result<Vec<SessionMetadata>> {
        // `archive` is now ignored for reads; routing picks the highest-priority
        // backend that has the session and supports `Dag`.
        Ok(self
            .read_walk_per_session(
                scroll_id,
                |reg| reg.backend.as_dag().is_some(),
                |backend| async move {
                    let d = backend.as_dag().expect("predicate ensured");
                    d.get_children(scroll_id).await.map(Some)
                },
            )
            .await?
            .unwrap_or_default())
    }

    /// Recursive DAG walk rooted at `root_scroll_id`.
    ///
    /// Matches the shape of `FileBasedArchivist::get_session_tree`: returns
    /// every edge reachable from `root_scroll_id` (children, grandchildren,
    /// …). Uses `DagBackend::get_dag_edges` per-parent plus a `seen` set to
    /// guard against cycles.
    pub async fn get_session_tree(
        &self,
        root_scroll_id: Uuid,
        archive: Option<String>,
    ) -> Result<Vec<DagEdge>> {
        // TODO(phase3): consider multi-backend DAG walk in a future phase —
        // current impl uses the default backend only. Consistent BFS across
        // a tree requires all `get_dag_edges` calls to target the SAME
        // backend as the root, which the walker API does not yet expose.
        let backend = self.resolve_backend(archive).await?;
        let dag = backend
            .as_dag()
            .ok_or_else(|| ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::Dag,
                backend: "selected".into(),
            })?;

        let mut out = Vec::new();
        let mut stack = vec![root_scroll_id];
        let mut seen = std::collections::HashSet::new();
        while let Some(parent) = stack.pop() {
            if !seen.insert(parent) {
                continue;
            }
            let edges = dag.get_dag_edges(parent).await?;
            for e in &edges {
                stack.push(e.child);
            }
            out.extend(edges);
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // Cleanup
    // ------------------------------------------------------------------

    /// Delete sessions that have zero messages.
    ///
    /// Ported from `FileBasedArchivist::cleanup_empty_sessions`. Pages through
    /// every session (including hidden ones) via `list_sessions_paged`, counts
    /// messages per session, and deletes those with zero. Meta sessions
    /// (`SessionKind::AcpConnection`) are skipped — they track connection
    /// events in `events.jsonl`, not messages, so an empty message log is
    /// expected.
    ///
    /// Returns `(deleted, total_scanned)`.
    pub async fn cleanup_empty_sessions(
        &self,
        archive: Option<String>,
    ) -> Result<(usize, usize)> {
        let backend = self.resolve_backend(archive).await?;

        let mut total: usize = 0;
        let mut deleted: usize = 0;
        let mut q = SessionListQuery {
            include_hidden: true,
            limit: MAX_PAGE_LIMIT,
            ..SessionListQuery::default()
        };

        loop {
            let page = backend.list_sessions_paged(q.clone()).await?;

            for session in page.items.iter() {
                total += 1;

                // Skip meta sessions - they track events, not messages, so a
                // zero message count is expected and not a signal of emptiness.
                if session.kind == SessionKind::AcpConnection {
                    tracing::debug!(
                        scroll_id = %session.scroll_id,
                        "Skipping meta session (AcpConnection) during cleanup"
                    );
                    continue;
                }

                let count = match backend.count_messages(session.scroll_id).await {
                    Ok(c) => c,
                    Err(e) => {
                        // Match legacy semantics: if we can't count messages,
                        // skip this session rather than risk deleting a
                        // non-empty one.
                        tracing::warn!(
                            scroll_id = %session.scroll_id,
                            error = %e,
                            "Failed to count messages for session, skipping cleanup"
                        );
                        continue;
                    }
                };

                if count == 0 {
                    match backend.delete_session(session.scroll_id).await {
                        Ok(()) => {
                            tracing::info!(
                                scroll_id = %session.scroll_id,
                                "Deleted empty session during cleanup"
                            );
                            deleted += 1;
                        }
                        Err(e) => {
                            tracing::warn!(
                                scroll_id = %session.scroll_id,
                                error = %e,
                                "Failed to delete empty session during cleanup"
                            );
                        }
                    }
                }
            }

            match page.next_cursor {
                Some(cursor) => q.cursor = Some(cursor),
                None => break,
            }
        }

        tracing::info!(
            deleted = deleted,
            total = total,
            "Completed empty session cleanup"
        );

        Ok((deleted, total))
    }
}
