riftenlabs-indexer/src/db/cauldron/pool.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 std::collections::VecDeque;

use anyhow::{Context, Result};
use bitcoin_hashes::hex::{FromHex, ToHex};
use log::{debug, info};
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
use rusqlite::{params, Connection};

pub fn create_table(conn: &Connection) {
    conn.execute(
        "
        CREATE TABLE pool (
            creation_utxo TEXT PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
            owner_pkh TEXT NOT NULL,
            token_id TEXT NOT NULL,
            withdrawn_in_utxo TEXT REFERENCES utxo_spending(spent_utxo_hash) ON DELETE SET NULL
        )",
        [],
    )
    .expect("failed to create table pool");

    conn.execute(
        "CREATE TABLE pool_history_entry (
            utxo TEXT PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
            pool TEXT REFERENCES pool(creation_utxo) ON DELETE CASCADE,
            txid TEXT REFERENCES tx(txid) ON DELETE CASCADE,
            tx_pos TEXT NOT NULL,
            mtp_timestamp BIGINT,
            first_seen_timestamp BIGINT
        )",
        [],
    )
    .expect("failed to create table pool_history_entry");
}

fn get_pool_by_utxo(conn: &Connection, utxo_hash: &OutPointHash) -> Result<Option<OutPointHash>> {
    let mut stmt = conn.prepare("SELECT pool FROM pool_history_entry WHERE utxo = ?")?;

    let mut row = stmt.query([utxo_hash.to_hex()])?;
    let utxo_hex: Option<String> = row.next()?.map(|r| r.get(0).unwrap());

    match utxo_hex {
        Some(utxo) => Ok(Some(
            OutPointHash::from_hex(&utxo).expect("invalid original_utxo utxo in db"),
        )),
        None => Ok(None),
    }
}

fn flag_as_withdrawn(
    conn: &Connection,
    pool_utxo: &OutPointHash,
    cauldron: &ParsedContract,
) -> Result<()> {
    conn.execute(
        "UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?",
        params![cauldron.spent_utxo_hash.to_hex(), pool_utxo.to_hex()],
    )
    .context("flagging pool as withdrawn")?;

    Ok(())
}

fn insert_new_pool(conn: &Connection, cauldron: &ParsedContract) -> Result<()> {
    // or replace, as it could have been added in mempool, then block
    conn.execute(
        "INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
        params![
            cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_hex(),
            cauldron.pkh.to_hex(),
            cauldron.token_id.expect("token id for new pool missing").to_hex(),
            None::<String>
        ]
    ).context("inserting new pool")?;

    Ok(())
}

fn insert_pool_history_entry(
    conn: &Connection,
    pool: &OutPointHash,
    cauldron: &ParsedContract,
    mtp_timestamp: Option<u64>,
    first_seen_timestamp: Option<u64>,
) -> Result<()> {
    conn.execute(
        "INSERT INTO pool_history_entry (utxo, pool, txid, tx_pos, mtp_timestamp, first_seen_timestamp)
         VALUES (?, ?, ?, ?, ?, ?)
         ON CONFLICT(utxo) DO UPDATE SET
            pool = excluded.pool,
            txid = excluded.txid,
            tx_pos = excluded.tx_pos,
            mtp_timestamp = COALESCE(excluded.mtp_timestamp, pool_history_entry.mtp_timestamp),
            first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, pool_history_entry.first_seen_timestamp)",
        params![
            cauldron
                .new_utxo_hash
                .expect("utxo hash on new pool history entry")
                .to_hex(),
            pool.to_hex(),
            cauldron
                .new_utxo_txid
                .expect("txid of new pool history entry")
                .to_hex(),
            cauldron
                .new_utxo_n
                .expect("utxo index of new pool history entry"),
            mtp_timestamp,
            first_seen_timestamp,
        ],
    )
    .context("inserting pool_history_entry")?;
    Ok(())
}

pub fn update_pool_history(
    conn: &Connection,
    cauldrons: Vec<ParsedContract>,
    mtp_timestamp: Option<u64>,
    first_seen_timestamp: Option<u64>,
) -> Result<()> {
    if cauldrons.is_empty() {
        return Ok(());
    }

    let mut queue = VecDeque::from(cauldrons);

    while let Some(current) = queue.pop_front() {
        let (pool_utxo, is_new) = match get_pool_by_utxo(conn, &current.spent_utxo_hash)? {
            Some(c) => (c, false),
            None => {
                // Not in our database. Check if child of another interaction in current batch.
                let has_parent = queue
                    .iter()
                    .any(|parent| Some(current.spent_utxo_hash) == parent.new_utxo_hash);

                if has_parent {
                    // Was child of current batch.
                    // Process later (after parent).
                    queue.push_back(current);
                    continue;
                }

                if let Some(utxo) = current.new_utxo_hash {
                    // Not in existing pool or child of a cauldron in current batch.
                    // This is a new pool.
                    assert!(!current.is_withdrawn);
                    (utxo, true)
                } else {
                    // This is a new pool that is immediately withdrawn. Just ignore.
                    assert!(current.is_withdrawn);
                    continue;
                }
            }
        };

        if current.is_withdrawn {
            info!(
                "LP {} withdrawn in {}",
                pool_utxo.to_hex(),
                current.spent_utxo_hash.to_hex()
            );
            flag_as_withdrawn(conn, &pool_utxo, &current)?;
        } else if is_new {
            info!(
                "Cauldron LP created in tx {}",
                current
                    .new_utxo_txid
                    .expect("expected txid in new LP")
                    .to_hex()
            );
            insert_new_pool(conn, &current)?;
            insert_pool_history_entry(
                conn,
                &pool_utxo,
                &current,
                mtp_timestamp,
                first_seen_timestamp,
            )?;
        } else {
            debug!("New entry for pool {}", pool_utxo.to_hex());
            insert_pool_history_entry(
                conn,
                &pool_utxo,
                &current,
                mtp_timestamp,
                first_seen_timestamp,
            )?;
        }
    }

    Ok(())
}