klieo-bus-memory 2.2.0

#![deny(missing_docs)]
#![deny(rust_2018_idioms)]
#![deny(rustdoc::broken_intra_doc_links)]

//! In-process implementations of `klieo-core`'s bus traits
//! ([`Pubsub`], [`RequestReply`],
//! [`KvStore`], [`JobQueue`]).
//!
//! Use this crate for tests, single-binary applications, and local dev.
//! For production multi-process deployment use `klieo-bus-nats`.
//!
//! # Limitations
//!
//! - **No durability across process restart.** Pubsub messages and
//!   queued jobs live in process memory.
//! - **Pubsub `AckHandle` is a no-op.** In-process pubsub has no
//!   redelivery semantics; the trait surface is preserved for
//!   consistency but `ack` / `nak` / `term` succeed without effect.
//! - **JobQueue lease-expiry is lazy** — checked on the next `claim()`
//!   call, not via a background task. Sufficient for tests and
//!   single-binary apps where claim cadence is high.
//!
//! # Quickstart
//!
//! ```no_run
//! use klieo_bus_memory::MemoryBus;
//! let bus = MemoryBus::new();
//! // bus.pubsub, bus.kv, bus.request_reply, bus.jobs are
//! // Arc<dyn …> ready to drop into AgentContext.
//! # let _ = bus;
//! ```

mod jobs;
mod kv;
mod pubsub;
mod request_reply;

pub use jobs::MemoryJobQueue;
pub use kv::MemoryKv;
pub use pubsub::MemoryPubsub;
pub use request_reply::MemoryRequestReply;

use klieo_core::{BusHandles, JobQueue, KvStore, Pubsub, RequestReply};
use std::sync::Arc;

/// Wired-together in-process bus. Build once, share across agents.
pub struct MemoryBus {
    /// Pub/sub.
    pub pubsub: Arc<dyn Pubsub>,
    /// Synchronous request/reply, riding on `pubsub`.
    pub request_reply: Arc<dyn RequestReply>,
    /// KV store.
    pub kv: Arc<dyn KvStore>,
    /// Durable job queue with DLQ + dedup-CAS, riding on `pubsub` + `kv`.
    pub jobs: Arc<dyn JobQueue>,
}

impl MemoryBus {
    /// Build a fresh in-process bus.
    pub fn new() -> Self {
        let pubsub_concrete = Arc::new(MemoryPubsub::new());
        let kv_concrete = Arc::new(MemoryKv::new());
        let rr_concrete = Arc::new(MemoryRequestReply::new(pubsub_concrete.clone()));
        let jobs_concrete = Arc::new(MemoryJobQueue::new(
            pubsub_concrete.clone(),
            kv_concrete.clone(),
        ));
        Self {
            pubsub: pubsub_concrete,
            request_reply: rr_concrete,
            kv: kv_concrete,
            jobs: jobs_concrete,
        }
    }

    /// Capability-shaped default — build a fresh in-process bus and
    /// wrap it in [`Arc`] for cheap cross-task sharing.
    ///
    /// Equivalent to `Arc::new(MemoryBus::new())`. Lead-with this
    /// constructor when wiring agents that hold the bus behind an
    /// `Arc<MemoryBus>`; reach for [`Self::new`] when you want owned
    /// access to the four trait handles via field destructuring.
    pub fn shared() -> Arc<Self> {
        Arc::new(Self::new())
    }
}

impl Default for MemoryBus {
    fn default() -> Self {
        Self::new()
    }
}

impl From<MemoryBus> for BusHandles {
    fn from(value: MemoryBus) -> Self {
        BusHandles::new(value.pubsub, value.kv, value.request_reply, value.jobs)
    }
}

#[cfg(test)]
mod lib_tests {
    use super::*;
    use bytes::Bytes;
    use klieo_core::bus::{Headers, Job};
    use klieo_core::ids::DurableName;
    use std::time::Duration;
    use tokio_stream::StreamExt;

    #[tokio::test]
    async fn shared_returns_arc_wired_and_usable() {
        let bus = MemoryBus::shared();
        bus.kv
            .put("b", "k", Bytes::from_static(b"shared"))
            .await
            .unwrap();
        let entry = bus.kv.get("b", "k").await.unwrap().unwrap();
        assert_eq!(entry.value, Bytes::from_static(b"shared"));
        // Cheap clone — same underlying state.
        let bus2 = Arc::clone(&bus);
        let entry2 = bus2.kv.get("b", "k").await.unwrap().unwrap();
        assert_eq!(entry2.value, Bytes::from_static(b"shared"));
    }

    #[tokio::test]
    async fn memory_bus_wires_all_four() {
        let bus = MemoryBus::new();
        bus.kv
            .put("b", "k", Bytes::from_static(b"v"))
            .await
            .unwrap();
        let entry = bus.kv.get("b", "k").await.unwrap().unwrap();
        assert_eq!(entry.value, Bytes::from_static(b"v"));
        let mut sub = bus
            .pubsub
            .subscribe("s", DurableName::new("d"))
            .await
            .unwrap();
        bus.pubsub
            .publish("s", Bytes::from_static(b"hi"), Headers::new())
            .await
            .unwrap();
        let msg = sub.next().await.unwrap().unwrap();
        assert_eq!(msg.payload, Bytes::from_static(b"hi"));
        let err = bus
            .request_reply
            .request("svc", Bytes::from_static(b"x"), Duration::from_millis(20))
            .await
            .unwrap_err();
        assert!(matches!(err, klieo_core::BusError::Timeout));
        let id = bus
            .jobs
            .enqueue("q", Job::new(Bytes::from_static(b"j")))
            .await
            .unwrap();
        let claimed = bus
            .jobs
            .claim("q", "w", Duration::from_secs(1))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(claimed.id, id);
    }

    #[tokio::test]
    async fn bus_handles_from_memory_bus_preserves_handles() {
        let bus = MemoryBus::new();
        let ps = Arc::as_ptr(&bus.pubsub);
        let kv = Arc::as_ptr(&bus.kv);
        let rr = Arc::as_ptr(&bus.request_reply);
        let jq = Arc::as_ptr(&bus.jobs);
        let handles: BusHandles = bus.into();
        assert!(std::ptr::addr_eq(Arc::as_ptr(&handles.pubsub), ps));
        assert!(std::ptr::addr_eq(Arc::as_ptr(&handles.kv), kv));
        assert!(std::ptr::addr_eq(Arc::as_ptr(&handles.request_reply), rr));
        assert!(std::ptr::addr_eq(Arc::as_ptr(&handles.jobs), jq));
    }
}