alkstore 0.1.0

//! Object-safety compile probes and signature-conformance checks for
//! the trait surface (the task's acceptance criterion): a trivial
//! local impl of every trait must construct as `Box<dyn ...>` —
//! proving the whole surface is object-safe and shape-complete.

use std::time::Duration;

use serde_json::{Value, json};

use crate::event_receiver::EventReceiver;
use crate::job::Job;
use crate::job_handle::JobHandle;
use crate::lock::Lock;
use crate::outbox::{Delivery, Outbox};
use crate::queue::Queue;
use crate::stream::StreamEvent;
use crate::stream_handle::StreamHandle;
use crate::tx::TxHandle;
use crate::wake_receiver::WakeReceiver;
use crate::{
    BoxedFuture, EnqueueOpts, Error, QueueOpts, Result, Schedule, ScheduleOpts, StopToken, Store,
    Wake,
};

#[derive(Default)]
struct ProbeTx;

impl TxHandle for ProbeTx {
    fn enqueue_tx<'a>(
        &'a mut self,
        _queue: &str,
        _opts: EnqueueOpts,
        _payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn publish_tx<'a>(
        &'a mut self,
        _stream: &str,
        _payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn publish_with_key_tx<'a>(
        &'a mut self,
        _stream: &str,
        _key: Option<String>,
        _payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn notify_tx<'a>(&'a mut self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
    fn save_offset_tx<'a>(
        &'a mut self,
        _stream: &str,
        _consumer: &str,
        _offset: i64,
    ) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
    fn get_job_tx<'a>(
        &'a mut self,
        _queue: &str,
        _job_id: i64,
    ) -> BoxedFuture<'a, Result<Option<Job>>> {
        Box::pin(std::future::ready(Ok(None)))
    }
    fn get_offset_tx<'a>(
        &'a mut self,
        _stream: &str,
        _consumer: &str,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(0)))
    }
    fn read_since_tx<'a>(
        &'a mut self,
        _stream: &str,
        _offset: i64,
        _limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn read_from_consumer_tx<'a>(
        &'a mut self,
        _stream: &str,
        _consumer: &str,
        _limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn outbox_enqueue_tx<'a>(
        &'a mut self,
        _outbox: &str,
        _opts: EnqueueOpts,
        _payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn commit(self: Box<Self>) -> BoxedFuture<'static, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
}

struct ProbeJob;

impl JobHandle for ProbeJob {
    fn job(&self) -> &Job {
        unreachable!()
    }
    fn ack(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
    fn retry(
        self: Box<Self>,
        _err: Option<String>,
        _delay: Option<i64>,
    ) -> BoxedFuture<'static, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
    fn fail(self: Box<Self>, _err: Option<String>) -> BoxedFuture<'static, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
    fn heartbeat<'a>(&'a self, _extend: i64) -> BoxedFuture<'a, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
}

struct ProbeQueue;

impl Queue for ProbeQueue {
    fn name(&self) -> &str {
        "probe"
    }
    fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn claim_one<'a>(
        &'a self,
        _worker_id: &str,
    ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>> {
        Box::pin(std::future::ready(Ok(None)))
    }
    fn claim_batch<'a>(
        &'a self,
        _worker_id: &str,
        _n: i64,
    ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn ack_batch<'a>(&'a self, _ids: &[i64]) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(0)))
    }
    fn cancel<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result<bool>> {
        Box::pin(std::future::ready(Ok(false)))
    }
    fn get_job<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result<Option<Job>>> {
        Box::pin(std::future::ready(Ok(None)))
    }
    fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(0)))
    }
}

struct ProbeStream;

impl StreamHandle for ProbeStream {
    fn name(&self) -> &str {
        "probe"
    }
    fn publish<'a>(&'a self, _payload: Value) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn publish_with_key<'a>(
        &'a self,
        _key: Option<String>,
        _payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn read_since<'a>(
        &'a self,
        _offset: i64,
        _limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn read_from_consumer<'a>(
        &'a self,
        _consumer: &str,
        _limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn save_offset<'a>(&'a self, _consumer: &str, _offset: i64) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
    fn get_offset<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(0)))
    }
    fn trim_to<'a>(&'a self, _horizon: i64) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(0)))
    }
    fn subscribe<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeEventReceiver) as Box<dyn EventReceiver>
        )))
    }
}

struct ProbeEventReceiver;

impl EventReceiver for ProbeEventReceiver {
    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
        Box::pin(std::future::ready(None))
    }
    fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
        Ok(None)
    }
    fn read_since<'a>(
        &'a mut self,
        _offset: i64,
        _limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        Box::pin(std::future::ready(Ok(Vec::new())))
    }
    fn save_offset(&mut self) -> Result<()> {
        Ok(())
    }
    fn offset(&self) -> i64 {
        0
    }
}

struct ProbeOutbox;

impl Outbox for ProbeOutbox {
    fn name(&self) -> &str {
        "probe"
    }
    fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
        Box::pin(std::future::ready(Ok(1)))
    }
    fn run_once<'a>(
        &'a mut self,
        _worker_id: &str,
        _delivery: &'a mut dyn Delivery,
    ) -> BoxedFuture<'a, Result<bool>> {
        Box::pin(std::future::ready(Ok(false)))
    }
}

struct ProbeLock;

impl Lock for ProbeLock {
    fn name(&self) -> &str {
        "probe"
    }
    fn renew<'a>(&'a self, _ttl: i64) -> BoxedFuture<'a, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
    fn release(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
        Box::pin(std::future::ready(Ok(true)))
    }
}

struct ProbeWakeReceiver;

impl WakeReceiver for ProbeWakeReceiver {
    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Wake>> {
        Box::pin(std::future::ready(None))
    }
    fn try_recv(&mut self) -> Result<Option<Wake>> {
        Err(Error::Closed)
    }
    fn recv_timeout<'a>(&'a mut self, _timeout: Duration) -> BoxedFuture<'a, Result<Option<Wake>>> {
        Box::pin(std::future::ready(Ok(None)))
    }
}

struct ProbeStore;

impl Store for ProbeStore {
    fn begin_tx(&self) -> BoxedFuture<'_, Result<Box<dyn TxHandle + Send>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeTx) as Box<dyn TxHandle + Send>
        )))
    }
    fn notify<'a>(&'a self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
    fn listen<'a>(&'a self, _channel: &str) -> BoxedFuture<'a, Result<Box<dyn WakeReceiver>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeWakeReceiver) as Box<dyn WakeReceiver>
        )))
    }
    fn stream<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<Box<dyn StreamHandle>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeStream) as Box<dyn StreamHandle>
        )))
    }
    fn queue<'a>(
        &'a self,
        _name: &str,
        _opts: QueueOpts,
    ) -> BoxedFuture<'a, Result<Box<dyn Queue>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeQueue) as Box<dyn Queue>
        )))
    }
    fn outbox<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<Box<dyn Outbox>>> {
        Box::pin(std::future::ready(Ok(
            Box::new(ProbeOutbox) as Box<dyn Outbox>
        )))
    }
    fn try_lock<'a>(
        &'a self,
        _name: &str,
        _owner: &str,
        _ttl: i64,
    ) -> BoxedFuture<'a, Result<Option<Box<dyn Lock>>>> {
        Box::pin(std::future::ready(Ok(None)))
    }
    fn schedule<'a>(
        &'a self,
        _name: &str,
        _spec: &str,
        _queue: &str,
        _payload: Value,
        _opts: ScheduleOpts,
    ) -> BoxedFuture<'a, Result<Schedule>> {
        Box::pin(std::future::ready(Ok(Schedule::new(
            "probe".into(),
            "@every 60s".into(),
            "q".into(),
            ScheduleOpts::default(),
        ))))
    }
    fn unschedule<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<bool>> {
        Box::pin(std::future::ready(Ok(false)))
    }
    fn run_schedules<'a>(&'a self, _stop: StopToken) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
}

#[test]
fn trait_objects_construct() {
    let _tx: Box<dyn TxHandle + Send> = Box::new(ProbeTx);
    let _job: Box<dyn JobHandle> = Box::new(ProbeJob);
    let _queue: Box<dyn Queue> = Box::new(ProbeQueue);
    let _stream: Box<dyn StreamHandle> = Box::new(ProbeStream);
    let _receiver: Box<dyn EventReceiver> = Box::new(ProbeEventReceiver);
    let _outbox: Box<dyn Outbox> = Box::new(ProbeOutbox);
    let _lock: Box<dyn Lock> = Box::new(ProbeLock);
    let _wake_receiver: Box<dyn WakeReceiver> = Box::new(ProbeWakeReceiver);
    let _store: Box<dyn Store> = Box::new(ProbeStore);
}

#[tokio::test]
async fn with_tx_commit_and_rollback_dispositions() {
    let store: Box<dyn Store> = Box::new(ProbeStore);

    let committed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let flag = committed.clone();
    store
        .with_tx(Box::new(move |_tx| {
            let flag = flag.clone();
            Box::pin(async move {
                flag.store(true, std::sync::atomic::Ordering::SeqCst);
                Ok(())
            }) as BoxedFuture<'_, Result<()>>
        }))
        .await
        .unwrap_or_else(|e| panic!("with_tx commit failed: {e}"));
    assert!(committed.load(std::sync::atomic::Ordering::SeqCst));

    let rolled_back = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let flag = rolled_back.clone();
    let r = store
        .with_tx(Box::new(move |tx| {
            let flag = flag.clone();
            Box::pin(async move {
                flag.store(true, std::sync::atomic::Ordering::SeqCst);
                let _ = tx
                    .enqueue_tx("q", EnqueueOpts::default(), json!({}))
                    .await?;
                Err::<(), _>(Error::InvalidName {
                    name: "probe".into(),
                })
            }) as BoxedFuture<'_, Result<()>>
        }))
        .await;
    assert!(matches!(r, Err(Error::InvalidName { .. })));
    assert!(rolled_back.load(std::sync::atomic::Ordering::SeqCst));
}