liminal-server 0.25.4

//! Tap records preserve channel facts and add terminal call facts.

use super::RECORD_VERSION;
use crate::console::{ConsoleError, wall::WallBase};
use crate::tap::{self, TapIdent, TapOutcome, TapSource};
use serde::Serialize;
use std::num::NonZeroU32;

/// What became of a tapped message, in the console's words.
///
/// Channel delivery and terminal call correlation name their outcomes here.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Outcome {
    /// Taken by at least one subscriber.
    Delivered,
    /// The correlated call reply arrived.
    Answered,
    /// No reply arrived before the call deadline.
    Unanswered,
    /// Admitted, and no subscriber matched.
    Nobody,
    /// Admitted; subscribers matched and none took it.
    Unaccepted,
    /// Refused at admission, by the schema, or by the library.
    Refused,
    /// The store could not say whether the append landed.
    Unresolved,
}

impl From<TapOutcome> for Outcome {
    fn from(outcome: TapOutcome) -> Self {
        match outcome {
            TapOutcome::Answered => Self::Answered,
            TapOutcome::Unanswered => Self::Unanswered,
            TapOutcome::Delivered => Self::Delivered,
            TapOutcome::NoSubscriber => Self::Nobody,
            TapOutcome::NoneAccepted => Self::Unaccepted,
            TapOutcome::Unresolved => Self::Unresolved,
            TapOutcome::Refused => Self::Refused,
        }
    }
}

/// One point a message crossed.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Hop {
    /// The device id.
    pub node: String,
    /// Milliseconds after the message's `t`.
    pub at_ms: u64,
}

/// The facts a tap carries about one message. Payloads never ride here.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct TapMessage {
    /// `<device>:<message id>` when admitted, `<device>:x<observation>`
    /// otherwise.
    pub id: String,
    /// The subject (channel name).
    pub subject: String,
    /// Payload size in bytes.
    pub size: u64,
    /// What became of it.
    pub outcome: Outcome,
    /// Wall time it was first seen, in milliseconds since the epoch.
    pub t: u64,
    /// The publishing connection's id, or `liminal.<device>` for the server.
    pub from: String,
    /// The station that published it: the registered worker identity of the
    /// publishing connection. Absent when that connection is not a registered
    /// worker, and for the server's own records.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from_station: Option<String>,
    /// This device's id.
    pub to: String,
    /// The stations that received it: each registered worker whose
    /// subscription's inbox on this device queued the message, once, in
    /// delivery order. A subscriber whose inbox refused it is not here; nor is
    /// one that is no registered worker (it is counted in the outcome, never
    /// named), nor a receiver on another device. `[]` when the fan-out was
    /// recorded and reached no registered worker here; absent when it was not
    /// recorded (a tap went live while the publish ran, or the tap could not
    /// lend it a buffer), so a reader shows its receivers as not known rather
    /// than as none.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub to_stations: Option<Vec<String>>,
    /// Where it went, in order.
    pub hops: Vec<Hop>,
    /// The id of the message it answers (its causal parent), when it has one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reply_to: Option<String>,
    /// Held by a rule. Rules do not exist yet, so always false.
    pub held: bool,
    /// Where a rule diverted it. Rules do not exist yet, so always absent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub diverted: Option<String>,
    /// Measured round trip and named refusal of a call.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fields: Option<tap::message::CallDetails>,
}

/// One `console.tap.<target>` record: one batch, every message in it admitted
/// at the one divisor `sampled`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct TapBatch {
    v: u64,
    #[serde(rename = "type")]
    record_type: &'static str,
    /// The node or stream the tap is on.
    pub target: String,
    /// One more than this tap'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 tap's ring since its previous batch.
    pub dropped: u64,
    /// The messages, oldest first.
    pub messages: Vec<TapMessage>,
}

impl TapBatch {
    /// A batch for `target`.
    #[must_use]
    pub const fn new(
        target: String,
        seq: u64,
        sampled: NonZeroU32,
        dropped: u64,
        messages: Vec<TapMessage>,
    ) -> Self {
        Self {
            v: RECORD_VERSION,
            record_type: "tap.batch",
            target,
            seq,
            sampled,
            dropped,
            messages,
        }
    }
}

/// The `console.tap.<target>` record for one tap batch on `device`.
///
/// Every message is named, timed and attributed in the console's terms. A
/// payload, which only a test grant can carry today, does not ride the
/// record: the console's reader names no field for one.
///
/// # Errors
/// Returns [`ConsoleError::Clock`] when a message's time passes u64
/// milliseconds.
pub fn tap_batch_record(
    device: &str,
    target: &str,
    batch: &tap::TapBatch,
    wall: &WallBase,
) -> Result<TapBatch, ConsoleError> {
    let messages = batch
        .messages
        .iter()
        .map(|message| tap_message(device, message, wall))
        .collect::<Result<Vec<_>, _>>()?;
    Ok(TapBatch::new(
        target.to_owned(),
        batch.seq,
        batch.sampled,
        batch.dropped,
        messages,
    ))
}

fn tap_message(
    device: &str,
    message: &tap::TapMessage,
    wall: &WallBase,
) -> Result<TapMessage, ConsoleError> {
    Ok(TapMessage {
        id: match message.ident {
            TapIdent::Call {
                connection,
                operation,
            } => format!("{device}:call:{connection}:{operation}"),
            TapIdent::Message(id) => format!("{device}:{id}"),
            TapIdent::Observation(observation) => format!("{device}:x{observation}"),
        },
        subject: message.subject.clone(),
        size: message.size,
        outcome: message.outcome.into(),
        t: wall.unix_ms_at(message.seen_at)?,
        from: match message.source {
            TapSource::Connection(connection) => connection.to_string(),
            TapSource::Server => message
                .station
                .as_deref()
                .map_or_else(|| format!("liminal.{device}"), str::to_owned),
        },
        from_station: message.station.as_deref().map(str::to_owned),
        to: device.to_owned(),
        to_stations: message.to_stations.as_deref().map(|stations| {
            stations
                .iter()
                .map(|reached| reached.station.to_string())
                .collect()
        }),
        hops: message
            .hops
            .iter()
            .map(|hop| Hop {
                node: hop.node.to_string(),
                at_ms: hop.at_ms,
            })
            .collect(),
        reply_to: message.reply_to.clone(),
        held: message.held,
        diverted: None,
        fields: message.call.clone(),
    })
}