//! The relay claims outbox rows with the SQLite queue's own atomic claim and
//! settles each row from the sink's verdict, with fakes at the sink boundary.

mod support;

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Notify;

use async_trait::async_trait;
use domain::{
    errors::DomainError,
    events::{DomainEvent, EventId},
    ports::{EventPublisher, EventRelaySink},
    value_objects::UserId,
};
use sqlite::SqliteEventQueue;
use uuid::Uuid;

#[derive(Default)]
struct FakeSink {
    fail: bool,
    /// When set, `relay` signals `started` then takes this long.
    slow: Option<Duration>,
    started: Notify,
    seen: Mutex<Vec<Option<EventId>>>,
}

#[async_trait]
impl EventRelaySink for FakeSink {
    async fn relay(&self, id: Option<&EventId>, _event: &DomainEvent) -> Result<(), DomainError> {
        self.seen.lock().unwrap().push(id.cloned());
        if let Some(delay) = self.slow {
            self.started.notify_one();
            tokio::time::sleep(delay).await;
        }
        if self.fail {
            Err(DomainError::InfrastructureError("bus down".into()))
        } else {
            Ok(())
        }
    }
}

async fn outbox() -> (sqlx::SqlitePool, Arc<dyn EventPublisher>, SqliteEventQueue) {
    let pool = support::memory_pool().await;
    sqlite::migrate(&pool).await.unwrap();
    let config = config::EventsConfig {
        queue_poll_interval_ms: 10,
        ..Default::default()
    };
    let queue = SqliteEventQueue::create(
        pool.clone(),
        sqlite::DbEventQueueConfig::from_config(&config),
    );
    (pool.clone(), Arc::new(queue.clone()), queue)
}

async fn row(pool: &sqlx::SqlitePool) -> (String, i32, Option<String>) {
    sqlx::query_as("SELECT status, attempts, event_id FROM event_queue")
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn wait_for(pool: &sqlx::SqlitePool, status: &str) {
    for _ in 0..200 {
        if row(pool).await.0 == status {
            return;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    panic!("row never reached {status}: {:?}", row(pool).await);
}

fn start(
    queue: SqliteEventQueue,
    sink: Arc<FakeSink>,
) -> (
    tokio::sync::watch::Sender<bool>,
    tokio::task::JoinHandle<()>,
) {
    let (stop, rx) = tokio::sync::watch::channel(false);
    let relay = bootstrap::outbox_relay::OutboxRelay::new(Arc::new(queue), sink);
    (stop, tokio::spawn(relay.run(rx)))
}

fn event() -> DomainEvent {
    DomainEvent::UserUpdated {
        user_id: UserId::from_uuid(Uuid::new_v4()),
    }
}

#[tokio::test]
async fn relay_marks_rows_done_only_after_publish_succeeds() {
    let (pool, publisher, queue) = outbox().await;
    publisher.publish(&event()).await.unwrap();
    let sink = Arc::new(FakeSink::default());

    let (stop, relay) = start(queue, sink.clone());
    wait_for(&pool, "done").await;
    relay.abort();
    drop(stop);

    let (_, attempts, event_id) = row(&pool).await;
    assert_eq!(attempts, 0);
    let seen = sink.seen.lock().unwrap();
    assert_eq!(seen.len(), 1);
    assert_eq!(
        seen[0].as_ref().map(|id| id.value().to_string()),
        event_id,
        "the sink gets the row's own event id"
    );
}

#[tokio::test]
async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
    let (pool, publisher, queue) = outbox().await;
    publisher.publish(&event()).await.unwrap();
    let sink = Arc::new(FakeSink {
        fail: true,
        ..Default::default()
    });

    let (stop, relay) = start(queue, sink.clone());
    for _ in 0..200 {
        if row(&pool).await.1 == 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    relay.abort();
    drop(stop);

    let (status, attempts, _) = row(&pool).await;
    assert_eq!((status.as_str(), attempts), ("pending", 1));
}

#[tokio::test]
async fn shutdown_lets_the_in_flight_row_finish_then_stops_the_relay() {
    let (pool, publisher, queue) = outbox().await;
    publisher.publish(&event()).await.unwrap();
    publisher.publish(&event()).await.unwrap();
    let sink = Arc::new(FakeSink {
        slow: Some(Duration::from_millis(200)),
        ..Default::default()
    });
    let (stop, relay) = start(queue, sink.clone());

    sink.started.notified().await;
    stop.send(true).unwrap();
    tokio::time::timeout(Duration::from_secs(5), relay)
        .await
        .expect("relay stops on shutdown")
        .unwrap();

    let statuses: Vec<String> = sqlx::query_scalar("SELECT status FROM event_queue ORDER BY id")
        .fetch_all(&pool)
        .await
        .unwrap();
    assert_eq!(statuses[0], "done", "in-flight row was completed");
    assert_ne!(statuses[0], "processing");
    assert_eq!(sink.seen.lock().unwrap().len(), 1, "no new row after stop");
}