skilj-core 0.1.1

//! The public plugin API - the one module every consuming application's
//! own code touches directly. See [docs/architecture.md §1](../../../docs/architecture.md#plugin-api-decide-project) for the full
//! reasoning behind this shape.

use crate::event_store::{Event, MissedOccurrencePolicy};
use crate::shared::{
    CommandDecision, PrivateField, QueryItemMapping, SensitiveField, Tag, TagMapping,
};
use schemars::JsonSchema;
use serde::{de::DeserializeOwned, Serialize};

/// See `CommandType::required_role`'s own doc comment and docs/
/// architecture.md §1.3.1 - the one deliberate proc-macro exception to
/// this module's otherwise macro-free plugin API. Re-exported here so a
/// consumer only ever needs `skilj_core::plugin::{CommandType,
/// requires_role}`, not a direct dependency on `skilj-macros` itself.
pub use skilj_macros::requires_role;

/// The fallback every plugin trait's `BOUNDED_CONTEXT` associated const
/// carries - see that const's own doc comment on `EventType`/
/// `CommandType`/`Projection`. Also what `skilj::SkiljBuilder::builder()`
/// seeds `current_bounded_context` with, so a single-bounded-context app
/// never has to call `.bounded_context(...)` at all: every `.event_type::<T>()`/
/// `.command_type::<T>()`/`.projection::<T>()` call registers under this
/// name until something changes it.
pub const DEFAULT_BOUNDED_CONTEXT: &str = "default";

/// Bridges a raw, type-erased `event_store::Event` into the
/// strongly-typed, hand-written per-bounded-context event enum
/// `decide()`/`project()` pattern-match over (§1.4's `BankingEvent`, say),
/// implemented once per bounded context, on that enum. See docs/
/// architecture.md §1.6 for the full reasoning; §1.7's decider bridge is
/// the one caller of `try_from_event` in this crate.
///
/// ```ignore
/// impl BoundedContextEvent for BankingEvent {
///     fn try_from_event(event: &Event) -> Option<Result<Self, serde_json::Error>> {
///         match event.event_type.name.as_str() {
///             "MoneyDeposited" => Some(serde_json::from_str(&event.payload).map(Self::MoneyDeposited)),
///             "MoneyWithdrawn" => Some(serde_json::from_str(&event.payload).map(Self::MoneyWithdrawn)),
///             _ => None,
///         }
///     }
/// }
/// ```
pub trait BoundedContextEvent: Sized {
    /// `None` when `event.event_type.name` doesn't match any variant this
    /// enum declares - never expected in practice (a caller only ever
    /// passes this bounded context's own events), so `None` is a
    /// defensive case, not a designed-for one. `Some(Err(..))` when the
    /// stored payload doesn't deserialize into the matched variant's
    /// payload type - narrower than it used to be (an externally- or
    /// directly-created event's own payload is now schema-checked before
    /// it's ever stored), but still reachable: a command-triggered or
    /// system-triggered event's payload came from `decide()`/
    /// `scheduled_payload`, which are never schema-checked at all (see
    /// the note above `entity CommandType`'s "Payload schema shape" in
    /// specs/skilj.allium), and even a schema-valid payload can still be
    /// stricter than the compiled Rust type expects.
    fn try_from_event(event: &Event) -> Option<Result<Self, serde_json::Error>>;
}

/// One step in an `upcast_payload` chain - see that function's own doc
/// comment. `to_version` is the schema_version this step's `transform`
/// produces: a payload whose own `Metadata.version` is strictly less
/// than `to_version` needs this step applied before whichever step (or
/// final deserialization) comes next.
#[derive(Clone, Copy)]
pub struct UpcastStep {
    /// The schema version `transform` produces.
    pub to_version: i64,
    /// Rewrites a payload written before `to_version` into that version's
    /// shape.
    pub transform: fn(serde_json::Value) -> serde_json::Value,
}

/// Sugar around the one pattern a hand-written `BoundedContextEvent::
/// try_from_event` already has every ingredient for: `event.metadata.version`
/// records the schema_version a payload was actually written under
/// (specs/skilj.allium's own note above, "that is what Metadata.version
/// is for"), and nothing stops a `try_from_event` arm from branching on
/// it before deserializing. This is that branch, generalised into a
/// declared chain of pure JSON transforms instead of a repeated
/// `match metadata.version { ... }` per event type. See
/// [docs/architecture.md §33](../../../docs/architecture.md#payload-upcasting) for the full "why not first-class
/// upcasting" investigation this is the one built piece of.
///
/// Not a replacement for `schema_is_backwards_compatible`'s additive-only
/// contract, and doesn't need to be: a genuinely reshaping change
/// (rename, retype, split/merge a field) still can't be registered as a
/// revision of an existing type - `Self::Payload`'s own derived JSON
/// Schema only ever has to describe the *current*, already-reshaped
/// value, because this runs entirely on the raw `serde_json::Value`
/// before `serde_json::from_value::<T>` ever sees it. The transform
/// chain lives in application code only; nothing about the registered
/// schema, `RegisterEventType`, or `RegisterCommandType` changes.
///
/// `chain` is walked once, in the order given, applying every step whose
/// `to_version` is strictly greater than `written_at_version` - a
/// payload written at version 1 runs every step in a `[..to 2, ..to 3]`
/// chain, one written at version 2 runs only the second, one written at
/// version 3 (or later) runs none. Steps are expected in ascending
/// `to_version` order; this does not sort them - an out-of-order chain
/// applies transforms in the wrong sequence, the same "caller's own
/// responsibility" register `consumed_event_types()` already trusts the
/// caller to get right.
///
/// ```ignore
/// // MoneyDeposited moved from a float-dollar `amount` (schema_version 1)
/// // to an integer-cent `amount_cents` (schema_version 2) - a genuine
/// // reshape schema_is_backwards_compatible would reject as a revision
/// // of the same type, handled here entirely in application code:
/// const MONEY_DEPOSITED_UPCASTS: &[UpcastStep] = &[UpcastStep {
///     to_version: 2,
///     transform: |mut v| {
///         if let Some(amount) = v.get("amount").and_then(|a| a.as_f64()) {
///             if let Some(obj) = v.as_object_mut() {
///                 obj.remove("amount");
///                 obj.insert("amount_cents".into(), ((amount * 100.0).round() as i64).into());
///             }
///         }
///         v
///     },
/// }];
///
/// "MoneyDeposited" => Some(
///     skilj_core::plugin::upcast_payload(
///         &event.payload,
///         event.metadata.version,
///         MONEY_DEPOSITED_UPCASTS,
///     )
///     .map(Self::MoneyDeposited),
/// ),
/// ```
pub fn upcast_payload<T: DeserializeOwned>(
    payload: &str,
    written_at_version: i64,
    chain: &[UpcastStep],
) -> Result<T, serde_json::Error> {
    let mut value: serde_json::Value = serde_json::from_str(payload)?;
    for step in chain {
        if written_at_version < step.to_version {
            value = (step.transform)(value);
        }
    }
    serde_json::from_value(value)
}

/// One command type a bounded context registers.
///
/// `decide()` is synchronous and pure: it receives only `payload` and
/// `matching_events`, nothing else - no database access, no I/O. That's
/// deliberate (§1.1): `ProcessCommand`'s optimistic-then-locked retry may
/// call `decide()` more than once per submission.
pub trait CommandType {
    /// The payload shape, and the source of its JSON Schema - derived via
    /// `schemars`, never hand-written (§1.2).
    type Payload: Serialize + DeserializeOwned + JsonSchema;

    /// The bounded context's own generated event enum, one variant per
    /// registered `EventType`. `matching_events` is typed against this so
    /// a missed match arm is a compile error, not a silently-ignored
    /// event type (§1.4). `: BoundedContextEvent` is what lets §1.7's
    /// decider bridge convert this bounded context's raw, stored `Event`s
    /// into it before calling `decide()`.
    type Event: BoundedContextEvent;

    /// The command type's name, unique within its bounded context - what
    /// GraphQL, REST and `EventSpec::event_type` name it by.
    const NAME: &'static str;

    /// Which bounded context `skilj::SkiljBuilder::auto_register()`
    /// registers this command type under - see `skilj_macros::auto_register`'s
    /// own doc comment for the `#[auto_register]` attribute that reads
    /// this. Irrelevant to manual `.bounded_context(name).command_type::<T>()`
    /// chaining, which ignores this const entirely and keeps working
    /// exactly as before.
    ///
    /// Defaults to `DEFAULT_BOUNDED_CONTEXT` ("default"), so a
    /// single-bounded-context app never has to override this at all. A
    /// multi-context app points it at its own module-level
    /// `BOUNDED_CONTEXT` const (the same one `skilj-demo`'s `banking`/
    /// `courses` modules already declare), e.g.
    /// `const BOUNDED_CONTEXT: &'static str = BOUNDED_CONTEXT;`.
    const BOUNDED_CONTEXT: &'static str = DEFAULT_BOUNDED_CONTEXT;

    /// The consistency tags this command's payload carries: each maps a tag
    /// key to a payload field. `decide()` is handed every event sharing at
    /// least one of the resulting tags. Empty by default.
    fn tag_mappings() -> Vec<TagMapping> {
        Vec::new()
    }

    /// See `plugin::EventType::owner_tag_key`'s own doc comment - identical
    /// role here: which of `tag_mappings()`'s own keys names this command
    /// type's "owner" dimension, `None` (the default) for one with no such
    /// notion. Registered and validated exactly the same way
    /// (`RegisterCommandType`'s own `valid_owner_tag_key`), and read by
    /// `event_store::command_owner_scope_satisfied` - `Command.consistency_tags`'
    /// own sibling check to `event_owner_scope_satisfied`'s `Event.tags`.
    /// Cross-tenant read fix (docs/architecture.md's own write-up of these
    /// passes).
    fn owner_tag_key() -> Option<&'static str> {
        None
    }

    /// Payload fields encrypted per subject, so `forgetSubject` makes them
    /// unreadable. Empty by default.
    fn sensitive_fields() -> Vec<SensitiveField> {
        Vec::new()
    }

    /// See `EventType::private_fields()`'s own doc comment - identical
    /// role, for `Command.payload` instead of `Event.payload`.
    fn private_fields() -> Vec<PrivateField> {
        Vec::new()
    }

    /// Opt-in: submitting over GraphQL with a write-level grant is always
    /// permitted; triggering over REST with a `CommandToken` needs this.
    fn rest_trigger_allowed() -> bool {
        false
    }

    /// Opt-in approval flow (issue #98, docs/architecture.md
    /// §214): when `true`, a submission - REST or GraphQL - creates a
    /// pending `CommandProposal` instead of running `decide()`. An
    /// approver with an active admin-level `RoleAccessMapping` on the
    /// bounded context then approves it over GraphQL, which runs the
    /// ordinary submission pipeline against the state at that time.
    ///
    /// A command may still be rejected by `decide()` at approval time;
    /// the proposal is then recorded as rejected, with the decision's own
    /// reason. `false` by default, so every existing command type keeps
    /// its immediate-submission behaviour.
    fn requires_approval() -> bool {
        false
    }

    /// Narrows who may approve a `requires_approval()` command type, on
    /// top of the admin grant every approver needs: `Some(name)` requires
    /// the approver's `Role.name` to equal `name`, `None` (the default)
    /// lets any active admin-level mapping approve. Ignored for a type
    /// that does not require approval.
    fn approval_role() -> Option<&'static str> {
        None
    }

    /// An optional approval deadline, in seconds from proposal creation.
    /// `None` (the default) means a proposal stays approvable until it is
    /// approved or rejected. An expired proposal is recorded as expired
    /// and cannot be approved.
    fn approval_deadline_seconds() -> Option<i64> {
        None
    }

    /// Whether the account that proposed a command may also approve it.
    /// `false` (the default) requires a different approver - a different
    /// human, not merely a different Role row; rejecting - withdrawing -
    /// one's own proposal is always allowed. Ignored for a type that does
    /// not require approval.
    fn approval_allows_self_approval() -> bool {
        false
    }

    /// An extra caller-facing gate, on top of the ordinary write-level
    /// `RoleAccessMapping` check: `None` (the default) means no extra
    /// restriction. Set this by writing `#[requires_role("name")]`
    /// directly above the `impl CommandType for ...` block - see
    /// `requires_role`'s own doc comment - rather than overriding this
    /// method by hand; the attribute is what makes the requirement read
    /// as part of the command's own declaration.
    ///
    /// Not a spec-level concept, not persisted anywhere: `Role.name` has
    /// no uniqueness guarantee in `specs/skilj.allium` (see `entity
    /// Role`), so this is only as safe as the deployment's own discipline
    /// keeping role names meaningful and non-colliding - a `skilj-core`
    /// concern, not something the engine itself enforces. Checked
    /// exclusively by `skilj-graphql`'s own mutation resolver (docs/
    /// architecture.md §1.3.1, [§8](../../../docs/architecture.md#open-for-a-future-pass) item 5) - REST triggering is untouched,
    /// since `CommandToken` is already its own, separate per-token grant.
    fn required_role() -> Option<&'static str> {
        None
    }

    /// Decides from the events the command's query selects. Implement
    /// this, or [`decide_with`](CommandType::decide_with) when the
    /// decision also needs a snapshot's state; the framework only calls
    /// `decide_with`, whose default calls this. The default here refuses
    /// every command, for a type that implements neither.
    fn decide(_payload: &Self::Payload, _matching_events: &[Self::Event]) -> CommandDecision {
        CommandDecision::Rejected {
            reason: "this CommandType implements neither decide() nor decide_with()".to_string(),
            kind: "decide_not_implemented".to_string(),
        }
    }

    /// What the framework calls to decide a command
    /// (docs/architecture.md §202): `decision.events()` is what
    /// `matching_events` is to `decide()`, and `decision.snapshot::<S>(value)`
    /// is snapshot `S`'s state for a tag value one of the query's
    /// [`QueryItemMapping::snapshot`] items derived, already brought up to
    /// date - nothing to fold. The default calls `decide()` with the
    /// events, so a type that needs no snapshot implements `decide()` alone.
    fn decide_with(
        payload: &Self::Payload,
        decision: &Decision<'_, Self::Event>,
    ) -> CommandDecision {
        Self::decide(payload, decision.events())
    }

    /// Which events `decide()` receives, and which later events conflict
    /// with it (docs/architecture.md §198, the DCB query model). Each item
    /// matches events of its types (any type when it names none) that
    /// carry all of its tags; `matching_events` is every event matching
    /// any item, in sequence order, except that an item with `latest`
    /// contributes only that many of its matches, the newest
    /// (`QueryItemMapping::latest()` for one, `last(n)` for n).
    ///
    /// Empty (the default) means one item per `tag_mappings()` entry, any
    /// type, every match: every event carrying any of the command's tags,
    /// as before queries existed. A narrower query reads less, and also
    /// conflicts less: an event carrying the command's tag but matching
    /// no item no longer makes it decide again.
    ///
    /// `vec![QueryItemMapping::nothing()]` is a command whose `decide()`
    /// needs no history (docs/architecture.md §201): it reads no events,
    /// and nothing committed after it is read conflicts with it.
    ///
    /// `build()` refuses a query naming an event type its bounded context
    /// doesn't register in the same builder, a tag mapping that isn't one
    /// of `tag_mappings()`, an item with neither types nor tags other than
    /// a lone `nothing()`, and a query together with `snapshot()`.
    fn consistency_query() -> Vec<QueryItemMapping> {
        Vec::new()
    }

    /// Superseded by snapshot items in `consistency_query()`
    /// (`QueryItemMapping::snapshot`, docs/architecture.md §202), which
    /// work with any number of tags and hand `decide_with()` the state
    /// already brought up to date; kept working for existing command
    /// types.
    ///
    /// Opt-in accelerator for `decide()` against a large tag-scoped
    /// history ([docs/architecture.md §19](../../../docs/architecture.md#optional-snapshotting-matching-events)'s "Problem 2") - names the
    /// `Snapshot::NAME` this command type's own decide() can fold
    /// forward from instead of replaying every matching event. `None`
    /// (the default) means no snapshot; every existing `CommandType`
    /// impl is completely unaffected.
    ///
    /// Only actually used when this command type derives exactly the
    /// one tag the named `Snapshot::TAG_KEY` is - a multi-tag command
    /// that names a snapshot anyway is a harmless misconfiguration,
    /// silently ignored (falls back to the ordinary `decide()` path),
    /// not a hard error - see `decide_from_snapshot`'s own doc comment
    /// for the *reachable* misconfiguration this trait can't catch for
    /// you.
    fn snapshot() -> Option<&'static str> {
        None
    }

    /// Called instead of `decide()` when `snapshot()` names a real,
    /// tag-matching `Snapshot` (see that method's own doc comment).
    /// `snapshot_state_json` is `Snapshot::State::default()`,
    /// JSON-encoded, when no snapshot has been written yet for this tag
    /// value, or when a stored one's own `snapshot_version` no longer
    /// matches `Snapshot::VERSION` - the model changed, so it's treated
    /// as absent rather than trusted; `events_since_snapshot` is then
    /// simply everything for this tag, the same set `decide()` would
    /// have seen. Correctness obligation: this must reach the identical
    /// `CommandDecision` `decide()` would over the *full* matching_events
    /// for the same payload - `snapshot_state_json` plus
    /// `events_since_snapshot` is meant to be exactly equivalent
    /// information, not an approximation.
    ///
    /// `snapshot_state_json` deliberately isn't a typed `Snapshot::State`,
    /// since linking `CommandType` to one specific `Snapshot` impl at
    /// the type level needs either a new required associated type
    /// (breaking every existing `CommandType` impl) or a *defaulted*
    /// one (`associated_type_defaults`, nightly-only). Deserialise it
    /// yourself, the same way `tag_mappings()`'s own `field` is already
    /// just a string, unchecked against `Payload` at compile time.
    ///
    /// The default implementation is a safe, non-panicking rejection -
    /// only ever reached if a `CommandType` overrides `snapshot()`
    /// without also overriding this: a real misconfiguration (unlike
    /// the multi-tag case above), but one this trait can still fail
    /// safely on rather than silently producing a wrong decision.
    ///
    /// `context` (issue #97) is the same [`DecisionContext`]
    /// `decide_with` reads through [`Decision::context`].
    fn decide_from_snapshot(
        _payload: &Self::Payload,
        _snapshot_state_json: &str,
        _events_since_snapshot: &[Self::Event],
        _context: &DecisionContext,
    ) -> CommandDecision {
        CommandDecision::Rejected {
            reason: "this CommandType declares snapshot() but never overrides \
                     decide_from_snapshot()"
                .to_string(),
            kind: "snapshot_misconfigured".to_string(),
        }
    }
}

/// One event type a bounded context registers. Unlike `CommandType`/
/// `Projection`, there's no behaviour to plug in here - an `EventType` is
/// pure declaration (its schema, its tag/sensitivity mappings, which
/// origins may create it).
pub trait EventType {
    /// The payload shape, and the source of its JSON Schema - derived via
    /// `schemars`, never hand-written.
    type Payload: Serialize + DeserializeOwned + JsonSchema;

    /// The event type's name, unique within its bounded context - what
    /// `EventSpec::event_type` and the API surfaces name it by.
    const NAME: &'static str;

    /// See `CommandType::BOUNDED_CONTEXT`'s own doc comment - identical
    /// role and default here, for `skilj::SkiljBuilder::auto_register()`'s
    /// `EventType` side.
    const BOUNDED_CONTEXT: &'static str = DEFAULT_BOUNDED_CONTEXT;

    /// The tags this event's payload carries: each maps a tag key to a payload
    /// field. A command whose tags overlap an event's sees that event in
    /// `decide()`. Empty by default.
    fn tag_mappings() -> Vec<TagMapping> {
        Vec::new()
    }

    /// Which of `tag_mappings()`'s own keys names this type's "owner"
    /// dimension, if it has one - `Some("company")` for a type whose
    /// events belong to different tenants sharing one bounded context,
    /// `None` (the default) for a type with no such notion. Unlike
    /// `plugin::Projection::OWNER_TAG_KEY` (deliberately Rust-only, no
    /// spec field), this one is registered and validated:
    /// `RegisterEventType`'s own `valid_owner_tag_key` requires it be
    /// null or name a real `tag_mappings()` key, since `EventType.tag_mappings`
    /// is already a real, spec'd, registered field this joins rather than
    /// a fresh Rust-only concept. Cross-tenant read fix
    /// (docs/architecture.md's own write-up of these passes).
    fn owner_tag_key() -> Option<&'static str> {
        None
    }

    /// Payload fields encrypted per subject, so `forgetSubject` makes them
    /// unreadable. Empty by default.
    fn sensitive_fields() -> Vec<SensitiveField> {
        Vec::new()
    }

    /// See `value PrivateField`'s own doc comment - a plain read-time
    /// redaction rule, opt-in per field, default empty like
    /// `sensitive_fields()` above. Private-field mechanism
    /// (docs/architecture.md's own write-up of this pass).
    fn private_fields() -> Vec<PrivateField> {
        Vec::new()
    }

    /// Opt-in: whether an external system holding an external-event token
    /// (`createExternalEventToken`) may create this event over REST, as the
    /// broker bridges do. `false` by default.
    fn external_creation_allowed() -> bool {
        false
    }

    /// Opt-in: whether this event may be created without a command, by a
    /// caller holding a direct-creation token (`createDirectCreationToken`).
    /// `false` by default.
    fn direct_creation_allowed() -> bool {
        false
    }

    /// Opt-in: whether event read tokens may be minted for this event type
    /// (`createEventReadToken`), so REST can serve its events. `false` by
    /// default.
    fn event_read_allowed() -> bool {
        false
    }

    /// Opt-in: `rule CreateSystemEvent`'s own gate, `EventType.
    /// system_triggered_allowed`. `false` by default, like every other
    /// creation-path flag on this trait - a type that never overrides
    /// this three plus `system_triggered_schedule`/`missed_occurrence_policy`
    /// is never scheduled at all, so `scheduled_payload` below is never
    /// called for it.
    fn system_triggered_allowed() -> bool {
        false
    }

    /// The 7-field Quartz-dialect cron expression (`cron` crate) the
    /// scheduler evaluates - required, and checked at registration time
    /// (see `Error::MissingScheduleOrPolicy`), when
    /// `system_triggered_allowed()` is `true`. `None` otherwise.
    fn system_triggered_schedule() -> Option<String> {
        None
    }

    /// See `enum MissedOccurrencePolicy`. Deliberately no default that
    /// silently picks one policy over another - the user's own call on
    /// this: "either way can be potentially dangerous", so a type opting
    /// into scheduling must say which it means, explicitly, by
    /// overriding this method; `register_event_type`'s own guard rejects
    /// registration outright if this is still `None` while
    /// `system_triggered_allowed()` is `true`.
    fn missed_occurrence_policy() -> Option<MissedOccurrencePolicy> {
        None
    }

    /// `scheduled_payload(event_type)` - the black box `rule
    /// CreateSystemEvent` names. Called at most once per eligible
    /// occurrence (see `event_store::create_system_event`'s own doc
    /// comment on why it's never called for one that's about to be
    /// rejected), never for a type that isn't `system_triggered_allowed`.
    /// Panics if reached without being overridden - a type opting into
    /// scheduling without ever producing a payload for it is a genuine
    /// programming error, not a runtime condition to handle gracefully.
    fn scheduled_payload() -> Self::Payload {
        unimplemented!(
            "{}::scheduled_payload() must be overridden - this event type is registered with \
             system_triggered_allowed() = true, so the scheduler needs a payload for every \
             occurrence it fires",
            Self::NAME
        )
    }
}

/// Type-erased access to a bounded context's own `EventType::
/// scheduled_payload()`, for the background scheduler in `db::
/// fire_system_event` - the `EventType` equivalent of `CommandDispatcher`/
/// `ProjectionDispatcher` above `Skilj` implements over its own compiled-in
/// registry.
pub trait EventDispatcher: Send + Sync {
    /// `scheduled_payload()`'s own output, JSON-serialised - `None` when
    /// no `(bounded_context, event_type)` pair matches anything
    /// registered in this process, the same "pair isn't registered at
    /// all" convention `CommandDispatcher::dispatch`/`ProjectionDispatcher::
    /// keys` already use. A genuine misconfiguration when it happens for
    /// a type the database itself marks `system_triggered_allowed` (see
    /// `@guarantee ScheduleStateIsShared`'s own expectation that every
    /// instance in a cluster registers the same scheduled types) -
    /// `db::fire_system_event`'s caller treats it as a skip, not a panic,
    /// the same tolerance `catch_up_bounded_context` already extends to
    /// an unregistered projection dispatcher.
    fn scheduled_payload(&self, bounded_context: &str, event_type: &str) -> Option<String>;
}

/// One projection a bounded context registers.
///
/// `project()` folds exactly one event into this projection's stored
/// state - synchronous, like `decide()`, and for the same reason (§1.1).
pub trait Projection {
    /// The state `project()` folds events into, stored as JSON. `Default` is
    /// the state before any event.
    type State: Serialize + DeserializeOwned + JsonSchema + Default;

    /// Same generated per-bounded-context event enum `CommandType::Event`
    /// uses - a projection folds any event type it consumes, not just
    /// one. Same `: BoundedContextEvent` bound and reasoning.
    type Event: BoundedContextEvent;

    /// The projection's name, unique within its bounded context - what
    /// `queryProjection` names it by.
    const NAME: &'static str;

    /// See `CommandType::BOUNDED_CONTEXT`'s own doc comment - identical
    /// role and default here, for `skilj::SkiljBuilder::auto_register()`'s
    /// `Projection` side.
    const BOUNDED_CONTEXT: &'static str = DEFAULT_BOUNDED_CONTEXT;

    /// The `EventType::NAME`s this projection actually folds - `RegisterProjection`'s
    /// own `consumed_event_types`. No default: unlike `tag_mappings()`/
    /// `sensitive_fields()`, there's no reasonable "consumes nothing"
    /// default for a projection to fall back to, and Rust has no way to
    /// infer this from `project()`'s own body (which variants of
    /// `Self::Event` it actually matches on isn't something the type
    /// system exposes) - so it's declared explicitly, the same
    /// "must specify" register `NAME` itself already uses.
    fn consumed_event_types() -> Vec<&'static str>;

    /// Whether this projection updates inline, in the same transaction as
    /// the events it consumes (`true`), or via a background consumer
    /// (`false`, the default) - see `Projection.sync` in the spec.
    fn sync() -> bool {
        false
    }

    /// Which instance(s) of this projection `event` updates - see the
    /// instance-data note on `entity Projection` in the spec. Defaults to
    /// one constant, unnamed key (`""`) - the "single shared value" case
    /// every projection built before this pass already is, needing no
    /// change at all to keep working exactly as it already does.
    /// Returning more than one key (a transfer event naming both the
    /// giver's and the receiver's own account id, say) folds this event
    /// into each of those instances independently, via its own call to
    /// `project()` - see that method's own doc comment for the `key`
    /// parameter it's handed each time.
    fn keys(_event: &Self::Event) -> Vec<String> {
        vec![String::new()]
    }

    /// The tag key (one of some consumed event type's own `tag_mappings`
    /// keys) that names this projection's "owner" dimension, if it has
    /// one - `Some("company")` for a projection whose instances belong to
    /// different tenants sharing one bounded context, `None` (the
    /// default) for a projection with no such notion, e.g. `keys()`
    /// already being a person's own id or the projection being unkeyed.
    /// Deliberately a Rust-only implementation detail with no spec entity
    /// field and no registration/admin-visible surface of its own - the
    /// identical treatment `Snapshot::TAG_KEY` already gets, and for the
    /// same reason: `RoleAccessMapping.scope` is what an admin actually
    /// grants and sees, not this.
    ///
    /// When set, each instance's own "owner" value is derived
    /// automatically at fold time from whichever event touched it: the
    /// value of the tag on `event.tags` whose key matches this one, if
    /// present. An event lacking that tag leaves an already-established
    /// owner untouched rather than clearing it. `query_projection`
    /// rejects a `scope`-restricted `RoleAccessMapping` whose `scope`
    /// does not match an instance's own derived owner - see that
    /// function's own doc comment and specs/skilj.allium's
    /// `owner_scope_satisfied`.
    const OWNER_TAG_KEY: Option<&'static str> = None;

    /// The `Role.name` required to query this projection at all, if any -
    /// `None` (the default, and every projection that predates this) for
    /// one with no such notion. Deliberately Rust-only, no spec entity
    /// field and no registration/admin-visible surface, the identical
    /// treatment `OWNER_TAG_KEY` just above gets and for the same reason.
    ///
    /// Unlike `OWNER_TAG_KEY`, which is derived per-instance from folded
    /// event tags, this is a fixed, whole-projection gate: it names no
    /// dimension to look up per instance, just one required team, so
    /// either every instance of this projection is reachable by a
    /// matching Role or none are - never a per-instance answer.
    /// Composes with `OWNER_TAG_KEY` rather than replacing it: a
    /// projection may declare both (staff-only *and* company-scoped),
    /// and `query_projection` checks both independently. See that
    /// function's own doc comment and specs/skilj.allium's
    /// `team_only_satisfied`.
    ///
    /// Genuinely different in kind from `private_fields`' own `Team`
    /// kind, even though the membership test is identical
    /// (`Role.name` equality) - `private_fields` redacts one field of
    /// one raw event/command, for one reader, traced back to who created
    /// that specific record; a projection's stored state is one shared
    /// value folded from possibly many events by `project()`, with no
    /// single record each field individually traces back to, so nothing
    /// short of a whole-instance gate generalizes to it (issue
    /// #17's own finding, closing the gap left by `private_fields`
    /// itself: it protects `queryEvents`/`fetchCommands`, surfaces no
    /// `Write`-level Role can reach anyway, while `ProjectionQuery` - the
    /// one surface a `Write`-level Role *can* reach - stayed exactly as
    /// open as before).
    const TEAM_ONLY: Option<&'static str> = None;

    /// How many partitions this projection's own `keys()` values hash
    /// into for async catch-up (issue #25,
    /// docs/architecture.md §51) - `1` (the default, and every
    /// projection that predates this) means every key hashes to the
    /// same partition, so `catch_up_bounded_context` takes its original,
    /// unpartitioned code path unchanged and order is exactly as it
    /// already was. A value greater than `1` lets multiple skilj
    /// instances each independently claim and fold a different slice of
    /// this projection's own keys concurrently, instead of every
    /// instance redundantly redoing all of it - see
    /// `catch_up_partitioned_projection`'s own doc comment for the
    /// mechanism. Deliberately Rust-only, no spec entity field and no
    /// registration/admin-visible surface, the identical treatment
    /// `OWNER_TAG_KEY`/`TEAM_ONLY` above already get: it's a pure
    /// work-distribution detail, invisible to what `project()` actually
    /// folds, so changing it between deploys needs no rebuild (unlike
    /// `schema`/`consumed_event_types`/`sync`, which change what gets
    /// folded - see `register_projection`'s own rebuild-trigger logic).
    /// Meaningful only for `sync() == false`; inert (never validated
    /// against, never rejected) for a sync one, the same stance
    /// `OWNER_TAG_KEY`/`TEAM_ONLY` already take toward combinations that
    /// don't make sense rather than erroring on them.
    const PARTITION_COUNT: u32 = 1;

    /// `key` is which instance is currently being folded - one of
    /// `Self::keys(event)`'s own return values, handed back so `project()`
    /// can tell them apart when an event touches more than one (compare
    /// `key` against the event's own fields to decide, e.g., whether this
    /// call is crediting or debiting). Ignored entirely by a projection
    /// that never overrides `keys()` - it only ever sees the one constant
    /// `""` instance, the same single fold every existing projection
    /// already does.
    fn project(state: &mut Self::State, event: &Self::Event, key: &str);
}

/// A folded, tag-scoped accelerator for `CommandType::decide()` against
/// a large `matching_events` history ([docs/architecture.md §19](../../../docs/architecture.md#optional-snapshotting-matching-events)'s
/// "Problem 2") - `CommandType::snapshot()`'s own doc comment is where a
/// command type opts in.
///
/// Deliberately **not** `Projection`, even though the shape rhymes
/// (`fold` vs. `project`, `TAG_KEY` vs. `keys()`, `VERSION` vs. a
/// restage): `Projection`s are eventually-consistent, rebuildable/
/// restageable, Admin-exposed read-model infrastructure with no
/// obligation to the timing/correctness bar a DCB consistency check
/// needs - letting `decide()`'s own inputs depend on `Projection`
/// machinery would mean an operator rebuilding a projection for
/// ordinary read-model reasons could silently corrupt what `decide()`
/// sees. `Snapshot` is its own trait, own storage
/// (`{schema}.snapshots`/`{schema}.snapshot_progress`), own background
/// catch-up task, own inspection endpoint - no shared code path a
/// read-model change could destabilise.
///
/// Scoped to one tag key, holding a state per value of it. A command
/// type of any shape uses it through snapshot items in its consistency
/// query (`QueryItemMapping::snapshot`, docs/architecture.md §202): any
/// number of them, over one snapshot or several, next to ordinary items,
/// each handing `decide_with()` the state for one value. An event
/// carrying several values of the key - a transfer between two accounts -
/// is folded into each value's state, through `fold_for`, which is told
/// the value.
///
/// One `Snapshot` definition is shared by every `CommandType` that tags
/// on its `TAG_KEY` and opts in - `banking.rs`'s `DepositMoney`/
/// `WithdrawMoney` both tag on `account` and both want the same running
/// balance, so they share one `AccountBalanceSnapshot` rather than each
/// computing their own, free to drift apart.
pub trait Snapshot {
    /// The state `fold()` builds per tag value, stored as JSON. `Default` is
    /// the state before any event.
    type State: Serialize + DeserializeOwned + JsonSchema + Default;

    /// Same generated per-bounded-context event enum `CommandType::Event`/
    /// `Projection::Event` use - a snapshot folds any event type
    /// carrying its `TAG_KEY`, not just one.
    type Event: BoundedContextEvent;

    /// The snapshot's name, unique within its bounded context - what
    /// `SnapshotState::snapshot` and `Decision::snapshot` name it by.
    const NAME: &'static str;

    /// See `CommandType::BOUNDED_CONTEXT`'s own doc comment - identical
    /// role and default here, for `skilj::SkiljBuilder::auto_register()`'s
    /// `Snapshot` side.
    const BOUNDED_CONTEXT: &'static str = DEFAULT_BOUNDED_CONTEXT;

    /// The single tag key this snapshot is scoped to - see this trait's
    /// own doc comment for why only one. The background catch-up task
    /// derives which tag *value* an event belongs to generically, from
    /// `event.tags`, so there's no `keys()`-equivalent method to
    /// implement here the way `Projection` needs one.
    const TAG_KEY: &'static str;

    /// Which tag key names this snapshot's "owner" dimension, if it has
    /// one and it differs from `TAG_KEY` itself - `None` (the default)
    /// for a snapshot with no such notion. Deliberately Rust-only, no
    /// spec entity field and no registration surface, the identical
    /// treatment `TAG_KEY` itself already gets (`Snapshot` is compiled,
    /// deployed configuration throughout - see this trait's own doc
    /// comment). When set, each stored row's own owner value is derived
    /// at fold time from whichever event touched it: the value of the
    /// tag on `event.tags` whose key matches this one, if present - not
    /// necessarily the same tag `TAG_KEY` itself reads (a snapshot keyed
    /// by `"account"` might still need owner-scoping by `"company"`).
    /// `inspectSnapshot` rejects a `scope`-restricted `RoleAccessMapping`
    /// whose `scope` does not match a stored row's own derived owner -
    /// see `access_control::scope_matches_owner` and
    /// `snapshot_query::inspect_snapshot_field`'s own doc comment. Cross-
    /// tenant read fix (docs/architecture.md's own write-up of these
    /// passes).
    const OWNER_TAG_KEY: Option<&'static str> = None;

    /// Bumped by hand whenever `fold()`'s own logic or `State`'s shape
    /// changes - a stored row at an older version is treated as if it
    /// doesn't exist (see `CommandType::decide_from_snapshot`'s own doc
    /// comment), never trusted. Not inferred: Rust has no way to detect
    /// a fold's own semantic change, only a decider declaring one can.
    const VERSION: u64;

    /// How many partitions this snapshot's own tag values hash into for
    /// catch-up (issue #25, docs/architecture.md §52 - the
    /// `Snapshot` twin of `Projection::PARTITION_COUNT`, see that
    /// const's own doc comment for the full reasoning, identical here
    /// with the derived tag *value* playing the role `Projection`'s own
    /// `keys()` return values play). `1` (the default) means every tag
    /// value hashes to the same partition, so `catch_up_snapshots` takes
    /// its original, unpartitioned code path unchanged.
    const PARTITION_COUNT: u32 = 1;

    /// Folds one event into `state`, in place - the identical shape
    /// `Projection::project` already has, minus the `key` parameter.
    /// Implement this, or [`fold_for`](Snapshot::fold_for) when the fold
    /// depends on which tag value it's folding for; the framework only
    /// calls `fold_for`, whose default calls this. The default here
    /// changes nothing.
    fn fold(_state: &mut Self::State, _event: &Self::Event) {}

    /// Folds `event` into the state for `tag_value`, one of the values of
    /// `TAG_KEY` the event carries (docs/architecture.md §202). An event
    /// carrying several - a transfer tagged with both accounts - is
    /// folded into each value's state, once per value, so a fold that
    /// treats them differently (a debit for one, a credit for the other)
    /// tells them apart by `tag_value`. The default calls `fold()`.
    fn fold_for(state: &mut Self::State, tag_value: &str, event: &Self::Event) {
        let _ = tag_value;
        Self::fold(state, event)
    }
}

/// One blind-index constraint (docs/architecture.md §205): the tag key
/// whose values are stored as a keyed hash of a payload field instead of
/// the field, so a command can check that no earlier event carries the
/// same value - an email address registered once - without the value
/// ever being stored in the clear. A type tags it with
/// [`TagMapping::blind::<Self>`](crate::shared::TagMapping::blind), which
/// may name a sensitive field; a command's consistency query reads it
/// like any other tag.
///
/// ```
/// use skilj_core::blind::normalize;
/// use skilj_core::plugin::{BlindErasure, BlindTag};
///
/// struct CustomerEmail;
/// impl BlindTag for CustomerEmail {
///     const BOUNDED_CONTEXT: &'static str = "crm";
///     const TAG_KEY: &'static str = "email";
///     const ERASURE: BlindErasure = BlindErasure::with_subject("customer");
///     fn normalize(value: &str) -> String {
///         normalize::email(value)
///     }
/// }
/// ```
///
/// Every instance serving the bounded context must declare it, with the
/// same `blind_index_key`: an instance that doesn't refuses to derive the
/// tag (`blind_tag_not_declared`) rather than store the value.
pub trait BlindTag {
    /// See `CommandType::BOUNDED_CONTEXT`'s own doc comment.
    const BOUNDED_CONTEXT: &'static str = DEFAULT_BOUNDED_CONTEXT;

    /// The tag key. One `BlindTag` per key per bounded context; the key's
    /// constraint is its own HMAC key, derived from the
    /// `blind_index_key` and `(BOUNDED_CONTEXT, TAG_KEY)`.
    const TAG_KEY: &'static str;

    /// Bumped by hand whenever [`normalize`](BlindTag::normalize)
    /// changes what it returns for any input. Values hashed under the old
    /// normalizer don't match new ones, so `build()` refuses a version
    /// that differs from the stored one rather than silently letting a
    /// taken value be taken again.
    const NORMALIZER_VERSION: u32 = 1;

    /// What `forgetSubject` does with the tag - see [`BlindErasure`].
    const ERASURE: BlindErasure;

    /// What a payload value becomes before it's hashed, so values that
    /// mean the same thing collide: [`crate::blind::normalize::email`]
    /// trims and lowercases. Must be deterministic. It can also add an
    /// application secret of its own - a pepper the database never sees.
    /// The default leaves the value as it is.
    fn normalize(value: &str) -> String {
        value.to_string()
    }
}

/// What `forgetSubject` does with a [`BlindTag`]'s hashes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum BlindErasure {
    /// The hash stays: the value is still taken after the subject is
    /// forgotten. For a rule that outlives the person - a trial used once,
    /// an address that was banned.
    Keep,
    /// Forgetting `subject_key`'s subject removes the tag from every event
    /// and command whose `subject_tag` tag carries the subject's value, so
    /// the value can be taken again and no hash of it is left. Every type
    /// tagging the blind key must also tag `subject_tag`; `build()`
    /// checks.
    WithSubject {
        /// The `SensitiveField` subject key whose forgetting removes the tag.
        subject_key: &'static str,
        /// The plain tag carrying the subject's value on every event and
        /// command that carries the blind tag.
        subject_tag: &'static str,
    },
}

impl BlindErasure {
    /// [`BlindErasure::WithSubject`] where the encryption subject key and
    /// the plain tag carrying the subject's value share one name - the
    /// usual case, e.g. `SensitiveField::new("email", "customer",
    /// "customer_id")` next to `TagMapping::new("customer", "customer_id")`.
    pub const fn with_subject(subject: &'static str) -> Self {
        BlindErasure::WithSubject {
            subject_key: subject,
            subject_tag: subject,
        }
    }
}

impl From<BlindErasure> for crate::blind::Erasure {
    fn from(erasure: BlindErasure) -> Self {
        match erasure {
            BlindErasure::Keep => crate::blind::Erasure::Keep,
            BlindErasure::WithSubject {
                subject_key,
                subject_tag,
            } => crate::blind::Erasure::WithSubject {
                subject_key: subject_key.to_string(),
                subject_tag: subject_tag.to_string(),
            },
        }
    }
}

/// One snapshot's state for one tag value, as a decision is handed it
/// (docs/architecture.md §202): the stored row with every later event
/// carrying the value folded in, JSON-encoded. Read typed through
/// [`Decision::snapshot`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SnapshotState {
    /// `Snapshot::NAME`.
    pub snapshot: String,
    /// The tag value this state was folded for.
    pub tag_value: String,
    /// The folded state, JSON-encoded.
    pub state_json: String,
    /// The position `state_json` holds every event carrying the value
    /// through: the stored row's `as_of_sequence`, or the last event
    /// folded on top of it; -1 for none.
    pub folded_through: i64,
}

/// The time a decision is made against, and, for a command a deadline
/// fired, that deadline's own schedule and expiry (issue #97).
///
/// `decide()` never reads a clock itself (`CommandType`'s own doc comment,
/// §1.1): the framework stamps every submission with the `now` it uses for
/// the command's own `created_at`, its idempotency row and the events it
/// appends, and hands that same instant here. A decision therefore stays
/// deterministic and replayable, and a test can drive it with any `now`.
/// `decide_with()` reads it through [`Decision::context`] (`Decision::now`
/// and `Decision::deadline` are the two shorthands); the legacy
/// `decide_from_snapshot` is handed it directly.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecisionContext {
    /// The instant the submission is stamped with - the same `now` the
    /// command's own `created_at` and idempotency row carry.
    pub now: chrono::DateTime<chrono::Utc>,
    /// Present only when this submission is a deadline firing
    /// (`db::fire_due_deadlines`); `None` for every ordinary submission.
    pub deadline: Option<DeadlineContext>,
}

impl DecisionContext {
    /// An ordinary submission's context: `now`, no deadline.
    pub fn at(now: chrono::DateTime<chrono::Utc>) -> Self {
        Self {
            now,
            deadline: None,
        }
    }

    /// A deadline firing's context (issue #97): `scheduled_at` is
    /// the fired row's own `fire_at`, `deadline_at` its optional
    /// `expires_at`.
    pub fn for_deadline(
        now: chrono::DateTime<chrono::Utc>,
        scheduled_at: chrono::DateTime<chrono::Utc>,
        deadline_at: Option<chrono::DateTime<chrono::Utc>>,
    ) -> Self {
        Self {
            now,
            deadline: Some(DeadlineContext::new(scheduled_at, deadline_at, now)),
        }
    }
}

/// What a deadline-fired command's `decide()` may reason about (
/// issue #97): `scheduled_at` is the deadline's own `fire_at` (when the
/// chain intended it to fire), `deadline_at` its optional `expires_at`,
/// and `expired` whether this firing happened after `deadline_at`. A
/// target command can refuse an action it fired too late for, and can tell
/// a genuine firing from a redrive or a retry by comparing
/// `DecisionContext::now` with `scheduled_at`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeadlineContext {
    /// The fired row's own `fire_at` - when the deadline was scheduled to
    /// fire.
    pub scheduled_at: chrono::DateTime<chrono::Utc>,
    /// The fired row's own `expires_at`, if it had one.
    pub deadline_at: Option<chrono::DateTime<chrono::Utc>>,
    /// `deadline_at` is `Some` and `now` is past it. Always `false` when
    /// there is no expiry at all.
    pub expired: bool,
}

impl DeadlineContext {
    /// A deadline's context: `expired` is whether `now` is past `deadline_at`.
    pub fn new(
        scheduled_at: chrono::DateTime<chrono::Utc>,
        deadline_at: Option<chrono::DateTime<chrono::Utc>>,
        now: chrono::DateTime<chrono::Utc>,
    ) -> Self {
        Self {
            scheduled_at,
            deadline_at,
            expired: deadline_at.is_some_and(|at| now > at),
        }
    }
}

/// What `CommandType::decide_with` decides from (docs/architecture.md
/// §202): the events the command's query selects, the state of every
/// snapshot item it derives, and the [`DecisionContext`] the decision is
/// made against (issue #97). Built by the framework; room to grow
/// without changing `decide_with`'s signature.
pub struct Decision<'a, E> {
    events: &'a [E],
    snapshots: &'a [SnapshotState],
    context: &'a DecisionContext,
}

impl<'a, E> Decision<'a, E> {
    /// Built by the framework for each decision.
    pub fn new(
        events: &'a [E],
        snapshots: &'a [SnapshotState],
        context: &'a DecisionContext,
    ) -> Self {
        Self {
            events,
            snapshots,
            context,
        }
    }

    /// The events the query's items other than snapshot items select, in
    /// sequence order, each once - `decide()`'s `matching_events`.
    pub fn events(&self) -> &'a [E] {
        self.events
    }

    /// Snapshot `S`'s state for `tag_value`, as of the command's read,
    /// with every event carrying the value folded in. `None` when no item
    /// of the query derived that value for `S`: a decision should treat
    /// that as a misconfiguration rather than an empty state.
    pub fn snapshot<S: Snapshot>(&self, tag_value: &str) -> Option<S::State> {
        self.snapshots
            .iter()
            .find(|s| s.snapshot == S::NAME && s.tag_value == tag_value)
            .and_then(|s| serde_json::from_str(&s.state_json).ok())
    }

    /// The time this decision is made against, and, for a deadline firing,
    /// the deadline's own context (issue #97).
    pub fn context(&self) -> &'a DecisionContext {
        self.context
    }

    /// Shorthand for `self.context().now` - the instant the submission is
    /// stamped with, so `decide()` can reason about time without reading a
    /// clock.
    pub fn now(&self) -> chrono::DateTime<chrono::Utc> {
        self.context.now
    }

    /// Shorthand for `self.context().deadline`: the deadline that fired
    /// this command, when one did (issue #97).
    pub fn deadline(&self) -> Option<&'a DeadlineContext> {
        self.context.deadline.as_ref()
    }
}

/// Type-erased dispatch to a bounded context's own typed `decide()` -
/// what `CommandTrigger`'s REST handler and GraphQL's `CommandSubmission`
/// resolver both call through to actually process a command. Lives here,
/// as a trait, rather than `skilj-rest`/
/// `skilj-graphql` reaching into whichever crate happens to build the
/// concrete registry (`skilj`'s `SkiljBuilder`, today): both surface
/// crates already depend on `skilj-core`, so this is the natural shared
/// boundary - the same "opaque handle both sides can reach" reasoning
/// `db::Pool` already uses. See docs/architecture.md §1.7/[§8](../../../docs/architecture.md#open-for-a-future-pass) item 4.
///
/// A consumer using `skilj-core` + `skilj-rest` directly, without the
/// `skilj` facade's builder, implements this by hand instead - nothing
/// about it is facade-specific.
pub trait CommandDispatcher: Send + Sync {
    /// `None` when no `(bounded_context, command_type)` pair matches
    /// anything registered - the caller decides what that means for its
    /// own wire contract (an unrecognised-command rejection, a 404,
    /// whatever fits), since neither existing wire contract has needed
    /// to yet and this trait shouldn't invent one on their behalf.
    fn dispatch(
        &self,
        bounded_context: &str,
        command_type: &str,
        payload: &str,
        matching_events: &[Event],
    ) -> Option<crate::error::Result<CommandDecision>>;

    /// `dispatch`, with the states of the query's snapshot items
    /// (docs/architecture.md §202) and the [`DecisionContext`] the
    /// decision is made against (issue #97) - what the command
    /// path calls. The default ignores `snapshots` and `context`, calling
    /// `dispatch`, right for a dispatcher whose command types declare no
    /// snapshot items and don't read time.
    fn dispatch_with(
        &self,
        bounded_context: &str,
        command_type: &str,
        payload: &str,
        matching_events: &[Event],
        snapshots: &[SnapshotState],
        context: &DecisionContext,
    ) -> Option<crate::error::Result<CommandDecision>> {
        let _ = (snapshots, context);
        self.dispatch(bounded_context, command_type, payload, matching_events)
    }

    /// The snapshots this dispatcher's command types name in
    /// [`QueryItemMapping::snapshot`] items (docs/architecture.md §202):
    /// what the command path reads their versions, defaults and folds
    /// from, before the lock and under it. `None`, the default, for a
    /// dispatcher whose command types declare none; a command whose query
    /// has one is then refused as a configuration error.
    fn snapshots(&self) -> Option<&dyn SnapshotDispatcher> {
        None
    }

    /// The registered command type's own `CommandType::required_role()` -
    /// see that method's own doc comment. Outer `None` mirrors
    /// `dispatch`'s "pair isn't registered at all" convention; inner
    /// `None` is a registered command type declaring no extra gate (the
    /// default). Meant to be checked by `skilj-graphql`'s mutation
    /// resolver *before* calling `dispatch` at all, so an unauthorised
    /// caller never reaches `decide()` - not folded into `dispatch`
    /// itself, since REST triggering never needs this check (`CommandToken`
    /// is its own, separate per-token grant) and shouldn't pay for it.
    fn required_role(
        &self,
        bounded_context: &str,
        command_type: &str,
    ) -> Option<Option<&'static str>>;

    /// The registered command type's own `CommandType::consistency_query()`
    /// (docs/architecture.md §198) - `None` when the pair isn't
    /// registered, or for a dispatcher that predates queries; either reads
    /// as the default query.
    fn consistency_query(
        &self,
        _bounded_context: &str,
        _command_type: &str,
    ) -> Option<Vec<QueryItemMapping>> {
        None
    }

    /// The registered command type's own `CommandType::snapshot()` -
    /// same outer/inner `Option` convention as `required_role`. Meant to
    /// be checked before `dispatch`, the same way `required_role` is:
    /// if this is `Some(Some(name))` and the command's own derived tags
    /// match `SnapshotDispatcher::tag_key(bc, name)` exactly, the caller
    /// should read that snapshot and call `dispatch_from_snapshot`
    /// instead of `dispatch` - see [docs/architecture.md §19](../../../docs/architecture.md#optional-snapshotting-matching-events).
    fn snapshot_name(
        &self,
        bounded_context: &str,
        command_type: &str,
    ) -> Option<Option<&'static str>>;

    /// `CommandType::decide_from_snapshot`'s own type-erased entry
    /// point - `dispatch`'s counterpart for the snapshot-accelerated
    /// path. `None` for the identical "pair isn't registered at all"
    /// case `dispatch` already has.
    fn dispatch_from_snapshot(
        &self,
        bounded_context: &str,
        command_type: &str,
        payload: &str,
        snapshot_state_json: &str,
        events_since_snapshot: &[Event],
        context: &DecisionContext,
    ) -> Option<crate::error::Result<CommandDecision>>;

    /// The blind tags this process declares, keyed for hashing
    /// (docs/architecture.md §205) - what every tag derivation on the
    /// command path, and event creation through the same surfaces, hashes
    /// a blind mapping with. Empty, the default, for a dispatcher whose
    /// bounded contexts declare none: a stored blind mapping is then
    /// refused with `blind_tag_not_declared`, never derived in the clear.
    fn blind_index(&self) -> crate::blind::BlindIndex {
        crate::blind::BlindIndex::default()
    }
}

/// Type-erased dispatch to a bounded context's own typed `project()` -
/// `CommandDispatcher`'s own counterpart for the other black box §1.1
/// draws the pure-function line around. What the new transactional
/// `db::insert_event_and_update_sync_projections` calls through to
/// actually fold an event into a `sync` projection's stored state - see
/// docs/architecture.md's own write-up of this pass ([§8](../../../docs/architecture.md#open-for-a-future-pass) item 6).
///
/// Async (`sync: false`, the default) projections are folded by the
/// background consumer `db::catch_up_bounded_context` - see its own doc
/// comment and docs/architecture.md's write-up of that pass. Same
/// `project()` call either way; the consumer doesn't know or care whether
/// its caller is a request (the sync path) or a poll tick (this one).
pub trait ProjectionDispatcher: Send + Sync {
    /// `None` when no `(bounded_context, projection_name)` pair matches
    /// anything registered - the same "pair isn't registered at all"
    /// convention `dispatch`/`project` below already use. `Some(vec![])`
    /// when the pair *is* registered but `event`'s own type isn't one
    /// this projection actually consumes (`Projection::keys` is never
    /// called for it) - `caught_up_to` still advances, no instance is
    /// touched, the identical "position always advances, state only
    /// changes when consumed" treatment this trait's own `project`
    /// already documents. Otherwise, `Projection::keys(event)` verbatim -
    /// every instance this one event updates, one call to `project`
    /// below per key.
    fn keys(
        &self,
        bounded_context: &str,
        projection_name: &str,
        event: &Event,
    ) -> Option<Vec<String>>;

    /// `None` when no `(bounded_context, projection_name)` pair matches
    /// anything registered - the same "pair isn't registered at all"
    /// convention `CommandDispatcher::dispatch` uses. `Some(Ok(new_state))`
    /// is the projection's `state_json` after folding `event` into it,
    /// for the one instance named by `key` (one of `keys`'s own return
    /// values) - `BoundedContextEvent::try_from_event` succeeding is not
    /// by itself permission to call `project()`; the caller only reaches
    /// this once per key `keys` returned.
    fn project(
        &self,
        bounded_context: &str,
        projection_name: &str,
        state_json: &str,
        event: &Event,
        key: &str,
    ) -> Option<crate::error::Result<String>>;

    /// `T::State::default()`, JSON-serialised - `None` for the identical
    /// "pair isn't registered at all" case `project`/`dispatch` already
    /// have. The only source `db::catch_up_bounded_context` has for a
    /// `ProjectionRebuild`'s own starting state when it begins (or
    /// resumes after a restage) folding one: a rebuild's `caught_up_to`
    /// resets to `None` on every `db::upsert_projection_rebuild` call,
    /// including a restage of an already-`building` row (see that
    /// function's own doc comment), so whatever state a previous attempt
    /// had accumulated is stale the moment that happens - only the
    /// dispatcher, not whichever caller triggered the reset, knows the
    /// *current* correct starting point.
    fn default_state(&self, bounded_context: &str, projection_name: &str) -> Option<String>;

    /// The registered projection's own `Projection::OWNER_TAG_KEY` -
    /// `CommandDispatcher::snapshot_name`'s identical `Option<Option<_>>`
    /// shape: outer `None` for the same "pair isn't registered at all"
    /// case every method here already has, inner `None` when the
    /// projection is registered but declares no owner dimension (the
    /// default, and every projection that predates this).
    fn owner_tag_key(
        &self,
        bounded_context: &str,
        projection_name: &str,
    ) -> Option<Option<&'static str>>;

    /// The registered projection's own `Projection::TEAM_ONLY` - the
    /// identical `Option<Option<_>>` shape `owner_tag_key` just above
    /// already has, for the same reason. Defaulted to `None` (unlike
    /// `owner_tag_key`, which has no default) - every call site here
    /// only ever `.flatten()`s the result, for which the outer "pair
    /// isn't registered" `None` and the inner "registered, no team
    /// declared" `Some(None)` already read identically, so a dispatcher
    /// with nothing team-gated needs no explicit override at all.
    fn team_only(
        &self,
        _bounded_context: &str,
        _projection_name: &str,
    ) -> Option<Option<&'static str>> {
        None
    }

    /// The registered projection's own `Projection::PARTITION_COUNT`
    /// (issue #25, docs/architecture.md §51) - `None` for the
    /// same "pair isn't registered at all" case every method here
    /// already has, treated by every real caller (`db::
    /// catch_up_bounded_context`) identically to `Some(1)`: unpartitioned.
    /// Defaulted (unlike `owner_tag_key`, which has no default) for the
    /// same reason `team_only` is - a dispatcher with nothing partitioned
    /// needs no explicit override at all.
    fn partition_count(&self, _bounded_context: &str, _projection_name: &str) -> Option<u32> {
        None
    }
}

/// Type-erased dispatch to a bounded context's own typed `Snapshot::fold` -
/// `CommandDispatcher`'s `dispatch_from_snapshot`/`snapshot_name`'s own
/// counterpart for the other half of the type-erasure boundary, and what
/// `db::catch_up_snapshots` (the background catch-up task -
/// [docs/architecture.md §19](../../../docs/architecture.md#optional-snapshotting-matching-events)) calls through to actually fold an event
/// into a stored snapshot row. Same "outer `None` = not registered"
/// convention every other dispatcher trait in this module already uses.
pub trait SnapshotDispatcher: Send + Sync {
    /// Every `Snapshot::NAME` registered for `bounded_context` in this
    /// process - `db::catch_up_snapshots`' own discovery mechanism,
    /// since (deliberately - see `Snapshot`'s own doc comment) there is
    /// no `snapshots` metadata table to enumerate the way
    /// `list_projections_for_bounded_context` does for `Projection`.
    fn snapshot_names(&self, bounded_context: &str) -> Vec<&'static str>;

    /// The registered snapshot's own `Snapshot::TAG_KEY`.
    fn tag_key(&self, bounded_context: &str, snapshot_name: &str) -> Option<&'static str>;

    /// The registered snapshot's own `Snapshot::OWNER_TAG_KEY` -
    /// `ProjectionDispatcher::owner_tag_key`'s identical `Option<Option<_>>`
    /// shape: outer `None` for the same "pair isn't registered at all"
    /// case every method here already has, inner `None` when the
    /// snapshot is registered but declares no owner dimension (the
    /// default, and every snapshot that predates this).
    fn owner_tag_key(
        &self,
        bounded_context: &str,
        snapshot_name: &str,
    ) -> Option<Option<&'static str>>;

    /// The registered snapshot's own `Snapshot::VERSION` - compared
    /// against a stored row's `snapshot_version` to decide whether it's
    /// still trustworthy (see `CommandType::decide_from_snapshot`'s own
    /// doc comment).
    fn version(&self, bounded_context: &str, snapshot_name: &str) -> Option<u64>;

    /// `Some(Ok(new_state))` is `state_json` with `event` folded into it
    /// via `Snapshot::fold` - `None` for the identical "pair isn't
    /// registered at all" case every dispatcher here already has.
    /// `event`'s own `BoundedContextEvent::try_from_event` failing is a
    /// real `Err`, not a silent skip - a stored event this snapshot's
    /// own `TAG_KEY` names should always convert cleanly.
    fn fold(
        &self,
        bounded_context: &str,
        snapshot_name: &str,
        state_json: &str,
        event: &Event,
    ) -> Option<crate::error::Result<String>>;

    /// `fold`, for one of the values of the snapshot's tag key `event`
    /// carries: `Snapshot::fold_for` (docs/architecture.md §202). What
    /// catch-up and the command path call. The default ignores
    /// `tag_value` and calls `fold`.
    fn fold_value(
        &self,
        bounded_context: &str,
        snapshot_name: &str,
        tag_value: &str,
        state_json: &str,
        event: &Event,
    ) -> Option<crate::error::Result<String>> {
        let _ = tag_value;
        self.fold(bounded_context, snapshot_name, state_json, event)
    }

    /// `T::State::default()`, JSON-serialised - the starting point for a
    /// tag value with no stored snapshot yet, or one whose stored
    /// `snapshot_version` no longer matches. `None` for the identical
    /// "pair isn't registered at all" case.
    fn default_state(&self, bounded_context: &str, snapshot_name: &str) -> Option<String>;

    /// The registered snapshot's own `Snapshot::PARTITION_COUNT`
    /// (issue #25, docs/architecture.md §52) -
    /// `ProjectionDispatcher::partition_count`'s identical shape and
    /// defaulting: `None` for the same "pair isn't registered at all"
    /// case every method here already has, treated by every real caller
    /// (`db::catch_up_snapshots`) identically to `Some(1)`: unpartitioned.
    fn partition_count(&self, _bounded_context: &str, _snapshot_name: &str) -> Option<u32> {
        None
    }
}

/// A single-hop, stateless reaction: when `Source` commits in its own
/// bounded context, submit `Target`'s own command into *its* bounded
/// context - the answer to "make messages cross bounded contexts
/// without needing an external system like Temporal" (docs/architecture.md's
/// own write-up of this pass) for the case that never needed Temporal's
/// own durable-timer/retry/compensation machinery in the first place:
/// both bounded contexts already live in the same Postgres database,
/// often the same process, so relaying one committed fact into another
/// bounded context's own command needs nothing more than a durable
/// cursor and the same `submit_command` path every other caller already
/// goes through.
///
/// Deliberately **not** a Saga/process manager - see this trait's own
/// design note in docs/architecture.md: no multi-step state is tracked
/// across calls, no compensation, no retry beyond `submit_command`'s own
/// idempotency-key mechanism. One commit causes at most one more commit
/// elsewhere, the same bounded, single-hop shape `submit_command`'s own
/// DCB-conflict retry already has - not an open-ended state machine.
///
/// A `Target` command that gets rejected by its own `decide()` is a
/// legitimate business outcome here too (§5.4's own "a business
/// rejection is not an error" register), not a failure to retry -
/// `route()` translating `Source`'s payload into a `Target` payload
/// `Target::decide()` goes on to reject is exactly as valid an outcome
/// as one it accepts.
pub trait CrossContextRoute {
    /// The `EventType` this route reacts to, in its own bounded context.
    type Source: EventType;
    /// The `CommandType` this route submits, in *its* bounded context -
    /// deliberately a command, not a raw event: `Target`'s own
    /// `decide()` still gets to accept, reject, or reshape the outcome,
    /// the same as every other command submission. A route that skipped
    /// straight to creating an event would bypass whatever domain rules
    /// the target bounded context wants to enforce on its own history.
    type Target: CommandType;

    /// This route's own stable identity - names its durable cursor row
    /// (`{Source::BOUNDED_CONTEXT}.cross_context_route_cursors`) and the
    /// idempotency key derived for every command it submits
    /// (`"{NAME}:{source_event_sequence}"`), so a redelivered/retried
    /// catch-up tick is exactly as safe as any other idempotency-keyed
    /// submission already is.
    const NAME: &'static str;

    /// Where this route's own durable cursor starts the first time it is
    /// ever registered - `Beginning` (the default) replays every
    /// `Source` occurrence ever committed, exactly what every route
    /// registered before this const existed already does. A route whose
    /// job is a side effect that must never re-fire for old history (the
    /// canonical case: `UserRegistered -> SendWelcomeEmail`) overrides
    /// this to `Latest`, the `CrossContextRoute`-side counterpart to
    /// `EventReadStartPosition::Latest` on the spec-backed `EventReadToken`
    /// track (docs/architecture.md's own write-up of this pass) - a
    /// separate Rust-only type rather than reusing that one, since a
    /// route has no spec entity of its own to hang a shared type off of
    /// (§36's own "no spec entity" note). See
    /// `db::catch_up_cross_context_route`'s own doc comment for exactly
    /// when this is read - only on this route's first-ever catch-up
    /// tick, never again once a cursor row exists.
    const START_FROM: CrossContextRouteStartFrom = CrossContextRouteStartFrom::Beginning;

    /// `None` skips this occurrence of `Source` entirely - no command
    /// submitted, cursor still advances (retrying can never produce a
    /// target payload for an occurrence this route itself decided
    /// doesn't apply). `Some(payload)` submits `Target` with that
    /// payload, going through the identical `decide()`/`submit_command`
    /// path every other caller of `Target` already does.
    fn route(
        source_payload: &<Self::Source as EventType>::Payload,
    ) -> Option<<Self::Target as CommandType>::Payload>;
}

/// See `CrossContextRoute::START_FROM`. Deliberately not
/// `#[derive(Default)]`: `Beginning` is the trait const's own default,
/// spelled out explicitly at every use site (`CrossContextRouteInfo::
/// start_from`) rather than leaned on implicitly, so a reader never has
/// to check whether "the default" here means this type's `Default` impl
/// or the trait const's - there is only ever one to check.
///
/// `AtSequence`/`AtTime` are this trait's own counterpart to
/// `EventReadStartPosition`'s identically-named extension
/// (docs/architecture.md's own write-up of this pass) - a route's first
/// tick seeds its cursor at a caller-chosen sequence, or at the highest
/// sequence already committed at or before a given point in time,
/// exactly like an `EventReadToken` minted the same way would. `AtTime`
/// carries a plain Unix timestamp (seconds, UTC) rather than
/// `chrono::DateTime<Utc>` for one structural reason: `START_FROM` is a
/// trait *associated const*, so every value it can ever hold has to be
/// const-evaluable at the implementing type's own definition site, and
/// `DateTime<Utc>` has no `const fn` constructor to build one from - an
/// `i64` literal is. Converted to a real `DateTime<Utc>` only where it's
/// actually compared against event timestamps
/// (`db::catch_up_cross_context_route`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum CrossContextRouteStartFrom {
    /// From the first event in the source bounded context.
    Beginning,
    /// From the events committed after the route first starts; earlier ones
    /// are skipped.
    Latest,
    /// From the event after this sequence number.
    AtSequence(i64),
    /// Unix timestamp, seconds, UTC - see this enum's own doc comment
    /// for why not `DateTime<Utc>`.
    AtTime(i64),
}

/// One registered route's own static identity - `CrossContextRouteDispatcher::routes()`'s
/// own element type, enumerated once at startup by the background
/// catch-up loop (`db::catch_up_cross_context_route`).
#[derive(Debug, Clone, Copy)]
pub struct CrossContextRouteInfo {
    /// `CrossContextRoute::NAME`: the route's identity and the name of its
    /// cursor.
    pub name: &'static str,
    /// The bounded context the source event type lives in.
    pub source_bounded_context: &'static str,
    /// The event type whose events the route reacts to.
    pub source_event_type: &'static str,
    /// The bounded context the target command type lives in.
    pub target_bounded_context: &'static str,
    /// The command type the route submits.
    pub target_command_type: &'static str,
    /// `CrossContextRoute::START_FROM`, carried through so
    /// `db::catch_up_cross_context_route` can act on it without needing
    /// the route's own concrete type.
    pub start_from: CrossContextRouteStartFrom,
}

/// Type-erased dispatch to a bounded context's own typed
/// `CrossContextRoute::route` - the same "outer `None` = not registered
/// at all" convention every other dispatcher trait in this module
/// already uses, one level deeper here since `route()` itself can also
/// legitimately produce nothing for a given occurrence.
pub trait CrossContextRouteDispatcher: Send + Sync {
    /// Every registered route, across every bounded context - routes
    /// aren't scoped to one bounded context the way `EventType`/
    /// `CommandType`/`Projection` are (a route by definition spans two),
    /// so unlike `ProjectionDispatcher::keys` and friends there is no
    /// `bounded_context` parameter to filter by here; the background
    /// catch-up loop calls this once and iterates the full list itself.
    fn routes(&self) -> Vec<CrossContextRouteInfo>;

    /// `route_name` names one of `routes()`'s own entries;
    /// `source_payload_json` is the triggering `Source` event's own
    /// stored payload. Outer `None` - `route_name` isn't registered at
    /// all (defensive; the caller only ever names one of its own
    /// `routes()`). `Some(Err(e))` - the stored payload didn't
    /// deserialize into `Source::Payload` (see `BoundedContextEvent::
    /// try_from_event`'s own doc comment on why this is still reachable
    /// even though payloads are schema-checked at write time). `Some(Ok(None))` -
    /// `route()` itself decided this occurrence doesn't apply. `Some(Ok(Some(payload)))` -
    /// the `Target` command's own JSON payload, ready to submit.
    fn route(
        &self,
        route_name: &str,
        source_payload_json: &str,
    ) -> Option<Result<Option<String>, serde_json::Error>>;
}

/// issue #20's own "where does this cursor start" const, one per
/// `ScheduleDeadline`/`CancelDeadline` implementor - identical in shape to
/// `CrossContextRouteStartFrom` but its own type rather than a reuse of
/// it: a deadline reactor isn't a route, and giving it a name that says
/// what it's actually for beats leaning on a type whose own doc comment
/// is written entirely in terms of routes. See `CrossContextRouteStartFrom`'s
/// own doc comment for why `AtTime` carries a plain Unix timestamp rather
/// than `DateTime<Utc>` - the identical reasoning (a trait associated
/// const has to be const-evaluable) applies here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum DeadlinePollStartFrom {
    /// From the first event in the source bounded context.
    Beginning,
    /// From the events committed after the reactor first starts; earlier ones
    /// are skipped.
    Latest,
    /// From the event after this sequence number.
    AtSequence(i64),
    /// Unix timestamp, seconds, UTC.
    AtTime(i64),
}
/// Missed-run policy for recurring deadlines (issue #95).
///
/// Mirrors the global scheduler's own `MissedOccurrencePolicy`
/// (`ReplayBacklog`/`FireOnce`/`Skip`) but is deliberately its own type:
/// a global schedule and a per-entity deadline are different reactor
/// families (`SkipMissedOccurrences` skips on the shared `event_types`
/// position; this one advances a private `deadlines` chain), and a name
/// saying what it's for beats leaning on a type whose own doc comment is
/// written entirely in terms of `event_types`. Same ordering the global
/// one uses: `Skip` never replays, `FireOnce` collapses a backlog to one
/// firing, `FireAll` steps through every missed occurrence.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum MissedRunPolicy {
    /// Missed occurrences are not fired; the chain resumes at the next
    /// occurrence still ahead.
    Skip,
    /// A backlog of missed occurrences fires once, then the chain resumes
    /// ahead.
    FireOnce,
    /// Every missed occurrence fires, in order.
    FireAll,
}

/// Recurrence specification for per-entity deadlines (issue
/// #95). `schedule` is the same 7-field Quartz-dialect cron expression
/// (`cron` crate) the global scheduler evaluates, so one mental model
/// covers both surfaces. `start_at`/`end_at` bound the occurrences
/// themselves; `valid_until` is the kill-switch - a pending row past it
/// never fires and begets nothing, even if an occurrence would still be
/// due. All three are optional: absent means unbounded that way.
#[derive(Debug, Clone)]
pub struct RecurrenceSpec {
    /// A 7-field Quartz-dialect cron expression (`cron` crate) the occurrences
    /// follow.
    pub schedule: String,
    /// What happens to occurrences missed while nothing could fire them.
    pub policy: MissedRunPolicy,
    /// No occurrence before this instant; `None` for no lower bound.
    pub start_at: Option<chrono::DateTime<chrono::Utc>>,
    /// No occurrence after this instant; `None` for no upper bound.
    pub end_at: Option<chrono::DateTime<chrono::Utc>>,
    /// A pending firing past this instant never fires and schedules nothing
    /// further; `None` for no limit.
    pub valid_until: Option<chrono::DateTime<chrono::Utc>>,
}

/// The next firing of a recurring deadline chain, given only the
/// previous firing - pure, no I/O, so both `db::fire_due_deadlines` and
/// the unit tests below evaluate the identical math. `prev_fire_at` is
/// the occurrence the chain just fired (or, at schedule time, the initial
/// `fire_at`); `now` bounds the catch-up, the same register
/// `event_store::create_system_event`'s own `nothing_later_is_due`
/// already is. Returns `None` when the chain ends: the schedule doesn't
/// parse, no occurrence is left, or the candidate fails the validity
/// window (`start_at`/`end_at`/`valid_until`, or `now` past
/// `valid_until`).
pub fn next_recurrence_fire_at(
    recurrence: &RecurrenceSpec,
    prev_fire_at: chrono::DateTime<chrono::Utc>,
    now: chrono::DateTime<chrono::Utc>,
) -> Option<chrono::DateTime<chrono::Utc>> {
    use crate::event_store::{latest_occurrence_at_or_before, next_occurrence_after};
    if now
        > recurrence
            .valid_until
            .unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
    {
        return None;
    }
    let candidate = match recurrence.policy {
        MissedRunPolicy::Skip => {
            // Missed occurrences are skipped outright: the first
            // occurrence strictly after whichever is later, the previous
            // firing or now.
            next_occurrence_after(&recurrence.schedule, prev_fire_at.max(now))?
        }
        MissedRunPolicy::FireOnce => {
            // A backlog collapses to a single firing of its latest
            // occurrence - the `fire_once` register §150 already is for
            // the global scheduler - and with no backlog this is just the
            // next occurrence after the previous firing.
            match latest_occurrence_at_or_before(&recurrence.schedule, now) {
                Some(latest) if latest > prev_fire_at => latest,
                _ => next_occurrence_after(&recurrence.schedule, prev_fire_at)?,
            }
        }
        MissedRunPolicy::FireAll => {
            // Step through every occurrence, missed or not.
            next_occurrence_after(&recurrence.schedule, prev_fire_at)?
        }
    };
    if candidate
        < recurrence
            .start_at
            .unwrap_or(chrono::DateTime::<chrono::Utc>::MIN_UTC)
    {
        return None;
    }
    if candidate
        > recurrence
            .end_at
            .unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
    {
        return None;
    }
    if candidate
        > recurrence
            .valid_until
            .unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
    {
        return None;
    }
    Some(candidate)
}

/// `ScheduleDeadline::schedule`'s own `Some` output - everything a
/// one-shot, per-entity timer needs: when it fires (`fire_at`), what it's
/// scoped to for a later `CancelDeadline` lookup (`tags`), and what to
/// submit when it does (`payload`, `Self::Target`'s own payload type).
/// `recurrence` (issue #95) turns the one-shot into the first
/// firing of a chain: after it fires, `db::fire_due_deadlines` derives
/// the next `fire_at` from the cron schedule and the missed-run policy,
/// until the validity window ends or a `CancelDeadline` cancels the
/// pending row. `None` - the default every existing schedule passes - is
/// the unchanged one-shot.
///
/// `expires_at` (issue #97) is an optional latest instant this
/// firing is still meaningful. It is carried to the fired command's
/// `decide_with` through [`DecisionContext::deadline`] as
/// `deadline_at`/`expired`, so a target that only makes sense soon after
/// its `fire_at` - an auto-reply, a reminder - can refuse one the
/// scheduler or a retry ran late. `None` (the default every existing
/// schedule passes) is unchanged: a firing is never `expired`. A window
/// probe: it does not stop the row firing, only tells the command it was
/// late - a chain that wants to stop outright uses
/// `RecurrenceSpec::valid_until`.
#[derive(Debug, Clone)]
pub struct DeadlineSpec<P> {
    /// When the deadline fires.
    pub fire_at: chrono::DateTime<chrono::Utc>,
    /// See this struct's own doc comment (issue #97).
    pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
    /// The tags a `CancelDeadline` matches to cancel this deadline before it
    /// fires.
    pub tags: Vec<Tag>,
    /// The target command's payload, submitted when the deadline fires.
    pub payload: P,
    /// Makes this the first firing of a recurring chain; `None` for a one-shot
    /// deadline.
    pub recurrence: Option<RecurrenceSpec>,
}

/// issue #20: a native one-shot, per-entity deadline/timer -
/// "cancel this order if not paid within 30 minutes," without reaching
/// for `skilj-temporal` (docs/architecture.md §34) or hand-building it on
/// top of the recurring, `EventType`-wide scheduler (`rule
/// CreateSystemEvent`). Deliberately the same shape as `CrossContextRoute`
/// just above, for the same reason: `decide()`'s only output is
/// `CommandDecision` (`Accepted { events }` / `Rejected`) - it has no
/// side-effect channel to schedule a timer through, and none should be
/// added there. So scheduling one is itself modelled as a pure reaction
/// to a committed event, with a background poller doing the actual
/// durable write, exactly like a route reacting to one.
///
/// Fires `Target` as a real command, not a raw event - the same reasoning
/// `CrossContextRoute::Target` already gives: whether a deadline is still
/// relevant is a decision `Target::decide()` gets to make against
/// *current* state at fire time, not one baked in back when the timer was
/// scheduled.
pub trait ScheduleDeadline {
    /// The event that, on commit, may schedule a new deadline.
    type Source: EventType;
    /// The command submitted when the deadline fires.
    type Target: CommandType;

    /// This schedule's own stable identity - names its durable cursor row
    /// and, combined with the triggering `Source` event's own sequence,
    /// the deterministic id of every row it inserts
    /// (`"{NAME}:{source_event_sequence}"`) - so a redelivered/retried
    /// catch-up tick re-inserts the identical row rather than a duplicate
    /// (`ON CONFLICT (id) DO NOTHING`), the same property a route's own
    /// idempotency key gives its command submissions, applied here to the
    /// row insert itself.
    const NAME: &'static str;

    /// See `DeadlinePollStartFrom`. `Beginning` (the default) replays
    /// every `Source` occurrence ever committed - the same default
    /// `CrossContextRoute::START_FROM` has, and for the same reason: it's
    /// what every schedule registered before this const existed already
    /// does.
    const START_FROM: DeadlinePollStartFrom = DeadlinePollStartFrom::Beginning;

    /// `None` skips this occurrence of `Source` entirely - no deadline
    /// scheduled, cursor still advances. `Some(spec)` inserts a `pending`
    /// row, fired the next time `spec.fire_at` comes due (or immediately,
    /// if it's already in the past by the time this runs - a deadline is
    /// a one-shot, so there's no missed-occurrence policy to choose
    /// between the way the recurring scheduler needs one).
    fn schedule(
        source_payload: &<Self::Source as EventType>::Payload,
    ) -> Option<DeadlineSpec<<Self::Target as CommandType>::Payload>>;
}

/// issue #20's own cancellation half - "cancellable if the
/// awaited thing happens first." Bound to a specific `ScheduleDeadline`
/// via `Deadline` (not just any deadline sharing a tag) so two unrelated
/// schedules that happen to reuse a tag key never cross-cancel each
/// other's rows.
///
/// Cancels **by tag**, not by id: the cancelling event (say, `OrderPaid`)
/// only ever carries its own payload/tags, never the opaque id a
/// `ScheduleDeadline` implementor generated deep inside its own row
/// insert - tag-based lookup is what actually lets "whichever happens
/// first" work without threading an id back out through some side
/// channel. Reuses the same tag-containment query
/// `list_events_for_bounded_context_matching_tags` already uses for
/// events (docs/architecture.md §19 Problem 1), applied to the
/// `deadlines` table's own `tags` column.
///
/// **Adding one to an existing schedule in a rolling deploy.** A due
/// deadline waits until every cancel targeting its schedule has processed
/// its source events up to the deadline's `fire_at` - but an instance only
/// knows the cancels declared in its own code. While instances without
/// the new cancel are still running, one of them can fire a deadline whose
/// cancelling event had already committed but not yet been processed.
/// Where that matters, deploy the new cancel while no such deadline can
/// come due, or roll out completely before relying on it
/// (docs/architecture.md §164).
pub trait CancelDeadline {
    /// The event that, on commit, may cancel one or more pending
    /// deadlines.
    type Source: EventType;
    /// Which `ScheduleDeadline`'s own pending rows this targets.
    type Deadline: ScheduleDeadline;

    /// This cancel reactor's own stable identity - names its durable
    /// cursor row, independent of `Self::Deadline::NAME`'s own cursor (a
    /// schedule and its cancel counterpart advance independently, each
    /// walking its own `Source` event stream).
    const NAME: &'static str;

    /// See `DeadlinePollStartFrom`; same default and reasoning as
    /// `ScheduleDeadline::START_FROM`.
    const START_FROM: DeadlinePollStartFrom = DeadlinePollStartFrom::Beginning;

    /// `None` skips this occurrence entirely - nothing cancelled, cursor
    /// still advances. `Some(tags)` cancels every still-`pending` row
    /// `Self::Deadline::NAME` owns whose own `tags` contain these -
    /// cancelling zero, one, or several rows is all a legitimate outcome
    /// here (a deadline that already fired, was already cancelled, or was
    /// never scheduled in the first place is not an error), the same
    /// register `CrossContextRoute::route` returning `None` already is.
    fn cancel_tags(source_payload: &<Self::Source as EventType>::Payload) -> Option<Vec<Tag>>;
}

/// One registered `ScheduleDeadline`'s own static identity -
/// `ScheduleDeadlineDispatcher::schedules()`'s own element type, the
/// `ScheduleDeadline`/`CancelDeadline` pair's counterpart to
/// `CrossContextRouteInfo`.
#[derive(Debug, Clone, Copy)]
pub struct ScheduleDeadlineInfo {
    /// `ScheduleDeadline::NAME`: the schedule's identity and the name of its
    /// cursor.
    pub name: &'static str,
    /// The bounded context the source event type lives in.
    pub source_bounded_context: &'static str,
    /// The event type whose events schedule deadlines.
    pub source_event_type: &'static str,
    /// The bounded context the target command type lives in.
    pub target_bounded_context: &'static str,
    /// The command type a firing submits.
    pub target_command_type: &'static str,
    /// `ScheduleDeadline::START_FROM`.
    pub start_from: DeadlinePollStartFrom,
}

/// One registered `CancelDeadline`'s own static identity.
#[derive(Debug, Clone, Copy)]
pub struct CancelDeadlineInfo {
    /// `CancelDeadline::NAME`: the cancel reactor's identity and the name of
    /// its cursor.
    pub name: &'static str,
    /// The bounded context the cancelling event type lives in.
    pub source_bounded_context: &'static str,
    /// The event type whose events cancel deadlines.
    pub source_event_type: &'static str,
    /// The paired `ScheduleDeadline::NAME` whose pending deadlines this
    /// cancels.
    pub deadline_schedule_name: &'static str,
    /// `Self::Deadline::Source::BOUNDED_CONTEXT` - where the paired
    /// `ScheduleDeadline`'s own `deadlines`/`deadline_cursors` rows
    /// physically live (mirroring `CrossContextRoute`'s own
    /// `source_bounded_context` placement choice for its cursor table),
    /// which can differ from this cancel reactor's own
    /// `source_bounded_context` above - the cancelling event and the
    /// deadline it cancels aren't required to live in the same bounded
    /// context, the same generality `CrossContextRoute::Source`/`Target`
    /// already has.
    pub deadline_schedule_bounded_context: &'static str,
    /// `Self::Deadline::Source::NAME` - the event type the paired
    /// `ScheduleDeadline` schedules from. A cancel waits until that
    /// schedule has processed every such event that precedes the
    /// cancelling one (docs/architecture.md §130).
    pub deadline_schedule_source_event_type: &'static str,
    /// `CancelDeadline::START_FROM`.
    pub start_from: DeadlinePollStartFrom,
}

/// Type-erased dispatch to a bounded context's own typed
/// `ScheduleDeadline::schedule` - `CrossContextRouteDispatcher`'s own
/// counterpart for this trait.
pub trait ScheduleDeadlineDispatcher: Send + Sync {
    /// Every registered schedule, across every bounded context - not
    /// scoped to one bounded context, the same reasoning
    /// `CrossContextRouteDispatcher::routes` already gives.
    fn schedules(&self) -> Vec<ScheduleDeadlineInfo>;

    /// `schedule_name` names one of `schedules()`'s own entries;
    /// `source_payload_json` is the triggering `Source` event's own
    /// stored payload. Outer `None` - not registered at all. `Some(Err(e))` -
    /// the stored payload didn't deserialize into `Source::Payload`.
    /// `Some(Ok(None))` - `schedule()` itself decided this occurrence
    /// doesn't apply. `Some(Ok(Some(spec)))` - ready to insert as a
    /// `pending` row, `Target`'s own JSON payload already resolved.
    fn schedule(
        &self,
        schedule_name: &str,
        source_payload_json: &str,
    ) -> Option<Result<Option<ErasedDeadlineSpec>, serde_json::Error>>;
}

/// `ScheduleDeadlineDispatcher::schedule`'s own type-erased output -
/// `DeadlineSpec<P>` with `P` resolved to its already-serialised JSON,
/// the same "erase the payload's own concrete type at the dispatcher
/// boundary" treatment every other dispatcher trait in this module
/// already gives its own typed payloads. `recurrence` is carried through
/// verbatim (issue #95) - the dispatcher boundary erases the
/// payload type only, never the schedule's own recurrence decision.
#[derive(Debug, Clone)]
pub struct ErasedDeadlineSpec {
    /// `DeadlineSpec::fire_at`.
    pub fire_at: chrono::DateTime<chrono::Utc>,
    /// `DeadlineSpec::expires_at`, carried through verbatim (
    /// issue #97).
    pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
    /// `DeadlineSpec::tags`.
    pub tags: Vec<Tag>,
    /// `DeadlineSpec::payload`, serialized to JSON.
    pub payload_json: String,
    /// `DeadlineSpec::recurrence`, carried through as is.
    pub recurrence: Option<RecurrenceSpec>,
}

/// Type-erased dispatch to a bounded context's own typed
/// `CancelDeadline::cancel_tags`.
pub trait CancelDeadlineDispatcher: Send + Sync {
    /// Every registered cancel reactor, across every bounded context.
    fn cancels(&self) -> Vec<CancelDeadlineInfo>;

    /// `cancel_name` names one of `cancels()`'s own entries;
    /// `source_payload_json` is the triggering `Source` event's own
    /// stored payload. Outer `None` - not registered at all. `Some(Err(e))` -
    /// the stored payload didn't deserialize into `Source::Payload`.
    /// `Some(Ok(None))` - `cancel_tags()` itself decided this occurrence
    /// doesn't apply. `Some(Ok(Some(tags)))` - cancel every still-`pending`
    /// row the paired `ScheduleDeadline` owns whose own tags contain
    /// these.
    fn cancel_tags(
        &self,
        cancel_name: &str,
        source_payload_json: &str,
    ) -> Option<Result<Option<Vec<Tag>>, serde_json::Error>>;
}