// Copyright (C) 2024-2026 Whiterun LLC
//
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html

use std::collections::HashMap;

use crate::db::cauldron::config::{config_get, config_set};
use crate::db::cauldron::spot::{self, build_spot_ohlc, Confirmed, SpotState, TokenKey};
use anyhow::Result;
use sqlx::{Row, SqlitePool};

/// Bumped whenever the maths that fills `ohlcv_1h` changes, so buckets materialised by
/// an older rule are discarded instead of being served forever — `rebuild_range` uses
/// `INSERT OR IGNORE`, so existing rows are never corrected in place.
///
/// 2: price switched from the signed net ratio to the gross volume ratio.
/// 3: price switched from the trades' execution average to the reserves they left
///    behind, so a sell can no longer print above the buy before it (see `spot`).
pub const OHLCV_VERSION: u32 = 3;
const OHLCV_VERSION_KEY: &str = "ohlcv_version";

pub async fn create_table(pool: &SqlitePool) {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS ohlcv_1h (
            token_id      BLOB    NOT NULL,
            bucket_ts     INTEGER NOT NULL,
            open          REAL    NOT NULL,
            high          REAL    NOT NULL,
            low           REAL    NOT NULL,
            close         REAL    NOT NULL,
            volume_sats   INTEGER NOT NULL,
            volume_tokens INTEGER NOT NULL,
            tx_count      INTEGER NOT NULL,
            PRIMARY KEY (token_id, bucket_ts)
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create ohlcv_1h table");
}

/// Discards `ohlcv_1h` when it was materialised under an older pricing rule.
///
/// Returns `true` when the table was cleared. The caller must then leave repopulation
/// to the incremental background task: re-materialising the whole history inline would
/// hold up Rocket's startup for as long as it takes, and the raw query path already
/// serves correct candles from `pool_history_entry` while the table refills.
pub async fn migrate_if_stale(read_pool: &SqlitePool, write_pool: &SqlitePool) -> Result<bool> {
    let stored = config_get(read_pool, OHLCV_VERSION_KEY)
        .await?
        .and_then(|v| v.parse::<u32>().ok());
    if stored == Some(OHLCV_VERSION) {
        return Ok(false);
    }

    let mut tx = write_pool.begin().await?;
    sqlx::query("DELETE FROM ohlcv_1h")
        .execute(&mut *tx)
        .await?;
    config_set(&mut *tx, OHLCV_VERSION_KEY, &OHLCV_VERSION.to_string()).await;
    tx.commit().await?;

    Ok(true)
}

/// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty.
pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result<Option<i64>> {
    let row: Option<(Option<i64>,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h")
        .fetch_optional(pool)
        .await?;
    Ok(row.and_then(|r| r.0))
}

/// Returns the earliest confirmed trade timestamp floored to the nearest 1-hour bucket,
/// or `None` if there are no confirmed trades.  Used to seed the initial backfill start
/// so the background task doesn't scan from Unix epoch 0.
pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result<Option<i64>> {
    let row: Option<(Option<i64>,)> = sqlx::query_as(
        "SELECT (MIN(phe.effective_timestamp) / 3600) * 3600
         FROM pool_history_entry AS phe
         WHERE phe.mtp_timestamp IS NOT NULL",
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.and_then(|r| r.0))
}

/// Gross traded volume per (token, bucket): `(volume_sats, volume_tokens, tx_count)`.
///
/// Volumes are gross sums of the absolute per-leg deltas. Summing the *signed*
/// deltas instead lets a multi-pool arbitrage transaction — which buys from one pool
/// and sells into others — cancel almost all of its token movement and report a
/// fraction of the volume it actually moved.
async fn load_bucket_volumes(
    read_pool: &SqlitePool,
    since_ts: i64,
    until_ts: i64,
) -> Result<HashMap<(TokenKey, i64), (i64, i64, i64)>> {
    let sql = r#"
WITH tx_trades AS (
  SELECT
    phe.token_id                              AS token_id,
    (phe.effective_timestamp / 3600) * 3600   AS bucket_ts,
    SUM(ABS(phe.sats_delta))                  AS vol_sats,
    SUM(ABS(phe.token_delta))                 AS vol_tokens
  FROM pool_history_entry AS phe
  JOIN tx ON tx.txid = phe.txid
  WHERE tx.blockhash IS NOT NULL
    AND phe.effective_timestamp >= ?
    AND phe.effective_timestamp < ?
  GROUP BY phe.token_id, phe.txid, phe.effective_timestamp
)
SELECT token_id, bucket_ts, SUM(vol_sats), SUM(vol_tokens), COUNT(*)
FROM tx_trades
GROUP BY token_id, bucket_ts
"#;

    let rows = sqlx::query(sql)
        .bind(since_ts)
        .bind(until_ts)
        .fetch_all(read_pool)
        .await?;

    Ok(rows
        .into_iter()
        .map(|r| {
            let token_id: Vec<u8> = r.get(0);
            let bucket_ts: i64 = r.get(1);
            ((token_id, bucket_ts), (r.get(2), r.get(3), r.get(4)))
        })
        .collect())
}

/// Materialise all 1-hour OHLCV buckets whose effective timestamp falls in
/// `[since_ts, until_ts)`, for confirmed transactions only.
///
/// Price is the aggregate pool spot price replayed across the range (see [`spot`]),
/// so it tracks what the pools were actually quoting rather than what the trades
/// averaged. Only buckets containing a pool change are stored: a bucket with no
/// events repeats the previous close exactly, and the read path reconstructs it by
/// carrying that close forward.
///
/// Two-phase approach: the reads run against `read_pool` (no write lock), then the
/// pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly).
/// Uses INSERT OR IGNORE so existing rows are never overwritten.
/// Returns the number of rows inserted.
pub async fn rebuild_range(
    read_pool: &SqlitePool,
    write_pool: &SqlitePool,
    since_ts: i64,
    until_ts: i64,
) -> Result<u64> {
    if since_ts >= until_ts {
        return Ok(0);
    }

    // Phase 1: read and replay using the read pool — no write lock held throughout.
    let mut events =
        spot::load_events(read_pool, None, since_ts, until_ts, Confirmed::Only).await?;
    if events.is_empty() {
        return Ok(0);
    }
    let mut snapshots = spot::load_snapshot(read_pool, None, since_ts, Confirmed::Only).await?;
    let volumes = load_bucket_volumes(read_pool, since_ts, until_ts).await?;

    struct Materialised {
        token_id: TokenKey,
        bucket_ts: i64,
        open: f64,
        high: f64,
        low: f64,
        close: f64,
        volume_sats: i64,
        volume_tokens: i64,
        tx_count: i64,
    }

    let mut pending: Vec<Materialised> = Vec::new();
    let token_ids: Vec<TokenKey> = events.keys().cloned().collect();
    for token_id in token_ids {
        let token_events = events.remove(&token_id).unwrap_or_default();
        let state = SpotState::new(snapshots.remove(&token_id).unwrap_or_default());
        for candle in build_spot_ohlc(state, &token_events, since_ts, until_ts, 3600) {
            if !candle.has_event {
                continue;
            }
            let (volume_sats, volume_tokens, tx_count) = volumes
                .get(&(token_id.clone(), candle.time))
                .copied()
                .unwrap_or((0, 0, 0));
            pending.push(Materialised {
                token_id: token_id.clone(),
                bucket_ts: candle.time,
                open: candle.open,
                high: candle.high,
                low: candle.low,
                close: candle.close,
                volume_sats,
                volume_tokens,
                tx_count,
            });
        }
    }

    if pending.is_empty() {
        return Ok(0);
    }

    // Phase 2: insert pre-computed rows inside a single transaction.
    // The write lock is held only for these fast INSERTs, not during the replay.
    let mut tx = write_pool.begin().await?;
    let mut inserted = 0u64;
    for row in pending {
        inserted += sqlx::query(
            "INSERT OR IGNORE INTO ohlcv_1h
             (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(row.token_id)
        .bind(row.bucket_ts)
        .bind(row.open)
        .bind(row.high)
        .bind(row.low)
        .bind(row.close)
        .bind(row.volume_sats)
        .bind(row.volume_tokens)
        .bind(row.tx_count)
        .execute(&mut *tx)
        .await?
        .rows_affected();
    }
    tx.commit().await?;

    Ok(inserted)
}

pub struct OhlcvRow {
    pub bucket_ts: i64,
    pub open: f64,
    pub high: f64,
    pub low: f64,
    pub close: f64,
    pub volume_sats: i64,
    pub volume_tokens: i64,
    pub tx_count: i64,
}

/// Return the materialised 1-hour candles for a single token in `[start, end)`.
/// Only buckets that had at least one trade are returned (gaps must be filled by the caller).
pub async fn get_active_candles(
    pool: &SqlitePool,
    token_blob: &[u8],
    start: i64,
    end: i64,
) -> Result<Vec<OhlcvRow>> {
    let rows = sqlx::query(
        "SELECT bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count
         FROM ohlcv_1h
         WHERE token_id = ? AND bucket_ts >= ? AND bucket_ts < ?
         ORDER BY bucket_ts ASC",
    )
    .bind(token_blob)
    .bind(start)
    .bind(end)
    .fetch_all(pool)
    .await?;

    Ok(rows
        .into_iter()
        .map(|r| OhlcvRow {
            bucket_ts: r.get(0),
            open: r.get(1),
            high: r.get(2),
            low: r.get(3),
            close: r.get(4),
            volume_sats: r.get(5),
            volume_tokens: r.get(6),
            tx_count: r.get(7),
        })
        .collect())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding, utxo_spending};
    use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
    use std::sync::atomic::{AtomicU64, Ordering};

    static OHLCV_TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    const HOUR: i64 = 3600;

    async fn test_pool() -> SqlitePool {
        let id = OHLCV_TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
        let uri = format!("file:ohlcv_test_{}?mode=memory&cache=shared", id);
        let opts = SqliteConnectOptions::new()
            .filename(&uri)
            .foreign_keys(false);
        SqlitePoolOptions::new().connect_with(opts).await.unwrap()
    }

    async fn setup_db(pool: &SqlitePool) {
        tx::create_table(pool).await;
        utxo_funding::create_table(pool).await;
        utxo_spending::create_table(pool).await;
        cauldron_pool::create_table(pool).await;
        create_table(pool).await; // ohlcv_1h + idx_phe_txid
    }

    /// One trade leg, inserted via raw SQL (FK disabled in tests).
    ///
    /// `reserves` is the state the leg leaves its pool holding — what the price is
    /// now read from — and `deltas` what it moved, which is what volume is read from.
    #[allow(clippy::too_many_arguments)]
    async fn insert_leg(
        conn: &SqlitePool,
        txid: [u8; 32],
        utxo: [u8; 32],
        pool_hash: [u8; 32],
        token_id: [u8; 32],
        ts: i64,
        confirmed: bool,
        reserves: (i64, i64),
        deltas: (i64, i64),
    ) {
        if confirmed {
            sqlx::query(
                "INSERT OR IGNORE INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)",
            )
            .bind(txid.as_slice())
            .bind([0xAA_u8; 32].as_slice())
            .bind(ts)
            .execute(conn)
            .await
            .unwrap();
        } else {
            sqlx::query("INSERT OR IGNORE INTO tx (txid, first_seen_timestamp) VALUES (?, ?)")
                .bind(txid.as_slice())
                .bind(ts)
                .execute(conn)
                .await
                .unwrap();
        }

        sqlx::query(
            "INSERT OR IGNORE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo)
             VALUES (?, ?, ?, NULL)",
        )
        .bind(pool_hash.as_slice())
        .bind([0u8; 20].as_slice())
        .bind(token_id.as_slice())
        .execute(conn)
        .await
        .unwrap();

        sqlx::query(
            "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id)
             VALUES (?, ?, ?, ?, 0, ?, ?, ?)",
        )
        .bind(utxo.as_slice())
        .bind(txid.as_slice())
        .bind([0u8; 32].as_slice())
        .bind(txid.as_slice())
        .bind(reserves.0)
        .bind(reserves.1)
        .bind(token_id.as_slice())
        .execute(conn)
        .await
        .unwrap();

        let seq: i64 =
            sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
                .fetch_one(conn)
                .await
                .unwrap();
        let ts_column = if confirmed {
            "mtp_timestamp"
        } else {
            "first_seen_timestamp"
        };
        sqlx::query(&format!(
            "INSERT INTO pool_history_entry
             (utxo, pool, token_id, txid, tx_pos, {ts_column}, sequence, sats, token_amount, sats_delta, token_delta)
             VALUES (?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)"
        ))
        .bind(utxo.as_slice())
        .bind(pool_hash.as_slice())
        .bind(token_id.as_slice())
        .bind(txid.as_slice())
        .bind(ts)
        .bind(seq)
        .bind(reserves.0)
        .bind(reserves.1)
        .bind(deltas.0)
        .bind(deltas.1)
        .execute(conn)
        .await
        .unwrap();
    }

    /// A confirmed single-pool trade whose reserves are irrelevant to the assertion.
    async fn insert_confirmed_trade(
        conn: &SqlitePool,
        txid: [u8; 32],
        utxo: [u8; 32],
        token_id: [u8; 32],
        ts: i64,
        sats_delta: i64,
        token_delta: i64,
    ) {
        insert_leg(
            conn,
            txid,
            utxo,
            [0xBB; 32],
            token_id,
            ts,
            true,
            (1000, 1000),
            (sats_delta, token_delta),
        )
        .await;
    }

    async fn insert_mempool_trade(
        conn: &SqlitePool,
        txid: [u8; 32],
        utxo: [u8; 32],
        token_id: [u8; 32],
        ts: i64,
    ) {
        insert_leg(
            conn,
            txid,
            utxo,
            [0xBB; 32],
            token_id,
            ts,
            false,
            (1000, 1000),
            (-1000, 25),
        )
        .await;
    }

    async fn closes(conn: &SqlitePool, token_id: [u8; 32]) -> Vec<(i64, f64)> {
        sqlx::query_as(
            "SELECT bucket_ts, close FROM ohlcv_1h WHERE token_id = ? ORDER BY bucket_ts",
        )
        .bind(token_id.as_slice())
        .fetch_all(conn)
        .await
        .unwrap()
    }

    /// `get_min_trade_bucket_ts` should floor a mid-hour timestamp to the hour boundary.
    #[tokio::test]
    async fn test_get_min_trade_bucket_ts_floors_to_hour() {
        let pool = test_pool().await;
        setup_db(&pool).await;

        // Trade at 1727963400 — not hour-aligned; floor to 1727960400
        insert_confirmed_trade(
            &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25,
        )
        .await;

        let result = get_min_trade_bucket_ts(&pool).await.unwrap();
        assert_eq!(
            result,
            Some(1727960400),
            "1727963400 should floor to 1727960400"
        );
    }

    /// `get_min_trade_bucket_ts` returns `None` when only unconfirmed (mempool) trades exist.
    #[tokio::test]
    async fn test_get_min_trade_bucket_ts_no_confirmed_trades() {
        let pool = test_pool().await;
        setup_db(&pool).await;

        insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await;

        let result = get_min_trade_bucket_ts(&pool).await.unwrap();
        assert_eq!(result, None, "mempool-only trades must not be returned");
    }

    /// Running `rebuild_range` twice on the same range inserts nothing on the second call
    /// because INSERT OR IGNORE skips rows that already exist.
    #[tokio::test]
    async fn test_rebuild_range_idempotent() {
        let pool = test_pool().await;
        setup_db(&pool).await;

        insert_confirmed_trade(
            &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25,
        )
        .await;

        let n1 = rebuild_range(&pool, &pool, 1727960400, 1727964000)
            .await
            .unwrap();
        assert!(n1 > 0, "first rebuild should insert at least one bucket");

        let n2 = rebuild_range(&pool, &pool, 1727960400, 1727964000)
            .await
            .unwrap();
        assert_eq!(
            n2, 0,
            "second rebuild must insert nothing (INSERT OR IGNORE)"
        );
    }

    /// Unconfirmed trades (tx.blockhash IS NULL) must not appear in `ohlcv_1h`.
    #[tokio::test]
    async fn test_rebuild_range_excludes_mempool() {
        let pool = test_pool().await;
        setup_db(&pool).await;

        insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await;

        let n = rebuild_range(&pool, &pool, 1727960400, 1727967600)
            .await
            .unwrap();
        assert_eq!(n, 0, "mempool trades must not be materialised");
    }

    /// The reason for `OHLCV_VERSION` 3.
    ///
    /// Reserves and deltas are mainnet GIRL's (token 63664918…f455) buy at
    /// 2026-08-10 15:45:54 followed by its sell at 2026-08-11 08:52:47. Priced by
    /// what the trades averaged, the sell materialises *above* the buy — 0.0836
    /// against 0.0784 per token — while the pool it traded against had just fallen
    /// from 0.0926 to 0.0760.
    #[tokio::test]
    async fn test_rebuild_range_prices_by_reserves_not_execution_average() {
        let pool = test_pool().await;
        setup_db(&pool).await;
        let token = [0x03_u8; 32];

        insert_leg(
            &pool,
            [0x01; 32],
            [0x11; 32],
            [0xB1; 32],
            token,
            HOUR,
            true,
            (540_052, 818_802_757_370_920),
            (100_000, -185_511_337_091_708),
        )
        .await;
        insert_leg(
            &pool,
            [0x02; 32],
            [0x12; 32],
            [0xB1; 32],
            token,
            2 * HOUR,
            true,
            (640_052, 691_199_193_943_404),
            (100_000, -127_603_563_427_516),
        )
        .await;
        insert_leg(
            &pool,
            [0x03; 32],
            [0x13; 32],
            [0xB1; 32],
            token,
            3 * HOUR,
            true,
            (580_053, 762_931_584_125_945),
            (-59_999, 71_732_390_182_541),
        )
        .await;

        rebuild_range(&pool, &pool, 0, 4 * HOUR).await.unwrap();

        let rows = closes(&pool, token).await;
        assert_eq!(rows.len(), 3);
        let buy = rows[1].1;
        let sell = rows[2].1;
        assert!(
            (buy * 1e8 - 0.0926).abs() < 1e-4,
            "buy must materialise at the price it created, got {}",
            buy * 1e8
        );
        assert!(
            (sell * 1e8 - 0.0760).abs() < 1e-4,
            "sell must materialise at the price it created, got {}",
            sell * 1e8
        );
        assert!(
            sell < buy,
            "a sell materialised at {} above the buy before it at {}",
            sell * 1e8,
            buy * 1e8
        );
    }

    /// A multi-pool arbitrage transaction prices at the reserves its legs left, and
    /// still reports every satoshi and token those legs moved.
    #[tokio::test]
    async fn test_rebuild_range_arb_prices_by_reserves_and_keeps_gross_volume() {
        let pool = test_pool().await;
        setup_db(&pool).await;
        let token = [0x03_u8; 32];
        let txid = [0x01_u8; 32];

        // Same transaction, two pools, opposite directions netting to +2 token units.
        insert_leg(
            &pool,
            txid,
            [0x02; 32],
            [0xB1; 32],
            token,
            HOUR,
            true,
            (1_000_000, 3_000_000),
            (-446_491_239, 1_334_527_069),
        )
        .await;
        insert_leg(
            &pool,
            txid,
            [0x04; 32],
            [0xB2; 32],
            token,
            HOUR,
            true,
            (2_000_000, 6_000_000),
            (384_906_040, -1_334_527_067),
        )
        .await;

        rebuild_range(&pool, &pool, 0, 2 * HOUR).await.unwrap();

        let (close, volume_sats, volume_tokens, tx_count): (f64, i64, i64, i64) = sqlx::query_as(
            "SELECT close, volume_sats, volume_tokens, tx_count FROM ohlcv_1h WHERE token_id = ?",
        )
        .bind(token.as_slice())
        .fetch_one(&pool)
        .await
        .unwrap();

        assert!(
            (close - 3_000_000.0 / 9_000_000.0).abs() < 1e-12,
            "close {close} must be the summed reserves the legs left"
        );
        // Net-ratio pricing would divide 61,585,199 sats by 2 token units.
        assert!(
            close < 61_585_199.0 / 2.0 / 1000.0,
            "netting artifact: {close}"
        );
        assert_eq!(volume_sats, 446_491_239 + 384_906_040);
        assert_eq!(volume_tokens, 1_334_527_069 + 1_334_527_067);
        assert_eq!(tx_count, 1, "one transaction, two legs");
    }

    /// A bucket with no pool change repeats the previous close exactly, so it is
    /// reconstructed on read rather than stored — otherwise every token would need a
    /// row for every hour it has ever existed.
    #[tokio::test]
    async fn test_rebuild_range_stores_only_buckets_with_events() {
        let pool = test_pool().await;
        setup_db(&pool).await;
        let token = [0x03_u8; 32];

        insert_leg(
            &pool,
            [0x01; 32],
            [0x11; 32],
            [0xB1; 32],
            token,
            HOUR,
            true,
            (700, 100),
            (700, 100),
        )
        .await;

        rebuild_range(&pool, &pool, 0, 10 * HOUR).await.unwrap();

        let rows = closes(&pool, token).await;
        assert_eq!(
            rows,
            vec![(HOUR, 7.0)],
            "only the bucket that moved is stored"
        );
    }

    /// The version key gates the wipe: stale tables are cleared exactly once.
    #[tokio::test]
    async fn test_migrate_if_stale_clears_once() {
        let pool = test_pool().await;
        setup_db(&pool).await;
        crate::db::cauldron::config::create_table(&pool).await;

        insert_confirmed_trade(
            &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25,
        )
        .await;
        rebuild_range(&pool, &pool, 1727960400, 1727964000)
            .await
            .unwrap();

        assert!(
            migrate_if_stale(&pool, &pool).await.unwrap(),
            "unversioned table must be wiped"
        );
        let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(remaining, 0, "stale buckets must be gone");

        rebuild_range(&pool, &pool, 1727960400, 1727964000)
            .await
            .unwrap();
        assert!(
            !migrate_if_stale(&pool, &pool).await.unwrap(),
            "a table at the current version must survive"
        );
        let kept: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(kept, 1, "rebuilt buckets must not be wiped again");
    }
}