server.rs

use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use chrono::TimeZone;
use rust_decimal::Decimal;

use axum::extract::{Path, Query, State};
use axum::response::sse::{Event, Sse};
use axum::response::{Html, IntoResponse, Json};
use axum::routing::get;
use axum::Router;
use tokio_stream::StreamExt;

use arb_common::types::ExchangeId;

use super::auth::AuthDb;
use super::extended_api::ExtendedApi;
use super::extractors::Session;
use super::state::DashState;
use super::web_storage::WebStorage;

#[derive(Clone)]
pub struct AppState {
    pub dash: Arc<DashState>,
    pub auth: Arc<Mutex<AuthDb>>,
    pub extended_api: Option<ExtendedApi>,
    /// Raw db_path retained for: (a) the SSE snapshot enrichment helper which
    /// is invoked from inside `spawn_blocking` and prefers a one-shot open,
    /// and (b) long-running background tasks in `web::optimizer` /
    /// `web::backtest` that need their own private connection. Day-to-day
    /// per-request handlers should use `web_db` instead.
    pub db_path: &'static str,
    /// Shared SQLite connection for short-lived web request handlers.
    /// See `web::web_storage::WebStorage` doc for the rationale.
    pub web_db: WebStorage,
    /// Active symbols from exchange_symbols DB table (loaded once at startup).
    pub active_symbols: Arc<Vec<String>>,
    /// Serializes concurrent credential edits so two admins cannot race a
    /// read-modify-write on `.env`. Held for the whole save flow (pause →
    /// validate → write).
    pub credentials_write_lock: Arc<tokio::sync::Mutex<()>>,
    /// Simple per-session rate limiter for credential endpoints. Maps
    /// user_id → timestamps of recent hits. Caller trims old entries.
    pub credentials_rate: Arc<std::sync::Mutex<std::collections::HashMap<i64, Vec<std::time::Instant>>>>,
    /// Snapshot of every exchange's `orderbook_depth` as it was at process
    /// start. Hot reload compares new TOML values against this map — any
    /// divergence means the live WS clients still hold the old subscription
    /// URL and the change requires a full bot restart to take effect.
    pub boot_orderbook_depths: Arc<std::collections::HashMap<String, crate::config::OrderbookDepth>>,
    /// Snapshot of every exchange's `ws_ob_mode` as it was at process start.
    /// Like `boot_orderbook_depths`: WS connection topology (single all-markets
    /// vs one-per-symbol) is decided when the OB client task spawns, so a
    /// hot-reload flip is silently ineffective until restart.
    pub boot_ws_ob_modes: Arc<std::collections::HashMap<String, crate::config::WsObMode>>,
    /// Rolling Tick→Recv histogram per exchange. Written by the executor
    /// on every market update (lock-free); read here on every `/api/state`
    /// call to produce the dashboard snapshot.
    pub tick_latency: Arc<crate::metrics::tick_latency::TickLatencyTracker>,
    /// Exchanges deliberately excluded from `tick_latency` at startup
    /// (e.g. Zero1 which doesn't publish event timestamps). Surfaced in
    /// the JSON snapshot so the UI can render greyed-out rows with a
    /// tooltip instead of silently hiding the venue.
    pub tick_latency_untracked: Arc<Vec<crate::metrics::tick_latency::TickLatencyUntrackedEntry>>,
}

/// Start the web dashboard server. Runs until the process exits.
pub async fn start(
    dash_state: Arc<DashState>,
    auth_db: Arc<Mutex<AuthDb>>,
    extended_api: Option<ExtendedApi>,
    addr: SocketAddr,
    db_path: &'static str,
    boot_orderbook_depths: std::collections::HashMap<String, crate::config::OrderbookDepth>,
    boot_ws_ob_modes: std::collections::HashMap<String, crate::config::WsObMode>,
    tick_latency: Arc<crate::metrics::tick_latency::TickLatencyTracker>,
    tick_latency_untracked: Vec<crate::metrics::tick_latency::TickLatencyUntrackedEntry>,
) -> anyhow::Result<()> {
    let dash_for_shutdown = dash_state.clone();
    let active_symbols = Arc::new({
        let db = crate::storage::sqlite::Storage::open(db_path)?;
        db.load_pairs_params().active_symbols()
    });
    let web_db = WebStorage::new(db_path)?;
    let state = AppState {
        dash: dash_state,
        auth: auth_db,
        extended_api,
        db_path,
        web_db,
        active_symbols,
        credentials_write_lock: Arc::new(tokio::sync::Mutex::new(())),
        credentials_rate: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
        boot_orderbook_depths: Arc::new(boot_orderbook_depths),
        boot_ws_ob_modes: Arc::new(boot_ws_ob_modes),
        tick_latency,
        tick_latency_untracked: Arc::new(tick_latency_untracked),
    };

    let app = Router::new()
        // Public routes (no auth)
        .route("/login", get(login_page))
        .route("/api/auth/login", axum::routing::post(super::auth_routes::login))

        // Auth management (require session)
        .route("/api/auth/logout", axum::routing::post(super::auth_routes::logout))
        .route("/api/auth/me", get(super::auth_routes::me))
        .route("/api/auth/password", axum::routing::post(super::auth_routes::change_password))

        // App routes (require session + permissions)
        .route("/", get(index))
        .route("/app.js", get(app_js))
        .route("/api/state", get(api_state))
        .route("/api/events", get(sse_events))
        .route("/api/log", get(api_log))
        .route("/api/logs/list", get(api_logs_list))
        .route("/api/logs/download/{filename}", get(api_logs_download))
        .route("/api/logs/download-session/{session_id}", get(api_logs_download_session))
        .route("/api/logs/download-all", get(api_logs_download_all))

        .route("/api/config", get(api_get_config).post(api_set_config))
        .route("/api/credentials/schemas", get(api_credentials_schemas))
        .route("/api/credentials", get(api_credentials_get))
        .route("/api/credentials/{exchange_id}", axum::routing::post(api_credentials_save))
        .route("/api/credentials/{exchange_id}/test", axum::routing::post(api_credentials_test))
        .route("/api/restart", axum::routing::post(api_restart))
        .route("/api/restart/binaries", get(api_restart_binaries))
        .route("/api/shutdown", axum::routing::post(api_shutdown))
        .route("/api/close-all", axum::routing::post(api_close_all))
        .route("/api/close-trade/{trade_id}", axum::routing::post(api_close_trade))
        .route("/api/halt/pause", axum::routing::post(api_halt_pause))
        .route("/api/halt/resume", axum::routing::post(api_halt_resume))
        .route("/api/halt/resume-api", axum::routing::post(api_halt_resume_api))

        // History (cross-session)
        .route("/api/history/sessions", get(api_history_sessions))
        .route("/api/history/sessions/{id}/trades", get(api_history_session_trades))

        // Extended exchange data (rewards/leaderboard only — funding moved to Parquet)
        .route("/api/extended/rewards", get(api_extended_rewards))

        // Backtest
        .route("/api/backtest", axum::routing::post(super::backtest::api_backtest_run))
        .route("/api/backtest/dates", get(super::backtest::api_backtest_dates))
        .route("/api/backtest/symbols", get(super::backtest::api_backtest_symbols))
        .route("/api/backtest/config", get(super::backtest::api_backtest_config))
        .route("/api/backtest/history", get(super::backtest::api_backtest_history_list))
        .route("/api/backtest/history/{filename}", get(super::backtest::api_backtest_history_get))

        // Optimizer
        .route("/api/optimizer/config", get(super::optimizer::api_optimizer_config))
        .route("/api/optimizer/symbols", get(super::optimizer::api_optimizer_symbols))
        .route("/api/optimizer/start", axum::routing::post(super::optimizer::api_optimizer_start))
        .route("/api/optimizer/running", get(super::optimizer::api_optimizer_running))
        .route("/api/optimizer/simulate", axum::routing::post(super::optimizer::api_optimizer_simulate))
        .route("/api/optimizer/cancel", axum::routing::post(super::optimizer::api_optimizer_cancel))
        .route("/api/optimizer/runs", get(super::optimizer::api_optimizer_runs))
        .route("/api/optimizer/runs/{id}", get(super::optimizer::api_optimizer_run_detail))
        .route("/api/optimizer/apply", axum::routing::post(super::optimizer::api_optimizer_apply))
        .route("/api/optimizer/tick-stats/{symbol}", get(super::optimizer::api_optimizer_tick_stats))

        // Config sync (push to remote servers)
        .route("/api/servers", get(api_get_servers))
        .route("/api/config/snapshot", get(api_config_snapshot))
        .route("/api/config/push", axum::routing::post(api_config_push))
        .route("/api/config/hot-reload", axum::routing::post(api_config_hot_reload))

        // Symbol overrides
        .route("/api/symbol-overrides", get(api_get_symbol_overrides).post(api_set_symbol_override))
        .route("/api/symbol-overrides/{symbol}/{ex1}/{ex2}", axum::routing::delete(api_delete_symbol_override))

        // Symbol blocks (disable/enable trading per symbol)
        .route("/api/symbol-blocks", get(api_get_symbol_blocks).post(api_create_symbol_block))
        .route("/api/symbol-blocks/{id}", axum::routing::delete(api_delete_symbol_block))

        // Admin routes
        .route("/api/auth/users", get(super::auth_routes::list_users).post(super::auth_routes::create_user))
        .route("/api/auth/users/{id}", axum::routing::put(super::auth_routes::update_user).delete(super::auth_routes::delete_user))
        .route("/api/auth/roles", get(super::auth_routes::list_roles).post(super::auth_routes::create_role))
        .route("/api/auth/roles/{name}", axum::routing::put(super::auth_routes::update_role).delete(super::auth_routes::delete_role))
        .with_state(state);

    // SO_REUSEADDR: allow binding even if the port is in TIME_WAIT from previous process.
    // Without this, restart fails with "Address already in use" for up to 120s on macOS.
    let socket = socket2::Socket::new(
        if addr.is_ipv6() { socket2::Domain::IPV6 } else { socket2::Domain::IPV4 },
        socket2::Type::STREAM,
        Some(socket2::Protocol::TCP),
    )?;
    socket.set_reuse_address(true)?;
    socket.set_nonblocking(true)?;
    socket.bind(&addr.into())?;
    socket.listen(1024)?;
    let listener = tokio::net::TcpListener::from_std(socket.into())?;
    tracing::info!(%addr, "Web dashboard listening");

    // Web server runs until process is killed by bot_start.sh.
    // SO_REUSEADDR + retry ensures new process can bind even while old holds port.
    // No graceful shutdown — keeps serving dashboard during cargo build (~2min).
    let _ = dash_for_shutdown; // used by engine shutdown, not web server
    axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await?;
    Ok(())
}

/// Serve login.html from disk (hot-reloadable) with compiled fallback.
async fn login_page() -> Html<String> {
    if let Ok(content) = tokio::fs::read_to_string("bot/src/web/login.html").await {
        return Html(content);
    }
    Html(include_str!("login.html").to_string())
}

/// Serve app.html — requires valid session, otherwise redirects to /login.
async fn index(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
) -> impl IntoResponse {
    // Check session server-side — redirect to login if invalid
    let authenticated = super::extractors::extract_session_token(&headers)
        .and_then(|token| {
            let auth = state.auth.lock().ok()?;
            auth.validate_session(&token).ok().flatten()
        })
        .is_some();

    if !authenticated {
        return axum::response::Redirect::to("/login").into_response();
    }

    let html = if let Ok(content) = tokio::fs::read_to_string("bot/src/web/app.html").await {
        content
    } else {
        include_str!("app.html").to_string()
    };
    Html(html).into_response()
}

/// Serve app.js — requires valid session.
async fn app_js(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
) -> impl IntoResponse {
    let authenticated = super::extractors::extract_session_token(&headers)
        .and_then(|token| {
            let auth = state.auth.lock().ok()?;
            auth.validate_session(&token).ok().flatten()
        })
        .is_some();

    if !authenticated {
        return axum::response::Redirect::to("/login").into_response();
    }

    let js = if let Ok(content) = tokio::fs::read_to_string("bot/src/web/app.js").await {
        content
    } else {
        include_str!("app.js").to_string()
    };
    ([("content-type", "application/javascript")], js).into_response()
}

async fn api_state(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }
    let mut snapshot = state.dash.read();
    // Enrich with SQLite data via the shared web connection (was opening
    // a fresh read-only conn per request, ~3 transient FDs each).
    let result = state.web_db.run(move |db| {
        let enrich_result = enrich_snapshot(&mut snapshot, db);
        let balance_delta = compute_balance_delta(&snapshot, db);
        (snapshot, enrich_result.is_ok(), balance_delta)
    }).await;
    let (snap, _ok, balance_delta) = result;

    // Attach the per-exchange Tick→Recv snapshot. The tracker lives in the
    // executor's hot path and is unaffected by the DashSnapshot rebuild
    // cadence — we just pull a fresh view here on every API call. Merging
    // into the parent JSON keeps the dashboard on a single poll path.
    let mut tick_snap = state.tick_latency.snapshot();
    tick_snap.untracked = (*state.tick_latency_untracked).clone();
    let mut body = match serde_json::to_value(&snap) {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(error = %e, "api_state: snapshot serialize failed");
            return Json(snap).into_response();
        }
    };
    if let Some(obj) = body.as_object_mut() {
        match serde_json::to_value(&tick_snap) {
            Ok(v) => { obj.insert("tick_latency".to_string(), v); }
            Err(e) => tracing::warn!(error = %e, "api_state: tick_latency serialize failed"),
        }
        match serde_json::to_value(&balance_delta) {
            Ok(v) => { obj.insert("balance_delta".to_string(), v); }
            Err(e) => tracing::warn!(error = %e, "api_state: balance_delta serialize failed"),
        }
    }
    Json(body).into_response()
}

/// Balance Δ = Σ current equity − Σ session-start equity. Sanity check
/// against Σ trade.pnl (computed through `derive_pnl`). When the two
/// disagree, accounting is losing fills somewhere — the 2026-04-18 MEGA
/// orphan scenario.
///
/// Returns per-exchange breakdown so the user sees WHICH exchange drifted.
/// Uses current `ExchangeHealthSnap.equity` as "current" (already fetched
/// by the periodic `refresh_balances` task) and the
/// `session_start_balances` table as "start". Exchanges missing from one
/// side contribute 0 to that side.
#[derive(serde::Serialize)]
struct BalanceDeltaEntry {
    exchange: String,
    start_usd: String,
    current_usd: String,
    delta_usd: String,
}

#[derive(serde::Serialize)]
struct BalanceDeltaSummary {
    /// True when we have both start + current for at least one exchange.
    /// Sessions that predate the feature report `available=false` and the
    /// UI should render "n/a" rather than a misleading zero delta.
    available: bool,
    total_start_usd: String,
    total_current_usd: String,
    total_delta_usd: String,
    per_exchange: Vec<BalanceDeltaEntry>,
}

fn compute_balance_delta(
    snap: &crate::web::state::DashSnapshot,
    db: &crate::storage::sqlite::Storage,
) -> BalanceDeltaSummary {
    use rust_decimal::Decimal;
    use std::str::FromStr;

    let session_id = snap.session_id as i64;
    let starts: std::collections::HashMap<i64, Decimal> = match db.load_session_start_balances(session_id) {
        Ok(rows) => rows.into_iter().collect(),
        Err(e) => {
            tracing::warn!(session_id, error = %e, "compute_balance_delta: load start balances failed");
            std::collections::HashMap::new()
        }
    };

    let mut total_start = Decimal::ZERO;
    let mut total_current = Decimal::ZERO;
    let mut per_exchange: Vec<BalanceDeltaEntry> = Vec::new();
    let mut any_known = false;

    // Resolve name → id once from the `exchanges` table (seeded with
    // lighter=1, extended=2, zero1=3, hyperliquid=4 and any hot-registered
    // others). Avoids a query per exchange below.
    let name_to_id: std::collections::HashMap<String, i64> = db
        .load_exchange_ids()
        .unwrap_or_default()
        .into_iter()
        .collect();

    for ex in &snap.health.exchanges {
        let Some(&exchange_id) = name_to_id.get(&ex.name) else { continue; };
        let start = starts.get(&exchange_id).copied().unwrap_or(Decimal::ZERO);
        let current = ex
            .equity
            .as_deref()
            .and_then(|s| Decimal::from_str(s.trim()).ok())
            .unwrap_or(Decimal::ZERO);
        if !start.is_zero() || !current.is_zero() {
            any_known = true;
        }
        total_start += start;
        total_current += current;
        per_exchange.push(BalanceDeltaEntry {
            exchange: ex.name.clone(),
            start_usd: start.to_string(),
            current_usd: current.to_string(),
            delta_usd: (current - start).to_string(),
        });
    }

    BalanceDeltaSummary {
        available: any_known && !starts.is_empty(),
        total_start_usd: total_start.to_string(),
        total_current_usd: total_current.to_string(),
        total_delta_usd: (total_current - total_start).to_string(),
        per_exchange,
    }
}

/// Log tail endpoint: returns last N lines from latest bot log.
#[derive(serde::Deserialize)]
struct LogQuery {
    #[serde(default = "default_log_lines")]
    lines: usize,
}
fn default_log_lines() -> usize { 1000 }

/// Cap on how many lines a single `/api/log` call may request. Prevents a
/// caller from forcing the server to read an arbitrarily large prefix of
/// the log file (memory + IO DoS).
const MAX_LOG_LINES: usize = 5000;

/// Approximate average log line length in bytes — used to size the initial
/// tail-read window. Production lines are 100-200 bytes; we round up so the
/// usual case fits in one read without seeking back further.
const LOG_LINE_AVG_BYTES: usize = 256;

async fn api_log(session: Session, Query(q): Query<LogQuery>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("log.view") { return e.into_response(); }
    let want = q.lines.clamp(1, MAX_LOG_LINES);
    let path = format!("{}/latest.log", arb_common::constants::LOG_DIR);

    // Tail-read: previously this handler `read_to_string`'d the whole log
    // file (often 8+ MB) on every poll, allocated a Vec<&str> for ALL
    // lines, and discarded everything except the last N. With the dashboard
    // auto-refreshing every few seconds that became sustained memory churn
    // and a contributing factor in the FD exhaustion incident on the
    // server. Now we seek to near the end and grow the read window only if
    // the requested line count isn't satisfied.
    let result = match tail_read_lines(&path, want).await {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(error = %e, path = %path, "api_log: tail_read_lines failed");
            return Json(serde_json::json!({ "lines": Vec::<String>::new(), "total": 0 })).into_response();
        }
    };

    Json(serde_json::json!({
        "lines": result.lines,
        "total": result.total_lines_in_window,
    })).into_response()
}

struct TailReadResult {
    lines: Vec<String>,
    /// Number of lines present in the window we actually scanned. Not the
    /// total lines in the whole file (we deliberately don't scan the whole
    /// file). The dashboard treats this as informational only.
    total_lines_in_window: usize,
}

async fn tail_read_lines(path: &str, want: usize) -> std::io::Result<TailReadResult> {
    use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom};

    let mut file = tokio::fs::File::open(path).await?;
    let file_size = file.metadata().await?.len();
    if file_size == 0 {
        return Ok(TailReadResult { lines: Vec::new(), total_lines_in_window: 0 });
    }

    // Estimate how many bytes are needed for `want` lines, rounded up to a
    // power-of-two-ish chunk so successive enlargements double cleanly. We
    // start at 4× the estimate so a single read is enough for the common
    // case (~256-byte lines × 1000 = 256 KB → start at 1 MB).
    let mut window: u64 = ((want as u64) * (LOG_LINE_AVG_BYTES as u64) * 4).max(64 * 1024);

    loop {
        let read_from = file_size.saturating_sub(window);
        let read_len = (file_size - read_from) as usize;
        file.seek(SeekFrom::Start(read_from)).await?;
        let mut buf = vec![0u8; read_len];
        file.read_exact(&mut buf).await?;

        // If we started mid-file, the first line in the buffer is almost
        // certainly truncated — drop everything up to (and including) the
        // first newline. When read_from == 0 we have the whole file and
        // must keep the first line.
        let usable: &[u8] = if read_from == 0 {
            &buf
        } else {
            match buf.iter().position(|&b| b == b'\n') {
                Some(pos) => &buf[pos + 1..],
                None => &[], // window contained no newline at all
            }
        };

        // `from_utf8_lossy` handles a chunk boundary that splits a UTF-8
        // sequence (rare in tracing output but possible for non-ASCII fields).
        let text = String::from_utf8_lossy(usable);
        let line_count = text.lines().count();

        // If we have enough lines, take the last `want` and return. Also
        // return when we already read the whole file — no point growing.
        if line_count >= want || read_from == 0 {
            let collected: Vec<String> = text
                .lines()
                .rev()
                .take(want)
                .map(|s| s.to_string())
                .collect::<Vec<_>>()
                .into_iter()
                .rev()
                .collect();
            return Ok(TailReadResult {
                lines: collected,
                total_lines_in_window: line_count,
            });
        }

        // Not enough lines — double the window and try again. Capped so a
        // pathologically sparse file (huge lines) can't drag the server
        // into reading the whole thing.
        let next_window = window.saturating_mul(2);
        if next_window >= file_size {
            window = file_size;
        } else if next_window > 16 * 1024 * 1024 {
            // 16 MB hard cap — beyond this, just return what we have.
            window = 16 * 1024 * 1024;
            if window <= read_from { // already past cap
                let collected: Vec<String> = text
                    .lines()
                    .rev()
                    .take(want)
                    .map(|s| s.to_string())
                    .collect::<Vec<_>>()
                    .into_iter()
                    .rev()
                    .collect();
                return Ok(TailReadResult {
                    lines: collected,
                    total_lines_in_window: line_count,
                });
            }
        } else {
            window = next_window;
        }
    }
}

/// Parse `bot_YYYY-MM-DD_HHMMSS.log` into epoch milliseconds (local timezone).
fn parse_log_filename_ms(name: &str) -> Option<u64> {
    let stem = name.strip_prefix("bot_")?.strip_suffix(".log")?;
    let (date, time) = stem.split_once('_')?;
    if date.len() != 10 || time.len() != 6 { return None; }
    let dt_str = format!("{} {}:{}:{}", date, &time[0..2], &time[2..4], &time[4..6]);
    let naive = chrono::NaiveDateTime::parse_from_str(&dt_str, "%Y-%m-%d %H:%M:%S").ok()?;
    let local = chrono::Local.from_local_datetime(&naive).single()?;
    Some(local.timestamp_millis() as u64)
}

/// Read log files from disk, match to bot_runs for session grouping.
struct LogFileEntry {
    name: String,
    start_ms: u64,
    end_ms: u64,
    size: u64,
    is_current: bool,
    session_id: Option<i64>,
}

async fn read_log_entries(web_db: &super::web_storage::WebStorage) -> Vec<LogFileEntry> {
    let log_dir = std::path::Path::new(arb_common::constants::LOG_DIR);

    // Read symlink target to identify current log
    let current_file = tokio::fs::read_link(log_dir.join("latest.log")).await
        .ok()
        .and_then(|p| p.file_name().map(|n| n.to_string_lossy().to_string()));

    // Read log files from disk
    let mut entries = Vec::new();
    let mut dir = match tokio::fs::read_dir(log_dir).await {
        Ok(d) => d,
        Err(_) => return entries,
    };
    while let Ok(Some(entry)) = dir.next_entry().await {
        let name = entry.file_name().to_string_lossy().to_string();
        if !name.starts_with("bot_") || !name.ends_with(".log") { continue; }
        let meta = match entry.metadata().await { Ok(m) => m, Err(_) => continue };
        let start_ms = parse_log_filename_ms(&name).unwrap_or(0);
        let end_ms = meta.modified().ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        entries.push(LogFileEntry {
            is_current: current_file.as_deref() == Some(&name),
            name, start_ms, end_ms, size: meta.len(),
            session_id: None,
        });
    }

    // Match log files to bot_runs by start time (within 60s tolerance).
    // Uses the shared web connection — previously this opened a fresh
    // raw rusqlite::Connection per call (Logs tab + 5s polling), burning
    // ~3 transient FDs each.
    let runs: Vec<(i64, i64)> = web_db.run(|db| {
        let mut stmt = match db.conn_ref().prepare(
            "SELECT started_at_ms, session_id FROM bot_runs WHERE session_id IS NOT NULL ORDER BY id"
        ) {
            Ok(s) => s,
            Err(_) => return Vec::new(),
        };
        stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
            .map(|rows| rows.filter_map(|r| r.ok()).collect())
            .unwrap_or_default()
    }).await;

    for entry in &mut entries {
        // Find closest bot_run within 60s of log file start
        for &(run_started_ms, sid) in &runs {
            let diff = (entry.start_ms as i64 - run_started_ms).unsigned_abs();
            if diff < 60_000 {
                entry.session_id = Some(sid);
                break;
            }
        }
    }

    // Sort newest first
    entries.sort_by_key(|b| std::cmp::Reverse(b.start_ms));
    entries
}

/// List all log files grouped by session.
async fn api_logs_list(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("log.view") { return e.into_response(); }

    let entries = read_log_entries(&state.web_db).await;

    // Group by session_id. Preserve order (newest first within each group).
    let mut sessions: Vec<serde_json::Value> = Vec::new();
    let mut session_map: std::collections::HashMap<i64, usize> = std::collections::HashMap::new();
    let mut ungrouped: Vec<serde_json::Value> = Vec::new();

    for e in &entries {
        let log_json = serde_json::json!({
            "name": e.name,
            "start_ms": e.start_ms,
            "end_ms": e.end_ms,
            "size": e.size,
            "is_current": e.is_current,
        });

        if let Some(sid) = e.session_id {
            if let Some(&idx) = session_map.get(&sid) {
                sessions[idx]["logs"].as_array_mut().unwrap().push(log_json);
                // Update session range
                let s = &mut sessions[idx];
                let cur_start = s["start_ms"].as_u64().unwrap_or(u64::MAX);
                let cur_end = s["end_ms"].as_u64().unwrap_or(0);
                if e.start_ms < cur_start { s["start_ms"] = serde_json::json!(e.start_ms); }
                if e.end_ms > cur_end { s["end_ms"] = serde_json::json!(e.end_ms); }
                let cur_size = s["total_size"].as_u64().unwrap_or(0);
                s["total_size"] = serde_json::json!(cur_size + e.size);
                if e.is_current { s["is_current"] = serde_json::json!(true); }
            } else {
                let idx = sessions.len();
                session_map.insert(sid, idx);
                sessions.push(serde_json::json!({
                    "session_id": sid,
                    "start_ms": e.start_ms,
                    "end_ms": e.end_ms,
                    "total_size": e.size,
                    "is_current": e.is_current,
                    "logs": [log_json],
                }));
            }
        } else {
            ungrouped.push(log_json);
        }
    }

    // Sort sessions by start_ms descending (newest first)
    sessions.sort_by(|a, b| {
        let a_ms = a["start_ms"].as_u64().unwrap_or(0);
        let b_ms = b["start_ms"].as_u64().unwrap_or(0);
        b_ms.cmp(&a_ms)
    });

    Json(serde_json::json!({
        "sessions": sessions,
        "ungrouped": ungrouped,
    })).into_response()
}

/// Download a single log file by name.
async fn api_logs_download(session: Session, Path(filename): Path<String>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("log.view") { return e.into_response(); }

    if !filename.starts_with("bot_") || !filename.ends_with(".log")
        || filename.contains('/') || filename.contains('\\') || filename.contains("..")
    {
        return (axum::http::StatusCode::BAD_REQUEST, "Invalid filename").into_response();
    }

    let path = format!("{}/{}", arb_common::constants::LOG_DIR, filename);
    match tokio::fs::read(&path).await {
        Ok(content) => {
            let headers = [
                (axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8"),
                (axum::http::header::CONTENT_DISPOSITION, &format!("attachment; filename=\"{}\"", filename)),
            ];
            (headers, content).into_response()
        }
        Err(_) => (axum::http::StatusCode::NOT_FOUND, "Log file not found").into_response(),
    }
}

/// Download all logs for a specific session (concatenated, oldest first).
async fn api_logs_download_session(session: Session, State(state): State<AppState>, Path(session_id): Path<i64>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("log.view") { return e.into_response(); }

    let entries = read_log_entries(&state.web_db).await;
    let mut files: Vec<&LogFileEntry> = entries.iter()
        .filter(|e| e.session_id == Some(session_id))
        .collect();
    files.sort_by_key(|e| e.start_ms); // oldest first for concatenation

    if files.is_empty() {
        return (axum::http::StatusCode::NOT_FOUND, "No logs for this session").into_response();
    }

    let mut combined = String::new();
    for f in &files {
        if !combined.is_empty() {
            combined.push_str("\n\n");
        }
        if files.len() > 1 {
            combined.push_str("══════════════════════════════════════════════════════════════\n");
            combined.push_str(&format!("  {}\n", f.name));
            combined.push_str("══════════════════════════════════════════════════════════════\n\n");
        }
        if let Ok(content) = tokio::fs::read_to_string(format!("{}/{}", arb_common::constants::LOG_DIR, f.name)).await {
            combined.push_str(&content);
        }
    }

    let filename = format!("session_{}.log", session_id);
    let headers = [
        (axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8"),
        (axum::http::header::CONTENT_DISPOSITION, &format!("attachment; filename=\"{}\"", filename)),
    ];
    (headers, combined).into_response()
}

/// Download all log files concatenated into a single file (oldest first).
async fn api_logs_download_all(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("log.view") { return e.into_response(); }

    let entries = read_log_entries(&state.web_db).await;
    let mut sorted: Vec<&LogFileEntry> = entries.iter().collect();
    sorted.sort_by_key(|e| e.start_ms); // oldest first

    let mut combined = String::new();
    for f in &sorted {
        if !combined.is_empty() {
            combined.push_str("\n\n");
        }
        combined.push_str("══════════════════════════════════════════════════════════════\n");
        combined.push_str(&format!("  {}\n", f.name));
        combined.push_str("══════════════════════════════════════════════════════════════\n\n");
        if let Ok(content) = tokio::fs::read_to_string(format!("{}/{}", arb_common::constants::LOG_DIR, f.name)).await {
            combined.push_str(&content);
        }
    }

    let headers = [
        (axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8"),
        (axum::http::header::CONTENT_DISPOSITION, "attachment; filename=\"arb_bot_all_logs.log\""),
    ];
    (headers, combined).into_response()
}

/// Read merged config: default.toml + local.toml overlay.
fn read_merged_config_table() -> toml::Value {
    let content = std::fs::read_to_string(arb_common::constants::CONFIG_DEFAULT_PATH).unwrap_or_default();
    let mut table: toml::Value = content.parse().unwrap_or(toml::Value::Table(Default::default()));
    arb_common::config::merge_local_overlay(&mut table, arb_common::constants::CONFIG_LOCAL_PATH);
    table
}

/// GET current config (read from toml).
async fn api_get_config(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("settings.view") { return e.into_response(); }
    use super::state::HotConfig;
    let table = read_merged_config_table();

    let s = |section: &str, key: &str| -> Option<&toml::Value> {
        table.get(section)?.get(key)
    };
    let f = |section: &str, key: &str, def: f64| -> f64 {
        s(section, key).and_then(|v| v.as_float().or_else(|| v.as_integer().map(|i| i as f64))).unwrap_or(def)
    };

    let config = HotConfig {
        max_order_size_usd: f("position", "max_order_size_usd", 25.0),
        leverage: s("position", "leverage").and_then(|v| v.as_integer()).unwrap_or(3) as u16,
        max_used_capital_usd: f("position", "max_used_capital_usd", 50.0),
        max_open_positions: s("position", "max_open_positions").and_then(|v| v.as_integer()).unwrap_or(5) as usize,
        min_entry_spread_bps: f("strategy", "min_entry_spread_bps", 35.0),
        include_fees_in_entry: s("strategy", "include_fees_in_entry").and_then(|v| v.as_bool()).unwrap_or(false),
        min_top_level_usd: f("strategy", "min_top_level_usd", 0.0),
        tick_max_spike_bps: f("strategy", "tick_max_spike_bps", 0.0),
        tick_spike_window_ms: f("strategy", "tick_spike_window_ms", 0.0),
        max_trade_divergence_bps: f("strategy", "max_trade_divergence_bps", 0.0),
        max_recv_staleness_secs: f("strategy", "max_recv_staleness_secs", 0.1),
        max_tick_age_ms: f("strategy", "max_tick_age_ms", 100.0),
        signal_cooldown_secs: f("strategy", "signal_cooldown_secs", 3.0),
        exit_mode: s("strategy", "exit_mode").and_then(|v| v.as_str()).unwrap_or("spread").to_string(),
        min_exit_spread_bps: f("strategy", "min_exit_spread_bps", 5.0),
        min_exit_pnl_pct: f("strategy", "min_exit_pnl_pct", 0.3),
        min_exit_checks: s("strategy", "min_exit_checks").and_then(|v| v.as_integer()).unwrap_or(1) as u32,
        include_fees_in_exit: s("strategy", "include_fees_in_exit").and_then(|v| v.as_bool()).unwrap_or(true),
        exchanges: std::collections::HashMap::new(),
        disabled_pairs: Vec::new(),
    };

    // Build exchanges JSON dynamically from [exchanges.*] TOML sections.
    let exchanges = {
        let mut map = serde_json::Map::new();
        if let Some(ex_table) = table.get("exchanges").and_then(|v| v.as_table()) {
            for (name, section) in ex_table {
                if let Some(sec) = section.as_table() {
                    let mut ex_json = serde_json::Map::new();
                    for (key, val) in sec {
                        match val {
                            toml::Value::String(s) => { ex_json.insert(key.clone(), serde_json::Value::String(s.clone())); }
                            toml::Value::Float(f) => { ex_json.insert(key.clone(), serde_json::json!(*f)); }
                            toml::Value::Integer(i) => { ex_json.insert(key.clone(), serde_json::json!(*i)); }
                            toml::Value::Boolean(b) => { ex_json.insert(key.clone(), serde_json::json!(*b)); }
                            toml::Value::Array(arr) => {
                                let json_arr: Vec<serde_json::Value> = arr.iter()
                                    .filter_map(|v| v.as_str().map(|s| serde_json::Value::String(s.to_string())))
                                    .collect();
                                ex_json.insert(key.clone(), serde_json::Value::Array(json_arr));
                            }
                            _ => {}
                        }
                    }
                    map.insert(name.clone(), serde_json::Value::Object(ex_json));
                }
            }
        }
        serde_json::Value::Object(map)
    };
    let mut result = serde_json::to_value(&config).unwrap_or_default();
    result["exchanges"] = exchanges;

    // Add exchange config schemas for dynamic UI rendering
    let schemas: Vec<serde_json::Value> = state.dash.exchange_schemas().iter()
        .map(|(name, fields)| serde_json::json!({ "name": name, "fields": fields }))
        .collect();
    result["exchange_schemas"] = serde_json::json!(schemas);

    // Add exchange name→id mapping + active symbols via the shared web
    // connection. Both reads run inside one spawn_blocking so the lock is
    // taken once.
    let (exchange_ids, active_syms, sym_exchanges) = state.web_db.run(|db| {
        let ids: Vec<serde_json::Value> = db.load_exchange_ids().unwrap_or_default()
            .into_iter()
            .map(|(name, id)| serde_json::json!({"name": name, "id": id}))
            .collect();
        let syms = db.load_active_common_symbols().unwrap_or_default();
        let pairs = db.load_active_symbol_exchanges().unwrap_or_default();
        let mut map: std::collections::BTreeMap<String, Vec<i64>> = std::collections::BTreeMap::new();
        for (sym, ex_id) in pairs {
            map.entry(sym).or_default().push(ex_id);
        }
        (ids, syms, map)
    }).await;
    result["exchange_ids"] = serde_json::json!(exchange_ids);
    result["active_symbols"] = serde_json::json!(active_syms);
    result["symbol_exchanges"] = serde_json::json!(sym_exchanges);

    // Extra global defaults for override pre-fill (not in HotConfig struct)
    result["signal_cooldown_secs"] = serde_json::json!(f("strategy", "signal_cooldown_secs", 3.0));
    result["min_exit_checks"] = serde_json::json!(f("strategy", "min_exit_checks", 1.0) as u32);
    result["max_positions_per_symbol"] = serde_json::json!(
        s("position", "max_positions_per_symbol").and_then(|v| v.as_integer()).unwrap_or(1)
    );

    Json(result).into_response()
}

/// POST new config: apply hot + save to default.toml.
async fn api_set_config(
    session: Session,
    State(state): State<AppState>,
    Json(config): Json<serde_json::Value>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("settings.edit") { return e.into_response(); }

    // Discover exchange names from merged config (authoritative ordering from [exchanges.*])
    let merged = read_merged_config_table();
    let exchange_names: Vec<String> = merged.get("exchanges")
        .and_then(|v| v.as_table())
        .map(|t| t.keys().cloned().collect())
        .unwrap_or_default();

    // Enforce exchange slippage limits before saving
    let mut config = config;
    clamp_exchange_slippage(&mut config, &exchange_names);

    let ex = config.get("exchanges");

    // Fields that cannot be hot-reloaded — changing them in [exchanges.NAME]
    // requires a bot restart. (key, default-if-missing, human-readable label)
    const RESTART_FIELDS: &[(&str, &str, &str)] = &[
        ("margin_mode", "isolated", "margin_mode"),
        ("ws_ob_mode", "per_market", "ws_ob_mode"),
    ];
    // Collect (exchange_name, field_label) pairs for changed restart-required fields.
    let mut restart_changes: Vec<(String, &'static str)> = Vec::new();
    for name in &exchange_names {
        for (field_key, default, label) in RESTART_FIELDS {
            let new_val = ex.and_then(|e| e.get(name))
                .and_then(|v| v.get(field_key))
                .and_then(|v| v.as_str());
            let Some(new_val) = new_val else { continue; };
            let current = merged.get("exchanges")
                .and_then(|t| t.get(name))
                .and_then(|t| t.get(field_key))
                .and_then(|v| v.as_str())
                .unwrap_or(default);
            if new_val != current {
                restart_changes.push((name.clone(), label));
            }
        }
    }

    // Save full config (strategy + exchange) to toml
    if let Err(e) = save_config_to_toml(&config).await {
        return Json(serde_json::json!({"ok": false, "error": e.to_string()})).into_response();
    }

    // Build HotExchangeConfig from the incoming JSON for each exchange.
    let build_hot_exchange = |name: &str| -> super::state::HotExchangeConfig {
        let v = ex.and_then(|e| e.get(name));
        let parse_slip = |val: Option<&serde_json::Value>, field: &str| -> Option<crate::config::SlippagePct> {
            let v = val?.get(field)?;
            if let Some(s) = v.as_str() {
                if s.eq_ignore_ascii_case("max") { return Some(crate::config::SlippagePct::Max); }
                s.parse::<f64>().ok().map(crate::config::SlippagePct::Fixed)
            } else {
                v.as_f64().map(crate::config::SlippagePct::Fixed)
            }
        };
        super::state::HotExchangeConfig {
            enabled: v.and_then(|v| v["enabled"].as_bool()),
            entry_order_type: v.and_then(|v| v["entry_order_type"].as_str()).map(String::from),
            entry_slippage_pct: parse_slip(v, "entry_slippage_pct"),
            exit_order_type: v.and_then(|v| v["exit_order_type"].as_str()).map(String::from),
            exit_slippage_pct: parse_slip(v, "exit_slippage_pct"),
            emergency_slippage_pct: parse_slip(v, "emergency_slippage_pct"),
            stop_loss_type: v.and_then(|v| v["stop_loss_type"].as_str()).map(String::from),
            allowed_with: v.and_then(|v| v["allowed_with"].as_array()).map(|arr| {
                arr.iter().filter_map(|e| e.as_str().map(String::from)).collect()
            }),
        }
    };

    let mut hot_exchanges = std::collections::HashMap::new();
    for name in &exchange_names {
        hot_exchanges.insert(name.clone(), build_hot_exchange(name));
    }

    // Extract hot config fields for engine live-reload
    let hot = super::state::HotConfig {
        max_order_size_usd: config["max_order_size_usd"].as_f64().unwrap_or(25.0),
        leverage: config["leverage"].as_u64().unwrap_or(3) as u16,
        max_used_capital_usd: config["max_used_capital_usd"].as_f64().unwrap_or(50.0),
        max_open_positions: config["max_open_positions"].as_u64().unwrap_or(5) as usize,
        min_entry_spread_bps: config["min_entry_spread_bps"].as_f64().unwrap_or(35.0),
        include_fees_in_entry: config["include_fees_in_entry"].as_bool().unwrap_or(false),
        min_top_level_usd: config["min_top_level_usd"].as_f64().unwrap_or(0.0),
        tick_max_spike_bps: config["tick_max_spike_bps"].as_f64().unwrap_or(0.0),
        tick_spike_window_ms: config["tick_spike_window_ms"].as_f64().unwrap_or(0.0),
        max_trade_divergence_bps: config["max_trade_divergence_bps"].as_f64().unwrap_or(0.0),
        max_recv_staleness_secs: config["max_recv_staleness_secs"].as_f64().unwrap_or(0.1),
        max_tick_age_ms: config["max_tick_age_ms"].as_f64().unwrap_or(100.0),
        signal_cooldown_secs: config["signal_cooldown_secs"].as_f64().unwrap_or(3.0),
        exit_mode: config["exit_mode"].as_str().unwrap_or("spread").to_string(),
        min_exit_spread_bps: config["min_exit_spread_bps"].as_f64().unwrap_or(5.0),
        min_exit_pnl_pct: config["min_exit_pnl_pct"].as_f64().unwrap_or(0.3),
        min_exit_checks: config["min_exit_checks"].as_u64().unwrap_or(1) as u32,
        include_fees_in_exit: config["include_fees_in_exit"].as_bool().unwrap_or(true),
        exchanges: hot_exchanges,
        disabled_pairs: config["disabled_pairs"].as_array()
            .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
            .unwrap_or_default(),
    };
    let version = state.dash.set_hot_config(hot);
    tracing::info!(version, user = %session.username, "Config updated via web UI");
    if !restart_changes.is_empty() {
        let parts: Vec<String> = restart_changes.iter()
            .map(|(name, label)| format!("{name}.{label}"))
            .collect();
        tracing::info!(changes = ?parts, "restart-required field changed in config");
        Json(serde_json::json!({
            "ok": true, "version": version,
            "restart_required": true,
            "restart_reason": format!("{} changed — requires restart to apply", parts.join(", "))
        })).into_response()
    } else {
        Json(serde_json::json!({"ok": true, "version": version})).into_response()
    }
}

/// Clamp numeric slippage fields for Extended to the hard exchange limit.
/// "max" values are not clamped — they resolve per-symbol at order time.
fn clamp_exchange_slippage(config: &mut serde_json::Value, exchange_names: &[String]) {
    use crate::constants::EXTENDED_MAX_SLIPPAGE_PCT;
    if !exchange_names.iter().any(|n| n == "extended") { return; }
    let fields = ["entry_slippage_pct", "exit_slippage_pct", "emergency_slippage_pct"];
    if let Some(ext) = config.pointer_mut("/exchanges/extended") {
        for field in &fields {
            // Skip "max" (string) values
            if ext.get(*field).and_then(|v| v.as_str()).map(|s| s.eq_ignore_ascii_case("max")).unwrap_or(false) {
                continue;
            }
            if let Some(val) = ext.get(*field).and_then(|v| v.as_f64())
                && val > EXTENDED_MAX_SLIPPAGE_PCT {
                    tracing::warn!(
                        field, configured = val, clamped = EXTENDED_MAX_SLIPPAGE_PCT,
                        "Extended slippage exceeds min exchange limit, clamped"
                    );
                    ext[*field] = serde_json::json!(EXTENDED_MAX_SLIPPAGE_PCT);
                }
        }
    }
}

async fn save_config_to_toml(config: &serde_json::Value) -> anyhow::Result<()> {
    let local_path = arb_common::constants::CONFIG_LOCAL_PATH;

    // Load existing local.toml as TOML table, or start from default.toml
    let mut table: toml::Table = if tokio::fs::try_exists(local_path).await.unwrap_or(false) {
        let content = tokio::fs::read_to_string(local_path).await?;
        content.parse().unwrap_or_default()
    } else {
        let content = tokio::fs::read_to_string(arb_common::constants::CONFIG_DEFAULT_PATH).await
            .unwrap_or_default();
        content.parse().unwrap_or_default()
    };

    // Helper: ensure section exists and set a value in it
    let set = |table: &mut toml::Table, section: &str, key: &str, val: toml::Value| {
        let entry = table.entry(section)
            .or_insert_with(|| toml::Value::Table(toml::Table::new()));
        if let Some(section_table) = entry.as_table_mut() {
            section_table.insert(key.to_string(), val);
        }
    };

    // Strategy
    set(&mut table, "strategy", "min_entry_spread_bps",
        toml::Value::Float(config["min_entry_spread_bps"].as_f64().unwrap_or(35.0)));
    set(&mut table, "strategy", "exit_mode",
        toml::Value::String(config["exit_mode"].as_str().unwrap_or("spread").to_string()));
    set(&mut table, "strategy", "min_exit_spread_bps",
        toml::Value::Float(config["min_exit_spread_bps"].as_f64().unwrap_or(5.0)));
    set(&mut table, "strategy", "min_exit_pnl_pct",
        toml::Value::Float(config["min_exit_pnl_pct"].as_f64().unwrap_or(0.3)));
    set(&mut table, "strategy", "include_fees_in_entry",
        toml::Value::Boolean(config["include_fees_in_entry"].as_bool().unwrap_or(false)));
    set(&mut table, "strategy", "include_fees_in_exit",
        toml::Value::Boolean(config["include_fees_in_exit"].as_bool().unwrap_or(true)));
    set(&mut table, "strategy", "min_top_level_usd",
        toml::Value::Float(config["min_top_level_usd"].as_f64().unwrap_or(0.0)));
    set(&mut table, "strategy", "tick_max_spike_bps",
        toml::Value::Float(config["tick_max_spike_bps"].as_f64().unwrap_or(0.0)));
    set(&mut table, "strategy", "tick_spike_window_ms",
        toml::Value::Float(config["tick_spike_window_ms"].as_f64().unwrap_or(0.0)));
    set(&mut table, "strategy", "max_trade_divergence_bps",
        toml::Value::Float(config["max_trade_divergence_bps"].as_f64().unwrap_or(0.0)));
    set(&mut table, "strategy", "max_recv_staleness_secs",
        toml::Value::Float(config["max_recv_staleness_secs"].as_f64().unwrap_or(0.1)));
    set(&mut table, "strategy", "max_tick_age_ms",
        toml::Value::Float(config["max_tick_age_ms"].as_f64().unwrap_or(100.0)));
    set(&mut table, "strategy", "signal_cooldown_secs",
        toml::Value::Float(config["signal_cooldown_secs"].as_f64().unwrap_or(3.0)));
    set(&mut table, "strategy", "min_exit_checks",
        toml::Value::Integer(config["min_exit_checks"].as_u64().unwrap_or(1) as i64));

    // Position
    set(&mut table, "position", "leverage",
        toml::Value::Integer(config["leverage"].as_u64().unwrap_or(3) as i64));
    set(&mut table, "position", "max_order_size_usd",
        toml::Value::Float(config["max_order_size_usd"].as_f64().unwrap_or(25.0)));
    set(&mut table, "position", "max_used_capital_usd",
        toml::Value::Float(config["max_used_capital_usd"].as_f64().unwrap_or(50.0)));
    set(&mut table, "position", "max_open_positions",
        toml::Value::Integer(config["max_open_positions"].as_u64().unwrap_or(5) as i64));

    // Exchange settings — iterate all exchanges from the incoming JSON
    if let Some(exchanges_json) = config.get("exchanges").and_then(|e| e.as_object()) {
        // Write exchange settings under [exchanges.NAME]
        let exchanges_table = table.entry("exchanges".to_string())
            .or_insert_with(|| toml::Value::Table(toml::Table::new()));
        if let Some(ex_root) = exchanges_table.as_table_mut() {
            for (name, ex_cfg) in exchanges_json {
                if let Some(obj) = ex_cfg.as_object() {
                    let ex_section = ex_root.entry(name.clone())
                        .or_insert_with(|| toml::Value::Table(toml::Table::new()));
                    if let Some(sec) = ex_section.as_table_mut() {
                        for (key, val) in obj {
                            let toml_val = match val {
                                serde_json::Value::String(s) => Some(toml::Value::String(s.clone())),
                                serde_json::Value::Number(n) => {
                                    if let Some(f) = n.as_f64() {
                                        // Preserve integer representation for whole numbers
                                        if f.fract() == 0.0 && f.abs() < i64::MAX as f64 {
                                            Some(toml::Value::Integer(f as i64))
                                        } else {
                                            Some(toml::Value::Float(f))
                                        }
                                    } else {
                                        None
                                    }
                                }
                                serde_json::Value::Bool(b) => Some(toml::Value::Boolean(*b)),
                                serde_json::Value::Array(arr) => {
                                    let toml_arr: Vec<toml::Value> = arr.iter()
                                        .filter_map(|v| v.as_str().map(|s| toml::Value::String(s.to_string())))
                                        .collect();
                                    Some(toml::Value::Array(toml_arr))
                                }
                                _ => None,
                            };
                            if let Some(tv) = toml_val {
                                sec.insert(key.clone(), tv);
                            }
                        }
                    }
                }
            }
        }

    }

    tokio::fs::write(local_path, toml::to_string_pretty(&table)?).await?;
    Ok(())
}

// ── Exchange credentials editing ────────────────────────────────────────────
//
// Reads and writes are scoped to the on-disk `.env` file (NOT process env),
// so the UI always reflects the authoritative source that the bot will read
// on its next start. Save flow is serialized with a tokio mutex to make
// concurrent admin edits safe, and rate-limited per session.

/// Max credential endpoint requests per session per minute.
const CREDENTIALS_RATE_LIMIT: usize = 10;
const CREDENTIALS_RATE_WINDOW_SECS: u64 = 60;

#[allow(clippy::result_large_err)] // Err is an axum Response returned by-value on the error path; boxing it would add an alloc to every early-return
fn credentials_rate_check(state: &AppState, user_id: i64) -> Result<(), axum::response::Response> {
    let now = std::time::Instant::now();
    let window = std::time::Duration::from_secs(CREDENTIALS_RATE_WINDOW_SECS);
    let mut map = match state.credentials_rate.lock() {
        Ok(m) => m,
        Err(_) => return Err((
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"ok": false, "error": "rate limiter poisoned"})),
        ).into_response()),
    };
    let entry = map.entry(user_id).or_default();
    entry.retain(|t| now.duration_since(*t) < window);
    if entry.len() >= CREDENTIALS_RATE_LIMIT {
        return Err((
            axum::http::StatusCode::TOO_MANY_REQUESTS,
            Json(serde_json::json!({
                "ok": false,
                "error": format!("rate limit: max {CREDENTIALS_RATE_LIMIT} credential requests per {CREDENTIALS_RATE_WINDOW_SECS}s"),
            })),
        ).into_response());
    }
    entry.push(now);
    Ok(())
}

/// Count "busy" conditions that should block credential rotation for the
/// given exchange. Returns (per_exchange_count, global_opening_inflight).
/// Callers that need to block for ANY exchange should check both.
fn count_exchange_busy(snap: &super::state::DashSnapshot, exchange_id: &str) -> (usize, usize) {
    let mut per = 0usize;
    for t in &snap.positions {
        if t.exchange_1 == exchange_id || t.exchange_2 == exchange_id { per += 1; }
    }
    for o in &snap.active_orders {
        if o.exchange == exchange_id { per += 1; }
    }
    // OpeningSnap has no exchange fields, so we can't attribute in-flight
    // opens to a specific exchange — treat them as a global block.
    let opening = snap.opening_trades.len();
    (per, opening)
}

/// GET /api/credentials/schemas — field layout + metadata for the Credentials UI.
async fn api_credentials_schemas(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("credentials.edit") { return e.into_response(); }
    if let Err(r) = credentials_rate_check(&state, session.user_id) { return r; }
    Json(serde_json::json!({ "schemas": super::credentials::schemas() })).into_response()
}

/// GET /api/credentials — presence + last-4 preview for every exchange,
/// plus per-exchange "busy" count so the UI can disable Edit.
async fn api_credentials_get(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("credentials.edit") { return e.into_response(); }
    if let Err(r) = credentials_rate_check(&state, session.user_id) { return r; }

    let env_path = std::path::Path::new(super::credentials::ENV_FILE);
    let snap = state.dash.read();
    let opening_inflight = snap.opening_trades.len();

    let schemas = super::credentials::schemas();
    let mut out = Vec::with_capacity(schemas.len());
    for sch in &schemas {
        let secrets = super::credentials::read_env_secrets_from_file(env_path, sch.exchange_id);
        let mut any_set = false;
        let mut fields_json = Vec::with_capacity(sch.fields.len());
        for f in &sch.fields {
            let raw = secrets.peek(&f.key).unwrap_or("");
            let set = !raw.is_empty();
            if set { any_set = true; }
            let preview = match f.kind {
                super::credentials::FieldKind::Secret => super::credentials::mask_secret(raw),
                _ => raw.to_string(),
            };
            fields_json.push(serde_json::json!({
                "key": f.key,
                "env_name": f.env_name,
                "label": f.label,
                "kind": f.kind,
                "required": f.required,
                "set": set,
                "preview": preview,
            }));
        }
        let (per_exchange, _) = count_exchange_busy(&snap, sch.exchange_id);
        let can_edit = per_exchange == 0 && opening_inflight == 0;
        out.push(serde_json::json!({
            "exchange_id": sch.exchange_id,
            "display_name": sch.display_name,
            "configured": any_set,
            "fields": fields_json,
            "open_positions": per_exchange,
            "opening_inflight": opening_inflight,
            "can_edit": can_edit,
        }));
    }

    Json(serde_json::json!({ "ok": true, "exchanges": out })).into_response()
}

#[derive(serde::Deserialize)]
struct CredentialsRequest {
    /// Map of `FIELD_KEY` (prefix-stripped, as in schema) → value.
    /// Empty values are ignored (= "leave as-is").
    fields: std::collections::HashMap<String, String>,
    #[serde(default)]
    restart: bool,
}

/// POST /api/credentials/{exchange_id}/test — validate candidate
/// credentials against the live exchange without persisting anything.
async fn api_credentials_test(
    session: Session,
    State(state): State<AppState>,
    axum::extract::Path(exchange_id): axum::extract::Path<String>,
    Json(body): Json<CredentialsRequest>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("credentials.edit") { return e.into_response(); }
    if let Err(r) = credentials_rate_check(&state, session.user_id) { return r; }

    let Some(schema) = super::credentials::schema_for(&exchange_id) else {
        return (axum::http::StatusCode::NOT_FOUND, Json(serde_json::json!({"ok": false, "error": "unknown exchange"}))).into_response();
    };

    let overrides = match allowlist_fields(&schema, body.fields) {
        Ok(m) => m,
        Err(e) => return (axum::http::StatusCode::BAD_REQUEST, Json(serde_json::json!({"ok": false, "error": e}))).into_response(),
    };

    let env_path = std::path::Path::new(super::credentials::ENV_FILE);
    match super::credentials::validate_credentials(env_path, &exchange_id, overrides).await {
        Ok(ok) => Json(serde_json::json!({"ok": true, "summary": ok.summary})).into_response(),
        Err(err) => {
            tracing::warn!(
                exchange = %exchange_id,
                user = %session.username,
                user_id = session.user_id,
                internal = %err.internal,
                "credential test failed",
            );
            Json(serde_json::json!({"ok": false, "error": err.external})).into_response()
        }
    }
}

/// POST /api/credentials/{exchange_id} — validate, persist to `.env`, and
/// (optionally) restart the bot. Refuses to write while the exchange has
/// any open positions OR any in-flight openings anywhere. Serialized via
/// `AppState::credentials_write_lock`.
async fn api_credentials_save(
    session: Session,
    State(state): State<AppState>,
    axum::extract::Path(exchange_id): axum::extract::Path<String>,
    Json(body): Json<CredentialsRequest>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("credentials.edit") { return e.into_response(); }
    if let Err(r) = credentials_rate_check(&state, session.user_id) { return r; }

    let Some(schema) = super::credentials::schema_for(&exchange_id) else {
        return (axum::http::StatusCode::NOT_FOUND, Json(serde_json::json!({"ok": false, "error": "unknown exchange"}))).into_response();
    };

    // Serialize with other credential saves across all admin sessions.
    let _write_guard = state.credentials_write_lock.lock().await;

    // ── 1. Pause engine BEFORE checking busy state ────────────────────────
    // Otherwise the engine could open a new position between our check and
    // our write. request_pause() is idempotent and cheap; if we bail out
    // below without restarting, we must resume it in the error path.
    state.dash.request_pause();

    // Give any in-flight open logic a tick to settle before we read snap.
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    // ── 2. Allowlist + non-empty filter ───────────────────────────────────
    let overrides = match allowlist_fields(&schema, body.fields) {
        Ok(m) => m,
        Err(e) => {
            state.dash.request_resume();
            return (axum::http::StatusCode::BAD_REQUEST, Json(serde_json::json!({"ok": false, "error": e}))).into_response();
        }
    };
    if overrides.is_empty() {
        state.dash.request_resume();
        return Json(serde_json::json!({"ok": false, "error": "no fields to update"})).into_response();
    }

    // ── 3. Busy check — per-exchange positions/orders AND global openings ──
    let snap = state.dash.read();
    let (per_exchange, opening_inflight) = count_exchange_busy(&snap, &exchange_id);
    if per_exchange > 0 {
        state.dash.request_resume();
        return (
            axum::http::StatusCode::CONFLICT,
            Json(serde_json::json!({
                "ok": false,
                "error": format!("exchange has {per_exchange} open position/order(s); close them first"),
            })),
        ).into_response();
    }
    if opening_inflight > 0 {
        state.dash.request_resume();
        return (
            axum::http::StatusCode::CONFLICT,
            Json(serde_json::json!({
                "ok": false,
                "error": format!("{opening_inflight} trade(s) in-flight (one leg filled, other pending); retry in a few seconds"),
            })),
        ).into_response();
    }
    drop(snap);

    // ── 4. Required-fields check against `.env` (not process env) ─────────
    let env_path = std::path::Path::new(super::credentials::ENV_FILE);
    let current = super::credentials::read_env_secrets_from_file(env_path, &exchange_id);
    for field in &schema.fields {
        if !field.required { continue; }
        let will_be_set = overrides.get(&field.key).map(|v| !v.is_empty()).unwrap_or(false)
            || current.peek(&field.key).map(|v| !v.is_empty()).unwrap_or(false);
        if !will_be_set {
            state.dash.request_resume();
            return (
                axum::http::StatusCode::BAD_REQUEST,
                Json(serde_json::json!({"ok": false, "error": format!("missing required field: {}", field.env_name)})),
            ).into_response();
        }
    }
    drop(current);

    // ── 5. Validate against the live exchange ────────────────────────────
    let resolved = match super::credentials::validate_credentials(env_path, &exchange_id, overrides.clone()).await {
        Ok(ok) => ok.resolved,
        Err(err) => {
            tracing::warn!(
                exchange = %exchange_id,
                user = %session.username,
                user_id = session.user_id,
                internal = %err.internal,
                "credentials save rejected — validation failed",
            );
            state.dash.request_resume();
            return (
                axum::http::StatusCode::BAD_REQUEST,
                Json(serde_json::json!({"ok": false, "error": format!("validation failed: {}", err.external)})),
            ).into_response();
        }
    };

    // ── 6. Re-check busy state AFTER validation (which took seconds) ─────
    // Engine is paused so no new opens, but we want to catch the very
    // narrow case of an in-flight opening that landed after our step 3
    // check.
    let snap2 = state.dash.read();
    let (per2, opening2) = count_exchange_busy(&snap2, &exchange_id);
    if per2 > 0 || opening2 > 0 {
        state.dash.request_resume();
        return (
            axum::http::StatusCode::CONFLICT,
            Json(serde_json::json!({
                "ok": false,
                "error": "exchange became busy during validation; retry",
            })),
        ).into_response();
    }
    drop(snap2);

    // ── 7. Write .env atomically ─────────────────────────────────────────
    // Merge rules:
    //   1. Non-empty user overrides are authoritative — they win over everything.
    //   2. Auto-resolved fields (e.g. Extended's derived STARK_PUBLIC_KEY,
    //      fetched VAULT_ID, Zero1's auto-picked ACCOUNT_ID) fill in fields
    //      the user left blank so they're persisted to .env after save.
    //   3. Empty overrides are ignored (do NOT clear existing .env values) —
    //      blank means "I didn't fill this in", not "wipe it". If resolved
    //      has a value for that field, it still gets written.
    let mut env_updates: std::collections::HashMap<String, String> =
        std::collections::HashMap::with_capacity(overrides.len() + resolved.len());
    for (k, v) in &overrides {
        if v.is_empty() { continue; }
        if let Some(field) = schema.find_field(k) {
            env_updates.insert(field.env_name.clone(), v.clone());
        }
    }
    for (k, v) in &resolved {
        if let Some(field) = schema.find_field(k) {
            env_updates.entry(field.env_name.clone()).or_insert_with(|| v.clone());
        }
    }
    if let Err(e) = super::credentials::write_env_file(env_path, &env_updates) {
        state.dash.request_resume();
        return Json(serde_json::json!({"ok": false, "error": format!("write .env: {e}")})).into_response();
    }

    // ── 8. Audit log (no values, structured) ──────────────────────────────
    let mut changed_keys: Vec<String> = env_updates.keys().cloned().collect();
    changed_keys.sort();
    tracing::warn!(
        event = "credentials_changed",
        exchange = %exchange_id,
        fields_changed = ?changed_keys,
        user = %session.username,
        user_id = session.user_id,
        ts_rfc3339 = %chrono::Utc::now().to_rfc3339(),
        "credentials updated via web UI",
    );

    // ── 9. Optional restart ──────────────────────────────────────────────
    let mut restarted = false;
    if body.restart {
        state.dash.request_shutdown();
        let args = ["bash".to_string(),
            "tools/bot_start.sh".to_string(),
            "--no-build".to_string(),
            "--resume".to_string()];
        let (stdout_cfg, stderr_cfg) = match std::fs::OpenOptions::new()
            .create(true).append(true).open("logs/restart.log")
        {
            Ok(f) => {
                let f2 = f.try_clone().unwrap_or_else(|_| std::fs::File::create("/dev/null").unwrap());
                (std::process::Stdio::from(f), std::process::Stdio::from(f2))
            }
            Err(_) => (std::process::Stdio::null(), std::process::Stdio::null()),
        };
        match std::process::Command::new(&args[0])
            .args(&args[1..])
            .stdin(std::process::Stdio::null())
            .stdout(stdout_cfg)
            .stderr(stderr_cfg)
            .spawn()
        {
            Ok(_) => {
                restarted = true;
                tracing::info!("Restart script spawned after credentials update (log: logs/restart.log)");
            }
            Err(e) => tracing::error!(error = %e, "Failed to spawn restart script after credentials update"),
        }
    } else {
        // Saved but no restart requested — resume trading.
        state.dash.request_resume();
    }

    Json(serde_json::json!({"ok": true, "restarted": restarted})).into_response()
}

/// Reject any field key not present in the schema; drop empty values.
fn allowlist_fields(
    schema: &super::credentials::ExchangeCredentialSchema,
    input: std::collections::HashMap<String, String>,
) -> Result<std::collections::HashMap<String, String>, String> {
    let mut out = std::collections::HashMap::new();
    for (k, v) in input {
        if schema.find_field(&k).is_none() {
            return Err(format!("unknown field: {k}"));
        }
        if v.is_empty() { continue; }
        out.insert(k, v);
    }
    Ok(out)
}

/// Close all positions (bot stays running, just closes everything).
async fn api_close_all(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("trades.close_all") { return e.into_response(); }
    tracing::warn!(user = %session.username, "Close-all requested via web UI");
    state.dash.request_close_all();
    Json(serde_json::json!({"ok": true, "message": "Closing all positions..."})).into_response()
}

/// Close a single trade by ID (both legs on both exchanges).
async fn api_close_trade(
    session: Session,
    State(state): State<AppState>,
    axum::extract::Path(trade_id): axum::extract::Path<String>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("trades.close") { return e.into_response(); }
    tracing::warn!(trade_id = %trade_id, user = %session.username, "Close single trade requested via web UI");
    state.dash.request_close_trade(trade_id.clone());
    Json(serde_json::json!({"ok": true, "message": format!("Closing trade {trade_id}...")})).into_response()
}

/// Pause trading (manual halt via web UI).
async fn api_halt_pause(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.pause") { return e.into_response(); }
    tracing::warn!(user = %session.username, "Manual pause requested via web UI");
    state.dash.request_pause();
    Json(serde_json::json!({"ok": true, "message": "Trading paused"})).into_response()
}

/// Resume trading (manual halt only). Returns remaining halts if any.
async fn api_halt_resume(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.pause") { return e.into_response(); }
    tracing::info!(user = %session.username, "Manual resume requested via web UI");
    let remaining = state.dash.request_resume();
    if remaining.is_empty() {
        Json(serde_json::json!({"ok": true, "message": "Trading resumed"})).into_response()
    } else {
        let problems: Vec<String> = remaining.iter().map(|(r, d)| format!("{}: {}", r, d)).collect();
        Json(serde_json::json!({
            "ok": true,
            "message": "Manual pause removed, but trading is still halted",
            "remaining_halts": problems
        })).into_response()
    }
}

/// Resume after permanent API errors halt. Resets cycle counter.
async fn api_halt_resume_api(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.pause") { return e.into_response(); }
    tracing::warn!(user = %session.username, "Manual API error halt resume via web UI");
    let remaining = state.dash.request_resume_api_halt();
    if remaining.is_empty() {
        Json(serde_json::json!({"ok": true, "message": "API error halt cleared, trading resumed"})).into_response()
    } else {
        let problems: Vec<String> = remaining.iter().map(|(r, d)| format!("{}: {}", r, d)).collect();
        Json(serde_json::json!({
            "ok": true,
            "message": "API halt cleared, but other halts remain",
            "remaining_halts": problems
        })).into_response()
    }
}

/// Shutdown bot: stops the engine and exits the process without restarting.
async fn api_shutdown(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.restart") { return e.into_response(); }
    tracing::warn!(user = %session.username, "Bot SHUTDOWN requested via web UI — process will exit");
    state.dash.request_pause();
    state.dash.request_shutdown();
    Json(serde_json::json!({"ok": true, "message": "Shutting down — process will exit without restart"})).into_response()
}

// ─── Config sync API (push to remote servers) ────────────────────────────────

/// GET /api/servers — list configured remote servers from [servers] in local.toml.
async fn api_get_servers(session: Session) -> impl IntoResponse {
    if let Err(e) = session.require_permission("settings.view") { return e.into_response(); }
    let servers = parse_servers_from_toml();
    Json(serde_json::json!({"servers": servers})).into_response()
}

/// Parse [servers.*] section from local.toml.
fn parse_servers_from_toml() -> Vec<serde_json::Value> {
    let path = arb_common::constants::CONFIG_LOCAL_PATH;
    let content = std::fs::read_to_string(path).unwrap_or_default();
    let table: toml::Table = content.parse().unwrap_or_default();
    let Some(servers) = table.get("servers").and_then(|v| v.as_table()) else {
        return Vec::new();
    };
    servers.iter().map(|(key, val)| {
        let t = val.as_table();
        serde_json::json!({
            "id": key,
            "name": t.and_then(|t| t.get("name")).and_then(|v| v.as_str()).unwrap_or(key),
            "host": t.and_then(|t| t.get("host")).and_then(|v| v.as_str()).unwrap_or(""),
        })
    }).collect()
}

/// GET /api/config/snapshot — full config snapshot: global config (TOML sections) + symbol overrides + manual blocks.
async fn api_config_snapshot(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("settings.view") { return e.into_response(); }

    // 1. Read local.toml as raw TOML string (we'll push this file as-is)
    let local_toml = tokio::fs::read_to_string(arb_common::constants::CONFIG_LOCAL_PATH)
        .await.unwrap_or_default();

    // 2 + 3. Symbol overrides + manual blocks via shared web connection.
    let (overrides, blocks) = state.web_db.run(|db| {
        let overrides = db.load_symbol_overrides().unwrap_or_default();
        let blocks: Vec<serde_json::Value> = db.load_active_blocks().unwrap_or_default()
            .into_iter()
            .filter(|(_, b)| matches!(b.block_type, crate::storage::BlockType::Manual | crate::storage::BlockType::Unprofitable))
            .map(|(_, b)| serde_json::json!({
                "symbol": b.symbol,
                "exchange_id": b.exchange_id,
                "exchange_id_2": b.exchange_id_2,
                "reason": b.reason,
            }))
            .collect();
        (overrides, blocks)
    }).await;

    Json(serde_json::json!({
        "local_toml": local_toml,
        "overrides": overrides,
        "manual_blocks": blocks,
    })).into_response()
}

/// One symbol block row prepared for push to a remote server:
/// (symbol, exchange_id, exchange_id_2, block_type, reason).
type PushBlockRow = (String, Option<i64>, Option<i64>, String, String);

/// POST /api/config/push — push config snapshot to a remote server via SSH.
async fn api_config_push(
    session: Session,
    State(state): State<AppState>,
    Json(body): Json<serde_json::Value>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("settings.edit") { return e.into_response(); }

    let server_id = body["server_id"].as_str().unwrap_or("");
    if server_id.is_empty() {
        return Json(serde_json::json!({"ok": false, "error": "server_id required"})).into_response();
    }

    // Look up server config
    let path = arb_common::constants::CONFIG_LOCAL_PATH;
    let content = std::fs::read_to_string(path).unwrap_or_default();
    let table: toml::Table = content.parse().unwrap_or_default();
    let srv = table.get("servers")
        .and_then(|v| v.as_table())
        .and_then(|s| s.get(server_id))
        .and_then(|v| v.as_table());
    let Some(srv) = srv else {
        return Json(serde_json::json!({"ok": false, "error": format!("Server '{server_id}' not found in [servers]")})).into_response();
    };
    let host = srv.get("host").and_then(|v| v.as_str()).unwrap_or("");
    let user = srv.get("user").and_then(|v| v.as_str()).unwrap_or("ubuntu");
    let ssh_key = srv.get("ssh_key").and_then(|v| v.as_str()).unwrap_or("");
    let remote_dir = srv.get("remote_dir").and_then(|v| v.as_str()).unwrap_or("arbitrage_bot");
    let db_path_remote = srv.get("db_path").and_then(|v| v.as_str()).unwrap_or("db/arb_bot.db");

    if host.is_empty() || ssh_key.is_empty() {
        return Json(serde_json::json!({"ok": false, "error": "Server config incomplete: host and ssh_key required"})).into_response();
    }

    // Expand ~ in ssh_key path
    let ssh_key_expanded = if ssh_key.starts_with('~') {
        if let Ok(home) = std::env::var("HOME") {
            ssh_key.replacen('~', &home, 1)
        } else { ssh_key.to_string() }
    } else { ssh_key.to_string() };

    // Sections to sync (strategy-related only). Everything else stays as-is on remote.
    const SYNC_SECTIONS: &[&str] = &["strategy", "position", "exchanges", "risk"];

    let local_toml_content = tokio::fs::read_to_string(arb_common::constants::CONFIG_LOCAL_PATH)
        .await.unwrap_or_default();
    let local_table: toml::Table = local_toml_content.parse().unwrap_or_default();

    // Read overrides + manual/unprofitable blocks via shared web connection.
    // Skip leverage/qty_min/liquidity blocks — those are auto-detected per environment.
    let (overrides, push_blocks) = state.web_db.run(|db| {
        let overrides = db.load_symbol_overrides().unwrap_or_default();
        let blocks: Vec<PushBlockRow> = db.load_active_blocks()
            .unwrap_or_default()
            .into_iter()
            .filter(|(_, b)| matches!(b.block_type, crate::storage::BlockType::Manual | crate::storage::BlockType::Unprofitable))
            .map(|(_, b)| (b.symbol, b.exchange_id, b.exchange_id_2, b.block_type.as_str().to_string(), b.reason))
            .collect();
        (overrides, blocks)
    }).await;

    let ssh_base = format!(
        "ssh -i '{}' -o StrictHostKeyChecking=no -o ConnectTimeout=10 -o BatchMode=yes {}@{}",
        ssh_key_expanded, user, host
    );

    // Helper: run SSH command with timeout
    let ssh_timeout = Duration::from_secs(30);
    let run_ssh = |cmd: String| async move {
        let child = tokio::process::Command::new("bash")
            .arg("-c").arg(&cmd)
            .output();
        match tokio::time::timeout(ssh_timeout, child).await {
            Ok(Ok(out)) => Ok(out),
            Ok(Err(e)) => Err(format!("SSH process error: {e}")),
            Err(_) => Err("SSH timeout (30s)".to_string()),
        }
    };

    tracing::info!(server = %server_id, user = %session.username, "Pushing config to remote server");

    // Step 1: Read remote local.toml, merge strategy sections, write back
    let read_cmd = format!("{ssh_base} 'cat {remote_dir}/config/local.toml 2>/dev/null || true'");
    let remote_toml_content = match run_ssh(read_cmd).await {
        Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout).to_string(),
        Ok(_) => {
            // File might not exist yet — that's OK, start from empty
            tracing::info!("Remote local.toml not found or empty, will create");
            String::new()
        }
        Err(e) => return Json(serde_json::json!({"ok": false, "error": format!("SSH read failed: {e}")})).into_response(),
    };

    let mut remote_table: toml::Table = remote_toml_content.parse().unwrap_or_default();

    // Merge only SYNC_SECTIONS from local into remote
    for section in SYNC_SECTIONS {
        if let Some(val) = local_table.get(*section) {
            remote_table.insert(section.to_string(), val.clone());
        }
    }

    let push_toml = toml::to_string_pretty(&remote_table).unwrap_or_default();
    let toml_cmd = format!(
        "{ssh_base} 'cat > {remote_dir}/config/local.toml' <<'TOML_EOF'\n{push_toml}\nTOML_EOF"
    );
    match run_ssh(toml_cmd).await {
        Err(e) => return Json(serde_json::json!({"ok": false, "error": e})).into_response(),
        Ok(out) if !out.status.success() => {
            let stderr = String::from_utf8_lossy(&out.stderr);
            return Json(serde_json::json!({"ok": false, "error": format!("SSH write local.toml failed: {stderr}")})).into_response();
        }
        _ => {}
    }

    let mut warnings: Vec<String> = Vec::new();

    // Step 2: Push symbol overrides via sqlite3 (always sync — DELETE ALL + INSERT)
    {
        let mut sql = String::from("DELETE FROM symbol_overrides;\n");
        for r in &overrides {
            let (id1, id2) = if r.exchange_id_1 < r.exchange_id_2 {
                (r.exchange_id_1, r.exchange_id_2)
            } else {
                (r.exchange_id_2, r.exchange_id_1)
            };
            let opt_f = |v: &Option<f64>| v.map(|f| f.to_string()).unwrap_or_else(|| "NULL".into());
            let opt_i = |v: &Option<i32>| v.map(|i| i.to_string()).unwrap_or_else(|| "NULL".into());
            let opt_i64 = |v: &Option<i64>| v.map(|i| i.to_string()).unwrap_or_else(|| "NULL".into());
            let opt_s = |v: &Option<String>| v.as_ref().map(|s| format!("'{}'", s.replace('\'', "''"))).unwrap_or_else(|| "NULL".into());
            let sym = r.symbol.replace('\'', "''");
            sql.push_str(&format!(
                "INSERT OR REPLACE INTO symbol_overrides (symbol,exchange_id_1,exchange_id_2,\
                 min_entry_spread_bps,take_profit_pct,min_exit_spread_bps,exit_mode,signal_cooldown_secs,\
                 min_exit_checks,min_top_level_usd,tick_max_spike_bps,\
                 max_recv_staleness_secs,max_tick_age_ms,\
                 max_positions_per_symbol,source_run_id,updated_at_ms) VALUES (\
                 '{sym}',{id1},{id2},{f0},{f1},{f2},{em},{f3},{i0},{f4},{f5},{f11},{f12},{i1},{i2},{ts});\n",
                f0 = opt_f(&r.min_entry_spread_bps), f1 = opt_f(&r.take_profit_pct),
                f2 = opt_f(&r.min_exit_spread_bps), em = opt_s(&r.exit_mode),
                f3 = opt_f(&r.signal_cooldown_secs), i0 = opt_i(&r.min_exit_checks),
                f4 = opt_f(&r.min_top_level_usd), f5 = opt_f(&r.tick_max_spike_bps),
                f11 = opt_f(&r.max_recv_staleness_secs), f12 = opt_f(&r.max_tick_age_ms),
                i1 = opt_i(&r.max_positions_per_symbol),
                i2 = opt_i64(&r.source_run_id), ts = r.updated_at_ms,
            ));
        }
        let sqlite_cmd = format!(
            "{ssh_base} 'sqlite3 {remote_dir}/{db_path_remote}' <<'SQL_EOF'\n{sql}SQL_EOF"
        );
        match run_ssh(sqlite_cmd).await {
            Err(e) => warnings.push(format!("overrides: {e}")),
            Ok(out) if !out.status.success() => {
                let stderr = String::from_utf8_lossy(&out.stderr);
                tracing::warn!(%stderr, "sqlite3 overrides push had errors");
                warnings.push(format!("overrides: {stderr}"));
            }
            _ => {}
        }
    }

    // Step 3: Push blocks via sqlite3 (manual + unprofitable; DELETE synced types + INSERT)
    {
        let now_ms = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64;
        let mut sql = String::from(
            "DELETE FROM symbol_blocks WHERE block_type IN ('manual','unprofitable') AND unblocked_at_ms IS NULL;\n"
        );
        for (symbol, ex1, ex2, block_type, reason) in &push_blocks {
            let sym = symbol.replace('\'', "''");
            let bt = block_type.replace('\'', "''");
            let reason_escaped = reason.replace('\'', "''");
            sql.push_str(&format!(
                "INSERT INTO symbol_blocks (symbol,exchange_id,exchange_id_2,block_type,reason,blocked_at_ms) \
                 VALUES ('{sym}',{ex1},{ex2},'{bt}','{reason_escaped}',{now_ms});\n",
                ex1 = ex1.map(|v| v.to_string()).unwrap_or_else(|| "NULL".into()),
                ex2 = ex2.map(|v| v.to_string()).unwrap_or_else(|| "NULL".into()),
            ));
        }
        let sqlite_cmd = format!(
            "{ssh_base} 'sqlite3 {remote_dir}/{db_path_remote}' <<'SQL_EOF'\n{sql}SQL_EOF"
        );
        match run_ssh(sqlite_cmd).await {
            Err(e) => warnings.push(format!("blocks: {e}")),
            Ok(out) if !out.status.success() => {
                let stderr = String::from_utf8_lossy(&out.stderr);
                tracing::warn!(%stderr, "sqlite3 blocks push had errors");
                warnings.push(format!("blocks: {stderr}"));
            }
            _ => {}
        }
    }

    // Step 4: Trigger hot-reload on remote bot via SSH curl to localhost
    let bot_port = srv.get("bot_port").and_then(|v| v.as_integer()).unwrap_or(3030);
    let mut hot_reloaded = false;
    let reload_cmd = format!(
        "{ssh_base} 'curl -sf -X POST http://127.0.0.1:{bot_port}/api/config/hot-reload --max-time 10'"
    );
    match run_ssh(reload_cmd).await {
        Ok(out) if out.status.success() => { hot_reloaded = true; }
        Ok(out) => {
            let stderr = String::from_utf8_lossy(&out.stderr);
            warnings.push(format!("hot-reload failed: {stderr} — push succeeded but bot needs manual restart"));
        }
        Err(e) => warnings.push(format!("hot-reload SSH error: {e}")),
    }

    let server_name = srv.get("name").and_then(|v| v.as_str()).unwrap_or(server_id);
    tracing::info!(server = %server_id, host, warnings = ?warnings, hot_reloaded, "Config push completed");
    let mut resp = serde_json::json!({
        "ok": true,
        "message": format!("Config pushed to {} ({})", server_name, host),
        "overrides_count": overrides.len(),
        "blocks_count": push_blocks.len(),
        "hot_reloaded": hot_reloaded,
    });
    if !warnings.is_empty() {
        resp["warnings"] = serde_json::json!(warnings);
        resp["message"] = serde_json::json!(format!(
            "Config pushed to {} ({}) with warnings", server_name, host
        ));
    }
    Json(resp).into_response()
}

/// POST /api/config/hot-reload — re-read TOML and DB, apply hot config + reload overrides + sync blocks DashMap.
/// Localhost-only (called via SSH curl from `api_config_push`). No session required.
/// Returns `requires_restart=true` if any field that can only be consumed at
/// bot startup has changed — today that's per-exchange `orderbook_depth`,
/// which freezes the WS subscription URL when the OB client task spawns.
async fn api_config_hot_reload(
    axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo<SocketAddr>,
    State(state): State<AppState>,
) -> impl IntoResponse {
    // Reject non-localhost callers — endpoint is internal-only.
    if !addr.ip().is_loopback() {
        return (axum::http::StatusCode::FORBIDDEN, "localhost only").into_response();
    }

    use super::state::HotConfig;
    let table = read_merged_config_table();

    let s = |section: &str, key: &str| -> Option<&toml::Value> {
        table.get(section)?.get(key)
    };
    let f = |section: &str, key: &str, def: f64| -> f64 {
        s(section, key).and_then(|v| v.as_float().or_else(|| v.as_integer().map(|i| i as f64))).unwrap_or(def)
    };

    // Build per-exchange hot config from TOML
    let mut hot_exchanges: std::collections::HashMap<String, super::state::HotExchangeConfig> = std::collections::HashMap::new();
    if let Some(ex_table) = table.get("exchanges").and_then(|v| v.as_table()) {
        for (name, sec) in ex_table {
            let parse_slip = |key: &str| -> Option<crate::config::SlippagePct> {
                let v = sec.get(key)?;
                if let Some(s) = v.as_str() {
                    if s.eq_ignore_ascii_case("max") { return Some(crate::config::SlippagePct::Max); }
                    s.parse::<f64>().ok().map(crate::config::SlippagePct::Fixed)
                } else {
                    v.as_float().map(crate::config::SlippagePct::Fixed)
                }
            };
            hot_exchanges.insert(name.clone(), super::state::HotExchangeConfig {
                enabled: sec.get("enabled").and_then(|v| v.as_bool()),
                entry_order_type: sec.get("entry_order_type").and_then(|v| v.as_str()).map(String::from),
                entry_slippage_pct: parse_slip("entry_slippage_pct"),
                exit_order_type: sec.get("exit_order_type").and_then(|v| v.as_str()).map(String::from),
                exit_slippage_pct: parse_slip("exit_slippage_pct"),
                emergency_slippage_pct: parse_slip("emergency_slippage_pct"),
                stop_loss_type: sec.get("stop_loss_type").and_then(|v| v.as_str()).map(String::from),
                allowed_with: sec.get("allowed_with").and_then(|v| v.as_array()).map(|arr| {
                    arr.iter().filter_map(|e| e.as_str().map(String::from)).collect()
                }),
            });
        }
    }

    let hot = HotConfig {
        max_order_size_usd: f("position", "max_order_size_usd", 25.0),
        leverage: s("position", "leverage").and_then(|v| v.as_integer()).unwrap_or(3) as u16,
        max_used_capital_usd: f("position", "max_used_capital_usd", 50.0),
        max_open_positions: s("position", "max_open_positions").and_then(|v| v.as_integer()).unwrap_or(5) as usize,
        min_entry_spread_bps: f("strategy", "min_entry_spread_bps", 35.0),
        include_fees_in_entry: s("strategy", "include_fees_in_entry").and_then(|v| v.as_bool()).unwrap_or(false),
        min_top_level_usd: f("strategy", "min_top_level_usd", 0.0),
        tick_max_spike_bps: f("strategy", "tick_max_spike_bps", 0.0),
        tick_spike_window_ms: f("strategy", "tick_spike_window_ms", 0.0),
        max_trade_divergence_bps: f("strategy", "max_trade_divergence_bps", 0.0),
        max_recv_staleness_secs: f("strategy", "max_recv_staleness_secs", 0.1),
        max_tick_age_ms: f("strategy", "max_tick_age_ms", 100.0),
        signal_cooldown_secs: f("strategy", "signal_cooldown_secs", 3.0),
        exit_mode: s("strategy", "exit_mode").and_then(|v| v.as_str()).unwrap_or("spread").to_string(),
        min_exit_spread_bps: f("strategy", "min_exit_spread_bps", 5.0),
        min_exit_pnl_pct: f("strategy", "min_exit_pnl_pct", 0.3),
        min_exit_checks: s("strategy", "min_exit_checks").and_then(|v| v.as_integer()).unwrap_or(1) as u32,
        include_fees_in_exit: s("strategy", "include_fees_in_exit").and_then(|v| v.as_bool()).unwrap_or(true),
        exchanges: hot_exchanges,
        disabled_pairs: Vec::new(),
    };

    // Detect boot-only fields that changed. Both `orderbook_depth` and
    // `ws_ob_mode` are consumed when the OB WS client task spawns —
    // `orderbook_depth` freezes the subscription URL (`?depth=1`), and
    // `ws_ob_mode` decides whether one all-markets WS or one-per-symbol
    // is created. Neither can be changed without tearing the client down,
    // so flipping them in TOML has no effect until a full bot restart.
    //
    // Parsing route: round-trip each `[exchanges.X]` section through the
    // real `ExchangeConfig` deserializer rather than re-implementing field
    // matching here. Every field has `#[serde(default)]`, so partial or
    // missing sections deserialize cleanly. This way `OrderbookDepth` and
    // `WsObMode` semantics (including fail-open on typos and snake_case
    // renaming) live in exactly one place — `bot/src/config.rs` — and any
    // future boot-only field added there only needs a new comparison block
    // below, not a parallel parser.
    let mut restart_required_changes: Vec<String> = Vec::new();
    if let Some(ex_table) = table.get("exchanges").and_then(|v| v.as_table()) {
        for (name, sec) in ex_table {
            let parsed: crate::config::ExchangeConfig =
                match sec.clone().try_into() {
                    Ok(c) => c,
                    Err(e) => {
                        // Malformed section — skip the boot-only diff for it
                        // rather than fail the whole hot reload. The hot
                        // config validation below still has its own path.
                        tracing::warn!(
                            exchange = %name,
                            error = %e,
                            "Hot reload: failed to parse exchange section for boot-only diff",
                        );
                        continue;
                    }
                };

            let boot_depth = state.boot_orderbook_depths
                .get(name)
                .copied()
                .unwrap_or_default();
            if parsed.orderbook_depth != boot_depth {
                restart_required_changes.push(format!(
                    "exchanges.{name}.orderbook_depth: {boot_depth:?} → {:?}",
                    parsed.orderbook_depth,
                ));
            }

            let boot_mode = state.boot_ws_ob_modes
                .get(name)
                .copied()
                .unwrap_or_default();
            if parsed.ws_ob_mode != boot_mode {
                restart_required_changes.push(format!(
                    "exchanges.{name}.ws_ob_mode: {boot_mode:?} → {:?}",
                    parsed.ws_ob_mode,
                ));
            }
        }
    }
    let requires_restart = !restart_required_changes.is_empty();
    if requires_restart {
        tracing::warn!(
            changes = ?restart_required_changes,
            "Hot reload: boot-only fields changed — a full bot restart is required for these to take effect"
        );
    }

    // Apply hot config — bumps version, executor will reload config + symbol overrides on next tick.
    let version = state.dash.set_hot_config(hot);

    // Sync blocks DashMap from DB: replace manual+unprofitable, keep auto-blocks.
    let mut blocks_synced = 0usize;
    if let Some(bs) = state.dash.blocked_symbols() {
        let db_blocks: Vec<crate::storage::SymbolBlock> = state.web_db.run(|db| {
            db.load_active_blocks()
                .map(|rows| rows.into_iter().map(|(_, b)| b).collect())
                .unwrap_or_default()
        }).await;
        if !db_blocks.is_empty() || !bs.is_empty() {
            // Step 1: remove existing manual+unprofitable from DashMap
            let mut to_clean: Vec<crate::types::SymbolId> = Vec::new();
            for entry in bs.iter() {
                let sym_id = *entry.key();
                let has_synced = entry.value().iter().any(|b| matches!(
                    b.block_type,
                    crate::storage::BlockType::Manual | crate::storage::BlockType::Unprofitable
                ));
                if has_synced { to_clean.push(sym_id); }
            }
            for sym_id in &to_clean {
                if let Some(mut entry) = bs.get_mut(sym_id) {
                    entry.retain(|b| !matches!(
                        b.block_type,
                        crate::storage::BlockType::Manual | crate::storage::BlockType::Unprofitable
                    ));
                }
                bs.remove_if(sym_id, |_, blocks| blocks.is_empty());
            }
            // Step 2: insert manual+unprofitable from DB
            for b in db_blocks {
                if !matches!(b.block_type, crate::storage::BlockType::Manual | crate::storage::BlockType::Unprofitable) {
                    continue;
                }
                if let Some(sym_id) = state.dash.symbol_name_to_id(&b.symbol) {
                    bs.entry(sym_id).or_default().push(b);
                    blocks_synced += 1;
                }
            }
        }
    }

    tracing::info!(version, blocks_synced, requires_restart, "Hot reload via /api/config/hot-reload");
    Json(serde_json::json!({
        "ok": true,
        "version": version,
        "blocks_synced": blocks_synced,
        "requires_restart": requires_restart,
        "restart_required_changes": restart_required_changes,
    })).into_response()
}

// ─── Symbol overrides API ─────────────────────────────────────────────────

async fn api_get_symbol_overrides(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }

    // Pull both the override rows and the per-source-run summaries inside a
    // single shared-connection lock — minimizes DB churn and matches the
    // previous behavior.
    type LoadResult = anyhow::Result<(
        Vec<crate::storage::sqlite::SymbolOverrideRow>,
        std::collections::HashMap<i64, serde_json::Value>,
    )>;
    let load: LoadResult = state.web_db.run(|db| {
        let rows = db.load_symbol_overrides()?;
        let mut run_summaries: std::collections::HashMap<i64, serde_json::Value> = std::collections::HashMap::new();
        if let Ok(mut stmt) = db.conn_ref().prepare(
            "SELECT r.id, r.tick_range_desc, r.best_sharpe, r.best_profit,
                    res.net_profit, res.trade_count, res.win_rate, res.sharpe_ratio
             FROM optimizer_runs r
             LEFT JOIN optimizer_results res ON res.run_id = r.id AND res.rank = 1
             WHERE r.id IN (SELECT DISTINCT source_run_id FROM symbol_overrides WHERE source_run_id IS NOT NULL)"
        ) {
            let _ = stmt.query_map([], |row| {
                let id: i64 = row.get(0)?;
                let tick_range: Option<String> = row.get(1)?;
                let period = tick_range.as_ref().and_then(|desc| {
                    let parts: Vec<&str> = desc.split(" — ").collect();
                    if parts.len() == 2 {
                        let t0 = chrono::NaiveDateTime::parse_from_str(parts[0].trim_end_matches(" UTC"), "%Y-%m-%d %H:%M:%S").ok()?;
                        let t1 = chrono::NaiveDateTime::parse_from_str(parts[1].trim_end_matches(" UTC"), "%Y-%m-%d %H:%M:%S").ok()?;
                        let hours = (t1 - t0).num_hours();
                        if hours < 24 { Some(format!("{}h", hours)) } else { Some(format!("{}d", hours / 24)) }
                    } else { None }
                });
                Ok((id, serde_json::json!({
                    "period": period,
                    "profit": row.get::<_, Option<f64>>(4)?,
                    "trades": row.get::<_, Option<i64>>(5)?,
                    "win_rate": row.get::<_, Option<f64>>(6)?,
                    "sharpe": row.get::<_, Option<f64>>(7)?,
                })))
            }).map(|qrows| {
                for r in qrows.flatten() { run_summaries.insert(r.0, r.1); }
            });
        }
        Ok((rows, run_summaries))
    }).await;

    let (rows, run_summaries) = match load {
        Ok(v) => v,
        Err(e) => return Json(serde_json::json!({"error": format!("{e}")})).into_response(),
    };

    // Load market params + last prices once for min_stake_usd computation
    let pairs = match crate::storage::sqlite::Storage::open(state.db_path) {
        Ok(db) => db.load_pairs_params(),
        Err(_) => arb_common::config::PairsParams::default(),
    };
    let last_prices = crate::web::optimizer::load_last_prices();

    // Enrich with exchange names + run summary + min_stake_usd
    let enriched: Vec<serde_json::Value> = rows.iter().map(|r| {
        let mut v = serde_json::to_value(r).unwrap_or_default();
        v["exchange_name_1"] = serde_json::Value::String(ExchangeId::from_db_id(r.exchange_id_1).name().to_string());
        v["exchange_name_2"] = serde_json::Value::String(ExchangeId::from_db_id(r.exchange_id_2).name().to_string());
        let min_qty = pairs.get_min_qty(&r.symbol);
        let price = last_prices.get(&r.symbol).copied().unwrap_or(0.0);
        if min_qty > 0.0 && price > 0.0 {
            v["min_stake_usd"] = serde_json::json!(min_qty * price);
        }
        if let Some(run_id) = r.source_run_id
            && let Some(summary) = run_summaries.get(&run_id) {
                v["run_summary"] = summary.clone();
            }
        v
    }).collect();
    Json(serde_json::json!({"overrides": enriched})).into_response()
}

async fn api_set_symbol_override(session: Session, State(state): State<AppState>, Json(body): Json<serde_json::Value>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.config") { return e.into_response(); }
    let symbol = body["symbol"].as_str().unwrap_or("").to_string();
    let ex1 = body["exchange_id_1"].as_i64().unwrap_or(0);
    let ex2 = body["exchange_id_2"].as_i64().unwrap_or(0);
    if symbol.is_empty() || ex1 == 0 || ex2 == 0 {
        return Json(serde_json::json!({"error": "symbol, exchange_id_1, exchange_id_2 required"})).into_response();
    }
    let exit_mode_val = body["exit_mode"].as_str();
    if let Some(em) = exit_mode_val
        && em != "spread" && em != "pnl" {
            return Json(serde_json::json!({"error": "exit_mode must be 'spread' or 'pnl'"})).into_response();
        }
    let now_ms = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64;
    let row = crate::storage::sqlite::SymbolOverrideRow {
        symbol,
        exchange_id_1: ex1,
        exchange_id_2: ex2,
        min_entry_spread_bps: body["min_entry_spread_bps"].as_f64(),
        take_profit_pct: body["take_profit_pct"].as_f64(),
        min_exit_spread_bps: body["min_exit_spread_bps"].as_f64(),
        exit_mode: body["exit_mode"].as_str().map(String::from),
        signal_cooldown_secs: body["signal_cooldown_secs"].as_f64(),
        min_exit_checks: body["min_exit_checks"].as_i64().map(|v| v as i32),
        min_top_level_usd: body["min_top_level_usd"].as_f64(),
        tick_max_spike_bps: body["tick_max_spike_bps"].as_f64(),
        max_recv_staleness_secs: body["max_recv_staleness_secs"].as_f64(),
        max_tick_age_ms: body["max_tick_age_ms"].as_f64(),
        max_positions_per_symbol: body["max_positions_per_symbol"].as_i64().map(|v| v as i32),
        source_run_id: body["source_run_id"].as_i64(),
        updated_at_ms: now_ms,
    };
    let result: anyhow::Result<()> = state.web_db.run(move |db| {
        db.upsert_symbol_override(&row)
    }).await;
    match result {
        Ok(()) => {
            // Bump hot config version to trigger reload in engine
            state.dash.bump_override_version();
            Json(serde_json::json!({"ok": true})).into_response()
        }
        Err(e) => Json(serde_json::json!({"error": format!("{e}")})).into_response(),
    }
}

async fn api_delete_symbol_override(session: Session, State(state): State<AppState>, Path((symbol, ex1, ex2)): Path<(String, i64, i64)>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.config") { return e.into_response(); }
    let result: anyhow::Result<()> = state.web_db.run(move |db| {
        db.delete_symbol_override(&symbol, ex1, ex2)
    }).await;
    match result {
        Ok(()) => {
            state.dash.bump_override_version();
            Json(serde_json::json!({"ok": true})).into_response()
        }
        Err(e) => Json(serde_json::json!({"error": format!("{e}")})).into_response(),
    }
}

// ─── Symbol blocks API ──────────────────────────────────────────────────────

async fn api_get_symbol_blocks(session: Session, State(state): State<AppState>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }
    let result: anyhow::Result<Vec<serde_json::Value>> = state.web_db.run(|db| {
        let rows = db.load_active_blocks()?;
        Ok(rows.into_iter().map(|(_, b)| {
            let ex1_name = b.exchange_id
                .map(|eid| ExchangeId::from_db_id(eid).name().to_string())
                .unwrap_or_else(|| "all".to_string());
            let ex2_name = b.exchange_id_2
                .map(|eid| ExchangeId::from_db_id(eid).name().to_string());
            serde_json::json!({
                "id": b.db_id,
                "symbol": b.symbol,
                "exchange_id": b.exchange_id,
                "exchange_id_2": b.exchange_id_2,
                "exchange_name_1": ex1_name,
                "exchange_name_2": ex2_name,
                "block_type": b.block_type.as_str(),
                "reason": b.reason,
                "blocked_at_ms": b.blocked_at_ms,
            })
        }).collect())
    }).await;
    match result {
        Ok(enriched) => Json(serde_json::json!({"blocks": enriched})).into_response(),
        Err(e) => Json(serde_json::json!({"error": format!("{e}")})).into_response(),
    }
}

async fn api_create_symbol_block(session: Session, State(state): State<AppState>, Json(body): Json<serde_json::Value>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.config") { return e.into_response(); }
    let symbol = body["symbol"].as_str().unwrap_or("").to_string();
    let block_type = body["block_type"].as_str().unwrap_or("manual").to_string();
    let reason = body["reason"].as_str().unwrap_or("").to_string();
    let req_ex1 = body["exchange_id_1"].as_i64().unwrap_or(0);
    let req_ex2 = body["exchange_id_2"].as_i64().unwrap_or(0);
    if symbol.is_empty() {
        return Json(serde_json::json!({"error": "symbol required"})).into_response();
    }

    // Resolve pair, validate uniqueness, and insert — all under one shared lock.
    enum CreateOutcome {
        Created { db_id: i64, ex1: i64, ex2: i64 },
        Error(String),
    }
    let symbol_for_db = symbol.clone();
    let block_type_for_db = block_type.clone();
    let reason_for_db = reason.clone();
    let outcome = state.web_db.run(move |db| {
        let (ex1, ex2) = if req_ex1 > 0 && req_ex2 > 0 {
            (req_ex1.min(req_ex2), req_ex1.max(req_ex2))
        } else {
            let ids: Vec<i64> = match db.conn_ref().prepare(
                "SELECT DISTINCT exchange_id FROM exchange_symbols WHERE symbol = ?1 AND status = 'active' ORDER BY exchange_id"
            ) {
                Ok(mut stmt) => stmt
                    .query_map(rusqlite::params![symbol_for_db], |row| row.get(0))
                    .map(|rows| rows.filter_map(|r| r.ok()).collect())
                    .unwrap_or_default(),
                Err(e) => return CreateOutcome::Error(format!("DB error: {e}")),
            };
            if ids.len() < 2 {
                return CreateOutcome::Error(format!("Symbol {symbol_for_db} not active on 2+ exchanges"));
            }
            (ids[0].min(ids[1]), ids[0].max(ids[1]))
        };

        if let Ok(active) = db.load_active_blocks() {
            let already = active.iter().any(|(_, b)| {
                b.symbol == symbol_for_db && b.exchange_id == Some(ex1) && b.exchange_id_2 == Some(ex2)
            });
            if already {
                return CreateOutcome::Error(format!("{symbol_for_db} already blocked on this pair"));
            }
        }

        match db.insert_symbol_block(&symbol_for_db, Some(ex1), Some(ex2), &block_type_for_db, &reason_for_db) {
            Ok(db_id) => CreateOutcome::Created { db_id, ex1, ex2 },
            Err(e) => CreateOutcome::Error(format!("{e}")),
        }
    }).await;

    match outcome {
        CreateOutcome::Created { db_id, ex1, ex2 } => {
            // Update runtime DashMap for immediate effect
            if let Some(bs) = state.dash.blocked_symbols()
                && let Some(sym_id) = state.dash.symbol_name_to_id(&symbol) {
                    bs.entry(sym_id).or_default().push(crate::storage::SymbolBlock {
                        db_id: Some(db_id),
                        symbol: symbol.clone(),
                        exchange_id: Some(ex1),
                        exchange_id_2: Some(ex2),
                        block_type: crate::storage::BlockType::from_tag(&block_type),
                        reason: reason.clone(),
                        blocked_at_ms: crate::utils::time::current_unix_ms(),
                        expires_at_ms: None,
                    });
                }
            let pair_label = format!("{} ↔ {}", ExchangeId::from_db_id(ex1).name(), ExchangeId::from_db_id(ex2).name());
            tracing::info!(symbol = %symbol, pair = %pair_label, block_type = %block_type, "Symbol BLOCKED via web UI");
            Json(serde_json::json!({"ok": true, "id": db_id, "pair": pair_label})).into_response()
        }
        CreateOutcome::Error(msg) => Json(serde_json::json!({"error": msg})).into_response(),
    }
}

async fn api_delete_symbol_block(session: Session, State(state): State<AppState>, Path(id): Path<i64>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.config") { return e.into_response(); }
    let result: anyhow::Result<Option<crate::storage::SymbolBlock>> = state.web_db.run(move |db| {
        db.unblock_by_id(id, "unblocked via web UI")
    }).await;
    match result {
        Ok(Some(block)) => {
            // Remove from runtime DashMap
            if let Some(bs) = state.dash.blocked_symbols()
                && let Some(sym_id) = state.dash.symbol_name_to_id(&block.symbol) {
                    if let Some(mut entry) = bs.get_mut(&sym_id) {
                        entry.retain(|b| b.db_id != Some(id));
                    }
                    bs.remove_if(&sym_id, |_, blocks| blocks.is_empty());
                }
            tracing::info!(symbol = %block.symbol, "Symbol UNBLOCKED via web UI");
            Json(serde_json::json!({"ok": true, "symbol": block.symbol})).into_response()
        }
        Ok(None) => Json(serde_json::json!({"error": "Block not found or already unblocked"})).into_response(),
        Err(e) => Json(serde_json::json!({"error": format!("{e}")})).into_response(),
    }
}

/// Restart bot: pauses trading, spawns bot_start.sh (builds while web stays alive),
/// then bot_start.sh starts new process and kills this one.
/// Accepts query param ?mode=resume|clean|reset (default: clean).
/// "reset" = clean + wipe database (factory reset).
async fn api_restart(session: Session, State(state): State<AppState>, query: axum::extract::Query<std::collections::HashMap<String, String>>) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.restart") { return e.into_response(); }
    let mode_str = query.get("mode").map(|s| s.as_str()).unwrap_or("clean");
    let mode = match mode_str {
        "resume" | "clean" | "reset" => mode_str.to_string(),
        other => {
            tracing::warn!(mode = %other, "Invalid restart mode");
            return Json(serde_json::json!({"ok": false, "error": format!("Invalid mode: {other}. Expected: resume, clean, reset")})).into_response();
        }
    };
    let git_pull = query.get("git_pull").map(|s| s == "true").unwrap_or(false);
    let use_old = query.get("old").map(|s| s == "true").unwrap_or(false);
    let mode_resp = mode.clone();
    tracing::warn!(%mode, %git_pull, %use_old, user = %session.username, "Bot restart requested via web UI");

    // Freeze trading immediately — no new entries while building/restarting
    state.dash.request_pause();
    state.dash.request_shutdown();
    // In clean/reset mode, request position close from the RUNNING executor (warm connections, market info loaded).
    // The new process will verify clean state, but closing here gives a head start.
    if mode == "clean" || mode == "reset" {
        state.dash.request_close_all();
        let label = if mode == "reset" { "factory reset" } else { "clean" };
        tracing::info!("Trading paused for restart — close_all requested ({label} mode)");
    } else {
        tracing::info!("Trading paused for restart");
    }

    // Spawn bot_start.sh SYNCHRONOUSLY before shutdown kills the runtime.
    // The child process lives independently of our process.
    // Redirect stdout/stderr to logs/restart.log so we can diagnose failures
    // (e.g. snapshot_old_pids output, kill_old_process steps, cargo build errors).
    {
        let old_flag = if use_old { " --old" } else { "" };
        let mut args = Vec::new();
        if git_pull {
            // git pull then bot_start.sh --no-build (binaries uploaded by CI, no local build needed)
            args.extend(["bash".to_string(), "-c".to_string(), format!(
                "cd '{}' && git pull origin master 2>/dev/null; exec bash tools/bot_start.sh --no-build --{}{}",
                std::env::current_dir().unwrap_or_default().display(), mode, old_flag
            )]);
        } else {
            let mut a = vec!["bash".to_string(), "tools/bot_start.sh".to_string(), "--no-build".to_string(), format!("--{}", mode)];
            if use_old { a.push("--old".to_string()); }
            args.extend(a);
        }
        // Open logs/restart.log in append mode. If that fails, fall back to null.
        let (stdout_cfg, stderr_cfg) = match std::fs::OpenOptions::new()
            .create(true).append(true).open("logs/restart.log")
        {
            Ok(f) => {
                let f2 = f.try_clone().unwrap_or_else(|_| std::fs::File::create("/dev/null").unwrap());
                (std::process::Stdio::from(f), std::process::Stdio::from(f2))
            }
            Err(_) => (std::process::Stdio::null(), std::process::Stdio::null()),
        };
        match std::process::Command::new(&args[0])
            .args(&args[1..])
            .stdin(std::process::Stdio::null())
            .stdout(stdout_cfg)
            .stderr(stderr_cfg)
            .spawn()
        {
            Ok(_) => tracing::info!("Restart script spawned (log: logs/restart.log)"),
            Err(e) => tracing::error!(error = %e, "Failed to spawn restart script"),
        }
    }

    let pull_msg = if git_pull { " with git pull" } else { "" };
    let bin_msg = if use_old { " on previous binary" } else { "" };
    Json(serde_json::json!({"ok": true, "message": format!("Restarting ({mode_resp} mode{pull_msg}{bin_msg})...")})).into_response()
}

/// Returns metadata about the deployable and previous binary so the restart
/// dialog can show whether `--old` would actually do something different.
/// Hashes are short prefixes — we only need a visual delta indicator, not a
/// full integrity check.
async fn api_restart_binaries(session: Session) -> impl IntoResponse {
    if let Err(e) = session.require_permission("bot.restart") { return e.into_response(); }
    fn describe(path: &str) -> serde_json::Value {
        match std::fs::metadata(path) {
            Ok(m) => {
                use std::io::Read;
                let sha = std::fs::File::open(path).ok().and_then(|mut f| {
                    use sha2::{Digest, Sha256};
                    let mut h = Sha256::new();
                    let mut buf = [0u8; 64 * 1024];
                    loop {
                        match f.read(&mut buf) {
                            Ok(0) => break,
                            Ok(n) => h.update(&buf[..n]),
                            Err(_) => return None,
                        }
                    }
                    Some(format!("{:x}", h.finalize()))
                });
                let mtime_ms = m.modified().ok()
                    .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                    .map(|d| d.as_millis() as i64);
                serde_json::json!({
                    "exists": true,
                    "size": m.len(),
                    "mtime_ms": mtime_ms,
                    "sha256": sha,
                })
            }
            Err(_) => serde_json::json!({"exists": false}),
        }
    }
    Json(serde_json::json!({
        "ok": true,
        "deploy":   describe("target/release/arb_bot"),
        "previous": describe("target/release/arb_bot.previous"),
    })).into_response()
}

/// Convert a `ClosedTradeRow` (DB) into the `ClosedTradeSnap` shape consumed
/// by the dashboard JS. Shared by the live SSE snapshot and the historical
/// session endpoint so both feed the unified trades table.
fn closed_trade_row_to_snap(r: crate::storage::sqlite::ClosedTradeRow) -> crate::web::state::ClosedTradeSnap {
    use crate::web::state::ClosedTradeSnap;
    let hold_ms = r.closed_at_ms - r.opened_at_ms;
    ClosedTradeSnap {
        id: r.id,
        session_seq: r.session_seq,
        symbol: r.symbol,
        exchange_1: crate::types::ExchangeId::from_db_id(r.exchange_id_1).name().to_string(),
        exchange_2: crate::types::ExchangeId::from_db_id(r.exchange_id_2).name().to_string(),
        direction: if r.is_long_1 { "A".into() } else { "B".into() },
        qty: r.qty,
        trade_state: r.state,
        entry_1: r.entry_price_1,
        entry_2: r.entry_price_2,
        exit_1: r.exit_price_1,
        exit_2: r.exit_price_2,
        gross_pnl: r.gross_pnl,
        net_pnl: r.net_pnl,
        net_pnl_pct: r.net_pnl_pct,
        entry_spread_bps: r.entry_spread_bps,
        fees: r.fees,
        exit_reason: r.exit_reason,
        hold_ms,
        closed_at_ms: r.closed_at_ms,
        signal_at_ms: r.signal_at_ms,
        cfg_entry_bps: r.cfg.as_ref().map(|c| c.entry_bps.to_string()),
        cfg_include_fees_in_entry: r.cfg.as_ref().map(|c| c.include_fees_in_entry),
        cfg_exit_mode: r.cfg.as_ref().map(|c| c.exit_mode.clone()),
        cfg_exit_bps: r.cfg.as_ref().map(|c| c.exit_bps.to_string()),
        cfg_exit_pnl_pct: r.cfg.as_ref().map(|c| c.exit_pnl_pct.to_string()),
        cfg_include_fees_in_exit: r.cfg.as_ref().map(|c| c.include_fees_in_exit),
        cfg_leverage: r.cfg.as_ref().map(|c| c.leverage),
        funding_1: {
            let is_zero = r.funding_1.parse::<rust_decimal::Decimal>()
                .map(|d| d.is_zero()).unwrap_or(true);
            if is_zero { None } else { Some(r.funding_1) }
        },
        funding_2: {
            let is_zero = r.funding_2.parse::<rust_decimal::Decimal>()
                .map(|d| d.is_zero()).unwrap_or(true);
            if is_zero { None } else { Some(r.funding_2) }
        },
    }
}

/// Convert a `ClosedOrderRow` (DB) into the `ClosedOrderSnap` shape consumed
/// by the dashboard JS. Mirrors the inline mapping previously embedded in
/// `enrich_snapshot`; shared with the historical session orders endpoint so
/// expandable trade rows render identically across live and history views.
fn closed_order_row_to_snap(r: crate::storage::sqlite::ClosedOrderRow) -> crate::web::state::ClosedOrderSnap {
    use crate::web::state::ClosedOrderSnap;
    let exchange_name = crate::types::ExchangeId::from_db_id(r.exchange_id).name();
    // `r.signal_at_ms` is resolved per-order in SQL: entry → trades.signal_at_ms,
    // close/exit → trades.exit_signal_at_ms (see load_closed_orders_for_session).
    // Both metrics work uniformly for entry and close orders.
    let tick_to_recv_ms = if r.exchange_ts_ms > 0 && r.recv_ts_ms > r.exchange_ts_ms {
        Some(r.recv_ts_ms - r.exchange_ts_ms)
    } else { None };
    let recv_to_signal_ms = if r.recv_ts_ms > 0 && r.signal_at_ms >= r.recv_ts_ms {
        Some(r.signal_at_ms - r.recv_ts_ms)
    } else { None };
    let sig_to_ord = if r.signal_at_ms > 0 && r.placed_at_ms >= r.signal_at_ms {
        r.placed_at_ms - r.signal_at_ms
    } else { 0 };
    ClosedOrderSnap {
        order_id: r.order_id,
        trade_id: r.trade_id,
        session_seq: r.session_seq,
        symbol: r.symbol,
        exchange: exchange_name.to_string(),
        side: r.side,
        order_type: r.order_type,
        purpose: r.purpose,
        qty: r.qty,
        limit_price: r.limit_price,
        status: r.status,
        filled_qty: r.filled_qty,
        avg_fill_price: r.avg_fill_price,
        fee: r.fee,
        placed_at_ms: r.placed_at_ms,
        closed_at_ms: r.closed_at_ms,
        order_to_confirm_ms: if r.first_fill_at_ms > 0 && r.placed_at_ms > 0 {
            r.first_fill_at_ms.saturating_sub(r.placed_at_ms)
        } else { 0 },
        duration_ms: if r.closed_at_ms > 0 && r.placed_at_ms > 0 {
            (r.closed_at_ms as u64).saturating_sub(r.placed_at_ms)
        } else { 0 },
        close_reason: r.close_reason,
        tick_to_recv_ms,
        recv_to_signal_ms,
        signal_price: r.signal_price,
        signal_to_order_ms: sig_to_ord,
    }
}

/// Enrich DashSnapshot with SQLite data (orders, config) using a borrowed
/// `Storage` connection. Caller is responsible for holding the connection
/// (`web_db.run(|db| enrich_snapshot(snap, db))`). Never called from the
/// executor — only from web handlers and SSE streams.
fn enrich_snapshot(snap: &mut crate::web::state::DashSnapshot, db: &crate::storage::sqlite::Storage) -> anyhow::Result<()> {
    use crate::web::state::TradeOrderSnap;

    for trade in &mut snap.positions {
        // 1. Orders for this trade
        if trade.orders.is_empty()
            && let Ok(rows) = db.query_orders_for_trade(&trade.id) {
                trade.orders = rows.into_iter().map(|r| {
                    let confirm_ms = if r.first_fill_at_ms > 0 && r.placed_at_ms > 0 {
                        r.first_fill_at_ms.saturating_sub(r.placed_at_ms)
                    } else { 0 };
                    // `r.signal_at_ms` is resolved per-order in SQL: entry →
                    // trades.signal_at_ms, close/exit → trades.exit_signal_at_ms
                    // (see query_orders_for_trade). Unknown (recovery/SL) → 0.
                    let tick_to_recv_ms = if r.exchange_ts_ms > 0 && r.recv_ts_ms > r.exchange_ts_ms {
                        Some(r.recv_ts_ms - r.exchange_ts_ms)
                    } else { None };
                    // `>=` so the triggering leg reports 0ms — its
                    // `signal_at_ms == recv_ts_ms` exactly.
                    let recv_to_signal_ms = if r.recv_ts_ms > 0 && r.signal_at_ms >= r.recv_ts_ms {
                        Some(r.signal_at_ms - r.recv_ts_ms)
                    } else { None };
                    let sto = if r.signal_at_ms > 0 && r.placed_at_ms >= r.signal_at_ms {
                        r.placed_at_ms - r.signal_at_ms
                    } else { 0 };
                    TradeOrderSnap {
                        order_id: r.order_id,
                        exchange: crate::types::ExchangeId::from_db_id(r.exchange_id).name().into(),
                        side: r.side,
                        order_type: r.order_type,
                        purpose: r.purpose,
                        qty: r.qty.to_string(),
                        limit_price: r.limit_price.to_string(),
                        status: r.status,
                        filled_qty: r.filled_qty,
                        avg_fill_price: r.avg_fill_price,
                        fee: r.fee,
                        order_to_confirm_ms: confirm_ms,
                        placed_at_ms: r.placed_at_ms,
                        signal_price: r.signal_price,
                        tick_to_recv_ms,
                        recv_to_signal_ms,
                        signal_to_order_ms: sto,
                    }
                }).collect();
            }

        // 2. Strategy config snapshot — per-trade frozen/hot-updated values from DB.
        if trade.cfg_entry_bps.is_none()
            && let Ok((entry_bps, inc_fee_entry, exit_mode, exit_bps, exit_pnl_pct, inc_fee_exit, lev)) = db.conn_ref()
                .query_row(
                    "SELECT cfg_entry_bps, cfg_include_fees_in_entry, cfg_exit_mode,
                            cfg_exit_bps, cfg_exit_pnl_pct, cfg_include_fees_in_exit,
                            cfg_leverage
                     FROM trades WHERE id=?1",
                    [&trade.id],
                    |r| Ok((
                        r.get::<_, Option<String>>(0)?,
                        r.get::<_, Option<i64>>(1)?,
                        r.get::<_, Option<String>>(2)?,
                        r.get::<_, Option<String>>(3)?,
                        r.get::<_, Option<String>>(4)?,
                        r.get::<_, Option<i64>>(5)?,
                        r.get::<_, Option<i64>>(6)?,
                    )),
                )
            {
                trade.cfg_entry_bps = entry_bps.and_then(|s| s.parse::<Decimal>().ok());
                trade.cfg_include_fees_in_entry = inc_fee_entry.map(|v| v != 0);
                trade.cfg_exit_mode = exit_mode;
                trade.cfg_exit_bps = exit_bps.and_then(|s| s.parse::<Decimal>().ok());
                trade.cfg_exit_pnl_pct = exit_pnl_pct.and_then(|s| s.parse::<Decimal>().ok());
                trade.cfg_include_fees_in_exit = inc_fee_exit.map(|v| v != 0);
                trade.cfg_leverage = lev.map(|v| v as u16);
            }
    }

    // 3. Closed positions (current session)
    if snap.closed_positions.is_empty()
        && let Ok(rows) = db.load_session_closed_trades() {
            snap.closed_positions = rows.into_iter().map(closed_trade_row_to_snap).collect();
        }

    // 4. Closed orders (current session)
    if snap.closed_orders.is_empty()
        && let Ok(rows) = db.load_session_closed_orders() {
            snap.closed_orders = rows.into_iter().map(closed_order_row_to_snap).collect();
        }

    Ok(())
}

/// SSE endpoint: pushes full state snapshot every 1 second.
async fn sse_events(
    session: Session,
    State(state): State<AppState>,
) -> Result<Sse<impl tokio_stream::Stream<Item = Result<Event, Infallible>>>, (axum::http::StatusCode, Json<serde_json::Value>)> {
    session.require_permission("dashboard.view")?;
    let stream = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(Duration::from_secs(1)))
        .then(move |_| {
            let dash = state.dash.clone();
            let web_db = state.web_db.clone();
            let tick_latency = state.tick_latency.clone();
            let tick_latency_untracked = state.tick_latency_untracked.clone();
            async move {
                let mut snap = dash.read();
                let snap = web_db.run(move |db| {
                    let _ = enrich_snapshot(&mut snap, db);
                    snap
                }).await;
                // Merge the per-exchange Tick→Recv snapshot into the same
                // JSON body the dashboard polls from `/api/state` — without
                // this the SSE stream silently clobbers the field every
                // second and the UI flashes "No data yet" forever.
                let mut tick_snap = tick_latency.snapshot();
                tick_snap.untracked = (*tick_latency_untracked).clone();
                let json = match serde_json::to_value(&snap) {
                    Ok(mut body) => {
                        if let Some(obj) = body.as_object_mut()
                            && let Ok(v) = serde_json::to_value(&tick_snap) {
                                obj.insert("tick_latency".to_string(), v);
                            }
                        serde_json::to_string(&body).unwrap_or_default()
                    }
                    Err(_) => serde_json::to_string(&snap).unwrap_or_default(),
                };
                Ok(Event::default().event("state").data(json))
            }
        });
    Ok(Sse::new(stream))
}

// ─── History endpoints ──────────────────────────────────────────────────────

/// GET /api/history/sessions — all sessions with aggregated stats.
async fn api_history_sessions(
    session: Session,
    State(state): State<AppState>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }
    let result: anyhow::Result<serde_json::Value> = state.web_db.run(|db| {
        let stats = db.load_all_time_stats()?;
        let sessions = db.load_all_sessions_summary()?;
        let sessions_json: Vec<serde_json::Value> = sessions.iter().map(|s| {
            serde_json::json!({
                "id": s.id,
                "started_at_ms": s.started_at_ms,
                "ended_at_ms": s.ended_at_ms,
                "state": s.state,
                "end_reason": s.end_reason,
                "trade_count": s.trade_count,
                "wins": s.wins,
                "losses": s.losses,
                "total_pnl": s.total_pnl,
                "total_fees": s.total_fees,
                "total_funding": s.total_funding,
                "leverage": s.leverage,
                "min_entry_spread_bps": s.min_entry_spread_bps,
                "exit_mode": s.exit_mode,
                "max_order_size_usd": s.max_order_size_usd,
            })
        }).collect();
        Ok(serde_json::json!({
            "ok": true,
            "stats": {
                "session_count": stats.session_count,
                "trade_count": stats.trade_count,
                "wins": stats.wins,
                "losses": stats.losses,
                "total_pnl": stats.total_pnl,
                "total_fees": stats.total_fees,
                "total_funding": stats.total_funding,
                "total_uptime_ms": stats.total_uptime_ms,
                "best_session_id": stats.best_session_id,
                "best_session_pnl": stats.best_session_pnl,
                "worst_session_id": stats.worst_session_id,
                "worst_session_pnl": stats.worst_session_pnl,
            },
            "sessions": sessions_json,
        }))
    }).await;
    match result {
        Ok(v) => Json(v).into_response(),
        Err(e) => Json(serde_json::json!({"ok": false, "error": e.to_string()})).into_response(),
    }
}

/// GET /api/history/sessions/{id}/trades — closed trades + orders for a
/// specific session. Trades are returned in the same `ClosedTradeSnap` shape
/// as `/api/state.closed_positions`, and orders in the `ClosedOrderSnap`
/// shape — so the dashboard JS can feed them through the unified `tradeRow`
/// renderer (with expandable per-trade order details) instead of building a
/// stripped-down history-only table.
async fn api_history_session_trades(
    session: Session,
    State(state): State<AppState>,
    Path(session_id): Path<i64>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }
    let result: anyhow::Result<serde_json::Value> = state.web_db.run(move |db| {
        let trades: Vec<crate::web::state::ClosedTradeSnap> = db
            .load_closed_trades_for_session(session_id)?
            .into_iter()
            .map(closed_trade_row_to_snap)
            .collect();
        let orders: Vec<crate::web::state::ClosedOrderSnap> = db
            .load_closed_orders_for_session(session_id)?
            .into_iter()
            .map(closed_order_row_to_snap)
            .collect();
        Ok(serde_json::json!({"ok": true, "trades": trades, "orders": orders}))
    }).await;
    match result {
        Ok(v) => Json(v).into_response(),
        Err(e) => Json(serde_json::json!({"ok": false, "error": e.to_string()})).into_response(),
    }
}

// ─── Extended exchange API endpoints ────────────────────────────────────────

/// GET /api/extended/rewards — points earned + leaderboard stats (live from API).
async fn api_extended_rewards(
    session: Session,
    State(state): State<AppState>,
) -> impl IntoResponse {
    if let Err(e) = session.require_permission("dashboard.view") { return e.into_response(); }
    let api = match &state.extended_api {
        Some(a) => a,
        None => return Json(serde_json::json!({"error": "Extended API not configured"})).into_response(),
    };
    let (rewards, stats) = tokio::join!(
        api.fetch_rewards(),
        api.fetch_leaderboard_stats(),
    );
    Json(serde_json::json!({
        "ok": true,
        "rewards": rewards.unwrap_or(serde_json::json!(null)),
        "leaderboard": stats.unwrap_or(serde_json::json!(null)),
    })).into_response()
}

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

    fn temp_log_path(label: &str) -> String {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let pid = std::process::id();
        format!("{}/arb_bot_test_log_{label}_{pid}_{nanos}.log", std::env::temp_dir().display())
    }

    /// Build a synthetic log file with `n` lines, each ~80 bytes long.
    async fn write_lines(path: &str, n: usize) {
        let mut content = String::with_capacity(n * 90);
        for i in 0..n {
            content.push_str(&format!(
                "2026-04-11T08:00:00.000000Z  INFO arb_bot::test_synthetic line_id={i:08} payload=xxxxxxxxxxxxxxxxx\n"
            ));
        }
        tokio::fs::write(path, content).await.unwrap();
    }

    #[tokio::test]
    async fn tail_read_returns_last_n_lines() {
        let path = temp_log_path("last_n");
        write_lines(&path, 5000).await;

        let res = tail_read_lines(&path, 100).await.expect("tail_read");
        assert_eq!(res.lines.len(), 100);
        // Last line in the file is line_id=00004999 (since we wrote 0..5000).
        assert!(res.lines.last().unwrap().contains("line_id=00004999"));
        // First returned line should be 100 from the end → line_id=00004900.
        assert!(res.lines.first().unwrap().contains("line_id=00004900"));

        let _ = tokio::fs::remove_file(&path).await;
    }

    #[tokio::test]
    async fn tail_read_returns_full_file_when_smaller_than_window() {
        let path = temp_log_path("small");
        write_lines(&path, 50).await;

        let res = tail_read_lines(&path, 1000).await.expect("tail_read");
        assert_eq!(res.lines.len(), 50);
        assert!(res.lines.first().unwrap().contains("line_id=00000000"));
        assert!(res.lines.last().unwrap().contains("line_id=00000049"));

        let _ = tokio::fs::remove_file(&path).await;
    }

    #[tokio::test]
    async fn tail_read_handles_empty_file() {
        let path = temp_log_path("empty");
        tokio::fs::write(&path, b"").await.unwrap();

        let res = tail_read_lines(&path, 1000).await.expect("tail_read");
        assert!(res.lines.is_empty());
        assert_eq!(res.total_lines_in_window, 0);

        let _ = tokio::fs::remove_file(&path).await;
    }

    #[tokio::test]
    async fn tail_read_drops_partial_first_line() {
        // Write a file where the first read window starts mid-line; the
        // tail_read must skip the truncated leading line, not include it.
        let path = temp_log_path("midline");
        // 2000 lines × ~95 bytes = ~190 KB. Default window for want=10 is
        // 10 * 256 * 4 = 10 KB → starts WAY past the start, so first read
        // will land in the middle of a line.
        write_lines(&path, 2000).await;

        let res = tail_read_lines(&path, 10).await.expect("tail_read");
        assert_eq!(res.lines.len(), 10);
        // Each returned line must be a complete log line starting with the
        // ISO timestamp prefix; partial lines would not start with `2026-`.
        for line in &res.lines {
            assert!(
                line.starts_with("2026-"),
                "line did not start with timestamp (truncated?): {line}"
            );
        }
        // The very last line should be line_id=00001999 (the actual last).
        assert!(res.lines.last().unwrap().contains("line_id=00001999"));

        let _ = tokio::fs::remove_file(&path).await;
    }
}