zc2 0.0.27

//! 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::sync::atomic::{AtomicU64, Ordering};

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

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

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

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

    /// Queue a completed transaction for the next flush.
    pub fn push_transaction(&self, tx: BufferedTransaction) {
        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)
    }

    /// 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, credits_amount, status) = match tx.tx_type.as_str() {
                    "cancel" => (
                        format!("Cancelled Job ({})", tx.worker_name.as_deref().unwrap_or(&tx.worker_id)),
                        0.0f64,
                        "failed",
                    ),
                    _ => (
                        format!("Compute Job ({})", tx.worker_name.as_deref().unwrap_or(&tx.worker_id)),
                        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);
                            }
                        }

                        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" => {
                                    // NOT durable — re-buffer this item for retry
                                    // next flush; leave its WAL entry Earned.
                                    self.pending.insert(tx.request_id.clone(), tx.clone());
                                }
                                // "applied" | "duplicate" | anything else durable
                                _ => {
                                    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 {
                            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) => {
                eprintln!("  [FLUSH] API sync failed: status {}", resp.status());
                // Return transactions to buffer on failure
                for tx in txs {
                    self.pending.insert(tx.request_id.clone(), tx);
                }
                0
            }
            Err(e) => {
                eprintln!("  [FLUSH] API sync failed: {}", e);
                // Return transactions to buffer on failure
                for tx in txs {
                    self.pending.insert(tx.request_id.clone(), tx);
                }
                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);
    }
}