//! Integration tests for the subscription stage using snapshot tests with paused tokio clock.
//!
//! Fixtures are pre-encoded WebSocket frames in a simple length-prefixed binary format:
//! Each frame is stored as: [4-byte big-endian length][frame bytes]
//! Multiple frames are concatenated in a single file.
//! This allows the fake to stream frames sequentially without a generator at test runtime.

mod common;

use atproto_devtool::commands::test::labeler::pipeline::{
    HttpTee, LabelerOptions, parse_target, run_pipeline,
};
use atproto_devtool::commands::test::labeler::subscription::{FrameHeader, SubscribeLabelsPayload};
use atproto_devtool::common::identity::{DnsResolver, HttpClient, IdentityError};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use url::Url;

/// Type alias for the response map in FakeHttpClient.
type FakeHttpResponses = Arc<Mutex<HashMap<String, (u16, Vec<u8>)>>>;

/// Fake HTTP client for testing.
struct FakeHttpClient {
    responses: FakeHttpResponses,
}

impl FakeHttpClient {
    fn new() -> Self {
        Self {
            responses: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    #[expect(
        dead_code,
        reason = "parallel helper to FakeRawHttpTee; currently unused because subscription tests use endpoint-only targets, kept for future crypto-stage wiring"
    )]
    fn add_response(&self, url: impl Into<String>, status: u16, body: Vec<u8>) {
        self.responses
            .lock()
            .unwrap()
            .insert(url.into(), (status, body));
    }
}

#[async_trait::async_trait]
impl HttpClient for FakeHttpClient {
    async fn get_bytes(&self, url: &Url) -> Result<(u16, Vec<u8>), IdentityError> {
        let url_str = url.as_str();
        self.responses
            .lock()
            .unwrap()
            .get(url_str)
            .cloned()
            .ok_or_else(|| IdentityError::DidResolutionFailed {
                status: 404,
                body: "Not found".to_string(),
            })
    }
}

/// Type alias for the DNS records map.
type FakeDnsRecords = Arc<Mutex<HashMap<String, Vec<String>>>>;

/// Fake DNS resolver for testing.
struct FakeDnsResolver {
    records: FakeDnsRecords,
}

impl FakeDnsResolver {
    fn new() -> Self {
        Self {
            records: Arc::new(Mutex::new(HashMap::new())),
        }
    }
}

#[async_trait::async_trait]
impl DnsResolver for FakeDnsResolver {
    async fn txt_lookup(&self, name: &str) -> Result<Vec<String>, IdentityError> {
        self.records
            .lock()
            .unwrap()
            .get(name)
            .cloned()
            .ok_or_else(|| IdentityError::DnsLookupFailed {
                source: Box::new(IdentityError::InvalidHandle),
            })
    }
}

/// Helper to render a report to a string.
fn render_report_to_string(
    report: &atproto_devtool::commands::test::labeler::report::LabelerReport,
) -> String {
    let mut buf = Vec::new();
    report
        .render(
            &mut buf,
            &atproto_devtool::commands::test::labeler::report::RenderConfig { no_color: true },
        )
        .expect("render failed");
    String::from_utf8(buf).expect("invalid utf-8")
}

/// Helper to build a FakeRawHttpTee preloaded with an empty single-page response.
///
/// Subscription tests don't exercise the HTTP stage, but the pipeline runs it
/// before subscription, so we supply a trivially-passing fake to avoid real
/// network calls against `https://example.com`.
fn make_passing_http_tee() -> common::FakeRawHttpTee {
    let tee = common::FakeRawHttpTee::new();
    tee.add_response(None, 200, br#"{"cursor":null,"labels":[]}"#.to_vec());
    tee
}

/// Helper to normalize elapsed time in snapshots.
///
/// Replaces `elapsed: <N>ms` with `elapsed: XXms`, advancing past each match
/// to avoid re-matching the replacement on the next iteration.
fn normalize_timing(rendered: String) -> String {
    let mut result = String::with_capacity(rendered.len());
    let mut rest = rendered.as_str();
    while let Some(pos) = rest.find("elapsed: ") {
        let after = pos + "elapsed: ".len();
        result.push_str(&rest[..after]);
        let tail = &rest[after..];
        if let Some(end) = tail.find("ms") {
            result.push_str("XXms");
            rest = &tail[end + 2..];
        } else {
            result.push_str(tail);
            return result;
        }
    }
    result.push_str(rest);
    result
}

/// Encodes a frame with length prefix: [4-byte big-endian length][frame bytes].
fn encode_frame_with_length(header: &FrameHeader, payload_bytes: &[u8]) -> Vec<u8> {
    let mut header_bytes = Vec::new();
    ciborium::ser::into_writer(header, &mut header_bytes).expect("encode header");

    let mut frame = header_bytes;
    frame.extend(payload_bytes);

    let len = frame.len() as u32;
    let mut result = len.to_be_bytes().to_vec();
    result.extend(frame);
    result
}

/// Generator test: generates all subscription fixture files.
/// Run with: cargo test --test labeler_subscription -- --ignored gen_fixtures
///
/// Fixture format: Each frame file contains WebSocket frames in a length-prefixed binary format:
/// [4-byte big-endian length][frame bytes]...
/// This allows load_frames_from_fixture to stream frames sequentially without decoding overhead.
#[test]
#[ignore]
fn gen_fixtures() {
    let base = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures/labeler/subscription");

    // Generate backfill_complete/frames.bin: 3 valid frames.
    let mut frames = Vec::new();
    for i in 0..3 {
        let header = FrameHeader {
            op: 1,
            t: Some("#labels".to_string()),
        };
        let payload = SubscribeLabelsPayload {
            seq: i,
            labels: vec![],
        };
        let mut payload_bytes = Vec::new();
        ciborium::ser::into_writer(&payload, &mut payload_bytes).expect("encode payload");
        frames.extend(encode_frame_with_length(&header, &payload_bytes));
    }
    std::fs::write(base.join("backfill_complete/frames.bin"), frames)
        .expect("write backfill_complete");

    // Generate backfill_exceeds_budget/frames.bin: 20 frames.
    let mut frames = Vec::new();
    for i in 0..20 {
        let header = FrameHeader {
            op: 1,
            t: Some("#labels".to_string()),
        };
        let payload = SubscribeLabelsPayload {
            seq: i,
            labels: vec![],
        };
        let mut payload_bytes = Vec::new();
        ciborium::ser::into_writer(&payload, &mut payload_bytes).expect("encode payload");
        frames.extend(encode_frame_with_length(&header, &payload_bytes));
    }
    std::fs::write(base.join("backfill_exceeds_budget/frames.bin"), frames)
        .expect("write backfill_exceeds_budget");

    // Generate empty_stream/.gitkeep (intentionally empty file to hold directory in git).
    std::fs::write(base.join("empty_stream/.gitkeep"), "").expect("write empty_stream");

    // Generate malformed_frame/frames.bin: 2 valid + 1 malformed frame.
    let mut frames = Vec::new();
    for i in 0..2 {
        let header = FrameHeader {
            op: 1,
            t: Some("#labels".to_string()),
        };
        let payload = SubscribeLabelsPayload {
            seq: i,
            labels: vec![],
        };
        let mut payload_bytes = Vec::new();
        ciborium::ser::into_writer(&payload, &mut payload_bytes).expect("encode payload");
        frames.extend(encode_frame_with_length(&header, &payload_bytes));
    }
    // Add a malformed frame: [4-byte BE length=3][0xFF, 0xFF, 0xFF] (invalid CBOR).
    frames.extend((3u32).to_be_bytes());
    frames.extend(&[0xFF, 0xFF, 0xFF]);
    std::fs::write(base.join("malformed_frame/frames.bin"), frames).expect("write malformed_frame");

    // Generate error_frame_malformed/frames.bin: error frame with invalid payload.
    let mut frames = Vec::new();
    let header = FrameHeader { op: -1, t: None };
    let mut frame_bytes = Vec::new();
    ciborium::ser::into_writer(&header, &mut frame_bytes).expect("encode header");
    frame_bytes.push(0xff); // Invalid CBOR for payload.

    let len = frame_bytes.len() as u32;
    let mut result = len.to_be_bytes().to_vec();
    result.extend(frame_bytes);
    frames.extend(result);
    std::fs::write(base.join("error_frame_malformed/frames.bin"), frames)
        .expect("write error_frame_malformed");

    println!("Generated all fixtures successfully");
}

/// Load and parse frames from a fixture file (length-prefixed format).
fn load_frames_from_fixture(path: &str) -> Vec<Vec<u8>> {
    let data = std::fs::read(path).unwrap_or_default();
    let mut frames = Vec::new();
    let mut pos = 0;

    while pos + 4 <= data.len() {
        let len_bytes = &data[pos..pos + 4];
        let len =
            u32::from_be_bytes([len_bytes[0], len_bytes[1], len_bytes[2], len_bytes[3]]) as usize;
        pos += 4;

        if pos + len > data.len() {
            break;
        }

        frames.push(data[pos..pos + len].to_vec());
        pos += len;
    }

    frames
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn backfill_completes_within_budget_passes() {
    let frames = load_frames_from_fixture(
        "tests/fixtures/labeler/subscription/backfill_complete/frames.bin",
    );
    let fake_ws = common::FakeWebSocketClient::new();

    // Add first connection (backfill): 3 frames with 10ms delay, then 600ms idle gap.
    fake_ws.add_script(common::FakeScript {
        frames: frames.clone(),
        inter_frame_delay: Duration::from_millis(10),
        final_wait: Some(Duration::from_millis(600)),
        transport_error: false,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn backfill_exceeds_budget_triggers_live_tail() {
    let backfill_frames = load_frames_from_fixture(
        "tests/fixtures/labeler/subscription/backfill_exceeds_budget/frames.bin",
    );
    let fake_ws = common::FakeWebSocketClient::new();

    // First connection (backfill): 20 frames @ 150ms delay = 3s total (exceeds 2s budget).
    fake_ws.add_script(common::FakeScript {
        frames: backfill_frames,
        inter_frame_delay: Duration::from_millis(150),
        final_wait: None,
        transport_error: false,
        mid_stream_error: false,
    });

    // Second connection (live-tail): 1 frame then 1s idle.
    let live_frames = load_frames_from_fixture(
        "tests/fixtures/labeler/subscription/backfill_complete/frames.bin",
    );
    fake_ws.add_script(common::FakeScript {
        frames: vec![live_frames[0].clone()],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: Some(Duration::from_secs(1)),
        transport_error: false,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn empty_stream_advisories() {
    let fake_ws = common::FakeWebSocketClient::new();

    // Single connection (backfill): empty stream returns no frames immediately,
    // so backfill outcome is NoFramesWithinBudget and live-tail is skipped.
    fake_ws.add_script(common::FakeScript {
        frames: vec![],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: None,
        transport_error: false,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = render_report_to_string(&report);

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn malformed_frame_emits_spec_violation() {
    let frames =
        load_frames_from_fixture("tests/fixtures/labeler/subscription/malformed_frame/frames.bin");
    let fake_ws = common::FakeWebSocketClient::new();

    // First connection (backfill): frames then stream close.
    fake_ws.add_script(common::FakeScript {
        frames,
        inter_frame_delay: Duration::from_millis(10),
        final_wait: None,
        transport_error: false,
        mid_stream_error: false,
    });

    // Second connection (live-tail after stream close): empty stream with long final_wait
    // to exhaust budget and report deterministic CleanHold outcome from explicit script.
    fake_ws.add_script(common::FakeScript {
        frames: vec![],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: Some(Duration::from_secs(2)),
        transport_error: false,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn error_frame_malformed_payload_spec_violation() {
    let frames = load_frames_from_fixture(
        "tests/fixtures/labeler/subscription/error_frame_malformed/frames.bin",
    );
    let fake_ws = common::FakeWebSocketClient::new();

    // First connection (backfill): frames then stream close.
    fake_ws.add_script(common::FakeScript {
        frames,
        inter_frame_delay: Duration::from_millis(0),
        final_wait: None,
        transport_error: false,
        mid_stream_error: false,
    });

    // Second connection (live-tail after stream close): empty stream with long final_wait
    // to exhaust budget and report deterministic CleanHold outcome from explicit script.
    fake_ws.add_script(common::FakeScript {
        frames: vec![],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: Some(Duration::from_secs(2)),
        transport_error: false,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn unreachable_endpoint_network_error() {
    let fake_ws = common::FakeWebSocketClient::new();

    // First connection returns transport error.
    fake_ws.add_script(common::FakeScript {
        frames: vec![],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: None,
        transport_error: true,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = render_report_to_string(&report);

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn live_tail_connect_failure_emits_network_error() {
    let backfill_frames = load_frames_from_fixture(
        "tests/fixtures/labeler/subscription/backfill_exceeds_budget/frames.bin",
    );
    let fake_ws = common::FakeWebSocketClient::new();

    // First connection (backfill): 20 frames @ 150ms delay = 3s total (exceeds 2s budget).
    fake_ws.add_script(common::FakeScript {
        frames: backfill_frames,
        inter_frame_delay: Duration::from_millis(150),
        final_wait: None,
        transport_error: false,
        mid_stream_error: false,
    });

    // Second connection (live-tail): returns transport error.
    fake_ws.add_script(common::FakeScript {
        frames: vec![],
        inter_frame_delay: Duration::from_millis(0),
        final_wait: None,
        transport_error: true,
        mid_stream_error: false,
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn mid_stream_transport_error_does_not_reset_idle_gap() {
    // Encode 2 frames to send before transport error.
    let mut frames = Vec::new();
    for i in 0..2 {
        let header = FrameHeader {
            op: 1,
            t: Some("#labels".to_string()),
        };
        let payload = SubscribeLabelsPayload {
            seq: i,
            labels: vec![],
        };
        let mut header_bytes = Vec::new();
        ciborium::ser::into_writer(&header, &mut header_bytes).expect("encode header");
        let mut payload_bytes = Vec::new();
        ciborium::ser::into_writer(&payload, &mut payload_bytes).expect("encode payload");

        let mut frame = header_bytes;
        frame.extend(payload_bytes);
        frames.push(frame);
    }

    let fake_ws = common::FakeWebSocketClient::new();

    // Single connection (backfill): 2 frames (10ms apart), then transport error, then final_wait
    // to exhaust idle-gap timer from last frame. The idle-gap should fire from the second frame,
    // not be reset by the transport error. This tests subscription.rs:417-419.
    fake_ws.add_script(common::FakeScript {
        frames,
        inter_frame_delay: Duration::from_millis(10),
        final_wait: Some(Duration::from_millis(700)), // 700ms > 500ms idle-gap.
        transport_error: false,
        mid_stream_error: true, // After all frames, yield one Transport error, then close.
    });

    let http = FakeHttpClient::new();
    let dns = FakeDnsResolver::new();
    let fake_tee = make_passing_http_tee();
    let target = parse_target("https://example.com/labeler", None).expect("parse failed");

    let opts = LabelerOptions {
        http: &http,
        dns: &dns,
        http_tee: HttpTee::Test(&fake_tee),
        ws_client: Some(&fake_ws),
        subscribe_timeout: Duration::from_secs(2),
        verbose: false,
    };

    let report = run_pipeline(target, opts).await;
    let rendered = normalize_timing(render_report_to_string(&report));

    insta::assert_snapshot!(rendered);
}

#[test]
fn subscribe_timeout_below_floor_rejected() {
    use atproto_devtool::commands::test::labeler::parse_subscribe_timeout;

    // Test 500ms is rejected.
    let result = parse_subscribe_timeout("500ms");
    assert!(result.is_err());
    let err = result.unwrap_err();
    assert!(err.contains("at least 1 second"), "error message: {err}");

    // Test 0 is rejected.
    let result = parse_subscribe_timeout("0");
    assert!(result.is_err());
}
