zc2 0.0.25

//! Dashboard API-based central ledger for credit management.
//!
//! This module provides a secure, centralized credit system:
//! - Credits managed via dashboard API (single source of truth)
//! - API key authentication via key format extraction
//! - Atomic transactions via dashboard API
//! - Full transaction history via API

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Transaction record
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LedgerTransaction {
    pub id: String,
    pub user_id: String,
    pub tx_type: TransactionType,
    pub amount: f64,
    pub balance_after: f64,
    pub timestamp: DateTime<Utc>,
    pub reference: Option<String>,
    pub authorized_by: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum TransactionType {
    Credit,
    Debit,
    Reserve,
    Commit,
    Cancel,
    Refund,
}

/// Ledger error types
#[derive(Debug)]
pub enum LedgerError {
    ConnectionFailed(String),
    Unauthorized(String),
    InsufficientCredits {
        required: f64,
        available: f64,
    },
    /// The billing authority could not be reached and no balance has ever been
    /// successfully loaded for this user, so the balance is *unknown* rather
    /// than zero. Distinct from InsufficientCredits on purpose: reporting
    /// "have 0.0000" when we simply failed to ask is actively misleading, and
    /// sent people hunting for a credits problem that did not exist.
    BalanceUnavailable {
        user_id: String,
    },
    InvalidReservation(String),
    InvalidAmount(String),
    DatabaseError(String),
}

impl std::fmt::Display for LedgerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ConnectionFailed(msg) => write!(f, "Ledger connection failed: {}", msg),
            Self::Unauthorized(msg) => write!(f, "Unauthorized: {}", msg),
            Self::InsufficientCredits {
                required,
                available,
            } => {
                write!(
                    f,
                    "Insufficient credits: need {}, have {}",
                    required, available
                )
            }
            Self::BalanceUnavailable { user_id } => {
                write!(
                    f,
                    "Balance for {} is unavailable: the billing authority could not be \
                     reached and no balance has been loaded yet. This is NOT a credits \
                     problem -- check connectivity to ZAKURO_API_URL.",
                    user_id
                )
            }
            Self::InvalidReservation(id) => write!(f, "Invalid reservation: {}", id),
            Self::InvalidAmount(msg) => write!(f, "Invalid amount: {}", msg),
            Self::DatabaseError(msg) => write!(f, "Database error: {}", msg),
        }
    }
}

/// Validate a credit amount arriving from any source (network or local).
/// Rejects NaN, infinities, and negatives. Zero is allowed (e.g. free jobs).
fn validate_amount(amount: f64) -> Result<(), LedgerError> {
    if !amount.is_finite() {
        return Err(LedgerError::InvalidAmount(format!(
            "non-finite amount: {}",
            amount
        )));
    }
    if amount < 0.0 {
        return Err(LedgerError::InvalidAmount(format!(
            "negative amount: {}",
            amount
        )));
    }
    Ok(())
}

/// Parse a dashboard balance response body. Accepts `credits_balance` or the
/// `balance` alias as a JSON number; returns None for malformed JSON, a
/// missing key, or a non-numeric value (so the caller can fail loudly rather
/// than treat a bad response as a 0.0 balance).
fn parse_balance_response(body: &str) -> Option<f64> {
    let parsed: serde_json::Value = serde_json::from_str(body).ok()?;
    parsed["credits_balance"]
        .as_f64()
        .or_else(|| parsed["balance"].as_f64())
}

/// Central ledger for credit management
pub struct Ledger {
    api_url: Option<String>,
    api_key: Option<String>,
    /// reservation_id → (user_id, amount)
    local_reservations: dashmap::DashMap<String, (String, f64)>,
    /// Local in-memory credits (used when API unavailable or for local ops)
    pub local_credits: dashmap::DashMap<String, f64>,
    /// Authoritative in-memory balances for P2P mode.
    /// Only populated for users this broker is authoritative for.
    pub authoritative_balances: dashmap::DashMap<String, f64>,
}

impl Ledger {
    /// Create a new ledger with the given api_url and api_key.
    pub fn new(api_url: Option<String>, api_key: Option<String>) -> Self {
        Self {
            api_url,
            api_key,
            local_reservations: dashmap::DashMap::new(),
            local_credits: dashmap::DashMap::new(),
            authoritative_balances: dashmap::DashMap::new(),
        }
    }

    /// Returns true when the broker is in API mode (dashboard API is the billing source).
    ///
    /// Standalone mode takes priority: if `ZAKURO_MASTER_KEY` is set, local credits
    /// are the billing source regardless of whether `api_url` is also configured.
    /// `api_url` may still be present for tx_buffer publishing even in standalone mode.
    pub fn is_api_mode(&self) -> bool {
        let master_key = std::env::var("ZAKURO_MASTER_KEY").unwrap_or_default();
        if !master_key.is_empty() {
            return false; // standalone billing always wins
        }
        self.api_url.is_some() && self.api_key.is_some()
    }

    /// Resolve user_id from an API key (Bearer token).
    /// Returns the zakuro_user_id associated with the key, or an error.
    pub fn resolve_user_from_api_key(&self, api_key: &str) -> Result<String, LedgerError> {
        Self::extract_user_from_key_format(api_key)
    }

    /// Extract user_id from key format: zk_{user_id}_{random_hex}
    /// ZAKURO_MASTER_KEY env var resolves to "admin". Any non-zk_ key also
    /// resolves to "admin" when ZAKURO_MASTER_KEY is not configured (standalone/dev mode).
    fn extract_user_from_key_format(api_key: &str) -> Result<String, LedgerError> {
        if api_key.is_empty() {
            return Err(LedgerError::Unauthorized("Empty API key".to_string()));
        }
        // Check master key from env → return "admin"
        let master_key = std::env::var("ZAKURO_MASTER_KEY").unwrap_or_default();
        if !master_key.is_empty() && api_key == master_key {
            return Ok("admin".to_string());
        }
        // Standard zk_ format: zk_{user_id}_{random_hex}
        if let Some(rest) = api_key.strip_prefix("zk_") {
            if let Some(pos) = rest.rfind('_') {
                let user_id = &rest[..pos];
                if !user_id.is_empty() {
                    return Ok(user_id.to_string());
                }
            }
            return Err(LedgerError::Unauthorized(
                "Invalid zk_ key format".to_string(),
            ));
        }
        // Fail closed: a non-zk_ key is only admin when it exactly matches a
        // configured ZAKURO_MASTER_KEY (handled above). With no master key set,
        // there is no admin identity — reject rather than granting admin to any
        // non-standard key (audit M4: closes the keyless admin-default bypass).
        Err(LedgerError::Unauthorized(
            "Invalid API key format (non-zk_ keys require a matching ZAKURO_MASTER_KEY)"
                .to_string(),
        ))
    }

    /// Get user's current balance — tries dashboard API first, falls back to local cache.
    pub fn get_balance(&self, user_id: &str) -> f64 {
        // In standalone mode (MASTER_KEY set), skip dashboard API — local_credits is authoritative.
        let master_key = std::env::var("ZAKURO_MASTER_KEY").unwrap_or_default();
        // Try dashboard API first (only when not in standalone mode)
        if master_key.is_empty() {
            if let (Some(ref api_url), Some(ref api_key)) = (&self.api_url, &self.api_key) {
                let endpoint = format!(
                    "{}/api/broker/balance/{}",
                    api_url.trim_end_matches('/'),
                    user_id
                );
                match ureq::get(&endpoint)
                    .header("X-Broker-Api-Key", api_key)
                    .call()
                {
                    Ok(resp) => {
                        let status = resp.status().as_u16();
                        if status == 200 {
                            let body = resp.into_body().read_to_string().unwrap_or_default();
                            match parse_balance_response(&body) {
                                Some(balance) => {
                                    // Populate BOTH credit maps so reserve() works correctly.
                                    // local_credits is consumed by reserve()/commit() (Standalone path).
                                    // authoritative_balances is used by local_reserve()/local_commit() (P2P Local path).
                                    self.local_credits.insert(user_id.to_string(), balance);
                                    self.authoritative_balances.insert(user_id.to_string(), balance);
                                    return balance;
                                }
                                // Fail loudly instead of caching a silent 0.0: a malformed
                                // response must NOT overwrite the cached balance.
                                None => eprintln!(
                                    "  [LEDGER] balance for {} unparseable from dashboard response; using cached value",
                                    user_id
                                ),
                            }
                        } else {
                            eprintln!("  [LEDGER] balance fetch for {} returned status {}; using cached value", user_id, status);
                        }
                    }
                    Err(e) => eprintln!(
                        "  [LEDGER] balance fetch for {} failed: {}; using cached value",
                        user_id, e
                    ),
                }
            }
        }
        // Fall back to whichever local cache has data
        self.authoritative_balances
            .get(user_id)
            .map(|v| *v)
            .or_else(|| self.local_credits.get(user_id).map(|v| *v))
            .unwrap_or(0.0)
    }

    /// True when a dashboard billing authority is configured — i.e. a failed
    /// lookup means "we could not ask", not "the balance is zero".
    ///
    /// In standalone / master-key mode there is no authority to be unreachable,
    /// so an absent user legitimately has 0 and must still be reported as out
    /// of credits (402). Mirrors the `standalone` test in
    /// `load_balance_if_needed` deliberately: if those two ever disagree, a
    /// standalone broker starts emitting BalanceUnavailable for ordinary
    /// empty accounts.
    pub fn has_billing_authority(&self) -> bool {
        self.api_url.is_some()
            && self.api_key.is_some()
            && std::env::var("ZAKURO_MASTER_KEY")
                .unwrap_or_default()
                .is_empty()
    }

    /// Like `get_balance`, but distinguishes "unknown" from "zero".
    ///
    /// Returns `None` only when the balance has never been successfully
    /// obtained *and* no cached value exists — i.e. we do not know it, as
    /// opposed to knowing it is 0. `get_balance` keeps collapsing that to 0.0
    /// for its many read-only callers (`/me`, logging, recovery); the billing
    /// paths use this instead so an unreachable authority cannot masquerade as
    /// an empty wallet.
    pub fn try_get_balance(&self, user_id: &str) -> Option<f64> {
        // Re-uses get_balance for the fetch + cache-populate side effects, then
        // reports whether anything actually landed in a cache. Deliberately not
        // a second copy of the fetch logic: two copies of a money path drift.
        let balance = self.get_balance(user_id);
        if self.authoritative_balances.contains_key(user_id)
            || self.local_credits.contains_key(user_id)
        {
            Some(balance)
        } else {
            None
        }
    }

    /// Reserve credits for a pending operation (local in-memory).
    ///
    /// Requires that `get_balance()` has been called first to populate local_credits.
    /// If local_credits is not yet populated for this user (race condition or first call),
    /// falls back to authoritative_balances as a safety net.
    pub fn reserve(
        &self,
        user_id: &str,
        amount: f64,
        _reference: &str,
    ) -> Result<String, LedgerError> {
        validate_amount(amount)?;
        let reservation_id = uuid::Uuid::new_v4().to_string();

        // Snapshot whether the balance is known BEFORE the entry() calls below:
        // they seed an absent key with 0.0, which would make a never-loaded
        // balance indistinguishable from a real zero by the time we report an
        // error.
        let balance_known = self.try_get_balance(user_id).is_some();

        // Load-or-insert the balance into local_credits. If the key is absent we
        // seed it from authoritative_balances (or 0.0) so the deduct below is not a
        // no-op — the old and_modify-only path reported a false InsufficientCredits.
        let seed = self
            .local_credits
            .get(user_id)
            .map(|v| *v)
            .or_else(|| self.authoritative_balances.get(user_id).map(|v| *v))
            .unwrap_or(0.0);

        let mut success = false;
        self.local_credits
            .entry(user_id.to_string())
            .and_modify(|b| {
                if *b >= amount {
                    *b -= amount;
                    success = true;
                }
            })
            .or_insert_with(|| {
                if seed >= amount {
                    success = true;
                    seed - amount
                } else {
                    seed
                }
            });
        if !success {
            // Same distinction as local_reserve: an unreachable billing
            // authority is not an empty wallet.
            if !balance_known && self.has_billing_authority() {
                return Err(LedgerError::BalanceUnavailable {
                    user_id: user_id.to_string(),
                });
            }
            let balance = self.local_credits.get(user_id).map(|v| *v).unwrap_or(0.0);
            return Err(LedgerError::InsufficientCredits {
                required: amount,
                available: balance,
            });
        }
        // Keep authoritative_balances in sync so /me and P2P billing see the correct balance.
        let balance_after = self.local_credits.get(user_id).map(|v| *v).unwrap_or(0.0);
        self.authoritative_balances
            .entry(user_id.to_string())
            .and_modify(|b| *b = balance_after)
            .or_insert(balance_after);
        self.local_reservations
            .insert(reservation_id.clone(), (user_id.to_string(), amount));
        Ok(reservation_id)
    }

    /// Commit a reservation (finalize the charge, refund difference).
    pub fn commit(&self, reservation_id: &str, actual_amount: f64) -> Result<f64, LedgerError> {
        if let Some((_, (user_id, reserved))) = self.local_reservations.remove(reservation_id) {
            // Cap the actual charge to [0, reserved]: non-finite or over-large
            // actual_amount charges the full reservation; negative charges nothing.
            let actual = if actual_amount.is_finite() {
                actual_amount.clamp(0.0, reserved)
            } else {
                reserved
            };
            let refund = (reserved - actual).max(0.0);
            if refund > 0.0 {
                self.local_credits
                    .entry(user_id.clone())
                    .and_modify(|b| *b += refund);
            }
            let final_bal = self.local_credits.get(&user_id).map(|v| *v).unwrap_or(0.0);
            // Keep authoritative_balances in sync so /me reflects the final balance.
            self.authoritative_balances
                .entry(user_id.to_string())
                .and_modify(|b| *b = final_bal)
                .or_insert(final_bal);
            return Ok(final_bal);
        }
        Err(LedgerError::InvalidReservation(reservation_id.to_string()))
    }

    /// Cancel a reservation (refund the full amount).
    pub fn cancel(&self, reservation_id: &str) -> Result<(), LedgerError> {
        if let Some((_, (user_id, amount))) = self.local_reservations.remove(reservation_id) {
            if amount > 0.0 {
                self.local_credits
                    .entry(user_id)
                    .and_modify(|b| *b += amount);
            }
        }
        Ok(())
    }

    /// Cancel a reservation from WAL replay (no DashMap entry — refund directly).
    pub fn cancel_from_wal(&self, user_id: &str, amount: f64) -> Result<(), LedgerError> {
        if amount <= 0.0 {
            return Ok(());
        }
        self.local_credits
            .entry(user_id.to_string())
            .and_modify(|b| *b += amount)
            .or_insert(amount);
        self.authoritative_balances
            .entry(user_id.to_string())
            .and_modify(|b| *b += amount);
        Ok(())
    }

    /// Commit from WAL replay (no DashMap entry — refund difference directly).
    pub fn commit_from_wal(
        &self,
        user_id: &str,
        reserved: f64,
        actual: f64,
    ) -> Result<f64, LedgerError> {
        let refund = reserved - actual;
        if refund > 0.0 {
            self.local_credits
                .entry(user_id.to_string())
                .and_modify(|b| *b += refund)
                .or_insert(refund);
            self.authoritative_balances
                .entry(user_id.to_string())
                .and_modify(|b| *b += refund);
        }
        Ok(self.get_balance(user_id))
    }

    /// Publish a transaction event via the dashboard API.
    #[allow(clippy::too_many_arguments)]
    pub fn publish_transaction(
        &self,
        _request_id: &str,
        user_id: &str,
        tx_type: &str,
        amount: f64,
        _balance_after: f64,
        worker_id: &str,
        duration_ms: f64,
        source_node: Option<&str>,
    ) {
        let (job_name, dashboard_type, status, credits_amount, compute_hours) = match tx_type {
            "commit" => {
                let job = if worker_id.is_empty() {
                    "Compute Job".to_string()
                } else {
                    format!("Compute Job ({})", worker_id)
                };
                let hours = if duration_ms > 0.0 {
                    duration_ms / 3_600_000.0
                } else {
                    0.0
                };
                (job, "job_execution", "completed", amount, hours)
            }
            "credit" => (
                "zkcr Added (Broker)".to_string(),
                "credit_purchase",
                "completed",
                amount,
                0.0,
            ),
            "cancel" => {
                let job = if worker_id.is_empty() {
                    "Cancelled Job".to_string()
                } else {
                    format!("Cancelled Job ({})", worker_id)
                };
                (job, "job_execution", "failed", 0.0, 0.0)
            }
            _ => return,
        };

        let compute_hours_opt: Option<f64> = if compute_hours > 0.0 {
            Some(compute_hours)
        } else {
            None
        };
        let duration_opt: Option<f64> = if duration_ms > 0.0 {
            Some(duration_ms)
        } else {
            None
        };

        if let (Some(ref api_url), Some(ref api_key)) = (&self.api_url, &self.api_key) {
            let worker_id_opt: Option<&str> = if worker_id.is_empty() {
                None
            } else {
                Some(worker_id)
            };
            let payload = serde_json::json!({
                "zakuro_user_id": user_id,
                "job_name": job_name,
                "transaction_type": dashboard_type,
                "credits_amount": credits_amount,
                "status": status,
                "compute_hours": compute_hours_opt,
                "worker_id": worker_id_opt,
                "source_node": source_node,
                "metadata": null,
                "duration_ms": duration_opt,
            });
            let endpoint = format!("{}/api/broker/transaction", api_url.trim_end_matches('/'));
            let body_str = serde_json::to_string(&payload).unwrap_or_default();
            let result = ureq::post(&endpoint)
                .header("X-Broker-Api-Key", api_key)
                .header("Content-Type", "application/json")
                .send(body_str.as_str());
            if let Err(e) = result {
                eprintln!("  [LEDGER] Failed to POST transaction via API: {}", e);
            }
        }
    }

    /// Get user info including balance.
    pub fn get_user_info(&self, user_id: &str) -> UserInfo {
        let balance = self.get_balance(user_id);
        UserInfo {
            user_id: user_id.to_string(),
            balance,
        }
    }

    /// Sync workers via Dashboard API instead of direct PostgreSQL access.
    /// This is the only supported worker sync method.
    ///
    /// `broker_wireguard_ip` is the broker's own WireGuard IP. It is used as the
    /// `wireguard_ip` for workers whose URI resolves to a loopback address (Docker
    /// workers share the broker's network namespace and have no independent IP).
    ///
    /// `broker_pubkey` is this broker's Ed25519 signing key, the same value sent to
    /// `/api/broker/node/register`. It is the join key between a worker row and the
    /// broker vouching for it: without it the dashboard can only fall back to the
    /// worker's own `last_seen`, which records that *something* called sync, not
    /// that a live broker stands behind it.
    pub fn sync_workers_via_api(
        zakuro_user_id: &str,
        workers: &[super::worker::Worker],
        api_url: &str,
        api_key: &str,
        node_name: Option<&str>,
        broker_wireguard_ip: Option<&str>,
        broker_pubkey: Option<&str>,
    ) -> Result<(), String> {
        use serde_json::json;

        if workers.is_empty() {
            return Ok(());
        }

        // Asked once per sync, not once per worker: every worker in this
        // container shares one supervisor, and N calls to answer one question
        // would be N chances to slow a sync that must finish promptly.
        // `None` whenever there is no supervisor to ask -- a node run outside
        // the compute image has none, and the sync carries on regardless.
        let health = super::node_health::worker_health();

        // Build worker sync payloads
        let worker_payloads: Vec<_> = workers
            .iter()
            .map(|worker| {
                let status_str = match worker.status {
                    super::worker::WorkerStatus::Healthy => "online",
                    super::worker::WorkerStatus::Busy => "online",
                    super::worker::WorkerStatus::Unhealthy => "offline",
                    super::worker::WorkerStatus::Draining => "offline",
                };

                let cpu_cores = worker.resources.cpus_available as i32;
                let ram_gb = (worker.resources.memory_available as f64 / (1024.0 * 1024.0 * 1024.0))
                    .round() as i32;

                // Docker workers share the broker's network namespace, so their URI IP
                // is loopback. Use the broker's own WireGuard IP instead.
                let effective_wireguard_ip = match worker.wireguard_ip.as_deref() {
                    Some("127.0.0.1") | Some("::1") | Some("localhost") | None => {
                        broker_wireguard_ip
                    }
                    Some(ip) => Some(ip),
                };

                let price_per_hour = worker.pricing.price_per_hour;

                json!({
                    "zakuro_user_id": zakuro_user_id,
                    "worker_id": &worker.name,  // Use stable name as worker_id
                    "name": &worker.name,
                    "status": status_str,
                    "gpu_model": worker.hardware.gpu_model.as_deref(),
                    "gpu_vram_gb": worker.hardware.gpu_vram_gb.map(|v| v as i32),
                    "cpu_model": worker.hardware.cpu_model.as_deref(),
                    "cpu_cores": cpu_cores,
                    "ram_gb": ram_gb,
                    "storage_gb": worker.hardware.storage_gb.map(|v| v as i32),
                    "source_node": node_name,
                    "wireguard_ip": effective_wireguard_ip,
                    "is_docker": worker.is_docker,
                    "price_per_hour": price_per_hour,
                    "min_charge": worker.pricing.min_charge,
                    "broker_pubkey": broker_pubkey,
                    // A COUNT, not a model name: gpu_model is never populated by
                    // any worker in this deployment, so a count is the only GPU
                    // fact we can actually report.
                    //
                    // Reports *available*, not total, to match cpu_cores and
                    // ram_gb above -- both derived from available figures. A
                    // total mixed in among them would make the three
                    // inconsistent in a way nothing downstream could detect.
                    "gpus": worker.resources.gpus_available as i32,
                    // What the node's own supervisor says about the worker
                    // process, from the s6 API over loopback inside the image.
                    // `status` here is what the BROKER thinks (reachable or
                    // not); these say whether the process itself is up, and
                    // how many times it has been restarted -- which is how a
                    // flapping worker looks different from a healthy one.
                    // All three are null on a node with no supervisor.
                    "service_status": health.as_ref().map(|h| h.status.as_str()),
                    "service_restart_count": health.as_ref().and_then(|h| h.restart_count),
                    "service_uptime": health.as_ref().and_then(|h| h.uptime.as_deref()),
                })
            })
            .collect();

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

        // Body is just the list of workers
        let payload_str = serde_json::to_string(&worker_payloads)
            .map_err(|e| format!("Failed to serialize payload: {}", e))?;

        let response = 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());

        match response {
            Ok(resp) => {
                if resp.status().as_u16() == 200 {
                    Ok(())
                } else {
                    Err(format!("API returned status {}", resp.status()))
                }
            }
            Err(e) => Err(format!("Failed to sync workers via API: {}", e)),
        }
    }
}

/// User info response
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UserInfo {
    pub user_id: String,
    pub balance: f64,
}

// --- P2P local credit operations (zero API calls on hot path) ---

impl Ledger {
    /// Load a user's balance from the dashboard API if not already cached in authoritative_balances.
    /// Returns the cached balance.
    pub fn load_balance_if_needed(&self, user_id: &str) -> f64 {
        // In standalone mode (no dashboard, or MASTER_KEY set), local_credits is the source of truth.
        // Sync it to authoritative_balances so the P2P billing path sees the correct balance.
        let standalone = self.api_url.is_none()
            || !std::env::var("ZAKURO_MASTER_KEY")
                .unwrap_or_default()
                .is_empty();
        if standalone {
            if let Some(bal) = self.local_credits.get(user_id).map(|v| *v) {
                self.authoritative_balances.insert(user_id.to_string(), bal);
                return bal;
            }
        }

        if let Some(balance) = self.authoritative_balances.get(user_id) {
            return *balance;
        }

        // First access — load from dashboard API.
        //
        // Only cache a value we actually obtained. Unconditionally inserting
        // get_balance()'s result cached 0.0 for a *failed* fetch, which then
        // looked indistinguishable from a real zero balance to every later
        // caller -- the bug this pairs with. On failure, leave the cache empty
        // so the next call retries and try_get_balance() can still say
        // "unknown".
        match self.try_get_balance(user_id) {
            Some(balance) => {
                self.authoritative_balances
                    .insert(user_id.to_string(), balance);
                balance
            }
            None => 0.0,
        }
    }

    /// Get the authoritative in-memory balance (returns None if not cached).
    pub fn get_authoritative_balance(&self, user_id: &str) -> Option<f64> {
        self.authoritative_balances.get(user_id).map(|v| *v)
    }

    /// Reserve credits in-memory (P2P local path — zero API calls).
    /// Returns (reservation_id, balance_before) on success.
    pub fn local_reserve(
        &self,
        user_id: &str,
        amount: f64,
        _reference: &str,
    ) -> Result<(String, f64), LedgerError> {
        validate_amount(amount)?;
        let reservation_id = uuid::Uuid::new_v4().to_string();

        // Ensure balance is loaded
        self.load_balance_if_needed(user_id);

        // Snapshot before the entry() below seeds an absent key with 0.0 --
        // otherwise "never loaded" is indistinguishable from "genuinely zero"
        // by the time the error is built.
        let balance_known = self.try_get_balance(user_id).is_some();

        // Atomic deduct from DashMap. Use or_insert_with so an absent key is
        // load-or-inserted (load_balance_if_needed seeds it in practice, but guard
        // against the and_modify no-op that reported a false InsufficientCredits).
        let mut success = false;
        let mut balance_before = 0.0;
        self.authoritative_balances
            .entry(user_id.to_string())
            .and_modify(|b| {
                balance_before = *b;
                if *b >= amount {
                    *b -= amount;
                    success = true;
                }
            })
            .or_insert_with(|| {
                // Absent key: treat seeded balance as 0.0 (nothing loaded).
                if amount <= 0.0 {
                    success = true;
                }
                0.0
            });

        if !success {
            // Distinguish "we know you are out of credits" from "we never
            // managed to ask". Without this, a broker that cannot reach the
            // dashboard denies every job with "have 0.0000" -- observed on a
            // machine with 500 credits whose only problem was connectivity.
            if !balance_known && self.has_billing_authority() {
                return Err(LedgerError::BalanceUnavailable {
                    user_id: user_id.to_string(),
                });
            }
            let available = self
                .authoritative_balances
                .get(user_id)
                .map(|v| *v)
                .unwrap_or(0.0);
            return Err(LedgerError::InsufficientCredits {
                required: amount,
                available,
            });
        }

        // Track reservation for commit/cancel
        self.local_reservations
            .insert(reservation_id.clone(), (user_id.to_string(), amount));

        Ok((reservation_id, balance_before))
    }

    /// Commit a reservation in-memory (P2P local path — zero API calls).
    /// Returns balance_after.
    pub fn local_commit(&self, reservation_id: &str, actual_cost: f64) -> Result<f64, LedgerError> {
        if let Some((_, (user_id, reserved))) = self.local_reservations.remove(reservation_id) {
            let actual = if actual_cost.is_finite() {
                actual_cost.clamp(0.0, reserved)
            } else {
                reserved
            };
            let refund = (reserved - actual).max(0.0);
            if refund > 0.0 {
                self.authoritative_balances
                    .entry(user_id.clone())
                    .and_modify(|b| *b += refund);
            }
            let balance_after = self
                .authoritative_balances
                .get(&user_id)
                .map(|v| *v)
                .unwrap_or(0.0);
            Ok(balance_after)
        } else {
            Err(LedgerError::InvalidReservation(reservation_id.to_string()))
        }
    }

    /// Cancel a reservation in-memory (P2P local path — zero API calls).
    pub fn local_cancel(&self, reservation_id: &str) -> Result<(), LedgerError> {
        if let Some((_, (user_id, amount))) = self.local_reservations.remove(reservation_id) {
            if amount > 0.0 {
                self.authoritative_balances
                    .entry(user_id)
                    .and_modify(|b| *b += amount);
            }
        }
        // Idempotent — OK even if not found
        Ok(())
    }

    /// Add credits to a user's authoritative in-memory balance (P2P earn path).
    /// Loads balance from dashboard API first if not already cached.
    /// Returns the new balance after the addition.
    pub fn local_add_credits(&self, user_id: &str, amount: f64) -> f64 {
        // Reject non-finite / negative credit amounts — never poison the balance.
        if validate_amount(amount).is_err() {
            return self.load_balance_if_needed(user_id);
        }
        self.load_balance_if_needed(user_id);
        self.authoritative_balances
            .entry(user_id.to_string())
            .and_modify(|b| *b += amount)
            .or_insert(amount);
        self.authoritative_balances
            .get(user_id)
            .map(|v| *v)
            .unwrap_or(amount)
    }

    /// Get all authoritative balance entries (for flush thread).
    pub fn authoritative_balance_snapshot(&self) -> Vec<(String, f64)> {
        self.authoritative_balances
            .iter()
            .map(|e| (e.key().clone(), *e.value()))
            .collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_local_fallback() {
        let ledger = Ledger::new(None, None);
        // Seed local credits manually
        ledger.local_credits.insert("user1".to_string(), 100.0);

        // Reserve
        let reservation = ledger.reserve("user1", 10.0, "test");
        assert!(reservation.is_ok());
        assert_eq!(
            ledger.local_credits.get("user1").map(|v| *v).unwrap_or(0.0),
            90.0
        );

        // Commit partial
        let commit = ledger.commit(&reservation.unwrap(), 5.0);
        assert!(commit.is_ok());
        assert_eq!(
            ledger.local_credits.get("user1").map(|v| *v).unwrap_or(0.0),
            95.0
        );
    }

    #[test]
    fn test_extract_user_from_key() {
        // zk_ keys always parse to the embedded user_id.
        assert_eq!(
            Ledger::extract_user_from_key_format("zk_9000000001_abc123def456").unwrap(),
            "9000000001"
        );
        // A configured master key grants admin on exact match.
        std::env::set_var("ZAKURO_MASTER_KEY", "the-master-key");
        assert_eq!(
            Ledger::extract_user_from_key_format("the-master-key").unwrap(),
            "admin"
        );
        // A non-zk_, non-matching key is REJECTED (not defaulted to admin).
        assert!(Ledger::extract_user_from_key_format("garbage").is_err());
        // With NO master key configured, a non-zk_ key is still rejected
        // (the old admin-default bypass is gone — audit M4).
        std::env::remove_var("ZAKURO_MASTER_KEY");
        assert!(Ledger::extract_user_from_key_format("garbage").is_err());
        // zk_ keys keep working with no master key.
        assert_eq!(
            Ledger::extract_user_from_key_format("zk_42_xyz").unwrap(),
            "42"
        );
    }

    #[test]
    fn parse_balance_response_reads_credits_balance() {
        assert_eq!(
            super::parse_balance_response(r#"{"credits_balance": 42.5}"#),
            Some(42.5)
        );
    }

    #[test]
    fn parse_balance_response_reads_balance_alias() {
        assert_eq!(
            super::parse_balance_response(r#"{"balance": 7.0}"#),
            Some(7.0)
        );
    }

    #[test]
    fn parse_balance_response_rejects_malformed_or_missing() {
        assert_eq!(super::parse_balance_response("not json"), None);
        assert_eq!(super::parse_balance_response("{}"), None);
        assert_eq!(
            super::parse_balance_response(r#"{"credits_balance": "oops"}"#),
            None
        );
    }

    // ── H6: idempotency & over-charge characterization ───────────────

    /// Committing a reservation that was already cancelled is rejected
    /// (no double settlement).
    #[test]
    fn test_commit_after_cancel_is_invalid() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        let rid = ledger.reserve("u", 40.0, "r").unwrap();
        ledger.cancel(&rid).unwrap(); // full refund → balance 100
        match ledger.commit(&rid, 40.0) {
            Err(LedgerError::InvalidReservation(_)) => {}
            other => panic!(
                "commit after cancel must be InvalidReservation, got {:?}",
                other
            ),
        }
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 100.0).abs() < 1e-9);
    }

    /// Committing the same reservation twice is rejected the second time.
    #[test]
    fn test_double_commit_is_invalid() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        let rid = ledger.reserve("u", 40.0, "r").unwrap();
        let bal = ledger.commit(&rid, 40.0).unwrap(); // charges 40 → 60
        assert!((bal - 60.0).abs() < 1e-9);
        match ledger.commit(&rid, 40.0) {
            Err(LedgerError::InvalidReservation(_)) => {}
            other => panic!("second commit must be InvalidReservation, got {:?}", other),
        }
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 60.0).abs() < 1e-9);
    }

    /// local_commit after local_cancel is likewise rejected.
    #[test]
    fn test_local_commit_after_cancel_is_invalid() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("u", 100.0);
        let (rid, _) = ledger.local_reserve("u", 40.0, "r").unwrap();
        ledger.local_cancel(&rid).unwrap();
        match ledger.local_commit(&rid, 40.0) {
            Err(LedgerError::InvalidReservation(_)) => {}
            other => panic!(
                "local_commit after cancel must be InvalidReservation, got {:?}",
                other
            ),
        }
        assert!((ledger.load_balance_if_needed("u") - 100.0).abs() < 1e-9);
    }

    /// cancel is idempotent: cancelling an unknown/already-cancelled id is Ok
    /// and does not refund twice.
    #[test]
    fn test_cancel_is_idempotent() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        let rid = ledger.reserve("u", 40.0, "r").unwrap(); // balance 60
        ledger.cancel(&rid).unwrap(); // refund → 100
        ledger.cancel(&rid).unwrap(); // no-op, still 100
        ledger.cancel("never-existed").unwrap(); // unknown id → Ok, no change
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 100.0).abs() < 1e-9);
    }

    /// Over-charge is clamped to the reservation: committing more than reserved
    /// charges exactly the reserved amount (asserts the charged value).
    #[test]
    fn test_over_charge_clamped_to_reserved_exact() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        let rid = ledger.reserve("u", 30.0, "r").unwrap(); // balance 70
        let bal = ledger.commit(&rid, 999.0).unwrap(); // clamp charge to 30
        assert!(
            (bal - 70.0).abs() < 1e-9,
            "charged exactly the reserved 30, balance 70, got {}",
            bal
        );
    }
}