zc2 0.0.25

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

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)
    }

    /// 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 = match tx.tx_type.as_str() {
                    "commit" => "job_execution",
                    "cancel" => "job_execution",
                    "credit" => "credit_purchase",
                    other => other,
                };
                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()
    }
}