zc2 0.0.32

//! Transaction flush buffer for batched API writes.
//!
//! Instead of writing each transaction to the dashboard API synchronously on the hot path,
//! completed transactions are queued here and flushed periodically (every 30s)
//! in a single batch. This removes the dashboard API call from the critical path when P2P mode
//! is enabled.

use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use std::io::Write;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use super::wal::{Wal, WalStatus};

/// Sentinel user id a broker assigns when it executes a job with no known
/// requester (local mode, no `X-Zakuro-User` header — see
/// `server::execute_prepare` and `deploy::serve::bill_served_request`, the
/// two sites that produce it). Local execution is single-node and free, and
/// the hub rejects this user by design (`User not found: anonymous`), so
/// there is nothing to settle and nothing to gain by buffering it.
pub const ANONYMOUS_USER: &str = "anonymous";

/// Number of failed flush attempts a transaction gets before it is
/// quarantined instead of re-queued. Bounds how long a permanently-rejected
/// transaction can block everything queued behind it in the same batch.
const MAX_FLUSH_ATTEMPTS: u32 = 5;

/// A buffered transaction waiting to be flushed to the dashboard API.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BufferedTransaction {
    pub request_id: String,
    pub user_id: String,
    pub tx_type: String,
    pub amount: f64,
    pub balance_after: f64,
    pub worker_id: String,
    pub duration_ms: f64,
    /// Name of the broker node that processed this transaction (executor origin)
    pub source_node: Option<String>,
    /// Name of the worker that executed the job (executor destination)
    pub worker_name: Option<String>,
    /// Worker URI / address (destination endpoint)
    pub worker_uri: Option<String>,
    /// Worker price in credits per hour (zkcr/hr)
    pub price_per_hour: f64,
}

/// Accumulates completed transactions and balance snapshots for periodic API flush.
pub struct TransactionBuffer {
    /// Pending transactions keyed by request_id.
    pending: DashMap<String, BufferedTransaction>,
    /// Authoritative balance snapshots keyed by user_id (latest value wins).
    balance_snapshots: DashMap<String, f64>,
    /// Count of transactions flushed since start.
    total_flushed: AtomicU64,
    /// Failed-flush attempt count per request_id. Kept alongside `pending`
    /// (not inside `BufferedTransaction`) so re-inserting a transaction on
    /// failure doesn't require touching every construction site. Survives
    /// re-insertion because it's keyed by `request_id`, not by the queue
    /// slot; cleared once a transaction is quarantined or durably flushed.
    attempts: DashMap<String, u32>,
    /// Set once the anonymous-sentinel skip has been logged, so a busy local
    /// broker logs it at most once per process instead of once per call.
    anonymous_warned: AtomicBool,
}

/// Internal `tx_type` → the dashboard's `transaction_type`. `credit` is how a
/// provider's peer earnings (`earn-{request_id}`) are booked, and the hub's
/// earnings query relies on it landing as `credit_purchase`.
pub(crate) fn dashboard_type(tx_type: &str) -> &str {
    match tx_type {
        "commit" | "cancel" => "job_execution",
        "credit" => "credit_purchase",
        other => other,
    }
}

/// Human-readable job name for a transaction, as sent to the dashboard and
/// as logged when a transaction is quarantined. Shared so the two stay in
/// sync (a quarantine log line matching the name the hub actually saw).
fn job_name_for(tx: &BufferedTransaction) -> String {
    match tx.tx_type.as_str() {
        "cancel" => format!(
            "Cancelled Job ({})",
            tx.worker_name.as_deref().unwrap_or(&tx.worker_id)
        ),
        _ => format!(
            "Compute Job ({})",
            tx.worker_name.as_deref().unwrap_or(&tx.worker_id)
        ),
    }
}

/// Summarize the hub's `errors[]` array (each entry keyed by job_name, not
/// request_id — see the call site) into one string suitable for every
/// per-item "failed" transaction quarantined from the same response.
fn summarize_item_errors(errors: &serde_json::Value) -> String {
    match errors.as_array() {
        Some(errors) if !errors.is_empty() => {
            let joined: Vec<String> = errors
                .iter()
                .take(3)
                .map(|e| {
                    format!(
                        "{}: {}",
                        e["job_name"].as_str().unwrap_or("?"),
                        e["error"].as_str().unwrap_or("unknown")
                    )
                })
                .collect();
            let mut summary = format!(
                "hub reported {} item failure(s): {}",
                errors.len(),
                joined.join("; ")
            );
            if errors.len() > 3 {
                summary.push_str(&format!(" … and {} more", errors.len() - 3));
            }
            summary
        }
        _ => "hub reported this item as failed (no error detail provided)".to_string(),
    }
}

impl TransactionBuffer {
    pub fn new() -> Self {
        Self {
            pending: DashMap::new(),
            balance_snapshots: DashMap::new(),
            total_flushed: AtomicU64::new(0),
            attempts: DashMap::new(),
            anonymous_warned: AtomicBool::new(false),
        }
    }

    /// Queue a completed transaction for the next flush.
    ///
    /// The anonymous sentinel (`ANONYMOUS_USER`) is refused here: the hub
    /// rejects it unconditionally, local execution has nothing to settle,
    /// and buffering it would just occupy a slot the batch would have to
    /// fail (or quarantine) on every flush. This is the single choke point —
    /// callers don't need to check the sentinel themselves.
    pub fn push_transaction(&self, tx: BufferedTransaction) {
        if tx.user_id == ANONYMOUS_USER {
            if !self.anonymous_warned.swap(true, Ordering::Relaxed) {
                eprintln!(
                    "  [FLUSH] Skipping hub sync for anonymous-user transaction {} \
                     (local execution; nothing to settle)",
                    tx.request_id
                );
            }
            return;
        }
        self.pending.insert(tx.request_id.clone(), tx);
    }

    /// Record an authoritative balance snapshot for a user.
    pub fn snapshot_balance(&self, user_id: &str, balance: f64) {
        self.balance_snapshots.insert(user_id.to_string(), balance);
    }

    /// Number of pending transactions.
    pub fn pending_count(&self) -> usize {
        self.pending.len()
    }

    /// Total transactions flushed since start.
    pub fn total_flushed(&self) -> u64 {
        self.total_flushed.load(Ordering::Relaxed)
    }

    /// Pending transactions, cloned. Test-only: pins what `flush_to_api` will send.
    #[cfg(test)]
    pub(crate) fn pending_snapshot(&self) -> Vec<BufferedTransaction> {
        self.pending.iter().map(|e| e.value().clone()).collect()
    }

    /// Drain all pending items, returning them.
    fn drain(&self) -> (Vec<BufferedTransaction>, Vec<(String, f64)>) {
        // Drain transactions
        let mut txs = Vec::with_capacity(self.pending.len());
        let keys: Vec<String> = self.pending.iter().map(|e| e.key().clone()).collect();
        for key in keys {
            if let Some((_, tx)) = self.pending.remove(&key) {
                txs.push(tx);
            }
        }

        // Drain balance snapshots
        let mut balances = Vec::with_capacity(self.balance_snapshots.len());
        let bkeys: Vec<String> = self
            .balance_snapshots
            .iter()
            .map(|e| e.key().clone())
            .collect();
        for key in bkeys {
            if let Some((uid, bal)) = self.balance_snapshots.remove(&key) {
                balances.push((uid, bal));
            }
        }

        (txs, balances)
    }

    /// Sibling path of the WAL used for permanently-rejected transactions:
    /// same directory resolution as the WAL itself (`~/.zakuro/wal.jsonl` →
    /// `~/.zakuro/wal-rejected.jsonl`), so no separate config/env is needed.
    fn rejected_path(wal: &Wal) -> std::path::PathBuf {
        let wal_path = wal.path();
        let stem = wal_path
            .file_stem()
            .and_then(|s| s.to_str())
            .unwrap_or("wal");
        let ext = wal_path
            .extension()
            .and_then(|s| s.to_str())
            .unwrap_or("jsonl");
        wal_path.with_file_name(format!("{stem}-rejected.{ext}"))
    }

    /// Append one permanently-rejected transaction to `wal-rejected.jsonl`
    /// and log a single line naming its request id, job name and error.
    /// Never panics and never drops the transaction silently: a write
    /// failure is logged, but the transaction is still gone from `pending`
    /// (matching the "never re-insert a quarantined item" requirement) —
    /// the alternative of re-queuing it would put it right back where it
    /// started (the whole point of quarantining is to stop that loop).
    fn quarantine(&self, tx: &BufferedTransaction, wal: &Wal, error: &str) {
        let job_name = job_name_for(tx);
        let path = Self::rejected_path(wal);
        let line = serde_json::json!({
            "request_id": tx.request_id,
            "user_id": tx.user_id,
            "job_name": job_name,
            "tx_type": tx.tx_type,
            "amount": tx.amount,
            "worker_id": tx.worker_id,
            "error": error,
            "quarantined_at": chrono::Utc::now().to_rfc3339(),
        });

        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let write_result = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .and_then(|mut f| writeln!(f, "{}", line));
        if let Err(e) = write_result {
            eprintln!(
                "  [FLUSH] Failed to write quarantine file {}: {}",
                path.display(),
                e
            );
        }

        eprintln!(
            "  [FLUSH] Quarantined {} (job \"{}\"): {}",
            tx.request_id, job_name, error
        );
    }

    /// Record one failed flush attempt for a whole-request-level failure
    /// (transport error or non-200 response). Re-queues the transaction for
    /// the next flush unless it has now failed `MAX_FLUSH_ATTEMPTS` times, in
    /// which case it is quarantined instead so it stops blocking everything
    /// queued behind it.
    fn record_failure(&self, tx: BufferedTransaction, wal: &Wal, error: &str) {
        let attempts = {
            let mut entry = self.attempts.entry(tx.request_id.clone()).or_insert(0);
            *entry += 1;
            *entry
        };
        if attempts >= MAX_FLUSH_ATTEMPTS {
            self.attempts.remove(&tx.request_id);
            self.quarantine(&tx, wal, error);
        } else {
            self.pending.insert(tx.request_id.clone(), tx);
        }
    }

    /// Clear any attempt count for a transaction that has now durably
    /// reached the hub, so a stray counter can't outlive its transaction.
    fn record_success(&self, request_id: &str) {
        self.attempts.remove(request_id);
    }

    /// Flush all pending transactions to production dashboard via API.
    /// Returns the number of transactions flushed.
    ///
    /// `wal` is used to mark WAL-backed entries (currently: peer-earn credits,
    /// keyed `earn-{request_id}`) `Committed` once the batch has been durably
    /// delivered to the dashboard. This is the ONLY place earn WAL entries are
    /// marked Committed — until this succeeds they stay `Earned` (replay-eligible),
    /// so a crash before this point re-delivers the earn on restart, and a crash
    /// after it does not (the dashboard already has it, and dedups by
    /// request_id, so even a racing double-delivery is a harmless no-op).
    pub fn flush_to_api(&self, api_url: &str, api_key: &str, wal: &Wal) -> usize {
        let (txs, _balances) = self.drain();

        if txs.is_empty() {
            return 0;
        }

        let tx_count = txs.len();

        // Build JSON payload for batch-sync API
        let payload: Vec<serde_json::Value> = txs
            .iter()
            .map(|tx| {
                // Map internal tx_type to dashboard transaction_type
                let dashboard_type = dashboard_type(tx.tx_type.as_str());
                let job_name = job_name_for(tx);
                let (credits_amount, status) = match tx.tx_type.as_str() {
                    "cancel" => (0.0f64, "failed"),
                    _ => (tx.amount, "completed"),
                };
                serde_json::json!({
                    "request_id": tx.request_id,
                    "zakuro_user_id": tx.user_id,
                    "job_name": job_name,
                    "transaction_type": dashboard_type,
                    "credits_amount": credits_amount,
                    "status": status,
                    "duration_ms": if tx.duration_ms > 0.0 { serde_json::Value::Number(serde_json::Number::from_f64(tx.duration_ms).unwrap()) } else { serde_json::Value::Null },
                    "worker_id": tx.worker_name,
                    "source_node": tx.source_node,
                    "compute_hours": if tx.duration_ms > 0.0 { Some(tx.duration_ms / 3_600_000.0) } else { None::<f64> },
                    "executor": tx.worker_name,
                    "destination": tx.worker_uri,
                    "price_per_hour": if tx.price_per_hour > 0.0 { Some(tx.price_per_hour) } else { None::<f64> },
                    "metadata": null
                })
            })
            .collect();

        let endpoint = format!("{}/api/broker/batch-sync", api_url.trim_end_matches('/'));

        let payload_str = match serde_json::to_string(&payload) {
            Ok(s) => s,
            Err(e) => {
                eprintln!("  [FLUSH] Failed to serialize payload: {}", e);
                // Return transactions to buffer on failure
                for tx in txs {
                    self.pending.insert(tx.request_id.clone(), tx);
                }
                return 0;
            }
        };

        match ureq::post(&endpoint)
            .config()
            .http_status_as_error(false)
            .build()
            .header("X-Broker-Api-Key", api_key)
            .header("Content-Type", "application/json")
            .send(payload_str.as_str())
        {
            Ok(resp) if resp.status().as_u16() == 200 => {
                let body = resp.into_body().read_to_string().unwrap_or_default();
                let parsed: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                let inserted = parsed["inserted"].as_u64().unwrap_or(tx_count as u64);
                let failed = parsed["failed"].as_u64().unwrap_or(0);
                let success = parsed["success"].as_bool().unwrap_or(true);

                self.total_flushed.fetch_add(inserted, Ordering::Relaxed);

                // Honor per-item results when the dashboard provides them
                // (batch-sync isolates each item and reports applied/duplicate/
                // failed per request_id). Fall back to whole-batch-on-200
                // behavior only for older dashboards that omit `results[]`.
                match parsed["results"].as_array() {
                    Some(results) => {
                        // Build request_id → status map. Items with a null/absent
                        // request_id can't be matched to an earn WAL entry (and
                        // aren't earns), so they're irrelevant to WAL marking.
                        let mut item_status: std::collections::HashMap<&str, &str> =
                            std::collections::HashMap::new();
                        for r in results {
                            if let (Some(rid), Some(st)) =
                                (r["request_id"].as_str(), r["status"].as_str())
                            {
                                item_status.insert(rid, st);
                            }
                        }

                        // `results[]` keys per-item status by request_id, which is
                        // reliable. The hub's `errors[]` array, on the other hand,
                        // keys per-item detail by job_name — and job names are not
                        // unique (two concurrent jobs on the same worker share one),
                        // so an error entry can't be reliably matched back to a
                        // single failed request_id. Rather than guess that mapping,
                        // every item reported "failed" here carries the same
                        // aggregate summary of everything the hub reported.
                        let item_error_summary = summarize_item_errors(&parsed["errors"]);

                        for tx in &txs {
                            // Default to "applied" when the dashboard returned
                            // results but didn't mention this request_id (e.g.
                            // null request_id on a non-earn tx): it's not a
                            // reported failure, so treat as durable and drop.
                            let status = item_status
                                .get(tx.request_id.as_str())
                                .copied()
                                .unwrap_or("applied");
                            match status {
                                "failed" => {
                                    // batch-sync dedups by request_id (it checks
                                    // for an existing row before inserting), and
                                    // zc already sends request_id in the payload,
                                    // so retrying a per-item failure can't
                                    // double-book a balance. A per-row failure
                                    // can also be transient (e.g. a database
                                    // hiccup on the hub's side), so it gets the
                                    // same attempt-counted grace as a
                                    // whole-request failure rather than being
                                    // quarantined on the spot.
                                    self.record_failure(tx.clone(), wal, &item_error_summary);
                                }
                                // "applied" | "duplicate" | anything else durable
                                _ => {
                                    self.record_success(&tx.request_id);
                                    if tx.request_id.starts_with("earn-") {
                                        let _ = wal.update_status(
                                            &tx.request_id,
                                            WalStatus::Committed,
                                            Some(tx.amount),
                                            Some(tx.duration_ms),
                                        );
                                    }
                                }
                            }
                        }
                    }
                    None => {
                        // Back-compat: older dashboard without per-item results.
                        // HTTP 200 → treat the whole batch as durable and mark
                        // every earn WAL entry Committed.
                        for tx in &txs {
                            self.record_success(&tx.request_id);
                            if tx.request_id.starts_with("earn-") {
                                let _ = wal.update_status(
                                    &tx.request_id,
                                    WalStatus::Committed,
                                    Some(tx.amount),
                                    Some(tx.duration_ms),
                                );
                            }
                        }
                    }
                }

                if !success || failed > 0 {
                    eprintln!(
                        "  [FLUSH] Partial sync to {}: {} inserted, {} failed",
                        api_url, inserted, failed
                    );
                    if let Some(errors) = parsed["errors"].as_array() {
                        for e in errors.iter().take(3) {
                            eprintln!(
                                "  [FLUSH]   → {}: {}",
                                e["job_name"].as_str().unwrap_or("?"),
                                e["error"].as_str().unwrap_or("unknown")
                            );
                        }
                        if errors.len() > 3 {
                            eprintln!("  [FLUSH]   … and {} more errors", errors.len() - 3);
                        }
                    }
                } else {
                    println!(
                        "  [FLUSH] Synced {} transactions to {} via API",
                        inserted, api_url
                    );
                }
                inserted as usize
            }
            Ok(resp) => {
                let status = resp.status();
                eprintln!("  [FLUSH] API sync failed: status {}", status);
                let error = format!("status {}", status);
                // Requeue each transaction unless it has now hit
                // MAX_FLUSH_ATTEMPTS, in which case it's quarantined instead
                // of blocking everything else in the batch forever.
                for tx in txs {
                    self.record_failure(tx, wal, &error);
                }
                0
            }
            Err(e) => {
                eprintln!("  [FLUSH] API sync failed: {}", e);
                let error = e.to_string();
                // A transport error counts as a failed attempt too.
                for tx in txs {
                    self.record_failure(tx, wal, &error);
                }
                0
            }
        }
    }
}

impl std::fmt::Debug for TransactionBuffer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TransactionBuffer")
            .field("pending", &self.pending.len())
            .field("balance_snapshots", &self.balance_snapshots.len())
            .field("total_flushed", &self.total_flushed())
            .finish()
    }
}

/// The hub's earnings query depends on `flush_to_api` actually sending
/// `credit` transactions as `transaction_type: "credit_purchase"` in the
/// batch-sync payload — not just on `dashboard_type` returning the right
/// string in isolation (pinned separately in
/// `server::earn_prefix_contract_tests::dashboard_types_are_stable`).
#[cfg(test)]
mod earn_credit_purchase_payload_tests {
    use super::*;

    /// A hub that captures the batch-sync request body and replies success.
    /// Built on the shared `bind_loopback_http` rather than a new hand-rolled
    /// `tiny_http` server, the same pattern as
    /// `server::billing_price_pin_tests::mock_sync_hub`.
    fn capturing_hub() -> (String, std::sync::mpsc::Receiver<String>) {
        let (server, port) = crate::integration_tests::integration::bind_loopback_http();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            for mut req in server.incoming_requests() {
                let mut body = String::new();
                let _ = std::io::Read::read_to_string(req.as_reader(), &mut body);
                let _ = tx.send(body);
                let _ = req.respond(tiny_http::Response::from_string(
                    r#"{"success":true,"inserted":1,"failed":0}"#,
                ));
            }
        });
        (format!("http://127.0.0.1:{port}"), rx)
    }

    #[test]
    fn flush_sends_a_credit_transaction_as_credit_purchase() {
        let path = format!(
            "/tmp/zc_test_flush_dashboard_type_{}_{}.jsonl",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        let _ = std::fs::remove_file(&path);
        let wal = Wal::open(&path).expect("open test WAL");

        let buf = TransactionBuffer::new();
        buf.push_transaction(BufferedTransaction {
            request_id: "earn-r1".to_string(),
            user_id: "owner".to_string(),
            tx_type: "credit".to_string(),
            amount: 1.5,
            balance_after: 1.5,
            worker_id: "w0".to_string(),
            duration_ms: 10.0,
            source_node: None,
            worker_name: Some("w0".to_string()),
            worker_uri: None,
            price_per_hour: 0.0,
        });

        let (hub, bodies) = capturing_hub();
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 1);

        let sent: serde_json::Value = serde_json::from_str(&bodies.recv().unwrap()).unwrap();
        assert_eq!(
            sent[0]["request_id"], "earn-r1",
            "the earn row must be the one sent"
        );
        assert_eq!(
            sent[0]["transaction_type"], "credit_purchase",
            "a `credit` tx must be sent to the hub as `credit_purchase`, or the \
             earnings summary silently drops it"
        );

        let _ = std::fs::remove_file(&path);
    }
}

/// Fault 1 (whole-request failure retried forever) and Fault 2 (per-item
/// rejections silently dropped) and Fault 3 (buffering the anonymous
/// sentinel) all quarantine/skip via this module's helpers, on the same
/// `bind_loopback_http` fake-hub pattern as `earn_credit_purchase_payload_tests`.
#[cfg(test)]
mod quarantine_tests {
    use super::*;

    fn test_wal(name: &str) -> (std::path::PathBuf, Wal) {
        let path = std::path::PathBuf::from(format!(
            "/tmp/zc_test_flush_quarantine_{}_{}_{}.jsonl",
            name,
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let _ = std::fs::remove_file(&path);
        let wal = Wal::open(&path).expect("open test WAL");
        (path, wal)
    }

    fn cleanup(wal_path: &std::path::Path, wal: &Wal) {
        let _ = std::fs::remove_file(wal_path);
        let _ = std::fs::remove_file(TransactionBuffer::rejected_path(wal));
    }

    fn tx(request_id: &str, user_id: &str) -> BufferedTransaction {
        BufferedTransaction {
            request_id: request_id.to_string(),
            user_id: user_id.to_string(),
            tx_type: "commit".to_string(),
            amount: 1.0,
            balance_after: 1.0,
            worker_id: "w0".to_string(),
            duration_ms: 10.0,
            source_node: None,
            worker_name: Some("w0".to_string()),
            worker_uri: None,
            price_per_hour: 0.0,
        }
    }

    fn read_rejected_lines(wal: &Wal) -> Vec<serde_json::Value> {
        let path = TransactionBuffer::rejected_path(wal);
        match std::fs::read_to_string(&path) {
            Ok(contents) => contents
                .lines()
                .filter(|l| !l.trim().is_empty())
                .map(|l| serde_json::from_str(l).expect("quarantine line must be valid JSON"))
                .collect(),
            Err(_) => Vec::new(),
        }
    }

    /// A hub that always answers every request with the given HTTP status
    /// (used for the whole-request-failure tests).
    fn hub_always(status: u16) -> String {
        let (server, port) = crate::integration_tests::integration::bind_loopback_http();
        std::thread::spawn(move || {
            for mut req in server.incoming_requests() {
                let mut body = String::new();
                let _ = std::io::Read::read_to_string(req.as_reader(), &mut body);
                let _ = req.respond(tiny_http::Response::from_string("").with_status_code(status));
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    /// A hub reachable at bind time but not actually listening — every send
    /// fails as a transport error (connection refused), not an HTTP status.
    fn dead_hub() -> String {
        let (server, port) = crate::integration_tests::integration::bind_loopback_http();
        drop(server); // closes the listening socket; nothing answers `port`
        format!("http://127.0.0.1:{port}")
    }

    /// A hub that returns `status` for the first `fail_times` requests, then
    /// a plain success for every request after that.
    fn hub_fail_then_succeed(fail_times: usize, status: u16) -> String {
        let (server, port) = crate::integration_tests::integration::bind_loopback_http();
        std::thread::spawn(move || {
            let mut seen = 0usize;
            for mut req in server.incoming_requests() {
                let mut body = String::new();
                let _ = std::io::Read::read_to_string(req.as_reader(), &mut body);
                seen += 1;
                if seen <= fail_times {
                    let _ =
                        req.respond(tiny_http::Response::from_string("").with_status_code(status));
                } else {
                    let _ = req.respond(tiny_http::Response::from_string(
                        r#"{"success":true,"inserted":1,"failed":0}"#,
                    ));
                }
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    /// A hub that answers 200 but reports `bad_rid` failed (via `results[]`
    /// and `errors[]`) on every request, while everything else applies.
    fn hub_with_item_failure(bad_rid: &'static str) -> String {
        hub_with_item_failure_then_succeed(bad_rid, usize::MAX)
    }

    /// A hub that answers 200 and reports `bad_rid` failed (via `results[]`
    /// and `errors[]`) for the first `fail_times` requests, then reports it
    /// (and everything else) applied from then on.
    fn hub_with_item_failure_then_succeed(bad_rid: &'static str, fail_times: usize) -> String {
        let (server, port) = crate::integration_tests::integration::bind_loopback_http();
        std::thread::spawn(move || {
            let mut seen = 0usize;
            for mut req in server.incoming_requests() {
                let mut body = String::new();
                let _ = std::io::Read::read_to_string(req.as_reader(), &mut body);
                seen += 1;
                let bad_still_fails = seen <= fail_times;
                let parsed: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                let items = parsed.as_array().cloned().unwrap_or_default();
                let results: Vec<serde_json::Value> = items
                    .iter()
                    .map(|item| {
                        let rid = item["request_id"].as_str().unwrap_or("");
                        let status = if rid == bad_rid && bad_still_fails {
                            "failed"
                        } else {
                            "applied"
                        };
                        serde_json::json!({"request_id": rid, "status": status})
                    })
                    .collect();
                let failed = results.iter().filter(|r| r["status"] == "failed").count();
                let inserted = results.len() - failed;
                let resp_body = serde_json::json!({
                    "success": failed == 0,
                    "inserted": inserted,
                    "failed": failed,
                    "results": results,
                    "errors": if failed > 0 {
                        serde_json::json!([{"job_name": "Compute Job (w0)", "error": "User not found: anonymous"}])
                    } else {
                        serde_json::json!([])
                    },
                })
                .to_string();
                let _ = req.respond(tiny_http::Response::from_string(resp_body));
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    /// Test 1: a hub that always 500s never gets a transaction accepted.
    /// After MAX_FLUSH_ATTEMPTS flushes it must be gone from `pending` and
    /// present (exactly once — the quarantine event fires only on the attempt
    /// that crosses the threshold) in the quarantine file.
    #[test]
    fn whole_request_failure_quarantines_after_max_attempts() {
        let (path, wal) = test_wal("always_500");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-bad", "alice"));

        let hub = hub_always(500);
        for _ in 0..MAX_FLUSH_ATTEMPTS {
            assert_eq!(buf.pending_count(), 1, "not yet quarantined");
            let flushed = buf.flush_to_api(&hub, "key", &wal);
            assert_eq!(flushed, 0);
        }

        assert_eq!(buf.pending_count(), 0, "must be gone from pending");
        let rejected = read_rejected_lines(&wal);
        assert_eq!(
            rejected.len(),
            1,
            "the quarantine event (and its one warning) must fire exactly once, \
             not once per attempt"
        );
        assert_eq!(rejected[0]["request_id"], "r-bad");

        cleanup(&path, &wal);
    }

    /// Test 2: a transaction added after the bad one has been quarantined
    /// must still reach an accepting hub — the earlier failure does not
    /// keep blocking the queue forever.
    #[test]
    fn quarantined_entry_does_not_block_later_transactions() {
        let (path, wal) = test_wal("unblocks_queue");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-bad", "alice"));

        let bad_hub = hub_always(500);
        for _ in 0..MAX_FLUSH_ATTEMPTS {
            buf.flush_to_api(&bad_hub, "key", &wal);
        }
        assert_eq!(buf.pending_count(), 0, "bad tx quarantined out of pending");

        buf.push_transaction(tx("r-good", "bob"));
        let good_hub_url = {
            let (server, port) = crate::integration_tests::integration::bind_loopback_http();
            std::thread::spawn(move || {
                for mut req in server.incoming_requests() {
                    let mut body = String::new();
                    let _ = std::io::Read::read_to_string(req.as_reader(), &mut body);
                    let _ = req.respond(tiny_http::Response::from_string(
                        r#"{"success":true,"inserted":1,"failed":0}"#,
                    ));
                }
            });
            format!("http://127.0.0.1:{port}")
        };

        let flushed = buf.flush_to_api(&good_hub_url, "key", &wal);
        assert_eq!(flushed, 1, "the later transaction must reach the hub");
        assert_eq!(buf.pending_count(), 0);

        cleanup(&path, &wal);
    }

    /// Test 3: a transport error (nothing listening) counts as a failed
    /// attempt too, and quarantines at the same threshold as an HTTP failure.
    #[test]
    fn transport_error_counts_as_an_attempt_and_quarantines() {
        let (path, wal) = test_wal("transport_error");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-unreachable", "alice"));

        let hub = dead_hub();
        for _ in 0..MAX_FLUSH_ATTEMPTS {
            buf.flush_to_api(&hub, "key", &wal);
        }

        assert_eq!(buf.pending_count(), 0);
        let rejected = read_rejected_lines(&wal);
        assert_eq!(rejected.len(), 1);
        assert_eq!(rejected[0]["request_id"], "r-unreachable");

        cleanup(&path, &wal);
    }

    /// Test 4: a 200 response with a per-item failure gets the same
    /// attempt-counted grace as a whole-request failure — batch-sync dedups
    /// by request_id, so retrying a per-item rejection can't double-book a
    /// balance, and a per-row failure can be transient (e.g. a database
    /// hiccup on the hub's side). It's requeued (not quarantined) on each of
    /// the first `MAX_FLUSH_ATTEMPTS - 1` flushes, and only quarantined once
    /// it has failed `MAX_FLUSH_ATTEMPTS` times. Other items in the same
    /// batch keep flushing normally throughout.
    #[test]
    fn per_item_rejection_inside_200_gets_the_same_attempt_grace() {
        let (path, wal) = test_wal("per_item");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-bad-item", "anonymous-ish"));
        buf.push_transaction(tx("r-good-item", "bob"));

        let hub = hub_with_item_failure("r-bad-item");
        for i in 0..MAX_FLUSH_ATTEMPTS {
            let flushed = buf.flush_to_api(&hub, "key", &wal);
            if i == 0 {
                assert_eq!(
                    flushed, 1,
                    "the good item in the same batch must still flush on the first attempt"
                );
            }
        }

        assert_eq!(buf.pending_count(), 0, "must be gone from pending");
        let rejected = read_rejected_lines(&wal);
        assert_eq!(
            rejected.len(),
            1,
            "quarantined exactly once, after MAX_FLUSH_ATTEMPTS, not on the first failure"
        );
        assert_eq!(rejected[0]["request_id"], "r-bad-item");
        assert!(
            rejected[0]["error"]
                .as_str()
                .unwrap_or("")
                .contains("User not found"),
            "the hub's per-item error text must be preserved"
        );

        cleanup(&path, &wal);
    }

    /// Test 4b: a per-item failure that succeeds on a later flush (the
    /// transient case the attempt grace exists for) must never be
    /// quarantined, and must reach the hub once the hub starts applying it.
    #[test]
    fn per_item_rejection_that_later_succeeds_is_not_quarantined() {
        let (path, wal) = test_wal("per_item_transient");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-transient-item", "bob"));

        let hub = hub_with_item_failure_then_succeed("r-transient-item", 2);
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 0); // attempt 1: item fails
        assert_eq!(buf.pending_count(), 1, "requeued, not quarantined");
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 0); // attempt 2: item fails
        assert_eq!(buf.pending_count(), 1, "requeued, not quarantined");
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 1); // attempt 3: item applies

        assert_eq!(buf.pending_count(), 0);
        assert!(
            read_rejected_lines(&wal).is_empty(),
            "a per-item failure that eventually succeeds must never be quarantined"
        );

        cleanup(&path, &wal);
    }

    /// Test 5: a transient failure that later succeeds must NOT be
    /// quarantined — attempts stay below the threshold and the transaction
    /// eventually reaches the hub.
    #[test]
    fn transient_failure_then_success_is_not_quarantined() {
        let (path, wal) = test_wal("transient");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-transient", "alice"));

        let hub = hub_fail_then_succeed(2, 500);
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 0); // attempt 1: fails
        assert_eq!(buf.pending_count(), 1, "requeued, not quarantined");
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 0); // attempt 2: fails
        assert_eq!(buf.pending_count(), 1, "requeued, not quarantined");
        assert_eq!(buf.flush_to_api(&hub, "key", &wal), 1); // attempt 3: succeeds

        assert_eq!(buf.pending_count(), 0);
        assert!(
            read_rejected_lines(&wal).is_empty(),
            "a transaction that eventually succeeds must never be quarantined"
        );

        cleanup(&path, &wal);
    }

    /// Test 6: `push_transaction` with the anonymous sentinel must not
    /// buffer it for the hub, while a normal user is buffered as usual.
    #[test]
    fn anonymous_sentinel_is_never_buffered() {
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-anon", ANONYMOUS_USER));
        assert_eq!(
            buf.pending_count(),
            0,
            "the anonymous sentinel must never be buffered for the hub"
        );

        buf.push_transaction(tx("r-real", "alice"));
        assert_eq!(
            buf.pending_count(),
            1,
            "a normal user must still be buffered"
        );
    }

    /// Test 7: every quarantine-file line is valid JSON, one per line, and
    /// carries the error text (already exercised structurally by the tests
    /// above via `read_rejected_lines`, which parses each line as JSON —
    /// this test additionally pins the one-line-per-entry and error-text
    /// shape directly).
    #[test]
    fn quarantine_file_lines_are_valid_json_with_error_text() {
        let (path, wal) = test_wal("valid_json_lines");
        let buf = TransactionBuffer::new();
        buf.push_transaction(tx("r-one", "alice"));
        buf.push_transaction(tx("r-two", "alice"));

        let hub = hub_always(500);
        for _ in 0..MAX_FLUSH_ATTEMPTS {
            buf.flush_to_api(&hub, "key", &wal);
        }

        let rejected_path = TransactionBuffer::rejected_path(&wal);
        let contents = std::fs::read_to_string(&rejected_path).expect("quarantine file exists");
        let lines: Vec<&str> = contents.lines().filter(|l| !l.trim().is_empty()).collect();
        assert_eq!(lines.len(), 2, "one line per quarantined transaction");
        for line in &lines {
            let parsed: serde_json::Value =
                serde_json::from_str(line).expect("each line must be standalone valid JSON");
            assert!(
                parsed["error"].as_str().is_some_and(|s| !s.is_empty()),
                "each quarantined line must carry the error text"
            );
        }

        cleanup(&path, &wal);
    }
}