liminal-server 0.24.1

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Instant;

use tokio::sync::Notify;

use liminal::channel::PublishRecord;
use liminal::protocol::MessageEnvelope;

use crate::ServerError;
use crate::console::ConsoleHub;
use crate::server::connection::services::{PublishOutcome, Reach};
use crate::server::self_streams::cost::CostWake;
use crate::server::self_streams::events::{PublishOrigin, ServerEvent};
use crate::server::self_streams::sink::{OriginPublication, SelfStreams};

#[derive(Clone, Debug, Default)]
pub(in crate::server) struct ServerSignals {
    state: Arc<SignalState>,
}

#[derive(Debug, Default)]
struct SignalState {
    wake: Arc<CostWake>,
    streams: OnceLock<Arc<SelfStreams>>,
    console: OnceLock<Arc<ConsoleHub>>,
    /// Woken by the console hub when it has records due; the self-stream
    /// publisher loop waits on it (SRV-012 R4).
    console_wake: Arc<Notify>,
    /// Set once a publisher owner runs the console streams.
    console_enabled: AtomicBool,
    connections_opened: AtomicU64,
    connections_closed: AtomicU64,
    already_crossed: std::sync::Mutex<std::collections::BTreeMap<String, u64>>,
    /// Server events whose `liminal.events` row has been queued (or
    /// refused), counted after the enqueue, for a test to wait on. Test-only.
    #[cfg(test)]
    events_written: (std::sync::Mutex<u64>, std::sync::Condvar),
}

/// The console's counters and the signals' own, read at one moment
/// (SRV-012 R8). Every console bookkeeping failure is one of the named
/// failure counters in `console`, each also `liminal_console_failures_total`.
/// Test-only: production reads these through the metrics.
#[cfg(test)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct SignalCounts {
    /// The console hub's counters.
    pub console: crate::console::ConsoleCountsSnapshot,
    /// Publishes the tap was shown while a tap subscription was live.
    pub tap_observations: u64,
    /// Of those, the server's own (its self-stream rows).
    pub tap_server_observations: u64,
    /// Connections registered with the supervisor.
    pub connections_opened: u64,
    /// Connections removed from the supervisor.
    pub connections_closed: u64,
}

impl ServerSignals {
    pub(in crate::server) fn console(&self) -> Option<&Arc<ConsoleHub>> {
        self.state.console.get()
    }

    pub(in crate::server) fn already_crossed(
        &self,
        link_id: &str,
        life: &crate::link::LinkLife,
    ) -> Result<(), String> {
        use crate::console::records::{Carrier, Link, LinkCounters, LinkState};
        let console = self
            .state
            .console
            .get()
            .ok_or("crossing console is not installed")?;
        let mut counters = self
            .state
            .already_crossed
            .lock()
            .map_err(|_| "crossing counter lock poisoned")?;
        let count = counters.entry(link_id.to_owned()).or_default();
        *count = count
            .checked_add(1)
            .ok_or("already crossed counter overflow")?;
        if *count == 1 {
            tracing::info!(
                link = link_id,
                "already_crossed: crossing answered from its publish receipt"
            );
        }
        let reason = life.reason()?;
        // Keep topology observations ordered with their counter increments.
        let observed = console
            .emit(crate::console::TopologyChange::Link {
                link_id: Link::to_peer(console.device(), life.peer()).id,
                peer: life.peer().to_owned(),
                carrier: Some(Carrier::Wire),
                state: if reason.is_some() {
                    LinkState::Down
                } else {
                    LinkState::Up
                },
                reason,
                counters: LinkCounters {
                    already_crossed: Some(*count),
                    ..LinkCounters::default()
                },
            })
            .map_err(|error| format!("crossing topology observation failed: {error}"));
        drop(counters);
        observed
    }

    pub(in crate::server) fn observe_link(
        &self,
        life: &crate::link::LinkLife,
    ) -> Result<(), String> {
        if let Some(console) = self.state.console.get() {
            let signals = self.clone();
            life.observe(
                console.device().to_owned(),
                Arc::new(move |change| signals.topology(change)),
            )?;
        }
        Ok(())
    }

    pub(in crate::server) fn wake(&self) -> Arc<CostWake> {
        Arc::clone(&self.state.wake)
    }

    pub(in crate::server) fn install(&self, streams: Arc<SelfStreams>) -> Result<(), ServerError> {
        streams.start()?;
        self.state
            .streams
            .set(streams)
            .map_err(|_| SelfStreams::failure("self-stream publisher was already installed"))
    }

    /// Installs the console hub whose tap sees every publish (SRV-012 R3).
    pub(in crate::server) fn install_console(
        &self,
        console: Arc<ConsoleHub>,
    ) -> Result<(), ServerError> {
        self.state
            .console
            .set(console)
            .map_err(|_| SelfStreams::failure("console hub was already installed"))
    }

    /// Turns the console streams on: the hub's wake now reaches the
    /// publisher loop, which pumps the console on its own thread.
    ///
    /// # Errors
    /// Returns a named failure when no hub is installed or the hub's
    /// notifier slot is poisoned.
    pub(in crate::server) fn enable_console(&self) -> Result<(), ServerError> {
        let console = self
            .state
            .console
            .get()
            .ok_or_else(|| SelfStreams::failure("console hub is not installed"))?;
        let wake = Arc::clone(&self.state.console_wake);
        console
            .set_notifier(Arc::new(move || wake.notify_one()))
            .map_err(|error| SelfStreams::failure(&error.to_string()))?;
        self.state.console_enabled.store(true, Ordering::Release);
        // A first pump at start only when a subscriber arrived before the
        // owner ran, so an idle server is never woken.
        console
            .wake_if_due()
            .map_err(|error| SelfStreams::failure(&error.to_string()))
    }

    /// Whether a publisher owner runs the console streams.
    pub(in crate::server) fn console_enabled(&self) -> bool {
        self.state.console_enabled.load(Ordering::Acquire)
    }

    /// The console's wake, once the console streams are on.
    pub(in crate::server) fn console_wake(&self) -> Option<Arc<Notify>> {
        self.console_enabled()
            .then(|| Arc::clone(&self.state.console_wake))
    }

    /// One topology change at the site that caused it, after every lock
    /// there is released (SRV-012 R7). A change the hub cannot apply is
    /// counted there (`topology_emit_failures`) and logged here by name.
    pub(in crate::server) fn topology(&self, change: crate::console::TopologyChange) {
        if let Some(console) = self.state.console.get()
            && let Err(error) = console.emit(change)
        {
            tracing::warn!(%error, "TOPOLOGY_EMIT_FAILED");
        }
    }

    /// Every publish on the one path, once its result is known: it arms the
    /// cost tick for a connection's work, and shows the console tap the
    /// publish (SRV-012 R3), which needs the envelope for the payload size
    /// and the causal parent (R14). A publish that recorded what it saw
    /// passes its record with the instant the server received it.
    pub(in crate::server) fn published(
        &self,
        channel: &str,
        origin: PublishOrigin<'_>,
        envelope: &MessageEnvelope,
        result: &Result<OriginPublication, ServerError>,
        recorded: Option<(&PublishRecord, Instant)>,
    ) {
        if !channel.starts_with("console.") && result.is_ok() {
            self.state.wake.arm(&origin);
        }
        if let Some(console) = self.state.console.get() {
            tap(console, channel, origin, envelope, result, recorded);
        }
    }

    /// A buffer for a publish to record its recipients in, only while a
    /// console tap is live; `None` otherwise, and then the publish records
    /// nothing (SRV-012 `to_stations`).
    pub(in crate::server) fn lend_recipients(&self) -> Option<PublishRecord> {
        let console = self.state.console.get()?;
        console.tap().lend_recipients().unwrap_or_else(|error| {
            console.observe_failed();
            tracing::warn!(%error, "TAP_RECIPIENTS_FAILED");
            None
        })
    }

    /// Takes back a buffer [`Self::lend_recipients`] lent.
    pub(in crate::server) fn give_back(&self, buffer: PublishRecord) {
        if let Some(console) = self.state.console.get()
            && let Err(error) = console.tap().give_back(buffer)
        {
            console.observe_failed();
            tracing::warn!(%error, "TAP_RECIPIENTS_FAILED");
        }
    }

    /// The library subscriber `subscriber` belongs to connection
    /// `connection`, so what it receives names that connection's station.
    pub(in crate::server) fn receiver(&self, subscriber: u64, connection: u64) {
        if let Some(console) = self.state.console.get()
            && let Err(error) = console.tap().name_receiver(subscriber, connection)
        {
            console.observe_failed();
            tracing::warn!(%error, "TAP_RECEIVER_FAILED");
        }
    }

    pub(in crate::server) fn event(&self, event: ServerEvent<'_>) {
        if let Some(streams) = self.state.streams.get() {
            streams.event(event);
        }
        #[cfg(test)]
        {
            let (written, changed) = &self.state.events_written;
            if let Ok(mut written) = written.lock() {
                *written += 1;
                changed.notify_all();
            }
        }
    }

    /// Waits up to `timeout` until `count` server events have had their row
    /// queued, answering whether they had. Test-only: the row is written
    /// later, by the owner, so a test that needs it written drains the queue
    /// after this.
    #[cfg(test)]
    pub(in crate::server) fn await_events(&self, count: u64, timeout: std::time::Duration) -> bool {
        let (written, changed) = &self.state.events_written;
        written
            .lock()
            .ok()
            .and_then(|guard| {
                changed
                    .wait_timeout_while(guard, timeout, |written| *written < count)
                    .ok()
            })
            .is_some_and(|(written, _)| *written >= count)
    }

    /// The installed self streams, kept available for their shutdown join.
    pub(in crate::server) fn streams(&self) -> Option<&Arc<SelfStreams>> {
        self.state.streams.get()
    }

    /// Every counter as it stands.
    #[cfg(test)]
    pub(in crate::server) fn counts(&self) -> SignalCounts {
        let console = self.state.console.get();
        SignalCounts {
            console: console.map(|console| console.counts()).unwrap_or_default(),
            tap_observations: console.map_or(0, |console| console.tap().observations()),
            tap_server_observations: console
                .map_or(0, |console| console.tap().server_observations()),
            connections_opened: self.state.connections_opened.load(Ordering::Relaxed),
            connections_closed: self.state.connections_closed.load(Ordering::Relaxed),
        }
    }

    pub(in crate::server) fn with_flush_barrier<T>(
        &self,
        flush: impl FnOnce() -> Result<T, ServerError>,
    ) -> Result<T, ServerError> {
        match self.state.streams.get() {
            Some(streams) => streams.with_flush_barrier(flush),
            None => flush(),
        }
    }

    pub(in crate::server) fn connection(&self, connection: u64, opened: bool) {
        let counter = if opened {
            &self.state.connections_opened
        } else {
            &self.state.connections_closed
        };
        counter.fetch_add(1, Ordering::Relaxed);
        self.event(if opened {
            ServerEvent::ConnectionOpened { connection }
        } else {
            ServerEvent::ConnectionClosed { connection }
        });
    }

    pub(in crate::server) fn lifecycle_batch<'a>(
        &self,
        conversation: u64,
        receipts: impl IntoIterator<Item = (u64, &'a str)>,
    ) -> Result<(), ServerError> {
        self.state.streams.get().map_or(Ok(()), |streams| {
            streams.lifecycle_batch(conversation, receipts)
        })
    }
}

/// SRV-011 R2: the tap's one hook. It drops by SUBJECT: nothing under
/// `console.` is observed, whatever its origin, so the console's own records
/// are never tapped, while `liminal.cost` and `liminal.events` stay tappable.
/// The fact borrows everything; the tap copies only what a kept message
/// holds (R11). Each message is attributed by its origin.
fn tap(
    console: &ConsoleHub,
    channel: &str,
    origin: PublishOrigin<'_>,
    envelope: &MessageEnvelope,
    result: &Result<OriginPublication, ServerError>,
    recorded: Option<(&PublishRecord, Instant)>,
) {
    if !crate::console::tappable(channel) {
        return;
    }
    let result = result.as_ref().map(|published| &published.outcome);
    let Some(outcome) = tap_outcome(result) else {
        // A dedup-suppressed re-publish reached nobody and is not tapped; it
        // is counted by name.
        if matches!(
            result,
            Ok(PublishOutcome {
                reach: Reach::Suppressed,
                ..
            })
        ) {
            console.suppressed();
        }
        return;
    };
    let fact = crate::tap::PublishFact {
        subject: channel,
        payload: &envelope.payload,
        outcome,
        message_id: result.ok().map(|published| published.message_id),
        source: match origin {
            PublishOrigin::Connection { id } => crate::tap::TapSource::Connection(id),
            PublishOrigin::Host { .. } | PublishOrigin::Server { .. } => {
                crate::tap::TapSource::Server
            }
        },
        reply_to: envelope
            .causal_context
            .parent_id
            .as_ref()
            .map(|parent| parent.0.as_str()),
        recorded: recorded.map(|(record, received_at)| crate::tap::Recorded {
            recipients: record.recipients(),
            received_at,
            appended_at: record.appended_at(),
        }),
    };
    let observed = match origin {
        PublishOrigin::Host { name } => console.tap().observe_embedded(&fact, name),
        PublishOrigin::Connection { .. } | PublishOrigin::Server { .. } => {
            console.tap().observe(&fact)
        }
    };
    if let Err(error) = observed {
        // Each failure of a Several counts on its own (Buttercup's V3).
        for _ in 0..error.failures() {
            console.observe_failed();
        }
        tracing::warn!(%error, channel, "TAP_OBSERVE_FAILED");
    }
}

/// The tap's word for a publish return: an accepted publish says whom it
/// reached ('nobody' and 'unaccepted' included, SRV-012 R10); a
/// dedup-suppressed re-publish has no word and is not tapped.
const fn tap_outcome(
    result: Result<&PublishOutcome, &ServerError>,
) -> Option<crate::tap::TapOutcome> {
    match result {
        Ok(PublishOutcome { reach, .. }) => match reach {
            Reach::Delivered => Some(crate::tap::TapOutcome::Delivered),
            Reach::NoSubscriber => Some(crate::tap::TapOutcome::NoSubscriber),
            Reach::NoneAccepted => Some(crate::tap::TapOutcome::NoneAccepted),
            Reach::Suppressed => None,
        },
        Err(ServerError::PublishUnresolved { .. }) => Some(crate::tap::TapOutcome::Unresolved),
        Err(
            ServerError::ReservedNamespace { .. }
            | ServerError::SelfStreamPublisherAlreadyFinished { .. }
            | ServerError::ConfigLoad { .. }
            | ServerError::ConfigValidation { .. }
            | ServerError::ListenerBind { .. }
            | ServerError::ListenerAccept { .. }
            | ServerError::ParticipantIncarnation { .. }
            | ServerError::ParticipantServiceFatal { .. }
            | ServerError::ConnectionFateRecoveryRequired { .. }
            | ServerError::ServerIncarnationExhausted
            | ServerError::ParticipantStartupRestore { .. }
            | ServerError::ConnectionIncarnationExhausted { .. }
            | ServerError::ConnectionPidCollision { .. }
            | ServerError::UnsupportedOperation { .. }
            | ServerError::PushReplyDisconnected { .. }
            | ServerError::PushReplyTimeout { .. }
            | ServerError::PushReplyExpired { .. }
            | ServerError::PushReplyRefused { .. }
            | ServerError::PushFrameExceedsOutboundCapacity { .. }
            | ServerError::ClusterJoin { .. }
            | ServerError::ClusterSync { .. }
            | ServerError::ShutdownTimeout { .. }
            | ServerError::ShutdownFlush { .. }
            | ServerError::ShutdownUnsettledConnectionFates { .. }
            | ServerError::HealthEndpoint { .. }
            | ServerError::ConnectionLimitReached { .. }
            | ServerError::ConnectionCapReached { .. }
            | ServerError::ParticipantLogMigration { .. }
            | ServerError::Agent(_),
        ) => Some(crate::tap::TapOutcome::Refused),
    }
}