//! Common test utilities shared across integration tests.
//!
//! This module uses the `tests/common/mod.rs` idiom because cargo treats each
//! `tests/*.rs` as a separate crate and requires this pattern to share code.
//! Some helpers may be unused in any given test binary, so we allow dead code at the module level.
//! The `#[expect]` attribute cannot be used here because different test binaries use different
//! subsets of helpers, causing the expect to fail in binaries where a helper happens to be unused.
#![allow(dead_code)]

use async_trait::async_trait;
use atproto_devtool::commands::test::labeler::http::{HttpStageError, RawHttpTee, RawXrpcResponse};
use atproto_devtool::commands::test::labeler::subscription::{
    FrameStream, SubscriptionStageError, WebSocketClient,
};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use url::Url;

/// Type alias for HTTP response map in tests.
pub type FakeHttpResponses = Arc<Mutex<HashMap<Option<String>, (reqwest::StatusCode, Vec<u8>)>>>;

/// Fake HTTP tee for testing, returns pre-defined responses.
pub struct FakeRawHttpTee {
    /// Map of cursor -> response bytes.
    responses: FakeHttpResponses,
    /// Whether to return a transport error.
    transport_error: Arc<Mutex<bool>>,
}

impl FakeRawHttpTee {
    /// Create a new FakeRawHttpTee.
    pub fn new() -> Self {
        Self {
            responses: Arc::new(Mutex::new(HashMap::new())),
            transport_error: Arc::new(Mutex::new(false)),
        }
    }

    /// Add a response for a given cursor.
    pub fn add_response(&self, cursor: Option<&str>, status: u16, body: Vec<u8>) {
        self.responses.lock().unwrap().insert(
            cursor.map(|s| s.to_string()),
            (reqwest::StatusCode::from_u16(status).unwrap(), body),
        );
    }

    /// Set the transport error flag to simulate network failures.
    pub fn set_transport_error(&self) {
        *self.transport_error.lock().unwrap() = true;
    }
}

impl Default for FakeRawHttpTee {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl RawHttpTee for FakeRawHttpTee {
    async fn query_labels(&self, cursor: Option<&str>) -> Result<RawXrpcResponse, HttpStageError> {
        if *self.transport_error.lock().unwrap() {
            // Return a realistic transport error simulating TCP connection refused.
            return Err(HttpStageError::Transport {
                message: "tcp connect: connection refused".into(),
                source: None,
            });
        }

        let cursor_key = cursor.map(|s| s.to_string());
        let responses = self.responses.lock().unwrap();

        match responses.get(&cursor_key) {
            Some((status, body)) => {
                let raw_body: Arc<[u8]> = Arc::from(body.as_slice());
                let decoded = serde_json::from_slice::<
                    atrium_api::com::atproto::label::query_labels::Output,
                >(body)
                .map_err(|source| HttpStageError::DecodeFailed {
                    raw_body: raw_body.clone(),
                    source,
                    source_url: "https://example.com/xrpc/com.atproto.label.queryLabels"
                        .to_string(),
                })?;
                Ok(RawXrpcResponse {
                    status: *status,
                    raw_body,
                    decoded,
                    source_url: "https://example.com/xrpc/com.atproto.label.queryLabels"
                        .to_string(),
                })
            }
            None => {
                // Return a realistic transport error for unconfigured cursor.
                Err(HttpStageError::Transport {
                    message: "tcp connect: connection refused".into(),
                    source: None,
                })
            }
        }
    }
}

/// Fake WebSocket client for testing subscription stage with scripted responses.
///
/// Two construction styles are supported:
///
/// * `new()` + `add_script()` — the subscription-test style. Each `connect()`
///   consumes exactly one script. Calling `connect()` when no script is queued
///   panics, so subscription tests must explicitly declare every connection
///   the stage is expected to make.
/// * `empty()` — the identity-test style. Every `connect()` returns an
///   empty-stream placeholder that closes immediately. Identity tests construct
///   one of these to satisfy the pipeline's subscription stage without asserting
///   on its output.
pub struct FakeWebSocketClient {
    /// Scripts queued for each connection.
    scripts: Arc<Mutex<Vec<FakeScript>>>,
    /// If true, `connect()` silently returns an empty stream when no script is queued.
    /// If false, `connect()` panics when no script is queued (forcing subscription tests
    /// to declare exactly the connections they expect).
    silent_default: bool,
}

/// A script for a single WebSocket connection.
pub struct FakeScript {
    /// Frames to return (if no transport error).
    pub frames: Vec<Vec<u8>>,
    /// Delay between frames.
    pub inter_frame_delay: Duration,
    /// Optional final wait after all frames (simulates idle gap or continued streaming).
    pub final_wait: Option<Duration>,
    /// Whether to return a transport error instead of returning frames.
    pub transport_error: bool,
    /// If set, after all `frames` are yielded the stream yields one `Transport`
    /// error, then closes. Used to test that a mid-stream transport error does
    /// not reset the idle-gap timer.
    pub mid_stream_error: bool,
}

impl FakeWebSocketClient {
    /// Create a FakeWebSocketClient for subscription tests.
    ///
    /// Every `connect()` consumes exactly one script added via `add_script()`.
    /// Calling `connect()` with no script queued panics.
    pub fn new() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            silent_default: false,
        }
    }

    /// Create a FakeWebSocketClient that silently returns an empty stream on every connect.
    ///
    /// Intended for identity tests that must satisfy the pipeline's subscription
    /// stage but do not assert on its output.
    pub fn empty() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            silent_default: true,
        }
    }

    /// Add a script to the queue for the next connection.
    pub fn add_script(&self, script: FakeScript) {
        self.scripts.lock().unwrap().push(script);
    }
}

impl Default for FakeWebSocketClient {
    fn default() -> Self {
        Self::new()
    }
}

/// A fake frame stream returned by FakeWebSocketClient.
struct FakeFrameStream {
    frames: Vec<Vec<u8>>,
    current_frame: usize,
    inter_frame_delay: Duration,
    final_wait: Option<Duration>,
    mid_stream_error: bool,
    mid_stream_error_yielded: bool,
}

#[async_trait]
impl FrameStream for FakeFrameStream {
    async fn next_frame(&mut self) -> Option<Result<Vec<u8>, SubscriptionStageError>> {
        // Return frames one by one with inter-frame delays.
        if self.current_frame < self.frames.len() {
            if self.current_frame > 0 {
                tokio::time::sleep(self.inter_frame_delay).await;
            }
            let frame = self.frames[self.current_frame].clone();
            self.current_frame += 1;
            return Some(Ok(frame));
        }

        // All frames consumed. If mid-stream error mode is on and we haven't yielded the
        // error yet, sleep one inter-frame delay and yield it once — this lets the stage
        // observe a transport error without the idle-gap timer being reset.
        if self.mid_stream_error && !self.mid_stream_error_yielded {
            self.mid_stream_error_yielded = true;
            tokio::time::sleep(self.inter_frame_delay).await;
            return Some(Err(SubscriptionStageError::Transport {
                message: "fake mid-stream transport error".to_string(),
                source: None,
            }));
        }

        // Apply final_wait once (long enough to let the stage's idle-gap or budget timer fire).
        if let Some(wait_duration) = self.final_wait.take() {
            tokio::time::sleep(wait_duration).await;
        }

        // Stream closed.
        None
    }

    async fn close(&mut self) {
        // Noop for fake.
    }
}

#[async_trait]
impl WebSocketClient for FakeWebSocketClient {
    async fn connect(&self, _url: &Url) -> Result<Box<dyn FrameStream>, SubscriptionStageError> {
        let mut scripts = self.scripts.lock().unwrap();

        if scripts.is_empty() {
            if self.silent_default {
                // Silent-default mode (for identity tests): return an empty stream that
                // closes immediately. Do not consume any script slot.
                return Ok(Box::new(FakeFrameStream {
                    frames: vec![],
                    current_frame: 0,
                    inter_frame_delay: Duration::from_millis(0),
                    final_wait: None,
                    mid_stream_error: false,
                    mid_stream_error_yielded: false,
                }));
            }
            // Script-driven mode (for subscription tests): panic loudly so the test
            // author notices they forgot to declare a connection.
            panic!(
                "FakeWebSocketClient: no script queued for this connect() call. \
                Each subscription test must declare exactly the scripts it expects \
                the stage to consume. Use fake_ws.add_script() for each connect() \
                the stage will make. (Identity tests should use FakeWebSocketClient::empty() instead.)"
            );
        }

        let script = scripts.remove(0);

        if script.transport_error {
            return Err(SubscriptionStageError::Transport {
                message: "fake transport error".to_string(),
                source: None,
            });
        }

        Ok(Box::new(FakeFrameStream {
            frames: script.frames,
            current_frame: 0,
            inter_frame_delay: script.inter_frame_delay,
            final_wait: script.final_wait,
            mid_stream_error: script.mid_stream_error,
            mid_stream_error_yielded: false,
        }))
    }
}
