hyperhive/swarm-wire/src/router_todo.rs

//! The router's todo queue: one work-queue stream, one subject and one
//! durable consumer per agent, and the op each message carries.
//!
//! The notification router publishes a [`RouterTodoOp`] on [`subject`] for
//! the agent it addresses. That agent's harness drains its own consumer,
//! [`consumer_name`], applies each op to its todo store as subsystem
//! [`TODO_SUBSYSTEM`], and acks it, which removes it from the stream. An
//! agent that is down finds its ops waiting, up to [`MAX_AGE`].
//!
//! The store dedups on `(subsystem, key)`, so a redelivered op is harmless.
//!
//! Three crates name these: the router publishes, the harness consumes, and
//! the auth-callout responder grants each its side.

use std::time::Duration;

use serde::{Deserialize, Serialize};

/// The stream the ops are queued in. The router creates it.
pub const STREAM: &str = "ROUTER_TODOS";

/// Every subject [`STREAM`] captures sits below this, one token per agent.
pub const SUBJECT_PREFIX: &str = "$SWARM.router.todo";

/// How long an op waits for its agent to take it.
pub const MAX_AGE: Duration = Duration::from_hours(30 * 24);

/// How many ops one agent's subject holds before the oldest go.
pub const MAX_MSGS_PER_SUBJECT: i64 = 1000;

/// The router's publish dedup window: an op republished with the same
/// `Nats-Msg-Id` inside it is dropped by the stream.
pub const DUPLICATE_WINDOW: Duration = Duration::from_mins(2);

/// The todo subsystem every op is applied under, so the router's todos never
/// collide with another producer's keys.
pub const TODO_SUBSYSTEM: &str = "router";

/// The subject the router queues `agent`'s ops on.
#[must_use]
pub fn subject(agent: &str) -> String {
    format!("{SUBJECT_PREFIX}.{agent}")
}

/// Every [`subject`] at once: the stream's subjects and the router's publish
/// grant.
#[must_use]
pub fn subject_wildcard() -> String {
    subject("*")
}

/// The durable consumer `agent`'s harness drains: `todo-<agent>`.
#[must_use]
pub fn consumer_name(agent: &str) -> String {
    format!("todo-{agent}")
}

/// What an op does to the agent's todo list.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Op {
    /// Add the todo, or update the one under the same key.
    Upsert,
    /// Remove the todo under the key, or with [`RouterTodoOp::prefix`] every
    /// router todo whose key starts with it.
    Clear,
}

/// One change to one agent's router todos.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RouterTodoOp {
    pub op: Op,
    /// The todo's key under [`TODO_SUBSYSTEM`], such as
    /// `forge:o/r#7:assignment`.
    pub key: String,
    /// For [`Op::Clear`]: `key` is a prefix. Ignored for an upsert.
    #[serde(default)]
    pub prefix: bool,
    /// The line the agent reads. Empty for a clear.
    #[serde(default)]
    pub summary: String,
    /// Which flow produced the op: `router:<flow-id>`.
    pub source: String,
}

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

    #[test]
    fn the_names_are_the_agreed_ones() {
        assert_eq!(subject("atlas"), "$SWARM.router.todo.atlas");
        assert_eq!(subject_wildcard(), "$SWARM.router.todo.*");
        assert_eq!(consumer_name("atlas"), "todo-atlas");
    }

    #[test]
    fn an_op_reads_back_with_defaults_for_what_a_clear_leaves_out() {
        let op: RouterTodoOp = serde_json::from_str(
            r#"{"op":"clear","key":"forge:o/r#1:","prefix":true,"source":"router:closed"}"#,
        )
        .expect("decodes");
        assert_eq!(op.op, Op::Clear);
        assert!(op.prefix);
        assert_eq!(op.summary, "");
        let upsert: RouterTodoOp = serde_json::from_str(
            r#"{"op":"upsert","key":"k","summary":"s","source":"router:assigned"}"#,
        )
        .expect("decodes");
        assert!(!upsert.prefix);
    }
}