obix 0.12.3

//! Handler-controlled transaction scoping for outbox event-handler jobs.
//!
//! Every persistent event delivered to an
//! [`SingletonSubscriber`](super::SingletonSubscriber) comes with an
//! [`EventCtx`] that the handler must resolve into a [`Handled`] token by
//! choosing exactly one of three entry verbs — a monotone cost ladder:
//!
//! | verb | meaning | cost |
//! |------|---------|------|
//! | [`EventCtx::skip`] | not my event | zero — no transaction is opened |
//! | [`EventCtx::collect_with`] (and the [`collect`](EventCtx::collect) sugar) | contribute an item to the pending batch's accumulator | zero at collect time — one [`flush`](super::SingletonSubscriber::flush) call per batch applies all items |
//! | [`EventCtx::consume`] | land the pending batch first, then do my work in a fresh op | my event is its own atomic unit, fenced from history |
//!
//! and — when an op was taken — the single exit verb
//! [`IsolatedOp::commit`], which lands the op (work + checkpoint,
//! atomically) when the invocation returns.
//!
//! A pending batch is **only ever collected items**: no transaction is left
//! open across event invocations. It exists only while there is ready
//! persistent backlog — the runner never awaits the stream while items are
//! pending, so a pending stream is itself a flush trigger. Batching
//! therefore rides bursts that already happened and adds no latency at low
//! traffic. Ephemeral events travel on their own stream
//! and are handled between batches — they never interrupt a batch, a
//! transaction never spans the foreign `handle_ephemeral` await, and between
//! batches the two streams race fairly so neither can starve the other.
//! Handlers that only consume one stream should declare it via
//! [`SUBSCRIPTION`](super::SingletonSubscriber::SUBSCRIPTION) — the other
//! stream is then never subscribed at all.
//!
//! Every flush — whichever of the triggers fires — first hands all collected
//! items to the handler's [`flush`](super::SingletonSubscriber::flush) inside
//! a transaction opened for the landing, then persists the checkpoint at the
//! last *fully handled* sequence (skips included), then commits: items, work
//! and pointer are inseparable. A failed flush rolls everything back and
//! replays the whole batch (items are re-collected), so collected work must
//! tolerate wholesale replay.
//!
//! # The two ctx types, and why they differ
//!
//! [`EventCtx`] and [`KeyedEventCtx`] are facades over the same internals,
//! not a fork — but they deliberately expose different verbs:
//!
//! | capability | singleton | keyed |
//! |------------|-----------|-------|
//! | ephemeral delivery | yes — by presence | no, statically |
//! | [`pause_until`](KeyedEventCtx::pause_until), staged chains | no — by presence | yes |
//! | dormancy / wake | no | yes |
//!
//! The asymmetry is the **presence contract**, not scheduling mechanics. A
//! singleton subscriber is always on, and that presence is precisely what
//! licenses its ephemeral subscription: ephemeral events cannot be replayed,
//! so only an always-present consumer may hear them. A verb that pauses
//! consumption contradicts the property that defines the mode — which is why
//! the pause and staged verbs are keyed-only, and why nothing is gained by
//! adding a pause-less staged variant to the singleton.
//!
//! A single-instance flow that genuinely needs to pause or stage is
//! persistent-only by definition. Host it as a keyed subscriber with one
//! static key — a legitimate and intended shape (foreign-system relays with
//! backpressure, single-instance exporters), which also brings dormancy for
//! free.

use serde::{Deserialize, Serialize};

use std::marker::PhantomData;

use job::CurrentJob;

use crate::out::lane::{InsertOrder, Lane};
use crate::out::subscription::StreamPosition;
use crate::sequence::{CommitSequence, EventSequence};

/// Error type shared with the handler trait methods.
pub(crate) type HandlerError = Box<dyn std::error::Error + Send + Sync>;

/// Persisted execution state of an outbox event-handler job: the sequence of
/// the last fully handled persistent event, plus (keyed subscribers only)
/// where the member last paused.
#[derive(Default, Clone, Serialize, Deserialize)]
pub(crate) struct OutboxEventJobState {
    pub(crate) sequence: EventSequence,
    /// The cursor on the commit-ordered lane. Its presence is what marks a
    /// subscription as checkpointed under `Ordering::Commit`; on the insert
    /// lane it stays `None` and `sequence` is the cursor.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) commit_sequence: Option<CommitSequence>,
    /// Where and until when the member last paused. Read at run start: a
    /// wake that finds nothing persisted beyond the paused event is answered
    /// by pausing again, without the subscriber being invoked. Stale by
    /// construction once the cursor is past `sequence`.
    ///
    /// `serde(default)` so execution-state rows written before pausing
    /// existed still decode; `skip_serializing_if` so a singleton
    /// subscriber's state stays byte-identical to what it always wrote.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) paused: Option<PausedState>,
}

/// The pause slot: the event the cursor is parked before, and the instant
/// the member asked to be woken at.
#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct PausedState {
    pub(crate) sequence: EventSequence,
    pub(crate) until: chrono::DateTime<chrono::Utc>,
}

/// Book-keeping the runner shares with [`EventCtx`].
pub(crate) struct BatchTracker {
    /// Number of events that collected items since the last flush. Nonzero
    /// means the accumulator is dirty: a batch is pending, and no checkpoint
    /// may be persisted before the items are flushed. Also what
    /// `max_batch_size` bounds — with no deferred ops, the pending batch is
    /// exactly its collected events.
    pub(crate) collected: usize,
    /// Highest position whose checkpoint has been persisted, on the
    /// subscription's own lane. Dynamic because the ctx types are lane-free.
    pub(crate) persisted: StreamPosition,
    /// When the checkpoint was last persisted (any flush or standalone write).
    pub(crate) last_persist: tokio::time::Instant,
}

/// Copies a keyed member's durable cursor into its `subscriptions` row,
/// inside the same transaction that persists the job checkpoint it mirrors.
///
/// Indirected through a trait object rather than a `Tables` type parameter
/// because everything in this module is shared with the singleton path,
/// which has no subscription row to mirror into and passes `None`. It also
/// keeps the mirror atomic with the checkpoint by construction: both writes
/// land on the same op, so the copy can never claim progress the job did not
/// commit.
pub(crate) trait CheckpointMirror: Send + Sync {
    fn mirror<'a>(
        &'a self,
        op: &'a mut es_entity::DbOp<'static>,
        checkpoint: EventSequence,
    ) -> futures::future::BoxFuture<'a, Result<(), sqlx::Error>>;
}

pub(crate) struct CtxParts<'inv> {
    pub(crate) op_slot: &'inv mut Option<es_entity::DbOp<'static>>,
    pub(crate) current_job: &'inv mut CurrentJob,
    /// Mutable so the pause verbs can record where the member paused in the
    /// state the runner is about to persist. `sequence` remains the runner's
    /// alone.
    pub(crate) state: &'inv mut OutboxEventJobState,
    pub(crate) tracker: &'inv mut BatchTracker,
    /// `Some` for keyed members, `None` for singletons — see
    /// [`CheckpointMirror`].
    pub(crate) mirror: Option<&'inv dyn CheckpointMirror>,
}

/// Proof that a persistent event was resolved in one of the legal ways.
///
/// Only obtainable from [`EventCtx::skip`], [`EventCtx::collect_with`] (or
/// its [`collect`](EventCtx::collect) sugar) or [`IsolatedOp::commit`] — the
/// type system forces every
/// [`handle_persistent`](super::SingletonSubscriber::handle_persistent)
/// invocation to decide the transactional fate of its event.
///
/// The token is branded with the invocation's lifetime, so it cannot leave
/// the invocation that minted it: handlers are `'static`, so stashing a
/// token for a later invocation does not compile —
///
/// ```compile_fail
/// use obix::{EventCtx, Handled};
///
/// struct Evil {
///     stash: std::sync::Mutex<Option<Handled<'static>>>,
/// }
///
/// fn stash_it(ctx: EventCtx<'_>, evil: &Evil) {
///     // error[E0521]: borrowed data escapes outside of function
///     *evil.stash.lock().unwrap() = Some(ctx.skip());
/// }
/// ```
///
/// Combined with the entry verbs consuming the [`EventCtx`] (each invocation
/// can mint exactly one token), the returned token is always *the* token of
/// the current invocation, of the kind that actually happened.
#[must_use = "return the Handled token from handle_persistent"]
pub struct Handled<'inv> {
    pub(crate) outcome: Outcome,
    pub(crate) _invocation: PhantomData<&'inv ()>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Outcome {
    Skip,
    Collect,
    Commit,
    /// Keyed-only: my cursor stays *before* this event until the given time.
    /// Only [`KeyedEventCtx::pause_until`] mints this — the singleton runner
    /// never constructs a ctx that can, so this variant is unreachable from
    /// [`EventCtx`].
    Pause(chrono::DateTime<chrono::Utc>),
    /// Keyed-only: commit the staged op left in `op_slot`, but the cursor
    /// still stays *before* this event until the given time. Only
    /// [`StagedOp::pause_until`] mints this.
    CommitAndPause(chrono::DateTime<chrono::Utc>),
}

/// Per-event decision point handed to
/// [`handle_persistent`](super::SingletonSubscriber::handle_persistent).
///
/// Generic over the handler's
/// [`Batch`](super::SingletonSubscriber::Batch) accumulator `B` (defaulting
/// to `()` for handlers that never collect). See the [module docs](self)
/// for the semantics of the three entry verbs.
///
/// # Presence contract
///
/// A singleton subscriber is **always on**, and that presence is what
/// licenses its ephemeral subscription: ephemeral events cannot be replayed,
/// so only an always-present consumer may hear them. This ctx therefore has
/// no pausing verb — no `pause_until`, no staged chain.
/// Their absence is semantic, not an omission: a verb that suspends
/// consumption would contradict the property that defines the mode.
///
/// ```compile_fail
/// use obix::{EventCtx, Handled};
///
/// fn pause_it<'inv>(ctx: EventCtx<'inv>) -> Handled<'inv> {
///     // error[E0599]: no method named `pause_until` found for struct `EventCtx`
///     //
///     // A flow that pauses or stages is persistent-only by definition —
///     // host it as a keyed subscriber; a single static key is legitimate.
///     ctx.pause_until()
/// }
/// ```
///
/// The control for the negative test above — same signature, with a verb a
/// singleton ctx does have. If this stops compiling, the `compile_fail`
/// above has started passing for the wrong reason:
///
/// ```
/// use obix::{EventCtx, Handled};
///
/// fn resolve_it<'inv>(ctx: EventCtx<'inv>) -> Handled<'inv> {
///     ctx.skip()
/// }
/// ```
#[must_use = "resolve the EventCtx via skip / collect / consume"]
pub struct EventCtx<'inv, B = ()> {
    pub(crate) parts: CtxParts<'inv>,
    pub(crate) batch: &'inv mut B,
    pub(crate) flusher: &'inv dyn ItemFlush<B>,
}

impl<'inv, B> EventCtx<'inv, B> {
    /// This event is not for me — no transaction is opened, an open batch op
    /// is left untouched, and the checkpoint advances lazily (piggybacked on
    /// the next flush, or persisted on the configured checkpoint interval).
    pub fn skip(self) -> Handled<'inv> {
        Handled {
            outcome: Outcome::Skip,
            _invocation: PhantomData,
        }
    }

    /// Contribute to the pending batch's accumulator — a pure memory write:
    /// no transaction is opened and no statement is executed now. The runner
    /// hands the accumulated batch to the handler's
    /// [`flush`](super::SingletonSubscriber::flush) exactly once per batch
    /// landing, inside the transaction that commits the checkpoint.
    ///
    /// Collected work shares fate with its neighbors and must tolerate
    /// whole-batch replay: on a failed flush the events replay and their
    /// items are re-collected.
    ///
    /// For `Vec` and `HashMap` accumulators the [`collect`](Self::collect)
    /// sugar is usually more convenient.
    pub fn collect_with(self, f: impl FnOnce(&mut B)) -> Handled<'inv> {
        f(self.batch);
        self.parts.tracker.collected += 1;
        Handled {
            outcome: Outcome::Collect,
            _invocation: PhantomData,
        }
    }

    /// Land the pending batch first (its collected items and its checkpoint,
    /// at the last fully handled sequence), then hand back a fresh op: this
    /// event is its own atomic unit, sharing no fate with history — and none
    /// with the future either, since [`IsolatedOp`] only offers
    /// [`commit`](IsolatedOp::commit).
    ///
    /// This is the failure-isolation fence: if this event's work fails, only
    /// this event replays.
    pub async fn consume(self) -> Result<IsolatedOp<'inv>, HandlerError>
    where
        B: Default,
    {
        let EventCtx {
            mut parts,
            batch,
            flusher,
        } = self;
        flush_batch(&mut parts, batch, flusher, "consume_entry").await?;
        *parts.op_slot = Some(
            es_entity::DbOp::init_with_clock(parts.current_job.pool(), parts.current_job.clock())
                .await?,
        );
        let op = parts.op_slot.as_mut().expect("just materialized above");
        Ok(IsolatedOp { op })
    }
}

impl<'inv, T> EventCtx<'inv, Vec<T>> {
    /// [`collect_with`](Self::collect_with) sugar for `Vec` accumulators:
    /// append one item to the pending batch.
    pub fn collect(self, item: T) -> Handled<'inv> {
        self.collect_with(|batch| batch.push(item))
    }
}

impl<'inv, K, V, S> EventCtx<'inv, std::collections::HashMap<K, V, S>>
where
    K: std::hash::Hash + Eq,
    S: std::hash::BuildHasher,
{
    /// [`collect_with`](Self::collect_with) sugar for `HashMap` accumulators:
    /// keyed last-write-wins insert. Persistent events arrive in ascending
    /// sequence, so within a batch this naturally keeps the newest item per
    /// key — the coalescing fold (N updates per key → 1 flushed entry).
    pub fn collect(self, key: K, value: V) -> Handled<'inv> {
        self.collect_with(|batch| {
            batch.insert(key, value);
        })
    }
}

/// An op holding exactly this event's work, fenced from the batch history.
/// Implements [`AtomicOperation`](es_entity::AtomicOperation). The only exit
/// is [`commit`](Self::commit) — isolation from future events is guaranteed
/// by construction, and there is no mutable access to the raw
/// [`es_entity::DbOp`] (only a shared [`Deref`](std::ops::Deref) view):
/// committing, rolling back, or swapping out the underlying op is
/// unrepresentable, so work and checkpoint can only land together, through
/// the runner.
#[must_use = "exit with .commit() to produce the Handled token"]
pub struct IsolatedOp<'inv> {
    op: &'inv mut es_entity::DbOp<'static>,
}

impl<'inv> IsolatedOp<'inv> {
    /// Land my work and my checkpoint, atomically, when the invocation
    /// returns.
    pub fn commit(self) -> Handled<'inv> {
        Handled {
            outcome: Outcome::Commit,
            _invocation: PhantomData,
        }
    }
}

impl std::ops::Deref for IsolatedOp<'_> {
    type Target = es_entity::DbOp<'static>;

    fn deref(&self) -> &Self::Target {
        self.op
    }
}

es_entity::delegate_atomic_operation!(IsolatedOp<'_>, { s => s.op });

pub(crate) type BoxFuture<'a, T> =
    std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;

/// Object-safe bridge from the runner (and [`EventCtx::consume`]'s
/// entry fence) to the handler's typed
/// [`flush`](super::SingletonSubscriber::flush) — erases the handler type so
/// [`EventCtx`] only needs to know the accumulator `B`. The whole
/// [`OutboxEventJobState`] is passed because only the implementor knows its `L`.
pub(crate) trait ItemFlush<B>: Send + Sync {
    fn flush_items<'a>(
        &'a self,
        op: &'a mut es_entity::DbOp<'static>,
        items: B,
        state: &'a OutboxEventJobState,
    ) -> BoxFuture<'a, Result<(), HandlerError>>;

    /// Where the subscription's cursor sits, on the handler's own lane.
    fn position_of(&self, state: &OutboxEventJobState) -> StreamPosition;
}

/// Restricted view of the batch op handed to
/// [`flush`](super::SingletonSubscriber::flush) — everything an
/// [`AtomicOperation`](es_entity::AtomicOperation) can do, and nothing else —
/// plus the lane position this batch lands at.
///
/// Committing belongs to the runner: after `flush` returns `Ok`, the
/// checkpoint is written and the transaction commits — items, work and
/// pointer land atomically. There is no access to the raw
/// [`es_entity::DbOp`], mirroring [`IsolatedOp`]'s sealing.
pub struct FlushOp<'a, L = InsertOrder>
where
    L: Lane,
{
    op: &'a mut es_entity::DbOp<'static>,
    position: L::Position,
}

impl<'a, L> FlushOp<'a, L>
where
    L: Lane,
{
    pub(crate) fn new(op: &'a mut es_entity::DbOp<'static>, position: L::Position) -> Self {
        Self { op, position }
    }

    /// Where this batch lands: the last *fully handled* event on this lane, and
    /// the checkpoint this transaction is about to commit. Folding `max` over the
    /// flushed items understates it when the batch ended on skipped events.
    pub fn position(&self) -> L::Position {
        self.position
    }
}

es_entity::delegate_atomic_operation!([<L: Lane>] FlushOp<'_, L>, { s => s.op });

/// A batch flush failed. Carries the sequence range actually at fault, so
/// the failure is not misattributed to the (innocent) event whose verb
/// happened to trigger the landing — e.g. a later event entering
/// [`consume`](EventCtx::consume).
///
/// Propagates through `handle_persistent` as a boxed error; downcast to
/// re-attribute in logs or traces.
#[derive(Debug)]
pub struct FlushError {
    /// Which trigger landed the batch (`"backlog_drained"`, `"batch_full"`,
    /// `"commit"`, `"consume_entry"`, `"shutdown"`, `"stream_closed"`,
    /// `"undecodable_event"`, and for keyed subscribers `"pause_entry"` and
    /// `"staged_pause"`).
    pub reason: &'static str,
    /// The batch covers positions strictly after this (the last durable
    /// checkpoint)…
    pub after: StreamPosition,
    /// …through this (the last fully handled event).
    pub through: StreamPosition,
    pub source: HandlerError,
}

impl std::fmt::Display for FlushError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "flush of batch ({}, {}] failed (reason={}): {}",
            self.after, self.through, self.reason, self.source
        )
    }
}

impl std::error::Error for FlushError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.source.as_ref())
    }
}

/// Land the pending batch, if any: hand collected items to the handler's
/// flush (inside a transaction opened for the landing) → checkpoint at the
/// last fully handled sequence → commit. Also lands a runner-committed op
/// left in `op_slot` by [`EventCtx::consume`]. No-op when nothing is pending.
#[tracing::instrument(
    name = "outbox.flush_batch",
    skip_all,
    fields(
        reason = reason,
        collected = parts.tracker.collected,
        checkpoint_seq = u64::from(parts.state.sequence),
    ),
    err
)]
pub(crate) async fn flush_batch<B: Default>(
    parts: &mut CtxParts<'_>,
    batch: &mut B,
    flusher: &dyn ItemFlush<B>,
    reason: &'static str,
) -> Result<(), HandlerError> {
    if parts.op_slot.is_none() && parts.tracker.collected == 0 {
        return Ok(());
    }
    if parts.tracker.collected > 0 {
        if parts.op_slot.is_none() {
            *parts.op_slot = Some(
                es_entity::DbOp::init_with_clock(
                    parts.current_job.pool(),
                    parts.current_job.clock(),
                )
                .await?,
            );
        }
        // Drain before the call: on error the items are dropped with the op,
        // and the replayed events re-collect them — the accumulator never
        // leaks stale state into a retry.
        let items = std::mem::take(batch);
        parts.tracker.collected = 0;
        let state = &*parts.state;
        let op = parts.op_slot.as_mut().expect("op was materialized above");
        let through = flusher.position_of(state);
        if let Err(source) = flusher.flush_items(op, items, state).await {
            return Err(Box::new(FlushError {
                reason,
                after: parts.tracker.persisted,
                through,
                source,
            }));
        }
    }
    let mut op = parts
        .op_slot
        .take()
        .expect("a pending batch always has an op by now");
    parts
        .current_job
        .update_execution_state_in_op(&mut op, parts.state)
        .await?;
    if let Some(mirror) = parts.mirror {
        mirror.mirror(&mut op, parts.state.sequence).await?;
    }
    op.commit().await?;
    parts.tracker.persisted = flusher.position_of(parts.state);
    parts.tracker.last_persist = tokio::time::Instant::now();
    Ok(())
}

/// Persist the checkpoint on its own — used for skip-only stretches where no
/// work op ever materialized, bounded by the configured checkpoint interval.
#[tracing::instrument(
    name = "outbox.persist_checkpoint",
    skip_all,
    fields(checkpoint_seq = u64::from(state.sequence)),
    err
)]
pub(crate) async fn persist_checkpoint(
    current_job: &mut CurrentJob,
    state: &OutboxEventJobState,
    mirror: Option<&dyn CheckpointMirror>,
) -> Result<(), HandlerError> {
    let mut op = es_entity::DbOp::init_with_clock(current_job.pool(), current_job.clock()).await?;
    current_job
        .update_execution_state_in_op(&mut op, state)
        .await?;
    if let Some(mirror) = mirror {
        mirror.mirror(&mut op, state.sequence).await?;
    }
    op.commit().await?;
    Ok(())
}

// === Keyed subscribers ===
//
// [`KeyedEventCtx`] below is the keyed counterpart of [`EventCtx`] — a facade
// over the exact same internals (`CtxParts`, `Outcome`, `flush_batch`) rather
// than a fork of them. It adds two capabilities past the shared verb set:
//
//   - the pause verb ([`KeyedEventCtx::pause_until`]), which mints
//     `Outcome::Pause`;
//   - staged processing ([`StagedOp`] / [`Suspended`]), which lets one
//     event be processed across N committed transactions with external I/O
//     between them, and whose paused exit mints `Outcome::CommitAndPause`.
//
// Both live on a distinct type because the split is CONTRACTUAL, not just a
// convenient type-gate. A singleton subscriber is always present, and that
// presence is what licenses its ephemeral subscription (unreplayable events
// need an always-present consumer); a verb that pauses consumption would
// contradict it. So the type-gating is a consequence of the semantics
// rather than the reason for it — the singleton runner never constructs a
// ctx that can reach these, and its `Outcome::Pause`/`CommitAndPause` arm is
// an `unreachable!` by construction.

/// Per-event decision point handed to
/// [`KeyedSubscriber::handle`](super::KeyedSubscriber::handle) — the keyed
/// counterpart of [`EventCtx`], sharing its verb semantics;
/// [`pause_until`](Self::pause_until) and [`consume`](Self::consume)'s staged
/// chain are the keyed-only additions.
#[must_use = "resolve the KeyedEventCtx via skip / collect / consume / pause_until"]
pub struct KeyedEventCtx<'inv, B = ()> {
    pub(crate) parts: CtxParts<'inv>,
    pub(crate) batch: &'inv mut B,
    pub(crate) flusher: &'inv dyn ItemFlush<B>,
    /// Sequence of the event being processed. Distinct from
    /// `parts.state.sequence`, which is the last *fully handled* sequence —
    /// i.e. strictly before this one.
    pub(crate) event_seq: EventSequence,
}

impl<'inv, B> KeyedEventCtx<'inv, B> {
    /// Identical to [`EventCtx::skip`].
    pub fn skip(self) -> Handled<'inv> {
        Handled {
            outcome: Outcome::Skip,
            _invocation: PhantomData,
        }
    }

    /// Identical to [`EventCtx::collect_with`].
    pub fn collect_with(self, f: impl FnOnce(&mut B)) -> Handled<'inv> {
        f(self.batch);
        self.parts.tracker.collected += 1;
        Handled {
            outcome: Outcome::Collect,
            _invocation: PhantomData,
        }
    }

    /// Take this event's work, in one stage or many.
    ///
    /// Same entry fence as [`EventCtx::consume`]: the pending batch (its
    /// collected items and its checkpoint, at the last fully-handled
    /// sequence — strictly before this event) lands first, then a fresh op
    /// is opened. The difference is the exits: a [`StagedOp`] can be
    /// [`commit`](StagedOp::commit)ted (cursor advances, exactly the
    /// isolated-commit semantics), or it can [`suspend`](StagedOp::suspend)
    /// — land this stage and hand back a [`Suspended`] with **no transaction
    /// open**, so the subscriber can do external I/O before
    /// [`resume`](Suspended::resume) opens the next stage's op.
    ///
    /// The single-transaction case is the one-stage degenerate case: consume,
    /// work, [`commit`](StagedOp::commit).
    ///
    /// Interim stages are fenced before the cursor and replayed on crash:
    /// nothing a `suspend` landed is lost, but the event itself is re-read
    /// and re-handled from its first stage until a `commit` advances past
    /// it, so what an interim stage lands must tolerate landing again.
    pub async fn consume(self) -> Result<StagedOp<'inv>, HandlerError>
    where
        B: Default,
    {
        let KeyedEventCtx {
            mut parts,
            batch,
            flusher,
            event_seq,
        } = self;
        flush_batch(&mut parts, batch, flusher, "consume_entry").await?;
        let op =
            es_entity::DbOp::init_with_clock(parts.current_job.pool(), parts.current_job.clock())
                .await?;
        Ok(StagedOp {
            op,
            parts,
            event_seq,
        })
    }

    /// My cursor stays *before* this event until `at` — entry and exit in
    /// one, nothing to record.
    ///
    /// The runner lands any pending batch first (the same fence as
    /// [`consume`](Self::consume)'s entry — checkpoint at
    /// the last fully-handled sequence, which is pre-this-event), persists
    /// the checkpoint if dirty, then ends the run rescheduled at `at`. Does
    /// **not** advance the checkpoint past this event: the next run re-reads
    /// and re-evaluates it, so a paused event is retried, not skipped.
    ///
    /// The resume time is domain knowledge (e.g. a retry schedule owned by
    /// the delivery entity) — the one fact obix cannot derive on its own.
    /// Everything else about parking and waking (passivation, generations,
    /// wake) is derivable and stays internal.
    ///
    /// A pause is cut short by traffic, not by time alone: a wake-key match
    /// for an event *behind* the paused one re-delivers the paused event
    /// early, so a subscriber that still cannot proceed pauses again. A wake
    /// with nothing persisted beyond the paused event — the match for that
    /// very event landing after the pause did — is answered by the runner
    /// pausing again, without the subscriber being invoked.
    pub fn pause_until(self, at: chrono::DateTime<chrono::Utc>) -> Handled<'inv> {
        let KeyedEventCtx {
            parts, event_seq, ..
        } = self;
        parts.state.paused = Some(PausedState {
            sequence: event_seq,
            until: at,
        });
        Handled {
            outcome: Outcome::Pause(at),
            _invocation: PhantomData,
        }
    }
}

impl<'inv, T> KeyedEventCtx<'inv, Vec<T>> {
    /// [`collect_with`](Self::collect_with) sugar for `Vec` accumulators.
    pub fn collect(self, item: T) -> Handled<'inv> {
        self.collect_with(|batch| batch.push(item))
    }
}

impl<'inv, K, V, S> KeyedEventCtx<'inv, std::collections::HashMap<K, V, S>>
where
    K: std::hash::Hash + Eq,
    S: std::hash::BuildHasher,
{
    /// [`collect_with`](Self::collect_with) sugar for `HashMap` accumulators.
    pub fn collect(self, key: K, value: V) -> Handled<'inv> {
        self.collect_with(|batch| {
            batch.insert(key, value);
        })
    }
}

/// One stage of a keyed subscriber's processing of one event: an open op,
/// whose work lands whichever exit is taken.
///
/// Implements [`AtomicOperation`](es_entity::AtomicOperation) — use it like
/// any atomic operation, then take exactly one exit:
///
/// | exit | meaning | cursor |
/// |------|---------|--------|
/// | [`commit`](Self::commit) | processing of this event is done | advances past the event |
/// | [`suspend`](Self::suspend) | this stage is done, the chain continues | unmoved |
/// | [`pause_until`](Self::pause_until) | this stage is done, processing pauses until `at` | unmoved |
///
/// The verbs are about the *event*, not the transaction: every exit lands
/// this stage's writes, and what differs is what happens to the cursor.
/// `commit` means what [`IsolatedOp::commit`] means — work and checkpoint
/// land together — so a keyed subscriber that never stages reads exactly
/// like a singleton one: consume, work, commit. `suspend` and `pause_until`
/// land the work and leave the checkpoint where it was.
///
/// As with [`IsolatedOp`], there is no mutable access to the raw
/// [`es_entity::DbOp`] — the op can only land through one of the exits, so
/// no stage can commit without the runner knowing what it meant.
///
/// Sealed to its invocation by the same brand [`Handled`] carries, so a
/// staged op cannot be stashed and resumed from a later event:
///
/// ```compile_fail
/// use obix::StagedOp;
///
/// struct Evil {
///     stash: std::sync::Mutex<Option<StagedOp<'static>>>,
/// }
///
/// async fn stash_it(op: StagedOp<'_>, evil: &Evil) {
///     // error[E0521]: borrowed data escapes outside of function
///     *evil.stash.lock().unwrap() = Some(op);
/// }
/// ```
#[must_use = "exit with .commit() / .suspend() / .pause_until()"]
pub struct StagedOp<'inv> {
    op: es_entity::DbOp<'static>,
    parts: CtxParts<'inv>,
    event_seq: EventSequence,
}

impl<'inv> StagedOp<'inv> {
    /// This stage is done, the event is not: the returned [`Suspended`]
    /// holds **no open transaction**, so the subscriber can await external
    /// I/O before [`resume`](Suspended::resume) opens the next stage's op.
    ///
    /// The cursor does not move — this event is still being processed, and a
    /// crash here replays it (with everything this stage committed already
    /// durable).
    pub async fn suspend(self) -> Result<Suspended<'inv>, HandlerError> {
        let StagedOp {
            op,
            parts,
            event_seq,
        } = self;
        op.commit().await?;
        Ok(Suspended { parts, event_seq })
    }

    /// This stage is done and processing pauses, with the cursor still
    /// parked *before* this event, until `at`.
    ///
    /// The op's work lands, but the checkpoint the runner folds in is still
    /// pre-this-event: the next run re-reads the event and re-evaluates. The
    /// resume time is domain knowledge (a retry schedule owned by the
    /// consumer's entities) — the one fact obix cannot derive.
    pub fn pause_until(self, at: chrono::DateTime<chrono::Utc>) -> Handled<'inv> {
        let StagedOp {
            op,
            parts,
            event_seq,
        } = self;
        parts.state.paused = Some(PausedState {
            sequence: event_seq,
            until: at,
        });
        *parts.op_slot = Some(op);
        Handled {
            outcome: Outcome::CommitAndPause(at),
            _invocation: PhantomData,
        }
    }

    /// Processing of this event is done: the runner folds the checkpoint at
    /// this event's sequence into the same transaction, so work and cursor
    /// advance together — what [`IsolatedOp::commit`] does for a singleton.
    pub fn commit(self) -> Handled<'inv> {
        let StagedOp { op, parts, .. } = self;
        *parts.op_slot = Some(op);
        Handled {
            outcome: Outcome::Commit,
            _invocation: PhantomData,
        }
    }
}

impl std::ops::Deref for StagedOp<'_> {
    type Target = es_entity::DbOp<'static>;

    fn deref(&self) -> &Self::Target {
        &self.op
    }
}

es_entity::delegate_atomic_operation!(StagedOp<'_>, { s => s.op });

/// The gap between two stages of processing one event: **no transaction is
/// open**, which is exactly the point — this is where a subscriber awaits
/// external I/O.
///
/// [`resume`](Self::resume) opens the next stage's op (it comes from the
/// job's pool and clock, and is traced like any other, which is why this is
/// ctx-mediated rather than a side-op the consumer opens itself);
/// [`pause_until`](Self::pause_until) pauses instead.
///
/// Sealed to its invocation, exactly as [`StagedOp`] is:
///
/// ```compile_fail
/// use obix::Suspended;
///
/// struct Evil {
///     stash: std::sync::Mutex<Option<Suspended<'static>>>,
/// }
///
/// async fn stash_it(staged: Suspended<'_>, evil: &Evil) {
///     // error[E0521]: borrowed data escapes outside of function
///     *evil.stash.lock().unwrap() = Some(staged);
/// }
/// ```
#[must_use = "continue with .resume() or pause with .pause_until()"]
pub struct Suspended<'inv> {
    parts: CtxParts<'inv>,
    event_seq: EventSequence,
}

impl<'inv> Suspended<'inv> {
    /// Processing of the event resumes: the next stage's op is opened.
    pub async fn resume(self) -> Result<StagedOp<'inv>, HandlerError> {
        let Suspended { parts, event_seq } = self;
        let op =
            es_entity::DbOp::init_with_clock(parts.current_job.pool(), parts.current_job.clock())
                .await?;
        Ok(StagedOp {
            op,
            parts,
            event_seq,
        })
    }

    /// Pause with nothing further to record: the cursor stays *before* this
    /// event until `at`, exactly as [`KeyedEventCtx::pause_until`] does.
    pub fn pause_until(self, at: chrono::DateTime<chrono::Utc>) -> Handled<'inv> {
        let Suspended { parts, event_seq } = self;
        parts.state.paused = Some(PausedState {
            sequence: event_seq,
            until: at,
        });
        Handled {
            outcome: Outcome::Pause(at),
            _invocation: PhantomData,
        }
    }
}

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

    /// Execution-state rows written before pausing existed have no `paused`
    /// field. They must still decode — a keyed subscriber upgrading into
    /// this version resumes from its checkpoint rather than restarting.
    #[test]
    fn pre_pause_execution_state_still_decodes() {
        let state: OutboxEventJobState =
            serde_json::from_str(r#"{"sequence":42}"#).expect("legacy state must decode");
        assert_eq!(u64::from(state.sequence), 42);
        assert!(state.paused.is_none());
    }

    /// And a state with no pause serializes back to exactly what a singleton
    /// subscriber has always written — the new field adds no bytes.
    #[test]
    fn state_without_a_pause_serializes_unchanged() {
        let state = OutboxEventJobState {
            sequence: EventSequence::from(7u64),
            commit_sequence: None,
            paused: None,
        };
        assert_eq!(
            serde_json::to_string(&state).expect("serializes"),
            r#"{"sequence":7}"#
        );
    }
}