//! Crash-injection tests for C2 + C12.
//!
//! Phase 4c verified the runtime contracts via post-condition assertions
//! on the happy path. These tests verify the same contracts hold under
//! failure: a configured backend method returns `Err` mid-flight (the
//! moral equivalent of a kill -9 between two backend writes), and the
//! tests assert that no partial state is observable and that a fresh
//! step recovers correctly.
//!
//! For MemoryStore, atomic operations are guarded by a single Mutex,
//! so the `commit_reactor_batch` atomicity is structurally guaranteed
//! at the impl level. These tests therefore focus on the runner-level
//! recovery contract: when an inner backend method returns Err, the
//! next step picks up where it left off without duplicating state.
//!
//! Faults are armed before the call and consumed atomically — armed
//! fault returns Err exactly once, then the wrapper passes through to
//! the inner backend for subsequent calls. This models the "process
//! crashed once mid-step, restarted, retry succeeds" recovery shape.

use std::sync::Arc;

use anyhow::{anyhow, Result};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use causal::checkpoint_store::{CheckpointStore, ReactorCheckpoint};
use causal::contexts::Ctx;
use causal::event_log::EventLogBackend;
use causal::event::Event;
use causal::projector::Projector;
use causal::memory_store::MemoryStore;
use causal::projection_runner::{ProjectionRunner, StepOutcome};
use causal::reactor::Events;
use causal::reactor_runner::ReactorRunner;
use causal::reactor::Reactor;
use causal::types::{WriteResult, LogCursor, EventData, RecordedEvent, StreamRevision};

// ─────────────────────────────────────────────────────────────────────
// Fault injector — wraps MemoryStore; lets tests arm one Err per fault
// point. Once consumed, the wrapper passes through.
// ─────────────────────────────────────────────────────────────────────

#[derive(Clone, Debug)]
enum FaultPoint {
    Append,
    CheckpointSet,
}

struct FaultInjector {
    inner: Arc<MemoryStore>,
    armed: Mutex<Option<FaultPoint>>,
}

impl FaultInjector {
    fn new(inner: Arc<MemoryStore>) -> Arc<Self> {
        Arc::new(Self { inner, armed: Mutex::new(None) })
    }

    fn arm(&self, fault: FaultPoint) {
        *self.armed.lock() = Some(fault);
    }

    fn take_if_matches(&self, target: FaultPoint) -> bool {
        let mut armed = self.armed.lock();
        match (&*armed, &target) {
            (Some(FaultPoint::Append),        FaultPoint::Append)
          | (Some(FaultPoint::CheckpointSet), FaultPoint::CheckpointSet) => {
                *armed = None;
                true
            }
            _ => false,
        }
    }
}

#[async_trait]
impl EventLogBackend for FaultInjector {
    async fn read_all(
        &self, after: LogCursor, limit: usize,
    ) -> Result<Vec<RecordedEvent>> {
        EventLogBackend::read_all(self.inner.as_ref(), after, limit).await
    }
    async fn read_stream(
        &self, aggregate_type: &str, aggregate_id: Uuid,
        after: Option<StreamRevision>,
    ) -> Result<Vec<RecordedEvent>> {
        EventLogBackend::read_stream(
            self.inner.as_ref(), aggregate_type, aggregate_id, after,
        ).await
    }
    async fn latest_position(&self) -> Result<LogCursor> {
        EventLogBackend::latest_position(self.inner.as_ref()).await
    }

    async fn append_to_stream(
        &self,
        aggregate_type: &str,
        aggregate_id: Uuid,
        expected: causal::types::StreamState,
        events: Vec<EventData>,
    ) -> Result<WriteResult> {
        // All appends flow through this one primitive now, so the
        // `Append` fault point fires here.
        if self.take_if_matches(FaultPoint::Append) {
            return Err(anyhow!("fault: append"));
        }
        EventLogBackend::append_to_stream(
            self.inner.as_ref(), aggregate_type, aggregate_id, expected, events,
        ).await
    }
}

#[async_trait]
impl CheckpointStore for FaultInjector {
    async fn get(&self, consumer_id: &str) -> Result<Option<LogCursor>> {
        CheckpointStore::get(self.inner.as_ref(), consumer_id).await
    }
    async fn set(&self, consumer_id: &str, pos: LogCursor) -> Result<()> {
        if self.take_if_matches(FaultPoint::CheckpointSet) {
            return Err(anyhow!("fault: checkpoint set"));
        }
        CheckpointStore::set(self.inner.as_ref(), consumer_id, pos).await
    }
}

#[async_trait]
impl ReactorCheckpoint for FaultInjector {
    async fn record_reactor_attempt(
        &self,
        consumer_id: &str,
        source_event_id: Uuid,
    ) -> Result<u32> {
        self.inner.record_reactor_attempt(consumer_id, source_event_id).await
    }

    async fn clear_reactor_attempts(
        &self,
        consumer_id: &str,
        source_event_id: Uuid,
    ) -> Result<()> {
        self.inner.clear_reactor_attempts(consumer_id, source_event_id).await
    }
}

// ─────────────────────────────────────────────────────────────────────
// Test fixtures
// ─────────────────────────────────────────────────────────────────────

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Recorded { id: Uuid, occurred_at: DateTime<Utc> }
impl Event for Recorded {
    const CATEGORY: &'static str = "test";
    fn event_type(&self) -> &str { "recorded" }
    fn stream_id(&self) -> Uuid { self.id }
    fn occurred_at(&self) -> Option<DateTime<Utc>> { Some(self.occurred_at) }
}

async fn append_n(store: &MemoryStore, n: usize) {
    for _ in 0..n {
        let payload = Recorded { id: Uuid::new_v4(), occurred_at: Utc::now() };
        let ev = EventData {
            event_id: Uuid::new_v4(),
            causation_id: None,
            correlation_id: Uuid::new_v4(),
            event_type: format!("{}:{}", <Recorded as Event>::CATEGORY, payload.event_type()),
            payload: serde_json::to_value(&payload).unwrap(),
            created_at: Utc::now(),
            category: None,
            stream_id: None,
            metadata: serde_json::Map::new(),
            ephemeral: None,
            persistent: true,
        };
        causal::append_event(store, ev).await.unwrap();
    }
}

#[derive(Default, Clone)]
struct CountingProjector {
    seen_event_ids: Arc<Mutex<Vec<Uuid>>>,
}

#[async_trait]
impl Projector for CountingProjector {
    type Event = Recorded;
    const GROUP_NAME: &'static str = "counting";
    async fn project(
        &self, _fact: &Recorded, ctx: Ctx<'_>,
    ) -> Result<()> {
        // Idempotent: track unique event_ids; record every call too.
        self.seen_event_ids.lock().push(ctx.event_id);
        Ok(())
    }
}

// ─────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn cursor_set_failure_after_project_redelivers_idempotently() {
    // Models: project() returned Ok, then process crashed before
    // cursor.set persisted. On restart, the fact redelivers; idempotent
    // project absorbs the duplicate.
    let inner = Arc::new(MemoryStore::new());
    append_n(&inner, 3).await;
    let injector = FaultInjector::new(inner.clone());

    let m = CountingProjector::default();
    let runner = ProjectionRunner::new(
        m.clone(),
        "m1",
        injector.clone() as Arc<dyn EventLogBackend>,
        injector.clone() as Arc<dyn CheckpointStore>,
    );

    // Arm fault on the FIRST checkpoint.set — happens after fact[0]
    // was projected successfully.
    injector.arm(FaultPoint::CheckpointSet);

    let result = runner.step(10).await;
    assert!(result.is_err(), "first step errored on cursor set");

    // Materialize was called once; cursor did NOT advance.
    assert_eq!(m.seen_event_ids.lock().len(), 1);
    assert!(
        injector.get("m1").await.unwrap().is_none(),
        "cursor must not have advanced past the failed set"
    );

    // Next step (no fault armed) should redeliver fact[0] AND continue
    // through fact[1], fact[2]. Idempotent projector absorbs the
    // duplicate fact[0] call without ill effect.
    let outcome = runner.step(10).await.unwrap();
    assert!(matches!(outcome, StepOutcome::Progressed { applied: 3 }));
    assert_eq!(m.seen_event_ids.lock().len(), 4,
               "fact[0] called twice (idempotent), fact[1] + fact[2] once each");

    // Cursor at last fact's position.
    let cursor = injector.get("m1").await.unwrap().unwrap();
    let events = EventLogBackend::read_all(
        inner.as_ref(), LogCursor::ZERO, 10,
    ).await.unwrap();
    assert_eq!(cursor, events.last().unwrap().position);
}

// Reactor + outbox fault tests
// ─────────────────────────────────────────────────

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Trigger { id: Uuid, occurred_at: DateTime<Utc> }
impl Event for Trigger {
    const CATEGORY: &'static str = "test";
    fn event_type(&self) -> &str { "trigger" }
    fn stream_id(&self) -> Uuid { self.id }
    fn occurred_at(&self) -> Option<DateTime<Utc>> { Some(self.occurred_at) }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Echoed;
impl Event for Echoed {
    // Distinct category from the trigger ("test") — self-trigger
    // discipline: a reactor must not emit into its own trigger category,
    // else it would consume its own output. Now that outputs land in the
    // log immediately (no outbox), this is enforced in practice.
    const CATEGORY: &'static str = "echo";
    fn event_type(&self) -> &str { "echoed" }
    fn stream_id(&self) -> Uuid { Uuid::nil() }
}

struct EmitOne;
#[async_trait]
impl Reactor for EmitOne {
    type Trigger = Trigger;
    const GROUP_NAME: &'static str = "emit-one";
    async fn react(
        &self, _trigger: &Trigger, _ctx: Ctx<'_>,
    ) -> Result<Events> {
        let mut out = Events::new();
        out.push(Echoed);
        Ok(out)
    }
}

async fn append_trigger(store: &MemoryStore) -> Uuid {
    let event_id = Uuid::new_v4();
    let payload = Trigger { id: Uuid::new_v4(), occurred_at: Utc::now() };
    let ev = EventData {
        event_id,
        causation_id: None,
        correlation_id: Uuid::new_v4(),
        event_type: format!("{}:{}", <Trigger as Event>::CATEGORY, payload.event_type()),
        payload: serde_json::to_value(&payload).unwrap(),
        created_at: Utc::now(),
        category: None,
        stream_id: None,
        metadata: serde_json::Map::new(),
        ephemeral: None,
        persistent: true,
    };
    causal::append_event(store, ev).await.unwrap();
    event_id
}

#[tokio::test]
async fn reactor_append_then_checkpoint_crash_redelivers_idempotently() {
    // New crash model (no outbox): the runner appends the output, then
    // advances the cursor. A crash AFTER append but BEFORE checkpoint.set
    // leaves the output in the log with the cursor un-advanced. On the
    // next step the trigger redelivers; the runner re-reacts and
    // re-appends — the log's deterministic-event_id dedup (C1) absorbs
    // the duplicate, so exactly ONE output exists and the cursor advances.
    let inner = Arc::new(MemoryStore::new());
    append_trigger(&inner).await;
    let injector = FaultInjector::new(inner.clone());

    let runner = ReactorRunner::new(
        EmitOne,
        "r.crash",
        injector.clone() as Arc<dyn EventLogBackend>,
        injector.clone() as Arc<dyn ReactorCheckpoint>,
    );

    // Crash at the cursor advance, AFTER the output append.
    injector.arm(FaultPoint::CheckpointSet);
    assert!(runner.step(10).await.is_err());

    // Output is in the log; cursor NOT advanced.
    let after_crash =
        EventLogBackend::read_all(inner.as_ref(), LogCursor::ZERO, 10).await.unwrap();
    let outs1 = after_crash
        .iter()
        .filter(|e| e.event_type == "echo:echoed")
        .count();
    assert_eq!(outs1, 1, "output appended before the crash");
    assert!(inner.get("r.crash").await.unwrap().is_none(), "cursor not advanced");

    // Recovery: next step re-reacts + re-appends (deduped on event_id),
    // then advances the cursor.
    runner.step(10).await.unwrap();
    let after_recovery =
        EventLogBackend::read_all(inner.as_ref(), LogCursor::ZERO, 10).await.unwrap();
    let outs2 = after_recovery
        .iter()
        .filter(|e| e.event_type == "echo:echoed")
        .count();
    assert_eq!(outs2, 1, "re-append deduped on event_id — exactly one output");
    assert!(inner.get("r.crash").await.unwrap().is_some(), "cursor advanced on recovery");
}

// (The outbox/relay crash-recovery tests were removed with slice 3 —
// reactor outputs now append directly; their crash model is covered by
// `reactor_append_then_checkpoint_crash_redelivers_idempotently` above.)
