liminal-server 0.24.1

use std::error::Error;
use std::path::Path;

use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest,
    DetachAttemptToken, DetachRequest, EnrollBound, EnrollmentRequest, EnrollmentToken, Generation,
    LeaveAttemptToken, LeaveRequest, MarkerAck, ParticipantAck, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::restore_checkpoint_history::{CheckpointHistoryRoot, ROOT_ENCODED_BYTES};
use super::tests::{dispatch, open_disk_store_for_tests, test_participant_config};

/// What one owner of a conversation holds, read for a comparison: the
/// conversation state, and separately the checkpoint publication the owner
/// holds (its validated metadata and history root), because a live owner and
/// a cold reopen of the same bytes publish at different points (DUR-010
/// publishes at a `CHECKPOINT_EVERY` crossing or a seal; a rebuild publishes
/// the captured history on the first load that finds no checkpoint) and the
/// difference is exactly that record.
#[derive(Debug, PartialEq, Eq)]
pub(super) struct AuthoritySnapshot {
    /// Every field of the authority but the checkpoint publication.
    pub(super) state: String,
    /// The checkpoint metadata and history root the owner holds.
    pub(super) checkpoint: String,
    /// Whether the held history root is the conversation's empty root: the
    /// owner published no identity history.
    pub(super) empty_root: bool,
}

pub(super) fn authority_snapshot(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<AuthoritySnapshot, Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let mut owner = cell
        .lock()
        .map_err(|_| "conversation authority lock was poisoned")?;
    let authority = owner
        .as_mut()
        .ok_or("conversation authority must be restored")?;
    let metadata = authority.checkpoint_metadata.take();
    let root = authority.checkpoint_history_root.take();
    // The warm's pending cross-check is work a restore owes, not state.
    let check = authority.checkpoint_history_check.take();
    let state = format!("{authority:#?}");
    authority.checkpoint_history_check = check;
    let checkpoint = format!("{metadata:#?}\n{root:#?}");
    let empty_root = root == Some(CheckpointHistoryRoot::empty(conversation_id));
    authority.checkpoint_metadata = metadata;
    authority.checkpoint_history_root = root;
    drop(owner);
    Ok(AuthoritySnapshot {
        state,
        checkpoint,
        empty_root,
    })
}

/// A cold reopen of the bytes a live owner committed holds the same
/// conversation state, and the same checkpoint publication unless the live
/// owner published none yet, in which case the reopen holds the one its
/// rebuild published, over the captured history and never the empty root.
pub(super) fn assert_cold_matches_live(
    live: &AuthoritySnapshot,
    cold: &AuthoritySnapshot,
    step: &str,
) {
    assert_eq!(
        live.state, cold.state,
        "live/cold state diverged after {step}"
    );
    if live.checkpoint == cold.checkpoint {
        return;
    }
    assert!(
        live.empty_root,
        "after {step} the live owner holds a published checkpoint that a cold reopen of the same \
         bytes does not hold unchanged:\nlive {}\ncold {}",
        live.checkpoint, cold.checkpoint
    );
    assert!(
        !cold.empty_root,
        "after {step} the live owner published no checkpoint and the cold reopen's rebuild \
         published none either, yet they differ:\nlive {}\ncold {}",
        live.checkpoint, cold.checkpoint
    );
}

fn cold_reopen_matches_live(
    handler: ProductionParticipantHandler,
    data_dir: &Path,
    conversation_id: u64,
    history_step: &str,
) -> Result<ProductionParticipantHandler, Box<dyn Error>> {
    let live = authority_snapshot(&handler, conversation_id)?;
    drop(handler);
    let store = open_disk_store_for_tests(data_dir)?;
    let restored = ProductionParticipantHandler::new(store, test_participant_config())?;
    let cold = authority_snapshot(&restored, conversation_id)?;
    assert_cold_matches_live(&live, &cold, history_step);
    Ok(restored)
}

fn require_enrolled(value: ServerValue) -> Result<EnrollBound, Box<dyn Error>> {
    let ServerValue::EnrollBound(receipt) = value else {
        return Err(format!("history enrollment did not bind: {value:?}").into());
    };
    Ok(receipt)
}

fn walk_enrollment_history(
    data_dir: &Path,
    conversation_id: u64,
    first_connection: ConnectionIncarnation,
    second_connection: ConnectionIncarnation,
) -> Result<(ProductionParticipantHandler, EnrollBound, EnrollBound), Box<dyn Error>> {
    let store = open_disk_store_for_tests(data_dir)?;
    let mut handler = ProductionParticipantHandler::new(store, test_participant_config())?;
    let first = require_enrolled(dispatch(
        &handler,
        first_connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([0xD0; 16]),
        }),
    )?)?;
    handler = cold_reopen_matches_live(handler, data_dir, conversation_id, "initial enrollment")?;

    let second = require_enrolled(dispatch(
        &handler,
        second_connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([0xD1; 16]),
        }),
    )?)?;
    handler =
        cold_reopen_matches_live(handler, data_dir, conversation_id, "subsequent enrollment")?;
    Ok((handler, first, second))
}

fn walk_attach_detach_and_ack_history(
    mut handler: ProductionParticipantHandler,
    data_dir: &Path,
    conversation_id: u64,
    first_connection: ConnectionIncarnation,
    second_connection: ConnectionIncarnation,
    first: &EnrollBound,
    second: &EnrollBound,
) -> Result<ProductionParticipantHandler, Box<dyn Error>> {
    let attached = dispatch(
        &handler,
        first_connection,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id,
            participant_id: first.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: first.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([0xD2; 16]),
            accept_marker_delivery_seq: None,
        }),
    )?;
    let ServerValue::AttachBound(attached) = attached else {
        return Err(format!("history attach did not bind: {attached:?}").into());
    };
    assert_eq!(attached.capability_generation().get(), 2);
    handler = cold_reopen_matches_live(handler, data_dir, conversation_id, "attach")?;

    let detached = dispatch(
        &handler,
        first_connection,
        ClientRequest::Detach(DetachRequest {
            conversation_id,
            participant_id: first.participant_id(),
            capability_generation: attached.capability_generation(),
            detach_attempt_token: DetachAttemptToken::new([0xD3; 16]),
        }),
    )?;
    assert!(matches!(detached, ServerValue::DetachCommitted(_)));
    handler = cold_reopen_matches_live(handler, data_dir, conversation_id, "detach")?;

    let acknowledged = dispatch(
        &handler,
        second_connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id,
            participant_id: second.participant_id(),
            capability_generation: Generation::ONE,
            through_seq: 5,
        }),
    )?;
    assert!(matches!(acknowledged, ServerValue::AckCommitted(_)));
    handler = cold_reopen_matches_live(handler, data_dir, conversation_id, "participant ack")?;

    let marker_ack = dispatch(
        &handler,
        second_connection,
        ClientRequest::MarkerAck(MarkerAck {
            conversation_id,
            participant_id: second.participant_id(),
            capability_generation: Generation::ONE,
            marker_delivery_seq: 5,
        }),
    )?;
    assert!(matches!(
        marker_ack,
        ServerValue::MarkerMismatch(_) | ServerValue::MarkerNotDelivered(_)
    ));
    cold_reopen_matches_live(handler, data_dir, conversation_id, "marker ack")
}

fn walk_record_and_leave_history(
    mut handler: ProductionParticipantHandler,
    data_dir: &Path,
    conversation_id: u64,
    connection: ConnectionIncarnation,
    participant: &EnrollBound,
) -> Result<(), Box<dyn Error>> {
    for (index, token) in [0xD4_u8, 0xD5].into_iter().enumerate() {
        let recorded = dispatch(
            &handler,
            connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id,
                participant_id: participant.participant_id(),
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new([token; 16]),
                payload: vec![token],
            }),
        )?;
        assert!(matches!(recorded, ServerValue::RecordCommitted(_)));
        let step = format!("ordinary record {}", index + 1);
        handler = cold_reopen_matches_live(handler, data_dir, conversation_id, &step)?;
    }

    let left = dispatch(
        &handler,
        connection,
        ClientRequest::Leave(LeaveRequest {
            conversation_id,
            participant_id: participant.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: participant.attach_secret(),
            leave_attempt_token: LeaveAttemptToken::new([0xD6; 16]),
        }),
    )?;
    assert!(matches!(left, ServerValue::LeaveCommitted(_)));
    let _ = cold_reopen_matches_live(handler, data_dir, conversation_id, "Leave")?;
    Ok(())
}

#[test]
fn every_transition_history_matches_live_state_after_cold_restore() -> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let data_dir = home.path().join("durability");
    let conversation_id = 906;
    let first_connection = ConnectionIncarnation::new(88, 1);
    let second_connection = ConnectionIncarnation::new(88, 2);
    let (handler, first, second) = walk_enrollment_history(
        &data_dir,
        conversation_id,
        first_connection,
        second_connection,
    )?;
    let handler = walk_attach_detach_and_ack_history(
        handler,
        &data_dir,
        conversation_id,
        first_connection,
        second_connection,
        &first,
        &second,
    )?;
    walk_record_and_leave_history(
        handler,
        &data_dir,
        conversation_id,
        second_connection,
        &second,
    )
}

fn record(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant: &EnrollBound,
    conversation_id: u64,
    token: u8,
) -> Result<(), Box<dyn Error>> {
    let recorded = dispatch(
        handler,
        connection,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id,
            participant_id: participant.participant_id(),
            capability_generation: Generation::ONE,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new([token; 16]),
            payload: vec![token],
        }),
    )?;
    if !matches!(recorded, ServerValue::RecordCommitted(_)) {
        return Err(format!("record {token:#x} did not commit: {recorded:?}").into());
    }
    Ok(())
}

/// One identity's digest in the verified record.
const DIGEST_BYTES: usize = 32;

/// The admissions the owner holds, and the history root it holds them over.
fn admissions_and_root(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<(usize, Option<CheckpointHistoryRoot>), Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let owner = cell
        .lock()
        .map_err(|_| "conversation authority lock was poisoned")?;
    let authority = owner
        .as_ref()
        .ok_or("conversation authority must be restored")?;
    let held = (
        authority.committed_admissions.len(),
        authority.checkpoint_history_root,
    );
    drop(owner);
    Ok(held)
}

/// The door's room, every boot from 11 October: a post after the last
/// checkpoint adds an admission the index at that checkpoint's root cannot
/// hold, and the warm's cross-check, run on the owner after its tail had
/// replayed, refused the room as "retained history identity differs from its
/// authenticated checkpoint". The warm checks what the checkpoint's body held.
/// And it read all 50,011 of them on every boot: a root once verified is
/// recorded, a boot over it reads none, and a checkpoint one post later reads
/// exactly that post's admission and writes exactly its digest and the root
/// to the record, not the history.
#[test]
fn the_warms_history_check_reads_only_what_changed_since_the_root_it_verified()
-> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let data_dir = home.path().join("durability");
    let conversation_id = 907;
    let connection = ConnectionIncarnation::new(89, 1);
    let handler = ProductionParticipantHandler::new(
        open_disk_store_for_tests(&data_dir)?,
        test_participant_config(),
    )?;
    let participant = require_enrolled(dispatch(
        &handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([0xE0; 16]),
        }),
    )?)?;
    record(&handler, connection, &participant, conversation_id, 0xE1)?;
    drop(handler);

    // The first cold load finds no checkpoint, rebuilds, and publishes one
    // over every identity it holds.
    let handler = ProductionParticipantHandler::new(
        open_disk_store_for_tests(&data_dir)?,
        test_participant_config(),
    )?;
    handler.ensure_restored(conversation_id)?;
    assert!(
        !authority_snapshot(&handler, conversation_id)?.empty_root,
        "the rebuild published no identity history"
    );
    let (_, published) = admissions_and_root(&handler, conversation_id)?;
    record(&handler, connection, &participant, conversation_id, 0xE2)?;
    let (live_admissions, live_root) = admissions_and_root(&handler, conversation_id)?;
    assert_eq!(live_root, published, "the post crossed a checkpoint");
    drop(handler);

    // The second restores from that checkpoint and replays the post behind it.
    let handler = ProductionParticipantHandler::new(
        open_disk_store_for_tests(&data_dir)?,
        test_participant_config(),
    )?;
    handler.ensure_restored(conversation_id)?;
    assert_eq!(
        admissions_and_root(&handler, conversation_id)?,
        (live_admissions, published),
        "the restore did not come from the published checkpoint with the post replayed behind it"
    );
    let first = handler.verify_identity_history(conversation_id)?;
    assert!(first.checked > 0, "the warm checked no identity");
    assert_eq!(
        first.checked, first.identities,
        "with no verified root recorded, every identity is read"
    );
    // The first record: every identity's digest, and the root.
    assert_eq!(
        first.recorded_bytes,
        first.identities * DIGEST_BYTES + ROOT_ENCODED_BYTES
    );
    // Taken once: a second warm pass has nothing left to check.
    assert_eq!(
        handler.verify_identity_history(conversation_id)?,
        super::lazy_restore::HistoryCheckRead::default()
    );
    drop(handler);

    // A second boot over the same checkpoint covers every identity and reads
    // none: the root is recorded as verified.
    let handler = ProductionParticipantHandler::new(
        open_disk_store_for_tests(&data_dir)?,
        test_participant_config(),
    )?;
    handler.ensure_restored(conversation_id)?;
    assert_eq!(
        handler.verify_identity_history(conversation_id)?,
        super::lazy_restore::HistoryCheckRead {
            checked: 0,
            identities: first.identities,
            recorded_bytes: 0,
        }
    );

    // A checkpoint that takes in the one post behind the verified root: the
    // next boot reads that post's admission and nothing else.
    assert!(
        handler.publish_owed_checkpoint(conversation_id)?,
        "the live owner could not publish its checkpoint"
    );
    let (_, crossed) = admissions_and_root(&handler, conversation_id)?;
    assert_ne!(crossed, published, "the publication did not move the root");
    drop(handler);
    let handler = ProductionParticipantHandler::new(
        open_disk_store_for_tests(&data_dir)?,
        test_participant_config(),
    )?;
    handler.ensure_restored(conversation_id)?;
    assert_eq!(
        handler.verify_identity_history(conversation_id)?,
        // The record takes that one digest and the root: what changed, not
        // the history.
        super::lazy_restore::HistoryCheckRead {
            checked: 1,
            identities: first.identities + 1,
            recorded_bytes: DIGEST_BYTES + ROOT_ENCODED_BYTES,
        }
    );
    Ok(())
}