liminal-server 0.20.1

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

use liminal::durability::{DurableStore, bridge::block_on, open_ephemeral};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    ParticipantAck, ServerValue,
};

use super::super::ProductionParticipantHandler;
use super::super::log::OperationLog;
use super::super::outbox::ConversationOutboxLimits;
use super::super::outbox_log::OutboxLog;
use super::super::restore_checkpoint::{
    CheckpointLiveState, CheckpointPositions, CheckpointSchema,
};
use super::super::restore_checkpoint_history::CheckpointHistoryRoot;
use super::super::restore_checkpoint_history_capture::capture_identity_history;
use super::super::tests::{dispatch_tracked, test_participant_config};
use super::ConversationAuthority;
use super::checkpoint_body::CheckpointOwnerBody;
use super::checkpoint_owner::ConversationAuthorityRestoreContext;
use crate::config::types::ParticipantConfig;
use crate::server::participant::ParticipantConnectionConversations;

type TestResult<T = ()> = Result<T, Box<dyn Error>>;

struct Fixture {
    owner: ConversationAuthority,
    positions: CheckpointPositions,
    root: Vec<u8>,
    config: ParticipantConfig,
}

struct FrozenFixture {
    held: Result<Vec<u8>, String>,
    pruned: Result<Vec<u8>, String>,
    positions: CheckpointPositions,
    root: Vec<u8>,
    config: ParticipantConfig,
}

#[derive(serde::Serialize, serde::Deserialize)]
struct RedFixture {
    held: Vec<u8>,
    pruned: Vec<u8>,
    positions: [u64; 12],
    schema: u8,
    root: Vec<u8>,
}

fn freeze_fixture() -> TestResult<FrozenFixture> {
    let mut fixture = fixture()?;
    assert_eq!(fixture.owner.observer_progress_witnesses().len(), 1);
    let held = fixture
        .owner
        .encode_live()
        .map_err(|error| format!("{error:?}"));
    fixture.owner.prune_durable_observer_witnesses()?;
    assert_eq!(fixture.owner.observer_progress_witnesses().len(), 0);
    assert_eq!(
        fixture.owner.observer_progress_pruned_through(),
        fixture.positions.observer_progress
    );
    let pruned = fixture
        .owner
        .encode_live()
        .map_err(|error| format!("{error:?}"));
    let frozen = FrozenFixture {
        held,
        pruned,
        positions: fixture.positions,
        root: fixture.root,
        config: fixture.config,
    };
    if let Some(path) = std::env::var_os("LIMINAL_OBSERVER_RED_FIXTURE_EXPORT") {
        std::fs::write(path, serde_json::to_vec(&red_fixture(&frozen)?)?)?;
    }
    Ok(frozen)
}

#[test]
fn checkpoint_observer_held_and_pruned_capture_restore_keep_native_sources() -> TestResult {
    let fixture = freeze_fixture()?;
    let imported = std::env::var_os("LIMINAL_OBSERVER_RED_FIXTURE_IMPORT")
        .map(|path| imported_fixture(&std::fs::read(path)?))
        .transpose()?;
    let restore = imported.as_ref().unwrap_or(&fixture);
    let controls = [
        (
            "held witness capture",
            fixture
                .held
                .as_ref()
                .map_err(|error| error.clone().into())
                .and_then(|bytes| capture_control(bytes, "held witness capture", 1)),
        ),
        (
            "pruned bookkeeping capture",
            fixture
                .pruned
                .as_ref()
                .map_err(|error| error.clone().into())
                .and_then(|bytes| capture_control(bytes, "pruned bookkeeping capture", 0)),
        ),
        (
            "held witness restore",
            restore
                .held
                .as_ref()
                .map_err(|error| error.clone().into())
                .and_then(|bytes| restore_control(restore, bytes, "held witness restore", 1)),
        ),
        (
            "pruned bookkeeping restore",
            restore
                .pruned
                .as_ref()
                .map_err(|error| error.clone().into())
                .and_then(|bytes| restore_control(restore, bytes, "pruned bookkeeping restore", 0)),
        ),
    ];
    let mut failures = Vec::new();
    for (name, result) in controls {
        if let Err(error) = result {
            println!("observer_checkpoint_control {name} FAILED: {error:?}");
            failures.push(format!("{name}: {error:?}"));
        } else {
            println!("observer_checkpoint_control {name} passed");
        }
    }
    println!(
        "observer_checkpoint_controls total=4 passed={} failed={}",
        4 - failures.len(),
        failures.len()
    );
    if !failures.is_empty() {
        return Err(failures.join("; ").into());
    }
    let held = fixture.held.as_ref().map_err(String::as_str)?;
    let pruned = fixture.pruned.as_ref().map_err(String::as_str)?;
    assert_ne!(
        CheckpointOwnerBody::decode_canonical(held)?.observer_progress_witnesses,
        CheckpointOwnerBody::decode_canonical(pruned)?.observer_progress_witnesses
    );
    refusal_controls(&fixture, pruned)
}

fn fixture() -> TestResult<Fixture> {
    let config = test_participant_config();
    let conversation_id = 59;
    let incarnation = ConnectionIncarnation::new(89, 7);
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
    let mut conversations = ParticipantConnectionConversations::default();
    let enrolled = dispatch_tracked(
        &handler,
        incarnation,
        &mut conversations,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([61; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(receipt) = enrolled else {
        return Err("observer fixture enrollment refused".into());
    };
    // The enrollee's own Attached record at sequence 1 is sender-excluded
    // from its recipient snapshot, so it is never an ack target. A peer's
    // enrollment commits the next Attached record, which every bound slot
    // but the peer's own is owed: that sequence is the one delivery the
    // observed participant can acknowledge.
    let peer = dispatch_tracked(
        &handler,
        ConnectionIncarnation::new(89, 8),
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([62; 16]),
        }),
    )?;
    assert!(
        matches!(peer, ServerValue::EnrollBound(_)),
        "observer fixture peer enrollment refused: {peer:?}"
    );
    let peer_attached_seq = 2;
    let acknowledged = dispatch_tracked(
        &handler,
        incarnation,
        &mut conversations,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id,
            participant_id: receipt.participant_id(),
            capability_generation: Generation::ONE,
            through_seq: peer_attached_seq,
        }),
    )?;
    assert!(
        matches!(acknowledged, ServerValue::AckCommitted(_)),
        "observer fixture ack did not commit: {acknowledged:?}"
    );
    let log = OperationLog::new(Arc::clone(&store), conversation_id);
    let outbox_log = OutboxLog::new(Arc::clone(&store), conversation_id);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    let (mut owner, audit, schema, _) = block_on(ConversationAuthority::replay_for_rebuild(
        conversation_id,
        &log,
        &outbox_log,
        &config,
        limits,
    ))??;
    let history = block_on(capture_identity_history(
        store.as_ref(),
        &owner,
        CheckpointHistoryRoot::empty(conversation_id),
    ))??;
    owner.retain_checkpoint_metadata(history.root, &audit)?;
    let extension = owner
        .outbox
        .as_ref()
        .ok_or("observer fixture outbox absent")?
        .next_extension_sequence();
    let positions = owner.checkpoint_positions(schema, owner.next_log_sequence, extension)?;
    assert_eq!(positions.observer_progress, peer_attached_seq);
    Ok(Fixture {
        owner,
        positions,
        root: history.root.encode_canonical().to_vec(),
        config,
    })
}

fn capture_control(encoded: &[u8], name: &str, held: usize) -> TestResult {
    let section = CheckpointOwnerBody::decode_canonical(encoded)?.observer_progress_witnesses;
    assert!(!section.is_empty(), "{name} omitted observer bookkeeping");
    let state: serde_json::Value = serde_json::from_slice(section)?;
    assert_eq!(
        state["witnesses"]
            .as_array()
            .ok_or("witness list absent")?
            .len(),
        held
    );
    let pruned = usize::from(held == 0);
    assert_eq!(
        state["pruned_sources"]
            .as_array()
            .ok_or("pruned source list absent")?
            .len(),
        pruned
    );
    for source in state["witnesses"]
        .as_array()
        .ok_or("witness list absent")?
        .iter()
        .chain(
            state["pruned_sources"]
                .as_array()
                .ok_or("pruned source list absent")?,
        )
    {
        assert!(
            !source["source"]
                .as_array()
                .ok_or("native source absent")?
                .is_empty()
        );
    }
    println!("{name}: held={held} pruned_native_sources={pruned}");
    Ok(())
}

fn restore_control(fixture: &FrozenFixture, encoded: &[u8], name: &str, held: usize) -> TestResult {
    let context = ConversationAuthorityRestoreContext::new(
        fixture.positions,
        &fixture.root,
        &fixture.config,
    )?;
    let restored =
        ConversationAuthority::decode_live(encoded, &fixture.positions, &fixture.root, &context)?;
    assert_eq!(restored.observer_progress_witnesses().len(), held);
    let pruned_through = if held == 0 {
        fixture.positions.observer_progress
    } else {
        0
    };
    assert_eq!(restored.observer_progress_pruned_through(), pruned_through);
    assert_eq!(
        restored.encode_live()?,
        encoded,
        "{name} changed saved owner bytes"
    );
    println!("{name}: exact_owner_bytes={} held={held}", encoded.len());
    Ok(())
}

fn refusal_controls(fixture: &FrozenFixture, encoded: &[u8]) -> TestResult {
    let mut body = CheckpointOwnerBody::decode_canonical(encoded)?;
    let section: serde_json::Value = serde_json::from_slice(body.observer_progress_witnesses)?;
    let actual_source = serde_json::to_string(&section["pruned_sources"][0]["source"])?;
    let field = format!("\"source\":{actual_source}");
    let section = std::str::from_utf8(body.observer_progress_witnesses)?;
    assert!(section.contains(&field));
    let missing = section.replacen(&field, "\"source\":[]", 1).into_bytes();
    body.observer_progress_witnesses = &missing;
    let missing = body.encode_canonical()?;
    let context = ConversationAuthorityRestoreContext::new(
        fixture.positions,
        &fixture.root,
        &fixture.config,
    )?;
    assert!(
        ConversationAuthority::decode_live(&missing, &fixture.positions, &fixture.root, &context)
            .is_err()
    );
    body.observer_progress_witnesses = &[];
    let omitted = body.encode_canonical()?;
    assert!(
        ConversationAuthority::decode_live(&omitted, &fixture.positions, &fixture.root, &context)
            .is_err()
    );
    Ok(())
}

fn red_fixture(fixture: &FrozenFixture) -> TestResult<RedFixture> {
    let p = fixture.positions;
    Ok(RedFixture {
        held: fixture.held.as_ref().map_err(String::as_str)?.clone(),
        pruned: fixture.pruned.as_ref().map_err(String::as_str)?.clone(),
        positions: [
            p.conversation_id,
            p.covered_log_sequence,
            p.next_event_ordinal,
            p.next_log_sequence,
            p.next_seq,
            p.next_order,
            p.next_participant,
            p.extension_sequence,
            p.highest_delivery_seq,
            p.observer_progress,
            p.merged_base_sequence,
            p.merged_extension_sequence,
        ],
        schema: match p.schema {
            CheckpointSchema::V2Prefix => 2,
            CheckpointSchema::V3Suffix => 3,
            CheckpointSchema::V4Suffix => 4,
        },
        root: fixture.root.clone(),
    })
}

fn imported_fixture(bytes: &[u8]) -> TestResult<FrozenFixture> {
    let saved: RedFixture = serde_json::from_slice(bytes)?;
    let p = saved.positions;
    let schema = match saved.schema {
        2 => CheckpointSchema::V2Prefix,
        3 => CheckpointSchema::V3Suffix,
        4 => CheckpointSchema::V4Suffix,
        _ => return Err("observer red fixture schema is invalid".into()),
    };
    let positions = CheckpointPositions {
        conversation_id: p[0],
        covered_log_sequence: p[1],
        next_event_ordinal: p[2],
        next_log_sequence: p[3],
        next_seq: p[4],
        next_order: p[5],
        next_participant: p[6],
        extension_sequence: p[7],
        highest_delivery_seq: p[8],
        observer_progress: p[9],
        merged_base_sequence: p[10],
        merged_extension_sequence: p[11],
        schema,
    };
    positions.validate()?;
    Ok(FrozenFixture {
        held: Ok(saved.held),
        pruned: Ok(saved.pruned),
        positions,
        root: saved.root,
        config: test_participant_config(),
    })
}