//! AZ-676 integration tests — SubscribeVideo RPC, ai_locked atomic
//! coordination, and per-mode delivery semantics.

use std::net::TcpListener;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use tokio::time::timeout;
use tokio_stream::StreamExt;
use tonic::transport::{Channel, Endpoint};
use tonic::Request;

use shared::contracts::TelemetrySink;
use shared::models::frame::{Frame, PixelFormat as SharedPixelFormat};
use telemetry_stream::internal::video::VideoPath;
use telemetry_stream::{
    video_message, SubscribeVideoRequest, TelemetryStream, TelemetryStreamClient,
    TelemetryStreamConfig, VideoMode,
};

fn bind_ephemeral() -> (TcpListener, u16) {
    let l = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral");
    let port = l.local_addr().unwrap().port();
    (l, port)
}

async fn connect(port: u16) -> TelemetryStreamClient<Channel> {
    let url = format!("http://127.0.0.1:{port}");
    let endpoint = Endpoint::from_shared(url)
        .unwrap()
        .connect_timeout(Duration::from_secs(2));
    for _ in 0..50 {
        if let Ok(c) = TelemetryStreamClient::connect(endpoint.clone()).await {
            return c;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    panic!("gRPC client failed to connect");
}

fn make_frame(seq: u64, payload_len: usize) -> Frame {
    Frame {
        seq,
        capture_ts_monotonic_ns: seq * 1_000_000,
        decode_ts_monotonic_ns: seq * 1_000_000 + 10_000,
        pixels: Arc::new(Bytes::from(vec![(seq & 0xff) as u8; payload_len])),
        width: 1920,
        height: 1080,
        pix_fmt: SharedPixelFormat::Nv12,
        ai_locked: false,
    }
}

/// AC-1 — rtsp_forward emits exactly the configured URL in the
/// session-start message; no frames flow.
#[tokio::test]
async fn ac1_rtsp_forward_emits_url_only() {
    // Arrange
    let (listener, port) = bind_ephemeral();
    let cfg = TelemetryStreamConfig {
        video_path: VideoPath::RtspForward {
            url: "rtsp://camera.local:8554/stream0".to_string(),
        },
        ..TelemetryStreamConfig::default()
    };
    let server = TelemetryStream::with_config(cfg);
    let handle = server.handle();
    let (_join, _guard) = server.spawn_grpc_server_on(listener).unwrap();

    let mut client = connect(port).await;

    // Act
    let mut stream = client
        .subscribe_video(Request::new(SubscribeVideoRequest {
            client_id: "op_1".to_string(),
        }))
        .await
        .unwrap()
        .into_inner();

    let first = timeout(Duration::from_secs(2), stream.next())
        .await
        .expect("session-start within 2s")
        .expect("stream open")
        .expect("ok status");

    // Push a frame anyway — in rtsp_forward mode it must NOT flow.
    handle.push_frame(make_frame(1, 1024)).await.unwrap();

    let second = timeout(Duration::from_millis(500), stream.next()).await;

    // Assert
    let kind = first.kind.unwrap();
    match kind {
        video_message::Kind::Start(start) => {
            assert_eq!(start.mode, VideoMode::RtspForward as i32);
            assert_eq!(start.rtsp_url, "rtsp://camera.local:8554/stream0");
        }
        other => panic!("expected Start, got {other:?}"),
    }
    assert!(
        second.is_err(),
        "no further messages expected in rtsp_forward mode"
    );
    assert_eq!(handle.video_snapshot().mode, "rtsp_forward");
}

/// AC-2 — bytes_inline forwards encoded frames to subscribed clients.
#[tokio::test]
async fn ac2_bytes_inline_forwards_frames() {
    // Arrange
    let (listener, port) = bind_ephemeral();
    let cfg = TelemetryStreamConfig {
        video_path: VideoPath::BytesInline,
        // Generous capacity so the test client keeps up without lag.
        video_capacity: 256,
        ..TelemetryStreamConfig::default()
    };
    let server = TelemetryStream::with_config(cfg);
    let handle = server.handle();
    let (_join, _guard) = server.spawn_grpc_server_on(listener).unwrap();

    let mut client = connect(port).await;
    let mut stream = client
        .subscribe_video(Request::new(SubscribeVideoRequest {
            client_id: "op_inline".to_string(),
        }))
        .await
        .unwrap()
        .into_inner();

    // Drain the session-start first.
    let start = timeout(Duration::from_secs(2), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert!(matches!(start.kind.unwrap(), video_message::Kind::Start(_)));

    // Wait until the server has registered the session before
    // publishing so no frames are emitted before the broadcast has a
    // receiver.
    for _ in 0..100 {
        if handle.video_snapshot().video_session_count == 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert_eq!(handle.video_snapshot().video_session_count, 1);

    // Act — publish 100 frames; verify the client gets each one in
    // monotonically increasing sequence.
    let total: u64 = 100;
    for seq in 0..total {
        // Tiny pixel payload so the test isn't expensive.
        handle.push_frame(make_frame(seq, 64)).await.unwrap();
    }

    let mut received = 0u64;
    let mut last_seq: Option<u64> = None;
    while received < total {
        let msg = timeout(Duration::from_secs(2), stream.next())
            .await
            .expect("ac2 stalled — frame not received in 2s")
            .expect("stream open")
            .expect("ok status");
        match msg.kind.unwrap() {
            video_message::Kind::Frame(f) => {
                if let Some(prev) = last_seq {
                    assert!(f.seq > prev, "monotonic seq violated: {prev} → {}", f.seq);
                }
                last_seq = Some(f.seq);
                received += 1;
            }
            video_message::Kind::Start(_) => panic!("unexpected second Start"),
        }
    }

    // Assert
    assert_eq!(received, total);
    let snap = handle.video_snapshot();
    assert_eq!(snap.published_frames, total);
    assert_eq!(snap.bytes_inline_drops_total, 0);
}

/// AC-3 — ai_locked flips true on first subscriber, false when the
/// last subscriber disconnects.
#[tokio::test]
async fn ac3_ai_locked_toggles_on_session_start_and_stop() {
    // Arrange
    let (listener, port) = bind_ephemeral();
    let cfg = TelemetryStreamConfig {
        video_path: VideoPath::BytesInline,
        ..TelemetryStreamConfig::default()
    };
    let server = TelemetryStream::with_config(cfg);
    let handle = server.handle();
    let ai_locked = server.ai_locked_handle();
    let (_join, _guard) = server.spawn_grpc_server_on(listener).unwrap();

    // No clients yet → false.
    assert!(!ai_locked.load(Ordering::Acquire));

    // Act 1 — first subscriber connects; flag must flip to true.
    let mut c1 = connect(port).await;
    let mut s1 = c1
        .subscribe_video(Request::new(SubscribeVideoRequest {
            client_id: "op_a".to_string(),
        }))
        .await
        .unwrap()
        .into_inner();
    let _start = timeout(Duration::from_secs(2), s1.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    for _ in 0..100 {
        if ai_locked.load(Ordering::Acquire) {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert!(
        ai_locked.load(Ordering::Acquire),
        "ai_locked MUST be true once first session is active"
    );

    // Act 2 — second subscriber connects; flag stays true.
    let mut c2 = connect(port).await;
    let mut s2 = c2
        .subscribe_video(Request::new(SubscribeVideoRequest {
            client_id: "op_b".to_string(),
        }))
        .await
        .unwrap()
        .into_inner();
    let _start = timeout(Duration::from_secs(2), s2.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    for _ in 0..100 {
        if handle.video_snapshot().video_session_count == 2 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert!(ai_locked.load(Ordering::Acquire));
    assert_eq!(handle.video_snapshot().video_session_count, 2);

    // Act 3 — drop second client; one session left, still locked.
    drop(s2);
    drop(c2);
    for _ in 0..100 {
        if handle.video_snapshot().video_session_count == 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert_eq!(handle.video_snapshot().video_session_count, 1);
    assert!(ai_locked.load(Ordering::Acquire));

    // Act 4 — drop last client; ai_locked flips to false.
    drop(s1);
    drop(c1);
    for _ in 0..100 {
        if !ai_locked.load(Ordering::Acquire) {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(
        !ai_locked.load(Ordering::Acquire),
        "ai_locked MUST be false after last session leaves"
    );
    assert_eq!(handle.video_snapshot().video_session_count, 0);
}

/// Empty client_id is rejected at the boundary (parity with Subscribe).
#[tokio::test]
async fn empty_client_id_rejected() {
    // Arrange
    let (listener, port) = bind_ephemeral();
    let cfg = TelemetryStreamConfig {
        video_path: VideoPath::BytesInline,
        ..TelemetryStreamConfig::default()
    };
    let server = TelemetryStream::with_config(cfg);
    let _h = server.handle();
    let (_join, _guard) = server.spawn_grpc_server_on(listener).unwrap();

    let mut client = connect(port).await;

    // Act
    let err = client
        .subscribe_video(Request::new(SubscribeVideoRequest {
            client_id: String::new(),
        }))
        .await
        .expect_err("empty client_id must error");

    // Assert
    assert_eq!(err.code(), tonic::Code::InvalidArgument);
}