#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
//! Phase 1 chokepoint tests for runtime/store/input-ledger recovery replay.

use std::sync::Arc;

use chrono::Utc;
use meerkat_core::BlobStore;
use meerkat_core::lifecycle::run_primitive::RunApplyBoundary;
use meerkat_core::lifecycle::{InputId, RunId};
use meerkat_runtime::PersistentRuntimeDriver;
use meerkat_runtime::identifiers::LogicalRuntimeId;
use meerkat_runtime::input::{
    Input, InputDurability, InputHeader, InputOrigin, InputVisibility, PromptInput,
};
use meerkat_runtime::input_state::{
    InputLifecycleState, InputState, InputStateSeed, StoredInputState,
};
use meerkat_runtime::store::{InMemoryRuntimeStore, RuntimeStore};
use meerkat_runtime::traits::RuntimeDriver;
use meerkat_store::MemoryBlobStore;
use uuid::Uuid;

fn memory_blob_store() -> Arc<dyn BlobStore> {
    Arc::new(MemoryBlobStore::new())
}

fn make_runtime_id(label: &str) -> LogicalRuntimeId {
    LogicalRuntimeId::new(format!("phase1-recovery-{label}-{}", Uuid::now_v7()))
}

fn make_prompt(text: &str) -> Input {
    Input::Prompt(PromptInput {
        header: InputHeader {
            id: InputId::new(),
            timestamp: Utc::now(),
            source: InputOrigin::Operator,
            durability: InputDurability::Durable,
            visibility: InputVisibility::default(),
            idempotency_key: None,
            supersession_key: None,
            correlation_id: None,
        },
        text: text.into(),
        blocks: None,
        typed_turn_appends: Vec::new(),
        turn_metadata: None,
    })
}

fn stamp_runtime_metadata(state: &mut InputState, input: &Input) {
    let policy = meerkat_runtime::DefaultPolicyTable::resolve(input, true);
    let policy_version = policy.policy_version;
    state.runtime_semantics = Some(
        meerkat_runtime::ingress_types::RuntimeInputSemantics::from_policy_and_kind(
            &policy,
            input.kind(),
        ),
    );
    state.policy = Some(meerkat_runtime::input_state::PolicySnapshot {
        version: policy_version,
        decision: policy,
    });
}

fn applied_pending_state(input: &Input, run_id: &RunId, sequence: u64) -> StoredInputState {
    let mut state = InputState::new_accepted(input.id().clone());
    state.persisted_input = Some(input.clone());
    state.durability = Some(InputDurability::Durable);
    stamp_runtime_metadata(&mut state, input);
    // Simulate Accepted → Queued → Staged → Applied → AppliedPendingConsumption
    // by seeding the DSL-owned phase + run association alongside the shell.
    state.attempt_count = 1;
    StoredInputState {
        state,
        seed: InputStateSeed {
            phase: InputLifecycleState::AppliedPendingConsumption,
            last_run_id: Some(run_id.clone()),
            last_boundary_sequence: Some(sequence),
            terminal_outcome: None,
            attempt_count: 1,
        },
    }
}

fn sorted_ids(ids: impl IntoIterator<Item = InputId>) -> Vec<String> {
    let mut ids = ids.into_iter().map(|id| id.to_string()).collect::<Vec<_>>();
    ids.sort();
    ids
}

#[tokio::test]
async fn recovery_replay_red_ok_requeues_missing_boundary_contributors_through_persistent_driver() {
    let runtime_id = make_runtime_id("missing-receipt");
    let run_id = RunId::new();
    let first = make_prompt("first replay contribution");
    let second = make_prompt("second replay contribution");
    let first_id = first.id().clone();
    let second_id = second.id().clone();
    let expected = sorted_ids(vec![first_id.clone(), second_id.clone()]);
    let store: Arc<dyn RuntimeStore> = Arc::new(InMemoryRuntimeStore::new());

    store
        .persist_input_state(&runtime_id, &applied_pending_state(&first, &run_id, 0))
        .await
        .expect("persist first applied state");
    store
        .persist_input_state(&runtime_id, &applied_pending_state(&second, &run_id, 0))
        .await
        .expect("persist second applied state");

    let mut driver =
        PersistentRuntimeDriver::new(runtime_id.clone(), Arc::clone(&store), memory_blob_store());
    let report = driver.recover().await.expect("recover persistent driver");
    assert_eq!(
        report.inputs_recovered, 2,
        "missing boundary receipts should roll both contributors back into replay"
    );
    assert_eq!(
        sorted_ids(driver.active_input_ids()),
        expected,
        "recovery should preserve both contributors as active replay inputs"
    );

    for input_id in [&first_id, &second_id] {
        assert!(
            driver.input_state(input_id).is_some(),
            "driver should expose recovered input state"
        );
        assert_eq!(
            driver.inner_ref().input_phase(input_id),
            Some(InputLifecycleState::Queued),
            "DSL phase should be requeued after recovery"
        );

        let stored = store
            .load_input_state(&runtime_id, input_id)
            .await
            .expect("load persisted state")
            .expect("persisted input record");
        assert_eq!(stored.seed.phase, InputLifecycleState::Queued);
        assert_eq!(stored.seed.last_run_id, Some(run_id.clone()));
        assert_eq!(stored.seed.last_boundary_sequence, Some(0));
    }

    let replayed = vec![
        driver.dequeue_next().expect("first replay input").0,
        driver.dequeue_next().expect("second replay input").0,
    ];
    assert!(
        driver.dequeue_next().is_none(),
        "recovery should requeue only the missing-boundary contributors"
    );
    assert_eq!(
        sorted_ids(replayed),
        expected,
        "replay order should surface the recovered contributors exactly once"
    );
    let _ = RunApplyBoundary::RunStart;
}
