opendeviationbar-streaming 13.80.0

//! Niffler: immediate dead-letter replay on the flush thread.
//!
//! Phase 35: Dead-letter resilience -- automatic recovery.
//!
//! Scans the dead-letter directory for Parquet files written by
//! `dead_letter::write_dead_letter()` and replays them to ClickHouse
//! via JSONEachRow POST. Uses `.replaying` advisory lock to prevent
//! race with Python Charon (`kintsugi/dead_letter.py`).
//!
//! Called opportunistically every 60 seconds after successful flushes.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};

use arrow_array::RecordBatch;
use arrow_schema::DataType;

use super::config::ClickHouseWriterConfig;
use super::dead_letter::{DEAD_LETTER_DIR, DeadLetterError};

/// Maximum number of consecutive failures on a single file before it is quarantined.
const MAX_QUARANTINE_ATTEMPTS: u32 = 5;

/// Age (seconds since last status change) after which an orphaned `.replaying`
/// file is reclaimed back to `.parquet`. Must comfortably exceed the longest
/// plausible single-file replay (one bounded HTTP POST).
const STALE_REPLAYING_SECS: u64 = 600;

/// Per-file failure counts, persistent across replay cycles for the process
/// lifetime.
///
/// Issue #474 review: a per-call map can never reach
/// `MAX_QUARANTINE_ATTEMPTS` — each file is visited at most once per cycle,
/// so quarantine was unreachable. Counts are deliberately in-memory only:
/// on process restart a poison-worthy file gets another
/// `MAX_QUARANTINE_ATTEMPTS` retries before quarantine. That bounded
/// re-retry is acceptable — no on-disk counter state to corrupt or clean up.
fn failure_counts() -> &'static Mutex<HashMap<PathBuf, u32>> {
    static COUNTS: OnceLock<Mutex<HashMap<PathBuf, u32>>> = OnceLock::new();
    COUNTS.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Seconds since the file's status last changed (rename updates ctime, NOT
/// mtime — a dead-letter file written hours ago but renamed to `.replaying`
/// seconds ago must NOT look stale).
#[cfg(unix)]
fn seconds_since_status_change(md: &std::fs::Metadata) -> Option<u64> {
    use std::os::unix::fs::MetadataExt;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()?
        .as_secs();
    let ctime = u64::try_from(md.ctime()).ok()?;
    Some(now.saturating_sub(ctime))
}

#[cfg(not(unix))]
fn seconds_since_status_change(md: &std::fs::Metadata) -> Option<u64> {
    md.modified()
        .ok()
        .and_then(|m| m.elapsed().ok())
        .map(|d| d.as_secs())
}

/// Reclaim orphaned `.replaying` files back to `.parquet`.
///
/// A crash mid-replay (or a failed restore/poison rename) leaves a
/// `.replaying` file that is invisible to both Niffler (scans `.parquet`)
/// and Python Charon (skips `.replaying`) — permanently stranded data.
/// Age-gating on ctime avoids stealing a file another agent is actively
/// replaying; unreadable metadata is treated as not-stale (skip).
fn reclaim_stale_replaying(dead_letter_dir: &Path, min_age_secs: u64) {
    let Ok(entries) = std::fs::read_dir(dead_letter_dir) else {
        return;
    };
    for entry in entries.filter_map(|e| e.ok()) {
        let path = entry.path();
        if !path.extension().is_some_and(|ext| ext == "replaying") {
            continue;
        }
        let is_stale = entry
            .metadata()
            .ok()
            .and_then(|md| seconds_since_status_change(&md))
            .is_some_and(|age| age >= min_age_secs);
        if !is_stale {
            continue;
        }
        let restored = path.with_extension(""); // "x.parquet.replaying" -> "x.parquet"
        match std::fs::rename(&path, &restored) {
            Ok(()) => tracing::warn!(
                path = %restored.display(),
                "reclaimed stale .replaying orphan (crash mid-replay or failed rename); re-queued as .parquet"
            ),
            Err(e) => tracing::error!(
                path = %path.display(),
                error = %e,
                "failed to reclaim stale .replaying orphan; will retry next cycle"
            ),
        }
    }
}

/// Convert an Arrow `RecordBatch` to ClickHouse JSONEachRow format.
///
/// Returns a newline-separated string of JSON objects, one per row.
/// Handles Arrow types: Utf8, Float64, Int64, UInt32, UInt8, Boolean.
/// Nullable columns emit JSON `null` when the value is absent.
fn record_batch_to_json_each_row(batch: &RecordBatch) -> Result<String, DeadLetterError> {
    use arrow_array::{
        Array, BooleanArray, Float64Array, Int64Array, StringArray, UInt8Array, UInt32Array,
    };

    let schema = batch.schema();
    let num_rows = batch.num_rows();
    let num_cols = batch.num_columns();
    let mut lines = Vec::with_capacity(num_rows);

    for row_idx in 0..num_rows {
        let mut obj = serde_json::Map::with_capacity(num_cols);
        for col_idx in 0..num_cols {
            let field = schema.field(col_idx);
            let col = batch.column(col_idx);

            if col.is_null(row_idx) {
                obj.insert(field.name().clone(), serde_json::Value::Null);
                continue;
            }

            let value = match field.data_type() {
                DataType::Utf8 => {
                    let arr = col.as_any().downcast_ref::<StringArray>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not Utf8", field.name()))
                    })?;
                    serde_json::Value::String(arr.value(row_idx).to_string())
                }
                DataType::Float64 => {
                    let arr = col.as_any().downcast_ref::<Float64Array>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not Float64", field.name()))
                    })?;
                    serde_json::json!(arr.value(row_idx))
                }
                DataType::Int64 => {
                    let arr = col.as_any().downcast_ref::<Int64Array>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not Int64", field.name()))
                    })?;
                    serde_json::json!(arr.value(row_idx))
                }
                DataType::UInt32 => {
                    let arr = col.as_any().downcast_ref::<UInt32Array>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not UInt32", field.name()))
                    })?;
                    serde_json::json!(arr.value(row_idx))
                }
                DataType::UInt8 => {
                    let arr = col.as_any().downcast_ref::<UInt8Array>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not UInt8", field.name()))
                    })?;
                    serde_json::json!(arr.value(row_idx))
                }
                DataType::Boolean => {
                    let arr = col.as_any().downcast_ref::<BooleanArray>().ok_or_else(|| {
                        DeadLetterError::Arrow(format!("Column {} not Boolean", field.name()))
                    })?;
                    serde_json::json!(arr.value(row_idx))
                }
                dt => {
                    return Err(DeadLetterError::Arrow(format!(
                        "Unsupported Arrow type {:?} for column {}",
                        dt,
                        field.name()
                    )));
                }
            };
            obj.insert(field.name().clone(), value);
        }
        let json_str = serde_json::to_string(&obj)
            .map_err(|e| DeadLetterError::Arrow(format!("JSON serialization: {e}")))?;
        lines.push(json_str);
    }

    Ok(lines.join("\n"))
}

/// Niffler replay: scan dead-letter directory and replay Parquet files to ClickHouse.
///
/// - Scans `DEAD_LETTER_DIR` for `.parquet` files
/// - For each file (sorted alphabetically = oldest first by timestamp):
///   1. Renames to `.replaying` as advisory lock (prevents Python Charon race)
///   2. Reads Parquet via Arrow reader
///   3. Converts to JSONEachRow and POSTs to ClickHouse
///   4. On success: deletes the `.replaying` file
///   5. On failure: renames `.replaying` back to `.parquet`, stops (CH likely down)
/// - Returns total rows replayed
pub fn niffler_replay(
    rt: &tokio::runtime::Runtime,
    http_client: &reqwest::Client,
    config: &ClickHouseWriterConfig,
) -> usize {
    niffler_replay_dir(rt, http_client, config, Path::new(DEAD_LETTER_DIR))
}

/// Niffler replay with a custom directory (for testing).
pub fn niffler_replay_dir(
    rt: &tokio::runtime::Runtime,
    http_client: &reqwest::Client,
    config: &ClickHouseWriterConfig,
    dead_letter_dir: &Path,
) -> usize {
    if !dead_letter_dir.exists() {
        return 0;
    }

    // Issue #474 review: reclaim crash-orphaned .replaying files FIRST so
    // they re-enter this cycle's queue instead of being stranded forever.
    reclaim_stale_replaying(dead_letter_dir, STALE_REPLAYING_SECS);

    // Collect and sort .parquet files (oldest first by filename timestamp)
    let mut parquet_files: Vec<PathBuf> = match std::fs::read_dir(dead_letter_dir) {
        Ok(entries) => entries
            .filter_map(|e| e.ok())
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|ext| ext == "parquet"))
            .collect(),
        Err(e) => {
            tracing::warn!(error = %e, "failed to read dead-letter directory");
            return 0;
        }
    };
    parquet_files.sort();

    if parquet_files.is_empty() {
        return 0;
    }

    let mut total_rows = 0;
    // Per-file failure attempts for permanent errors (corruption, schema
    // mismatch, ...). Process-lifetime static — see failure_counts() doc for
    // why a per-call map made quarantine unreachable.
    let mut file_failure_count = failure_counts()
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());

    for parquet_path in parquet_files {
        // Advisory lock: rename to .replaying (prevents Python Charon race)
        let replaying_path = parquet_path.with_extension("parquet.replaying");
        if let Err(e) = std::fs::rename(&parquet_path, &replaying_path) {
            tracing::warn!(
                path = %parquet_path.display(),
                error = %e,
                "failed to acquire .replaying advisory lock, skipping"
            );
            continue;
        }

        // Read Parquet file
        let file = match std::fs::File::open(&replaying_path) {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(
                    path = %replaying_path.display(),
                    error = %e,
                    "failed to open .replaying file (permanent file-specific error)"
                );
                quarantine_on_permanent_error(
                    &replaying_path,
                    &parquet_path,
                    &mut file_failure_count,
                    "file open",
                );
                continue; // Try next file instead of breaking
            }
        };

        let reader =
            match parquet::arrow::arrow_reader::ParquetRecordBatchReader::try_new(file, 8192) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(
                        path = %replaying_path.display(),
                        error = %e,
                        "failed to create Parquet reader (permanent file-specific error)"
                    );
                    quarantine_on_permanent_error(
                        &replaying_path,
                        &parquet_path,
                        &mut file_failure_count,
                        "Parquet reader creation",
                    );
                    continue; // Try next file instead of breaking
                }
            };

        let batches: Vec<RecordBatch> = match reader.into_iter().collect::<Result<_, _>>() {
            Ok(b) => b,
            Err(e) => {
                tracing::error!(
                    path = %replaying_path.display(),
                    error = %e,
                    "failed to read Parquet batches (permanent file-specific error)"
                );
                quarantine_on_permanent_error(
                    &replaying_path,
                    &parquet_path,
                    &mut file_failure_count,
                    "Parquet batch read",
                );
                continue; // Try next file instead of breaking
            }
        };

        let batch_row_count: usize = batches.iter().map(|b| b.num_rows()).sum();
        if batch_row_count == 0 {
            let _ = std::fs::remove_file(&replaying_path);
            continue;
        }

        // Convert all batches to JSONEachRow and POST to ClickHouse
        let mut all_json_lines = Vec::new();
        let mut conversion_failed = false;
        for batch in &batches {
            match record_batch_to_json_each_row(batch) {
                Ok(lines) => all_json_lines.push(lines),
                Err(e) => {
                    tracing::error!(
                        path = %replaying_path.display(),
                        error = %e,
                        "failed to convert Parquet batch to JSONEachRow (permanent file-specific error)"
                    );
                    conversion_failed = true;
                    break;
                }
            }
        }

        if conversion_failed {
            quarantine_on_permanent_error(
                &replaying_path,
                &parquet_path,
                &mut file_failure_count,
                "Parquet→JSON conversion",
            );
            continue; // Try next file instead of breaking
        }

        let body = all_json_lines.join("\n");
        let insert_sql = format!(
            "INSERT INTO {}.{} FORMAT JSONEachRow",
            config.database, config.table
        );

        let post_result = rt.block_on(async {
            http_client
                .post(&config.url)
                .query(&[
                    ("database", config.database.as_str()),
                    ("query", insert_sql.as_str()),
                    ("wait_end_of_query", "1"),
                ])
                .header("Content-Type", "application/json")
                .body(body)
                .send()
                .await
        });

        match post_result {
            Ok(resp) if resp.status().is_success() => {
                let _ = std::fs::remove_file(&replaying_path);
                total_rows += batch_row_count;
                tracing::info!(
                    rows = batch_row_count,
                    path = %parquet_path.display(),
                    "niffler replayed dead-letter file"
                );
                // Clear failure count on success
                file_failure_count.remove(&parquet_path);
            }
            Ok(resp) => {
                let status = resp.status().as_u16();
                tracing::warn!(
                    status,
                    path = %parquet_path.display(),
                    "niffler replay POST failed (transient CH error), deferring to next cycle"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break; // Stop on first transient ClickHouse failure (CH likely down)
            }
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    path = %parquet_path.display(),
                    "niffler replay network error (transient), deferring to next cycle"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break; // Stop on first transient network failure
            }
        }
    }

    total_rows
}

/// Handle a permanent (file-specific) failure by incrementing the failure counter.
/// If the counter reaches `MAX_QUARANTINE_ATTEMPTS`, rename the file to `.poison`
/// and log an error. Otherwise, restore the file to `.parquet` for retry.
fn quarantine_on_permanent_error(
    replaying_path: &Path,
    parquet_path: &Path,
    failure_count: &mut HashMap<PathBuf, u32>,
    error_context: &str,
) {
    let count = failure_count.entry(parquet_path.to_path_buf()).or_insert(0);
    *count += 1;

    if *count >= MAX_QUARANTINE_ATTEMPTS {
        let poison_path = parquet_path.with_extension("parquet.poison");
        if let Err(e) = std::fs::rename(replaying_path, &poison_path) {
            // File stays as .replaying — the stale-orphan reclaim at the
            // next cycle start returns it to .parquet, and the persistent
            // failure count re-triggers quarantine immediately.
            tracing::error!(
                path = %poison_path.display(),
                error = %e,
                context = error_context,
                attempts = *count,
                "failed to quarantine poisoned file (rename to .poison failed); stale-reclaim will retry"
            );
        } else {
            // Quarantined — drop the counter so the map stays bounded.
            failure_count.remove(parquet_path);
            tracing::error!(
                path = %poison_path.display(),
                context = error_context,
                max_attempts = MAX_QUARANTINE_ATTEMPTS,
                "quarantined poisoned dead-letter file after max attempts; future cycles will skip this file"
            );
        }
    } else {
        if let Err(e) = std::fs::rename(replaying_path, parquet_path) {
            tracing::error!(
                path = %parquet_path.display(),
                error = %e,
                context = error_context,
                attempts = *count,
                "failed to restore .parquet file (rename back failed)"
            );
        } else {
            tracing::warn!(
                path = %parquet_path.display(),
                context = error_context,
                attempts = *count,
                max_attempts = MAX_QUARANTINE_ATTEMPTS,
                "file-specific error detected; will retry on next cycle (or quarantine at {} attempts)",
                MAX_QUARANTINE_ATTEMPTS
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::clickhouse_writer::dead_letter::dead_letter_schema;
    use crate::clickhouse_writer::row::ClickHouseBarRow;
    use crate::live_engine::CompletedBar;
    use opendeviationbar_core::OpenDeviationBar;
    use opendeviationbar_core::fixed_point::FixedPoint;
    use std::sync::Arc;

    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;
        bar.duration_us = 100_000_000;
        bar.lookback_trade_count = Some(200);
        bar.lookback_ofi = Some(0.1);

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

    fn test_dead_letter_dir() -> PathBuf {
        let pid = std::process::id();
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        let dir = std::env::temp_dir().join(format!("opendeviationbar-niffler-test-{pid}-{nanos}"));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    /// Helper: write a dead-letter Parquet file to a custom directory.
    fn write_test_parquet(dir: &Path, filename: &str, rows: &[ClickHouseBarRow]) -> PathBuf {
        let schema = dead_letter_schema();
        // Reuse the internal rows_to_record_batch from dead_letter module
        // by writing via the public API and renaming
        let path = dir.join(filename);

        let batch =
            crate::clickhouse_writer::dead_letter::rows_to_record_batch_public(rows, &schema)
                .unwrap();

        let props = parquet::file::properties::WriterProperties::builder()
            .set_compression(parquet::basic::Compression::ZSTD(
                parquet::basic::ZstdLevel::try_new(3).unwrap(),
            ))
            .build();
        let file = std::fs::File::create(&path).unwrap();
        let mut writer =
            parquet::arrow::ArrowWriter::try_new(file, std::sync::Arc::new(schema), Some(props))
                .unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();
        path
    }

    fn test_config(url: &str) -> ClickHouseWriterConfig {
        ClickHouseWriterConfig {
            url: url.to_string(),
            max_rows: 500,
            flush_period_ms: 60_000,
            max_retries: 0,
            ..Default::default()
        }
    }

    #[test]
    fn test_niffler_replay_no_directory() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config("http://127.0.0.1:1");
        let nonexistent = Path::new("/tmp/opendeviationbar-niffler-nonexistent-dir");

        let result = niffler_replay_dir(&rt, &client, &config, nonexistent);
        assert_eq!(result, 0);
    }

    #[test]
    fn test_niffler_replay_empty_directory() {
        let dir = test_dead_letter_dir();
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config("http://127.0.0.1:1");

        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(result, 0);

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Start a wiremock server on a background thread and return the URI.
    /// The server stays alive until the returned JoinHandle's thread ends.
    fn start_mock_server(status: u16) -> (String, std::thread::JoinHandle<()>) {
        let (tx, rx) = std::sync::mpsc::channel();
        let handle = std::thread::spawn(move || {
            let mock_rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            mock_rt.block_on(async {
                let mock_server = wiremock::MockServer::start().await;
                wiremock::Mock::given(wiremock::matchers::method("POST"))
                    .respond_with(wiremock::ResponseTemplate::new(status))
                    .mount(&mock_server)
                    .await;
                tx.send(mock_server.uri()).unwrap();
                // Keep server alive until test is done
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
            });
        });
        let uri = rx.recv().unwrap();
        (uri, handle)
    }

    #[test]
    fn test_niffler_replay_success_deletes_file() {
        let dir = test_dead_letter_dir();
        let rows = vec![test_row(1, 10), test_row(11, 20)];
        let parquet_path = write_test_parquet(&dir, "BTCUSDT_250_1000.parquet", &rows);
        assert!(parquet_path.exists());

        let (uri, _server) = start_mock_server(200);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config(&uri);

        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(result, 2);

        // File should be deleted after successful replay
        assert!(!parquet_path.exists());
        // .replaying should also not exist
        assert!(!parquet_path.with_extension("parquet.replaying").exists());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_replay_failure_restores_file() {
        let dir = test_dead_letter_dir();
        let rows = vec![test_row(1, 10)];
        let parquet_path = write_test_parquet(&dir, "BTCUSDT_250_2000.parquet", &rows);
        assert!(parquet_path.exists());

        let (uri, _server) = start_mock_server(503);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config(&uri);

        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(result, 0);

        // File should be restored (not deleted)
        assert!(parquet_path.exists());
        // .replaying should not exist
        assert!(!parquet_path.with_extension("parquet.replaying").exists());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_replay_ignores_non_parquet() {
        let dir = test_dead_letter_dir();
        // Create a non-parquet file
        std::fs::write(dir.join("notes.txt"), "not a parquet file").unwrap();
        // Create a .replaying file (should also be ignored)
        std::fs::write(dir.join("test.parquet.replaying"), "locked").unwrap();

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config("http://127.0.0.1:1");

        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(result, 0);

        // Non-parquet files should still exist
        assert!(dir.join("notes.txt").exists());
        assert!(dir.join("test.parquet.replaying").exists());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_replay_uses_replaying_extension() {
        // Verify the .replaying advisory lock mechanism
        let dir = test_dead_letter_dir();
        let rows = vec![test_row(1, 10)];
        let parquet_path = write_test_parquet(&dir, "BTCUSDT_250_3000.parquet", &rows);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        // Use unreachable server to trigger failure, proving the rename happened
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_millis(100))
            .build()
            .unwrap();
        let config = test_config("http://192.0.2.1:1"); // RFC 5737 TEST-NET, unreachable

        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(result, 0);

        // File should be restored to .parquet (not left as .replaying)
        assert!(parquet_path.exists());
        assert!(!parquet_path.with_extension("parquet.replaying").exists());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_quarantine_corrupted_parquet_with_healthy_file() {
        // Test Issue #474: corrupted file should NOT block healthy files.
        // Setup: create two files, corrupt one, verify the other replays.
        let dir = test_dead_letter_dir();

        // Write a corrupted file (truncated, invalid magic)
        let corrupted_path = dir.join("01_corrupted.parquet");
        std::fs::write(&corrupted_path, b"INVALID_PARQUET_DATA").unwrap();

        // Write a healthy file
        let healthy_rows = vec![test_row(1, 10)];
        let healthy_path = write_test_parquet(&dir, "02_healthy.parquet", &healthy_rows);

        // Mock server that accepts both
        let (uri, _server) = start_mock_server(200);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config(&uri);

        // Cycle 1: healthy file replays despite the corrupted sibling.
        let result = niffler_replay_dir(&rt, &client, &config, &dir);
        assert_eq!(
            result, 1,
            "Healthy file should have been replayed despite corrupted file"
        );
        assert!(
            !healthy_path.exists(),
            "Healthy file should be deleted after successful replay"
        );

        // Corrupted file: restored to .parquet after cycle 1 (attempt 1 of
        // MAX_QUARANTINE_ATTEMPTS — failure counts persist ACROSS cycles).
        assert!(
            corrupted_path.exists(),
            "Corrupted file should be restored for retry before max attempts"
        );

        // Remaining cycles: failure count accumulates to the quarantine cap.
        for _ in 1..MAX_QUARANTINE_ATTEMPTS {
            let result = niffler_replay_dir(&rt, &client, &config, &dir);
            assert_eq!(result, 0);
        }

        // Corrupted file should be quarantined to .poison (not deleted)
        let poison_path = corrupted_path.with_extension("parquet.poison");
        assert!(
            poison_path.exists(),
            "Corrupted file should be quarantined to .poison after max attempts"
        );

        // Original corrupted .parquet should NOT exist (renamed to .poison)
        assert!(
            !corrupted_path.exists(),
            "Original corrupted file should be renamed away"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_reclaim_stale_replaying_orphan() {
        // Issue #474 review: a crash mid-replay strands a .replaying file
        // forever (Niffler scans .parquet; Charon skips .replaying). The
        // cycle-start reclaim must return aged orphans to .parquet.
        let dir = test_dead_letter_dir();
        let orphan = dir.join("orphan.parquet.replaying");
        std::fs::write(&orphan, b"stranded").unwrap();

        // Fresh file + real threshold: NOT reclaimed (age gate).
        reclaim_stale_replaying(&dir, STALE_REPLAYING_SECS);
        assert!(orphan.exists(), "Fresh .replaying must not be stolen");
        assert!(!dir.join("orphan.parquet").exists());

        // Zero threshold simulates an aged orphan: reclaimed to .parquet.
        reclaim_stale_replaying(&dir, 0);
        assert!(!orphan.exists(), "Aged orphan should be reclaimed");
        assert!(
            dir.join("orphan.parquet").exists(),
            "Reclaimed orphan should re-enter the queue as .parquet"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_quarantine_after_max_attempts() {
        // Test that a corrupted file is quarantined after MAX_QUARANTINE_ATTEMPTS failures.
        let dir = test_dead_letter_dir();

        // Write a corrupted file
        let corrupted_path = dir.join("corrupted.parquet");
        std::fs::write(&corrupted_path, b"BAD").unwrap();

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        // Use unreachable server (forces network error on successful parse, but we won't get there)
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_millis(100))
            .build()
            .unwrap();
        let config = test_config("http://192.0.2.1:1");

        // Run niffler replay MAX_QUARANTINE_ATTEMPTS + 1 times to trigger quarantine
        for attempt in 1..=MAX_QUARANTINE_ATTEMPTS + 1 {
            let result = niffler_replay_dir(&rt, &client, &config, &dir);
            assert_eq!(result, 0);

            if attempt < MAX_QUARANTINE_ATTEMPTS {
                // Before max attempts, file should still be .parquet (retryable)
                assert!(
                    corrupted_path.exists(),
                    "File should be restored to .parquet before max attempts (attempt {})",
                    attempt
                );
            }
        }

        // After MAX_QUARANTINE_ATTEMPTS, file should be quarantined to .poison
        let poison_path = corrupted_path.with_extension("parquet.poison");
        assert!(
            poison_path.exists(),
            "Corrupted file should be renamed to .poison after {} attempts",
            MAX_QUARANTINE_ATTEMPTS
        );
        assert!(
            !corrupted_path.exists(),
            "Original file should not exist after quarantine"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_niffler_skips_poison_files() {
        // Verify that .poison files are skipped (not processed).
        let dir = test_dead_letter_dir();

        // Create a .poison file
        std::fs::write(dir.join("poison.parquet.poison"), "ignored").unwrap();

        // Create a healthy file
        let healthy_rows = vec![test_row(1, 10)];
        let healthy_path = write_test_parquet(&dir, "healthy.parquet", &healthy_rows);

        let (uri, _server) = start_mock_server(200);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let client = reqwest::Client::new();
        let config = test_config(&uri);

        let result = niffler_replay_dir(&rt, &client, &config, &dir);

        // Only the healthy file should be replayed
        assert_eq!(result, 1);
        assert!(!healthy_path.exists());

        // .poison file should still exist (not processed)
        assert!(dir.join("poison.parquet.poison").exists());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_record_batch_to_json_each_row() {
        let rows = vec![test_row(1, 10)];
        let schema = dead_letter_schema();
        let batch =
            crate::clickhouse_writer::dead_letter::rows_to_record_batch_public(&rows, &schema)
                .unwrap();

        let json = record_batch_to_json_each_row(&batch).unwrap();
        let lines: Vec<&str> = json.lines().collect();
        assert_eq!(lines.len(), 1);

        // Parse as JSON and verify key fields
        let parsed: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
        assert_eq!(parsed["symbol"], "BTCUSDT");
        assert_eq!(parsed["threshold_decimal_bps"], 250);
        assert_eq!(parsed["first_agg_trade_id"], 1);
        assert_eq!(parsed["last_agg_trade_id"], 10);
        // Nullable field that is set
        assert_eq!(parsed["lookback_trade_count"], 200);
        // Nullable field that is null
        assert!(parsed["lookback_duration_us"].is_null());
    }
}