rustrade-data 0.12.1

//! The London Strategic Edge stream transformer.
//!
//! Decoding is delegated wholesale to [`StatelessTransformer`] — one provider frame maps to one
//! event and nothing about it needs state. What this type adds is the consuming half of
//! [`resume`](super::resume): skipping the events a reconnect's replay re-delivers; and, on the
//! options dataset, the count of prints for contracts nobody registered.

use super::{
    live::subscribes_per_underlying,
    resume::{LseResumeKey, LseResumeState, LseWatermark},
    tick::LseMessage,
};
use crate::exchange::osi;
use crate::{
    Identifier,
    error::DataError,
    event::{MarketEvent, MarketIter},
    exchange::Connector,
    subscriber::shared_stream::AttachedTransformer,
    subscription::{Map, SubscriptionKind},
    transformer::{ExchangeTransformer, stateless::StatelessTransformer},
};
use chrono::{DateTime, Utc};
use fnv::{FnvHashMap, FnvHashSet};
use rustrade_instrument::exchange::ExchangeId;
use rustrade_integration::{
    Transformer, protocol::websocket::WsMessage, subscription::SubscriptionId,
};
use serde::Deserialize;
use smol_str::SmolStr;
use std::{
    cmp::Ordering,
    collections::BTreeMap,
    sync::Arc,
    time::{Duration, Instant},
};
use tokio::sync::mpsc;
use tracing::{info, warn};

/// How often the options dataset reports the prints it dropped for unregistered contracts.
///
/// A minute keeps a busy connection to one line per underlying per minute — at one registered
/// contract on a busy index ETF, that line stands in for thousands of dropped prints.
pub const UNREGISTERED_REPORT_INTERVAL: Duration = Duration::from_secs(60);

/// Transformer for every London Strategic Edge subscription kind.
///
/// Carries no resume state unless the caller opted in via
/// [`LseSubscriber::with_resume`](super::live::LseSubscriber::with_resume); without it this is a
/// direct pass-through to [`StatelessTransformer`].
#[derive(Debug)]
pub struct LseTransformer<Exchange, InstrumentKey, Kind> {
    inner: StatelessTransformer<Exchange, InstrumentKey, Kind, LseMessage>,
    resume: Option<ResumeTracker>,

    /// Present on the options dataset only, where most of what arrives was never registered.
    unregistered: Option<UnregisteredContracts>,
}

/// The options dataset's count of prints for contracts nobody registered.
///
/// # Why a count and not an error per print
/// One subscribe delivers an underlying's whole chain, so a stream registering a handful of
/// contracts receives prints for thousands it did not — measured at around 45 a second on one busy
/// underlying. Resolved like any other frame, each would become an "unidentifiable" error, drowning
/// every real one. They are expected, so they are dropped; but silently dropping them would hide
/// the one case that is a mistake — a registered contract spelled differently from the one that
/// trades, whose prints would all land here — so they are counted per underlying and reported
/// every [`UNREGISTERED_REPORT_INTERVAL`].
#[derive(Debug)]
struct UnregisteredContracts {
    exchange: ExchangeId,
    registered: FnvHashSet<SubscriptionId>,

    /// Ordered so the report lists underlyings in the same order every time.
    tally: BTreeMap<SmolStr, UnregisteredTally>,
    window_start: Instant,
}

/// What one underlying delivered for unregistered contracts since the last report.
#[derive(Debug, Default)]
struct UnregisteredTally {
    prints: u64,
    contracts: FnvHashSet<SubscriptionId>,
}

/// What a stream needs to resume: the shared watermarks, the kind they are filed under, and the
/// replay windows the connection opened for it.
#[derive(Debug)]
pub(super) struct ResumeContext {
    pub(super) state: Arc<LseResumeState>,

    /// [`SubscriptionKind::as_str`] of the stream — see [`LseResumeKey`].
    pub(super) kind: &'static str,

    /// Each subscription a replay window was opened for, and the instant it opens at.
    ///
    /// Only these expect a replay. A subscription the connection already held when the stream
    /// attached was never re-sent, so the provider replays nothing for it, and expecting one would
    /// read every live tick as a replay that fell short.
    pub(super) starts: FnvHashMap<SubscriptionId, DateTime<Utc>>,
}

/// Per-connection resume bookkeeping: the shared watermark, and what this connection still owes
/// the replay window.
#[derive(Debug)]
struct ResumeTracker {
    state: Arc<LseResumeState>,

    /// The dataset this stream serves, which partitions the shared state alongside the kind — see
    /// [`LseResumeKey`]. Held as a value because the transformer's `Exchange` type parameter is
    /// only reachable where its bounds are in scope, which the per-tick path is not.
    exchange: ExchangeId,

    /// The kind this stream serves, which partitions the shared state — see
    /// [`LseResumeKey`]. Held as a string because the transformer's `Kind` type parameter has no
    /// value to call [`SubscriptionKind::as_str`] on.
    kind: &'static str,

    /// Keyed by subscription alone: the dataset and the kind are this stream's, so both are
    /// constant across every entry here.
    pending: FnvHashMap<SubscriptionId, PendingDrop>,
}

/// Events still expected to arrive twice for one subscription, and the instant they carry.
#[derive(Copy, Clone, Debug)]
struct PendingDrop {
    time_exchange: DateTime<Utc>,
    remaining: usize,

    /// Whether the window was opened *before* this stream's own watermark.
    ///
    /// A symbol has one replay window per connection, opened at the earliest watermark among the
    /// streams holding it, so a stream that had delivered further receives ticks it already
    /// delivered before reaching its own watermark. Those are dropped silently, until the first
    /// tick past the watermark closes the window. Where the window opened exactly at the watermark,
    /// an older tick contradicts the inclusive `start` and is reported instead.
    opened_earlier: bool,

    /// Whether a tick past the watermark has arrived, closing the replay window for good.
    passed: bool,

    /// Whether the out-of-contract-ordering warning has already fired for this subscription.
    ///
    /// A provider replaying from *before* the requested instant would otherwise warn once per
    /// replayed tick, which on a busy symbol is tens of thousands of identical lines.
    warned_older: bool,

    /// Whether the provider announced a replay window opening later than the one requested.
    ///
    /// A clamp explains a shortfall at the resume instant completely — the window simply does not
    /// reach it — so the shortfall must not also be reported as the provider's history changing
    /// under the subscription. Two reports with different causes for one event is worse than one,
    /// because only one of them is true.
    ///
    /// Set when the boundary frame is read, which the provider was measured to send before the
    /// window it announces. The suppression is therefore ordered, not absolute: see the
    /// `Ordering::Greater` arm of [`should_drop`](ResumeTracker::should_drop) for what happens
    /// under the reverse order, and why it is tolerated.
    clamped: bool,
}

impl<Exchange, InstrumentKey, Kind> LseTransformer<Exchange, InstrumentKey, Kind>
where
    Exchange: Connector + Send,
    InstrumentKey: Clone + Send + Sync,
    Kind: SubscriptionKind + Send,
    Kind::Event: Sync,
    MarketIter<InstrumentKey, Kind::Event>: From<(ExchangeId, InstrumentKey, LseMessage)>,
{
    /// Construct a transformer, optionally resuming from previously delivered events.
    ///
    /// `resume` carries the kind this stream serves alongside the shared state, because the state
    /// is partitioned by dataset *and* kind — see [`LseResumeKey`] — and this type's `Kind`
    /// parameter has no value to derive the latter from. The dataset comes from `Exchange::ID`.
    ///
    /// The drop counters are read here, once per resumed subscription, rather than consulted per
    /// tick. Recording still takes the shared lock once per emitted tick; the critical section
    /// there is a hash lookup and a field update.
    pub(super) async fn new(
        instrument_map: Map<InstrumentKey>,
        initial_snapshots: &[MarketEvent<InstrumentKey, Kind::Event>],
        ws_sink_tx: mpsc::UnboundedSender<WsMessage>,
        resume: Option<ResumeContext>,
    ) -> Result<Self, DataError> {
        let unregistered = subscribes_per_underlying(Exchange::ID).then(|| {
            UnregisteredContracts::new(Exchange::ID, instrument_map.0.keys().cloned().collect())
        });

        let inner =
            StatelessTransformer::init(instrument_map, initial_snapshots, ws_sink_tx).await?;

        Ok(Self {
            inner,
            resume: resume.map(Self::tracker),
            unregistered,
        })
    }
}

impl<Exchange, InstrumentKey, Kind> LseTransformer<Exchange, InstrumentKey, Kind>
where
    Exchange: Connector,
{
    /// The bookkeeping for resuming from `context`.
    fn tracker(
        ResumeContext {
            state,
            kind,
            starts,
        }: ResumeContext,
    ) -> ResumeTracker {
        ResumeTracker {
            pending: starts
                .into_iter()
                .filter_map(|(subscription, start)| {
                    let key = LseResumeKey::new(Exchange::ID, subscription.clone(), kind);
                    let watermark = state.watermark(&key)?;
                    Some((subscription, PendingDrop::new(watermark, start)))
                })
                .collect(),
            state,
            exchange: Exchange::ID,
            kind,
        }
    }
}

/// What the shared connection hands a London Strategic Edge stream on attaching: what it needs to
/// resume, if it resumes.
///
/// Built by the connection from the subscriber's resume state and the replay windows it opened for
/// the stream, and handed to the stream's [`LseTransformer`] before any frame reaches it.
#[derive(Debug, Default)]
pub struct LseResume(pub(super) Option<ResumeContext>);

/// Resume from what the connection handed over, if anything.
///
/// Replaces any resume state the transformer was built with: the connection's is the one that
/// knows which replay windows it opened.
impl<Exchange, InstrumentKey, Kind> AttachedTransformer<LseResume>
    for LseTransformer<Exchange, InstrumentKey, Kind>
where
    Exchange: Connector,
{
    fn attached(&mut self, LseResume(resume): LseResume) {
        self.resume = resume.map(Self::tracker);
    }
}

impl PendingDrop {
    /// What remains to skip of a window opened at `start` for a stream last delivered at
    /// `watermark`.
    fn new(watermark: LseWatermark, start: DateTime<Utc>) -> Self {
        Self {
            time_exchange: watermark.time_exchange,
            remaining: watermark.count_at_time,
            opened_earlier: start < watermark.time_exchange,
            passed: false,
            warned_older: false,
            clamped: false,
        }
    }
}

impl<Exchange, InstrumentKey, Kind> ExchangeTransformer<Exchange, InstrumentKey, Kind>
    for LseTransformer<Exchange, InstrumentKey, Kind>
where
    Exchange: Connector + Send,
    InstrumentKey: Clone + Send + Sync,
    Kind: SubscriptionKind + Send,
    Kind::Event: Sync,
    MarketIter<InstrumentKey, Kind::Event>: From<(ExchangeId, InstrumentKey, LseMessage)>,
{
    /// Initialise without resumption.
    ///
    /// This is the trait's only seam and it is a static function, so it cannot carry the caller's
    /// resume state; the stream that wants resumption bypasses it and constructs the transformer
    /// directly.
    ///
    /// Initial snapshots are forwarded rather than dropped here, so this layer stops being the
    /// place they are lost. That is as far as forwarding reaches: the inner
    /// [`StatelessTransformer`]'s own `init` discards the argument unconditionally, as every
    /// stateless connector's does, and this provider's streams fetch no snapshot for it to
    /// discard. A future LSE kind that does fetch one would
    /// need a transformer not backed by `StatelessTransformer` before the snapshot reached it —
    /// forwarding is what makes that a visible gap rather than one hidden behind a `&[]` at the
    /// call site.
    async fn init(
        instrument_map: Map<InstrumentKey>,
        initial_snapshots: &[MarketEvent<InstrumentKey, Kind::Event>],
        ws_sink_tx: mpsc::UnboundedSender<WsMessage>,
    ) -> Result<Self, DataError> {
        Self::new(instrument_map, initial_snapshots, ws_sink_tx, None).await
    }
}

impl<Exchange, InstrumentKey, Kind> Transformer for LseTransformer<Exchange, InstrumentKey, Kind>
where
    Exchange: Connector,
    InstrumentKey: Clone,
    Kind: SubscriptionKind,
    MarketIter<InstrumentKey, Kind::Event>: From<(ExchangeId, InstrumentKey, LseMessage)>,
{
    type Error = DataError;
    type Input = LseMessage;
    type Output = MarketEvent<InstrumentKey, Kind::Event>;
    type OutputIter = Vec<Result<Self::Output, Self::Error>>;

    fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
        // The replay boundary is consumed here rather than forwarded. It is the only evidence that
        // a requested window was clamped to the provider's retention, and it cannot be caught
        // during the handshake -- see `LseMessage::ReplayStarted`. No decoder has a market event to
        // build from it either way.
        if let LseMessage::ReplayStarted {
            subscription_id,
            from,
        } = &input
        {
            let (subscription_id, from) = (subscription_id.clone(), *from);

            if let Some(resume) = self.resume.as_mut() {
                resume.warn_if_clamped(&subscription_id, from);
            }

            return vec![];
        }

        // A rejection. The shared connection consumes these rather than routing them -- one names
        // no symbol, so there is no stream to route it to -- but one arriving here by any other
        // route must still be reported: nothing downstream can act on it, and the stream simply
        // stops producing events for whatever was dropped, so a rejection is the last thing that
        // may reach the consumer as silence.
        if let LseMessage::Error { code, message } = &input {
            warn!(
                exchange = %Exchange::ID,
                code = code.as_deref().unwrap_or("unknown"),
                message = message.as_deref().unwrap_or("no message"),
                "London Strategic Edge rejected something after the handshake completed; the \
                 rejection does not name a symbol, so check whether a subscription has gone quiet",
            );

            return vec![];
        }

        // A print for an option contract nobody registered: counted, not converted into an error.
        // The report is checked on every option tick, registered or not, so it keeps its cadence
        // however few unregistered prints arrive.
        if let Some(unregistered) = self.unregistered.as_mut()
            && let LseMessage::Tick(tick) = &input
        {
            let dropped = unregistered.absorb(&tick.subscription_id);
            unregistered.report_if_due(Instant::now());

            if dropped {
                return vec![];
            }
        }

        // Nothing to track when the caller did not opt in, and a control frame carries no instant
        // to track. Both go straight through.
        let Some((subscription_id, time_exchange, replay)) =
            self.resume.as_ref().and_then(|_| match &input {
                LseMessage::Tick(tick) => Some((
                    tick.subscription_id.clone(),
                    tick.time_exchange,
                    tick.replay,
                )),
                _ => None,
            })
        else {
            return self.inner.transform(input);
        };

        if let Some(resume) = self.resume.as_mut()
            && resume.should_drop(&subscription_id, time_exchange, replay)
        {
            return vec![];
        }

        let output = self.inner.transform(input);

        // Record against what was actually emitted. An unidentifiable instrument yields an error
        // rather than an event, and a watermark advanced past an event nobody received would
        // leave a gap on the next reconnect.
        if let Some(resume) = self.resume.as_ref()
            && output.iter().any(Result::is_ok)
        {
            resume.record(subscription_id, time_exchange);
        }

        output
    }
}

impl ResumeTracker {
    /// Whether this event was already delivered by the connection this one replaced.
    fn should_drop(
        &mut self,
        subscription_id: &SubscriptionId,
        time_exchange: DateTime<Utc>,
        replay: bool,
    ) -> bool {
        let exchange = self.exchange;
        let Some(pending) = self.pending.get_mut(subscription_id) else {
            return false;
        };

        match time_exchange.cmp(&pending.time_exchange) {
            // Inside the instant the watermark names, and the previous connection is known to have
            // delivered this many events there. `start` is inclusive, so the provider replays them
            // all and they are skipped by position.
            Ordering::Equal if pending.remaining > 0 => {
                pending.remaining -= 1;

                // The skip stays positional and does NOT require `replay`, deliberately: the flag
                // is absent on live ticks, so a provider that stopped stamping replayed ones would
                // turn every reconnect into a duplicate flood if the drop depended on it. But a
                // tick arriving unstamped inside the window is a live event being discarded as a
                // duplicate, which only happens if the replay under-delivered at this instant, so
                // it is reported. Bounded by the count the watermark carries.
                if !replay {
                    warn!(
                        %exchange,
                        subscription = %subscription_id,
                        instant = %time_exchange,
                        "London Strategic Edge delivered an unreplayed tick inside the resume \
                         window and it was skipped as an already-delivered duplicate; a genuinely \
                         new tick may have been lost",
                    );
                }

                true
            }
            // The instant is exhausted; anything further carrying it is genuinely new.
            Ordering::Equal => false,
            Ordering::Greater => {
                let undelivered = pending.remaining;
                let instant = pending.time_exchange;

                // The window is closed for good, but the entry is KEPT. It carries the instant that
                // went out as `start`, which is the only record of what was requested, and the
                // clamp check reads it out of here when the boundary frame arrives -- removing it
                // would make a clamp undetectable for any subscription whose first replayed tick
                // beat its own `replay_started`. Zeroing the count closes the drop window just as
                // removal did: `Ordering::Equal` with nothing remaining emits.
                pending.remaining = 0;
                pending.passed = true;

                // The replay held fewer events at the watermark's instant than were delivered from
                // it. Positional skipping assumes the replay reproduces what went out live -- an
                // assumption that holds on every measurement taken, including one instant carrying
                // 141 events -- so a shortfall means the provider's history changed underneath the
                // subscription. No event is lost by it, but the assumption is load-bearing enough
                // that its failure must be visible rather than inferred from a gap much later.
                //
                // Unless the window was clamped, which explains the shortfall completely: the
                // window does not reach the resume instant at all, so of course nothing replayed at
                // it. The clamp is reported with the right cause by `warn_if_clamped`.
                //
                // `clamped` is only set once the boundary frame has been seen, so the suppression
                // holds for the measured order -- boundary first -- and not for its reverse. Should
                // a replayed tick beat its own boundary frame, this warning fires before the clamp
                // is known and BOTH lines appear, of which only the clamp's is true. That is
                // tolerated rather than closed: the alternative is to defer this warning until
                // something proves no clamp is coming, and the only frame that could prove it is
                // `replay_complete`, whose delivery and ordering are unmeasured -- so deferring
                // would risk losing a true report entirely to avoid an extra false one. A redundant
                // line is the cheaper failure.
                if undelivered > 0 && !pending.clamped {
                    warn!(
                        %exchange,
                        subscription = %subscription_id,
                        instant = %instant,
                        undelivered,
                        "London Strategic Edge replay held fewer ticks at the resume instant than \
                         were delivered from it; the provider's history changed under the resume",
                    );
                }

                false
            }
            // Older than the watermark, inside a window the connection opened earlier than it on
            // another stream's behalf: this stream delivered it before the connection was lost.
            Ordering::Less if pending.opened_earlier && !pending.passed => true,
            // Older than the watermark. `start` is inclusive, so the provider should never send
            // this; emitting is the safe answer, since dropping would discard a real event on the
            // strength of an assumption that has already been contradicted.
            Ordering::Less => {
                // The mirror image of the shortfall above, and the same load-bearing assumption
                // contradicted in the other direction -- most plausibly a `start` filter flooring
                // to a coarser resolution than the microseconds it was given. Its consequence is
                // worse, because everything between the served instant and the watermark is a
                // duplicate delivered into consumer state, so it must not be the quiet one. Fired
                // once per subscription: the condition holds for every replayed tick below the
                // watermark, and on a busy symbol that is tens of thousands of them.
                //
                // The entry now outlives the replay window -- it is kept so a late boundary frame
                // can still be recognised as a clamp -- so this arm is also reachable long after
                // the replay ended, for a live tick stamped before the resume instant. That is a
                // different cause with the same consequence, and the wording names neither rather
                // than asserting the wrong one.
                if !pending.warned_older {
                    pending.warned_older = true;

                    warn!(
                        %exchange,
                        subscription = %subscription_id,
                        instant = %time_exchange,
                        watermark = %pending.time_exchange,
                        "London Strategic Edge delivered a tick older than the resume instant, \
                         which an inclusive `start` should make impossible during the replay and \
                         a monotonic feed should make impossible after it; it is emitted rather \
                         than dropped, so expect duplicates back to the instant shown",
                    );
                }

                false
            }
        }
    }

    /// Warn when the provider opened a replay window later than the one asked for.
    ///
    /// # ⚠️ Silent clamping is the failure this exists to surface
    /// The provider retains 24 hours. A `start` older than that is **moved forward and served with
    /// no error** — a window requested 48 hours back was answered with one beginning exactly 24
    /// hours back. The `from` of the [`ReplayStarted`](LseMessage::ReplayStarted) frame is the only
    /// signal that it happened.
    ///
    /// The watermark is what went out as `start`, so it *is* the requested instant and no separate
    /// record of the request is needed. A window opened exactly at it is not a clamp: `start` is
    /// inclusive, so that is precisely the window asked for.
    ///
    /// A detected clamp is also recorded against the subscription, because it explains away the
    /// shortfall [`should_drop`](Self::should_drop) would otherwise report with a different and
    /// wrong cause.
    fn warn_if_clamped(&mut self, subscription_id: &SubscriptionId, from: DateTime<Utc>) {
        let Some(requested) = self.clamped_from(subscription_id, from) else {
            return;
        };

        if let Some(pending) = self.pending.get_mut(subscription_id) {
            pending.clamped = true;
        }

        warn!(
            exchange = %self.exchange,
            subscription = %subscription_id,
            requested = %requested,
            serving_from = %from,
            "London Strategic Edge clamped the requested replay window; events between the two \
             instants are beyond the provider's retention and are permanently lost",
        );
    }

    /// The instant a window was requested from, if the provider opened it later than that.
    ///
    /// `None` for a subscription this stream holds no watermark for — one that asked for no window.
    /// An entry is never removed once made, so a boundary frame that arrives *after* the replay has
    /// already passed the watermark is still answered correctly: the provider is measured to
    /// announce the window first, on both a crypto and an FX symbol, but under a clamp every
    /// replayed tick sits past the watermark by definition, so a single reordered frame would
    /// otherwise be enough to lose the only evidence the window was cut short.
    fn clamped_from(
        &self,
        subscription_id: &SubscriptionId,
        from: DateTime<Utc>,
    ) -> Option<DateTime<Utc>> {
        let requested = self.pending.get(subscription_id)?.time_exchange;

        // A window opened *at* the requested instant is not a clamp: `start` is inclusive, so that
        // is precisely the window that was asked for, and warning on it would make this fire on
        // every ordinary resume.
        (from > requested).then_some(requested)
    }

    /// Advance the shared watermark for an event this stream emitted.
    fn record(&self, subscription_id: SubscriptionId, time_exchange: DateTime<Utc>) {
        self.state.record(
            &LseResumeKey::new(self.exchange, subscription_id, self.kind),
            time_exchange,
        );
    }
}

impl UnregisteredContracts {
    fn new(exchange: ExchangeId, registered: FnvHashSet<SubscriptionId>) -> Self {
        Self {
            exchange,
            registered,
            tally: BTreeMap::new(),
            window_start: Instant::now(),
        }
    }

    /// Count a print if nobody registered its contract. `true` means it was counted and must be
    /// dropped; `false` means it is registered and passes through.
    fn absorb(&mut self, subscription_id: &SubscriptionId) -> bool {
        if self.registered.contains(subscription_id) {
            return false;
        }

        // A subscription identifier is the symbol, and every symbol on this channel is an OSI
        // contract, so the root is always found; were one not, it is tallied under itself rather
        // than lost.
        let symbol = subscription_id.as_ref();
        let underlying = osi::root(symbol).unwrap_or(symbol);

        // Looked up before inserting so the common case -- an underlying already tallied this
        // window -- builds no owned key.
        let tally = match self.tally.get_mut(underlying) {
            Some(tally) => tally,
            None => self.tally.entry(SmolStr::new(underlying)).or_default(),
        };
        tally.prints += 1;
        tally.contracts.insert(subscription_id.clone());

        true
    }

    /// Report what was counted and start a new window, if the current one has run its course.
    fn report_if_due(&mut self, now: Instant) {
        let window = now.saturating_duration_since(self.window_start);
        if window < UNREGISTERED_REPORT_INTERVAL {
            return;
        }

        for (underlying, tally) in std::mem::take(&mut self.tally) {
            info!(
                exchange = %self.exchange,
                %underlying,
                prints = tally.prints,
                contracts = tally.contracts.len(),
                window_secs = window.as_secs(),
                "London Strategic Edge delivered option prints for contracts nobody registered, \
                 and dropped them; register a contract to receive it, and if one you registered on \
                 this underlying is silent, check its OSI spelling, since a misspelled contract's \
                 prints are counted here",
            );
        }

        self.window_start = now;
    }
}

/// Bound required by [`StatelessTransformer`]'s decode path, restated here so this module's own
/// bounds stay legible.
const _: fn() = || {
    fn assert_identifiable<T: Identifier<Option<SubscriptionId>> + for<'de> Deserialize<'de>>() {}
    assert_identifiable::<LseMessage>();
};

#[cfg(test)]
#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
mod tests {
    use super::*;
    use crate::{exchange::lse::LseCrypto, subscription::trade::PublicTrades};
    use rustrade_integration::subscription::SubscriptionId;

    type Subject = LseTransformer<LseCrypto, u8, PublicTrades>;

    /// The kind these tests serve, and the partition their watermarks are filed under.
    const KIND: &str = "public_trades";

    fn id() -> SubscriptionId {
        super::super::mapper::subscription_id("BTC/USD")
    }

    fn key() -> LseResumeKey {
        LseResumeKey::new(ExchangeId::LseCrypto, id(), KIND)
    }

    /// A live tick, which carries no `replay` key at all.
    fn tick(ts: &str, price: &str) -> LseMessage {
        serde_json::from_str(&format!(
            r#"{{"type":"tick","symbol":"BTC/USD","ts":"{ts}","price":{price},
                "bid":{price},"ask":42001.0,"volume":0.00155}}"#
        ))
        .unwrap()
    }

    /// A tick served from the provider's replay buffer, which the provider stamps.
    fn replayed_tick(ts: &str, price: &str) -> LseMessage {
        serde_json::from_str(&format!(
            r#"{{"type":"tick","symbol":"BTC/USD","ts":"{ts}","replay":true,"price":{price},
                "bid":{price},"ask":42001.0,"volume":0.00155}}"#
        ))
        .unwrap()
    }

    /// A stream on a connection of its own: any window opens exactly at its own watermark.
    async fn subject(resume: Option<Arc<LseResumeState>>) -> Subject {
        let starts = resume
            .as_ref()
            .and_then(|state| state.watermark(&key()))
            .map(|watermark| (id(), watermark.time_exchange));

        subject_with(resume, starts).await
    }

    /// A stream whose window for [`id`] the connection opened at `start`, or never opened.
    async fn subject_with(
        resume: Option<Arc<LseResumeState>>,
        start: Option<(SubscriptionId, DateTime<Utc>)>,
    ) -> Subject {
        let map = Map([(id(), 1_u8)]
            .into_iter()
            .collect::<FnvHashMap<SubscriptionId, u8>>());
        let (tx, _rx) = mpsc::unbounded_channel();

        let resume = resume.map(|state| ResumeContext {
            state,
            kind: KIND,
            starts: start.into_iter().collect(),
        });

        Subject::new(map, &[], tx, resume).await.unwrap()
    }

    /// A symbol has one replay window per connection, opened at the earliest watermark among the
    /// streams holding it. A stream that had delivered further drops what it already delivered —
    /// silently, and exactly — and emits everything past its own watermark.
    #[tokio::test]
    async fn a_window_opened_before_the_watermark_drops_what_this_stream_already_delivered() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));
        first.transform(tick("2026-01-02T09:00:00.000002Z", "2.0"));
        first.transform(tick("2026-01-02T09:00:00.000002Z", "3.0"));

        // Another stream on the connection resumes from further back.
        let mut second = subject_with(
            Some(Arc::clone(&state)),
            Some((id(), at("2026-01-02T09:00:00Z"))),
        )
        .await;

        for (ts, price) in [
            ("2026-01-02T09:00:00.000000Z", "0.5"),
            ("2026-01-02T09:00:00.000001Z", "1.0"),
            ("2026-01-02T09:00:00.000002Z", "2.0"),
            ("2026-01-02T09:00:00.000002Z", "3.0"),
        ] {
            assert!(
                second.transform(replayed_tick(ts, price)).is_empty(),
                "{ts} was delivered before the connection was lost"
            );
        }

        assert_eq!(
            second
                .transform(replayed_tick("2026-01-02T09:00:00.000003Z", "4.0"))
                .len(),
            1,
            "the first tick past this stream's own watermark is new"
        );
    }

    /// The silent drop covers the replay window only. Once a tick passes the watermark, an older
    /// one is the monotonicity contradiction it always was: emitted, not swallowed.
    #[tokio::test]
    async fn an_older_tick_after_the_window_closes_is_emitted_even_when_it_opened_earlier() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000002Z", "1.0"));

        let mut second = subject_with(
            Some(Arc::clone(&state)),
            Some((id(), at("2026-01-02T09:00:00Z"))),
        )
        .await;
        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000003Z", "2.0"))
                .len(),
            1
        );
        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000001Z", "0.5"))
                .len(),
            1,
            "the window is closed; an older tick must be emitted, not dropped"
        );
    }

    /// A stream joining a symbol the connection already streams is given no replay, so it must not
    /// expect one: nothing is skipped at its watermark, and nothing is reported as a shortfall.
    #[tokio::test]
    async fn a_subscription_with_no_window_opened_skips_nothing() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));

        let mut joined = subject_with(Some(Arc::clone(&state)), None).await;
        assert!(joined.resume.as_ref().unwrap().pending.is_empty());
        assert_eq!(
            joined
                .transform(tick("2026-01-02T09:00:00.000001Z", "1.0"))
                .len(),
            1
        );
    }

    #[tokio::test]
    async fn without_resume_every_tick_is_emitted() {
        let mut subject = subject(None).await;

        assert_eq!(
            subject.transform(tick("2026-01-02T09:00:00Z", "1.0")).len(),
            1
        );
        assert_eq!(
            subject.transform(tick("2026-01-02T09:00:00Z", "1.0")).len(),
            1
        );
    }

    #[tokio::test]
    async fn with_resume_the_watermark_advances_on_emitted_ticks() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(Arc::clone(&state))).await;

        subject.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));
        subject.transform(tick("2026-01-02T09:00:00.000001Z", "2.0"));

        let watermark = state.watermark(&key()).unwrap();
        assert_eq!(
            watermark.time_exchange,
            "2026-01-02T09:00:00.000001Z"
                .parse::<DateTime<Utc>>()
                .unwrap()
        );
        assert_eq!(watermark.count_at_time, 2);
    }

    /// The core of the mechanism: a reconnect replays the whole of the watermark's instant because
    /// `start` is inclusive, and exactly the already-delivered prefix is skipped.
    #[tokio::test]
    async fn a_replay_skips_exactly_the_ticks_already_delivered_at_the_resume_instant() {
        let state = Arc::new(LseResumeState::new());

        // First connection delivers three ticks sharing one instant.
        let mut first = subject(Some(Arc::clone(&state))).await;
        for price in ["1.0", "2.0", "3.0"] {
            first.transform(tick("2026-01-02T09:00:00.000001Z", price));
        }
        assert_eq!(state.watermark(&key()).unwrap().count_at_time, 3);

        // Reconnect: the provider replays all three, then a fourth that is genuinely new.
        let mut second = subject(Some(Arc::clone(&state))).await;
        for price in ["1.0", "2.0", "3.0"] {
            assert!(
                second
                    .transform(replayed_tick("2026-01-02T09:00:00.000001Z", price))
                    .is_empty(),
                "an already-delivered tick must not be emitted twice"
            );
        }

        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000001Z", "4.0"))
                .len(),
            1,
            "a fourth tick at the same instant is new and must be emitted"
        );
        assert_eq!(state.watermark(&key()).unwrap().count_at_time, 4);
    }

    /// The gap half of the contract: everything after the resume instant is new, always.
    #[tokio::test]
    async fn a_replay_emits_every_tick_past_the_resume_instant() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));

        let mut second = subject(Some(Arc::clone(&state))).await;
        assert!(
            second
                .transform(replayed_tick("2026-01-02T09:00:00.000001Z", "1.0"))
                .is_empty()
        );
        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000002Z", "2.0"))
                .len(),
            1,
            "one microsecond later is past the resume instant and must be emitted"
        );
    }

    /// The drop window closes for good once the instant is passed, so a later tick that happens to
    /// carry the resume instant again cannot be swallowed.
    #[tokio::test]
    async fn the_drop_window_does_not_reopen_after_the_resume_instant_is_passed() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));
        first.transform(tick("2026-01-02T09:00:00.000001Z", "2.0"));

        let mut second = subject(Some(Arc::clone(&state))).await;
        // Replay is short by one, then moves past the instant -- the window closes here.
        assert!(
            second
                .transform(replayed_tick("2026-01-02T09:00:00.000001Z", "1.0"))
                .is_empty()
        );
        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000002Z", "2.0"))
                .len(),
            1
        );
        assert_eq!(
            second
                .transform(tick("2026-01-02T09:00:00.000001Z", "3.0"))
                .len(),
            1,
            "the window is closed; a tick at the old instant must be emitted, not dropped"
        );
    }

    /// A subscription with no history resumes as a fresh subscription would.
    #[tokio::test]
    async fn a_subscription_with_no_watermark_drops_nothing() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(state)).await;

        assert_eq!(
            subject.transform(tick("2026-01-02T09:00:00Z", "1.0")).len(),
            1
        );
    }

    #[tokio::test]
    async fn a_control_frame_is_neither_emitted_nor_recorded() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(Arc::clone(&state))).await;

        let frame: LseMessage =
            serde_json::from_str(r#"{"type":"replay_complete","symbol":"BTC/USD","rows":41}"#)
                .unwrap();

        assert!(subject.transform(frame).is_empty());
        assert_eq!(state.watermark(&key()), None);
    }

    /// The no-dedupe lock, restated at the transformer: identical repeats are genuine distinct
    /// prints and only the resume prefix may ever be skipped.
    #[tokio::test]
    async fn identical_live_ticks_are_never_deduplicated() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(state)).await;

        let first = subject.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));
        let second = subject.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));

        assert_eq!(first.len(), 1);
        assert_eq!(second.len(), 1, "an identical repeat is a distinct print");
    }

    /// An inclusive `start` says this cannot happen, so it means the assumption behind positional
    /// skipping has been contradicted — most plausibly a `start` filter flooring to a coarser
    /// resolution. Emitting is the safe answer: dropping would discard a real event on the strength
    /// of an assumption already known to be wrong. A `warn!` carries the signal.
    #[tokio::test]
    async fn a_replayed_tick_older_than_the_resume_instant_is_emitted_rather_than_dropped() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000002Z", "1.0"));

        let mut second = subject(Some(Arc::clone(&state))).await;
        assert_eq!(
            second
                .transform(replayed_tick("2026-01-02T09:00:00.000001Z", "0.5"))
                .len(),
            1,
            "a tick below the watermark must be emitted, not silently swallowed"
        );

        // The window itself is untouched by it: the tick at the watermark is still skipped.
        assert!(
            second
                .transform(replayed_tick("2026-01-02T09:00:00.000002Z", "1.0"))
                .is_empty()
        );
    }

    /// The skip stays positional rather than conditioning on `replay`, because the flag is absent
    /// on live ticks: a provider that stopped stamping replayed ones would otherwise turn every
    /// reconnect into a duplicate flood. The cost is this case, which is reported by a `warn!`
    /// rather than changed.
    #[tokio::test]
    async fn an_unstamped_tick_inside_the_resume_window_is_still_skipped() {
        let state = Arc::new(LseResumeState::new());

        let mut first = subject(Some(Arc::clone(&state))).await;
        first.transform(tick("2026-01-02T09:00:00.000001Z", "1.0"));

        let mut second = subject(Some(Arc::clone(&state))).await;
        assert!(
            second
                .transform(tick("2026-01-02T09:00:00.000001Z", "1.0"))
                .is_empty()
        );
    }

    /// A resumed subject whose `pending` holds a watermark at `instant`, as a reconnect's would.
    async fn resumed_at(state: &Arc<LseResumeState>, instant: &str) -> Subject {
        let mut first = subject(Some(Arc::clone(state))).await;
        first.transform(tick(instant, "1.0"));

        subject(Some(Arc::clone(state))).await
    }

    fn at(spelling: &str) -> DateTime<Utc> {
        spelling.parse::<DateTime<Utc>>().unwrap()
    }

    /// The boundary frame announcing a window opening at `from`.
    fn boundary(from: &str) -> LseMessage {
        serde_json::from_str(&format!(
            r#"{{"type":"replay_started","symbol":"BTC/USD","from":"{from}"}}"#
        ))
        .unwrap()
    }

    /// This subject's per-connection bookkeeping for the one subscription these tests use.
    fn pending_for(subject: &Subject) -> PendingDrop {
        *subject
            .resume
            .as_ref()
            .unwrap()
            .pending
            .get(&id())
            .unwrap_or_else(|| panic!("the resumed subject must hold an entry for {}", id()))
    }

    /// The provider retains 24 hours and moves an older `start` forward with no error, so the
    /// boundary frame's `from` is the only evidence the requested window was not the served one.
    #[tokio::test]
    async fn a_replay_window_opened_later_than_requested_is_reported() {
        let state = Arc::new(LseResumeState::new());
        let subject = resumed_at(&state, "2026-08-12T10:00:00Z").await;

        assert_eq!(
            subject
                .resume
                .as_ref()
                .unwrap()
                .clamped_from(&id(), at("2026-08-13T10:00:00Z")),
            Some(at("2026-08-12T10:00:00Z")),
        );
    }

    /// `start` is inclusive, so a window opened exactly where it was asked for is the window that
    /// was asked for — reporting it would make the check noise on every ordinary resume.
    #[tokio::test]
    async fn a_replay_window_opened_where_requested_is_not_a_clamp() {
        let state = Arc::new(LseResumeState::new());
        let subject = resumed_at(&state, "2026-08-14T10:16:55.161234Z").await;

        assert_eq!(
            subject
                .resume
                .as_ref()
                .unwrap()
                .clamped_from(&id(), at("2026-08-14T10:16:55.161234Z")),
            None,
        );
    }

    /// A subscription that asked for no window cannot have had one clamped.
    #[tokio::test]
    async fn a_replay_boundary_for_a_subscription_with_no_watermark_is_ignored() {
        let state = Arc::new(LseResumeState::new());
        let subject = subject(Some(state)).await;

        assert_eq!(
            subject
                .resume
                .as_ref()
                .unwrap()
                .clamped_from(&id(), at("2026-08-13T10:00:00Z")),
            None,
        );
    }

    /// A clamp explains the shortfall at the resume instant completely — the window does not reach
    /// it — so the shortfall must not *also* be reported as the provider's history changing under
    /// the subscription. Two reports with different causes for one event is worse than one, because
    /// only one of them is true.
    #[tokio::test]
    async fn a_clamped_window_is_not_also_reported_as_a_changed_history() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = resumed_at(&state, "2026-08-12T10:00:00Z").await;

        // The provider answers with a window opening a day later than asked for.
        subject.transform(boundary("2026-08-13T10:00:00Z"));

        let pending = pending_for(&subject);
        assert!(
            pending.clamped,
            "the clamp must be recorded, not only logged"
        );
        assert_eq!(
            pending.remaining, 1,
            "the drop window is still open until a tick passes the resume instant",
        );

        // Every replayed tick under a clamp sits past the watermark by definition, so the first one
        // closes the window with the count unspent. That shortfall has one cause and it is the
        // clamp.
        assert_eq!(
            subject
                .transform(replayed_tick("2026-08-13T10:00:00.000001Z", "1.0"))
                .len(),
            1,
        );

        let pending = pending_for(&subject);
        assert_eq!(pending.remaining, 0, "the drop window must be closed");
        assert!(
            pending.clamped,
            "the clamp flag is what suppresses the wrong-cause report and must survive the frame \
             that closes the window",
        );
    }

    /// The boundary frame is measured to arrive first, but a single reordered frame must not lose
    /// the only evidence the window was cut short. The entry is therefore kept rather than removed
    /// once the replay passes the watermark, so the clamp is still detected when the first replayed
    /// tick beats its own boundary frame.
    #[tokio::test]
    async fn a_clamp_is_still_detected_when_a_replayed_tick_beats_the_boundary_frame() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = resumed_at(&state, "2026-08-12T10:00:00Z").await;

        // The tick arrives first and closes the drop window, as every tick under a clamp does.
        assert_eq!(
            subject
                .transform(replayed_tick("2026-08-13T10:00:00.000001Z", "1.0"))
                .len(),
            1,
        );

        // The boundary frame follows. The requested instant must still be readable.
        assert_eq!(
            subject
                .resume
                .as_ref()
                .unwrap()
                .clamped_from(&id(), at("2026-08-13T10:00:00Z")),
            Some(at("2026-08-12T10:00:00Z")),
            "the requested instant is the only record of what was asked for and must outlive the \
             drop window",
        );
    }

    /// A rejection raised after the handshake stopped reading carries no instrument and no instant.
    /// It is consumed for its `warn!` and must not reach a decoder or the watermark.
    #[tokio::test]
    async fn a_rejection_yields_no_market_event_and_does_not_advance_the_watermark() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(Arc::clone(&state))).await;

        let rejection: LseMessage = serde_json::from_str(
            r#"{"type":"error","code":"INVALID_START","message":"Invalid start"}"#,
        )
        .unwrap();

        assert!(subject.transform(rejection).is_empty());
        assert_eq!(state.watermark(&key()), None);
    }

    /// The frame is the only evidence a requested window was clamped, and it reaches the stream
    /// rather than the handshake. It must yield no market event of its own.
    mod unregistered_contracts {
        use super::*;
        use crate::exchange::lse::LseOptions;

        type OptionsSubject = LseTransformer<LseOptions, u8, PublicTrades>;

        const REGISTERED: &str = "SPY260930C00700000";

        fn option_tick(symbol: &str) -> LseMessage {
            serde_json::from_str(&format!(
                r#"{{"type":"tick","symbol":"{symbol}","ts":"2026-01-02T15:00:00+00:00",
                    "price":1.25,"bid":null,"ask":null,"volume":3,"name":"label"}}"#
            ))
            .unwrap()
        }

        async fn options_subject() -> OptionsSubject {
            let map = Map([(
                super::super::super::mapper::subscription_id(REGISTERED),
                1_u8,
            )]
            .into_iter()
            .collect::<FnvHashMap<SubscriptionId, u8>>());
            let (tx, _rx) = mpsc::unbounded_channel();

            OptionsSubject::new(map, &[], tx, None).await.unwrap()
        }

        fn tally(subject: &OptionsSubject) -> Vec<(String, u64, usize)> {
            subject
                .unregistered
                .as_ref()
                .unwrap()
                .tally
                .iter()
                .map(|(underlying, tally)| {
                    (underlying.to_string(), tally.prints, tally.contracts.len())
                })
                .collect()
        }

        #[tokio::test]
        async fn a_registered_contract_is_emitted() {
            let mut subject = options_subject().await;

            let output = subject.transform(option_tick(REGISTERED));
            assert_eq!(output.len(), 1);
            assert!(output[0].is_ok());
            assert!(tally(&subject).is_empty());
        }

        /// The chain delivers thousands of contracts nobody asked for. Each one surfaced as an
        /// error would bury every real error on the stream.
        #[tokio::test]
        async fn an_unregistered_contract_is_counted_rather_than_raised() {
            let mut subject = options_subject().await;

            assert!(
                subject
                    .transform(option_tick("SPY260930C00701000"))
                    .is_empty()
            );
            assert!(
                subject
                    .transform(option_tick("SPY260930C00701000"))
                    .is_empty()
            );
            assert!(
                subject
                    .transform(option_tick("SPY260930P00650000"))
                    .is_empty()
            );
            assert!(
                subject
                    .transform(option_tick("QQQ260930C00500000"))
                    .is_empty()
            );

            assert_eq!(
                tally(&subject),
                [("QQQ".to_owned(), 1, 1), ("SPY".to_owned(), 3, 2)]
            );
        }

        #[tokio::test]
        async fn the_count_is_kept_until_the_interval_has_run() {
            let mut subject = options_subject().await;
            subject.transform(option_tick("SPY260930C00701000"));

            let unregistered = subject.unregistered.as_mut().unwrap();
            let opened = unregistered.window_start;

            unregistered.report_if_due(opened + UNREGISTERED_REPORT_INTERVAL / 2);
            assert_eq!(tally(&subject).len(), 1);
        }

        #[tokio::test]
        async fn a_report_starts_a_new_window() {
            let mut subject = options_subject().await;
            subject.transform(option_tick("SPY260930C00701000"));

            let unregistered = subject.unregistered.as_mut().unwrap();
            let due = unregistered.window_start + UNREGISTERED_REPORT_INTERVAL;

            unregistered.report_if_due(due);
            assert!(tally(&subject).is_empty());
            assert_eq!(subject.unregistered.as_ref().unwrap().window_start, due);
        }

        /// Only the options dataset fans out, so only it counts. Elsewhere an unregistered symbol
        /// is a genuine mismatch and stays an error.
        #[tokio::test]
        async fn other_datasets_keep_no_count() {
            assert!(subject(None).await.unregistered.is_none());
        }
    }

    #[tokio::test]
    async fn a_replay_boundary_is_consumed_without_emitting_an_event() {
        let state = Arc::new(LseResumeState::new());
        let mut subject = subject(Some(Arc::clone(&state))).await;

        let boundary: LseMessage = serde_json::from_str(
            r#"{"type":"replay_started","symbol":"BTC/USD","from":"2026-01-02T09:00:00Z"}"#,
        )
        .unwrap();

        assert!(subject.transform(boundary).is_empty());
        assert_eq!(
            state.watermark(&key()),
            None,
            "a boundary frame must not advance the watermark"
        );
    }
}