// 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 anyhow::Result;
use sqlx::{Row, SqlitePool};

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");
}

/// 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))
}

/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls
/// in `[since_ts, until_ts)`.
///
/// Two-phase approach: the slow aggregation SELECT runs 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: aggregate using the read pool — no write lock held during the slow CTE.
    let select_sql = r#"
WITH per_pool_tx_raw AS (
  SELECT
    phe.token_id,
    phe.txid,
    phe.effective_timestamp AS ts,
    phe.utxo,
    phe.sats_delta,
    phe.token_delta,
    phe.sequence
  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 < ?
),
per_pool_tx AS (
  SELECT
    token_id,
    txid,
    ts,
    (ts / 3600) * 3600 AS bucket_ts,
    utxo,
    MIN(sequence)         AS min_sequence,
    SUM(sats_delta)       AS signed_sats,
    SUM(token_delta)      AS signed_tokens,
    SUM(ABS(sats_delta))  AS vol_sats,
    SUM(ABS(token_delta)) AS vol_tokens
  FROM per_pool_tx_raw
  GROUP BY token_id, txid, ts, utxo
),
tx_trades AS (
  SELECT
    token_id,
    txid,
    ts,
    bucket_ts,
    MIN(min_sequence)  AS min_sequence,
    SUM(signed_sats)   AS signed_sats,
    SUM(signed_tokens) AS signed_tokens,
    SUM(vol_sats)      AS vol_sats,
    SUM(vol_tokens)    AS vol_tokens
  FROM per_pool_tx
  GROUP BY token_id, txid, ts
),
priceable AS (
  SELECT
    token_id,
    bucket_ts,
    ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price,
    ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC,  min_sequence ASC)  AS rn_asc,
    ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, min_sequence DESC) AS rn_desc
  FROM tx_trades
  WHERE signed_tokens != 0
),
ohlc AS (
  SELECT
    token_id,
    bucket_ts,
    MAX(CASE WHEN rn_asc  = 1 THEN price END) AS open,
    MAX(CASE WHEN rn_desc = 1 THEN price END) AS close,
    MAX(price) AS high,
    MIN(price) AS low
  FROM priceable
  GROUP BY token_id, bucket_ts
),
vol AS (
  SELECT
    token_id,
    bucket_ts,
    SUM(vol_sats)   AS volume_sats,
    SUM(vol_tokens) AS volume_tokens,
    COUNT(*)        AS tx_count
  FROM tx_trades
  GROUP BY token_id, bucket_ts
)
SELECT
  ohlc.token_id,
  ohlc.bucket_ts,
  ohlc.open,
  ohlc.high,
  ohlc.low,
  ohlc.close,
  vol.volume_sats,
  vol.volume_tokens,
  vol.tx_count
FROM ohlc
JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts
"#;

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

    if rows.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 aggregation.
    let mut tx = write_pool.begin().await?;
    let mut inserted = 0u64;
    for row in &rows {
        let token_id: Vec<u8> = row.get(0);
        let bucket_ts: i64 = row.get(1);
        let open: f64 = row.get(2);
        let high: f64 = row.get(3);
        let low: f64 = row.get(4);
        let close: f64 = row.get(5);
        let volume_sats: i64 = row.get(6);
        let volume_tokens: i64 = row.get(7);
        let tx_count: i64 = row.get(8);

        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(token_id)
        .bind(bucket_ts)
        .bind(open)
        .bind(high)
        .bind(low)
        .bind(close)
        .bind(volume_sats)
        .bind(volume_tokens)
        .bind(tx_count)
        .execute(&mut *tx)
        .await?
        .rows_affected();
    }
    tx.commit().await?;

    Ok(inserted)
}

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

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

    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;
        cauldron_pool::create_table(pool).await;
        create_table(pool).await; // ohlcv_1h + idx_phe_txid
    }

    /// Insert a single confirmed trade via raw SQL (FK disabled in tests).
    async fn insert_confirmed_trade(
        pool: &SqlitePool,
        txid: [u8; 32],
        utxo: [u8; 32],
        token_id: [u8; 32],
        mtp_ts: i64,
        sats_delta: i64,
        token_delta: i64,
    ) {
        let blockhash = [0xAA_u8; 32];
        sqlx::query("INSERT INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)")
            .bind(txid.as_slice())
            .bind(blockhash.as_slice())
            .bind(mtp_ts)
            .execute(pool)
            .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, 1000, 1000, ?)",
        )
        .bind(utxo.as_slice())
        .bind(txid.as_slice())
        .bind([0u8; 32].as_slice())
        .bind(txid.as_slice())
        .bind(token_id.as_slice())
        .execute(pool)
        .await
        .unwrap();
        let seq: i64 =
            sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
                .fetch_one(pool)
                .await
                .unwrap();
        sqlx::query(
            "INSERT INTO pool_history_entry
             (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, sequence, sats, token_amount, sats_delta, token_delta)
             VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, ?, ?)",
        )
        .bind(utxo.as_slice())
        .bind([0xBB_u8; 32].as_slice()) // dummy pool hash (FK disabled)
        .bind(token_id.as_slice())
        .bind(txid.as_slice())
        .bind(mtp_ts)
        .bind(seq)
        .bind(sats_delta)
        .bind(token_delta)
        .execute(pool)
        .await
        .unwrap();
    }

    /// Insert a mempool-only trade (no blockhash on the tx row).
    async fn insert_mempool_trade(
        pool: &SqlitePool,
        txid: [u8; 32],
        utxo: [u8; 32],
        token_id: [u8; 32],
        first_seen_ts: i64,
    ) {
        sqlx::query("INSERT INTO tx (txid, first_seen_timestamp) VALUES (?, ?)")
            .bind(txid.as_slice())
            .bind(first_seen_ts)
            .execute(pool)
            .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, 1000, 1000, ?)",
        )
        .bind(utxo.as_slice())
        .bind(txid.as_slice())
        .bind([0u8; 32].as_slice())
        .bind(txid.as_slice())
        .bind(token_id.as_slice())
        .execute(pool)
        .await
        .unwrap();
        let seq: i64 =
            sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
                .fetch_one(pool)
                .await
                .unwrap();
        sqlx::query(
            "INSERT INTO pool_history_entry
             (utxo, pool, token_id, txid, tx_pos, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta)
             VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, -1000, 25)",
        )
        .bind(utxo.as_slice())
        .bind([0xBB_u8; 32].as_slice())
        .bind(token_id.as_slice())
        .bind(txid.as_slice())
        .bind(first_seen_ts)
        .bind(seq)
        .execute(pool)
        .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");
    }
}

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())
}