sim-cli 0.10.0

use std::time::Duration;

use clap::Parser;
use eyre::Result;
use sim_cli::ApiKeyArg;
use simulator_api::{
    AvailableRange, BundleBuildRequest, BundleBuildStatus, BundleBuildStatusResponse,
};
use simulator_client::BacktestClient;

#[derive(Parser, Debug)]
pub struct BuildBundleArgs {
    #[command(flatten)]
    pub api_key: ApiKeyArg,

    /// First slot of the range to build (inclusive).
    pub start_slot: Option<u64>,
    /// Last slot of the range to build (inclusive).
    pub end_slot: Option<u64>,

    /// Build forward from the latest snapshot slot; requires `--bundle-size`.
    #[arg(long)]
    pub latest: bool,

    /// Slots per bundle. Omit to use the pipeline default of 50,000 slots.
    #[arg(long)]
    pub bundle_size: Option<u64>,

    /// Dedupe key: resubmitting the same key returns the existing build instead
    /// of triggering (or re-charging for) a new one.
    #[arg(long)]
    pub idempotency_key: Option<String>,

    /// Poll until the build reaches a terminal state instead of returning on submit.
    #[arg(long)]
    pub wait: bool,

    /// Seconds between status polls when `--wait` is set.
    #[arg(long, default_value_t = 15, value_parser = clap::value_parser!(u64).range(1..))]
    pub poll_secs: u64,

    /// Print the (final) status response as JSON.
    #[arg(long)]
    pub json: bool,
}

pub async fn build_bundle(args: BuildBundleArgs, url: String) -> Result<()> {
    let client = BacktestClient::builder()
        .url(url)
        .api_key(args.api_key.get())
        .build();

    let (start_slot, end_slot) = if args.latest
        && let Some(size) = args.bundle_size
    {
        let start = client.get_latest_snapshot_slot().await?.snapshot_slot + 1;
        let end = start + size;
        (start, end)
    } else if let Some(start) = args.start_slot
        && let Some(end) = args.end_slot
    {
        (start, end)
    } else {
        eyre::bail!(
            "specify a slot range: pass both START_SLOT and END_SLOT, or --latest with --bundle-size"
        );
    };

    println!("Requesting bundle build for: {start_slot}-{end_slot}...");

    let mut resp = client
        .build_bundle(&BundleBuildRequest {
            start_slot,
            end_slot,
            bundle_size: args.bundle_size,
            idempotency_key: args.idempotency_key,
        })
        .await?;
    // ETA "what's cached" inference; best-effort (empty = cold estimate).
    let ranges = if args.json {
        Vec::new()
    } else {
        client.available_ranges().await.unwrap_or_default()
    };
    if !args.json {
        println!("{}", format_status("issued", &resp, &ranges));
    }

    if args.wait {
        while matches!(resp.status, BundleBuildStatus::InProgress) {
            tokio::time::sleep(Duration::from_secs(args.poll_secs)).await;
            resp = client.get_bundle_status(&resp.request_id).await?;
            if !args.json {
                println!("{}", format_status("status", &resp, &ranges));
            }
        }
    }

    if args.json {
        println!("{}", serde_json::to_string_pretty(&resp)?);
    }
    Ok(())
}

#[derive(Parser, Debug)]
pub struct BuildStatusArgs {
    /// Request id returned by `build-bundle`.
    pub request_id: String,

    #[command(flatten)]
    pub api_key: ApiKeyArg,

    /// Print the status response as JSON.
    #[arg(long)]
    pub json: bool,
}

pub async fn get_status(args: BuildStatusArgs, url: String) -> Result<()> {
    let client = BacktestClient::builder()
        .url(url)
        .api_key(args.api_key.get())
        .build();
    let resp = client.get_bundle_status(&args.request_id).await?;
    if args.json {
        println!("{}", serde_json::to_string_pretty(&resp)?);
    } else {
        let ranges = client.available_ranges().await.unwrap_or_default();
        println!("{}", format_status("status", &resp, &ranges));
    }
    Ok(())
}

fn format_status(
    label: &str,
    resp: &BundleBuildStatusResponse,
    ranges: &[AvailableRange],
) -> String {
    let size = resp
        .bundle_size
        .map_or_else(|| "default".to_string(), |s| s.to_string());
    let eta = match resp.status {
        BundleBuildStatus::InProgress => match estimate_remaining_secs(resp, ranges, now_unix_ms())
        {
            0 => " eta finishing up".to_string(),
            secs => format!(" eta {}", format_duration(secs)),
        },
        _ => String::new(),
    };
    format!(
        "{label}: {} [{}] slots {}..={} size={size}{eta}",
        resp.request_id,
        resp.status.as_str(),
        resp.start_slot,
        resp.end_slot,
    )
}

// Per-phase cost model (seconds); tune as real timings firm up. ETA = phases
// still ahead − elapsed-since-submission, recomputed each poll.
const ORCHESTRATOR_SPINUP_SECS: u64 = 180;
const SNAPSHOT_DISCOVERY_SECS: u64 = 60;
const SNAPSHOT_IMPORT_SECS: u64 = 1500; // download (3m+7m) + transform (3m+12m)
const BLOCK_IMPORT_SPINUP_SECS: u64 = 180;
const BLOCK_RATE_HISTORIC: f64 = 130.0; // blk/s via jetstreamer
const BLOCK_RATE_RECENT: f64 = 45.0; // blk/s via RPC (current/prev epoch)
const BUNDLE_BUILD_SPINUP_SECS: u64 = 180;
const INTERIM_BUNDLE_SECS: u64 = 480; // 4-12m band midpoint
const FINAL_BUNDLE_SECS: u64 = 480; // 4-12m band midpoint
const SIMULATION_SECS_PER_SLOT: f64 = 0.2;
const SLOTS_PER_PARTITION: u64 = 1000;
/// `end_slot` within ~2 epochs of the latest available data is "recent" (RPC-rate).
const RECENT_WINDOW_SLOTS: u64 = 2 * 432_000;

fn estimate_remaining_secs(
    resp: &BundleBuildStatusResponse,
    ranges: &[AvailableRange],
    now_ms: u64,
) -> u64 {
    let total = estimate_total_secs(resp.start_slot, resp.end_slot, ranges);
    let elapsed = now_ms.saturating_sub(resp.created_at_unix_ms) / 1000;
    total.saturating_sub(elapsed)
}

fn estimate_total_secs(start: u64, end: u64, ranges: &[AvailableRange]) -> u64 {
    let snapshot_cached = ranges.iter().any(|r| range_covers(r, start));
    let blocks_cached = ranges
        .iter()
        .any(|r| range_covers(r, start) && range_covers(r, end));

    // Snapshot import and block import run concurrently → phase-2 is the slower;
    // block import grows with the range, so a big cold range can dominate.
    let snapshot_secs = if snapshot_cached {
        0
    } else {
        SNAPSHOT_IMPORT_SECS
    };
    let block_secs = if blocks_cached {
        0
    } else {
        let rate = if is_recent(end, ranges) {
            BLOCK_RATE_RECENT
        } else {
            BLOCK_RATE_HISTORIC
        };
        BLOCK_IMPORT_SPINUP_SECS + (partition_span(start, end) as f64 / rate) as u64
    };
    let phase_two = snapshot_secs.max(block_secs);

    let gap = replay_gap(start, ranges);
    let (interim, simulation) = if gap > 0 {
        (
            INTERIM_BUNDLE_SECS,
            (gap as f64 * SIMULATION_SECS_PER_SLOT) as u64,
        )
    } else {
        (0, 0)
    };

    ORCHESTRATOR_SPINUP_SECS
        + SNAPSHOT_DISCOVERY_SECS
        + phase_two
        + BUNDLE_BUILD_SPINUP_SECS
        + interim
        + simulation
        + FINAL_BUNDLE_SECS
}

fn range_covers(r: &AvailableRange, slot: u64) -> bool {
    r.bundle_start_slot <= slot && r.max_bundle_end_slot.is_some_and(|end| slot <= end)
}

/// `[start, end]` snapped out to 1000-slot partition boundaries.
fn partition_span(start: u64, end: u64) -> u64 {
    let lo = (start / SLOTS_PER_PARTITION) * SLOTS_PER_PARTITION;
    let hi = (end / SLOTS_PER_PARTITION) * SLOTS_PER_PARTITION + (SLOTS_PER_PARTITION - 1);
    hi.saturating_sub(lo) + 1
}

fn is_recent(end: u64, ranges: &[AvailableRange]) -> bool {
    let latest = ranges
        .iter()
        .filter_map(|r| r.max_bundle_end_slot)
        .max()
        .unwrap_or(end);
    latest.saturating_sub(end) <= RECENT_WINDOW_SLOTS
}

/// Slots replayed from the nearest available boundary up to `start`; 0 when a
/// range already covers `start` or nothing anchors below it.
fn replay_gap(start: u64, ranges: &[AvailableRange]) -> u64 {
    if ranges.iter().any(|r| range_covers(r, start)) {
        return 0;
    }
    let anchor = ranges
        .iter()
        .filter_map(|r| r.max_bundle_end_slot)
        .filter(|&end| end <= start)
        .max()
        .or_else(|| {
            ranges
                .iter()
                .map(|r| r.bundle_start_slot)
                .filter(|&s| s <= start)
                .max()
        })
        .unwrap_or(start);
    start.saturating_sub(anchor)
}

fn now_unix_ms() -> u64 {
    chrono::Utc::now().timestamp_millis().max(0) as u64
}

fn format_duration(secs: u64) -> String {
    let (hours, minutes) = (secs / 3600, (secs % 3600) / 60);
    if hours > 0 {
        format!("~{hours}h{minutes:02}m")
    } else {
        format!("~{minutes}m")
    }
}

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

    #[test]
    fn parses_positional_range_and_flags() {
        let args = BuildBundleArgs::try_parse_from([
            "build-bundle",
            "100",
            "200",
            "--api-key",
            "test-key",
            "--bundle-size",
            "500",
            "--wait",
        ])
        .unwrap();
        assert_eq!((args.start_slot, args.end_slot), (Some(100), Some(200)));
        assert_eq!(args.bundle_size, Some(500));
        assert!(args.wait);
    }

    #[test]
    fn poll_secs_rejects_zero() {
        assert!(
            BuildBundleArgs::try_parse_from([
                "build-bundle",
                "1",
                "2",
                "--api-key",
                "k",
                "--wait",
                "--poll-secs",
                "0",
            ])
            .is_err()
        );
    }

    fn range(start: u64, end: u64) -> AvailableRange {
        AvailableRange {
            bundle_start_slot: start,
            bundle_start_slot_utc: None,
            max_bundle_end_slot: Some(end),
            max_bundle_end_slot_utc: None,
            max_bundle_size: Some(end - start + 1),
        }
    }

    fn in_progress(start: u64, end: u64, created_at_unix_ms: u64) -> BundleBuildStatusResponse {
        BundleBuildStatusResponse {
            request_id: "req-1".to_string(),
            start_slot: start,
            end_slot: end,
            bundle_size: None,
            status: BundleBuildStatus::InProgress,
            created_at_unix_ms,
        }
    }

    #[test]
    fn formats_status_with_default_size() {
        let resp = in_progress(1, 2, now_unix_ms());
        let line = format_status("issued", &resp, &[]);
        assert!(line.contains("req-1"));
        assert!(line.contains("in_progress"));
        assert!(line.contains("size=default"));
        assert!(line.contains(" eta "));
    }

    #[test]
    fn cached_range_is_cheaper_than_cold() {
        let covered = [range(0, 2_000_000)];
        let warm = estimate_total_secs(1_000_000, 1_100_000, &covered);
        let cold = estimate_total_secs(1_000_000, 1_100_000, &[]);
        assert!(warm < cold, "warm {warm} should be < cold {cold}");
    }

    #[test]
    fn block_import_dominates_for_big_cold_range() {
        let (start, end) = (1_000_000, 3_000_000); // 2M slots, uncovered
        // With no ranges the build counts as recent (slow 45 blk/s), so the block
        // path far exceeds the snapshot import — phase-2 must take the block path.
        let block_path = BLOCK_IMPORT_SPINUP_SECS
            + (partition_span(start, end) as f64 / BLOCK_RATE_RECENT) as u64;
        assert!(block_path > SNAPSHOT_IMPORT_SECS);
        assert!(estimate_total_secs(start, end, &[]) >= block_path);
    }

    #[test]
    fn countdown_decreases_and_floors_at_zero() {
        let t0 = 1_000_000_000_000;
        let resp = in_progress(1_000_000, 1_100_000, t0);
        let r0 = estimate_remaining_secs(&resp, &[], t0);
        let r1 = estimate_remaining_secs(&resp, &[], t0 + 60_000);
        assert!(r1 < r0 && r1 + 60 == r0);
        assert_eq!(estimate_remaining_secs(&resp, &[], t0 + 10_000_000_000), 0);
    }

    #[test]
    fn recent_is_pricier_than_historic() {
        // A high "latest" marker that covers neither build window.
        let latest = [range(20_000_000, 21_000_000)];
        let recent = estimate_total_secs(21_200_000, 21_300_000, &latest); // near latest -> 45 blk/s
        let historic = estimate_total_secs(1_000_000, 1_100_000, &latest); // far below -> 130 blk/s
        assert!(recent > historic);
    }
}