//! RFC-019 Stage C — `PostgresBackend::subscribe_signal_delivery`
//! integration test. Gated on `FF_PG_TEST_URL`.
//!
//! Mirrors `subscribe_lease_history.rs`: synthetic direct-INSERT into
//! `ff_signal_event` to exercise the LISTEN + catch-up paths without
//! dragging the full HMAC + suspension state machine into scope.

use ff_backend_postgres::{apply_migrations, PostgresBackend};
use ff_core::backend::ScannerFilter;
use ff_core::engine_backend::EngineBackend;
use ff_core::partition::PartitionConfig;
use ff_core::stream_events::SignalDeliveryEffect;
use ff_core::stream_subscribe::{decode_postgres_event_cursor, StreamCursor};
use futures::StreamExt;
use sqlx::postgres::PgPoolOptions;
use std::time::Duration;

async fn setup_or_skip() -> Option<sqlx::PgPool> {
    let url = std::env::var("FF_PG_TEST_URL").ok()?;
    let bootstrap = PgPoolOptions::new()
        .max_connections(8)
        .connect(&url)
        .await
        .expect("connect to FF_PG_TEST_URL");
    apply_migrations(&bootstrap)
        .await
        .expect("apply_migrations clean");
    Some(bootstrap)
}

#[tokio::test]
#[ignore = "requires a live Postgres; set FF_PG_TEST_URL"]
async fn subscribe_signal_delivery_yields_typed_event() {
    let Some(pool) = setup_or_skip().await else {
        eprintln!("FF_PG_TEST_URL not set — skipping");
        return;
    };

    let backend = PostgresBackend::from_pool(pool.clone(), PartitionConfig::default());

    let mut stream = backend
        .subscribe_signal_delivery(StreamCursor::empty(), &ScannerFilter::default())
        .await
        .expect("subscribe");

    tokio::time::sleep(Duration::from_millis(150)).await;

    let exec_uuid = uuid::Uuid::new_v4();
    let sid = uuid::Uuid::new_v4();
    let wp = uuid::Uuid::new_v4();
    sqlx::query(
        "INSERT INTO ff_signal_event \
         (execution_id, signal_id, waitpoint_id, source_identity, delivered_at_ms, partition_key) \
         VALUES ($1, $2, $3, 'unit-test', $4, 0)",
    )
    .bind(exec_uuid.to_string())
    .bind(sid.to_string())
    .bind(wp.to_string())
    .bind(1_700_000_000_000_i64)
    .execute(&pool)
    .await
    .expect("insert synthetic event");

    let event = tokio::time::timeout(Duration::from_secs(5), stream.next())
        .await
        .expect("timeout waiting for event")
        .expect("stream ended")
        .expect("stream error");

    assert_eq!(event.execution_id.to_string(), format!("{{fp:0}}:{exec_uuid}"));
    assert_eq!(event.signal_id.0, sid);
    assert_eq!(event.waitpoint_id.as_ref().map(|w| w.0), Some(wp));
    assert_eq!(event.source_identity.as_deref(), Some("unit-test"));
    assert_eq!(event.effect, SignalDeliveryEffect::Satisfied);
    assert_eq!(event.at.0, 1_700_000_000_000);

    let decoded = decode_postgres_event_cursor(&event.cursor)
        .expect("cursor decode")
        .expect("some event_id");
    assert!(decoded > 0, "event_id should be positive, got {decoded}");
}

#[tokio::test]
#[ignore = "requires a live Postgres; set FF_PG_TEST_URL"]
async fn subscribe_signal_delivery_cursor_resume() {
    let Some(pool) = setup_or_skip().await else {
        return;
    };
    let backend = PostgresBackend::from_pool(pool.clone(), PartitionConfig::default());

    let tail: i64 = sqlx::query_scalar(
        "SELECT COALESCE(MAX(event_id), 0) FROM ff_signal_event WHERE partition_key = 0",
    )
    .fetch_one(&pool)
    .await
    .unwrap();

    let exec_uuid = uuid::Uuid::new_v4();
    let sid_a = uuid::Uuid::new_v4();
    sqlx::query(
        "INSERT INTO ff_signal_event \
         (execution_id, signal_id, waitpoint_id, source_identity, delivered_at_ms, partition_key) \
         VALUES ($1, $2, NULL, NULL, $3, 0)",
    )
    .bind(exec_uuid.to_string())
    .bind(sid_a.to_string())
    .bind(1_700_000_000_001_i64)
    .execute(&pool)
    .await
    .unwrap();

    let sid_b = uuid::Uuid::new_v4();
    sqlx::query(
        "INSERT INTO ff_signal_event \
         (execution_id, signal_id, waitpoint_id, source_identity, delivered_at_ms, partition_key) \
         VALUES ($1, $2, NULL, NULL, $3, 0)",
    )
    .bind(exec_uuid.to_string())
    .bind(sid_b.to_string())
    .bind(1_700_000_000_002_i64)
    .execute(&pool)
    .await
    .unwrap();

    let start_cursor = ff_core::stream_subscribe::encode_postgres_event_cursor(tail);
    let mut stream = backend
        .subscribe_signal_delivery(start_cursor, &ScannerFilter::default())
        .await
        .expect("subscribe resume");

    let first = tokio::time::timeout(Duration::from_secs(5), stream.next())
        .await
        .expect("timeout #1")
        .expect("stream ended")
        .expect("err");
    assert_eq!(first.signal_id.0, sid_a);

    let resume_cursor = first.cursor.clone();
    drop(stream);

    let mut stream2 = backend
        .subscribe_signal_delivery(resume_cursor.clone(), &ScannerFilter::default())
        .await
        .expect("subscribe resume after drop");
    let second = tokio::time::timeout(Duration::from_secs(5), stream2.next())
        .await
        .expect("timeout #2")
        .expect("stream ended")
        .expect("err");

    assert_ne!(second.cursor, resume_cursor, "cursor must advance");
    assert_eq!(second.signal_id.0, sid_b);
}
