zenkey-fleet 0.11.1

//! Attachments are a wire fact and the engine carries them (#117): on the
//! subscribe path (`SampleView`), the fan-in path (`FleetAnswer`), and the
//! fetch ladder (`FetchedValue`) — refcounted like the payload, `None` when
//! the wire carried none (absence is a fact too, never a default).
//!
//! Event-driven: the matching badge proves routability before publishing.
//! Ports are ephemeral (`util::peer_pair`), so two test runs at once
//! cannot collide.

use std::time::Duration;

use zenkey::qos::QosProfile;
use zenkey_fleet::declare_publication;

mod util;
use util::peer_pair;

const KEY: &str = "v1/h-eeeeeeeeeeee/state/demo/health";

/// The Monitor delivers the attachment beside the payload — and a sample
/// without one delivers `None`, not an empty buffer.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_watched_sample_carries_its_attachment() {
    let (a, b) = peer_pair().await;

    let monitor = zenkey_fleet::Monitor::start(&b, zenkey_fleet::MonitorSpec::default())
        .await
        .expect("monitor");
    let mut events = monitor.events();
    monitor.watch(KEY).await.expect("watch");

    let publication = declare_publication(&a, KEY, QosProfile::Transition, None)
        .await
        .expect("declare");
    let matching = publication.matching_events().await.expect("events");
    assert!(
        tokio::time::timeout(util::SETTLE, matching.recv())
            .await
            .expect("matching within 5s")
            .expect("listener alive")
    );

    publication
        .send(b"{}".to_vec(), Some(b"meta".to_vec()))
        .await
        .expect("send with attachment");
    publication
        .send(b"{}".to_vec(), None)
        .await
        .expect("send without");

    let mut views = Vec::new();
    while views.len() < 2 {
        let item = tokio::time::timeout(util::SETTLE, events.recv())
            .await
            .expect("event within 5s")
            .expect("stream alive");
        if let zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::Sample(s)) = item {
            views.push(s);
        }
    }
    let first = views[0].attachment.as_ref().expect("first carried one");
    assert_eq!(first.to_bytes().as_ref(), b"meta");
    // #120: the wire's actual QoS axes ride the view and match the profile
    // the publication declared.
    assert!(
        views[0].qos_matches(QosProfile::Transition),
        "declared transition, observed {:?}/{:?}/{:?}/express={}",
        views[0].priority,
        views[0].congestion_control,
        views[0].reliability,
        views[0].express
    );
    assert!(
        views[1].attachment.is_none(),
        "no attachment on the wire is None, not an empty buffer"
    );
}

/// `fleet_get` answers carry the replier's attachment; a replier that sends
/// none yields `None`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_fleet_answer_carries_the_reply_attachment() {
    let (a, b) = peer_pair().await;

    let _queryable = a
        .declare_queryable(KEY)
        .callback(move |query| {
            let with = query.key_expr().as_str().to_string();
            tokio::spawn(async move {
                query
                    .reply(with, b"{\"ok\":true}".to_vec())
                    .attachment(b"who-answered".to_vec())
                    .await
                    .expect("reply");
            });
        })
        .await
        .expect("queryable");

    // Settle: loop the GET until the queryable answers (wait-routable).
    let answers = loop {
        let answers = zenkey_fleet::fleet_get(
            &zenkey_fleet::Fleet::new(&b, ""),
            KEY,
            &zenkey_fleet::GetOpts::new(Duration::from_millis(500)),
        )
        .await
        .expect("get");
        if !answers.is_empty() {
            break answers;
        }
    };
    let att = answers[0].attachment.as_ref().expect("attachment carried");
    assert_eq!(att.to_bytes().as_ref(), b"who-answered");
}

/// The fetch ladder's window rung carries the attachment too — what the
/// zengui detail pane renders.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_fetched_value_carries_the_attachment() {
    let (a, b) = peer_pair().await;

    let publication = declare_publication(&a, KEY, QosProfile::Sampled, None)
        .await
        .expect("declare");
    let matching = publication.matching_events().await.expect("events");

    let fetch = tokio::spawn(async move {
        zenkey_fleet::fetch_value(
            &b,
            KEY,
            zenkey_fleet::FetchSpec {
                // No storage in this fixture: keep the GET rungs short so the
                // window rung (the one under test) opens quickly.
                get_timeout: Duration::from_millis(300),
                window: Duration::from_secs(5),
            },
        )
        .await
    });

    // The fetch's window subscriber raises the badge; then publish into it.
    assert!(
        tokio::time::timeout(util::SETTLE, matching.recv())
            .await
            .expect("matching within 5s")
            .expect("listener alive")
    );
    publication
        .send(b"{\"v\":1}".to_vec(), Some(b"tag".to_vec()))
        .await
        .expect("send");

    let outcome = fetch.await.expect("join").expect("fetch");
    match outcome {
        zenkey_fleet::FetchOutcome::Value(v) => {
            let att = v.attachment.expect("attachment carried");
            assert_eq!(att.to_bytes().as_ref(), b"tag");
        }
        other => panic!("expected a value, got {other:?}"),
    }
}