//! Host lifecycle event bus (scripting-and-hooks RFC §4, phase 0). //! //! A process-wide broadcast bus + bounded catch-up ring for **lifecycle events**: client //! connect/disconnect, session and stream start/end, pairing decisions, virtual-display //! create/release, library mutations, host start/stop. Fire sites on BOTH planes call //! [`emit`]; consumers ([`EventBus::subscribe`]) get a ring-backed catch-up plus a live //! tail — the shape `GET /api/v1/events` (SSE, phase 1) and the hook runner (phase 2) //! consume. Until those land, the bus is host-internal. //! //! Design notes (mirrors [`crate::log_capture`], the shipped ring precedent): //! - Events carry a monotonically increasing `seq` (1-based) and a wall-clock `ts_ms`. //! A consumer resumes with `since = last seen seq`; one that fell off the ring gets //! `dropped = true` and should resync via the REST snapshots. //! - The wire shape is **versioned and additive-only** within a major ([`SCHEMA_VERSION`]): //! fields and kinds may be added, never removed or renamed. The JSON snapshot tests below //! are the review gate — a failing snapshot IS a schema change. //! - **Payload hygiene: events never carry secrets** — no PINs, no tokens, no key material. //! Client names and fingerprints are fine (already exposed via the management API). //! - Emission is fire-and-forget and cheap (a mutex push + a non-blocking broadcast send); //! nothing here sits in a streaming hot path. Slow consumers lag (`RecvError::Lagged`) //! rather than buffering unboundedly; the SSE layer disconnects them. use serde::{Deserialize, Serialize}; use std::collections::VecDeque; use std::sync::{Mutex, OnceLock}; use std::time::{SystemTime, UNIX_EPOCH}; use tokio::sync::broadcast; use utoipa::ToSchema; /// Wire-shape major version, carried on every event. Additive-only within a major; removing /// or renaming a field is a new major (and, at the API layer, a new endpoint negotiation). pub const SCHEMA_VERSION: u32 = 1; /// Catch-up ring capacity. Events are small (a few hundred bytes) and low-rate (lifecycle, /// not per-frame), so 1024 spans hours of ordinary host activity. const RING_CAPACITY: usize = 1024; /// Live-tail channel depth per subscriber before a slow consumer starts lagging. const BROADCAST_CAPACITY: usize = 256; /// One host lifecycle event, as it will appear on the wire (`data:` of one SSE frame). #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] pub struct HostEvent { /// Monotonic sequence number (1-based) — a consumer resumes with `since = last seen`. pub seq: u64, /// Unix timestamp in milliseconds (the [`crate::log_capture::LogEntry`] convention). pub ts_ms: u64, /// Wire-shape version ([`SCHEMA_VERSION`]). pub schema: u32, /// The event kind + payload, flattened: `"kind": "stream.started", …payload…`. #[serde(flatten)] pub kind: EventKind, } /// Which protocol plane an event originated from. Hooks and scripts filter on it — a hook /// that fires for native clients but not Moonlight clients is a bug, not a v2 feature. #[derive(Serialize, Deserialize, ToSchema, Clone, Copy, Debug, PartialEq, Eq)] #[serde(rename_all = "lowercase")] pub enum Plane { /// The native punktfunk/1 plane (QUIC). Native, /// The GameStream/Moonlight compat plane (`--gamestream`). Gamestream, } /// Why a client went away. `Quit` is a deliberate user "stop" (the typed close code); /// `Timeout` is a transport idle timeout (the client vanished); `Error` is everything else. #[derive(Serialize, Deserialize, ToSchema, Clone, Copy, Debug, PartialEq, Eq)] #[serde(rename_all = "lowercase")] pub enum DisconnectReason { Quit, Timeout, Error, } /// The connecting/disconnecting client's identity. #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] pub struct ClientRef { /// Client-supplied device name; may be empty (an anonymous or compat-plane client). pub name: String, /// Hex SHA-256 certificate fingerprint, when the client presented one. #[serde(skip_serializing_if = "Option::is_none")] pub fingerprint: Option, pub plane: Plane, } /// A live A/V session (the plane-neutral notion the Dashboard shows). #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] pub struct SessionRef { /// Host-local session id (unique within this host process). pub id: u64, /// Short client label (cert-fingerprint prefix, or peer IP for an anonymous client). pub client: String, /// Negotiated mode, `WxH@Hz` (e.g. `"3840x2160@120"`). pub mode: String, pub hdr: bool, } /// A live video stream (what the stream marker file reflects). #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] pub struct StreamRef { /// Negotiated mode, `WxH@Hz`. pub mode: String, pub hdr: bool, /// Client-supplied device name; may be empty. pub client: String, /// The launched app/title for this stream, when one was requested (store-qualified id on /// the native plane, app title on the GameStream plane). #[serde(skip_serializing_if = "Option::is_none")] pub app: Option, pub plane: Plane, } /// A device in the pairing flow. #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] pub struct DeviceRef { /// Sanitized device name (the pairing store's copy). pub name: String, /// Hex certificate fingerprint. pub fingerprint: String, pub plane: Plane, } /// The event catalog (RFC §4). Serialized internally tagged as `"kind": "."`, /// flattened into [`HostEvent`]. **Additive-only** within [`SCHEMA_VERSION`]. #[derive(Serialize, Deserialize, ToSchema, Clone, Debug)] #[serde(tag = "kind")] pub enum EventKind { #[serde(rename = "client.connected")] ClientConnected { client: ClientRef }, #[serde(rename = "client.disconnected")] ClientDisconnected { client: ClientRef, reason: DisconnectReason, }, #[serde(rename = "session.started")] SessionStarted { session: SessionRef }, #[serde(rename = "session.ended")] SessionEnded { session: SessionRef }, #[serde(rename = "stream.started")] StreamStarted { stream: StreamRef }, #[serde(rename = "stream.stopped")] StreamStopped { stream: StreamRef }, #[serde(rename = "pairing.pending")] PairingPending { device: DeviceRef }, #[serde(rename = "pairing.completed")] PairingCompleted { device: DeviceRef }, #[serde(rename = "pairing.denied")] PairingDenied { device: DeviceRef }, #[serde(rename = "display.created")] DisplayCreated { /// The virtual-display backend that minted it (`VirtualDisplay::name`). backend: String, /// `WxH@Hz`. mode: String, }, #[serde(rename = "display.released")] DisplayReleased { /// How many kept displays this release retired. count: u32, }, #[serde(rename = "library.changed")] LibraryChanged { /// What mutated the library: `"manual"` today; a provider id once the provider /// API (RFC §8) lands. source: String, }, #[serde(rename = "host.started")] HostStarted { version: String, /// Whether the GameStream/Moonlight compat plane is enabled. gamestream: bool, }, #[serde(rename = "host.stopping")] HostStopping, } impl EventKind { /// The wire kind string (`"stream.started"`, …) — for filters and log lines. pub fn name(&self) -> &'static str { match self { EventKind::ClientConnected { .. } => "client.connected", EventKind::ClientDisconnected { .. } => "client.disconnected", EventKind::SessionStarted { .. } => "session.started", EventKind::SessionEnded { .. } => "session.ended", EventKind::StreamStarted { .. } => "stream.started", EventKind::StreamStopped { .. } => "stream.stopped", EventKind::PairingPending { .. } => "pairing.pending", EventKind::PairingCompleted { .. } => "pairing.completed", EventKind::PairingDenied { .. } => "pairing.denied", EventKind::DisplayCreated { .. } => "display.created", EventKind::DisplayReleased { .. } => "display.released", EventKind::LibraryChanged { .. } => "library.changed", EventKind::HostStarted { .. } => "host.started", EventKind::HostStopping => "host.stopping", } } } impl EventKind { /// The client/device name this event carries, if any — the `filter.client` axis of hooks /// and scripts. (For `session.*` this is the short client *label* the Dashboard shows — /// cert-fingerprint prefix or peer IP — since that is what the event carries.) pub fn client_name(&self) -> Option<&str> { match self { EventKind::ClientConnected { client } | EventKind::ClientDisconnected { client, .. } => Some(&client.name), EventKind::SessionStarted { session } | EventKind::SessionEnded { session } => { Some(&session.client) } EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => { Some(&stream.client) } EventKind::PairingPending { device } | EventKind::PairingCompleted { device } | EventKind::PairingDenied { device } => Some(&device.name), _ => None, } } /// The certificate fingerprint this event carries, if any. pub fn fingerprint(&self) -> Option<&str> { match self { EventKind::ClientConnected { client } | EventKind::ClientDisconnected { client, .. } => client.fingerprint.as_deref(), EventKind::PairingPending { device } | EventKind::PairingCompleted { device } | EventKind::PairingDenied { device } => Some(&device.fingerprint), _ => None, } } /// The protocol plane this event carries, if any. pub fn plane(&self) -> Option { match self { EventKind::ClientConnected { client } | EventKind::ClientDisconnected { client, .. } => Some(client.plane), EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => { Some(stream.plane) } EventKind::PairingPending { device } | EventKind::PairingCompleted { device } | EventKind::PairingDenied { device } => Some(device.plane), _ => None, } } /// The launched app id/title this event carries, if any. pub fn app(&self) -> Option<&str> { match self { EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => { stream.app.as_deref() } _ => None, } } } /// Does `pattern` select `kind`? Exact kind names (`stream.started`) or `domain.*` prefixes /// matched on the dot boundary (`stream.*` matches `stream.started`, never `streamx.started`). /// One vocabulary for the SSE `?kinds=` filter and the hooks `on:` field. pub fn kind_matches(pattern: &str, kind: &str) -> bool { match pattern.strip_suffix(".*") { Some(prefix) => kind .strip_prefix(prefix) .is_some_and(|rest| rest.starts_with('.')), None => pattern == kind, } } /// Formats a mode as the wire's `WxH@Hz` string. pub fn mode_str(width: u32, height: u32, hz: u32) -> String { format!("{width}x{height}@{hz}") } /// One consumer's view: the ring-backed catch-up plus a live-tail receiver, taken atomically /// (no event can fall between `catch_up` and the first `rx.recv()`, and none is in both). pub struct Subscription { /// Events with `seq > since`, oldest first. pub catch_up: Vec, /// True when events between `since` and the first caught-up one were already evicted — /// the consumer should resync via the REST snapshots (the `LogPage.dropped` contract). pub dropped: bool, /// The live tail. A consumer that can't keep up sees `RecvError::Lagged`. pub rx: broadcast::Receiver, } /// The process-wide event bus: a bounded seq-numbered ring (catch-up) + a broadcast channel /// (live tail). pub struct EventBus { inner: Mutex, tx: broadcast::Sender, } struct Ring { events: VecDeque, next_seq: u64, } impl EventBus { fn new() -> Self { let (tx, _) = broadcast::channel(BROADCAST_CAPACITY); Self { inner: Mutex::new(Ring { events: VecDeque::with_capacity(RING_CAPACITY), next_seq: 1, }), tx, } } /// Stamp, ring-buffer, and broadcast one event. Fire-and-forget: no receivers is fine /// (the ring still records it for a later subscriber's catch-up). pub fn emit(&self, kind: EventKind) { let ts_ms = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); let mut ring = self.inner.lock().unwrap_or_else(|e| e.into_inner()); let ev = HostEvent { seq: ring.next_seq, ts_ms, schema: SCHEMA_VERSION, kind, }; ring.next_seq += 1; if ring.events.len() == RING_CAPACITY { ring.events.pop_front(); } ring.events.push_back(ev.clone()); // Send while still holding the ring lock: it serializes with `subscribe` (which also // takes the lock), so an event lands either in a subscriber's catch-up or on its live // tail — never both, never neither. `send` is non-blocking, the hold is trivial. let _ = self.tx.send(ev); } /// A live-tail-only subscription (no catch-up, no cursor) — for host-internal consumers /// like the hook runner that only care about events from now on. pub fn subscribe_live(&self) -> broadcast::Receiver { self.tx.subscribe() } /// Subscribe with a resume cursor: events with `seq > since` come back as catch-up, the /// returned receiver carries everything after. `since = 0` means "from the ring start". pub fn subscribe(&self, since: u64) -> Subscription { let ring = self.inner.lock().unwrap_or_else(|e| e.into_inner()); let rx = self.tx.subscribe(); let first_seq = ring.events.front().map_or(ring.next_seq, |e| e.seq); let dropped = since != 0 && since.saturating_add(1) < first_seq; let catch_up = ring .events .iter() .filter(|e| e.seq > since) .cloned() .collect(); Subscription { catch_up, dropped, rx, } } } /// The process-wide bus — a `OnceLock` singleton (the [`crate::log_capture::ring`] shape) so /// fire sites across both planes and the API layer share it without threading an `Arc`. pub fn bus() -> &'static EventBus { static BUS: OnceLock = OnceLock::new(); BUS.get_or_init(EventBus::new) } /// Emit one lifecycle event on the process-wide bus. Cheap and non-blocking — safe from any /// thread, including RAII `Drop` paths. pub fn emit(kind: EventKind) { bus().emit(kind); } #[cfg(test)] mod tests { use super::*; fn ev(name: &str) -> EventKind { EventKind::LibraryChanged { source: name.to_string(), } } #[test] fn seq_is_monotonic_and_catch_up_resumes() { let bus = EventBus::new(); for i in 0..5 { bus.emit(ev(&format!("m{i}"))); } let sub = bus.subscribe(0); assert_eq!( sub.catch_up.iter().map(|e| e.seq).collect::>(), vec![1, 2, 3, 4, 5] ); assert!(!sub.dropped); assert!(sub.catch_up.iter().all(|e| e.schema == SCHEMA_VERSION)); // Resume from a cursor mid-ring. let sub = bus.subscribe(3); assert_eq!( sub.catch_up.iter().map(|e| e.seq).collect::>(), vec![4, 5] ); assert!(!sub.dropped); // Cursor at the tip: empty catch-up, not a gap. let sub = bus.subscribe(5); assert!(sub.catch_up.is_empty()); assert!(!sub.dropped); } #[test] fn eviction_reports_dropped() { let bus = EventBus::new(); for i in 0..(RING_CAPACITY + 50) { bus.emit(ev(&format!("m{i}"))); } // Seqs 1..=50 were evicted; a cursor inside the gap must flag it. let sub = bus.subscribe(10); assert!(sub.dropped); assert_eq!(sub.catch_up.first().map(|e| e.seq), Some(51)); // A fresh consumer (since = 0) is a backfill, not a gap. let sub = bus.subscribe(0); assert!(!sub.dropped); assert_eq!(sub.catch_up.len(), RING_CAPACITY); } #[tokio::test] async fn live_tail_continues_exactly_after_catch_up() { let bus = EventBus::new(); bus.emit(ev("before-1")); bus.emit(ev("before-2")); let mut sub = bus.subscribe(0); assert_eq!(sub.catch_up.len(), 2); // Emitted after subscribe → on the live tail only, starting at exactly seq 3. bus.emit(ev("after")); let live = sub.rx.recv().await.expect("live event"); assert_eq!(live.seq, 3); assert_eq!(live.kind.name(), "library.changed"); // Nothing duplicated: the tail holds only what wasn't in the catch-up. assert!(sub.rx.try_recv().is_err()); } /// The wire shape IS the contract (additive-only, RFC §4): these snapshots are the review /// gate — if one fails, the change renames/removes a field and needs a schema-version bump, /// not a test update. #[test] fn wire_shape_snapshots() { let ev = HostEvent { seq: 4182, ts_ms: 1_700_000_000_000, schema: 1, kind: EventKind::StreamStarted { stream: StreamRef { mode: mode_str(3840, 2160, 120), hdr: true, client: "Living Room TV".into(), app: Some("steam:570".into()), plane: Plane::Native, }, }, }; assert_eq!( serde_json::to_string(&ev).unwrap(), r#"{"seq":4182,"ts_ms":1700000000000,"schema":1,"kind":"stream.started","stream":{"mode":"3840x2160@120","hdr":true,"client":"Living Room TV","app":"steam:570","plane":"native"}}"# ); let ev = HostEvent { seq: 1, ts_ms: 1_700_000_000_000, schema: 1, kind: EventKind::ClientDisconnected { client: ClientRef { name: "Deck".into(), fingerprint: Some("b1c2".into()), plane: Plane::Gamestream, }, reason: DisconnectReason::Timeout, }, }; assert_eq!( serde_json::to_string(&ev).unwrap(), r#"{"seq":1,"ts_ms":1700000000000,"schema":1,"kind":"client.disconnected","client":{"name":"Deck","fingerprint":"b1c2","plane":"gamestream"},"reason":"timeout"}"# ); let ev = HostEvent { seq: 2, ts_ms: 1_700_000_000_000, schema: 1, kind: EventKind::HostStopping, }; assert_eq!( serde_json::to_string(&ev).unwrap(), r#"{"seq":2,"ts_ms":1700000000000,"schema":1,"kind":"host.stopping"}"# ); } #[test] fn wire_shape_roundtrips() { let ev = HostEvent { seq: 7, ts_ms: 3, schema: 1, kind: EventKind::PairingPending { device: DeviceRef { name: "iPad Pro".into(), fingerprint: "ab12".into(), plane: Plane::Native, }, }, }; let json = serde_json::to_string(&ev).unwrap(); let back: HostEvent = serde_json::from_str(&json).unwrap(); assert_eq!(back.seq, 7); assert_eq!(back.kind.name(), "pairing.pending"); match back.kind { EventKind::PairingPending { device } => { assert_eq!(device.name, "iPad Pro"); assert_eq!(device.plane, Plane::Native); } other => panic!("wrong kind: {other:?}"), } } }