//! The SSE subscribe stream and its reconnect state machine.

use crate::common::*;

// ── SSE subscribe / reconnect state machine ──
//
// These drive the real `subscribe()` -> `SyncNotifyStream::next_change()` path
// against wiremock so the reconnect state machine (subscribe.rs) is exercised
// end to end: a live notification, a transparent reconnect across a stream
// drop, the fatal auth-rejection exit, and the give-up-after-N-failures cap.

const SUBSCRIBE_PATH: &str = "/api/v1/sync/subscribe";

/// One complete SSE "changed" block, body-terminated so the client sees a full
/// event and then end-of-stream.
fn sse_changed_block() -> &'static str {
    "event: changed\n\n"
}

#[tokio::test]
async fn subscribe_yields_changed_event() {
    let kit = MockKit::start().await;
    kit.get(SUBSCRIBE_PATH).text(sse_changed_block()).await;

    let client = kit.authed();
    let mut stream = client.subscribe().await.expect("subscribe should succeed");

    // The first block is delivered inside the initial response body, so this
    // returns without any reconnect.
    assert_eq!(stream.next_change().await, Some(()));
}

#[tokio::test]
async fn subscribe_reconnects_transparently_after_stream_drop() {
    let kit = MockKit::start().await;
    // Every connection serves one block then ends (Content-Length terminates the
    // body). Consuming the first event, then reading past it, drops the stream
    // and forces a reconnect that must transparently yield the next event.
    kit.get(SUBSCRIBE_PATH).text(sse_changed_block()).await;

    let client = kit.authed();
    let mut stream = client.subscribe().await.expect("subscribe should succeed");

    // #1 comes from the initial connection; #2 can only arrive after the stream
    // drops (body EOF) and reconnect() re-establishes it.
    assert_eq!(stream.next_change().await, Some(()));
    assert_eq!(stream.next_change().await, Some(()));

    // subscribe() opened one connection; the second event required at least one
    // reconnect, so the server saw two or more subscribe requests.
    let hits = kit.hits(SUBSCRIBE_PATH).await;
    assert!(
        hits >= 2,
        "expected a reconnect (>=2 subscribe requests), got {hits}"
    );
}

#[tokio::test]
async fn subscribe_stream_closes_on_auth_rejection() {
    let kit = MockKit::start().await;
    // First connection opens cleanly but carries no event and ends immediately,
    // forcing a reconnect. The reconnect is rejected for auth -> fatal, so the
    // stream ends with `None` rather than retrying.
    kit.get(SUBSCRIBE_PATH).once().text("").await;
    kit.get(SUBSCRIBE_PATH).code(401).text("unauthorized").await;

    let client = kit.authed();
    let mut stream = client.subscribe().await.expect("initial subscribe is 200");

    // Empty body -> EOF -> reconnect -> 401 -> fatal -> None.
    assert_eq!(stream.next_change().await, None);
}

// Note: the "give up after MAX_RECONNECT_ATTEMPTS consecutive failures" path is
// deliberately not covered end-to-end here. Exercising it against a live mock
// server would incur the real exponential backoff (growing to a 60s cap, minutes
// of wall-clock), and `tokio::time::pause()` cannot collapse it: the wiremock
// server runs on its own runtime, so a paused clock auto-advances past the
// cross-thread request to the nearest timer and the reconnect never completes.
// The two pieces of that path are covered separately: the backoff schedule by
// `reconnect_delay_grows_then_caps` (subscribe.rs), and the fatal-exit mechanics
// (returning `None` and clearing state) by the auth-rejection test above.