klieo-ops 3.3.0

//! `JobQueueEscalation` default impl. Tickets are published as `Job`s on
//! the configured `JobQueue` under subject `escalation.{severity}`.
//! Resolution + listing are kept in an in-memory map keyed by
//! `EscalationId`.
//!
//! Production deployments back the `JobQueue` with NATS JetStream
//! (`klieo-bus-nats`), giving the escalation channel persistence +
//! at-least-once delivery semantics. The in-memory map for resolution
//! state is a Phase B limitation — to be extended into the KvStore in
//! M6 follow-up or M8.

use super::trait_::{
    Escalation, EscalationError, EscalationFilter, EscalationId, EscalationState, EscalationTicket,
    Resolution, ResolutionOutcome, Severity,
};
use crate::audit::OpsAuditSink;
use crate::ops_event::OpsEvent;
use async_trait::async_trait;
use bytes::Bytes;
use dashmap::DashMap;
use klieo_core::bus::Job;
use klieo_core::ids::RunId;
use klieo_core::memory::EpisodicMemory;
use klieo_core::JobQueue;
use std::sync::Arc;
use ulid::Ulid;

/// Default `Escalation` impl. Writes tickets to a `JobQueue`; tracks
/// resolution state in an in-process DashMap (sufficient for single-process
/// deployments; multi-process needs a follow-up KV-backed state store).
pub struct JobQueueEscalation {
    jobs: Arc<dyn JobQueue>,
    tickets: DashMap<EscalationId, (EscalationTicket, EscalationState)>,
    audit: Option<OpsAuditSink>,
}

impl JobQueueEscalation {
    /// Construct from a shared `JobQueue`.
    #[must_use]
    pub fn new(jobs: Arc<dyn JobQueue>) -> Self {
        Self {
            jobs,
            tickets: DashMap::new(),
            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(OpsAuditSink::new(
            episodic,
            run_id,
            "klieo.ops.escalation.audit",
        ));
        self
    }

    async fn emit(&self, event: OpsEvent) {
        if let Some(sink) = &self.audit {
            sink.emit(event).await;
        }
    }

    fn subject_for(severity: Severity) -> String {
        let label = match severity {
            Severity::Critical => "critical",
            Severity::High => "high",
            Severity::Medium => "medium",
            Severity::Low => "low",
        };
        format!("escalation.{label}")
    }
}

#[async_trait]
impl Escalation for JobQueueEscalation {
    async fn raise(&self, ticket: EscalationTicket) -> Result<EscalationId, EscalationError> {
        let id = EscalationId(format!("esc_{}", Ulid::new()));
        let subject = Self::subject_for(ticket.severity);
        let body = serde_json::to_vec(&ticket)
            .map_err(|e| EscalationError::Internal(format!("serialise ticket: {e}")))?;
        let job = Job::new(Bytes::from(body));
        self.jobs
            .enqueue(&subject, job)
            .await
            .map_err(|e| EscalationError::Unavailable(format!("enqueue: {e}")))?;
        self.tickets
            .insert(id.clone(), (ticket.clone(), EscalationState::Raised));
        let severity_str = format!("{:?}", ticket.severity).to_lowercase();
        let provenance_root = ticket.provenance.as_ref().map(|p| p.merkle_root.clone());
        self.emit(OpsEvent::EscalationRaised {
            tenant: ticket.tenant.clone(),
            ticket: id.0.clone(),
            severity: severity_str,
            reason: ticket.reason.clone(),
            provenance_root,
        })
        .await;
        Ok(id)
    }

    async fn resolve(
        &self,
        id: EscalationId,
        resolution: Resolution,
    ) -> Result<(), EscalationError> {
        // Phase A: quorum check is the caller's responsibility (M8 FourEyesGate
        // wires it). Here we simply transition state to the resolution's
        // outcome and keep the resolution payload alongside the ticket for
        // later inspection.
        let mut entry = self
            .tickets
            .get_mut(&id)
            .ok_or_else(|| EscalationError::UnknownTicket(id.clone()))?;
        let new_state = match resolution.outcome {
            ResolutionOutcome::Approved => EscalationState::Resolved,
            ResolutionOutcome::Denied => EscalationState::AutoDenied,
            ResolutionOutcome::TimedOut => EscalationState::AutoDenied,
            ResolutionOutcome::Halted => EscalationState::Halted,
        };
        let tenant = entry.0.tenant.clone();
        entry.1 = new_state;
        drop(entry);
        let outcome_str = format!("{:?}", resolution.outcome).to_lowercase();
        self.emit(OpsEvent::EscalationResolved {
            tenant,
            ticket: id.0.clone(),
            outcome: outcome_str,
            reason: resolution.reason.clone(),
        })
        .await;
        Ok(())
    }

    async fn list(&self, filter: EscalationFilter) -> Vec<EscalationTicket> {
        self.tickets
            .iter()
            .filter(|entry| {
                let (ticket, state) = entry.value();
                if let Some(states) = &filter.states {
                    if !states.contains(state) {
                        return false;
                    }
                }
                if let Some(min_sev) = filter.min_severity {
                    let order = |s: Severity| match s {
                        Severity::Low => 0u8,
                        Severity::Medium => 1,
                        Severity::High => 2,
                        Severity::Critical => 3,
                    };
                    if order(ticket.severity) < order(min_sev) {
                        return false;
                    }
                }
                if let Some(tenant) = &filter.tenant {
                    if ticket.tenant.as_ref() != Some(tenant) {
                        return false;
                    }
                }
                true
            })
            .map(|entry| entry.value().0.clone())
            .collect()
    }
}