klieo-ops 3.5.0

//! JobQueue passthrough channel adapter. Re-enqueues each ticket onto a
//! caller-specified target subject — useful when an internal ticketing
//! / ITSM service has its own JobQueue consumer in the same NATS cluster.

use super::{ChannelAdapter, ChannelError};
use crate::escalation::trait_::EscalationTicket;
use async_trait::async_trait;
use bytes::Bytes;
use klieo_core::bus::Job;
use klieo_core::JobQueue;
use std::sync::Arc;

/// Re-enqueue tickets onto a target JobQueue subject.
pub struct JobQueueChannel {
    jobs: Arc<dyn JobQueue>,
    target_subject: String,
}

impl JobQueueChannel {
    /// Build with the target subject.
    #[must_use]
    pub fn new(jobs: Arc<dyn JobQueue>, target_subject: impl Into<String>) -> Self {
        Self {
            jobs,
            target_subject: target_subject.into(),
        }
    }
}

#[async_trait]
impl ChannelAdapter for JobQueueChannel {
    async fn deliver(&self, ticket: &EscalationTicket) -> Result<(), ChannelError> {
        let body = serde_json::to_vec(ticket)
            .map_err(|e| ChannelError::Internal(format!("serialise ticket: {e}")))?;
        let job = Job::new(Bytes::from(body));
        self.jobs
            .enqueue(&self.target_subject, job)
            .await
            .map_err(|e| {
                ChannelError::Unavailable(format!("enqueue {}: {e}", self.target_subject))
            })?;
        Ok(())
    }

    fn name(&self) -> &'static str {
        "JobQueueChannel"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::escalation::trait_::{EscalationTicket, Severity};
    use klieo_bus_memory::MemoryBus;

    #[tokio::test]
    async fn jobqueue_channel_enqueues_ticket() {
        let bus = MemoryBus::new();
        let ch = JobQueueChannel::new(bus.jobs.clone(), "itsm.incoming");
        let ticket = EscalationTicket {
            tenant: None,
            severity: Severity::Medium,
            reason: "test".into(),
            provenance: None,
        };
        ch.deliver(&ticket).await.expect("enqueue");
    }
}