kanade_shared/

subject.rs

/// Most dot-separated labels a pc_id may carry. A dotted pc_id (an OS
/// hostname such as `host.example.com`) becomes several NATS subject
/// tokens, and the role-level permission block in
/// `configs/nats-server.users.conf` can only enumerate that many label
/// counts where the pc_id sits in the middle of a subject. Keep the two in
/// step: the conformance test asserts the config against this constant.
pub const MAX_PC_ID_LABELS: usize = 4;

/// Reject a pc_id the role-level NATS permissions cannot cover. The only
/// refusal is "too many labels": every other pc_id that was valid before
/// stays valid.
pub fn validate_pc_id(pc_id: &str) -> Result<(), String> {
    let labels = pc_id.split('.').count();
    if labels > MAX_PC_ID_LABELS {
        return Err(format!(
            "pc_id {pc_id:?} has {labels} dot-separated labels; at most {MAX_PC_ID_LABELS} are supported"
        ));
    }
    Ok(())
}

pub const COMMANDS_ALL: &str = "commands.all";

pub fn commands_group(name: &str) -> String {
    format!("commands.group.{name}")
}

pub fn commands_pc(pc_id: &str) -> String {
    format!("commands.pc.{pc_id}")
}

/// `notifications.all` — broadcast end-user notification (SPEC
/// §2.2.1 / Phase E). Mirrors [`COMMANDS_ALL`] but on the
/// notification fan-out plane: the backend publishes here, the
/// `NOTIFICATIONS` stream retains it, and every agent forwards it to
/// the Client Apps in matching user sessions.
pub const NOTIFICATIONS_ALL: &str = "notifications.all";

/// Subject prefix for [`notifications_group`]. Exposed so callers that
/// parse a subject back into its group name (the backend's notification
/// audience resolver) strip against the same string the builder emits —
/// if the format ever changes, both move together instead of the parser
/// silently failing to match.
pub const NOTIFICATIONS_GROUP_PREFIX: &str = "notifications.group.";

/// Subject prefix for [`notifications_pc`]. See
/// [`NOTIFICATIONS_GROUP_PREFIX`].
pub const NOTIFICATIONS_PC_PREFIX: &str = "notifications.pc.";

/// `notifications.group.{group_name}` — group-scoped end-user
/// notification. Sibling of [`commands_group`] on the notification
/// plane.
pub fn notifications_group(name: &str) -> String {
    format!("{NOTIFICATIONS_GROUP_PREFIX}{name}")
}

/// `notifications.pc.{pc_id}` — single-PC end-user notification.
/// Sibling of [`commands_pc`] on the notification plane.
pub fn notifications_pc(pc_id: &str) -> String {
    format!("{NOTIFICATIONS_PC_PREFIX}{pc_id}")
}

/// `notif-amend` — ephemeral fleet-wide control channel for post-send
/// operations on a notification (currently: recall). Deliberately NOT
/// under `notifications.>` so the `NOTIFICATIONS` JetStream stream doesn't
/// retain it (these are live control pushes, not history). Published via
/// **core NATS** (fire-and-forget); every agent subscribes and forwards a
/// `notifications.amended` push to its clients, each of which applies it
/// only if it holds the referenced id — so a single broadcast needs no
/// per-audience routing (an id a client doesn't have is a no-op). The
/// durable half of a recall is the stream message deletion; this is just
/// the "remove it from screens that are showing it now" half.
pub const NOTIFICATIONS_AMEND_SUBJECT: &str = "notif-amend";

/// `events.notifications.acked.{pc_id}.{user_sid}.{notif_id}` — the
/// agent publishes this when a user clicks "確認" on a notification
/// (SPEC §2.2.2 / Phase E). The `{user_sid}` segment distinguishes
/// concurrent users on a shared PC (Fast User Switching / RDP). Lives
/// under `events.>` so the existing `EVENTS` stream retains it; the
/// backend's notification-acks projector consumes the narrowed
/// [`EVENTS_NOTIFICATIONS_ACKED_FILTER`] to build the SPA's
/// per-recipient confirmation view.
///
/// Subject spelling is fixed by SPEC §2.2.2 / §2.12.8 (`events.>`), so
/// acks ride the `EVENTS` stream's retention (shorter than the 90-day
/// `NOTIFICATIONS` history), not the notification stream's. That only
/// bounds **re-projection** after a `-WipeDb`: the durable source of
/// truth for ack_status is the `notification_acks` SQLite table, which
/// persists independently — so a live fleet keeps full ack history;
/// only a DB wipe truncates re-derivable acks to the EVENTS window,
/// the same limitation every `events.*`-projected table already has.
pub fn events_notifications_acked(pc_id: &str, user_sid: &str, notif_id: &str) -> String {
    format!("events.notifications.acked.{pc_id}.{user_sid}.{notif_id}")
}

/// `events.notifications.unacked.{pc_id}.{user_sid}.{notif_id}` — the
/// agent publishes this when a user *retracts* a prior "確認" (the
/// read↔unread toggle). Mirror of [`events_notifications_acked`]; the
/// backend's notification-acks projector consumes it on the same
/// consumer (see [`EVENTS_NOTIFICATIONS_FILTER`]) so ack and unack for
/// one recipient stay strictly ordered on the `EVENTS` stream. Layer
/// split: the **agent** tombstones the `notifications_read` KV (so the
/// user's own `notifications.list` goes back to unread); this event is
/// the **projector's** half — it stamps `unacked_at` on the read-model
/// row and appends the revoke to the audit log. Same retention caveat as
/// the acked subject: the durable source of truth is SQLite, the stream
/// only bounds re-projection after a `-WipeDb`.
pub fn events_notifications_unacked(pc_id: &str, user_sid: &str, notif_id: &str) -> String {
    format!("events.notifications.unacked.{pc_id}.{user_sid}.{notif_id}")
}

// `commands_exec` (subject `commands.exec.<job_id>`) was removed in
// v0.22.1. The STREAM_EXEC stream now catches the existing
// `commands.{all,group.X,pc.Y}` subjects directly, so the dedicated
// per-exec subject isn't needed any more. See
// `kanade-agent::command_replay` for how reconnecting agents catch
// up on missed messages.

pub fn results(request_id: &str) -> String {
    format!("results.{request_id}")
}

pub fn heartbeat(pc_id: &str) -> String {
    format!("heartbeat.{pc_id}")
}

/// `host_perf.<pc_id>` — Phase 1 of the perf telemetry pipeline. The
/// agent publishes a whole-host CPU / Memory / Disk I/O / Network
/// snapshot here on the cadence set by `host_perf_interval` in
/// agent_config (default 60 s). Distinct subject from `heartbeat.<pc_id>`
/// so the periodic heartbeat publisher stays untouched and pre-host_perf
/// backends that don't subscribe simply ignore the new traffic.
pub fn host_perf(pc_id: &str) -> String {
    format!("host_perf.{pc_id}")
}

/// `process_perf.<pc_id>` — Phase 2: top-N per-process snapshot
/// published only while `process_perf_enabled` is `true` AND the
/// `process_perf_expires_at` deadline is in the future. Separate
/// subject from `host_perf.<pc_id>` because process-perf is an
/// opt-in investigation mode — having its own subject lets the
/// projector skip the heavy table entirely for hosts that never
/// turn it on.
pub fn process_perf(pc_id: &str) -> String {
    format!("process_perf.{pc_id}")
}

/// `obs.<pc_id>` — per-PC observability event stream (Issue #246).
/// The agent publishes one [`crate::wire::ObsEvent`] per timeline
/// event (sign-in / out, power on / off, sleep / resume, agent
/// milestones, diagnostic bundle pointers). Distinct from
/// `events.started.*` (in-flight script lifecycle) and
/// `host_perf.<pc_id>` (numeric telemetry) — `obs.*` is the
/// semantic-event stream the SPA Timeline page consumes.
pub fn obs(pc_id: &str) -> String {
    format!("obs.{pc_id}")
}

/// `obs.>` — filter the backend projector subscribes to so a new
/// PC starts flowing into the timeline without any per-PC SUB
/// registration. Pairs with [`obs`] for publish.
pub const OBS_FILTER: &str = "obs.>";

/// `kill.<exec_id>` — Spec §2.6 Layer 3 abort signal. The exec_id is
/// the deployment / scheduler-fire UUID (formerly named `job_id`
/// pre-v0.29; renamed for accuracy — every `Command.exec_id` is a
/// per-deploy UUID, not a job-catalog id).
pub fn kill(exec_id: &str) -> String {
    format!("kill.{exec_id}")
}

pub fn inventory(pc_id: &str, category: &str) -> String {
    format!("inventory.{pc_id}.{category}")
}

/// `events.started.<exec_id>.<pc_id>` — v0.30 / PR α' lifecycle
/// event published by the agent just before spawning a script's
/// child process. Lets the backend project an in-flight row into
/// `execution_results` (with `finished_at = NULL`) so the SPA
/// Activity table can show running rows alongside finished ones.
/// Backend subscribes via [`EVENTS_STARTED_FILTER`].
pub fn events_started(exec_id: &str, pc_id: &str) -> String {
    format!("events.started.{exec_id}.{pc_id}")
}

/// Wildcard the backend events projector consumes on STREAM_EVENTS.
/// Narrow (`events.started.>`) rather than the whole `events.>` so
/// future event types can carry their own filters without rerouting
/// the started subset.
pub const EVENTS_STARTED_FILTER: &str = "events.started.>";

/// Wildcard the backend notification-acks projector consumes on
/// `STREAM_EVENTS`. Narrow (`events.notifications.acked.>`) rather
/// than the whole `events.>` so the projector only wakes for ack
/// events and not the high-volume `events.started.*` lifecycle
/// traffic (which the events projector handles separately).
/// NOTE: since unack landed, the backend projector consumes the broader
/// [`EVENTS_NOTIFICATIONS_FILTER`] (ack + unack on one ordered consumer), so
/// this narrower `acked.>` filter is **no longer the active consumer filter**.
/// Kept as a named constant for the ack-only subject shape (docs + tests).
pub const EVENTS_NOTIFICATIONS_ACKED_FILTER: &str = "events.notifications.acked.>";

/// Wildcard the backend notification-acks projector consumes on
/// `STREAM_EVENTS` once it handles both ack **and** unack. Broader than
/// [`EVENTS_NOTIFICATIONS_ACKED_FILTER`] (`acked.>`) so a single durable
/// consumer sees both `events.notifications.acked.*` and
/// `events.notifications.unacked.*` in stream-sequence order — the only
/// way `ack → unack → re-ack` stays correctly serialised (a second
/// consumer would race the DELETE past a later INSERT). Still a strict
/// subset of `events.>` so STREAM_EVENTS retains it without a config
/// change, and still narrower than the high-volume `events.started.*`
/// traffic the events projector owns.
pub const EVENTS_NOTIFICATIONS_FILTER: &str = "events.notifications.>";

pub const INVENTORY_HW: &str = "hw";
pub const INVENTORY_SW: &str = "sw";
pub const INVENTORY_NET: &str = "net";

/// `logs.fetch.<pc_id>` — request/reply: operator (or backend) sends
/// a `LogsRequest`; the addressed agent replies with the tail of its
/// local log file. On-demand only, no stream.
pub fn logs_fetch(pc_id: &str) -> String {
    format!("logs.fetch.{pc_id}")
}

/// `job.tail.<pc_id>` — request/reply for the live tail of a
/// still-running job's stdout/stderr. The operator (or backend, on
/// the SPA's behalf) sends a [`crate::wire::JobTailRequest`] carrying
/// the `result_id`; the addressed agent replies with the current
/// ring-buffer tail from its in-memory live registry. On-demand only,
/// no stream — the SPA polls this every few seconds (same shape as
/// `logs.fetch.<pc_id>`) while a job is in flight. Distinct subject
/// from `logs.fetch.<pc_id>` (whole-agent log file) because this is
/// scoped to a single job's captured output, not the agent's log.
pub fn job_tail(pc_id: &str) -> String {
    format!("job.tail.{pc_id}")
}

/// `agents.<pc_id>.ping` — v0.38 / #133 request/reply for the
/// active "ping" round-trip. The agent answers with a fresh
/// `Heartbeat` on demand instead of the backend waiting up to ~30 s
/// for the next periodic heartbeat tick to land. Distinct subject
/// from `heartbeat.<pc_id>` so the periodic publisher is unaffected
/// and old agents that don't subscribe simply time the request out.
pub fn ping(pc_id: &str) -> String {
    format!("agents.{pc_id}.ping")
}

/// `remote.ctrl.<pc_id>` — #1140 control plane for a remote-assistance
/// session: start, stop, retune.
///
/// Addressed by **pc_id, not session id**, because the agent has to already
/// be subscribed when the first request arrives — a session-scoped control
/// subject would have nothing listening at the moment an operator clicks
/// "connect". Request/reply: the backend sends a [`crate::wire::RemoteCtrl`]
/// and the agent answers with a [`crate::wire::RemoteCtrlReply`] carrying
/// the screen geometry (so the SPA can size its canvas before the first
/// frame lands) or a refusal reason.
pub fn remote_ctrl(pc_id: &str) -> String {
    format!("remote.ctrl.{pc_id}")
}

/// `remote.frame.<session_id>` — agent → backend, one encoded tile per
/// message.
///
/// Session-scoped rather than pc-scoped so two operators viewing the same
/// machine never have to demultiplex each other, and so a torn-down
/// session's late frames land on a subject nobody reads any more.
///
/// **Core NATS only — deliberately not JetStream.** A frame is worthless the
/// moment it has been displayed; retaining megabytes per second of them
/// would cost storage and add latency for nothing. `remote.*` sits outside
/// the `events.>` / `obs.>` / `commands.*` filters precisely so no stream
/// catches it — the same reasoning [`NOTIFICATIONS_AMEND_SUBJECT`] records.
///
/// The payload is the **raw encoded image**, not JSON: tile metadata rides
/// in NATS headers (see [`crate::wire::FrameMeta`]). Base64 inside a JSON
/// envelope inflates every frame by a third, and #1142 measured a typical
/// session at ~4.0 Mbps — so the envelope alone would cost ~1.3 Mbps per
/// viewer to carry nothing.
pub fn remote_frame(session_id: &str) -> String {
    format!("remote.frame.{session_id}")
}

/// `remote.input.<session_id>` — backend → agent, one input event per
/// message. JSON, unlike frames: these are small and structured, and the
/// encoding overhead that matters for pixels is irrelevant for a mouse move.
pub fn remote_input(session_id: &str) -> String {
    format!("remote.input.{session_id}")
}

// v0.14: subject::inventory_request was retired alongside the
// hardcoded inventory loop. On-demand collection now goes through
// the normal exec path (`kanade exec configs/jobs/inventory-
// hw.yaml`) — Command + ExecResult + the inventory-fact projector
// give operators the same effect with no extra subject.

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

    #[test]
    fn validate_pc_id_limits_label_count_only() {
        for ok in ["pc", "a.b", "a.b.c.d", "PC 1/x?", "", "m1air.local"] {
            assert!(validate_pc_id(ok).is_ok(), "{ok:?}");
        }
        let err = validate_pc_id("a.b.c.d.e").unwrap_err();
        assert!(err.contains("at most 4"), "{err}");
    }

    #[test]
    fn commands_all_constant() {
        assert_eq!(COMMANDS_ALL, "commands.all");
    }

    #[test]
    fn commands_group_formats_name() {
        assert_eq!(commands_group("canary"), "commands.group.canary");
        assert_eq!(commands_group("wave1"), "commands.group.wave1");
    }

    #[test]
    fn commands_pc_formats_id() {
        assert_eq!(commands_pc("pc-01"), "commands.pc.pc-01");
        assert_eq!(commands_pc("PC1234"), "commands.pc.PC1234");
    }

    #[test]
    fn notifications_all_constant() {
        assert_eq!(NOTIFICATIONS_ALL, "notifications.all");
    }

    #[test]
    fn notifications_group_formats_name() {
        assert_eq!(
            notifications_group("tokyo-office"),
            "notifications.group.tokyo-office"
        );
    }

    #[test]
    fn notifications_pc_formats_id() {
        assert_eq!(notifications_pc("PC1234"), "notifications.pc.PC1234");
    }

    #[test]
    fn notifications_amend_subject_constant() {
        // Pin the value: backend publisher and every agent subscriber must
        // agree on it, and it must stay OUTSIDE `notifications.>` so the
        // NOTIFICATIONS stream never retains these ephemeral control messages.
        assert_eq!(NOTIFICATIONS_AMEND_SUBJECT, "notif-amend");
        assert!(!NOTIFICATIONS_AMEND_SUBJECT.starts_with("notifications."));
    }

    #[test]
    fn events_notifications_acked_formats_all_segments() {
        assert_eq!(
            events_notifications_acked("PC1234", "S-1-5-21-1001", "notif-9f3a"),
            "events.notifications.acked.PC1234.S-1-5-21-1001.notif-9f3a"
        );
    }

    #[test]
    fn events_notifications_acked_filter_is_narrow_wildcard() {
        assert_eq!(
            EVENTS_NOTIFICATIONS_ACKED_FILTER,
            "events.notifications.acked.>"
        );
        // Must stay a strict subset of the EVENTS stream's `events.>`
        // subjects so STREAM_EVENTS retains it without a config change.
        assert!(EVENTS_NOTIFICATIONS_ACKED_FILTER.starts_with("events."));
    }

    #[test]
    fn events_notifications_unacked_formats_all_segments() {
        assert_eq!(
            events_notifications_unacked("PC1234", "S-1-5-21-1001", "notif-9f3a"),
            "events.notifications.unacked.PC1234.S-1-5-21-1001.notif-9f3a"
        );
    }

    #[test]
    fn events_notifications_filter_covers_acked_and_unacked() {
        assert_eq!(EVENTS_NOTIFICATIONS_FILTER, "events.notifications.>");
        // The broadened filter must subsume both the acked and unacked
        // subjects so one durable consumer serialises them in order.
        let acked = events_notifications_acked("PC1", "S-1", "n1");
        let unacked = events_notifications_unacked("PC1", "S-1", "n1");
        assert!(acked.starts_with("events.notifications."));
        assert!(unacked.starts_with("events.notifications."));
        // Still a subset of the EVENTS stream's retained subjects.
        assert!(EVENTS_NOTIFICATIONS_FILTER.starts_with("events."));
    }

    #[test]
    fn remote_subjects_format_their_scope() {
        assert_eq!(remote_ctrl("PC1234"), "remote.ctrl.PC1234");
        assert_eq!(remote_frame("sess-9f3a"), "remote.frame.sess-9f3a");
        assert_eq!(remote_input("sess-9f3a"), "remote.input.sess-9f3a");
    }

    #[test]
    fn remote_subjects_stay_outside_every_retained_prefix() {
        // The whole plane is core-NATS-only. If any of these ever started
        // with a retained prefix, JetStream would quietly begin storing a
        // video stream — this test is the guard on that.
        for s in [remote_ctrl("PC1"), remote_frame("s1"), remote_input("s1")] {
            for retained in [
                "events.",
                "obs.",
                "commands.",
                "notifications.",
                "inventory.",
            ] {
                assert!(!s.starts_with(retained), "{s} collides with {retained}");
            }
        }
    }

    #[test]
    fn remote_ctrl_is_pc_scoped_and_frame_is_session_scoped() {
        // Encodes the design decision: control must be reachable before a
        // session exists, frames must not be.
        assert!(remote_ctrl("PC1").ends_with("PC1"));
        assert!(remote_frame("sess-1").ends_with("sess-1"));
    }

    #[test]
    fn results_formats_request_id() {
        assert_eq!(results("req-1"), "results.req-1");
    }

    #[test]
    fn heartbeat_formats_pc_id() {
        assert_eq!(heartbeat("pc-01"), "heartbeat.pc-01");
    }

    #[test]
    fn host_perf_formats_pc_id() {
        assert_eq!(host_perf("pc-01"), "host_perf.pc-01");
        assert_eq!(host_perf("PC1234"), "host_perf.PC1234");
    }

    #[test]
    fn process_perf_formats_pc_id() {
        assert_eq!(process_perf("pc-01"), "process_perf.pc-01");
        assert_eq!(process_perf("PC1234"), "process_perf.PC1234");
    }

    #[test]
    fn obs_formats_pc_id() {
        assert_eq!(obs("pc-01"), "obs.pc-01");
        assert_eq!(obs("PC1234"), "obs.PC1234");
    }

    #[test]
    fn obs_filter_constant() {
        assert_eq!(OBS_FILTER, "obs.>");
    }

    #[test]
    fn kill_formats_exec_id() {
        assert_eq!(kill("exec-uuid-1"), "kill.exec-uuid-1");
    }

    #[test]
    fn logs_fetch_formats_pc_id() {
        assert_eq!(logs_fetch("pc-01"), "logs.fetch.pc-01");
    }

    #[test]
    fn ping_formats_pc_id() {
        assert_eq!(ping("pc-01"), "agents.pc-01.ping");
    }

    #[test]
    fn job_tail_formats_pc_id() {
        assert_eq!(job_tail("pc-01"), "job.tail.pc-01");
        assert_eq!(job_tail("PC1234"), "job.tail.PC1234");
    }

    #[test]
    fn events_started_formats_exec_id_and_pc_id() {
        assert_eq!(
            events_started("exec-uuid-1", "pc-01"),
            "events.started.exec-uuid-1.pc-01",
        );
    }

    #[test]
    fn events_started_filter_is_narrow_wildcard() {
        assert_eq!(EVENTS_STARTED_FILTER, "events.started.>");
    }

    #[test]
    fn inventory_formats_pc_id_and_category() {
        assert_eq!(inventory("pc-01", "hw"), "inventory.pc-01.hw");
        assert_eq!(inventory("pc-01", INVENTORY_HW), "inventory.pc-01.hw");
        assert_eq!(inventory("pc-01", INVENTORY_SW), "inventory.pc-01.sw");
        assert_eq!(inventory("pc-01", INVENTORY_NET), "inventory.pc-01.net");
    }
}