//! The transactional seam's mem arm (ADR-007): `begin_tx` allocates an
//! **overlay** private to the caller's handle; the eleven `*_tx`
//! methods stage effects in it; commit merges the overlay atomically
//! and fires the overlay's pending wakes; drop discards it.
//!
//! - **Staging** (engine-mem.md §The tx overlay): every `*_tx` call
//!   validates its names at the entry point (before any effect),
//!   encodes the payload (`Codec` typed, ADR-023 §1), reads one clock
//!   instant, resolves/stamps identically to the no-handle paths
//!   (ADR-020 §3 — `enqueue_tx` stamps the plain derived defaults
//!   300/3/5/none, `outbox_enqueue_tx` the outbox's 60/5/5 set), and
//!   appends the effect to the handle-private overlay. Ids and offsets
//!   allocate at stage time from the engine-wide allocators (the
//!   pg-shaped story — a rolled-back tx's allocation gaps out; ids and
//!   offsets are monotone and never reused, gaps ordinary).
//! - **Reads inside the tx** merge the overlay over the committed
//!   state — read-your-own-writes with no round-trips:
//!   `get_job_tx`/`get_offset_tx`/`read_since_tx`/
//!   `read_from_consumer_tx` see the transaction's own staged
//!   enqueues/publishes/saves and the committed rows underneath; every
//!   other store view sees none of it pre-commit (an overlay is
//!   invisible to every other view until its commit — no isolation
//!   questions can arise across handles).
//! - **Commit-merge + wake-at-commit**: `commit` consumes the handle;
//!   the merge lands queues then streams (the canonical lock order;
//!   two sequential guarded sections on a non-awaiting path — each
//!   section's committed view is self-consistent at its boundary) and
//!   the overlay's pending wakes fire after the merged state is
//!   visible — a tx-staged
//!   notify delivers nothing until the merge, one wake per staged
//!   notify (per-notify, never coalesced). That discipline is what the
//!   pg engine had to fix retroactively (`pg-fix-tx-wake`); here it is
//!   structural, pinned from birth.
//! - **Drop = discard** (ADR-021 §4): dropping the handle without
//!   commit discards the overlay — no-ghosts trivially, because
//!   uncommitted state never touched committed state; the discard is
//!   panic-safe through an unwinding drop (the overlay is plain local
//!   data; dropping runs nothing).
//! - **Numeric domains** (ADR-023 §2) at the trait-impl entry: the
//!   stream reads' extents clamp empty (`limit <= 0`), the enqueue
//!   boundaries are total (a negative delay resolves to a ready time
//!   in the past — ready now), and durations never ride the tx seam.

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;

use serde_json::Value;

use alkstore::{
    BoxedFuture, EnqueueOpts, Error, Job, JobState, Result, StreamEvent, TxHandle,
    validate_local_name, validate_shared_name,
};

use crate::clock::Clock;
use crate::queue::{build_job_row, get_job_row, job_from_row};
use crate::resolution::{
    Stamps, outbox_backing_queue_name, outbox_default_stamps, plain_queue_default_stamps,
    resolve_enqueue_opts, stamps_with_override,
};
use crate::state::Guard;
use crate::state::{EventRow, JobRow, State};
use crate::stream::{
    get_offset_row, has_offset_row, insert_event, read_events, save_offset_row, stream_wake_channel,
};
use crate::wakes::WakeBus;

/// The handle-private overlay: every staged effect, in stage order.
#[derive(Debug, Default)]
struct Overlay {
    /// Staged enqueue rows (ids allocated at stage time).
    jobs: HashMap<i64, JobRow>,
    /// Staged event rows per stream (offsets allocated at stage time;
    /// stage order = offset order — the allocators are globally
    /// monotone).
    events: HashMap<String, Vec<EventRow>>,
    /// Staged notifies — channels in stage order, duplicates allowed
    /// (per-notify delivery, never coalesced).
    notifies: Vec<String>,
    /// Staged checkpoint saves (stream, consumer) → the staged max.
    saves: HashMap<(String, String), i64>,
}

/// One staged enqueue effect — the shared body of `enqueue_tx` and
/// `outbox_enqueue_tx` (the only difference: the queue name and the
/// stamp source).
fn stage_enqueued(
    overlay: &mut Overlay,
    state: &State,
    clock: &Clock,
    queue: &str,
    opts: &EnqueueOpts,
    payload: Value,
    stamps: Stamps,
) -> Result<i64> {
    let bytes = alkstore::encode_payload(&payload)?;
    let now = clock.unix_now();
    let resolved = resolve_enqueue_opts(now, opts);
    let id = state.allocate_job_id();
    let row = build_job_row(
        id,
        queue.to_string(),
        bytes,
        now,
        &resolved,
        opts.priority,
        &stamps,
    );
    overlay.jobs.insert(id, row);
    Ok(id)
}

/// The staged-job read: overlay first, then the committed live/dead
/// tables (dead rows visible — the auto-commit twin's shape, ADR-010
/// §1) — both queue-scoped against the argument.
fn merged_get_job(overlay: &Overlay, state: &State, queue: &str, job_id: i64) -> Option<Job> {
    if let Some(row) = overlay.jobs.get(&job_id) {
        return if row.queue == queue {
            Some(job_from_row(row.clone(), JobState::Pending, None, None))
        } else {
            None
        };
    }
    let q = state.queues.guard();
    let job = get_job_row(&q, job_id)?;
    (job.queue == queue).then_some(job)
}

/// The merged event read: the committed log and the staged rows under
/// one offset-ASC ordering (the global-FIFO shape holds across the
/// merge — stage-time offsets are globally monotone). The extent guard
/// (ADR-023 §2) lives here: `limit <= 0` yields the empty `Vec`.
fn merged_read_since(
    overlay: &Overlay,
    state: &State,
    stream: &str,
    offset: i64,
    limit: i64,
) -> Vec<EventRow> {
    if limit <= 0 {
        return Vec::new();
    }
    let committed = {
        let s = state.streams.guard();
        read_events(&s, stream, offset, limit)
    };
    let mut merged = committed;
    if let Some(staged) = overlay.events.get(stream) {
        merged.extend(staged.iter().filter(|row| row.offset > offset).cloned());
    }
    merged.sort_by_key(|row| row.offset);
    merged.truncate(limit as usize);
    merged
}

/// The merged checkpoint read (absent everywhere = 0).
fn merged_get_offset(overlay: &Overlay, state: &State, stream: &str, consumer: &str) -> i64 {
    let (committed, present) = {
        let s = state.streams.guard();
        (
            get_offset_row(&s, stream, consumer),
            has_offset_row(&s, stream, consumer),
        )
    };
    let staged = overlay
        .saves
        .get(&(stream.to_string(), consumer.to_string()));
    // The in-tx read shows what the commit would produce: a staged
    // save against an existing checkpoint rides the monotone guard
    // (the staged save only merges when it is ahead, so the read is
    // the max), while a staged save *seeding* an absent checkpoint is
    // stored raw — a negative seed reads raw in-tx too (ADR-023 §2's
    // boundary/offset annotation), not clamped against a phantom zero.
    match staged {
        Some(s) if present => (*s).max(committed),
        Some(s) => *s,
        None => committed,
    }
}

fn event_of(row: EventRow, stream: &str) -> StreamEvent {
    StreamEvent::from_row(
        row.offset,
        stream.to_string(),
        row.key,
        row.payload,
        row.created_at,
    )
}

/// The caller-held mem transaction handle (constructed only through
/// [`crate::MemStore`]'s `begin_tx`; consumers see
/// `Box<dyn TxHandle + Send>`). [`TxHandle::commit`] consumes it;
/// dropping it discards the overlay.
pub struct MemTxHandle {
    overlay: Overlay,
    state: Arc<State>,
    clock: Arc<Clock>,
    wakes: Arc<WakeBus>,
}

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

impl MemTxHandle {
    pub(crate) fn begin(state: Arc<State>, clock: Arc<Clock>, wakes: Arc<WakeBus>) -> Self {
        Self {
            overlay: Overlay::default(),
            state,
            clock,
            wakes,
        }
    }
}

impl TxHandle for MemTxHandle {
    fn enqueue_tx<'a>(
        &'a mut self,
        queue: &str,
        opts: EnqueueOpts,
        payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        if let Err(e) = validate_shared_name(queue) {
            return Box::pin(async move { Err(e) });
        }
        match stage_enqueued(
            &mut self.overlay,
            &self.state,
            &self.clock,
            queue,
            &opts,
            payload,
            stamps_with_override(plain_queue_default_stamps(), &opts),
        ) {
            Ok(id) => Box::pin(async move { Ok(id) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn publish_tx<'a>(&'a mut self, stream: &str, payload: Value) -> BoxedFuture<'a, Result<i64>> {
        let stream = stream.to_string();
        Box::pin(async move { self.publish_with_key_tx(&stream, None, payload).await })
    }

    fn publish_with_key_tx<'a>(
        &'a mut self,
        stream: &str,
        key: Option<String>,
        payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        if let Err(e) = validate_shared_name(stream) {
            return Box::pin(async move { Err(e) });
        }
        if let Some(k) = &key
            && (k.is_empty() || k.trim().is_empty())
        {
            let bad = k.clone();
            return Box::pin(async move { Err(Error::InvalidName { name: bad }) });
        }
        match self.stage_publish(stream, key, payload) {
            Ok(offset) => Box::pin(async move { Ok(offset) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn notify_tx<'a>(&'a mut self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>> {
        if let Err(e) = validate_shared_name(channel) {
            return Box::pin(async move { Err(e) });
        }
        match self.stage_notify(channel, payload) {
            Ok(()) => Box::pin(async { Ok(()) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn save_offset_tx<'a>(
        &'a mut self,
        stream: &str,
        consumer: &str,
        offset: i64,
    ) -> BoxedFuture<'a, Result<()>> {
        if let Err(e) = validate_shared_name(stream) {
            return Box::pin(async move { Err(e) });
        }
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        self.stage_save_offset(stream, consumer, offset);
        Box::pin(async { Ok(()) })
    }

    fn get_job_tx<'a>(
        &'a mut self,
        queue: &str,
        job_id: i64,
    ) -> BoxedFuture<'a, Result<Option<Job>>> {
        if let Err(e) = validate_shared_name(queue) {
            return Box::pin(async move { Err(e) });
        }
        let job = merged_get_job(&self.overlay, &self.state, queue, job_id);
        Box::pin(async move { Ok(job) })
    }

    fn get_offset_tx<'a>(
        &'a mut self,
        stream: &str,
        consumer: &str,
    ) -> BoxedFuture<'a, Result<i64>> {
        if let Err(e) = validate_shared_name(stream) {
            return Box::pin(async move { Err(e) });
        }
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        let offset = merged_get_offset(&self.overlay, &self.state, stream, consumer);
        Box::pin(async move { Ok(offset) })
    }

    fn read_since_tx<'a>(
        &'a mut self,
        stream: &str,
        offset: i64,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if let Err(e) = validate_shared_name(stream) {
            return Box::pin(async move { Err(e) });
        }
        let events = merged_read_since(&self.overlay, &self.state, stream, offset, limit);
        let stream = stream.to_string();
        Box::pin(async move { Ok(events.into_iter().map(|r| event_of(r, &stream)).collect()) })
    }

    fn read_from_consumer_tx<'a>(
        &'a mut self,
        stream: &str,
        consumer: &str,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if let Err(e) = validate_shared_name(stream) {
            return Box::pin(async move { Err(e) });
        }
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        let events = {
            let from = merged_get_offset(&self.overlay, &self.state, stream, consumer);
            merged_read_since(&self.overlay, &self.state, stream, from, limit)
        };
        let stream = stream.to_string();
        Box::pin(async move { Ok(events.into_iter().map(|r| event_of(r, &stream)).collect()) })
    }

    fn outbox_enqueue_tx<'a>(
        &'a mut self,
        outbox: &str,
        opts: EnqueueOpts,
        payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        if let Err(e) = validate_shared_name(outbox) {
            return Box::pin(async move { Err(e) });
        }
        let backing = outbox_backing_queue_name(outbox);
        match stage_enqueued(
            &mut self.overlay,
            &self.state,
            &self.clock,
            &backing,
            &opts,
            payload,
            stamps_with_override(outbox_default_stamps(), &opts),
        ) {
            Ok(id) => Box::pin(async move { Ok(id) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn commit(self: Box<Self>) -> BoxedFuture<'static, Result<()>> {
        // Consume the handle: the overlay moves out (the one-shot
        // discipline); a dropped-before-commit handle discards instead
        // — nothing runs on that path, panic-safe through the unwind.
        let MemTxHandle {
            overlay,
            state,
            clock: _,
            wakes,
        } = *self;
        Box::pin(async move {
            merge_and_wake(state, wakes, overlay);
            Ok(())
        })
    }
}

impl MemTxHandle {
    fn stage_publish(&mut self, stream: &str, key: Option<String>, payload: Value) -> Result<i64> {
        let bytes = alkstore::encode_payload(&payload)?;
        let now = self.clock.unix_now();
        let offset = self.state.allocate_event_offset();
        let row = EventRow {
            offset,
            key,
            payload: bytes,
            created_at: now,
        };
        self.overlay
            .events
            .entry(stream.to_string())
            .or_default()
            .push(row);
        Ok(offset)
    }

    fn stage_notify(&mut self, channel: &str, payload: Value) -> Result<()> {
        // The encoding run is the codec-typed entry check (the
        // measured serialization, ADR-020 §4); the wake contract's
        // delivery is the channel alone — the bytes are nowhere stored.
        let encoded = alkstore::encode_payload(&payload)?;
        let _ = encoded;
        self.overlay.notifies.push(channel.to_string());
        Ok(())
    }

    fn stage_save_offset(&mut self, stream: &str, consumer: &str, offset: i64) {
        let entry = self
            .overlay
            .saves
            .entry((stream.to_string(), consumer.to_string()))
            .or_insert(i64::MIN);
        if offset > *entry {
            *entry = offset;
        }
    }
}

/// The commit-merge: move every staged effect into the committed state
/// (queues then streams, the canonical lock order; two sequential
/// guarded sections on a non-awaiting path) then fire the overlay's
/// pending wakes — the
/// wake-at-commit discipline: nothing delivers until the merge lands.
fn merge_and_wake(state: Arc<State>, wakes: Arc<WakeBus>, mut overlay: Overlay) {
    {
        let mut q = state.queues.guard();
        for (_, row) in std::mem::take(&mut overlay.jobs) {
            q.live.insert(row.id, row);
        }
    }
    {
        let mut s = state.streams.guard();
        for (stream, rows) in overlay.events.iter() {
            for row in rows {
                insert_event(&mut s, stream, row.clone());
            }
        }
        for ((stream, consumer), offset) in overlay.saves.iter() {
            save_offset_row(&mut s, stream, consumer, *offset);
        }
    }
    // Wake-at-commit: staged notifies in stage order (duplicates ride —
    // per-notify), plus one wake per staged stream on its engine-internal
    // feed (receivers drain by cursor, so one wake per stream is enough —
    // the feed is plumbing, not contract surface; never coalesced across
    // commits).
    let mut fired: VecDeque<String> = VecDeque::with_capacity(overlay.notifies.len());
    fired.extend(overlay.notifies.iter().cloned());
    for stream in overlay.events.keys() {
        fired.push_back(stream_wake_channel(stream));
    }
    wakes.fire_commit_wakes(fired);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::MemStore;
    use alkstore::{Delivery, Outbox, Store, TxHandle};

    fn store() -> MemStore {
        let store = MemStore::new();
        store.clock.freeze(1_000_000);
        store
    }

    async fn listener(store: &MemStore, channel: &str) -> Box<dyn alkstore::WakeReceiver> {
        store.listen(channel).await.unwrap()
    }

    async fn listener_quiet(rx: &mut dyn alkstore::WakeReceiver) {
        let outcome = tokio::time::timeout(std::time::Duration::from_millis(60), rx.recv()).await;
        assert!(
            matches!(outcome, Err(_) | Ok(None)),
            "no wake may arrive in the quiet window"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn tx_staged_notify_delivers_nothing_pre_commit_one_wake_post() {
        let store = store();
        let mut rx = listener(&store, "orders").await;
        let mut tx = store.begin_tx().await.unwrap();
        tx.notify_tx("orders", serde_json::json!({"n": 1}))
            .await
            .unwrap();
        listener_quiet(rx.as_mut()).await;
        assert!(tx.commit().await.is_ok());
        let wake = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
            .await
            .expect("the wake delivers")
            .expect("a wake");
        assert_eq!(wake.channel, "orders");
        // Per-notify: a second staged notify delivers its own wake.
        let mut tx = store.begin_tx().await.unwrap();
        tx.notify_tx("orders", serde_json::json!(2)).await.unwrap();
        tx.commit().await.unwrap();
        assert!(rx.recv().await.is_some());
        assert!(
            rx.try_recv().unwrap().is_none(),
            "never coalesced — exactly one wake per staged notify"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn enqueue_tx_stamps_the_plain_defaults_and_commits_atomically() {
        let store = store();
        let mut tx = store.begin_tx().await.unwrap();
        let id = tx
            .enqueue_tx(
                "emails",
                EnqueueOpts {
                    delay: Some(30),
                    max_attempts: Some(7),
                    ..Default::default()
                },
                serde_json::json!({"to": "a@b.c"}),
            )
            .await
            .unwrap();
        let staged = tx.get_job_tx("emails", id).await.unwrap().unwrap();
        assert_eq!(
            staged.visibility_timeout_s, 300,
            "the plain defaults stamp (no handle carries opts)"
        );
        assert_eq!(staged.max_attempts, 7);
        assert_eq!(staged.backoff_base_s, 5);
        assert_eq!(staged.dead_letter_retention_s, None);
        assert_eq!(staged.run_at, 1_000_030);
        assert_eq!(staged.state, JobState::Pending);
        let committed_view = store
            .queue("emails", alkstore::QueueOpts::default())
            .await
            .unwrap();
        assert!(
            committed_view.get_job(id).await.unwrap().is_none(),
            "other views see none pre-commit"
        );
        tx.commit().await.unwrap();
        let job = committed_view.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.queue, "emails");
        assert_eq!(
            job.payload,
            serde_json::to_vec(&serde_json::json!({"to": "a@b.c"})).unwrap()
        );
        assert_eq!(job.created_at, 1_000_000);
        assert_eq!(
            job.run_at, 1_000_030,
            "the resolved values, not the raw fields"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn publish_tx_reads_and_merge_ordering_plus_wake_at_commit() {
        let store = store();
        let mut feed = store.wakes().attach(&stream_wake_channel("orders"));
        let mut tx = store.begin_tx().await.unwrap();
        let a = tx.publish_tx("orders", serde_json::json!(1)).await.unwrap();
        let b = tx
            .publish_with_key_tx("orders", Some("cust-7".into()), serde_json::json!(2))
            .await
            .unwrap();
        assert!(a < b);
        let seen = tx.read_since_tx("orders", 0, 10).await.unwrap();
        assert_eq!(
            seen.len(),
            2,
            "the tx's own staged publishes are visible inside the tx"
        );
        assert_eq!(seen[0].key, None);
        assert_eq!(seen[1].key.as_deref(), Some("cust-7"));
        assert_eq!(
            seen.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![a, b]
        );
        let handle = store.stream("orders").await.unwrap();
        assert!(
            handle.read_since(0, 10).await.unwrap().is_empty(),
            "other views see none pre-commit"
        );
        assert!(
            feed.rx().try_recv().is_err(),
            "no wake fires before the merge"
        );
        tx.commit().await.unwrap();
        let committed = handle.read_since(0, 10).await.unwrap();
        assert_eq!(
            committed.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![a, b]
        );
        assert_eq!(committed[1].payload_as::<i64>().unwrap(), 2);
        assert!(
            feed.rx().try_recv().is_ok(),
            "one wake fired at the commit (after the merge)"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn drop_without_commit_discards_the_whole_overlay_no_ghosts() {
        let store = store();
        let mut rx = listener(&store, "ch").await;
        let handle = store.stream("s").await.unwrap();
        let committed_anchor = handle.publish(serde_json::json!("anchor")).await.unwrap();
        {
            let mut tx = store.begin_tx().await.unwrap();
            tx.enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(1))
                .await
                .unwrap();
            tx.publish_tx("s", serde_json::json!(2)).await.unwrap();
            tx.notify_tx("ch", serde_json::json!(3)).await.unwrap();
            tx.save_offset_tx("s", "c1", 999).await.unwrap();
            drop(tx);
        }
        let q = store
            .queue("q", alkstore::QueueOpts::default())
            .await
            .unwrap();
        assert!(q.get_job(1).await.unwrap().is_none(), "no ghost job");
        let events = handle.read_since(0, 10).await.unwrap();
        assert_eq!(
            events.len(),
            1,
            "no ghost event — only the committed anchor"
        );
        assert_eq!(events[0].offset, committed_anchor);
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            0,
            "no ghost offset save"
        );
        listener_quiet(rx.as_mut()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn other_views_see_none_pre_commit_and_rollback_leaves_no_read_subjects() {
        let store = store();
        let handle = store.stream("s").await.unwrap();
        let mut tx = store.begin_tx().await.unwrap();
        let offset = tx.publish_tx("s", serde_json::json!(1)).await.unwrap();
        tx.save_offset_tx("s", "c1", offset).await.unwrap();
        assert_eq!(
            tx.get_offset_tx("s", "c1").await.unwrap(),
            offset,
            "the tx's own save is visible inside"
        );
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            0,
            "other views see none pre-commit"
        );
        drop(tx);
        assert!(
            handle.read_since(0, 10).await.unwrap().is_empty(),
            "post-rollback the read's subject is gone"
        );
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            0,
            "no-ghosts for reads"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn save_offset_tx_is_monotone_and_only_the_staged_max_merges() {
        let store = store();
        let mut tx = store.begin_tx().await.unwrap();
        tx.save_offset_tx("s", "c", 5).await.unwrap();
        tx.save_offset_tx("s", "c", 3).await.unwrap();
        assert_eq!(
            tx.get_offset_tx("s", "c").await.unwrap(),
            5,
            "monotone across staged saves"
        );
        tx.commit().await.unwrap();
        let handle = store.stream("s").await.unwrap();
        assert_eq!(handle.get_offset("c").await.unwrap(), 5);
        // A staged save below the committed checkpoint silently no-ops.
        let mut tx = store.begin_tx().await.unwrap();
        tx.save_offset_tx("s", "c", 2).await.unwrap();
        assert_eq!(tx.get_offset_tx("s", "c").await.unwrap(), 5);
        tx.commit().await.unwrap();
        assert_eq!(handle.get_offset("c").await.unwrap(), 5);
    }

    #[cold]
    #[inline(never)]
    fn mid_flight_panic() -> alkstore::Error {
        panic!("mid-flight, after staging all four write kinds")
    }

    #[tokio::test(flavor = "current_thread")]
    async fn with_tx_ok_commits_err_rolls_back_and_panic_rolls_back_through_the_unwind() {
        let store = std::sync::Arc::new(store());
        // Ok ⇒ commit.
        let committed_id = {
            let s = store.clone();
            let captured = std::sync::Arc::new(std::sync::Mutex::new(0i64));
            let sink = captured.clone();
            s.with_tx(Box::new(move |tx| {
                Box::pin(async move {
                    let id = tx
                        .enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(1))
                        .await?;
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) = id;
                    Ok(())
                })
            }))
            .await
            .unwrap();
            *std::sync::Arc::try_unwrap(captured)
                .unwrap_or_else(|_| panic!("the sink outlives the run"))
                .lock()
                .unwrap_or_else(|p| p.into_inner())
        };
        assert!(committed_id > 0);
        // Err ⇒ rollback.
        let rolled = {
            let s = store.clone();
            s.with_tx(Box::new(|tx| {
                Box::pin(async move {
                    tx.enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(2))
                        .await?;
                    Err::<(), alkstore::Error>(alkstore::Error::database(std::io::Error::other(
                        "business write failed",
                    )))
                })
            }))
            .await
        };
        assert!(rolled.is_err());
        // Panic across the closure ⇒ rollback through the unwind's drop
        // (the overlay discard runs nothing — panic-safe by construction).
        let probe_store = store.clone();
        let spawned = tokio::spawn(async move {
            probe_store
                .with_tx(Box::new(|tx| {
                    Box::pin(async move {
                        tx.enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(3))
                            .await?;
                        tx.publish_tx("s", serde_json::json!(4)).await?;
                        tx.notify_tx("ch", serde_json::json!(5)).await?;
                        tx.save_offset_tx("s", "c", 42).await?;
                        Err::<(), alkstore::Error>(mid_flight_panic())
                    })
                }))
                .await
        });
        let joined = spawned.await;
        assert!(joined.is_err(), "the panic surfaced");
        assert!(
            joined.unwrap_err().is_panic(),
            "the JoinError carries the panic"
        );
        // No residue of any kind (the uniform no-ghosts list).
        let q = store
            .queue("q", alkstore::QueueOpts::default())
            .await
            .unwrap();
        assert!(
            q.get_job(1).await.unwrap().is_some(),
            "the Ok leg committed"
        );
        for id in [2, 3] {
            assert!(
                q.get_job(id).await.unwrap().is_none(),
                "no ghost job (err/panic)"
            );
        }
        let stream = store.stream("s").await.unwrap();
        assert!(
            stream.read_since(0, 10).await.unwrap().is_empty(),
            "no ghost event"
        );
        assert_eq!(
            stream.get_offset("c").await.unwrap(),
            0,
            "no ghost offset save"
        );
        let mut rx = listener(&store, "ch").await;
        assert!(rx.try_recv().unwrap().is_none(), "no ghost notify wake");
        // The store remains usable: a fresh tx commits through the same
        // seam (the next id gaps past the discarded allocations').
        let usable = {
            let s = store.clone();
            let captured = std::sync::Arc::new(std::sync::Mutex::new(0i64));
            let sink = captured.clone();
            s.with_tx(Box::new(move |tx| {
                Box::pin(async move {
                    let id = tx
                        .enqueue_tx("q", EnqueueOpts::default(), serde_json::json!(6))
                        .await?;
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) = id;
                    Ok(())
                })
            }))
            .await
            .unwrap();
            *std::sync::Arc::try_unwrap(captured)
                .unwrap_or_else(|_| panic!("the sink outlives the run"))
                .lock()
                .unwrap_or_else(|p| p.into_inner())
        };
        let next = store
            .queue("q", alkstore::QueueOpts::default())
            .await
            .unwrap();
        assert!(next.get_job(usable).await.unwrap().is_some());
        assert!(usable > 3, "ids allocate forward — gaps are legal");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn get_job_tx_is_queue_scoped_and_sees_dead_rows() {
        let store = store();
        let mut tx = store.begin_tx().await.unwrap();
        let id = tx
            .enqueue_tx("emails", EnqueueOpts::default(), serde_json::json!(1))
            .await
            .unwrap();
        tx.commit().await.unwrap();
        let q = store
            .queue("emails", alkstore::QueueOpts::default())
            .await
            .unwrap();
        let claimed = q.claim_batch("w", 1).await.unwrap().pop().unwrap();
        claimed.fail(Some("boom".into())).await.unwrap();
        let mut tx = store.begin_tx().await.unwrap();
        let dead = tx.get_job_tx("emails", id).await.unwrap().unwrap();
        assert_eq!(
            dead.state,
            JobState::Dead,
            "dead rows are get_job_tx-visible"
        );
        assert_eq!(dead.last_error.as_deref(), Some("boom"));
        assert!(
            tx.get_job_tx("other", id).await.unwrap().is_none(),
            "queue-scoped: another queue reads None"
        );
    }

    struct NoDelivery;
    impl Delivery for NoDelivery {
        fn deliver<'a>(
            &'a mut self,
            _job: Box<dyn alkstore::JobHandle>,
        ) -> BoxedFuture<'a, Result<()>> {
            Box::pin(async { Ok(()) })
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn outbox_enqueue_tx_stamps_the_60_5_5_set() {
        let store = store();
        // The concrete handle (crate-internal): the staging inspection
        // reads the overlay before the commit-merge.
        let mut tx = MemTxHandle::begin(
            store.state.clone(),
            store.clock.clone(),
            store.wakes().clone(),
        );
        let id = tx
            .outbox_enqueue_tx(
                "mail",
                EnqueueOpts {
                    delay: Some(5),
                    max_attempts: Some(2),
                    ..Default::default()
                },
                serde_json::json!(1),
            )
            .await
            .unwrap();
        let staged = tx.overlay.jobs.get(&id).unwrap().clone();
        assert_eq!(staged.queue, outbox_backing_queue_name("mail"));
        assert_eq!(staged.visibility_timeout_s, 60);
        assert_eq!(staged.max_attempts, 2, "the per-job override rides");
        assert_eq!(staged.backoff_base_s, 5);
        assert_eq!(staged.dead_letter_retention_s, None);
        assert_eq!(staged.run_at, 1_000_005);
        let err = tx
            .get_job_tx(&outbox_backing_queue_name("mail"), id)
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::ReservedName { .. }),
            "the derived name is unreachable by get_job_tx"
        );
        Box::new(tx).commit().await.unwrap();
        // Commit makes it claimable by run_once exactly when the
        // business write commits.
        store.clock.advance(10);
        let mut ob = crate::outbox::MemOutbox::new(
            "mail".to_string(),
            store.state.clone(),
            store.clock.clone(),
        );
        assert!(
            ob.run_once("w", &mut NoDelivery).await.unwrap(),
            "the committed outbox job is claimable"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn entry_validation_on_every_tx_surface() {
        let store = store();
        let mut tx = MemTxHandle::begin(
            store.state.clone(),
            store.clock.clone(),
            store.wakes().clone(),
        );
        for bad in ["", "   ", "__alkstore_q"] {
            assert!(
                tx.enqueue_tx(bad, EnqueueOpts::default(), serde_json::json!(1))
                    .await
                    .is_err()
            );
            assert!(tx.publish_tx(bad, serde_json::json!(1)).await.is_err());
            assert!(
                tx.publish_with_key_tx(bad, None, serde_json::json!(1))
                    .await
                    .is_err()
            );
            assert!(tx.notify_tx(bad, serde_json::json!(1)).await.is_err());
            assert!(tx.save_offset_tx(bad, "c", 1).await.is_err());
            assert!(tx.get_job_tx(bad, 1).await.is_err());
            assert!(tx.get_offset_tx(bad, "c").await.is_err());
            assert!(tx.read_since_tx(bad, 0, 1).await.is_err());
            assert!(tx.read_from_consumer_tx(bad, "c", 1).await.is_err());
            assert!(
                tx.outbox_enqueue_tx(bad, EnqueueOpts::default(), serde_json::json!(1))
                    .await
                    .is_err()
            );
        }
        // Consumer-local identifiers take the non-empty rule only.
        let err = tx.save_offset_tx("s", "", 1).await.unwrap_err();
        assert!(matches!(err, Error::InvalidName { .. }));
        let err = tx.read_from_consumer_tx("s", "", 1).await.unwrap_err();
        assert!(matches!(err, Error::InvalidName { .. }));
        assert!(
            tx.overlay.jobs.is_empty(),
            "validation errors stage nothing"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn numeric_domain_guards_on_the_tx_reads() {
        let store = store();
        let mut tx = store.begin_tx().await.unwrap();
        assert!(tx.read_since_tx("s", 0, 0).await.unwrap().is_empty());
        assert!(tx.read_since_tx("s", 0, -5).await.unwrap().is_empty());
        assert!(
            tx.read_from_consumer_tx("s", "c", -1)
                .await
                .unwrap()
                .is_empty()
        );
        tx.commit().await.unwrap();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn concurrent_txs_merge_serially_without_isolation_questions() {
        let store = store();
        let mut t1 = store.begin_tx().await.unwrap();
        let mut t2 = store.begin_tx().await.unwrap();
        let a = t1.publish_tx("s", serde_json::json!(1)).await.unwrap();
        let b = t2.publish_tx("s", serde_json::json!(2)).await.unwrap();
        t1.commit().await.unwrap();
        t2.commit().await.unwrap();
        let handle = store.stream("s").await.unwrap();
        let events = handle.read_since(0, 10).await.unwrap();
        assert_eq!(
            events.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![a, b],
            "both commits merged; reads yield offset ASC — no isolation questions across handles"
        );
        assert!(
            store.wakes().commit_watermark() >= 2,
            "each commit fired one boundary"
        );
    }
}