frame-conv 0.2.0

//! F-3a R3/R6 — the subscription pattern against a live liminal 0.3.0
//! server: the cross-pattern composition arc, delivery at the proven
//! strength (ordered, contiguous, sender-excluded), membership-forward
//! join, and typed schema refusal on the event surface.

#![allow(clippy::expect_used, clippy::unwrap_used)]

mod support;

use std::collections::HashMap;
use std::error::Error;
use std::time::{Duration, Instant};

use frame_conv::{
    ConversationHandle, ConversationSeq, InboundRequest, ParticipantRef, RequestOutcome,
    SubscriptionItem,
};
use serde::{Deserialize, Serialize};
use support::{FileStore, QUANTUM, RunningServer, attachment, store_dir};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct AddContact {
    name: String,
    priority: u32,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct ContactAdded {
    name: String,
    total: u32,
}

/// One observed event row: body, publisher, sequence.
type EventRow = (ContactAdded, ParticipantRef, ConversationSeq);

/// Collects up to `want` typed events, skipping lifecycle items, within
/// `budget`.
fn collect_events(
    handle: &mut ConversationHandle<FileStore>,
    want: usize,
    budget: Duration,
) -> Result<Vec<EventRow>, Box<dyn Error>> {
    let mut events = Vec::new();
    let until = Instant::now() + budget;
    while events.len() < want && Instant::now() < until {
        match handle.next_event::<ContactAdded>(Duration::from_secs(1))? {
            Some(SubscriptionItem::Event {
                body,
                publisher,
                seq,
            }) => events.push((body, publisher, seq)),
            Some(_lifecycle) => {}
            None => {}
        }
    }
    Ok(events)
}

/// R3's composition acceptance (the cross-pattern arc BUILD-PLAN names):
/// participant A changes state via request-response; subscriber C observes
/// the resulting typed update through its subscription — proven by content.
#[test]
fn composition_arc_request_changes_state_subscriber_observes() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("sub-composition")?;

    let (mut owner, owner_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("owner.lpcr")),
    )?;
    let conversation = owner_grant.conversation;

    // The subscriber joins BEFORE the state change so the update is in its
    // membership-forward window.
    let (mut subscriber, _sub_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("subscriber.lpcr")),
    )?;

    let owner_thread = std::thread::spawn(move || -> Result<(), String> {
        let mut contacts: HashMap<String, u32> = HashMap::new();
        let inbound = owner
            .next_request::<AddContact>(Duration::from_secs(25))
            .map_err(|error| error.to_string())?;
        let Some(InboundRequest::Valid(request)) = inbound else {
            return Err(format!("owner expected a valid request: {inbound:?}"));
        };
        contacts.insert(request.body.name.clone(), request.body.priority);
        let total = u32::try_from(contacts.len()).map_err(|error| error.to_string())?;
        let added = ContactAdded {
            name: request.body.name,
            total,
        };
        owner
            .reply(request.correlation, &added)
            .map_err(|error| error.to_string())?;
        owner
            .publish_event(&added)
            .map_err(|error| error.to_string())?;
        Ok(())
    });

    let (mut requester, _req_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("requester.lpcr")),
    )?;
    let outcome = requester.request::<AddContact, ContactAdded>(
        &AddContact {
            name: "ada".to_owned(),
            priority: 7,
        },
        Duration::from_secs(25),
    )?;
    let RequestOutcome::Replied { reply, .. } = outcome else {
        return Err(format!("expected the correlated reply, observed {outcome:?}").into());
    };
    assert_eq!(reply.name, "ada");

    owner_thread.join().expect("owner thread must not panic")?;

    let events = collect_events(&mut subscriber, 1, 3 * QUANTUM)?;
    let [(event, publisher, _seq)] = events.as_slice() else {
        return Err(format!("subscriber expected exactly the update event: {events:?}").into());
    };
    assert_eq!(
        event,
        &ContactAdded {
            name: "ada".to_owned(),
            total: 1,
        },
        "the subscriber must observe the state change BY CONTENT"
    );
    assert_eq!(
        *publisher, owner_grant.participant,
        "the update must carry the owner's verified identity"
    );

    std::fs::remove_dir_all(&stores)?;
    server.shutdown()?;
    Ok(())
}

/// R3 — delivery at the proven strength (F-0c assertion 5): events arrive
/// in send order on ascending contiguous sequences matching the publisher's
/// receipts; the publisher never observes its own events (sender
/// exclusion); zero gaps on either side.
#[test]
fn events_ordered_contiguous_sender_excluded() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("sub-ordering")?;

    let (mut publisher, _pub_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();

    let (mut subscriber, _sub_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("subscriber.lpcr")),
    )?;

    let mut receipts = Vec::new();
    for total in 1..=3_u32 {
        let receipt = publisher.publish_event(&ContactAdded {
            name: "ada".to_owned(),
            total,
        })?;
        receipts.push(receipt.seq);
    }
    assert!(
        receipts
            .windows(2)
            .all(|pair| pair[1].value() == pair[0].value() + 1),
        "publish receipts must occupy ascending contiguous sequences: {receipts:?}"
    );

    let events = collect_events(&mut subscriber, 3, 3 * QUANTUM)?;
    let observed: Vec<(u32, ConversationSeq)> = events
        .iter()
        .map(|(body, _, seq)| (body.total, *seq))
        .collect();
    let expected: Vec<(u32, ConversationSeq)> = receipts
        .iter()
        .enumerate()
        .map(|(index, seq)| (u32::try_from(index).unwrap_or(0) + 1, *seq))
        .collect();
    assert_eq!(
        observed, expected,
        "delivery order/sequences must match send order and the receipts"
    );

    // Sender exclusion: the publisher observes ONLY the subscriber's join,
    // never a copy of its own events.
    let mut publisher_items = Vec::new();
    let until = Instant::now() + QUANTUM + Duration::from_secs(2);
    while Instant::now() < until {
        match publisher.next_event::<ContactAdded>(Duration::from_secs(1))? {
            Some(SubscriptionItem::Event { body, .. }) => {
                return Err(format!("publisher observed its own event: {body:?}").into());
            }
            Some(item) => publisher_items.push(format!("{item:?}")),
            None => {}
        }
    }
    assert!(
        publisher_items
            .iter()
            .all(|item| item.contains("PeerJoined")),
        "publisher must observe only the join lifecycle: {publisher_items:?}"
    );

    assert_eq!(publisher.anomaly_counters().gaps, 0);
    assert_eq!(subscriber.anomaly_counters().gaps, 0);

    std::fs::remove_dir_all(&stores)?;
    server.shutdown()?;
    Ok(())
}

/// R3 — the late-joiner pin (F-0c full characterization, test-1 finding):
/// joining is membership-forward; history before the join is NOT delivered,
/// and the fresh cursor space raises no gap.
#[test]
fn late_joiner_is_membership_forward_no_history_replay() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("sub-latejoin")?;

    let (mut publisher, _pub_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();

    // History BEFORE the subscriber exists.
    publisher.publish_event(&ContactAdded {
        name: "before-join".to_owned(),
        total: 1,
    })?;

    let (mut subscriber, _sub_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("subscriber.lpcr")),
    )?;

    publisher.publish_event(&ContactAdded {
        name: "after-join".to_owned(),
        total: 2,
    })?;

    // The subscriber observes exactly the post-membership event; a further
    // full quantum of silence proves the pre-join event never arrives.
    let events = collect_events(&mut subscriber, 2, 3 * QUANTUM)?;
    let [(event, _, _)] = events.as_slice() else {
        return Err(format!(
            "membership-forward join must deliver exactly the post-join event: {events:?}"
        )
        .into());
    };
    assert_eq!(event.name, "after-join");
    assert_eq!(
        subscriber.anomaly_counters().gaps,
        0,
        "a membership-forward baseline is not a gap"
    );

    std::fs::remove_dir_all(&stores)?;
    server.shutdown()?;
    Ok(())
}

/// R1/R6 — a schema-invalid event surfaces typed on the subscription
/// surface (never dropped, never a panic), and the stream continues.
#[test]
fn schema_invalid_event_surfaces_typed() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("sub-schema")?;

    let (mut publisher, pub_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();

    let (mut subscriber, _sub_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("subscriber.lpcr")),
    )?;

    // A well-formed envelope whose payload is NOT the subscriber's typed
    // event, then a valid event proving the stream continues.
    publisher.publish_event(&serde_json::json!({"weird": true}))?;
    publisher.publish_event(&ContactAdded {
        name: "valid".to_owned(),
        total: 1,
    })?;

    let mut saw_invalid = false;
    let mut saw_valid = false;
    let until = Instant::now() + 3 * QUANTUM;
    while !(saw_invalid && saw_valid) && Instant::now() < until {
        match subscriber.next_event::<ContactAdded>(Duration::from_secs(1))? {
            Some(SubscriptionItem::SchemaInvalid {
                publisher: from,
                detail,
                ..
            }) => {
                assert_eq!(from, pub_grant.participant);
                assert!(!detail.is_empty(), "the refusal must carry its detail");
                saw_invalid = true;
            }
            Some(SubscriptionItem::Event { body, .. }) => {
                assert_eq!(body.name, "valid");
                saw_valid = true;
            }
            Some(_) | None => {}
        }
    }
    assert!(saw_invalid, "the schema-invalid event must surface typed");
    assert!(saw_valid, "the stream must continue past the invalid event");

    std::fs::remove_dir_all(&stores)?;
    server.shutdown()?;
    Ok(())
}