liminal-server 0.24.1

//! DUR-027 R3: the observer stream carries a retire row at schema version 3; a
//! restore folds it out through the aggregate's own retirement, and a retire
//! the aggregate refuses is a corrupt stream.

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

use liminal::durability::{DurableStore, bridge::block_on};
use liminal_protocol::wire::{ConversationId, DeliverySeq};

use super::log::OperationLogError;
use super::observer::{OBSERVER_STREAM_KEY, ObserverLog, ObserverRow, encode_for_tests};
use super::tests_w3_restore_fixture::new_store;

/// What every test here answers.
type TestResult = Result<(), Box<dyn Error>>;

fn track(conversation_id: ConversationId, observer_progress: DeliverySeq) -> ObserverRow {
    ObserverRow::Track {
        conversation_id,
        observer_progress,
    }
}

const fn retire(conversation_id: ConversationId) -> ObserverRow {
    ObserverRow::Retire { conversation_id }
}

/// Writes `rows` as the stream's first rows, encoded as the appender stores
/// them, with one flush.
fn stream_of(rows: &[ObserverRow]) -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let store = new_store()?;
    let payloads = rows
        .iter()
        .map(encode_for_tests)
        .collect::<Result<Vec<_>, _>>()?;
    block_on(store.append_all(OBSERVER_STREAM_KEY, payloads, 0))??;
    block_on(store.flush())??;
    Ok(store)
}

/// The restore's refusal, or the aggregate's rows it answered instead.
fn refusal(store: Arc<dyn DurableStore>) -> Result<OperationLogError, Box<dyn Error>> {
    match block_on(ObserverLog::new(store).restore())? {
        Ok(restored) => Err(format!(
            "restore answered {:?} instead of refusing",
            restored.aggregate.progress_rows()
        )
        .into()),
        Err(error) => Ok(error),
    }
}

#[test]
fn a_retire_row_folds_its_conversation_out_and_leaves_the_others() -> TestResult {
    let store = stream_of(&[track(1, 4), track(2, 5), track(3, 6), retire(2)])?;
    let restored = block_on(ObserverLog::new(store).restore())??;
    assert_eq!(restored.aggregate.progress_rows(), vec![(1, 4), (3, 6)]);
    assert_eq!(restored.aggregate.observer_progress(2), None);
    assert_eq!(restored.next_sequence, 4);
    Ok(())
}

#[test]
fn a_retired_id_tracks_again_after_its_retire_row() -> TestResult {
    // The aggregate keeps no retired set; the server's tombstone refuses a
    // retired id before a track is ever written (R2).
    let store = stream_of(&[track(1, 4), retire(1), track(1, 0)])?;
    let restored = block_on(ObserverLog::new(store).restore())??;
    assert_eq!(restored.aggregate.progress_rows(), vec![(1, 0)]);
    Ok(())
}

#[test]
fn a_retire_of_an_armed_conversation_is_a_corrupt_stream() -> TestResult {
    let store = stream_of(&[
        track(1, 4),
        ObserverRow::Arms { arms: vec![(1, 4)] },
        retire(1),
    ])?;
    let error = refusal(store)?;
    assert!(
        matches!(error, OperationLogError::CorruptRow { .. }),
        "{error:?}"
    );
    Ok(())
}

#[test]
fn a_retire_of_an_untracked_conversation_is_a_corrupt_stream() -> TestResult {
    let store = stream_of(&[track(1, 4), retire(9)])?;
    let error = refusal(store)?;
    assert!(
        matches!(error, OperationLogError::CorruptRow { .. }),
        "{error:?}"
    );
    Ok(())
}

#[test]
fn a_retire_row_is_stored_at_schema_version_3() -> TestResult {
    let stored: serde_json::Value = serde_json::from_slice(&encode_for_tests(&retire(7))?)?;
    assert_eq!(stored["schema_version"], 3);
    assert_eq!(stored["row"]["row"], "retire");
    assert_eq!(stored["row"]["conversation_id"], 7);
    Ok(())
}

#[test]
fn a_retire_row_at_another_schema_version_is_refused_by_its_version() -> TestResult {
    let store = new_store()?;
    let forged = br#"{"schema_version":1,"row":{"row":"retire","conversation_id":1}}"#.to_vec();
    block_on(store.append_all(
        OBSERVER_STREAM_KEY,
        vec![encode_for_tests(&track(1, 4))?, forged],
        0,
    ))??;
    block_on(store.flush())??;
    let error = refusal(store)?;
    assert!(
        matches!(error, OperationLogError::SchemaVersion(1)),
        "{error:?}"
    );
    Ok(())
}