liminal-server 0.25.4

use std::sync::Arc;

use liminal::channel::{ChannelSupervisor, InboxInstall};
use liminal::durability::DurableStore;
use serde_json::json;

use super::config::AgentsConfig;
use super::presence::{ConnectionObservation, Presence};
use super::records::{AgentEvent, AgentHarness, AgentSpec};
use super::retained::Retained;
use super::test_support::{Fixture, TestResult, approved};

#[test]
fn retained_presence_is_current_first_then_changes_with_named_departure() -> TestResult {
    let fixture = Fixture::new()?;
    let registry = fixture.registry()?;
    registry.execute_approved(&approved(
        "create-a",
        "a",
        "owner",
        AgentEvent::Create(AgentSpec {
            harness: AgentHarness::Cdx,
            purpose: "Build one service.".into(),
            service: None,
        }),
    )?)?;
    let supervisor = ChannelSupervisor::new()?;
    let retained = Retained::new(AgentsConfig::default(), supervisor.clone(), "fixture");
    let presence = Presence::default();
    presence.connect(
        &registry,
        &retained,
        ConnectionObservation {
            name: "a",
            holder: "a",
            connection: 7,
            at_ms: 2,
        },
        || Ok(()),
    )?;
    let mut watch = retained.subscribe(1, "agents.a.presence", &[], None)?;
    let current = watch
        .try_next()
        .ok_or("current value was not queued first")?;
    let value: serde_json::Value = serde_json::from_slice(&current.payload)?;
    assert_eq!(value, json!({"present":true,"at_ms":2,"connection":7}));
    let refused = presence.connect(
        &registry,
        &retained,
        ConnectionObservation {
            name: "a",
            holder: "a",
            connection: 8,
            at_ms: 3,
        },
        || Ok(()),
    );
    assert_eq!(
        refused.err().ok_or("duplicate admitted")?.to_string(),
        "agent a is already present on connection 7, held since unix ms 2; the seat frees when \
         that connection closes or its peer stops answering keepalive (80 s), so give this \
         client a seat of its own"
    );
    presence.close(&registry, &retained, 7, "client_close", 4)?;
    let gone = watch.try_next().ok_or("gone change was not queued")?;
    let value: serde_json::Value = serde_json::from_slice(&gone.payload)?;
    assert_eq!(
        value,
        json!({"present":false,"at_ms":4,"reason":"client_close"})
    );
    presence.close(&registry, &retained, 7, "client_close", 5)?;
    assert!(watch.try_next().is_none());
    presence.connect(
        &registry,
        &retained,
        ConnectionObservation {
            name: "a",
            holder: "a",
            connection: 8,
            at_ms: 6,
        },
        || Ok(()),
    )?;
    assert_eq!(presence.connection("a")?, Some(8));
    drop(watch);
    supervisor.shutdown();
    Ok(())
}

#[test]
fn retained_watchers_are_bounded_and_a_slow_watcher_is_shed() -> TestResult {
    let supervisor = ChannelSupervisor::new()?;
    let bounds = AgentsConfig {
        max_watchers: 1,
        watcher_depth: 1,
        ..AgentsConfig::default()
    };
    let retained = Retained::new(bounds, supervisor.clone(), "fixture");
    retained.publish(&[("agents.a.state".into(), json!({"state":"gone"}))])?;
    let first = retained.subscribe(1, "agents.a.state", &[], None)?;
    assert_eq!(
        retained
            .subscribe(2, "agents.a.state", &[], None)
            .err()
            .ok_or("unbounded watchers")?
            .to_string(),
        "auth.lys.agents.max_watchers: limit 1, count 1"
    );
    retained.publish(&[("agents.a.state".into(), json!({"state":"present"}))])?;
    assert!(first.is_overflowed());
    drop(first);
    let mut replacement = retained.subscribe(2, "agents.a.state", &[], None)?;
    let value = replacement
        .try_next()
        .ok_or("replacement lacks current state")?;
    assert_eq!(
        serde_json::from_slice::<serde_json::Value>(&value.payload)?,
        json!({"state":"present"})
    );
    drop(replacement);
    supervisor.shutdown();
    Ok(())
}

#[test]
fn a_server_restart_records_a_previously_present_ticket_gone() -> TestResult {
    let fixture = Fixture::new()?;
    let registry = fixture.registry()?;
    registry.execute_approved(&approved(
        "create-a",
        "a",
        "owner",
        AgentEvent::Create(AgentSpec {
            harness: AgentHarness::Cdx,
            purpose: "Build one service.".into(),
            service: None,
        }),
    )?)?;
    let supervisor = ChannelSupervisor::new()?;
    let retained = Retained::new(AgentsConfig::default(), supervisor.clone(), "fixture");
    Presence::default().connect(
        &registry,
        &retained,
        ConnectionObservation {
            name: "a",
            holder: "a",
            connection: 7,
            at_ms: 2,
        },
        || Ok(()),
    )?;
    drop(registry);
    let store: Arc<dyn DurableStore> = fixture.counts;
    let agents = super::runtime::Agents::open(
        store,
        AgentsConfig::default(),
        "fixture-network".into(),
        supervisor.clone(),
        "fixture",
        None,
        None,
    )?;
    assert_eq!(agents.registry().record("a")?.connection(), None);
    assert_eq!(
        agents.retained().value("agents.a.state")?,
        Some(json!({"state":"gone"}))
    );
    let installed = InboxInstall {
        budget: liminal::channel::ConnectionInboxBudget::new(1),
        depth_cap: 1,
        notifier: None,
        capacity: None,
    };
    assert!(
        agents
            .retained()
            .subscribe(3, "agents.a.state", &[], Some(installed))
            .is_err()
    );
    supervisor.shutdown();
    Ok(())
}