//! Micro-profile of the per-`deliver` primitives, to find where the saga
//! write-path round trips actually go (hotspot H3 in tasks/PERFORMANCE.md)
//! before optimising. Times each primitive in isolation against local PG.
//!
//! Run: `MIRE_PROFILE=1 cargo test --release -p mire-sagas --test pg_profile
//! -- --nocapture --test-threads=1`. Gated on MIRE_PROFILE + DATABASE_URL.

#![allow(dead_code)]

use std::time::{Duration, Instant};

use mire::{Aggregate, AggregateRoot, EventData, EventStore, ExpectedVersion};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
enum Ev {
    Tick,
}
impl EventData for Ev {
    fn event_type(&self) -> &'static str {
        "profile.tick"
    }
}
#[derive(Default)]
struct A;
impl Aggregate for A {
    type Event = Ev;
    fn stream_category() -> &'static str {
        "profile-agg"
    }
    fn apply(&mut self, _: &Ev) {}
}

fn p(label: &str, n: usize, total: Duration) {
    let per = total / n as u32;
    eprintln!("  {label:<34} {per:>9.2?}/op   ({n} ops, {total:.2?} total)");
}

#[tokio::test]
async fn profile_deliver_primitives() {
    if std::env::var("MIRE_PROFILE").is_err() {
        eprintln!("skipping pg_profile; set MIRE_PROFILE=1 to run");
        return;
    }
    let Ok(url) = std::env::var("DATABASE_URL") else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = sqlx::PgPool::connect(&url).await.unwrap();
    let store = EventStore::new(pool.clone());
    store.migrate().await.unwrap();
    mire_sagas::migrate(&store).await.unwrap();

    const N: usize = 500;
    let warm = uuid::Uuid::new_v4().simple().to_string();

    // Seed one stream so `load` has something to read (1 event).
    let seed_id = format!("seed-{warm}");
    {
        let mut root = AggregateRoot::<A>::new(&seed_id);
        root.record(Ev::Tick);
        store.save(&mut root).await.unwrap();
    }

    eprintln!("\nper-deliver primitive cost (local PG, {N} ops each):");

    // 1. instance-lease acquire + release cycle (what a leased op pays on
    //    top of its real work — H3).
    let t = Instant::now();
    for i in 0..N {
        let inst = format!("lease-{warm}-{i}");
        let outcome = mire_sagas::instance_lease::acquire(
            &pool,
            "profile-saga",
            &inst,
            "w",
            Duration::from_secs(15),
        )
        .await
        .unwrap();
        let fence = match outcome {
            mire_sagas::instance_lease::AcquireOutcome::Acquired { fence } => fence,
            mire_sagas::instance_lease::AcquireOutcome::Contended => unreachable!("fresh instance"),
        };
        mire_sagas::instance_lease::release(&pool, "profile-saga", &inst, "w", fence)
            .await
            .unwrap();
    }
    p("lease acquire+release", N, t.elapsed());

    // 2. load (read stream version + events) — what deliver does to rehydrate.
    let t = Instant::now();
    for _ in 0..N {
        let _ = store.load::<A>(&seed_id).await.unwrap();
    }
    p("load (1-event stream)", N, t.elapsed());

    // 3. route query against es_saga_awaits (the lookup deliver does today,
    //    which routing-from-envelope would eliminate).
    let t = Instant::now();
    for _ in 0..N {
        let _: Vec<String> = sqlx::query_scalar(
            "SELECT step FROM es_saga_awaits \
             WHERE saga_name = $1 AND instance_id = $2 AND event_type = $3 \
             ORDER BY step LIMIT 2",
        )
        .bind("profile-saga")
        .bind(&seed_id)
        .bind("profile.tick")
        .fetch_all(&pool)
        .await
        .unwrap();
    }
    p("route query (es_saga_awaits)", N, t.elapsed());

    // 4. a transactional commit like the saga commit: BEGIN + append 1 event
    //    + COMMIT (via the composable append_in_tx path).
    let commit_id = format!("commit-{warm}");
    {
        let mut root = AggregateRoot::<A>::new(&commit_id);
        root.record(Ev::Tick);
        store.save(&mut root).await.unwrap();
    }
    let t = Instant::now();
    for _ in 0..N {
        let mut tx = store.begin_transaction().await.unwrap();
        // append one event to a throwaway stream inside the tx (mirrors the
        // saga commit's tx-scoped append shape).
        let one = format!("commit-ev-{warm}-{}", uuid::Uuid::new_v4().simple());
        tx.append::<Ev>(
            &one,
            "profile-agg",
            ExpectedVersion::NoStream,
            &[mire::Event::new(Ev::Tick)],
        )
        .await
        .unwrap();
        tx.commit().await.unwrap();
    }
    p("tx commit (BEGIN+append+COMMIT)", N, t.elapsed());

    // 5. single-statement append (the hot-path shape) for comparison — what
    //    the saga commit would approach if collapsed (P3.7).
    let t = Instant::now();
    for _ in 0..N {
        let one = format!("ss-ev-{warm}-{}", uuid::Uuid::new_v4().simple());
        store
            .append::<Ev>(
                &one,
                "profile-agg",
                ExpectedVersion::NoStream,
                &[mire::Event::new(Ev::Tick)],
            )
            .await
            .unwrap();
    }
    p("single-statement append (no tx)", N, t.elapsed());

    eprintln!();
}
