opendeviationbar-streaming 13.78.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::path::{Path, PathBuf};

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

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

/// 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;
    }

    // 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;

    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"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break;
            }
        };

        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"
                    );
                    let _ = std::fs::rename(&replaying_path, &parquet_path);
                    break;
                }
            };

        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"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break;
            }
        };

        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"
                    );
                    conversion_failed = true;
                    break;
                }
            }
        }

        if conversion_failed {
            let _ = std::fs::rename(&replaying_path, &parquet_path);
            break;
        }

        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"
                );
            }
            Ok(resp) => {
                let status = resp.status().as_u16();
                tracing::warn!(
                    status,
                    path = %parquet_path.display(),
                    "niffler replay POST failed, restoring file"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break; // Stop on first failure (CH likely down)
            }
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    path = %parquet_path.display(),
                    "niffler replay network error, restoring file"
                );
                let _ = std::fs::rename(&replaying_path, &parquet_path);
                break; // Stop on first failure
            }
        }
    }

    total_rows
}

#[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_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());
    }
}