liminal-server 0.23.1

//! What a tap carries: the filter, the grant, the publish it is shown, and the
//! per-message facts it keeps.

use liminal::channel::Recipient;
use std::num::NonZeroU32;
use std::sync::Arc;
use std::time::Instant;

/// Which messages a subscription wants: those whose subject (the channel
/// name) equals `subject`, or starts with its prefix when it ends in `*`.
/// An empty subject or a lone `*` matches everything.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TapFilter {
    /// Exact subject, or a prefix followed by `*`.
    pub subject: String,
}

impl TapFilter {
    /// Whether `subject` is inside this filter.
    #[must_use]
    pub fn matches(&self, subject: &str) -> bool {
        self.subject.strip_suffix('*').map_or_else(
            || self.subject.is_empty() || self.subject == subject,
            |prefix| subject.starts_with(prefix),
        )
    }
}

/// The right to see payloads on a tap. Nothing in the server mints one until
/// grants (SRV-011 R4) exist, so no payload leaves through a tap today.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PayloadGrant {
    _sealed: (),
}

impl PayloadGrant {
    /// Mints a grant. Test-only until R4.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn server_issued() -> Self {
        Self { _sealed: () }
    }
}

/// What became of a tapped publish, by the cause the server knows. A
/// dedup-suppressed re-publish is never tapped: its original was.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TapOutcome {
    /// Accepted and taken by at least one subscriber.
    Delivered,
    /// The correlated reply arrived and was released.
    Answered,
    /// No reply arrived before the call deadline.
    Unanswered,
    /// Accepted, and no subscriber matched it.
    NoSubscriber,
    /// Accepted; subscribers matched and none took it.
    NoneAccepted,
    /// The append's outcome is not known.
    Unresolved,
    /// Refused (admission, schema, or a library failure).
    Refused,
}

/// Who published.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TapSource {
    /// A connection, by its id.
    Connection(u64),
    /// An in-process publisher: the server or an embedded host named by its station.
    Server,
}

/// How a tapped message is named: by the server's message id when it was
/// admitted, by the tap's observation number when it was not.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TapIdent {
    /// The transport-scoped message id an admitted publish was given.
    Message(u64),
    /// A connection-local call operation, separate from channel message ids.
    Call {
        /// Calling connection.
        connection: u64,
        /// Correlation operation.
        operation: u64,
    },
    /// The tap's own count of the publish, for one that has no message id.
    Observation(u64),
}

/// One publish as the hook hands it to the tap. Borrowed throughout: the tap
/// copies only what an admitted message keeps.
#[derive(Clone, Copy, Debug)]
pub struct PublishFact<'a> {
    /// The subject (channel name).
    pub subject: &'a str,
    /// The payload; its length is the message's size, and it is copied only
    /// for a subscription holding a [`PayloadGrant`].
    pub payload: &'a [u8],
    /// What became of it.
    pub outcome: TapOutcome,
    /// The message id, when the publish was admitted and given one.
    pub message_id: Option<u64>,
    /// Who published.
    pub source: TapSource,
    /// The id of the message it answers (the envelope's causal parent).
    pub reply_to: Option<&'a str>,
    /// What the publish recorded; `None` when it recorded nothing: no tap
    /// was live when it began, or the tap could not lend it a record. The
    /// tap then first sees the message as it keeps it, and knows neither
    /// whom it reached nor when.
    pub recorded: Option<Recorded<'a>>,
}

/// What a recording publish saw, all together, so a fact can never carry
/// recipients' queue instants without the receipt they are measured from
/// (Hermes 7 Oct 06:55).
#[derive(Clone, Copy, Debug)]
pub struct Recorded<'a> {
    /// The subscriber pid of every local subscription whose inbox queued the
    /// message, with the instant it did, as the library's fan-out recorded
    /// them; empty when it reached none.
    pub recipients: &'a [Recipient],
    /// When the server received the publish, before admission: the message's
    /// first sight (lane brief ruling 6).
    pub received_at: Instant,
    /// When a durable channel's append of the message completed, before its
    /// fan-out; `None` for one that is not durable.
    pub appended_at: Option<Instant>,
}

/// A station a message reached: its id, shared rather than copied, and the
/// instant the first of its subscriptions' inboxes queued the message.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Reached {
    /// The station.
    pub station: Arc<str>,
    /// When its first subscription's inbox queued the message.
    pub queued_at: Instant,
}

/// One point a message crossed: the node, shared rather than copied, and how
/// long after the message was first seen.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Hop {
    /// The node the message crossed.
    pub node: Arc<str>,
    /// Milliseconds after the message was first seen.
    pub at_ms: u64,
}

/// The per-message facts a tap keeps.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TapMessage {
    /// Its id.
    pub ident: TapIdent,
    /// The subject (channel name).
    pub subject: String,
    /// Payload size in bytes.
    pub size: u64,
    /// What became of it.
    pub outcome: TapOutcome,
    /// When the server first saw it, on the monotonic clock; the record's
    /// wall time is the tap's wall base plus the elapsed time to here.
    pub seen_at: Instant,
    /// Who published it.
    pub source: TapSource,
    /// The station that published it: the registered worker identity of
    /// the publishing connection, shared rather than copied. `None` for a
    /// connection that is not a registered worker, and for the server. An embedded
    /// host carries its named publisher identity here.
    pub station: Option<Arc<str>>,
    /// The stations that received it: each registered worker whose
    /// subscription's inbox queued the message, once, in the fan-out's order,
    /// shared by every tap that kept it: empty when the recorded fan-out
    /// reached no registered worker. `None` when its recipients were not
    /// recorded, so whom it reached is not known.
    pub to_stations: Option<Arc<[Reached]>>,
    /// Where it went, in order.
    pub hops: Vec<Hop>,
    /// The id of the message it answers, when it has one.
    pub reply_to: Option<String>,
    /// Held by a rule. Always `false` until rules (R4) exist.
    pub held: bool,
    /// The payload, only on a subscription carrying a [`PayloadGrant`].
    pub payload: Option<Vec<u8>>,
    /// Terminal call facts; absent on a channel publish.
    pub call: Option<CallDetails>,
}

/// What a drain hands the subscriber: messages that were all admitted at
/// one divisor.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TapBatch {
    /// One more than this subscription's previous batch, from 1.
    pub seq: u64,
    /// N in "sampled 1 in N" for every message in this batch.
    pub sampled: NonZeroU32,
    /// Messages evicted from the subscription's ring since its previous
    /// batch.
    pub dropped: u64,
    /// The messages, oldest first.
    pub messages: Vec<TapMessage>,
}

/// Terminal call fields the console can read without a payload grant.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct CallDetails {
    /// Request receipt to correlated reply receipt, in milliseconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub latency_ms: Option<u64>,
    /// The server's named refusal, when it refused the call.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refusal: Option<String>,
}

/// One call as seen at its terminal correlation or deadline.
#[derive(Clone, Copy, Debug)]
pub struct CallFact<'a> {
    /// Called subject.
    pub subject: &'a str,
    /// Request payload bytes.
    pub size: u64,
    /// Calling connection.
    pub connection: u64,
    /// Pending-reply operation, when admitted.
    pub operation: Option<u64>,
    /// The calling station, captured at receipt.
    pub station: Option<&'a Arc<str>>,
    /// The serving station, captured at dispatch.
    pub served_by: Option<&'a Arc<str>>,
    /// Request receipt.
    pub received_at: Instant,
    /// Dispatch to the serving station.
    pub forwarded_at: Instant,
    /// Reply, refusal or deadline observation.
    pub finished_at: Instant,
    /// Terminal outcome.
    pub outcome: TapOutcome,
    /// The server's refusal words.
    pub refusal: Option<&'a str>,
}