minerva 0.2.0

extern crate alloc;

use alloc::sync::Arc;

use crate::kairos::{Clock, DynClock, SkewExceeded, TimeSource, ToU16};

use super::{Event, VersionVector};

/// A sender-side event producer: the *send* half of causal broadcast, the
/// inverse of the [`Ideal`](crate::metis::Ideal) gate.
///
/// A producer owns one station's send-side causal context and mints
/// [`Event`]s with two operations.
///
/// [`produce`](Self::produce) is the Birman-Schiper-Stephenson send step:
/// count this event in the station's own knowledge, snapshot that knowledge
/// as the event's dependencies, and mint a stamp that already dominates
/// everything observed. [`observe`](Self::observe) is the dual. It folds a
/// received event's dependencies into knowledge and advances the clock by the
/// Hybrid Logical Clock receive rule. This is the only way a producer
/// learns another station's progress. [`try_observe`](Self::try_observe) is
/// its bounded form, refusing wholesale when forward skew exceeds a
/// caller-set bound.
///
/// # Clock ownership
///
/// `C` names how the producer holds its [`Clock`]: any sealed
/// [`ClockCarrier`]. The default owns an erased clock by value. A station has
/// *one* logical clock by invariant. Some stations have stamping sites beyond
/// the producer, such as a seal or a ledger append. Such a station must share
/// that clock with every site. Otherwise it mints colliding stamps under one
/// `station_id`. Such a station passes `Producer<Arc<Clock<TS>>>` or
/// `Producer<&Clock<TS>>` instead. Semantics do not depend on `C`, and the
/// lock-free clock already tolerates concurrent stamping.
///
/// Share the clock across stamping *sites*, never the producer role: one
/// station still runs one producer. Two producers over one clock would each
/// self-count independently and mint colliding dots.
///
/// # Composition
///
/// A producer composes with an [`Ideal`](crate::metis::Ideal) rather than
/// owning one: it *observes* each arrival on receipt, building the knowledge
/// a later `produce` snapshots, while the buffer *delivers* once causal
/// order permits.
///
/// Feeding arrivals is safe in any order. Knowledge folds by join, and the
/// clock only advances. Thus, a later `produce` dominates every observed
/// stamp whatever the order, though exact clock values may vary with it. A
/// node never feeds its own produced events back through its producer, since
/// they were self-counted; a stray self-echo is harmless but ticks the clock.
///
/// # Not generic over the payload
///
/// The producer holds only the clock and knowledge, both payload-agnostic,
/// so the operations carry the payload. This is the one asymmetry with
/// [`Event<T>`](Event) and [`Ideal<T>`](crate::metis::Ideal).
/// `C` is not an exception: it abstracts how the clock is *held*, never what
/// it stamps.
///
/// # Totality
///
/// Every operation is infallible and panic-free, except
/// [`try_observe`](Self::try_observe). Rejection is that method's whole point.
/// [`new`](Self::new) cannot fail because the [`Clock`] already validated its
/// `station_id`.
pub struct Producer<C = DynClock> {
    /// Identity (`clock.station_id()`), stamp source, *and* high-water. Identity is
    /// single-sourced here, so the producer never holds a `station_id` that could
    /// disagree with the clock; the clock is also the high-water, since each
    /// [`observe`](Self::observe) folds a receipt in via
    /// [`Clock::observe`](crate::kairos::Clock::observe), so a plain
    /// [`now`](crate::kairos::Clock::now) mint already dominates every observed stamp.
    /// Held through a `ClockCarrier` (owned, borrowed, or shared).
    clock: C,
    /// Observed-knowledge: `knowledge[self]` is the count of events this station has
    /// produced, `knowledge[k]` the count of station `k`'s events it has observed (received).
    /// Receipt-level, so generally ahead of what any buffer has *delivered* to the application.
    knowledge: VersionVector,
}

/// An ownership carrier that gives a producer one validated HLC.
///
/// This trait is sealed: producers accept an owned [`Clock`], `&Clock<TS>`, or
/// `Arc<Clock<TS>>`, but callers cannot substitute a different minting machine under
/// the same contract.
pub trait ClockCarrier: private::Sealed {
    /// The clock's physical-time source.
    type Source: TimeSource;

    /// Borrows the carried clock.
    fn clock(&self) -> &Clock<Self::Source>;
}

impl<TS: TimeSource> ClockCarrier for Clock<TS> {
    type Source = TS;

    fn clock(&self) -> &Clock<Self::Source> {
        self
    }
}

impl<TS: TimeSource> ClockCarrier for &Clock<TS> {
    type Source = TS;

    fn clock(&self) -> &Clock<Self::Source> {
        self
    }
}

impl<TS: TimeSource> ClockCarrier for Arc<Clock<TS>> {
    type Source = TS;

    fn clock(&self) -> &Clock<Self::Source> {
        self
    }
}

mod private {
    use super::Arc;
    use crate::kairos::{Clock, TimeSource};

    pub trait Sealed {}

    impl<TS: TimeSource> Sealed for Clock<TS> {}
    impl<TS: TimeSource> Sealed for &Clock<TS> {}
    impl<TS: TimeSource> Sealed for Arc<Clock<TS>> {}
}

impl<C: ClockCarrier> Producer<C> {
    /// Creates a producer for `clock`'s station: empty knowledge, no events yet.
    ///
    /// `clock` is any [`ClockCarrier`]: pass a [`Clock`] by value for the
    /// sole-owner shape, or an `Arc<Clock<TS>>` / `&Clock<TS>` to share the station's one
    /// clock with its other stamping sites (see the type-level *Clock ownership*
    /// section).
    ///
    /// Infallible; the [`Clock`] already validated its `station_id` at construction.
    #[must_use]
    pub const fn new(clock: C) -> Self {
        Self {
            clock,
            knowledge: VersionVector::new(),
        }
    }

    /// Mints the next event: the Birman-Schiper-Stephenson send step.
    ///
    /// Advances this station's own count, snapshots the resulting knowledge as the
    /// event's `deps` (so `deps[self]` is the event's own causal sequence number,
    /// `>= 1`), mints a [`Kairos`](crate::kairos::Kairos) with [`now`](crate::kairos::Clock::now), and returns
    /// the event to broadcast. A plain `now` suffices because the clock *is* the
    /// high-water: every [`observe`](Self::observe) already folded each receipt in via
    /// [`observe`](crate::kairos::Clock::observe), so the minted stamp strictly exceeds
    /// every stamp this station has produced or observed.
    pub fn produce<P>(&mut self, kairotic: impl ToU16, payload: P) -> Event<P> {
        let _ = self.knowledge.increment(self.clock.clock().station_id());
        let deps = self.knowledge.clone();
        let stamp = self.clock.clock().now(kairotic);
        Event {
            stamp,
            deps,
            payload,
        }
    }

    /// *Observes* a received event: folds its dependencies into knowledge and advances the
    /// clock past its stamp with the Hybrid Logical Clock receive rule
    /// ([`observe`](crate::kairos::Clock::observe)).
    ///
    /// This is *receipt*, not application *delivery*. "Deliver" in
    /// Birman-Schiper-Stephenson vocabulary is the ordered, once-only release
    /// to the application, which is
    /// [`pop_ready`](crate::metis::Ideal::pop_ready); a node observes an event
    /// on arrival, before any buffer releases it. So an
    /// observed-but-undelivered event already shapes future stamps ---
    /// receipt-level causality, deliberately stronger than the
    /// application-visible kind
    /// (`docs/metis-production-recording-decision.adoc`).
    ///
    /// Knowledge folds by join, so it is order-independent. The clock's
    /// receive rule is *not* a join: it ticks once per call. Exact clock
    /// values therefore depend on how many events are observed. The guarantee
    /// [`produce`](Self::produce) rests on still holds in any order.
    ///
    /// For *remote* events only: a producer self-counts in `produce` and must
    /// not observe its own output. A stray self-echo leaves knowledge
    /// untouched and only ticks the clock, so it is harmless but not a
    /// literal no-op.
    pub fn observe<P>(&mut self, event: &Event<P>) {
        self.knowledge = self.knowledge.merge(&event.deps);
        self.clock.clock().observe(event.stamp);
    }

    /// The bounded, trust-aware [`observe`](Self::observe): folds the event in only
    /// if its stamp's forward skew is within `max_forward_skew`, else rejects it and
    /// leaves the producer untouched.
    ///
    /// [`Clock::try_observe`](crate::kairos::Clock::try_observe)'s skew check
    /// lifted over the whole receipt: on `Ok` the effect is exactly
    /// [`observe`](Self::observe), on `Err` neither knowledge nor clock has
    /// moved.
    ///
    /// That all-or-nothing pairing is why this exists rather than the caller
    /// composing the two. Merging a rejected event's `deps` while refusing its
    /// stamp is unsafe. A later [`produce`](Self::produce) could then snapshot
    /// dependencies that cover an event its own stamp does not dominate. That
    /// breaks the embedding of causal order in stamp order, which the whole
    /// layer rests on.
    ///
    /// The bound defends only when the event's origin is authenticated.
    /// Without authentication an adversary simply rotates the claimed
    /// identity. Verify the sender *before* the fold, and apply your own
    /// policy to a rejection: minerva owns the mechanism, never the trust.
    ///
    /// # Errors
    ///
    /// [`SkewExceeded`] when the stamp's forward skew over the current time-source
    /// reading exceeds `max_forward_skew`; knowledge and clock are both unchanged.
    pub fn try_observe<P>(
        &mut self,
        event: &Event<P>,
        max_forward_skew: u64,
    ) -> Result<(), SkewExceeded> {
        self.clock
            .clock()
            .try_observe(event.stamp, max_forward_skew)?;
        self.knowledge = self.knowledge.merge(&event.deps);
        Ok(())
    }

    /// This producer's station identity, single-sourced from its [`Clock`].
    #[must_use]
    pub fn station_id(&self) -> u32 {
        self.clock.clock().station_id()
    }

    /// The station's observed-knowledge, read-only: `knowledge[self]` is how many
    /// events it has produced, `knowledge[k]` how many of station `k`'s it has
    /// observed (received).
    #[must_use]
    pub const fn knowledge(&self) -> &VersionVector {
        &self.knowledge
    }
}