riftenlabs-indexer/src/rpc/mod.rs

// Copyright (C) 2024 Riften Labs AS
//
// 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::{bail, Context, Result};
use rusqlite::{params, Connection};
use rust_decimal::prelude::*;
use serde::Serialize;
use std::{
    collections::HashMap,
    time::{SystemTime, UNIX_EPOCH},
};

pub mod tvl;

#[allow(clippy::type_complexity)]
pub fn list_tokens_by_volume(
    connection: &Connection,
    seconds: usize,
    limit: usize,
) -> Result<Vec<(String, u64, u64, u64, u64, u64, u64)>> {
    // List token ID's by volume last n seconds
    let mut statement = connection
        .prepare(
            "
            WITH TradeData AS (
                SELECT
                    uf1.token_id,
                    ABS(uf1.sats - COALESCE(uf2.sats, 0)) as trade_volume
                FROM utxo_funding uf1
                INNER JOIN utxo_funding uf2 ON uf1.spent_utxo_hash = uf2.new_utxo_hash
                JOIN tx ON uf1.txid = tx.txid
                WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) >= (strftime('%s', 'now') - ?)
            ),
            TVLData AS (
                SELECT
                    token_id,
                    SUM(sats) as tvl_sats,
                    SUM(token_amount) as tvl_tokens
                FROM utxo_funding
                WHERE new_utxo_hash NOT IN (SELECT spent_utxo_hash FROM utxo_spending)
                GROUP BY token_id
            ),
            HighestUnspentUTXO AS (
                SELECT
                    token_id,
                    MAX(sats) as highest_sats,
                    MAX(token_amount) as token_amount
                FROM utxo_funding
                WHERE new_utxo_hash NOT IN (SELECT spent_utxo_hash FROM utxo_spending)
                GROUP BY token_id
            ),
            AggregateTradeData AS (
                SELECT
                    COALESCE(td.token_id, tvl.token_id) as token_id,
                    COALESCE(SUM(td.trade_volume), 0) as total_trade_volume,
                    COALESCE(COUNT(td.token_id), 0) as number_of_trades,
                    tvl.tvl_sats,
                    tvl.tvl_tokens,
                    hu.highest_sats,
                    hu.token_amount
                FROM TVLData tvl
                LEFT JOIN TradeData td ON tvl.token_id = td.token_id
                LEFT JOIN HighestUnspentUTXO hu ON tvl.token_id = hu.token_id
                GROUP BY tvl.token_id
            )
            SELECT
                token_id,
                total_trade_volume,
                number_of_trades,
                tvl_sats,
                tvl_tokens,
                highest_sats,
                token_amount
            FROM AggregateTradeData
            ORDER BY total_trade_volume DESC, tvl_sats DESC
            LIMIT ?;
        ",
        )
        .unwrap();

    let tokens: Vec<(String, u64, u64, u64, u64, u64, u64)> = statement
        .query_and_then([seconds, limit], |row| {
            let token_id: String = row.get(0)?;
            let trade_volume: u64 = row.get(1)?;
            let trade_count: u64 = row.get(2)?;
            let tvl_sats: u64 = row.get(3)?;
            let tvl_token: u64 = row.get(4)?;
            let best_contract_sats: u64 = row.get(5)?;
            let best_contracts_token: u64 = row.get(6)?;
            Ok((
                token_id,
                trade_volume,
                trade_count,
                tvl_sats,
                tvl_token,
                best_contract_sats,
                best_contracts_token,
            ))
        })
        .unwrap()
        .map(|row: Result<(String, u64, u64, u64, u64, u64, u64)>| row.unwrap())
        .collect();

    Ok(tokens)
}

struct PriceInterval {
    start: i64,
    step: i64,
    sats: i64,
    tokens: i64,
    min: f64,
    max: f64,
}

impl PriceInterval {
    pub fn new(start: i64, step: i64) -> Self {
        Self {
            start,
            step,
            sats: 0,
            tokens: 0,
            min: f64::MAX,
            max: f64::MIN,
        }
    }

    pub fn next(&self) -> Self {
        let new_start = self.start + self.step;
        Self::new(new_start, self.step)
    }

    pub fn end(&self) -> i64 {
        self.start + self.step
    }

    pub fn avg_price(&self) -> Option<f64> {
        if self.tokens == 0 {
            return None;
        }
        Some(self.sats as f64 / self.tokens as f64)
    }

    pub fn add_pool(&mut self, sats: i64, tokens: i64) {
        if tokens != 0 {
            let price = sats as f64 / tokens as f64;
            if price > self.max {
                self.max = price
            }
            if price < self.min {
                self.min = price
            }
        }
        self.sats += sats;
        self.tokens += tokens;
    }

    pub fn to_result(&self) -> Option<(i64, f64, f64, f64)> {
        if self.tokens == 0 {
            None
        } else {
            Some((
                self.start,
                self.avg_price().expect("avg price not calculated"),
                self.max,
                self.min,
            ))
        }
    }
}

// Get the current price of a given token
pub fn current_price(db: &Connection, token_id: &str) -> Result<(u64, u64, f64)> {
    let sql = "SELECT uf.sats, uf.token_amount
        FROM utxo_funding uf
        LEFT JOIN utxo_spending us ON uf.new_utxo_hash = us.spent_utxo_hash
        WHERE us.spent_utxo_hash IS NULL AND uf.token_id = ?";

    let mut statement = db.prepare(sql)?;

    let mut rows = statement.query(params![token_id])?;

    let mut sum_sats: u64 = 0;
    let mut sum_tokens: u64 = 0;

    while let Some(row) = rows.next()? {
        let sats: i64 = row.get(0)?;
        let tokens: i64 = row.get(1)?;

        sum_sats += sats as u64;
        sum_tokens += tokens as u64;
    }

    let price = Decimal::from_u64(sum_sats).context("overflow")?
        / Decimal::from_u64(sum_tokens).context("overflow")?;

    Ok((sum_sats, sum_tokens, price.to_f64().context("overflow")?))
}

#[allow(clippy::type_complexity)]
pub fn historic_price(
    connection: &Connection,
    timestamp_start: i64, // Start timestamp in posix
    timestamp_end: i64,   // End timestamp in posix
    step_size: i64,       // Interval in seconds (e.g., 600 for 10 minutes)
    token_id: &str,
) -> Result<Vec<(i64, f64, f64, f64)>> {
    if timestamp_start > timestamp_end {
        bail!("Start cannot be higher than end");
    }

    let total_intervals = (timestamp_end - timestamp_start) / step_size;

    const MAX_INTERVALS: i64 = 10000;
    if total_intervals > MAX_INTERVALS {
        bail!(
            "Too many intervals ({} > {})",
            total_intervals,
            MAX_INTERVALS
        );
    }

    // Prepare and execute the SQL query for the current interval
    let sql = "
    SELECT
        COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp,
        utxo_funding.sats,
        utxo_funding.token_amount
    FROM
        utxo_funding
    LEFT JOIN
        tx ON utxo_funding.txid = tx.txid
    WHERE
        utxo_funding.token_id = ? AND
        effective_timestamp >= ? AND
        effective_timestamp < ?
    ORDER BY
        effective_timestamp ASC
";
    let mut statement = connection.prepare(sql)?;

    let mut rows = statement.query(params![token_id, timestamp_start, timestamp_end])?;

    let mut result: Vec<(i64, f64, f64, f64)> = Vec::with_capacity(total_intervals as usize);

    let mut current_interval = PriceInterval::new(timestamp_start, step_size);

    while let Some(row) = rows.next()? {
        let timestamp: i64 = row.get(0)?;
        let sats: i64 = row.get(1)?;
        let tokens: i64 = row.get(2)?;

        if timestamp >= current_interval.end() {
            if let Some(r) = current_interval.to_result() {
                result.push(r);
            }
            loop {
                current_interval = current_interval.next();
                if timestamp < current_interval.end() {
                    break;
                }
            }
        }

        current_interval.add_pool(sats, tokens);
    }

    // final trade window
    if let Some(r) = current_interval.to_result() {
        result.push(r);
    }

    Ok(result)
}

#[derive(serde::Serialize)]
pub struct PoolYield {
    token_id: String,
    txid: String,
    tx_pos: i64,
    sats: i64,
    tokens: i64,

    pool_yield: f64,
    apy: f64,
}

pub fn pools_by_apy(connection: &Connection) -> Result<Vec<PoolYield>> {
    let sql = "
        WITH OriginalData AS (
            SELECT
                p.creation_utxo,
                uf.sats AS original_sats,
                uf.token_amount AS original_token_amount,
                uf.token_id as token_id,
                COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS original_timestamp
            FROM
                pool p
            JOIN utxo_funding uf ON p.creation_utxo = uf.new_utxo_hash
            JOIN tx ON uf.txid = tx.txid
            WHERE
                p.withdrawn_in_utxo IS NULL
                AND uf.sats >= 1000000
        ),
        LatestData AS (
            SELECT
                phe.pool,
                uf.sats AS latest_sats,
                uf.token_amount AS latest_token_amount,
                COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS latest_timestamp,
                uf.new_utxo_txid,
                uf.new_utxo_n
            FROM
                pool_history_entry phe
            JOIN utxo_funding uf ON phe.utxo = uf.new_utxo_hash
            JOIN tx ON uf.txid = tx.txid
            WHERE
                phe.pool IN (SELECT creation_utxo FROM OriginalData)
            ORDER BY
                COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) DESC
        )
        SELECT
            od.original_sats,
            od.original_token_amount,
            od.original_timestamp,
            ld.latest_sats,
            ld.latest_token_amount,
            od.token_id,
            ld.new_utxo_txid,
            ld.new_utxo_n
        FROM
            OriginalData od
        JOIN
            LatestData ld ON od.creation_utxo = ld.pool;
        ";

    let mut statement = connection.prepare(sql)?;

    let current_timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64;

    let pool_rows = statement.query_map(params![], |row| {
        let original_sats: i64 = row.get(0)?;
        let original_tokens: i64 = row.get(1)?;
        let original_timestamp: i64 = row.get(2)?;

        let latest_sats: i64 = row.get(3)?;
        let latest_token_amount: i64 = row.get(4)?;
        let token_id = row.get(5)?;

        let txid = row.get(6)?;
        let tx_pos = row.get(7)?;

        assert!(current_timestamp >= original_timestamp);

        let original_k_sr = f64::sqrt((original_sats * original_tokens) as f64);
        let latest_k_sr = f64::sqrt((latest_sats * latest_token_amount) as f64);

        let pool_yield = ((latest_k_sr - original_k_sr) / original_k_sr) * 100.;
        let days_elapsed = (current_timestamp - original_timestamp) as f64 / 86400.0;

        let apy: f64 = if days_elapsed > 0.0 {
            let years_elapsed = 365.25 / days_elapsed;
            (((pool_yield / 100.0) + 1.0).powf(years_elapsed) - 1.0) * 100.0
        } else {
            0.0
        };

        Ok(PoolYield {
            token_id,
            txid,
            tx_pos,
            sats: latest_sats,
            tokens: latest_token_amount,
            pool_yield,
            apy,
        })
    })?;

    let mut pools = Vec::new();
    for pool_row in pool_rows {
        let pool_data = pool_row?;
        pools.push(pool_data);
    }

    // Sort pools by highest APY first
    pools.sort_by(|a, b| {
        b.apy
            .partial_cmp(&a.apy)
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    pools.truncate(1000);

    Ok(pools)
}

#[derive(Serialize)]
pub struct ContractCount {
    active: u64,
    ended: u64,
}

pub fn contract_count(db: &Connection) -> Result<ContractCount> {
    let active: u64 = db.query_row(
        "SELECT COUNT(*) FROM pool WHERE withdrawn_in_utxo IS NULL",
        [],
        |row| row.get(0),
    )?;

    let ended: u64 = db.query_row(
        "SELECT COUNT(*) FROM pool WHERE withdrawn_in_utxo IS NOT NULL",
        [],
        |row| row.get(0),
    )?;

    Ok(ContractCount { active, ended })
}

fn all_time_volume(db: &Connection, end_timestamp: u64) -> Result<Vec<(String, i64)>> {
    let sql = "
    SELECT
        uf1.token_id,
        SUM(ABS(uf1.sats - COALESCE(uf2.sats, 0))) AS total_volume_sats
    FROM utxo_funding uf1
    INNER JOIN utxo_funding uf2 ON uf1.spent_utxo_hash = uf2.new_utxo_hash
    JOIN tx ON uf1.txid = tx.txid
    WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) <= ?
    GROUP BY uf1.token_id";

    let mut stmt = db.prepare(sql)?;

    let volume_iter = stmt.query_map(params![end_timestamp], |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
    })?;

    let mut volumes = Vec::new();
    for volume in volume_iter {
        volumes.push(volume?);
    }

    Ok(volumes)
}

fn period_volume(
    db: &Connection,
    begin_timestamp: u64,
    end_timestamp: u64,
) -> Result<Vec<(String, i64)>> {
    let sql = "
    SELECT
        uf1.token_id,
        SUM(ABS(uf1.sats - COALESCE(uf2.sats, 0))) AS total_volume_sats
    FROM utxo_funding uf1
    INNER JOIN utxo_funding uf2 ON uf1.spent_utxo_hash = uf2.new_utxo_hash
    JOIN tx ON uf1.txid = tx.txid
    WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) BETWEEN ? AND ?
    GROUP BY uf1.token_id
    ";

    let mut stmt = db.prepare(sql)?;
    let volume_iter = stmt.query_map(params![begin_timestamp, end_timestamp], |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
    })?;

    let mut volumes = Vec::new();
    for volume in volume_iter {
        volumes.push(volume?);
    }

    Ok(volumes)
}

pub fn contract_volume(
    db: &Connection,
    end_timestamp: u64,
) -> Result<HashMap<String, (i64, i64, i64)>> {
    let one_day_seconds = 86400_u64;
    let thirty_days_seconds = 30 * 86400_u64;

    let one_day_begin_timestamp = end_timestamp
        .checked_sub(one_day_seconds)
        .context("timestamp underflow")?;
    let thirty_days_begin_timestamp = end_timestamp
        .checked_sub(thirty_days_seconds)
        .context("timestamp underflow")?;

    let all_time = all_time_volume(db, end_timestamp)?;
    let one_day = period_volume(db, one_day_begin_timestamp, end_timestamp)?;
    let thirty_days = period_volume(db, thirty_days_begin_timestamp, end_timestamp)?;

    let mut result = HashMap::new();

    for (token_id, volume) in all_time.into_iter() {
        result.insert(token_id, (volume, 0, 0));
    }

    for (token_id, day_volume) in one_day.into_iter() {
        if let Some((_, _, one_day_volume)) = result.get_mut(&token_id) {
            *one_day_volume = day_volume;
        } else {
            result.insert(token_id.clone(), (0, 0, day_volume));
        }
    }

    for (token_id, month_volume) in thirty_days.into_iter() {
        if let Some((_, thirty_day_volume, _)) = result.get_mut(&token_id) {
            *thirty_day_volume = month_volume;
        } else {
            result.insert(token_id.clone(), (0, month_volume, 0));
        }
    }

    Ok(result)
}