opendeviationbar-streaming 13.78.0

//! Dedicated flush thread for batching and POSTing bars to ClickHouse.
//!
//! Issue #318: Rust-native ClickHouse writer -- flush thread.
//!
//! The flush thread runs on a dedicated `std::thread` with its own single-threaded
//! tokio runtime. It receives bars via a bounded `mpsc::SyncSender` channel and
//! flushes them when:
//!
//! 1. Buffer reaches `max_rows` bars (N-trigger)
//! 2. `flush_period_ms` elapses since last flush (T-trigger)
//! 3. An explicit `Flush` or `Shutdown` command arrives
//!
//! Uses raw `reqwest::Client` POST with JSONEachRow format for INSERT, giving
//! direct access to `X-ClickHouse-Summary` response headers for row count
//! verification (WRITER-07).
//!
//! Retry uses `backon::ExponentialBuilder` with transient error classification.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::Receiver;
use std::time::{Duration, Instant};

use backon::{ExponentialBuilder, Retryable};

use super::config::ClickHouseWriterConfig;
use super::dead_letter;
use super::dedup::compute_batch_dedup_token;
use super::niffler;
use super::row::ClickHouseBarRow;
use super::summary::{SUMMARY_HEADER, parse_summary, verify_written_rows};

/// Commands sent from the sink to the flush thread.
pub enum FlushCommand {
    /// A new bar row to buffer.
    Bar(Box<ClickHouseBarRow>),
    /// Explicit flush request (drain buffer now).
    Flush,
    /// Graceful shutdown (flush remaining, then exit).
    Shutdown,
}

/// Atomic counters for flush thread observability.
#[derive(Debug, Default)]
pub struct FlushThreadMetrics {
    /// Total bars successfully flushed to ClickHouse.
    pub bars_flushed: AtomicU64,
    /// Total bars that failed all retries.
    pub bars_failed: AtomicU64,
    /// Number of flush operations performed.
    pub flush_count: AtomicU64,
    /// Last flush latency in microseconds.
    pub last_flush_latency_us: AtomicU64,
    /// Number of dead-lettered batches (Phase 35).
    pub dead_letter_count: AtomicU64,
    /// Total bars replayed from dead-letter files by Niffler (Phase 35).
    pub dead_letter_replayed: AtomicU64,
}

/// Errors that can occur during a flush POST to ClickHouse.
#[derive(Debug)]
enum FlushError {
    /// HTTP response with non-2xx status code.
    Http { status: u16, body: String },
    /// Network-level error (connection refused, DNS, timeout, etc.).
    Network(reqwest::Error),
    /// Serialization error (should not happen in practice).
    Serialization(String),
}

impl std::fmt::Display for FlushError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            FlushError::Http { status, body } => {
                write!(f, "HTTP {}: {}", status, body)
            }
            FlushError::Network(e) => write!(f, "Network error: {}", e),
            FlushError::Serialization(e) => write!(f, "Serialization error: {}", e),
        }
    }
}

/// Classify whether a `FlushError` is transient (retryable).
///
/// Classification:
/// - HTTP 429, 500, 502, 503 -> transient (retry)
/// - Network/timeout errors -> transient (retry)
/// - HTTP 400, 401, other 4xx -> fatal (no retry)
/// - Serialization errors -> fatal (no retry)
fn is_transient_flush_error(err: &FlushError) -> bool {
    match err {
        FlushError::Http { status, .. } => matches!(status, 429 | 500 | 502 | 503),
        FlushError::Network(_) => true,
        FlushError::Serialization(_) => false,
    }
}

/// Spawn the dedicated flush thread.
///
/// The thread owns a single-threaded tokio runtime for async `reqwest`.
/// It loops on `rx.recv_timeout()` to implement the T-trigger (flush on timeout).
pub fn spawn_flush_thread(
    rx: Receiver<FlushCommand>,
    config: ClickHouseWriterConfig,
    metrics: Arc<FlushThreadMetrics>,
) -> std::thread::JoinHandle<()> {
    std::thread::Builder::new()
        .name("ch-flush".into())
        .spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("failed to create tokio runtime for flush thread");

            let http_client = reqwest::Client::new();

            let mut buffer: Vec<ClickHouseBarRow> = Vec::with_capacity(config.max_rows);
            let timeout = Duration::from_millis(config.flush_period_ms);
            let mut last_niffler_check = Instant::now();
            let niffler_interval = Duration::from_secs(60);

            // Macro to run niffler replay if 60s elapsed since last check.
            // Niffler MUST NOT block normal flush flow (Phase 35 anti-pattern).
            macro_rules! maybe_niffler {
                () => {
                    if last_niffler_check.elapsed() >= niffler_interval {
                        let replayed = niffler::niffler_replay(&rt, &http_client, &config);
                        if replayed > 0 {
                            metrics
                                .dead_letter_replayed
                                .fetch_add(replayed as u64, Ordering::Relaxed);
                            tracing::info!(replayed, "niffler replayed dead-letter bars");
                        }
                        last_niffler_check = Instant::now();
                    }
                };
            }

            loop {
                match rx.recv_timeout(timeout) {
                    Ok(FlushCommand::Bar(row)) => {
                        buffer.push(*row);
                        if buffer.len() >= config.max_rows {
                            do_flush(&rt, &http_client, &mut buffer, &config, &metrics);
                            maybe_niffler!();
                        }
                    }
                    Ok(FlushCommand::Flush) => {
                        if !buffer.is_empty() {
                            do_flush(&rt, &http_client, &mut buffer, &config, &metrics);
                            maybe_niffler!();
                        }
                    }
                    Ok(FlushCommand::Shutdown) => {
                        if !buffer.is_empty() {
                            do_flush(&rt, &http_client, &mut buffer, &config, &metrics);
                        }
                        // Final niffler pass on shutdown
                        let replayed = niffler::niffler_replay(&rt, &http_client, &config);
                        if replayed > 0 {
                            metrics
                                .dead_letter_replayed
                                .fetch_add(replayed as u64, Ordering::Relaxed);
                            tracing::info!(replayed, "niffler final replay on shutdown");
                        }
                        break;
                    }
                    Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                        if !buffer.is_empty() {
                            do_flush(&rt, &http_client, &mut buffer, &config, &metrics);
                            maybe_niffler!();
                        }
                    }
                    Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                        if !buffer.is_empty() {
                            do_flush(&rt, &http_client, &mut buffer, &config, &metrics);
                        }
                        break;
                    }
                }
            }

            tracing::info!(
                bars_flushed = metrics.bars_flushed.load(Ordering::Relaxed),
                bars_failed = metrics.bars_failed.load(Ordering::Relaxed),
                flush_count = metrics.flush_count.load(Ordering::Relaxed),
                "flush thread exiting"
            );
        })
        .expect("failed to spawn flush thread")
}

/// Flush the buffer to ClickHouse with retry via raw reqwest POST.
///
/// Uses JSONEachRow format and parses X-ClickHouse-Summary from response
/// headers for row count verification (WRITER-07).
fn do_flush(
    rt: &tokio::runtime::Runtime,
    http_client: &reqwest::Client,
    buffer: &mut Vec<ClickHouseBarRow>,
    config: &ClickHouseWriterConfig,
    metrics: &FlushThreadMetrics,
) {
    let start = Instant::now();
    let batch_size = buffer.len() as u64;

    // Compute per-batch dedup token and set metadata on all rows
    let dedup_token = compute_batch_dedup_token(buffer);
    let source_start_ts = buffer.iter().map(|r| r.open_time_us).min().unwrap_or(0);
    let source_end_ts = buffer.iter().map(|r| r.close_time_us).max().unwrap_or(0);

    for row in buffer.iter_mut() {
        row.cache_key.clone_from(&dedup_token);
        row.source_start_ts = source_start_ts;
        row.source_end_ts = source_end_ts;
    }

    // Serialize rows to JSONEachRow format (one JSON object per line).
    // Borrow buffer for serialization -- keep rows available for dead-lettering on failure.
    let body = match serialize_json_each_row(buffer) {
        Ok(b) => b,
        Err(e) => {
            tracing::error!(error = %e, "failed to serialize batch to JSONEachRow");
            metrics.bars_failed.fetch_add(batch_size, Ordering::Relaxed);
            metrics.dead_letter_count.fetch_add(1, Ordering::Relaxed);
            // Dead-letter the rows that failed serialization
            let failed_rows = std::mem::take(buffer);
            match dead_letter::write_dead_letter(&failed_rows) {
                Ok(path) => {
                    tracing::warn!(
                        path = %path.display(),
                        rows = failed_rows.len(),
                        "dead-lettered serialization-failed batch to Parquet"
                    );
                }
                Err(dl_err) => {
                    tracing::error!(
                        error = %dl_err,
                        rows = failed_rows.len(),
                        "CRITICAL: failed to dead-letter serialization-failed batch -- bars lost"
                    );
                }
            }
            return;
        }
    };

    let insert_sql = format!(
        "INSERT INTO {}.{} FORMAT JSONEachRow",
        config.database, config.table
    );
    let url = config.url.clone();
    let database = config.database.clone();
    let token = dedup_token.clone();
    let max_retries = config.max_retries as usize;

    let result = rt.block_on(async {
        let body_ref = &body;
        let url_ref = &url;
        let database_ref = &database;
        let insert_sql_ref = &insert_sql;
        let token_ref = &token;

        (|| async move {
            let resp = http_client
                .post(url_ref)
                .query(&[
                    ("database", database_ref.as_str()),
                    ("query", insert_sql_ref.as_str()),
                    ("wait_end_of_query", "1"),
                    ("insert_deduplication_token", token_ref.as_str()),
                ])
                .header("Content-Type", "application/json")
                .body(body_ref.clone())
                .send()
                .await
                .map_err(FlushError::Network)?;

            let status = resp.status().as_u16();
            if resp.status().is_success() {
                Ok(resp)
            } else {
                let body = resp.text().await.unwrap_or_default();
                Err(FlushError::Http { status, body })
            }
        })
        .retry(
            ExponentialBuilder::default()
                .with_min_delay(Duration::from_secs(1))
                .with_max_delay(Duration::from_secs(16))
                .with_max_times(max_retries)
                .with_jitter(),
        )
        .when(is_transient_flush_error)
        .await
    });

    let latency_us = start.elapsed().as_micros() as u64;
    metrics
        .last_flush_latency_us
        .store(latency_us, Ordering::Relaxed);

    match result {
        Ok(response) => {
            // Parse X-ClickHouse-Summary header for row count verification
            match response.headers().get(SUMMARY_HEADER) {
                Some(header_value) => {
                    if let Ok(header_str) = header_value.to_str() {
                        if let Some(summary) = parse_summary(header_str) {
                            verify_written_rows(&summary, batch_size);
                        } else {
                            tracing::warn!(
                                header_value = header_str,
                                "failed to parse X-ClickHouse-Summary header JSON"
                            );
                        }
                    } else {
                        tracing::warn!("X-ClickHouse-Summary header contains non-UTF8 value");
                    }
                }
                None => {
                    tracing::warn!("X-ClickHouse-Summary header missing from response");
                }
            }

            // Success: clear the buffer
            buffer.clear();
            metrics
                .bars_flushed
                .fetch_add(batch_size, Ordering::Relaxed);
            metrics.flush_count.fetch_add(1, Ordering::Relaxed);
            tracing::debug!(batch_size, latency_us, dedup_token, "flush successful");
        }
        Err(e) => {
            metrics.bars_failed.fetch_add(batch_size, Ordering::Relaxed);
            metrics.dead_letter_count.fetch_add(1, Ordering::Relaxed);
            // Dead-letter the failed rows to Parquet
            let failed_rows = std::mem::take(buffer);
            match dead_letter::write_dead_letter(&failed_rows) {
                Ok(path) => {
                    tracing::error!(
                        batch_size,
                        error = %e,
                        dedup_token,
                        path = %path.display(),
                        "flush failed after all retries - bars dead-lettered to Parquet"
                    );
                }
                Err(dl_err) => {
                    tracing::error!(
                        batch_size,
                        error = %e,
                        dl_error = %dl_err,
                        dedup_token,
                        "CRITICAL: flush failed AND dead-letter write failed -- bars lost"
                    );
                }
            }
        }
    }
}

/// Serialize a batch of rows to JSONEachRow format (one JSON object per line).
fn serialize_json_each_row(rows: &[ClickHouseBarRow]) -> Result<String, FlushError> {
    let mut lines = Vec::with_capacity(rows.len());
    for row in rows {
        let json =
            serde_json::to_string(row).map_err(|e| FlushError::Serialization(e.to_string()))?;
        lines.push(json);
    }
    Ok(lines.join("\n"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::OpenDeviationBar;
    use opendeviationbar_core::fixed_point::FixedPoint;
    use std::sync::Arc;

    use crate::live_engine::CompletedBar;

    /// Helper: create a test ClickHouseBarRow with a given trade ID range.
    fn test_row(first_tid: i64, last_tid: i64) -> ClickHouseBarRow {
        let mut bar = OpenDeviationBar::default();
        bar.open = FixedPoint::from_str("50000.0").unwrap();
        bar.high = FixedPoint::from_str("50100.0").unwrap();
        bar.low = FixedPoint::from_str("49900.0").unwrap();
        bar.close = FixedPoint::from_str("50050.0").unwrap();
        bar.vwap = FixedPoint::from_str("50025.0").unwrap();
        bar.open_time = 1_700_000_000_000_000;
        bar.close_time = 1_700_000_100_000_000;
        bar.first_agg_trade_id = first_tid;
        bar.last_agg_trade_id = last_tid;
        bar.individual_trade_count = 100;
        bar.agg_record_count = 50;

        let completed = CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar,
        };
        ClickHouseBarRow::from_completed_bar(&completed)
    }

    // -----------------------------------------------------------------------
    // Error classification tests (updated from clickhouse::error to FlushError)
    // -----------------------------------------------------------------------

    #[test]
    fn test_is_transient_flush_error_429() {
        let err = FlushError::Http {
            status: 429,
            body: "Too Many Requests".into(),
        };
        assert!(is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_transient_flush_error_503() {
        let err = FlushError::Http {
            status: 503,
            body: "Service Unavailable".into(),
        };
        assert!(is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_transient_flush_error_500() {
        let err = FlushError::Http {
            status: 500,
            body: "Internal Server Error".into(),
        };
        assert!(is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_transient_flush_error_502() {
        let err = FlushError::Http {
            status: 502,
            body: "Bad Gateway".into(),
        };
        assert!(is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_not_transient_flush_error_400() {
        let err = FlushError::Http {
            status: 400,
            body: "Bad Request".into(),
        };
        assert!(!is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_not_transient_flush_error_401() {
        let err = FlushError::Http {
            status: 401,
            body: "Unauthorized".into(),
        };
        assert!(!is_transient_flush_error(&err));
    }

    #[test]
    fn test_is_not_transient_serialization_error() {
        let err = FlushError::Serialization("bad data".into());
        assert!(!is_transient_flush_error(&err));
    }

    // -----------------------------------------------------------------------
    // Flush command and metrics tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_flush_command_enum() {
        let row = test_row(1, 10);
        let _cmd = FlushCommand::Bar(Box::new(row));
        let _cmd = FlushCommand::Flush;
        let _cmd = FlushCommand::Shutdown;
    }

    #[test]
    fn test_flush_thread_metrics_default() {
        let metrics = FlushThreadMetrics::default();
        assert_eq!(metrics.bars_flushed.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.flush_count.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.last_flush_latency_us.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.dead_letter_count.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.dead_letter_replayed.load(Ordering::Relaxed), 0);
    }

    // -----------------------------------------------------------------------
    // Flush thread behavior tests (unreachable server -> failure path)
    // -----------------------------------------------------------------------

    #[test]
    fn test_flush_on_shutdown_drains_buffer() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            max_rows: 500,
            flush_period_ms: 10_000,
            max_retries: 0,
            ..Default::default()
        };

        let (tx, rx) = std::sync::mpsc::sync_channel(100);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

        for i in 0..3 {
            tx.send(FlushCommand::Bar(Box::new(test_row(i * 10, i * 10 + 9))))
                .unwrap();
        }
        tx.send(FlushCommand::Shutdown).unwrap();
        handle.join().expect("flush thread panicked");

        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 3);
    }

    #[test]
    fn test_flush_on_row_count_triggers() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            max_rows: 3,
            flush_period_ms: 60_000,
            max_retries: 0,
            ..Default::default()
        };

        let (tx, rx) = std::sync::mpsc::sync_channel(100);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

        for i in 0..3 {
            tx.send(FlushCommand::Bar(Box::new(test_row(i * 10, i * 10 + 9))))
                .unwrap();
        }

        std::thread::sleep(Duration::from_millis(100));

        tx.send(FlushCommand::Shutdown).unwrap();
        handle.join().expect("flush thread panicked");

        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 3);
        assert_eq!(metrics.dead_letter_count.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_flush_on_timeout_triggers() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            max_rows: 500,
            flush_period_ms: 50,
            max_retries: 0,
            ..Default::default()
        };

        let (tx, rx) = std::sync::mpsc::sync_channel(100);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

        tx.send(FlushCommand::Bar(Box::new(test_row(1, 10))))
            .unwrap();

        std::thread::sleep(Duration::from_millis(200));

        tx.send(FlushCommand::Shutdown).unwrap();
        handle.join().expect("flush thread panicked");

        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_flush_explicit_flush_command() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            max_rows: 500,
            flush_period_ms: 60_000,
            max_retries: 0,
            ..Default::default()
        };

        let (tx, rx) = std::sync::mpsc::sync_channel(100);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

        tx.send(FlushCommand::Bar(Box::new(test_row(1, 5))))
            .unwrap();
        tx.send(FlushCommand::Bar(Box::new(test_row(6, 10))))
            .unwrap();
        tx.send(FlushCommand::Flush).unwrap();

        std::thread::sleep(Duration::from_millis(100));

        tx.send(FlushCommand::Shutdown).unwrap();
        handle.join().expect("flush thread panicked");

        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 2);
    }

    #[test]
    fn test_flush_on_channel_disconnect() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            max_rows: 500,
            flush_period_ms: 60_000,
            max_retries: 0,
            ..Default::default()
        };

        let (tx, rx) = std::sync::mpsc::sync_channel(100);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

        tx.send(FlushCommand::Bar(Box::new(test_row(1, 5))))
            .unwrap();
        drop(tx);

        handle.join().expect("flush thread panicked");

        assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_dedup_token_set_on_rows() {
        let row1 = test_row(1, 5);
        let row2 = test_row(6, 10);
        let mut buffer = vec![row1, row2];

        assert!(buffer[0].cache_key.is_empty());

        let dedup_token = compute_batch_dedup_token(&buffer);
        let source_start = buffer.iter().map(|r| r.open_time_us).min().unwrap_or(0);
        let source_end = buffer.iter().map(|r| r.close_time_us).max().unwrap_or(0);

        for row in buffer.iter_mut() {
            row.cache_key.clone_from(&dedup_token);
            row.source_start_ts = source_start;
            row.source_end_ts = source_end;
        }

        assert!(!buffer[0].cache_key.is_empty());
        assert_eq!(buffer[0].cache_key, buffer[1].cache_key);
        assert!(buffer[0].source_start_ts > 0);
        assert!(buffer[0].source_end_ts > 0);
    }

    #[test]
    fn test_summary_mismatch_detection() {
        use super::super::summary::ClickHouseSummary;

        let summary = ClickHouseSummary {
            read_rows: 0,
            written_rows: 0,
            written_bytes: 0,
            elapsed_ns: 1000,
        };
        assert!(!verify_written_rows(&summary, 5));

        let summary_ok = ClickHouseSummary {
            read_rows: 0,
            written_rows: 5,
            written_bytes: 12345,
            elapsed_ns: 1000,
        };
        assert!(verify_written_rows(&summary_ok, 5));
    }

    // -----------------------------------------------------------------------
    // Serialization tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_serialize_json_each_row() {
        let row1 = test_row(1, 5);
        let row2 = test_row(6, 10);
        let rows = vec![row1, row2];

        let body = serialize_json_each_row(&rows).expect("serialization should succeed");
        let lines: Vec<&str> = body.lines().collect();
        assert_eq!(lines.len(), 2);

        // Each line should be valid JSON
        for line in &lines {
            let _: serde_json::Value =
                serde_json::from_str(line).expect("each line should be valid JSON");
        }

        // First line should contain the symbol
        assert!(lines[0].contains("BTCUSDT"));
    }

    // -----------------------------------------------------------------------
    // Wiremock integration tests: verify summary header parsing on hot path
    // -----------------------------------------------------------------------

    #[test]
    fn test_flush_with_mock_server_summary_match() {
        // Test 1: Mock returns 200 + X-ClickHouse-Summary with matching written_rows
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            let mock_server = wiremock::MockServer::start().await;

            // Mock response with summary header showing 3 rows written
            wiremock::Mock::given(wiremock::matchers::method("POST"))
                .respond_with(
                    wiremock::ResponseTemplate::new(200).insert_header(
                        "X-ClickHouse-Summary",
                        r#"{"read_rows":"0","written_rows":"3","written_bytes":"12345","elapsed_ns":"1000000"}"#,
                    ),
                )
                .mount(&mock_server)
                .await;

            let config = ClickHouseWriterConfig {
                url: mock_server.uri(),
                max_rows: 500,
                flush_period_ms: 60_000,
                max_retries: 0,
                ..Default::default()
            };

            let (tx, rx) = std::sync::mpsc::sync_channel(100);
            let metrics = Arc::new(FlushThreadMetrics::default());
            let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

            // Send 3 bars then shutdown
            for i in 0..3 {
                tx.send(FlushCommand::Bar(Box::new(test_row(i * 10, i * 10 + 9))))
                    .unwrap();
            }
            tx.send(FlushCommand::Flush).unwrap();
            std::thread::sleep(Duration::from_millis(200));
            tx.send(FlushCommand::Shutdown).unwrap();
            handle.join().expect("flush thread panicked");

            // Flush succeeded: bars_flushed should be 3, bars_failed should be 0
            assert_eq!(metrics.bars_flushed.load(Ordering::Relaxed), 3);
            assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 0);
            assert_eq!(metrics.flush_count.load(Ordering::Relaxed), 1);
        });
    }

    #[test]
    fn test_flush_with_mock_server_summary_mismatch() {
        // Test 2: Mock returns 200 + X-ClickHouse-Summary with written_rows < batch size
        // (silent dedup scenario)
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            let mock_server = wiremock::MockServer::start().await;

            // Mock response with summary header showing only 1 of 3 rows written (dedup)
            wiremock::Mock::given(wiremock::matchers::method("POST"))
                .respond_with(
                    wiremock::ResponseTemplate::new(200).insert_header(
                        "X-ClickHouse-Summary",
                        r#"{"read_rows":"0","written_rows":"1","written_bytes":"4000","elapsed_ns":"500000"}"#,
                    ),
                )
                .mount(&mock_server)
                .await;

            let config = ClickHouseWriterConfig {
                url: mock_server.uri(),
                max_rows: 500,
                flush_period_ms: 60_000,
                max_retries: 0,
                ..Default::default()
            };

            let (tx, rx) = std::sync::mpsc::sync_channel(100);
            let metrics = Arc::new(FlushThreadMetrics::default());
            let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

            for i in 0..3 {
                tx.send(FlushCommand::Bar(Box::new(test_row(i * 10, i * 10 + 9))))
                    .unwrap();
            }
            tx.send(FlushCommand::Flush).unwrap();
            std::thread::sleep(Duration::from_millis(200));
            tx.send(FlushCommand::Shutdown).unwrap();
            handle.join().expect("flush thread panicked");

            // Flush succeeds (HTTP 200) but verify_written_rows logs warning
            // bars_flushed is still 3 (the HTTP call succeeded)
            assert_eq!(metrics.bars_flushed.load(Ordering::Relaxed), 3);
            assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 0);
        });
    }

    #[test]
    fn test_flush_with_mock_server_no_summary_header() {
        // Test 3: Mock returns 200 but NO X-ClickHouse-Summary header
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            let mock_server = wiremock::MockServer::start().await;

            // Mock response without summary header
            wiremock::Mock::given(wiremock::matchers::method("POST"))
                .respond_with(wiremock::ResponseTemplate::new(200))
                .mount(&mock_server)
                .await;

            let config = ClickHouseWriterConfig {
                url: mock_server.uri(),
                max_rows: 500,
                flush_period_ms: 60_000,
                max_retries: 0,
                ..Default::default()
            };

            let (tx, rx) = std::sync::mpsc::sync_channel(100);
            let metrics = Arc::new(FlushThreadMetrics::default());
            let handle = spawn_flush_thread(rx, config, Arc::clone(&metrics));

            tx.send(FlushCommand::Bar(Box::new(test_row(1, 5))))
                .unwrap();
            tx.send(FlushCommand::Flush).unwrap();
            std::thread::sleep(Duration::from_millis(200));
            tx.send(FlushCommand::Shutdown).unwrap();
            handle.join().expect("flush thread panicked");

            // Flush succeeds but tracing::warn about missing header
            assert_eq!(metrics.bars_flushed.load(Ordering::Relaxed), 1);
            assert_eq!(metrics.bars_failed.load(Ordering::Relaxed), 0);
        });
    }
}