liminal-server 0.24.1

use std::collections::BTreeSet;
use std::error::Error;
use std::sync::Arc;

use liminal::durability::{CountingStore, DurableStore, bridge::block_on};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::log::STREAM_PREFIX;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::route_keys::conversation_windows;
use super::routed_stream::read_head;
use super::tests::{dispatch, test_participant_config};
use super::tests_memory_store::MemoryStore;

#[test]
fn a_serving_conversation_crossing_the_checkpoint_boundary_uses_atomic_batches()
-> Result<(), Box<dyn Error>> {
    let ServingFixture {
        store,
        handler,
        connection,
        participant_id,
    } = serving_fixture()?;
    let route = checkpoint_conversation_route(10);
    let stream = format!("{STREAM_PREFIX}10");
    let mut boundaries = BTreeSet::new();
    let before = store.snapshot();
    for operation in 0_u64..257 {
        let mut token = [0; 16];
        token[..8].copy_from_slice(&operation.to_be_bytes());
        let result = dispatch(
            &handler,
            connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: 10,
                participant_id,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: operation.to_be_bytes().to_vec(),
            }),
        )?;
        assert!(
            matches!(result, ServerValue::RecordCommitted(_)),
            "{result:?}"
        );
        let head = block_on(read_head(store.as_ref(), &route, &stream))??
            .ok_or("serving operation omitted its routed head")?;
        if [255, 256, 257].contains(&head.next_sequence) {
            assert_checkpoint_boundary(store.as_ref(), head.next_sequence)?;
            boundaries.insert(head.next_sequence);
        }
    }
    assert_eq!(boundaries, BTreeSet::from([255, 256, 257]));
    let counts = store.snapshot().since(&before);
    assert_eq!(
        counts.apply_routed_batch_calls, 257,
        "every logical operation must commit once, including the checkpoint crossing"
    );
    assert_eq!(counts.append_calls, 0, "serving writes must be staged");
    assert_eq!(counts.append_all_calls, 0, "serving writes must be staged");
    assert_eq!(counts.flushes, 0, "the atomic batch owns durability");
    Ok(())
}

struct ServingFixture {
    store: Arc<CountingStore<MemoryStore>>,
    handler: ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
}

fn serving_fixture() -> Result<ServingFixture, Box<dyn Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let handler = ProductionParticipantHandler::new(
        Arc::clone(&store) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let connection = ConnectionIncarnation::new(10, 1);
    let enrolled = dispatch(
        &handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: 10,
            enrollment_token: EnrollmentToken::new([1; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(bound) = enrolled else {
        return Err(format!("enrollment did not bind: {enrolled:?}").into());
    };
    let recipient = dispatch(
        &handler,
        ConnectionIncarnation::new(10, 2),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: 10,
            enrollment_token: EnrollmentToken::new([2; 16]),
        }),
    )?;
    assert!(matches!(recipient, ServerValue::EnrollBound(_)));
    Ok(ServingFixture {
        store,
        handler,
        connection,
        participant_id: bound.participant_id(),
    })
}

fn assert_checkpoint_boundary(store: &dyn DurableStore, next: u64) -> Result<(), Box<dyn Error>> {
    let route = checkpoint_conversation_route(10);
    let bytes = block_on(store.get_routed(&route, &CheckpointLocator::routed_key(10)))??;
    if next == 255 {
        assert!(bytes.is_none(), "a checkpoint preceded the first crossing");
    } else {
        let bytes = bytes.ok_or("checkpoint locator is absent at the boundary")?;
        let locator = CheckpointLocator::decode_canonical(&bytes, 10)?;
        assert_eq!(locator.covered_sequence(), 255);
        let body = block_on(store.get_routed(&route, &locator.body_key().routed_key(10)))??
            .ok_or("checkpoint body is absent at the boundary")?;
        locator.body_key().verify_body(&body)?;
    }
    Ok(())
}

/// DUR-027 R5: the seal's own commit publishes its checkpoint, and the
/// retirement that follows it in the same restore unit cuts the conversation's
/// own windows of its route (DUR-028 R5).
/// A reopen therefore finds no owner, no routed row and no stream row.
#[test]
fn a_real_last_identity_seal_leaves_no_checkpoint_route_once_retired() -> Result<(), Box<dyn Error>>
{
    let fixture = super::tests_f8b_seal::last_identity_store()?;
    let store = Arc::clone(&fixture.store);
    let conversation_id = fixture.conversation_id;
    drop(fixture);
    let route = checkpoint_conversation_route(conversation_id);
    assert!(
        !block_on(store.scan_routed(&route, &[], &[0xff]))??.is_empty(),
        "the fixture minted no routed rows, so it cannot pin the cut"
    );
    for _boot in 0..2 {
        let handler = ProductionParticipantHandler::new(Arc::clone(&store), seal_fixture_config())?;
        assert!(handler.existing_cell(conversation_id)?.is_none());
        for (from, to) in conversation_windows(
            conversation_id,
            &[
                format!("{STREAM_PREFIX}{conversation_id}"),
                format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
            ],
        )? {
            assert!(block_on(store.scan_routed(&route, &from, &to))??.is_empty());
        }
        let stream = format!("{STREAM_PREFIX}{conversation_id}");
        assert!(block_on(store.read_from(&stream, 0, usize::MAX))??.is_empty());
    }
    Ok(())
}

fn seal_fixture_config() -> crate::config::types::ParticipantConfig {
    let mut config = test_participant_config();
    config.max_retained_record_rows = 4;
    config
}