1072 lines
34 KiB
Rust
1072 lines
34 KiB
Rust
// 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, VecDeque},
|
|
sync::atomic::AtomicI64,
|
|
};
|
|
|
|
use crate::db::blob::{blob_to_display_hex, display_hex_to_blob, FromBlob, ToBlob};
|
|
use crate::def::PoolID;
|
|
use anyhow::{Context, Result};
|
|
use bitcoin_hashes::hex::ToHex;
|
|
use bitcoincash::TokenID;
|
|
use log::{debug, info, warn};
|
|
use malachite::Integer;
|
|
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
|
|
use rust_decimal::prelude::Zero;
|
|
use serde::{Serialize, Serializer};
|
|
use sqlx::{Row, SqliteConnection, SqlitePool};
|
|
|
|
use crate::rpc::apy::PoolSnapshot;
|
|
|
|
fn serialize_integer_as_string<S>(integer: &Integer, serializer: S) -> Result<S::Ok, S::Error>
|
|
where
|
|
S: Serializer,
|
|
{
|
|
serializer.serialize_str(&integer.to_string())
|
|
}
|
|
|
|
pub async fn create_table(pool: &SqlitePool) {
|
|
sqlx::query(
|
|
"CREATE TABLE pool (
|
|
creation_utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
|
|
owner_pkh BLOB NOT NULL,
|
|
token_id BLOB NOT NULL,
|
|
withdrawn_in_utxo BLOB REFERENCES utxo_spending(spent_utxo_hash) ON DELETE SET NULL
|
|
)",
|
|
)
|
|
.execute(pool)
|
|
.await
|
|
.expect("failed to create table pool");
|
|
|
|
sqlx::query(
|
|
"CREATE TABLE pool_history_entry (
|
|
utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
|
|
pool BLOB REFERENCES pool(creation_utxo) ON DELETE CASCADE,
|
|
token_id BLOB NOT NULL,
|
|
txid BLOB REFERENCES tx(txid) ON DELETE CASCADE,
|
|
tx_pos INT NOT NULL,
|
|
mtp_timestamp BIGINT,
|
|
first_seen_timestamp BIGINT,
|
|
effective_timestamp BIGINT GENERATED ALWAYS AS (COALESCE(first_seen_timestamp, mtp_timestamp)),
|
|
sequence BIGINT NOT NULL,
|
|
sats BIGINT NOT NULL,
|
|
token_amount BIGINT NOT NULL,
|
|
sats_delta BIGINT NOT NULL,
|
|
token_delta BIGINT NOT NULL
|
|
)",
|
|
)
|
|
.execute(pool)
|
|
.await
|
|
.expect("failed to create table pool_history_entry");
|
|
|
|
sqlx::query(
|
|
"CREATE INDEX idx_pool_history_entry_pool_sequence ON pool_history_entry(pool, sequence)",
|
|
)
|
|
.execute(pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query("CREATE INDEX idx_pool_owner_pkh ON pool(owner_pkh)")
|
|
.execute(pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query("CREATE INDEX idx_pool_withdrawn_in_utxo ON pool(withdrawn_in_utxo)")
|
|
.execute(pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query("CREATE INDEX idx_pool_history_entry_pool_timestamp_sequence ON pool_history_entry (pool, effective_timestamp, sequence DESC)")
|
|
.execute(pool).await.unwrap();
|
|
sqlx::query("CREATE INDEX idx_pool_history_entry_token_id ON pool_history_entry(token_id)")
|
|
.execute(pool)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
async fn get_pool_by_utxo(
|
|
conn: &mut SqliteConnection,
|
|
utxo_hash: &OutPointHash,
|
|
) -> Result<Option<OutPointHash>> {
|
|
let row: Option<(Vec<u8>,)> =
|
|
sqlx::query_as("SELECT pool FROM pool_history_entry WHERE utxo = ?")
|
|
.bind(utxo_hash.to_blob())
|
|
.fetch_optional(&mut *conn)
|
|
.await?;
|
|
|
|
match row {
|
|
Some((blob,)) => Ok(Some(
|
|
OutPointHash::from_blob(&blob).expect("invalid original_utxo utxo in db"),
|
|
)),
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
pub async fn flag_as_withdrawn(
|
|
conn: &mut SqliteConnection,
|
|
pool_utxo: &OutPointHash,
|
|
cauldron: &ParsedContract,
|
|
) -> Result<()> {
|
|
sqlx::query("UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?")
|
|
.bind(cauldron.spent_utxo_hash.to_blob())
|
|
.bind(pool_utxo.to_blob())
|
|
.execute(&mut *conn)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("failed flag pool as withdrawn. Original error: {:?}", e))?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn insert_new_pool(conn: &mut SqliteConnection, cauldron: &ParsedContract) -> Result<()> {
|
|
sqlx::query(
|
|
"INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
|
|
)
|
|
.bind(cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_blob())
|
|
.bind(cauldron.pkh.to_blob())
|
|
.bind(cauldron.token_id.expect("token id for new pool missing").to_blob())
|
|
.bind(None::<Vec<u8>>)
|
|
.execute(&mut *conn)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("failed to insert new pool. Original error: {:?}", e))?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Next sequence number in the `pool_history_entry` table
|
|
static NEXT_SEQUENCE: AtomicI64 = AtomicI64::new(-10);
|
|
|
|
pub async fn initialize_seq(pool: &SqlitePool) {
|
|
let row: (i64,) = sqlx::query_as("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
NEXT_SEQUENCE.store(row.0, std::sync::atomic::Ordering::SeqCst);
|
|
}
|
|
|
|
#[allow(dead_code)]
|
|
pub fn dummy_init_seq() {
|
|
if NEXT_SEQUENCE.load(std::sync::atomic::Ordering::SeqCst) < 0 {
|
|
NEXT_SEQUENCE.store(42, std::sync::atomic::Ordering::SeqCst);
|
|
}
|
|
}
|
|
|
|
pub async fn insert_pool_history_entry(
|
|
conn: &mut SqliteConnection,
|
|
pool: &OutPointHash,
|
|
cauldron: &ParsedContract,
|
|
mtp_timestamp: Option<u64>,
|
|
first_seen_timestamp: Option<u64>,
|
|
sats_delta: i64,
|
|
token_delta: i64,
|
|
) -> Result<()> {
|
|
let next_seq = NEXT_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
assert!(next_seq >= 0);
|
|
|
|
sqlx::query(
|
|
"INSERT INTO pool_history_entry (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(utxo) DO UPDATE SET
|
|
pool = excluded.pool,
|
|
token_id = excluded.token_id,
|
|
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),
|
|
sequence = excluded.sequence",
|
|
)
|
|
.bind(cauldron.new_utxo_hash.expect("utxo hash on new pool history entry").to_blob())
|
|
.bind(pool.to_blob())
|
|
.bind(cauldron.token_id.expect("token id on new pool history entry").to_blob())
|
|
.bind(cauldron.new_utxo_txid.expect("txid of new pool history entry").to_blob())
|
|
.bind(cauldron.new_utxo_n.expect("utxo index of new pool history entry") as i64)
|
|
.bind(mtp_timestamp.map(|t| t as i64))
|
|
.bind(first_seen_timestamp.map(|t| t as i64))
|
|
.bind(next_seq)
|
|
.bind(cauldron.sats.map(|s| s as i64))
|
|
.bind(cauldron.token_amount)
|
|
.bind(sats_delta)
|
|
.bind(token_delta)
|
|
.execute(&mut *conn)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("failed to insert pool history entry. Original error: {:?}", e))?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn update_pool_history(
|
|
conn: &mut SqliteConnection,
|
|
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(&mut *conn, ¤t.spent_utxo_hash).await? {
|
|
Some(c) => (c, false),
|
|
None => {
|
|
let has_parent = queue
|
|
.iter()
|
|
.any(|parent| Some(current.spent_utxo_hash) == parent.new_utxo_hash);
|
|
|
|
if has_parent {
|
|
queue.push_back(current);
|
|
continue;
|
|
}
|
|
|
|
if let Some(utxo) = current.new_utxo_hash {
|
|
assert!(!current.is_withdrawn);
|
|
(utxo, true)
|
|
} else {
|
|
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(&mut *conn, &pool_utxo, ¤t).await?;
|
|
} 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(&mut *conn, ¤t).await?;
|
|
insert_pool_history_entry(
|
|
&mut *conn,
|
|
&pool_utxo,
|
|
¤t,
|
|
mtp_timestamp,
|
|
first_seen_timestamp,
|
|
0,
|
|
0,
|
|
)
|
|
.await?;
|
|
} else {
|
|
debug!("New entry for pool {}", pool_utxo.to_hex());
|
|
let prev_entry = get_pool_history_entry(&mut *conn, current.spent_utxo_hash).await?;
|
|
let sats_delta = if let Some(current_sats) = current.sats {
|
|
current_sats as i64 - prev_entry.sats as i64
|
|
} else {
|
|
0
|
|
};
|
|
let token_delta = if let Some(current_token_amount) = current.token_amount {
|
|
current_token_amount - prev_entry.token_amount as i64
|
|
} else {
|
|
0
|
|
};
|
|
insert_pool_history_entry(
|
|
&mut *conn,
|
|
&pool_utxo,
|
|
¤t,
|
|
mtp_timestamp,
|
|
first_seen_timestamp,
|
|
sats_delta,
|
|
token_delta,
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Strategy for which pool state to select when two pool states are available for a timestamp
|
|
#[derive(PartialEq, Eq)]
|
|
enum SnapshotSelection {
|
|
UseBefore,
|
|
UseAfter,
|
|
}
|
|
|
|
/// Get pools and their state nearest to given timestamp.
|
|
async fn get_nearest_entries(
|
|
pool: &SqlitePool,
|
|
timestamp: i64,
|
|
token_id: Option<&str>,
|
|
owner_pkh: Option<&str>,
|
|
resolution: SnapshotSelection,
|
|
) -> Result<HashMap<String, PoolSnapshot>> {
|
|
let token_blob = match token_id {
|
|
Some(id) => Some(display_hex_to_blob::<TokenID>(id)?),
|
|
None => None,
|
|
};
|
|
let owner_pkh_blob = match owner_pkh {
|
|
Some(pkh) => Some(hex::decode(pkh)?),
|
|
None => None,
|
|
};
|
|
|
|
let extra_filters = match (&token_blob, &owner_pkh_blob) {
|
|
(Some(_), Some(_)) => "token_id = ?2 AND owner_pkh = ?3",
|
|
(Some(_), None) => "token_id = ?2",
|
|
(None, Some(_)) => "owner_pkh = ?2",
|
|
(None, None) => "1=1",
|
|
};
|
|
|
|
let ts = "effective_timestamp";
|
|
|
|
let preferred = match resolution {
|
|
SnapshotSelection::UseBefore => "ts ASC",
|
|
SnapshotSelection::UseAfter => "ts DESC",
|
|
};
|
|
|
|
let query = format!(
|
|
"WITH NearestEntries AS (
|
|
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
|
|
FROM pool_history_entry phe
|
|
WHERE {ts} <= ?1
|
|
AND phe.pool IN (
|
|
SELECT creation_utxo
|
|
FROM pool
|
|
WHERE {extra_filters}
|
|
)
|
|
GROUP BY phe.pool
|
|
HAVING MAX({ts})
|
|
|
|
UNION ALL
|
|
|
|
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
|
|
FROM pool_history_entry phe
|
|
WHERE {ts} > ?1
|
|
AND phe.pool IN (
|
|
SELECT creation_utxo
|
|
FROM pool
|
|
WHERE {extra_filters}
|
|
)
|
|
GROUP BY phe.pool
|
|
HAVING MIN({ts})
|
|
)
|
|
SELECT * FROM NearestEntries
|
|
ORDER BY {preferred}",
|
|
);
|
|
|
|
let rows = match (&token_blob, &owner_pkh_blob) {
|
|
(Some(t), Some(p)) => {
|
|
sqlx::query(&query)
|
|
.bind(timestamp)
|
|
.bind(t)
|
|
.bind(p)
|
|
.fetch_all(pool)
|
|
.await?
|
|
}
|
|
(Some(t), None) => {
|
|
sqlx::query(&query)
|
|
.bind(timestamp)
|
|
.bind(t)
|
|
.fetch_all(pool)
|
|
.await?
|
|
}
|
|
(None, Some(p)) => {
|
|
sqlx::query(&query)
|
|
.bind(timestamp)
|
|
.bind(p)
|
|
.fetch_all(pool)
|
|
.await?
|
|
}
|
|
(None, None) => sqlx::query(&query).bind(timestamp).fetch_all(pool).await?,
|
|
};
|
|
let mut pools: HashMap<String, PoolSnapshot> = HashMap::default();
|
|
|
|
for row in rows {
|
|
let pool_blob: Vec<u8> = row.get(0);
|
|
let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
|
|
let sats: i64 = row.get(1);
|
|
let token_amount: i64 = row.get(2);
|
|
let ts_val: i64 = row.get(3);
|
|
|
|
let pool_snapshot = PoolSnapshot {
|
|
pool_id: pool_id.clone(),
|
|
sats: sats as u64,
|
|
token_amount: token_amount as u64,
|
|
timestamp: ts_val as u64,
|
|
};
|
|
|
|
if !pools.contains_key(&pool_snapshot.pool_id) {
|
|
pools.insert(pool_snapshot.pool_id.clone(), pool_snapshot);
|
|
}
|
|
}
|
|
|
|
Ok(pools)
|
|
}
|
|
|
|
/// A capital injection event: both-side-positive delta on a pool_history_entry row.
|
|
/// Carries the actual pool state immediately before and after the injection so callers
|
|
/// can split a pool period into clean sub-periods without synthetic values.
|
|
pub struct InjectionRecord {
|
|
pub sats_before: u64,
|
|
pub tokens_before: u64,
|
|
pub sats_after: u64,
|
|
pub tokens_after: u64,
|
|
pub timestamp: u64,
|
|
}
|
|
|
|
/// Returns injection events (both-side-positive deltas) for a set of pools within a time window.
|
|
/// An injection is a pool state where both sats_delta >= 0 and token_delta >= 0, meaning
|
|
/// capital was added rather than a normal trade occurring.
|
|
/// Callers use the k_before/k_after ratio to neutralize the K jump in APY calculations.
|
|
pub async fn get_injections_between(
|
|
pool: &SqlitePool,
|
|
pool_id_blobs: &[Vec<u8>],
|
|
min_start_ts: i64,
|
|
end_ts: i64,
|
|
) -> Result<HashMap<String, Vec<InjectionRecord>>> {
|
|
if pool_id_blobs.is_empty() {
|
|
return Ok(HashMap::new());
|
|
}
|
|
|
|
let placeholders = (1..=pool_id_blobs.len())
|
|
.map(|i| format!("?{i}"))
|
|
.collect::<Vec<_>>()
|
|
.join(",");
|
|
let n = pool_id_blobs.len();
|
|
let query = format!(
|
|
"SELECT pool, sats, token_amount, sats_delta, token_delta, effective_timestamp
|
|
FROM pool_history_entry
|
|
WHERE pool IN ({placeholders})
|
|
AND effective_timestamp > ?{start_bind}
|
|
AND effective_timestamp <= ?{end_bind}
|
|
AND sats_delta >= 0 AND token_delta >= 0
|
|
AND (sats_delta > 0 OR token_delta > 0)",
|
|
start_bind = n + 1,
|
|
end_bind = n + 2,
|
|
);
|
|
|
|
let mut query_builder = sqlx::query(&query);
|
|
for blob in pool_id_blobs {
|
|
query_builder = query_builder.bind(blob.clone());
|
|
}
|
|
query_builder = query_builder.bind(min_start_ts).bind(end_ts);
|
|
|
|
let rows = query_builder.fetch_all(pool).await?;
|
|
let mut result: HashMap<String, Vec<InjectionRecord>> = HashMap::new();
|
|
|
|
for row in rows {
|
|
let pool_blob: Vec<u8> = row.get(0);
|
|
let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
|
|
let sats: i64 = row.get(1);
|
|
let token_amount: i64 = row.get(2);
|
|
let sats_delta: i64 = row.get(3);
|
|
let token_delta: i64 = row.get(4);
|
|
let ts: i64 = row.get(5);
|
|
|
|
result.entry(pool_id).or_default().push(InjectionRecord {
|
|
sats_before: (sats - sats_delta).max(0) as u64,
|
|
tokens_before: (token_amount - token_delta).max(0) as u64,
|
|
sats_after: sats.max(0) as u64,
|
|
tokens_after: token_amount.max(0) as u64,
|
|
timestamp: ts as u64,
|
|
});
|
|
}
|
|
|
|
Ok(result)
|
|
}
|
|
|
|
/// Returns pool period snapshots filtered by token and/or owner PKH.
|
|
pub async fn get_pool_period_snapshot(
|
|
pool: &SqlitePool,
|
|
token_id: Option<&str>,
|
|
owner_pkh: Option<&str>,
|
|
start: i64,
|
|
end: i64,
|
|
) -> Result<Vec<(PoolSnapshot, PoolSnapshot)>> {
|
|
let pools_start = get_nearest_entries(
|
|
pool,
|
|
start,
|
|
token_id,
|
|
owner_pkh,
|
|
SnapshotSelection::UseBefore,
|
|
)
|
|
.await?;
|
|
let mut pools_end =
|
|
get_nearest_entries(pool, end, token_id, owner_pkh, SnapshotSelection::UseAfter).await?;
|
|
|
|
let mut pools: Vec<(PoolSnapshot, PoolSnapshot)> = Vec::default();
|
|
for (start_pool_id, start_pool) in pools_start {
|
|
let end_pool = match pools_end.remove(&start_pool_id) {
|
|
Some(end) => end,
|
|
None => {
|
|
warn!("Found no 'end pool' for {start_pool_id}");
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let duration = end_pool.timestamp.saturating_sub(start_pool.timestamp);
|
|
if !duration.is_zero() {
|
|
pools.push((start_pool, end_pool))
|
|
}
|
|
}
|
|
|
|
if !pools_end.is_empty() {
|
|
warn!("Found {} end pools not in start pools", pools_end.len());
|
|
}
|
|
|
|
Ok(pools)
|
|
}
|
|
|
|
/// Get the nearest pool history entries for a specific set of pool IDs.
|
|
async fn get_nearest_entries_by_pool_ids(
|
|
pool: &SqlitePool,
|
|
timestamp: i64,
|
|
pool_id_blobs: &[Vec<u8>],
|
|
resolution: SnapshotSelection,
|
|
) -> Result<HashMap<String, PoolSnapshot>> {
|
|
let placeholders: String = (2..=(pool_id_blobs.len() + 1))
|
|
.map(|i| format!("?{i}"))
|
|
.collect::<Vec<_>>()
|
|
.join(",");
|
|
let pool_filter = format!("phe.pool IN ({placeholders})");
|
|
|
|
let ts = "effective_timestamp";
|
|
let preferred = match resolution {
|
|
SnapshotSelection::UseBefore => "ts ASC",
|
|
SnapshotSelection::UseAfter => "ts DESC",
|
|
};
|
|
|
|
let query = format!(
|
|
"WITH NearestEntries AS (
|
|
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
|
|
FROM pool_history_entry phe
|
|
WHERE {ts} <= ?1 AND {pool_filter}
|
|
GROUP BY phe.pool
|
|
HAVING MAX({ts})
|
|
|
|
UNION ALL
|
|
|
|
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
|
|
FROM pool_history_entry phe
|
|
WHERE {ts} > ?1 AND {pool_filter}
|
|
GROUP BY phe.pool
|
|
HAVING MIN({ts})
|
|
)
|
|
SELECT * FROM NearestEntries
|
|
ORDER BY {preferred}",
|
|
);
|
|
|
|
let mut query_builder = sqlx::query(&query).bind(timestamp);
|
|
for blob in pool_id_blobs {
|
|
query_builder = query_builder.bind(blob.clone());
|
|
}
|
|
|
|
let rows = query_builder.fetch_all(pool).await?;
|
|
let mut pools: HashMap<String, PoolSnapshot> = HashMap::default();
|
|
|
|
for row in rows {
|
|
let pool_blob: Vec<u8> = row.get(0);
|
|
let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
|
|
let sats: i64 = row.get(1);
|
|
let token_amount: i64 = row.get(2);
|
|
let ts_val: i64 = row.get(3);
|
|
|
|
let pool_snapshot = PoolSnapshot {
|
|
pool_id: pool_id.clone(),
|
|
sats: sats as u64,
|
|
token_amount: token_amount as u64,
|
|
timestamp: ts_val as u64,
|
|
};
|
|
|
|
if !pools.contains_key(&pool_snapshot.pool_id) {
|
|
pools.insert(pool_snapshot.pool_id.clone(), pool_snapshot);
|
|
}
|
|
}
|
|
|
|
Ok(pools)
|
|
}
|
|
|
|
/// Returns pool period snapshots for a specific list of pool IDs.
|
|
/// Use this instead of `get_pool_period_snapshot` when filtering by pool ID —
|
|
/// it avoids a join to the `pool` table and works across future contract types.
|
|
pub async fn get_pool_period_snapshot_by_pool_ids(
|
|
pool: &SqlitePool,
|
|
pool_ids: &[String],
|
|
start: i64,
|
|
end: i64,
|
|
) -> Result<Vec<(PoolSnapshot, PoolSnapshot)>> {
|
|
let pool_id_blobs: Result<Vec<Vec<u8>>> = pool_ids
|
|
.iter()
|
|
.map(|id| display_hex_to_blob::<PoolID>(id))
|
|
.collect();
|
|
let pool_id_blobs = pool_id_blobs?;
|
|
|
|
let pools_start =
|
|
get_nearest_entries_by_pool_ids(pool, start, &pool_id_blobs, SnapshotSelection::UseBefore)
|
|
.await?;
|
|
let mut pools_end =
|
|
get_nearest_entries_by_pool_ids(pool, end, &pool_id_blobs, SnapshotSelection::UseAfter)
|
|
.await?;
|
|
|
|
let mut pools: Vec<(PoolSnapshot, PoolSnapshot)> = Vec::default();
|
|
for (start_pool_id, start_pool) in pools_start {
|
|
let end_pool = match pools_end.remove(&start_pool_id) {
|
|
Some(end) => end,
|
|
None => {
|
|
warn!("Found no 'end pool' for {start_pool_id}");
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let duration = end_pool.timestamp.saturating_sub(start_pool.timestamp);
|
|
if !duration.is_zero() {
|
|
pools.push((start_pool, end_pool))
|
|
}
|
|
}
|
|
|
|
if !pools_end.is_empty() {
|
|
warn!("Found {} end pools not in start pools", pools_end.len());
|
|
}
|
|
|
|
Ok(pools)
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct PoolHistoryEntry {
|
|
txid: String,
|
|
sats: u64,
|
|
token_amount: u64,
|
|
timestamp: u64,
|
|
#[serde(serialize_with = "serialize_integer_as_string")]
|
|
k: Integer,
|
|
}
|
|
|
|
async fn get_pool_history_entry(
|
|
conn: &mut SqliteConnection,
|
|
utxo_hash: OutPointHash,
|
|
) -> Result<PoolHistoryEntry> {
|
|
let row = sqlx::query(
|
|
"SELECT txid, sats, token_amount, effective_timestamp as timestamp FROM pool_history_entry WHERE utxo = ?",
|
|
)
|
|
.bind(utxo_hash.to_blob())
|
|
.fetch_optional(&mut *conn)
|
|
.await?
|
|
.context("no pool history entry found for UTXO hash")?;
|
|
|
|
let txid_blob: Vec<u8> = row.get(0);
|
|
let txid = blob_to_display_hex::<bitcoincash::Txid>(&txid_blob)?;
|
|
let sats: i64 = row.get(1);
|
|
let token_amount: i64 = row.get(2);
|
|
let timestamp: i64 = row.get(3);
|
|
let sats = sats as u64;
|
|
let token_amount = token_amount as u64;
|
|
Ok(PoolHistoryEntry {
|
|
txid,
|
|
sats,
|
|
token_amount,
|
|
timestamp: timestamp as u64,
|
|
k: Integer::from(sats) * Integer::from(token_amount),
|
|
})
|
|
}
|
|
|
|
pub async fn db_pool_history(
|
|
pool: &SqlitePool,
|
|
pool_id: &PoolID,
|
|
start_time: u64,
|
|
) -> Result<Vec<PoolHistoryEntry>> {
|
|
let query = "SELECT
|
|
phe.txid,
|
|
phe.sats,
|
|
phe.token_amount,
|
|
phe.effective_timestamp as timestamp
|
|
FROM
|
|
pool_history_entry phe
|
|
WHERE
|
|
phe.pool = ?1
|
|
AND timestamp >= ?2
|
|
ORDER BY
|
|
phe.sequence ASC;
|
|
";
|
|
|
|
let rows = sqlx::query(query)
|
|
.bind(pool_id.to_blob())
|
|
.bind(start_time as i64)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
|
|
let mut history: Vec<PoolHistoryEntry> = Vec::default();
|
|
|
|
for row in rows {
|
|
let txid_blob: Vec<u8> = row.get(0);
|
|
let txid = blob_to_display_hex::<bitcoincash::Txid>(&txid_blob)?;
|
|
let sats: i64 = row.get(1);
|
|
let token_amount: i64 = row.get(2);
|
|
let timestamp: i64 = row.get(3);
|
|
let sats = sats as u64;
|
|
let token_amount = token_amount as u64;
|
|
|
|
let k = Integer::from(sats) * Integer::from(token_amount);
|
|
|
|
history.push(PoolHistoryEntry {
|
|
txid,
|
|
sats,
|
|
token_amount,
|
|
timestamp: timestamp as u64,
|
|
k,
|
|
});
|
|
}
|
|
|
|
Ok(history)
|
|
}
|
|
|
|
pub async fn db_pool_get_details(pool: &SqlitePool, pool_id: &PoolID) -> Result<(String, String)> {
|
|
let row: (Vec<u8>, Vec<u8>) =
|
|
sqlx::query_as("SELECT token_id, owner_pkh FROM pool WHERE creation_utxo = ?1")
|
|
.bind(pool_id.to_blob())
|
|
.fetch_one(pool)
|
|
.await?;
|
|
let token_hex = blob_to_display_hex::<TokenID>(&row.0)?;
|
|
let owner_pkh_hex = hex::encode(&row.1);
|
|
Ok((token_hex, owner_pkh_hex))
|
|
}
|
|
|
|
pub async fn db_pool_id_from_utxo(
|
|
pool: &SqlitePool,
|
|
utxo_hash: &OutPointHash,
|
|
) -> Result<Option<String>> {
|
|
let row: Option<(Vec<u8>,)> =
|
|
sqlx::query_as("SELECT pool FROM pool_history_entry WHERE utxo = ?")
|
|
.bind(utxo_hash.to_blob())
|
|
.fetch_optional(pool)
|
|
.await?;
|
|
|
|
match row {
|
|
Some((pool_blob,)) => {
|
|
let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
|
|
Ok(Some(pool_id))
|
|
}
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
/// Get total volume in satoshis across all tokens for a given time period
|
|
pub async fn get_total_volume_sats(
|
|
pool: &SqlitePool,
|
|
start_timestamp: u64,
|
|
end_timestamp: u64,
|
|
) -> anyhow::Result<i64> {
|
|
let sql = "
|
|
SELECT
|
|
COALESCE(SUM(ABS(phe.sats_delta)), 0) AS total_volume_sats
|
|
FROM pool_history_entry phe
|
|
JOIN pool p ON phe.pool = p.creation_utxo
|
|
JOIN tx ON phe.txid = tx.txid
|
|
WHERE tx.effective_timestamp BETWEEN ? AND ?";
|
|
|
|
let row: (i64,) = sqlx::query_as(sql)
|
|
.bind(start_timestamp as i64)
|
|
.bind(end_timestamp as i64)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
|
|
Ok(row.0)
|
|
}
|
|
|
|
/// Get volume in both satoshis and tokens for a specific token for a given time period
|
|
pub async fn get_token_volume_sats(
|
|
pool: &SqlitePool,
|
|
start_timestamp: u64,
|
|
end_timestamp: u64,
|
|
token_id: &str,
|
|
) -> anyhow::Result<(i64, i64)> {
|
|
// CROSS JOIN forces tx-first join order: filter by timestamp (few rows),
|
|
// then look up phe by txid, then filter pool by token_id.
|
|
// Without CROSS JOIN, the planner starts from pool→phe (924K rows) → tx.
|
|
let sql = "
|
|
SELECT
|
|
COALESCE(SUM(ABS(phe.sats_delta)), 0) AS token_volume_sats,
|
|
COALESCE(SUM(ABS(phe.token_delta)), 0) AS token_volume_tokens
|
|
FROM tx
|
|
CROSS JOIN pool_history_entry phe ON tx.txid = phe.txid
|
|
JOIN pool p ON phe.pool = p.creation_utxo
|
|
WHERE tx.effective_timestamp BETWEEN ? AND ?
|
|
AND p.token_id = ?";
|
|
let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
|
|
let row: (i64, i64) = sqlx::query_as(sql)
|
|
.bind(start_timestamp as i64)
|
|
.bind(end_timestamp as i64)
|
|
.bind(token_blob)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
|
|
Ok(row)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::db::cauldron::prepare_tables;
|
|
use crate::db::cauldron::tx::{insert_block_tx, insert_mempool_tx};
|
|
use crate::db::cauldron::utxo_funding::insert_utxo_funding;
|
|
use bitcoin_hashes::hex::ToHex;
|
|
use bitcoin_hashes::Hash;
|
|
use bitcoincash::{BlockHash, PubkeyHash, Txid};
|
|
use riftenlabs_defi::cauldron::ParsedContract;
|
|
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
|
|
static POOL_TEST_COUNTER: AtomicU64 = AtomicU64::new(0);
|
|
|
|
async fn test_pool() -> SqlitePool {
|
|
let id = POOL_TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
|
|
let uri = format!("file:pool_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_test_db(pool: &SqlitePool) {
|
|
prepare_tables(pool).await;
|
|
dummy_init_seq();
|
|
}
|
|
|
|
/// Helper to insert a simple pool with two history entries for testing get_pool_period_snapshot
|
|
/// Inserts entries at t0 and t1 so the period has non-zero duration
|
|
async fn seed_test_pool(
|
|
pool: &SqlitePool,
|
|
token: TokenID,
|
|
owner_pkh: PubkeyHash,
|
|
t0: i64,
|
|
t1: i64,
|
|
sats: u64,
|
|
token_amount: i64,
|
|
) -> OutPointHash {
|
|
// Generate unique hashes based on owner_pkh to avoid PK conflicts in DB
|
|
let mut seed = [0u8; 32];
|
|
|
|
// Use the hex representation to seed the buffer if into_inner() is missing
|
|
// This ensures a unique, deterministic seed for each owner in the test
|
|
let pkh_hex = owner_pkh.to_hex();
|
|
let pkh_bytes = hex::decode(pkh_hex).unwrap();
|
|
seed[0..20].copy_from_slice(&pkh_bytes);
|
|
|
|
let pool_utxo_0 = OutPointHash::from_inner(seed);
|
|
let mut seed_tx = seed;
|
|
seed_tx[31] = 0xFF; // slightly different for txid
|
|
let txid0 = Txid::from_inner(seed_tx);
|
|
|
|
let txid1 = Txid::from_inner(seed); // different enough for test
|
|
|
|
let contract0 = ParsedContract {
|
|
pkh: owner_pkh,
|
|
is_withdrawn: false,
|
|
spent_utxo_hash: OutPointHash::all_zeros(),
|
|
new_utxo_hash: Some(pool_utxo_0),
|
|
new_utxo_txid: Some(txid0),
|
|
new_utxo_n: Some(0),
|
|
token_id: Some(token),
|
|
sats: Some(sats),
|
|
token_amount: Some(token_amount),
|
|
};
|
|
|
|
let mut conn = pool.acquire().await.unwrap();
|
|
insert_new_pool(&mut conn, &contract0).await.unwrap();
|
|
insert_utxo_funding(&mut conn, &vec![contract0.clone()], &txid0, true)
|
|
.await
|
|
.unwrap();
|
|
insert_block_tx(&mut conn, &txid0, &BlockHash::all_zeros(), t0)
|
|
.await
|
|
.unwrap();
|
|
insert_mempool_tx(&mut conn, &txid0, t0 as u64)
|
|
.await
|
|
.unwrap();
|
|
insert_pool_history_entry(
|
|
&mut conn,
|
|
&pool_utxo_0,
|
|
&contract0,
|
|
Some(t0 as u64),
|
|
Some(t0 as u64),
|
|
0,
|
|
0,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
// Use a different UTXO for the second entry in the history to satisfy DB constraints
|
|
let pool_utxo_1 = OutPointHash::from_inner(seed_tx);
|
|
let contract1 = ParsedContract {
|
|
pkh: owner_pkh,
|
|
is_withdrawn: false,
|
|
spent_utxo_hash: pool_utxo_0,
|
|
new_utxo_hash: Some(pool_utxo_1),
|
|
new_utxo_txid: Some(txid1),
|
|
new_utxo_n: Some(0),
|
|
token_id: Some(token),
|
|
sats: Some(sats + 100),
|
|
token_amount: Some(token_amount - 10),
|
|
};
|
|
|
|
insert_utxo_funding(&mut conn, &vec![contract1.clone()], &txid1, true)
|
|
.await
|
|
.unwrap();
|
|
insert_block_tx(&mut conn, &txid1, &BlockHash::all_zeros(), t1)
|
|
.await
|
|
.unwrap();
|
|
insert_mempool_tx(&mut conn, &txid1, t1 as u64)
|
|
.await
|
|
.unwrap();
|
|
insert_pool_history_entry(
|
|
&mut conn,
|
|
&pool_utxo_0, // Recording history for the original pool
|
|
&contract1,
|
|
Some(t1 as u64),
|
|
Some(t1 as u64),
|
|
100,
|
|
-10,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
pool_utxo_0
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_volume_no_trades_in_period() {
|
|
let pool = test_pool().await;
|
|
setup_test_db(&pool).await;
|
|
|
|
let start_timestamp = 1755508856u64;
|
|
let end_timestamp = 1755595256u64;
|
|
let token_id = "f6677f3d3805d70949b375d36e094ff0ec9ece2a2cb1fde6d8b0e90b368f1f63";
|
|
|
|
let result = get_token_volume_sats(&pool, start_timestamp, end_timestamp, token_id).await;
|
|
let (sats_volume, token_volume) = result.unwrap();
|
|
assert_eq!(sats_volume, 0);
|
|
assert_eq!(token_volume, 0);
|
|
|
|
let result = get_total_volume_sats(&pool, start_timestamp, end_timestamp).await;
|
|
let total_volume = result.unwrap();
|
|
assert_eq!(total_volume, 0);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_pool_period_snapshot_with_token_filter() {
|
|
let db_pool = test_pool().await;
|
|
setup_test_db(&db_pool).await;
|
|
|
|
let token = TokenID::from_inner([0x01; 32]);
|
|
let owner = PubkeyHash::all_zeros();
|
|
let t0 = 1700000000i64;
|
|
let t1 = 1700001000i64;
|
|
|
|
seed_test_pool(&db_pool, token, owner, t0, t1, 1000, 500).await;
|
|
|
|
let token_hex = token.to_hex();
|
|
let result =
|
|
get_pool_period_snapshot(&db_pool, Some(&token_hex), None, t0 - 100, t1 + 100).await;
|
|
|
|
let pools = result.expect("query should not fail");
|
|
assert!(
|
|
!pools.is_empty(),
|
|
"token filter should return matching pools (got 0)"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_pool_period_snapshot_with_pkh_filter() {
|
|
let db_pool = test_pool().await;
|
|
setup_test_db(&db_pool).await;
|
|
|
|
let token = TokenID::from_inner([0x02; 32]);
|
|
let owner = PubkeyHash::from_inner([0x11; 20]);
|
|
let t0 = 1700000000i64;
|
|
let t1 = 1700001000i64;
|
|
|
|
seed_test_pool(&db_pool, token, owner, t0, t1, 2000, 1000).await;
|
|
|
|
let pkh_hex = owner.to_hex();
|
|
let result =
|
|
get_pool_period_snapshot(&db_pool, None, Some(&pkh_hex), t0 - 100, t1 + 100).await;
|
|
|
|
let pools = result.expect("query should not fail");
|
|
assert!(
|
|
!pools.is_empty(),
|
|
"pkh filter should return matching pools (got 0)"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_pool_period_snapshot_by_pool_ids() {
|
|
let db_pool = test_pool().await;
|
|
setup_test_db(&db_pool).await;
|
|
|
|
let token = TokenID::from_inner([0x02; 32]);
|
|
let t0 = 1700000000i64;
|
|
let t1 = 1700001000i64;
|
|
|
|
let owner_1 = PubkeyHash::from_inner([0x11; 20]);
|
|
let owner_2 = PubkeyHash::from_inner([0x22; 20]);
|
|
|
|
let pool_id_1 = seed_test_pool(&db_pool, token, owner_1, t0, t1, 2000, 1000).await;
|
|
let pool_id_2 = seed_test_pool(&db_pool, token, owner_2, t0, t1, 3000, 1500).await;
|
|
|
|
// Filter by a single pool ID
|
|
let result = get_pool_period_snapshot_by_pool_ids(
|
|
&db_pool,
|
|
&[pool_id_1.to_hex()],
|
|
t0 - 100,
|
|
t1 + 100,
|
|
)
|
|
.await;
|
|
let pools = result.expect("single pool_id query should not fail");
|
|
assert_eq!(pools.len(), 1, "Should return exactly 1 pool");
|
|
|
|
// Filter by both pool IDs
|
|
let result = get_pool_period_snapshot_by_pool_ids(
|
|
&db_pool,
|
|
&[pool_id_1.to_hex(), pool_id_2.to_hex()],
|
|
t0 - 100,
|
|
t1 + 100,
|
|
)
|
|
.await;
|
|
let pools = result.expect("multi pool_id query should not fail");
|
|
assert_eq!(pools.len(), 2, "Should return 2 pools for both IDs");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_pool_period_snapshot_pool_id_format() {
|
|
let db_pool = test_pool().await;
|
|
setup_test_db(&db_pool).await;
|
|
|
|
let token = TokenID::from_inner([0x03; 32]);
|
|
let owner = PubkeyHash::all_zeros();
|
|
let t0 = 1700000000i64;
|
|
let t1 = 1700001000i64;
|
|
|
|
let expected_pool_id = seed_test_pool(&db_pool, token, owner, t0, t1, 3000, 1500).await;
|
|
|
|
let result = get_pool_period_snapshot(&db_pool, None, None, t0 - 100, t1 + 100).await;
|
|
|
|
let pools = result.expect("query should not fail");
|
|
assert!(!pools.is_empty(), "should return at least one pool");
|
|
|
|
let (start_pool, _end_pool) = &pools[0];
|
|
|
|
assert_eq!(
|
|
start_pool.pool_id.len(),
|
|
64,
|
|
"pool_id should be 64 hex chars, got: {}",
|
|
start_pool.pool_id
|
|
);
|
|
assert!(
|
|
start_pool.pool_id.chars().all(|c| c.is_ascii_hexdigit()),
|
|
"pool_id should be valid hex, got: {}",
|
|
start_pool.pool_id
|
|
);
|
|
|
|
assert_eq!(
|
|
start_pool.pool_id,
|
|
expected_pool_id.to_hex(),
|
|
"pool_id should match expected"
|
|
);
|
|
}
|
|
}
|