liminal-server 0.21.1

//! The held topology: one revision per real change, nothing for a repeat,
//! never a console line, and a document in the console's names.

use std::time::{Duration, Instant};

use super::error::ConsoleError;
use super::model::{FlowRole, ThisDevice, TopologyChange, TopologyModel};
use super::records::{LineLife, LineState, Mount, canonical_bytes};
use super::wall::WallBase;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn model(started: Instant) -> TopologyModel {
    TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: false,
        },
        ["orders.reply".to_owned()],
        [
            ("orders".to_owned(), LineLife::Kept, LineState::Live),
            (
                "console.topology".to_owned(),
                LineLife::Passing,
                LineState::Live,
            ),
            (
                "console.tap.orders".to_owned(),
                LineLife::Passing,
                LineState::Live,
            ),
        ],
        [(7, Mount::Websocket, Some("alice".to_owned()))],
        [],
    )
}

/// The wall clock the tests read: [`WALL_MS`] at the instant a document is
/// built.
const WALL_MS: u64 = 1_791_272_000_000;

fn document(
    model: &TopologyModel,
    built: Instant,
) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
    Ok(serde_json::from_slice(&canonical_bytes(
        &model.document(&WallBase::fixed(WALL_MS, built)).0,
        "topology",
    )?)?)
}

/// R7: with console.topology and two console.tap channels live, no line id
/// starts with console., at start or after a change names one.
#[test]
fn the_topology_has_no_console_line() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    assert!(!model.apply(TopologyChange::Channel {
        name: "console.tap.audit".to_owned(),
        life: LineLife::Passing,
        state: LineState::Live,
    }));
    let value = document(&model, started)?;
    let lines = value["lines"].as_array().ok_or("lines is an array")?;
    assert_eq!(lines.len(), 1);
    assert!(lines.iter().all(|line| {
        !line["id"]
            .as_str()
            .is_some_and(|id| id.starts_with("console."))
    }));
    Ok(())
}

/// Each real change moves the revision on by exactly one; a repeat moves it
/// not at all, so it publishes nothing.
#[test]
fn one_revision_per_real_change() {
    let started = Instant::now();
    let mut model = model(started);
    assert_eq!(model.rev(), 1);
    let changes = [
        TopologyChange::Service {
            subject: "billing".to_owned(),
            present: true,
        },
        TopologyChange::Channel {
            name: "orders".to_owned(),
            life: LineLife::Kept,
            state: LineState::Quiesced,
        },
        TopologyChange::ConnectionOpened {
            id: 8,
            mount: Mount::Tcp,
            participant: None,
        },
        TopologyChange::Peer {
            name: "edge".to_owned(),
            present: true,
        },
    ];
    for change in changes.clone() {
        assert!(model.apply(change));
    }
    assert_eq!(model.rev(), 5);
    for change in changes {
        assert!(!model.apply(change), "a repeat is not a change");
    }
    assert_eq!(model.rev(), 5);
    assert!(model.apply(TopologyChange::ConnectionClosed { id: 8 }));
    assert!(!model.apply(TopologyChange::ConnectionClosed { id: 8 }));
    assert_eq!(model.rev(), 6);
}

/// A peer appears as a device and an up link whose ends are both listed
/// devices, sorted; when it leaves, both are gone. This device carries its
/// version and its start in wall milliseconds; a peer carries neither.
#[test]
fn a_peer_joins_and_leaves_the_document() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    model.apply(TopologyChange::Peer {
        name: "edge".to_owned(),
        present: true,
    });
    let joined = document(&model, started + Duration::from_secs(3_600))?;
    assert_eq!(joined["rev"], 2);
    assert_eq!(joined["devices"][0]["started"], WALL_MS - 3_600_000);
    assert_eq!(joined["devices"][0]["liminal"], "1.2.3");
    assert_eq!(joined["devices"][1]["id"], "edge");
    assert!(joined["devices"][1].get("liminal").is_none());
    assert_eq!(
        joined["links"][0]["ends"],
        serde_json::json!(["edge", "local"])
    );
    assert_eq!(joined["links"][0]["state"], "up");
    assert_eq!(joined["connections"][0]["id"], "7");
    assert_eq!(joined["stations"][0]["device"], "local");

    model.apply(TopologyChange::Peer {
        name: "edge".to_owned(),
        present: false,
    });
    let left = document(&model, started)?;
    assert_eq!(left["rev"], 3);
    assert_eq!(left["devices"].as_array().map(Vec::len), Some(1));
    assert_eq!(left["links"].as_array().map(Vec::len), Some(0));
    Ok(())
}

/// R7: a registered worker is a station, and a line's stops are its
/// registered publishers then its registered subscribers, each once; an
/// unregistered connection is no stop, and a closed one leaves the line.
#[test]
fn a_line_runs_from_its_publisher_to_its_subscriber() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let flow = |connection, role| TopologyChange::Flow {
        connection,
        line: "orders".to_owned(),
        role,
        present: true,
    };
    for (connection, identity) in [(11, "ping"), (12, "pong")] {
        assert!(model.apply(TopologyChange::Worker {
            connection,
            identity: identity.to_owned(),
            subjects: Vec::new(),
            activities: Vec::new(),
        }));
    }
    assert!(model.apply(flow(12, FlowRole::Subscribes)));
    assert!(model.apply(flow(11, FlowRole::Publishes)));
    assert!(
        !model.apply(flow(11, FlowRole::Publishes)),
        "a repeat is no change"
    );
    assert!(
        !model.apply(flow(7, FlowRole::Subscribes)),
        "an unregistered connection"
    );
    let value = document(&model, started)?;
    assert_eq!(value["lines"][0]["id"], "orders");
    assert_eq!(
        value["lines"][0]["stops"],
        serde_json::json!(["ping", "local", "pong"])
    );
    let stations: Vec<&str> = value["stations"]
        .as_array()
        .ok_or("stations")?
        .iter()
        .filter_map(|station| station["id"].as_str())
        .collect();
    assert_eq!(stations, ["local", "orders.reply", "ping", "pong"]);
    let kinds: Vec<&str> = value["stations"]
        .as_array()
        .ok_or("stations")?
        .iter()
        .filter_map(|station| station["kind"].as_str())
        .collect();
    assert_eq!(kinds, ["exchange", "service", "worker", "worker"]);
    assert_eq!(
        value["devices"][0]["box"],
        serde_json::json!({"x": -140, "y": 60, "w": 1340, "h": 520})
    );

    assert!(model.apply(TopologyChange::ConnectionClosed { id: 12 }));
    let value = document(&model, started)?;
    assert_eq!(
        value["lines"][0]["stops"],
        serde_json::json!(["ping", "local"])
    );
    Ok(())
}

/// A worker that stops subscribing leaves the line's stops at once, without
/// waiting for its connection to close.
#[test]
fn a_flow_withdrawn_leaves_the_stops() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let flow = |present| TopologyChange::Flow {
        connection: 12,
        line: "orders".to_owned(),
        role: FlowRole::Subscribes,
        present,
    };
    assert!(model.apply(TopologyChange::Worker {
        connection: 12,
        identity: "pong".to_owned(),
        subjects: Vec::new(),
        activities: Vec::new(),
    }));
    assert!(model.apply(flow(true)));
    assert!(model.apply(flow(false)));
    assert!(!model.apply(flow(false)), "a repeat is no change");
    let value = document(&model, started)?;
    assert_eq!(value["lines"][0]["stops"], serde_json::json!(["local"]));
    Ok(())
}

/// This device's start is its incarnation's, in wall milliseconds, and the
/// same whenever the document is built: one built at T+0 and read at T+30 s
/// lets the reader show 30 s more than at T+0, where an uptime stamped at
/// T+0 would still read T+0 (Hermes 7 Oct 01:49).
#[test]
fn the_start_is_the_incarnations_whenever_the_document_is_built() -> TestResult {
    let built = Instant::now();
    let mut model = model(built);
    let incarnation = built.checked_sub(Duration::from_secs(5)).ok_or("clock")?;
    assert!(model.incarnation_started(incarnation));

    let (at_zero, fault) = model.document(&WallBase::fixed(WALL_MS, built));
    assert!(fault.is_none(), "{fault:?}");
    // The same wall clock, read 30 s later.
    let later = built + Duration::from_secs(30);
    let (at_thirty, _) = model.document(&WallBase::fixed(WALL_MS + 30_000, later));
    let started = at_zero.devices[0]
        .started
        .ok_or("no start on this device")?;
    assert_eq!(started, WALL_MS - 5_000);
    assert_eq!(at_thirty.devices[0].started, Some(started));
    let read_at_thirty = WALL_MS + 30_000;
    assert_eq!(
        read_at_thirty - started,
        35_000,
        "the uptime read at T+30 s"
    );
    let value = document(&model, built)?;
    assert!(
        value["devices"][0].get("uptime").is_none(),
        "no stale uptime field"
    );
    Ok(())
}

/// Buttercup's finding 3 (Hermes 03:05): this device carries its start or
/// the fault that kept it, never both and never neither. A start the wall
/// clock can give is `started` alone; one before the epoch is
/// `started_fault` alone, naming the clock, and the fault comes back beside
/// the document for the hub to count.
#[test]
fn a_device_has_its_start_or_its_fault_never_both() -> TestResult {
    let built = Instant::now();
    let model = model(built);
    let (good, none) = model.document(&WallBase::fixed(WALL_MS, built));
    assert!(none.is_none(), "{none:?}");
    assert_eq!(
        (
            good.devices[0].started.is_some(),
            good.devices[0].started_fault.is_none()
        ),
        (true, true)
    );
    // A wall base one second after the epoch, read with a start two seconds
    // before it.
    let mut early = model;
    let incarnation = built.checked_sub(Duration::from_secs(2)).ok_or("clock")?;
    early.incarnation_started(incarnation);
    let (faulted, fault) = early.document(&WallBase::fixed(1_000, built));
    assert!(
        matches!(fault, Some(ConsoleError::Clock { .. })),
        "{fault:?}"
    );
    assert_eq!(faulted.devices[0].started, None);
    assert_eq!(
        faulted.devices[0].started_fault.as_deref(),
        Some("console clock: a time is outside u64 milliseconds since the epoch")
    );
    for peer in &faulted.devices[1..] {
        assert_eq!((peer.started, peer.started_fault.as_deref()), (None, None));
    }
    Ok(())
}

/// Lane brief ruling 1 (Hermes 7 Oct 03:51): the document names this device
/// once, at the top level, as one of its devices' ids, whatever peers it
/// holds and however they sort.
#[test]
fn the_document_names_this_device_once() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    model.apply(TopologyChange::Peer {
        name: "alpha".to_owned(),
        present: true,
    });
    let value = document(&model, started)?;
    assert_eq!(value["this_device"], "local");
    let devices = value["devices"]
        .as_array()
        .ok_or("the document has no devices")?;
    let named = devices
        .iter()
        .filter(|device| device["id"] == value["this_device"])
        .count();
    assert_eq!(named, 1, "this_device names exactly one device: {value}");
    Ok(())
}

/// Lane brief ruling 5 (Hermes 7 Oct 03:51): a server with a persistent
/// store has a depot, `<device>.depot`, and a durable line passes it after
/// its publishers and before its subscribers, because the append comes
/// before the fan-out. A passing line never stops there, and a server with
/// no persistent store lists no depot.
#[test]
fn a_durable_line_passes_the_depot_before_its_subscribers() -> TestResult {
    let started = Instant::now();
    let with_depot = |depot| {
        let mut model = TopologyModel::new(
            ThisDevice {
                id: "local".to_owned(),
                liminal: "1.2.3".to_owned(),
                started,
                depot,
            },
            [],
            [
                ("orders".to_owned(), LineLife::Kept, LineState::Live),
                ("audit".to_owned(), LineLife::Passing, LineState::Live),
            ],
            [],
            [],
        );
        for (connection, identity) in [(11, "ping"), (12, "pong")] {
            model.apply(TopologyChange::Worker {
                connection,
                identity: identity.to_owned(),
                subjects: Vec::new(),
                activities: Vec::new(),
            });
        }
        for line in ["orders", "audit"] {
            for (connection, role) in [(11, FlowRole::Publishes), (12, FlowRole::Subscribes)] {
                model.apply(TopologyChange::Flow {
                    connection,
                    line: line.to_owned(),
                    role,
                    present: true,
                });
            }
        }
        model
    };
    let stops = |value: &serde_json::Value, line: &str| {
        value["lines"]
            .as_array()
            .and_then(|lines| lines.iter().find(|entry| entry["id"] == line))
            .map(|entry| entry["stops"].clone())
    };
    let kept = document(&with_depot(true), started)?;
    assert_eq!(
        stops(&kept, "orders"),
        Some(serde_json::json!(["ping", "local.depot", "local", "pong"]))
    );
    assert_eq!(
        stops(&kept, "audit"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    let depot = kept["stations"]
        .as_array()
        .and_then(|stations| {
            stations
                .iter()
                .find(|station| station["id"] == "local.depot")
        })
        .ok_or_else(|| format!("no depot station: {kept}"))?;
    assert_eq!(
        (&depot["kind"], &depot["device"]),
        (&serde_json::json!("depot"), &serde_json::json!("local"))
    );

    let memory = document(&with_depot(false), started)?;
    assert_eq!(
        stops(&memory, "orders"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    assert!(
        !memory.to_string().contains("local.depot"),
        "no persistent store, no depot: {memory}"
    );
    Ok(())
}

/// Lane brief ruling 2 (Hermes 7 Oct 03:51): this server's exchange is a
/// station whose id is the device's own, of kind `exchange`, and a stop on
/// every line exactly once: after the publishers and the depot, before the
/// subscribers, and alone on a line with no flows.
#[test]
fn every_line_passes_the_exchange_once() -> TestResult {
    let started = Instant::now();
    let mut model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: true,
        },
        [],
        [
            ("orders".to_owned(), LineLife::Kept, LineState::Live),
            ("audit".to_owned(), LineLife::Passing, LineState::Live),
            ("idle".to_owned(), LineLife::Passing, LineState::Live),
        ],
        [],
        [],
    );
    for (connection, identity) in [(11, "ping"), (12, "pong")] {
        model.apply(TopologyChange::Worker {
            connection,
            identity: identity.to_owned(),
            subjects: Vec::new(),
            activities: Vec::new(),
        });
    }
    for line in ["orders", "audit"] {
        for (connection, role) in [(11, FlowRole::Publishes), (12, FlowRole::Subscribes)] {
            model.apply(TopologyChange::Flow {
                connection,
                line: line.to_owned(),
                role,
                present: true,
            });
        }
    }
    let value = document(&model, started)?;
    let stops = |line: &str| {
        value["lines"]
            .as_array()
            .and_then(|lines| lines.iter().find(|entry| entry["id"] == line))
            .map(|entry| entry["stops"].clone())
    };
    assert_eq!(
        stops("orders"),
        Some(serde_json::json!(["ping", "local.depot", "local", "pong"]))
    );
    assert_eq!(
        stops("audit"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    assert_eq!(stops("idle"), Some(serde_json::json!(["local"])));
    let exchange = value["stations"]
        .as_array()
        .and_then(|stations| stations.iter().find(|station| station["id"] == "local"))
        .ok_or_else(|| format!("no exchange station: {value}"))?;
    assert_eq!(
        (&exchange["kind"], &exchange["device"]),
        (&serde_json::json!("exchange"), &serde_json::json!("local"))
    );
    Ok(())
}

/// SRV-027 R3: a worker's station carries its subjects and its activities
/// with the schema strings exactly as registered (nested objects and
/// non-ASCII descriptions untouched); a worker that declared none is raw
/// bytes, `activities: []`; and the station leaves with its connection.
#[test]
fn worker_station_carries_its_subjects_and_activities_exactly() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let input = r#"{"type":"object","properties":{"items":{"type":"array","items":{"type":"object","properties":{"text":{"type":"string","description":"Texte à plonger"}}}}}}"#;
    let output = r#"{"type":"object","properties":{"vectors":{"type":"array","description":"Vecteurs, 384 dimensions"}}}"#;
    assert!(model.apply(TopologyChange::Worker {
        connection: 21,
        identity: "tessera-embed".to_owned(),
        subjects: vec!["tessera.embed.request".to_owned()],
        activities: vec![super::records::StationActivity {
            name: "tessera.embed.request".to_owned(),
            input_schema_json: input.to_owned(),
            output_schema_json: output.to_owned(),
        }],
    }));
    assert!(model.apply(TopologyChange::Worker {
        connection: 22,
        identity: "raw-worker".to_owned(),
        subjects: vec!["raw.request".to_owned()],
        activities: Vec::new(),
    }));
    let with_workers = document(&model, started)?;
    let stations = with_workers["stations"].as_array().ok_or("no stations")?;
    let typed = stations
        .iter()
        .find(|station| station["id"] == "tessera-embed")
        .ok_or("the typed worker has no station")?;
    assert_eq!(
        typed["subjects"],
        serde_json::json!(["tessera.embed.request"])
    );
    assert_eq!(typed["activities"][0]["name"], "tessera.embed.request");
    assert_eq!(typed["activities"][0]["input_schema_json"], input);
    assert_eq!(typed["activities"][0]["output_schema_json"], output);
    let raw = stations
        .iter()
        .find(|station| station["id"] == "raw-worker")
        .ok_or("the raw worker has no station")?;
    assert_eq!(raw["subjects"], serde_json::json!(["raw.request"]));
    assert_eq!(raw["activities"], serde_json::json!([]));
    let exchange = stations
        .iter()
        .find(|station| station["id"] == "local")
        .ok_or("no exchange")?;
    assert!(exchange.get("activities").is_none());
    assert!(model.apply(TopologyChange::ConnectionClosed { id: 21 }));
    let after_close = document(&model, started)?;
    assert!(
        after_close["stations"]
            .as_array()
            .ok_or("no stations")?
            .iter()
            .all(|station| station["id"] != "tessera-embed")
    );
    Ok(())
}