alkstore/alkstore-mem/src/store.rs

//! The mem engine's store: `MemStore::new`, the engine-native
//! constructor (ADR-024 §4 — the one documented exception to the
//! open prose), and the guarded state it owns — plus the `Store` trait
//! surface this crate's whole point rides on.
//!
//! Isolation per open: each `new()` yields a fully isolated instance
//! sharing nothing — a fresh `State` (the guarded per-mechanism core),
//! a fresh `WakeBus` (the wake-routing substrate), a fresh `Clock` —
//! exactly the suite factory's isolation contract; two instances never
//! contend and never observe each other. There is no connection, no
//! file, no pool; teardown is drop — the engine-wide `Drop` runs the
//! close path (the wake bus empties, every open feed disconnects), and
//! the state is gone with the process's references, the honest-
//! ephemeral posture (ADR-016 carriers, ADR-024 §2: the dependency on
//! `alkstore-mem` *is* the ephemerality declaration).
//!
//! The `Store` surface wires the mechanism handles (core-owned trait
//! objects — engine types never appear in consumer signatures): `queue`
//! carries its `QueueOpts` stamps (queues are names, not config
//! owners), `stream`/`outbox` carry validated names, `try_lock` rides
//! the TTL registry, `schedule`/`unschedule`/`run_schedules` the
//! collapsed scheduler, `notify`/`listen` the wake substrate, and
//! `begin_tx` allocates the tx overlay (`crate::tx` — the load-bearing
//! seam; `with_tx` is the trait's provided composition).

use std::sync::Arc;

use alkstore::{BoxedFuture, Result, StopToken, Store, TxHandle};
use alkstore::{QueueOpts, ScheduleOpts};
use serde_json::Value;

use crate::clock::Clock;
use crate::queue::MemQueueHandle;
use crate::scheduler;
use crate::state::State;
use crate::stream::MemStreamHandle;
use crate::tx::MemTxHandle;
use crate::wakes::WakeBus;

/// The mem engine's store handle. Consumers hold
/// [`alkstore::Store`] — the trait is the contract surface, never the
/// engine type; `MemStore::new` is the engine-native constructor.
pub struct MemStore {
    pub(crate) state: Arc<State>,
    pub(crate) wakes: Arc<WakeBus>,
    pub(crate) clock: Arc<Clock>,
}

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

impl std::fmt::Debug for MemStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemStore").finish_non_exhaustive()
    }
}

impl MemStore {
    /// Construct a fully isolated mem engine instance. Engine-native:
    /// no connection string, no file path (ADR-024 §4's documented
    /// exception to the open prose). Teardown is drop; each instance
    /// shares nothing with any other.
    pub fn new() -> Self {
        Self {
            state: Arc::new(State::new()),
            wakes: Arc::new(WakeBus::new()),
            clock: Arc::new(Clock::new()),
        }
    }

    #[cfg(test)]
    pub(crate) fn wakes(&self) -> &Arc<WakeBus> {
        &self.wakes
    }

    #[cfg(test)]
    pub(crate) fn clock(&self) -> &Arc<Clock> {
        &self.clock
    }
}

impl Drop for MemStore {
    fn drop(&mut self) {
        self.wakes.close();
    }
}

impl Store for MemStore {
    fn begin_tx(&self) -> BoxedFuture<'_, Result<Box<dyn TxHandle + Send>>> {
        let handle = MemTxHandle::begin(self.state.clone(), self.clock.clone(), self.wakes.clone());
        Box::pin(async move { Ok(Box::new(handle) as Box<dyn TxHandle + Send>) })
    }

    fn notify<'a>(&'a self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>> {
        crate::notify::notify(self.wakes.clone(), channel, payload)
    }

    fn listen<'a>(
        &'a self,
        channel: &str,
    ) -> BoxedFuture<'a, Result<Box<dyn alkstore::WakeReceiver>>> {
        let wakes = self.wakes.clone();
        let owned = channel.to_string();
        Box::pin(async move { crate::notify::listen(wakes, &owned) })
    }

    fn stream<'a>(
        &'a self,
        name: &str,
    ) -> BoxedFuture<'a, Result<Box<dyn alkstore::StreamHandle>>> {
        let handle = alkstore::validate_shared_name(name).map(|()| {
            MemStreamHandle::new(
                name.to_string(),
                self.state.clone(),
                self.clock.clone(),
                self.wakes.clone(),
            )
        });
        Box::pin(async move { handle.map(|h| Box::new(h) as Box<dyn alkstore::StreamHandle>) })
    }

    fn queue<'a>(
        &'a self,
        name: &str,
        opts: QueueOpts,
    ) -> BoxedFuture<'a, Result<Box<dyn alkstore::Queue>>> {
        let handle = alkstore::validate_shared_name(name).map(|()| {
            Box::new(MemQueueHandle::new(
                name.to_string(),
                opts,
                self.state.clone(),
                self.clock.clone(),
            )) as Box<dyn alkstore::Queue>
        });
        Box::pin(async move { handle })
    }

    fn outbox<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<Box<dyn alkstore::Outbox>>> {
        let handle = alkstore::validate_shared_name(name).map(|()| {
            Box::new(crate::outbox::MemOutbox::new(
                name.to_string(),
                self.state.clone(),
                self.clock.clone(),
            )) as Box<dyn alkstore::Outbox>
        });
        Box::pin(async move { handle })
    }

    fn try_lock<'a>(
        &'a self,
        name: &str,
        owner: &str,
        ttl: i64,
    ) -> BoxedFuture<'a, Result<Option<Box<dyn alkstore::Lock>>>> {
        let outcome =
            crate::lock::try_lock(self.state.clone(), self.clock.clone(), name, owner, ttl);
        Box::pin(async move { outcome })
    }

    fn schedule<'a>(
        &'a self,
        name: &str,
        spec: &str,
        queue: &str,
        payload: Value,
        opts: ScheduleOpts,
    ) -> BoxedFuture<'a, Result<alkstore::Schedule>> {
        scheduler::schedule(
            self.state.clone(),
            self.clock.clone(),
            name,
            spec,
            queue,
            payload,
            opts,
        )
    }

    fn unschedule<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<bool>> {
        scheduler::unschedule(self.state.clone(), name)
    }

    fn run_schedules<'a>(&'a self, stop: StopToken) -> BoxedFuture<'a, Result<()>> {
        scheduler::run_schedules(self.state.clone(), self.clock.clone(), stop)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkstore::EnqueueOpts;

    #[test]
    fn two_new_instances_share_nothing() {
        let a = MemStore::new();
        let b = MemStore::new();
        assert!(
            !std::sync::Arc::ptr_eq(&a.state, &b.state),
            "the state cores are distinct instances"
        );
        assert!(
            !std::sync::Arc::ptr_eq(&a.wakes, &b.wakes),
            "the wake substrates are distinct instances"
        );
        assert!(
            !std::sync::Arc::ptr_eq(&a.clock, &b.clock),
            "the clocks are distinct instances"
        );
    }

    #[test]
    fn instances_observed_through_their_futures_share_nothing() {
        let a = MemStore::new();
        let b = MemStore::new();
        a.clock().freeze(1_000_000);
        assert_eq!(a.clock().unix_now(), 1_000_000, "instance a's clock froze");
        assert!(
            b.clock().unix_now() > 1_700_000_000,
            "instance b's clock kept the process time"
        );
    }

    #[test]
    fn wake_substrates_do_not_cross_instances() {
        let a = MemStore::new();
        let b = MemStore::new();
        a.wakes().fire_commit_wakes(vec!["orders".into()]);
        let mut late_feed = b.wakes().attach("orders");
        assert!(
            late_feed.rx().try_recv().is_err(),
            "instance a's commit fired before the feed attached — instance b's substrate never replayed it, and instance a's substrate is unreachable from b"
        );
        let mut a_feed = a.wakes().attach("orders");
        a.wakes().fire_commit_wakes(vec!["orders".into()]);
        assert!(
            matches!(
                a_feed.rx().try_recv(),
                Ok(wake) if wake.channel == "orders"
            ),
            "instance a's substrate routes its own attaches"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn engine_drop_closes_open_feeds() {
        let store = MemStore::new();
        let mut feed = store.wakes().attach("orders");
        std::mem::drop(store);
        assert!(
            feed.rx().recv().await.is_none(),
            "teardown is drop: the engine's close path disconnects every feed"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn constructor_entry_validation() {
        let store = MemStore::new();
        for bad in ["", "   ", "__alkstore_q"] {
            match store.queue(bad, QueueOpts::default()).await {
                Err(alkstore::Error::InvalidName { .. } | alkstore::Error::ReservedName { .. }) => {
                }
                _other => panic!("queue({bad:?}) rejects"),
            }
            assert!(store.stream(bad).await.is_err());
            assert!(store.outbox(bad).await.is_err());
        }
        // Nothing was created on a rejected name — the next queue with
        // a fresh name starts clean (ids allocate nowhere).
        let good = store.queue("good", QueueOpts::default()).await.unwrap();
        let id = good
            .enqueue(serde_json::json!(1), EnqueueOpts::default())
            .await
            .unwrap();
        assert_eq!(id, 1, "ids count from 1");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn begin_tx_grants_a_boxed_handle_that_commits() {
        let store = MemStore::new();
        let mut tx = store.begin_tx().await.unwrap();
        tx.enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(1))
            .await
            .unwrap();
        tx.commit().await.unwrap();
        let q = store.queue("q", QueueOpts::default()).await.unwrap();
        assert!(q.get_job(1).await.unwrap().is_some());
    }
}