klieo-ops 0.3.0

//! KV-backed WorkLog impl. Items stored under `worklog/items/<id>`;
//! adjacency list `worklog/parents/<id> -> [parent_ids]` and
//! `worklog/children/<id> -> [child_ids]`. Cycle detection via DFS on
//! `depend()`. Status transitions emit `OpsEvent::WorkTransition`.
//!
//! Phase B M7 caps the DAG at 10 000 items per WorkLog instance and
//! does not yet implement orphan TTL (the 24h sweep noted in design
//! spec § 2.5 is operator-cron territory; M7 follow-up adds the
//! re-dispatcher binary that owns this sweep).

use super::trait_::{WorkDag, WorkId, WorkItem, WorkLog, WorkLogError, WorkStatus};
use crate::emit::emit_ops_event;
use crate::ops_event::OpsEvent;
use async_trait::async_trait;
use bytes::Bytes;
use chrono::Utc;
use dashmap::DashMap;
use futures::future::try_join_all;
use futures_core::stream::BoxStream;
use klieo_core::ids::RunId;
use klieo_core::memory::EpisodicMemory;
use klieo_core::KvStore;
use std::collections::{HashSet, VecDeque};
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::StreamExt;
use ulid::Ulid;

#[allow(dead_code)]
const BUCKET_ITEMS: &str = "ops.worklog.items";
#[allow(dead_code)]
const BUCKET_PARENTS: &str = "ops.worklog.parents";
#[allow(dead_code)]
const BUCKET_CHILDREN: &str = "ops.worklog.children";

const DEFAULT_MAX_ITEMS: usize = 10_000;

/// Optional audit sink. When present, every state-changing call emits the
/// corresponding `OpsEvent` into `episodic` under `run_id`.
struct AuditSink {
    episodic: Arc<dyn EpisodicMemory>,
    run_id: RunId,
}

/// KV-backed worklog.
pub struct KvWorkLog {
    kv: Arc<dyn KvStore>,
    // In-process cache + adjacency for cycle detection. KV is the source
    // of truth across processes (Phase B follow-up wires multi-process
    // coordination); this cache speeds same-process operations.
    items: DashMap<WorkId, WorkItem>,
    parents: DashMap<WorkId, HashSet<WorkId>>,
    children: DashMap<WorkId, HashSet<WorkId>>,
    cap: usize,
    ready_tx: broadcast::Sender<WorkId>,
    audit: Option<AuditSink>,
}

impl KvWorkLog {
    /// Build with default 10 000-item cap.
    #[must_use]
    pub fn new(kv: Arc<dyn KvStore>) -> Self {
        Self::with_cap(kv, DEFAULT_MAX_ITEMS)
    }

    /// Build with a custom DAG size cap.
    #[must_use]
    pub fn with_cap(kv: Arc<dyn KvStore>, cap: usize) -> Self {
        let (tx, _rx) = broadcast::channel(1024);
        Self {
            kv,
            items: DashMap::new(),
            parents: DashMap::new(),
            children: DashMap::new(),
            cap,
            ready_tx: tx,
            audit: None,
        }
    }

    /// Wire an audit sink. When set, every state-changing call emits the
    /// corresponding `OpsEvent` into `episodic` under `run_id`.
    #[must_use]
    pub fn with_audit(mut self, episodic: Arc<dyn EpisodicMemory>, run_id: RunId) -> Self {
        self.audit = Some(AuditSink { episodic, run_id });
        self
    }

    async fn emit(&self, event: OpsEvent) {
        if let Some(sink) = &self.audit {
            if let Err(err) = emit_ops_event(&*sink.episodic, sink.run_id, event).await {
                tracing::warn!(
                    target: "klieo.ops.worklog.audit",
                    error = %err,
                    "audit emit failed"
                );
            }
        }
    }

    async fn persist_item(&self, item: &WorkItem) -> Result<(), WorkLogError> {
        let body = serde_json::to_vec(item)
            .map_err(|e| WorkLogError::Internal(format!("serialise item: {e}")))?;
        self.kv
            .put(BUCKET_ITEMS, &item.id.0, Bytes::from(body))
            .await
            .map(|_| ())
            .map_err(|e| WorkLogError::Storage(e.to_string()))
    }

    /// Returns true if adding child -> on would create a cycle.
    fn would_cycle(&self, child: &WorkId, on: &WorkId) -> bool {
        // BFS upward from `on` along parents; if we reach `child`, cycle.
        let mut queue: VecDeque<WorkId> = VecDeque::new();
        let mut visited: HashSet<WorkId> = HashSet::new();
        queue.push_back(on.clone());
        while let Some(node) = queue.pop_front() {
            if &node == child {
                return true;
            }
            if !visited.insert(node.clone()) {
                continue;
            }
            if let Some(parents) = self.parents.get(&node) {
                for parent in parents.iter() {
                    queue.push_back(parent.clone());
                }
            }
        }
        false
    }

    fn dependencies_satisfied(&self, id: &WorkId) -> bool {
        match self.parents.get(id) {
            Some(parents) => parents.iter().all(|p| {
                self.items
                    .get(p)
                    .is_some_and(|it| it.status == WorkStatus::Done)
            }),
            None => true,
        }
    }

    fn announce_ready(&self, id: WorkId) {
        if let Err(err) = self.ready_tx.send(id) {
            // No live subscribers; `ready_stream()` consumers may have dropped.
            // Surface so operators see the gap rather than silently lose readiness signals.
            tracing::debug!(
                target: "klieo.ops.worklog",
                work_id = %err.0,
                "ready-stream has no live receivers; dropping notification"
            );
        }
    }
}

#[async_trait]
impl WorkLog for KvWorkLog {
    async fn plan(&self, mut item: WorkItem) -> Result<WorkId, WorkLogError> {
        if self.items.len() >= self.cap {
            return Err(WorkLogError::CapExceeded { cap: self.cap });
        }
        let id = WorkId(format!("wrk_{}", Ulid::new()));
        item.id = id.clone();
        item.last_transition_at = Utc::now().to_rfc3339();
        // Initial status: Ready if no deps, else Planned.
        if item.depends_on.is_empty() {
            item.status = WorkStatus::Ready;
        } else {
            item.status = WorkStatus::Planned;
        }
        // Validate all parents exist before touching any shared state.
        let initial_deps = std::mem::take(&mut item.depends_on);
        for parent in &initial_deps {
            if !self.items.contains_key(parent) {
                return Err(WorkLogError::UnknownWorkItem(parent.clone()));
            }
        }
        // A freshly-created id has no inbound edges yet, so no edge from
        // initial_deps back to id can close a cycle — skip K*BFS calls.
        // Insert edges directly into adjacency maps.
        for parent in &initial_deps {
            self.parents
                .entry(id.clone())
                .or_default()
                .insert(parent.clone());
            self.children
                .entry(parent.clone())
                .or_default()
                .insert(id.clone());
        }
        // Persist only after all edges are registered (no half-state window).
        self.items.insert(id.clone(), item.clone());
        self.persist_item(&item).await?;
        self.emit(OpsEvent::WorkPlanned {
            tenant: item.tenant.clone(),
            work_id: id.0.clone(),
            title: item.title.clone(),
            depends_on: initial_deps.iter().map(|p| p.0.clone()).collect(),
        })
        .await;
        if self.dependencies_satisfied(&id) {
            self.announce_ready(id.clone());
        }
        Ok(id)
    }

    async fn depend(&self, child: WorkId, on: WorkId) -> Result<(), WorkLogError> {
        if !self.items.contains_key(&child) {
            return Err(WorkLogError::UnknownWorkItem(child));
        }
        if !self.items.contains_key(&on) {
            return Err(WorkLogError::UnknownWorkItem(on));
        }
        if self.would_cycle(&child, &on) {
            return Err(WorkLogError::CycleDetected { child, on });
        }
        self.parents
            .entry(child.clone())
            .or_default()
            .insert(on.clone());
        self.children
            .entry(on.clone())
            .or_default()
            .insert(child.clone());
        let tenant = self.items.get(&child).and_then(|e| e.tenant.clone());
        self.emit(OpsEvent::WorkDependencyAdded {
            tenant,
            child: child.0,
            on: on.0,
        })
        .await;
        Ok(())
    }

    async fn ready(&self, limit: usize) -> Vec<WorkId> {
        self.items
            .iter()
            .filter(|e| e.value().status == WorkStatus::Ready)
            .map(|e| e.key().clone())
            .take(limit)
            .collect()
    }

    async fn ready_stream(&self) -> BoxStream<'static, WorkId> {
        let rx = self.ready_tx.subscribe();
        let stream = BroadcastStream::new(rx).filter_map(|r| r.ok());
        Box::pin(stream)
    }

    async fn dispatch(&self, id: WorkId) -> Result<(), WorkLogError> {
        let mut entry = self
            .items
            .get_mut(&id)
            .ok_or_else(|| WorkLogError::UnknownWorkItem(id.clone()))?;
        if entry.status != WorkStatus::Ready {
            return Err(WorkLogError::Internal(format!(
                "cannot dispatch item in status {:?}",
                entry.status
            )));
        }
        entry.status = WorkStatus::InProgress;
        entry.last_transition_at = Utc::now().to_rfc3339();
        let snapshot = entry.clone();
        drop(entry);
        self.persist_item(&snapshot).await?;
        self.emit(OpsEvent::WorkDispatched {
            tenant: snapshot.tenant.clone(),
            work_id: id.0,
        })
        .await;
        Ok(())
    }

    async fn transition(&self, id: WorkId, status: WorkStatus) -> Result<(), WorkLogError> {
        let mut entry = self
            .items
            .get_mut(&id)
            .ok_or_else(|| WorkLogError::UnknownWorkItem(id.clone()))?;
        // Reject retrograde transitions from terminal states.
        if entry.status.is_terminal() && entry.status != status {
            return Err(WorkLogError::Internal(format!(
                "cannot transition from terminal status {:?} to {:?}",
                entry.status, status
            )));
        }
        let prev_status = entry.status;
        entry.status = status;
        entry.last_transition_at = Utc::now().to_rfc3339();
        let snapshot = entry.clone();
        let id_clone = id.clone();
        drop(entry);
        self.persist_item(&snapshot).await?;
        self.emit(OpsEvent::WorkTransition {
            tenant: snapshot.tenant.clone(),
            work_id: id.0.clone(),
            from: format!("{prev_status:?}").to_lowercase(),
            to: format!("{status:?}").to_lowercase(),
            reason: None,
        })
        .await;
        // If we just marked an item Done, check its children to see if
        // any become Ready. Collect all newly-Ready snapshots first, then
        // persist them concurrently to avoid a sequential network RTT per child.
        if status == WorkStatus::Done {
            let mut newly_ready_snaps: Vec<WorkItem> = Vec::new();
            if let Some(children) = self.children.get(&id_clone) {
                let child_ids: Vec<WorkId> = children.iter().cloned().collect();
                drop(children);
                for child in child_ids {
                    if self.dependencies_satisfied(&child) {
                        if let Some(mut child_entry) = self.items.get_mut(&child) {
                            if child_entry.status == WorkStatus::Planned {
                                child_entry.status = WorkStatus::Ready;
                                child_entry.last_transition_at = Utc::now().to_rfc3339();
                                newly_ready_snaps.push(child_entry.clone());
                            }
                        }
                    }
                }
            }
            let persists = newly_ready_snaps.iter().map(|snap| self.persist_item(snap));
            try_join_all(persists).await?;
            for snap in &newly_ready_snaps {
                self.announce_ready(snap.id.clone());
            }
        }
        Ok(())
    }

    async fn get(&self, id: WorkId) -> Option<WorkItem> {
        self.items.get(&id).map(|e| e.value().clone())
    }

    async fn list_by_status(&self, filter: WorkStatus, limit: usize) -> Vec<WorkItem> {
        self.items
            .iter()
            .filter(|e| e.value().status == filter)
            .map(|e| e.value().clone())
            .take(limit)
            .collect()
    }

    async fn dag(&self, root: WorkId) -> WorkDag {
        // BFS over children (downstream) AND parents (upstream) so the
        // returned snapshot is the connected subgraph containing `root`.
        let mut visited: HashSet<WorkId> = HashSet::new();
        let mut queue: VecDeque<WorkId> = VecDeque::new();
        queue.push_back(root);
        let mut out: Vec<WorkItem> = Vec::new();
        while let Some(node) = queue.pop_front() {
            if !visited.insert(node.clone()) {
                continue;
            }
            if let Some(item) = self.items.get(&node) {
                out.push(item.value().clone());
            }
            if let Some(children) = self.children.get(&node) {
                for c in children.iter() {
                    queue.push_back(c.clone());
                }
            }
            if let Some(parents) = self.parents.get(&node) {
                for p in parents.iter() {
                    queue.push_back(p.clone());
                }
            }
        }
        WorkDag { items: out }
    }
}