doctate/server/src/events.rs

//! Broadcast bus for UI live-update events.
//!
//! The web UI subscribes via the `/web/events` SSE route (see
//! `routes::events`). Background workers and request handlers call
//! [`emit`] whenever they mutate case data on disk; subscribed browsers
//! then schedule a debounced reload. The filesystem remains the source
//! of truth — these events are pure triggers, never state.

use std::path::Path;

use serde::Serialize;
use tokio::sync::broadcast;

/// Broadcast capacity. Oversized on purpose: even a "bulk analyze 50 cases"
/// operation emits at most ~5 events per case (Queued/Transcript/Oneliner/
/// Document/Done), well under 256. Subscribers more than 256 events behind
/// receive a `Lagged(n)` on their next recv; they are re-synced by their
/// next page reload anyway, so skipping is safe.
pub const EVENT_CHANNEL_CAPACITY: usize = 256;

/// All UI-observable mutations on a case. Serialized as a `kind` field
/// in snake_case so the client can filter with e.g. `evt.kind === "transcript_ready"`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CaseEventKind {
    RecordingUploaded,
    RecordingDeleted,
    TranscriptReady,
    TranscriptFailed,
    OnelinerUpdated,
    AnalysisQueued,
    DocumentReady,
    CaseClosed,
    CaseReopened,
    CasePurged,
    CaseReset,
}

/// A single state-change notification. Clone is required because
/// `tokio::sync::broadcast` hands every subscriber its own copy.
#[derive(Debug, Clone, Serialize)]
pub struct CaseEvent {
    pub user_slug: String,
    pub case_id: String,
    pub kind: CaseEventKind,
    pub ts: String,
}

/// Multi-producer, multi-consumer sender shared via `AppState`.
pub type EventSender = broadcast::Sender<CaseEvent>;

/// Construct a fresh broadcast channel sized with [`EVENT_CHANNEL_CAPACITY`].
/// The unused receiver is dropped — subscribers call
/// [`EventSender::subscribe`] on the returned sender.
pub fn channel() -> EventSender {
    let (tx, _rx) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
    tx
}

/// Best-effort publish. `broadcast::Sender::send` returns `Err` only when
/// there are zero subscribers — which is the normal state before any
/// browser connects — so we swallow the error silently. Never blocks.
pub fn emit(
    tx: &EventSender,
    user_slug: impl Into<String>,
    case_id: impl Into<String>,
    kind: CaseEventKind,
) {
    let _ = tx.send(CaseEvent {
        user_slug: user_slug.into(),
        case_id: case_id.into(),
        kind,
        ts: doctate_common::now_rfc3339(),
    });
}

/// Extract `<user_slug>` from a `<data_path>/<user_slug>/<case_id>/` path.
/// Returns the empty string on malformed input. Empty slugs still flow
/// through the admin filter (admins see them) but are filtered out for
/// non-admins, which is the safe default.
pub fn user_slug_of(case_dir: &Path) -> String {
    case_dir
        .parent()
        .and_then(|p| p.file_name())
        .and_then(|s| s.to_str())
        .map(str::to_owned)
        .unwrap_or_default()
}

/// Extract `<case_id>` from a `<data_path>/<user_slug>/<case_id>/` path.
/// Returns the empty string on malformed input.
pub fn case_id_of(case_dir: &Path) -> String {
    case_dir
        .file_name()
        .and_then(|s| s.to_str())
        .map(str::to_owned)
        .unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn two_subscribers_both_receive() {
        let tx = channel();
        let mut a = tx.subscribe();
        let mut b = tx.subscribe();
        emit(&tx, "alice", "c1", CaseEventKind::RecordingUploaded);
        let ea = a.recv().await.unwrap();
        let eb = b.recv().await.unwrap();
        assert_eq!(ea.user_slug, "alice");
        assert_eq!(eb.case_id, "c1");
        assert!(matches!(ea.kind, CaseEventKind::RecordingUploaded));
    }

    #[tokio::test]
    async fn slow_subscriber_sees_lagged() {
        let tx = channel();
        let mut rx = tx.subscribe();
        // Fill the channel past capacity without recv'ing.
        for _ in 0..(EVENT_CHANNEL_CAPACITY + 5) {
            emit(&tx, "u", "c", CaseEventKind::TranscriptReady);
        }
        let err = rx.recv().await.unwrap_err();
        assert!(matches!(err, broadcast::error::RecvError::Lagged(_)));
    }

    #[test]
    fn user_slug_extracted_from_case_dir() {
        let p = Path::new("/data/alice/550e8400-e29b-41d4-a716-446655440000");
        assert_eq!(user_slug_of(p), "alice");
    }

    #[test]
    fn user_slug_empty_on_root_path() {
        assert_eq!(user_slug_of(Path::new("/")), "");
    }

    #[test]
    fn emit_with_no_subscribers_does_not_panic() {
        let tx = channel();
        emit(&tx, "u", "c", CaseEventKind::DocumentReady);
    }
}