mod common;

use std::sync::Arc;

use eventcore_types::{Event, EventStore, EventStoreError, StreamId, StreamVersion, StreamWrites};
use serde::{Deserialize, Serialize};
use tokio::sync::Barrier;
use uuid::Uuid;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct TestEvent {
    stream_id: StreamId,
    payload: String,
}

impl Event for TestEvent {
    fn stream_id(&self) -> &StreamId {
        &self.stream_id
    }

    fn event_type_name() -> &'static str {
        "TestEvent"
    }
}

fn unique_stream_id(prefix: &str) -> StreamId {
    StreamId::try_new(format!("{}-{}", prefix, Uuid::now_v7())).expect("valid stream id")
}

fn build_single_stream_writes(
    stream_id: &StreamId,
    expected_version: StreamVersion,
    payload: &str,
) -> StreamWrites {
    StreamWrites::new()
        .register_stream(stream_id.clone(), expected_version)
        .and_then(|writes| {
            writes.append(TestEvent {
                stream_id: stream_id.clone(),
                payload: payload.to_string(),
            })
        })
        .expect("single stream writes should register stream and append event")
}

#[tokio::test]
#[tracing_test::traced_test]
async fn developer_retries_after_postgres_version_conflict() {
    // Given: A migrated Postgres store that enforces optimistic concurrency
    let store = common::create_test_store().await;

    // Unique stream ID per test run for parallel execution
    let stream_id = unique_stream_id("account/concurrency");

    // And: Two separate writes targeting the same stream at version 0
    let first_writes =
        build_single_stream_writes(&stream_id, StreamVersion::new(0), "first deposit");
    let second_writes =
        build_single_stream_writes(&stream_id, StreamVersion::new(0), "second deposit");

    // When: Both writes attempt to append concurrently
    let barrier = Arc::new(Barrier::new(3));
    let first_store = store.clone();
    let first_barrier = barrier.clone();
    let first_handle = tokio::spawn(async move {
        let _ = first_barrier.wait().await;
        first_store.append_events(first_writes).await
    });

    let second_store = store.clone();
    let second_barrier = barrier.clone();
    let second_handle = tokio::spawn(async move {
        let _ = second_barrier.wait().await;
        second_store.append_events(second_writes).await
    });

    let _ = barrier.wait().await;

    let first_result = first_handle
        .await
        .expect("first append task should join without panic");
    let second_result = second_handle
        .await
        .expect("second append task should join without panic");

    let success_count = [&first_result, &second_result]
        .into_iter()
        .filter(|result| result.is_ok())
        .count();
    let conflict_count = [&first_result, &second_result]
        .into_iter()
        .filter(|result| matches!(result, Err(EventStoreError::VersionConflict { .. })))
        .count();

    // And: Developer retries with expected version 1 after reloading state
    let retry_writes = build_single_stream_writes(
        &stream_id,
        StreamVersion::new(1),
        "retry deposit after conflict",
    );
    let retry_result = store.append_events(retry_writes).await;

    // Then: Exactly one write succeeds initially, the other reports conflict, retry succeeds, two events persist, and instrumentation logs conflict
    let read_result = store.read_stream::<TestEvent>(stream_id.clone()).await;
    let total_events = read_result
        .as_ref()
        .map(|reader| reader.len())
        .unwrap_or_default();
    let logs_contain_conflict = logs_contain("postgres.version_conflict");
    let retry_succeeded = retry_result.is_ok();
    let retry_state = match &retry_result {
        Ok(_) => "ok".to_string(),
        Err(err) => err.to_string(),
    };

    assert!(
        success_count == 1
            && conflict_count == 1
            && retry_succeeded
            && read_result.is_ok()
            && total_events == 2
            && logs_contain_conflict,
        "postgres store should surface version conflict, allow retry, persist events, and emit instrumentation; success_count={success_count}, conflict_count={conflict_count}, retry_state={retry_state}, total_events={total_events}, logs_contain_conflict={logs_contain_conflict}",
    );
}
