diff --git a/Cargo.lock b/Cargo.lock index eed5447..02011ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -249,9 +249,9 @@ dependencies = [ [[package]] name = "bitcoincash" -version = "0.32.2" +version = "0.32.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a604a00994f6080334adbade057bd3724aacbaf674b1bd85310835ab46edd819" +checksum = "91089268ed5067e7c0973476a0f8e3cc208802ec551950f43643a07b13711228" dependencies = [ "base58ck", "bech32", @@ -261,6 +261,7 @@ dependencies = [ "bitcoin_hashes", "hex-conservative", "hex_lit", + "num-bigint", "secp256k1", "serde", ] @@ -1707,6 +1708,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -2272,13 +2283,12 @@ dependencies = [ [[package]] name = "riftenlabs-defi" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9103375b1b484664066a5958a619f7dd8baab23069d2c0f53fd9df369ff3b34b" +version = "0.4.5" dependencies = [ "anyhow", "bitcoin_hashes", "bitcoincash", + "hex", "serde", ] diff --git a/Cargo.toml b/Cargo.toml index 6ebbcda..3c19fa2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ build = "build.rs" [dependencies] anyhow = { version = "1.0.94", features = ["backtrace"] } -bitcoincash = "0.32.2" +bitcoincash = "0.32.4" bitcoin_hashes = { version = "0.14.100", default-features = false } # Same version as used by bitcoincash electrum-client-netagnostic = { version = "0.21.2", features = ["use-rustls", "use-websocket"] } hex = { version = "0.4.3", features = ["serde"] } @@ -22,7 +22,7 @@ serde_json = "1.0.133" rocket = { version = "0.5.1", features = ["json"] } sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] } rayon = "1.10.0" -riftenlabs-defi = "0.4.1" +riftenlabs-defi = "0.4.5" rocket_cors = "0.6.0" log = "0.4" stderrlog = "0.6.0" diff --git a/src/chain.rs b/src/chain.rs index 93fbf5d..ba11537 100644 --- a/src/chain.rs +++ b/src/chain.rs @@ -71,6 +71,14 @@ impl BlockUndoer for StoreBlockUndoer { { was_indexed = true; } + if db::cauldron::tokenbch::delete_entries_for_block( + &self.db.cauldron_w, + &blockheader.block_hash(), + ) + .await? + { + was_indexed = true; + } if db::bcmr::delete_entries_for_block(&self.db.bcmr_w, &blockheader.block_hash()) .await? { diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index 644a5a9..1c5361e 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -22,6 +22,7 @@ pub mod pool; pub mod poolvisitor; pub mod priceseries; pub mod spot; +pub mod tokenbch; pub mod tokenlist; pub mod tokentoken; pub mod tx; @@ -50,6 +51,7 @@ pub async fn prepare_tables(pool: &SqlitePool) { pool::create_table(pool).await; tokentoken::create_table(pool).await; + tokenbch::create_table(pool).await; sqlx::query( "CREATE INDEX idx_utxo_funding_join ON utxo_funding(new_utxo_hash, sats, token_id);", diff --git a/src/db/cauldron/tokenbch.rs b/src/db/cauldron/tokenbch.rs new file mode 100644 index 0000000..7ca035b --- /dev/null +++ b/src/db/cauldron/tokenbch.rs @@ -0,0 +1,736 @@ +// 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 + +//! Indexing of token <-> native-BCH (`TokenBch`) delegation pools. +//! +//! A pool is a SINGLE bare-P2S UTXO holding both reserves — BCH in its satoshi +//! value, the token in its CashToken — with the per-pool config prepended to +//! the generic dispatcher (`0x3f OP_DROP `). States +//! are parsed by [`riftenlabs_defi::tokenbch_delegation`]; a pool is admitted +//! by the constant dispatcher tail alone (the platform NFTH pins the logic +//! hash — see [`crate::db::orbconstants`]). +//! +//! The fee side is per-pool runtime config (`fee_paid_in_token`): the accrued +//! `owed` is satoshis when false, token base units when true. `owed` and +//! `platform_fee_rate` are the only mutable config fields (per-state); the +//! rest is pinned per-pool at creation. LP teardowns are not +//! output-recognizable — ingestion passes a [`TeardownProbe`] for each +//! blob-presenting tx and a known pool UTXO spent without a successor state is +//! flagged withdrawn (mirroring [`crate::db::cauldron::tokentoken`], which +//! documents the shared mempool/block reconciliation and tx-FK cleanup +//! semantics). + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicI64, Ordering}; + +use anyhow::Result; + +use bitcoincash::{BlockHash, TokenID, Txid}; +use log::info; +use riftenlabs_defi::{chainutil::OutPointHash, tokenbch_delegation::ParsedTokenBchPool}; +use sqlx::{Row, SqliteConnection, SqlitePool}; + +use crate::db::blob::{blob_to_display_hex, FromBlob, ToBlob}; +use crate::db::cauldron::tokentoken::{serialize_u64_as_string, TeardownProbe}; + +/// Idempotent: also run as an always-on migration for existing databases. +/// (The abandoned draft schema never reached production; nothing to migrate.) +pub async fn create_table(pool: &SqlitePool) { + sqlx::query( + "CREATE TABLE IF NOT EXISTS tokenbch_pool ( + creation_utxo BLOB PRIMARY KEY, + nft_owner BLOB NOT NULL, + token_id BLOB NOT NULL, + pool_fee_rate INT NOT NULL, + min_fee INT NOT NULL, + virtual_x BIGINT NOT NULL, + virtual_y BIGINT NOT NULL, + fee_paid_in_token TINYINT NOT NULL, + withdrawn_in_txid BLOB REFERENCES tx(txid) ON DELETE SET NULL + )", + ) + .execute(pool) + .await + .expect("failed to create table tokenbch_pool"); + + sqlx::query( + "CREATE TABLE IF NOT EXISTS tokenbch_pool_history_entry ( + utxo BLOB PRIMARY KEY, + pool BLOB NOT NULL REFERENCES tokenbch_pool(creation_utxo) ON DELETE CASCADE, + spent_utxo BLOB, + txid BLOB NOT NULL 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, + owed BIGINT NOT NULL, + platform_fee_rate INT NOT NULL, + sats_delta BIGINT NOT NULL, + token_delta BIGINT NOT NULL + )", + ) + .execute(pool) + .await + .expect("failed to create table tokenbch_pool_history_entry"); + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_tb_history_pool_sequence ON tokenbch_pool_history_entry(pool, sequence)", + ) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_tb_history_spent ON tokenbch_pool_history_entry(spent_utxo)", + ) + .execute(pool) + .await + .unwrap(); + sqlx::query("CREATE INDEX IF NOT EXISTS idx_tb_pool_token ON tokenbch_pool(token_id)") + .execute(pool) + .await + .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_tb_history_txid ON tokenbch_pool_history_entry(txid)", + ) + .execute(pool) + .await + .unwrap(); +} + +/// Next sequence number in the `tokenbch_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 tokenbch_pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + NEXT_SEQUENCE.store(row.0, Ordering::SeqCst); +} + +#[allow(dead_code)] +pub fn dummy_init_seq() { + if NEXT_SEQUENCE.load(Ordering::SeqCst) < 0 { + NEXT_SEQUENCE.store(42, Ordering::SeqCst); + } +} + +/// The pool (creation utxo) whose state is recorded at `utxo_hash`. +async fn get_pool_by_utxo( + conn: &mut SqliteConnection, + utxo_hash: &OutPointHash, +) -> Result> { + let row: Option<(Vec,)> = + sqlx::query_as("SELECT pool FROM tokenbch_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)?)), + None => Ok(None), + } +} + +/// Latest (sats, token_amount) for the pool state recorded at `utxo_hash`. +async fn get_reserves_at( + conn: &mut SqliteConnection, + utxo_hash: &OutPointHash, +) -> Result<(i64, i64)> { + let row = + sqlx::query("SELECT sats, token_amount FROM tokenbch_pool_history_entry WHERE utxo = ?") + .bind(utxo_hash.to_blob()) + .fetch_optional(&mut *conn) + .await?; + match row { + Some(r) => Ok((r.get(0), r.get(1))), + None => Ok((0, 0)), + } +} + +/// Insert the immutable per-pool row. A no-op when the pool already exists +/// (a mempool creation confirming in a block replays the same state). +pub async fn insert_new_pool(conn: &mut SqliteConnection, pool: &ParsedTokenBchPool) -> Result<()> { + sqlx::query( + "INSERT INTO tokenbch_pool + (creation_utxo, nft_owner, token_id, pool_fee_rate, min_fee, + virtual_x, virtual_y, fee_paid_in_token, withdrawn_in_txid) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL) + ON CONFLICT(creation_utxo) DO NOTHING", + ) + .bind(pool.new_utxo_hash.to_blob()) + .bind(pool.config.nft_owner.to_vec()) + .bind(pool.token_id.to_blob()) + .bind(pool.config.pool_fee_rate as i64) + .bind(pool.config.min_fee as i64) + .bind(pool.config.virtual_x as i64) + .bind(pool.config.virtual_y as i64) + .bind(pool.config.fee_paid_in_token as i64) + .execute(&mut *conn) + .await + .map_err(|e| anyhow::anyhow!("failed to insert tokenbch pool: {e:?}"))?; + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +pub async fn insert_history_entry( + conn: &mut SqliteConnection, + creation_utxo: &OutPointHash, + pool: &ParsedTokenBchPool, + spent_utxo: Option<&OutPointHash>, + mtp_timestamp: Option, + first_seen_timestamp: Option, + sats_delta: i64, + token_delta: i64, +) -> Result<()> { + let next_seq = NEXT_SEQUENCE.fetch_add(1, Ordering::SeqCst); + assert!(next_seq >= 0, "tokenbch sequence not initialized"); + + sqlx::query( + "INSERT INTO tokenbch_pool_history_entry + (utxo, pool, spent_utxo, txid, tx_pos, + mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, + owed, platform_fee_rate, sats_delta, token_delta) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(utxo) DO UPDATE SET + pool = excluded.pool, + spent_utxo = excluded.spent_utxo, + txid = excluded.txid, + tx_pos = excluded.tx_pos, + mtp_timestamp = COALESCE(excluded.mtp_timestamp, tokenbch_pool_history_entry.mtp_timestamp), + first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, tokenbch_pool_history_entry.first_seen_timestamp), + sequence = excluded.sequence", + ) + .bind(pool.new_utxo_hash.to_blob()) + .bind(creation_utxo.to_blob()) + .bind(spent_utxo.map(|h| h.to_blob())) + .bind(pool.new_utxo_txid.to_blob()) + .bind(pool.new_utxo_n as i64) + .bind(mtp_timestamp.map(|t| t as i64)) + .bind(first_seen_timestamp.map(|t| t as i64)) + .bind(next_seq) + .bind(pool.sats as i64) + .bind(pool.token_amount) + .bind(pool.config.owed as i64) + .bind(pool.config.platform_fee_rate as i64) + .bind(sats_delta) + .bind(token_delta) + .execute(&mut *conn) + .await + .map_err(|e| anyhow::anyhow!("failed to insert tokenbch history entry: {e:?}"))?; + Ok(()) +} + +/// Apply a batch of parsed TokenBch states plus the batch's teardown probes. +/// +/// A state's `spent_utxo_hash` is only a CANDIDATE (the index-aligned input, +/// present even at creation): it links the state to a pool iff it matches a +/// recorded state, otherwise the state opens a new pool. The queue defers +/// states whose candidate parent is created later in the same batch. +/// +/// After the states are applied, each probe's spent outpoints are checked +/// against recorded pool states: a known pool UTXO spent by a blob-presenting +/// tx with no successor state is an LP teardown and flags the pool withdrawn. +pub async fn update_tokenbch_pool_history( + conn: &mut SqliteConnection, + pools: Vec, + teardown_probes: &[TeardownProbe], + mtp_timestamp: Option, + first_seen_timestamp: Option, +) -> Result<()> { + let mut queue = VecDeque::from(pools); + + while let Some(current) = queue.pop_front() { + let linked_parent = match current.spent_utxo_hash { + None => None, + Some(candidate) => match get_pool_by_utxo(&mut *conn, &candidate).await? { + Some(creation) => Some((candidate, creation)), + None => { + let has_parent = queue.iter().any(|p| candidate == p.new_utxo_hash); + if has_parent { + queue.push_back(current); + continue; + } + // Unknown candidate: an ordinary funding input at the + // index — this state is a creation. + None + } + }, + }; + + match linked_parent { + None => { + info!("TokenBch pool created in tx {}", current.new_utxo_txid); + insert_new_pool(&mut *conn, ¤t).await?; + insert_history_entry( + &mut *conn, + ¤t.new_utxo_hash, + ¤t, + None, + mtp_timestamp, + first_seen_timestamp, + 0, + 0, + ) + .await?; + } + Some((parent, creation_utxo)) => { + let (prev_sats, prev_token) = get_reserves_at(&mut *conn, &parent).await?; + insert_history_entry( + &mut *conn, + &creation_utxo, + ¤t, + Some(&parent), + mtp_timestamp, + first_seen_timestamp, + current.sats as i64 - prev_sats, + current.token_amount - prev_token, + ) + .await?; + } + } + } + + // Teardown pass: a known pool UTXO among a probe's spent outpoints, with + // no successor state recorded, closes the pool. + for probe in teardown_probes { + for spent in &probe.spent_utxo_hashes { + let r = sqlx::query( + "UPDATE tokenbch_pool SET withdrawn_in_txid = ?1 + WHERE withdrawn_in_txid IS NULL + AND creation_utxo IN ( + SELECT h.pool FROM tokenbch_pool_history_entry h + WHERE h.utxo = ?2 + AND NOT EXISTS ( + SELECT 1 FROM tokenbch_pool_history_entry s + WHERE s.spent_utxo = ?2 + ) + )", + ) + .bind(probe.txid.to_blob()) + .bind(spent.to_blob()) + .execute(&mut *conn) + .await?; + if r.rows_affected() > 0 { + info!("TokenBch pool torn down in tx {}", probe.txid); + } + } + } + + Ok(()) +} + +/// Remove TokenBch state recorded in the given block (chain reorg). +/// +/// Order-independent w.r.t. how reorged blocks are replayed: deletes the +/// block's history rows, drops pools left with no history (their creation was +/// undone), and reverts teardowns flagged in the block. +pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) -> Result { + let r1 = sqlx::query( + "DELETE FROM tokenbch_pool_history_entry + WHERE txid IN (SELECT txid FROM tx WHERE blockhash = ?)", + ) + .bind(blockhash.to_blob()) + .execute(pool) + .await?; + + sqlx::query( + "DELETE FROM tokenbch_pool + WHERE creation_utxo NOT IN (SELECT DISTINCT pool FROM tokenbch_pool_history_entry)", + ) + .execute(pool) + .await?; + + let r3 = sqlx::query( + "UPDATE tokenbch_pool SET withdrawn_in_txid = NULL + WHERE withdrawn_in_txid IN (SELECT txid FROM tx WHERE blockhash = ?)", + ) + .bind(blockhash.to_blob()) + .execute(pool) + .await?; + + Ok(r1.rows_affected() + r3.rows_affected() != 0) +} + +/// A single active TokenBch pool at its latest state, for the RPC layer. +#[derive(serde::Serialize)] +pub struct ActiveTokenBchPool { + pub pool_id: String, + pub nft_owner: String, + pub token_id: String, + /// RAW satoshi value — the BCH reserve. When the fee side is BCH + /// (`fee_paid_in_token` false), includes `owed`. + #[serde(serialize_with = "serialize_u64_as_string")] + pub sats: u64, + /// RAW token amount. When `fee_paid_in_token`, includes `owed`. + #[serde(serialize_with = "serialize_u64_as_string")] + pub token_amount: u64, + /// Accrued platform fee on the fee side, excluded from the tradeable + /// reserve. + #[serde(serialize_with = "serialize_u64_as_string")] + pub owed: u64, + /// Fee side: false = fees accrue in BCH satoshis, true = in token units. + pub fee_paid_in_token: bool, + /// Platform's cut, parts-per-1_000_000 (mutable by a platform sweep). + pub platform_fee_rate: u32, + /// LP's cut, parts-per-100_000. + pub pool_fee_rate: u32, + /// Minimum combined (platform + pool) fee in fee-side base units. + pub min_fee: u64, + /// Virtual reserve offset, ALWAYS the BCH side. + #[serde(serialize_with = "serialize_u64_as_string")] + pub virtual_x: u64, + /// Virtual reserve offset, ALWAYS the token side. + #[serde(serialize_with = "serialize_u64_as_string")] + pub virtual_y: u64, + /// Transaction that created the pool's current state. + pub txid: String, + /// Output index of the pool UTXO. + pub vout: u32, +} + +const ACTIVE_POOL_SELECT: &str = + "SELECT p.creation_utxo, p.nft_owner, p.token_id, p.pool_fee_rate, p.min_fee, + p.virtual_x, p.virtual_y, p.fee_paid_in_token, + phe.sats, phe.token_amount, phe.owed, phe.platform_fee_rate, + phe.txid, phe.tx_pos + FROM tokenbch_pool p + JOIN tokenbch_pool_history_entry phe ON phe.pool = p.creation_utxo + AND phe.sequence = ( + SELECT MAX(sequence) FROM tokenbch_pool_history_entry WHERE pool = p.creation_utxo + ) + WHERE p.withdrawn_in_txid IS NULL"; + +fn row_to_active_pool(row: &sqlx::sqlite::SqliteRow) -> Result { + let pool_blob: Vec = row.get(0); + let nft_owner_blob: Vec = row.get(1); + let token_blob: Vec = row.get(2); + let pool_fee_rate: i64 = row.get(3); + let min_fee: i64 = row.get(4); + let virtual_x: i64 = row.get(5); + let virtual_y: i64 = row.get(6); + let fee_paid_in_token: i64 = row.get(7); + let sats: i64 = row.get(8); + let token_amount: i64 = row.get(9); + let owed: i64 = row.get(10); + let platform_fee_rate: i64 = row.get(11); + let txid_blob: Vec = row.get(12); + let tx_pos: i64 = row.get(13); + + Ok(ActiveTokenBchPool { + pool_id: blob_to_display_hex::(&pool_blob)?, + nft_owner: hex::encode(&nft_owner_blob), + token_id: blob_to_display_hex::(&token_blob)?, + sats: sats as u64, + token_amount: token_amount as u64, + owed: owed as u64, + fee_paid_in_token: fee_paid_in_token != 0, + platform_fee_rate: platform_fee_rate as u32, + pool_fee_rate: pool_fee_rate as u32, + min_fee: min_fee as u64, + virtual_x: virtual_x as u64, + virtual_y: virtual_y as u64, + txid: blob_to_display_hex::(&txid_blob)?, + vout: tx_pos as u32, + }) +} + +/// Active pools holding the given token, each at its latest state. +pub async fn db_active_pools_for_token( + pool: &SqlitePool, + token: &TokenID, +) -> Result> { + let sql = format!("{ACTIVE_POOL_SELECT} AND p.token_id = ?1"); + let rows = sqlx::query(&sql) + .bind(token.to_blob()) + .fetch_all(pool) + .await?; + rows.iter().map(row_to_active_pool).collect() +} + +/// Every active pool at its latest state, with no token filter. +pub async fn db_all_active_pools(pool: &SqlitePool) -> Result> { + let rows = sqlx::query(ACTIVE_POOL_SELECT).fetch_all(pool).await?; + rows.iter().map(row_to_active_pool).collect() +} + +/// Distinct token ids that appear in at least one active TokenBch pool. +pub async fn db_pool_tokens(pool: &SqlitePool) -> Result> { + let rows = + sqlx::query("SELECT DISTINCT token_id FROM tokenbch_pool WHERE withdrawn_in_txid IS NULL") + .fetch_all(pool) + .await?; + + let mut out = Vec::with_capacity(rows.len()); + for row in rows { + let blob: Vec = row.get(0); + out.push(blob_to_display_hex::(&blob)?); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::cauldron::prepare_tables; + use crate::db::cauldron::tx::insert_block_tx; + + use bitcoin_hashes::Hash; + use bitcoincash::{BlockHash, Txid}; + use riftenlabs_defi::tokenbch_delegation::DelegationConfig; + use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + use std::sync::atomic::{AtomicU64, Ordering as AOrdering}; + + static TB_TEST_COUNTER: AtomicU64 = AtomicU64::new(0); + + async fn test_db() -> SqlitePool { + let id = TB_TEST_COUNTER.fetch_add(1, AOrdering::SeqCst); + let uri = format!("file:tb_test_{}?mode=memory&cache=shared", id); + let opts = SqliteConnectOptions::new() + .filename(&uri) + .foreign_keys(false); + let pool = SqlitePoolOptions::new().connect_with(opts).await.unwrap(); + prepare_tables(&pool).await; + dummy_init_seq(); + pool + } + + /// The always-run startup migration calls `create_table` on databases + /// that already have the tables; it must be a no-op then. + #[tokio::test] + async fn create_table_is_idempotent() { + let pool = test_db().await; + create_table(&pool).await; + } + + fn oph(b: u8) -> OutPointHash { + OutPointHash::from_byte_array([b; 32]) + } + fn tid(b: u8) -> TokenID { + TokenID::from_byte_array([b; 32]) + } + fn txid(b: u8) -> Txid { + Txid::from_byte_array([b; 32]) + } + + fn config(owed: u64, fee_paid_in_token: bool) -> DelegationConfig { + DelegationConfig { + owed, + platform_fee_rate: 600, + pool_fee_rate: 300, + min_fee: 1, + virtual_x: 0, + virtual_y: 0, + nft_owner: [0x11; 32], + fee_paid_in_token, + } + } + + /// A creation state: the candidate spend points at an unrelated funding + /// outpoint no consumer knows. + fn creation(new: u8, txid_b: u8, sats: u64, token: i64) -> ParsedTokenBchPool { + ParsedTokenBchPool { + config: config(0, false), + spent_utxo_hash: Some(oph(new.wrapping_add(200))), + new_utxo_hash: oph(new), + new_utxo_txid: txid(txid_b), + new_utxo_n: 0, + token_id: tid(0xBB), + token_amount: token, + sats, + } + } + + /// A swap spending `prev`, producing `new` in tx `txid_b`. + fn swap(prev: u8, new: u8, txid_b: u8, sats: u64, token: i64) -> ParsedTokenBchPool { + let mut p = creation(new, txid_b, sats, token); + p.spent_utxo_hash = Some(oph(prev)); + p.config.owed = 1; + p + } + + async fn seed_tx(pool: &SqlitePool, txid_val: &Txid, blockhash: &BlockHash) { + let mut conn = pool.acquire().await.unwrap(); + insert_block_tx(&mut conn, txid_val, blockhash, 1_700_000_000) + .await + .unwrap(); + } + + /// A state first seen in the mempool (first_seen only) and later confirmed + /// in a block (mtp only) must keep the mempool first_seen timestamp. + #[tokio::test] + async fn mempool_then_block_keeps_first_seen() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c = creation(1, 10, 1_000_000, 2_000_000); + seed_tx(&pool, &c.new_utxo_txid, &block).await; + + let mut conn = pool.acquire().await.unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c.clone()], &[], None, Some(1_700_000_100)) + .await + .unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c], &[], Some(1_700_000_500), None) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT mtp_timestamp, first_seen_timestamp, effective_timestamp + FROM tokenbch_pool_history_entry", + ) + .fetch_one(&pool) + .await + .unwrap(); + let mtp: Option = row.get(0); + let first_seen: Option = row.get(1); + let effective: Option = row.get(2); + assert_eq!(mtp, Some(1_700_000_500)); + assert_eq!(first_seen, Some(1_700_000_100)); + assert_eq!(effective, Some(1_700_000_100), "first_seen wins"); + } + + #[tokio::test] + async fn creation_then_swap_tracks_reserves_and_token_query() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c = creation(1, 10, 1_000_000, 2_000_000); + seed_tx(&pool, &c.new_utxo_txid, &block).await; + let s = swap(1, 2, 11, 1_000_500, 1_999_002); + seed_tx(&pool, &s.new_utxo_txid, &block).await; + + let mut conn = pool.acquire().await.unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c, s], &[], Some(1_700_000_000), None) + .await + .unwrap(); + drop(conn); + + let pools = db_active_pools_for_token(&pool, &tid(0xBB)).await.unwrap(); + assert_eq!(pools.len(), 1); + assert_eq!(pools[0].sats, 1_000_500); + assert_eq!(pools[0].token_amount, 1_999_002); + assert_eq!(pools[0].owed, 1, "owed from the latest state"); + assert!(!pools[0].fee_paid_in_token); + assert_eq!(pools[0].platform_fee_rate, 600); + assert_eq!(pools[0].pool_fee_rate, 300); + assert_eq!(pools[0].vout, 0); + + // Deltas recorded against the previous raw reserves. + let row = sqlx::query( + "SELECT sats_delta, token_delta FROM tokenbch_pool_history_entry WHERE utxo = ?", + ) + .bind(oph(2).to_blob()) + .fetch_one(&pool) + .await + .unwrap(); + let (ds, dt): (i64, i64) = (row.get(0), row.get(1)); + assert_eq!((ds, dt), (500, -998)); + + let tokens = db_pool_tokens(&pool).await.unwrap(); + assert_eq!(tokens, vec![tid(0xBB).to_string()]); + } + + /// The candidate spend at creation (an unrelated funding outpoint) must + /// NOT link two independent pools that happen to be created in sequence. + #[tokio::test] + async fn unknown_candidate_spend_opens_a_new_pool() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c1 = creation(1, 10, 1_000_000, 2_000_000); + let mut c2 = creation(2, 11, 3_000_000, 4_000_000); + c2.token_id = tid(0xCC); + seed_tx(&pool, &c1.new_utxo_txid, &block).await; + seed_tx(&pool, &c2.new_utxo_txid, &block).await; + + let mut conn = pool.acquire().await.unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c1, c2], &[], Some(1), None) + .await + .unwrap(); + drop(conn); + + let n: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM tokenbch_pool") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(n.0, 2, "two independent pools"); + } + + /// A teardown is a blob-presenting tx spending a known pool UTXO with no + /// successor: the probe flags the pool withdrawn; a reorg of that block + /// restores it; a reorg of the creation removes the pool entirely. + #[tokio::test] + async fn teardown_probe_hides_pool_and_reorg_restores_it() { + let pool = test_db().await; + let block_a = BlockHash::from_byte_array([0xA1; 32]); + let block_b = BlockHash::from_byte_array([0xB2; 32]); + + let c = creation(1, 10, 1_000_000, 2_000_000); + seed_tx(&pool, &c.new_utxo_txid, &block_a).await; + + let mut conn = pool.acquire().await.unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c], &[], Some(1), None) + .await + .unwrap(); + + let teardown_txid = txid(20); + seed_tx(&pool, &teardown_txid, &block_b).await; + let probe = TeardownProbe { + txid: teardown_txid, + spent_utxo_hashes: vec![oph(0xEE), oph(1)], + }; + update_tokenbch_pool_history(&mut conn, vec![], &[probe], Some(2), None) + .await + .unwrap(); + drop(conn); + + assert!( + db_all_active_pools(&pool).await.unwrap().is_empty(), + "torn-down pool should not be active" + ); + + // Reorg the teardown block: pool becomes active again. + delete_entries_for_block(&pool, &block_b).await.unwrap(); + assert_eq!(db_all_active_pools(&pool).await.unwrap().len(), 1); + + // Reorg the creation block: pool disappears entirely. + delete_entries_for_block(&pool, &block_a).await.unwrap(); + assert!(db_all_active_pools(&pool).await.unwrap().is_empty()); + assert!(db_pool_tokens(&pool).await.unwrap().is_empty()); + } + + /// A probe outpoint consumed by a successor state in the SAME batch is a + /// swap, not a teardown. + #[tokio::test] + async fn swap_probe_is_not_a_teardown() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c = creation(1, 10, 1_000_000, 2_000_000); + let s = swap(1, 2, 11, 1_000_500, 1_999_002); + seed_tx(&pool, &c.new_utxo_txid, &block).await; + seed_tx(&pool, &s.new_utxo_txid, &block).await; + + let probe = TeardownProbe { + txid: txid(11), + spent_utxo_hashes: vec![oph(0x22), oph(1)], + }; + + let mut conn = pool.acquire().await.unwrap(); + update_tokenbch_pool_history(&mut conn, vec![c, s], &[probe], Some(1), None) + .await + .unwrap(); + drop(conn); + + assert_eq!( + db_all_active_pools(&pool).await.unwrap().len(), + 1, + "swapped pool must stay active" + ); + } +} diff --git a/src/db/cauldron/tokentoken.rs b/src/db/cauldron/tokentoken.rs index 473c8f0..9d4c804 100644 --- a/src/db/cauldron/tokentoken.rs +++ b/src/db/cauldron/tokentoken.rs @@ -3,41 +3,103 @@ // 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 -//! Indexing of native token-A <-> token-B (`TokenToken`) AMM pools. +//! Indexing of token-A <-> token-B (`TokenToken`) delegation pools. //! -//! Mirrors [`crate::db::cauldron::pool`] but tracks both token reserves of a -//! pool (token A in the main UTXO, token B in the linked storage UTXO). Tables -//! live in `cauldron.db` so ingestion shares the block write-transaction and -//! `KEY_LAST_INDEXED` checkpoint. +//! A pool is two co-created bare-P2S UTXOs: a generic thin main (token A, +//! identical locking for every pool on the platform NFTH) and a per-pool +//! storage sibling (token B + the config) at `main_vout + 1`. States are +//! parsed by [`riftenlabs_defi::tokentoken_delegation`]; a pool is admitted by +//! the constant locking alone (the platform NFTH pins the logic hash, hence +//! the thin main — see [`crate::db::orbconstants`]). //! -//! Like cauldron pools, states are indexed both from confirmed blocks (with -//! `mtp_timestamp`) and from the mempool (with `first_seen_timestamp`); the -//! history upsert reconciles the two when a mempool tx confirms. +//! Swaps, platform sweeps and creations are output-recognizable and arrive as +//! [`ParsedDelegationPool`] states (`owed_a` and `platform_fee_rate` are the +//! only mutable config fields — per-state; the rest is pinned per-pool at +//! creation). LP teardowns are NOT output-recognizable: every pool spend must +//! present the logic blob in some input, so ingestion passes a +//! [`TeardownProbe`] for each blob-presenting tx and a known main UTXO spent +//! without a successor state is flagged withdrawn. +//! +//! Tables live in `cauldron.db` so ingestion shares the block +//! write-transaction and `KEY_LAST_INDEXED` checkpoint. Like cauldron pools, +//! states are indexed both from confirmed blocks (with `mtp_timestamp`) and +//! from the mempool (with `first_seen_timestamp`); the history upsert +//! reconciles the two when a mempool tx confirms. History rows reference +//! `tx(txid)` ON DELETE CASCADE, so an evicted mempool tx drops its states +//! (and ON DELETE SET NULL reverts a teardown flagged from an evicted tx). use std::collections::VecDeque; use std::sync::atomic::{AtomicI64, Ordering}; use anyhow::Result; -use bitcoincash::{BlockHash, TokenID}; +use bitcoincash::{BlockHash, TokenID, Txid}; use log::info; -use riftenlabs_defi::{chainutil::OutPointHash, tokentoken::ParsedTokenToken}; +use riftenlabs_defi::{chainutil::OutPointHash, tokentoken_delegation::ParsedDelegationPool}; use sqlx::{Row, SqliteConnection, SqlitePool}; use crate::db::blob::{blob_to_display_hex, FromBlob, ToBlob}; -/// Idempotent: also run as an always-on migration for existing databases -/// (the tables were added after `DB_VERSION` 5 without a version bump). +/// A blob-presenting transaction's spent outpoints, probed for LP teardowns: +/// a known pool main UTXO among them, spent with no successor state, closes +/// the pool. Shared with the tokenbch indexer. +#[derive(Debug, Clone)] +pub struct TeardownProbe { + /// The probing (blob-presenting) transaction. + pub txid: Txid, + /// Outpoint hashes of ALL its inputs. + pub spent_utxo_hashes: Vec, +} + +/// True if `table` exists but lacks `required_column` — i.e. it is a leftover +/// from an incompatible schema generation. +async fn table_is_outdated(pool: &SqlitePool, table: &str, required_column: &str) -> bool { + let exists: (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name = ?") + .bind(table) + .fetch_one(pool) + .await + .expect("failed to query sqlite_master"); + if exists.0 == 0 { + return false; + } + let has: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM pragma_table_info(?) WHERE name = ?") + .bind(table) + .bind(required_column) + .fetch_one(pool) + .await + .expect("failed to query table info"); + has.0 == 0 +} + +/// Idempotent: also run as an always-on migration for existing databases. +/// +/// The pre-delegation AMM tables (and the old delegation drafts) are dropped +/// wholesale when detected — the old TokenToken contract never reached +/// production, so there is nothing to migrate. pub async fn create_table(pool: &SqlitePool) { + if table_is_outdated(pool, "tokentoken_pool", "pool_fee_rate").await { + sqlx::query("DROP TABLE IF EXISTS tokentoken_pool_history_entry") + .execute(pool) + .await + .expect("failed to drop outdated tokentoken history table"); + sqlx::query("DROP TABLE IF EXISTS tokentoken_pool") + .execute(pool) + .await + .expect("failed to drop outdated tokentoken pool table"); + } + sqlx::query( "CREATE TABLE IF NOT EXISTS tokentoken_pool ( creation_utxo BLOB PRIMARY KEY, nft_owner BLOB NOT NULL, token_a_id BLOB NOT NULL, token_b_id BLOB NOT NULL, - fee_rate INT NOT NULL, + pool_fee_rate INT NOT NULL, min_fee INT NOT NULL, - withdrawn_in_txid BLOB + virtual_x BIGINT NOT NULL, + virtual_y BIGINT NOT NULL, + withdrawn_in_txid BLOB REFERENCES tx(txid) ON DELETE SET NULL )", ) .execute(pool) @@ -48,18 +110,18 @@ pub async fn create_table(pool: &SqlitePool) { "CREATE TABLE IF NOT EXISTS tokentoken_pool_history_entry ( utxo BLOB PRIMARY KEY, pool BLOB NOT NULL REFERENCES tokentoken_pool(creation_utxo) ON DELETE CASCADE, + spent_main_utxo BLOB, storage_utxo BLOB NOT NULL, - token_a_id BLOB NOT NULL, - token_b_id BLOB NOT NULL, - txid BLOB NOT NULL, + txid BLOB NOT NULL REFERENCES tx(txid) ON DELETE CASCADE, tx_pos INT NOT NULL, - storage_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, reserve_a BIGINT NOT NULL, reserve_b BIGINT NOT NULL, + owed_a BIGINT NOT NULL, + platform_fee_rate INT NOT NULL, main_sats BIGINT NOT NULL, storage_sats BIGINT NOT NULL, reserve_a_delta BIGINT NOT NULL, @@ -76,6 +138,12 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_tt_history_spent_main ON tokentoken_pool_history_entry(spent_main_utxo)", + ) + .execute(pool) + .await + .unwrap(); sqlx::query( "CREATE INDEX IF NOT EXISTS idx_tt_pool_pair ON tokentoken_pool(token_a_id, token_b_id)", ) @@ -109,6 +177,7 @@ pub fn dummy_init_seq() { } } +/// The pool (creation utxo) whose state is recorded at `utxo_hash`. async fn get_pool_by_main_utxo( conn: &mut SqliteConnection, utxo_hash: &OutPointHash, @@ -141,50 +210,38 @@ async fn get_reserves_at( } } -pub async fn insert_new_pool(conn: &mut SqliteConnection, pool: &ParsedTokenToken) -> Result<()> { +/// Insert the immutable per-pool row. A no-op when the pool already exists +/// (a mempool creation confirming in a block replays the same state). +pub async fn insert_new_pool( + conn: &mut SqliteConnection, + pool: &ParsedDelegationPool, +) -> Result<()> { sqlx::query( - "INSERT OR REPLACE INTO tokentoken_pool - (creation_utxo, nft_owner, token_a_id, token_b_id, fee_rate, min_fee, withdrawn_in_txid) - VALUES (?, ?, ?, ?, ?, ?, NULL)", + "INSERT INTO tokentoken_pool + (creation_utxo, nft_owner, token_a_id, token_b_id, pool_fee_rate, min_fee, + virtual_x, virtual_y, withdrawn_in_txid) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL) + ON CONFLICT(creation_utxo) DO NOTHING", ) - .bind( - pool.new_main_utxo_hash - .expect("new pool main utxo missing") - .to_blob(), - ) - .bind(pool.nft_owner.to_vec()) - .bind(pool.token_a_id.expect("token a id missing").to_blob()) - .bind(pool.token_b_id.expect("token b id missing").to_blob()) - .bind(pool.fee_rate as i64) - .bind(pool.min_fee as i64) + .bind(pool.new_main_utxo_hash.to_blob()) + .bind(pool.config.nft_owner.to_vec()) + .bind(pool.token_a_id.to_blob()) + .bind(pool.token_b_id.to_blob()) + .bind(pool.config.pool_fee_rate as i64) + .bind(pool.config.min_fee as i64) + .bind(pool.config.virtual_x as i64) + .bind(pool.config.virtual_y as i64) .execute(&mut *conn) .await .map_err(|e| anyhow::anyhow!("failed to insert tokentoken pool: {e:?}"))?; Ok(()) } -pub async fn flag_as_withdrawn( - conn: &mut SqliteConnection, - creation_utxo: &OutPointHash, - pool: &ParsedTokenToken, -) -> Result<()> { - let txid = pool - .new_main_utxo_txid - .expect("withdraw must carry the withdrawing txid"); - sqlx::query("UPDATE tokentoken_pool SET withdrawn_in_txid = ? WHERE creation_utxo = ?") - .bind(txid.to_blob()) - .bind(creation_utxo.to_blob()) - .execute(&mut *conn) - .await - .map_err(|e| anyhow::anyhow!("failed to flag tokentoken pool withdrawn: {e:?}"))?; - Ok(()) -} - #[allow(clippy::too_many_arguments)] pub async fn insert_history_entry( conn: &mut SqliteConnection, creation_utxo: &OutPointHash, - pool: &ParsedTokenToken, + pool: &ParsedDelegationPool, mtp_timestamp: Option, first_seen_timestamp: Option, reserve_a_delta: i64, @@ -195,35 +252,36 @@ pub async fn insert_history_entry( sqlx::query( "INSERT INTO tokentoken_pool_history_entry - (utxo, pool, storage_utxo, token_a_id, token_b_id, txid, tx_pos, storage_tx_pos, + (utxo, pool, spent_main_utxo, storage_utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, reserve_a, reserve_b, - main_sats, storage_sats, reserve_a_delta, reserve_b_delta) + owed_a, platform_fee_rate, main_sats, storage_sats, + reserve_a_delta, reserve_b_delta) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(utxo) DO UPDATE SET pool = excluded.pool, + spent_main_utxo = excluded.spent_main_utxo, storage_utxo = excluded.storage_utxo, txid = excluded.txid, tx_pos = excluded.tx_pos, - storage_tx_pos = excluded.storage_tx_pos, mtp_timestamp = COALESCE(excluded.mtp_timestamp, tokentoken_pool_history_entry.mtp_timestamp), first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, tokentoken_pool_history_entry.first_seen_timestamp), sequence = excluded.sequence", ) - .bind(pool.new_main_utxo_hash.expect("new main utxo hash missing").to_blob()) + .bind(pool.new_main_utxo_hash.to_blob()) .bind(creation_utxo.to_blob()) - .bind(pool.new_storage_utxo_hash.expect("new storage utxo hash missing").to_blob()) - .bind(pool.token_a_id.expect("token a id missing").to_blob()) - .bind(pool.token_b_id.expect("token b id missing").to_blob()) - .bind(pool.new_main_utxo_txid.expect("main txid missing").to_blob()) - .bind(pool.new_main_utxo_n.expect("main vout missing") as i64) - .bind(pool.new_storage_utxo_n.expect("storage vout missing") as i64) + .bind(pool.spent_main_utxo_hash.map(|h| h.to_blob())) + .bind(pool.new_storage_utxo_hash.to_blob()) + .bind(pool.new_main_utxo_txid.to_blob()) + .bind(pool.new_main_utxo_n as i64) .bind(mtp_timestamp.map(|t| t as i64)) .bind(first_seen_timestamp.map(|t| t as i64)) .bind(next_seq) - .bind(pool.token_a_amount.expect("reserve a missing")) - .bind(pool.token_b_amount.expect("reserve b missing")) - .bind(pool.main_sats.expect("main sats missing") as i64) - .bind(pool.storage_sats.expect("storage sats missing") as i64) + .bind(pool.token_a_amount) + .bind(pool.token_b_amount) + .bind(pool.config.owed_a as i64) + .bind(pool.config.platform_fee_rate as i64) + .bind(pool.main_sats as i64) + .bind(pool.storage_sats as i64) .bind(reserve_a_delta) .bind(reserve_b_delta) .execute(&mut *conn) @@ -232,52 +290,48 @@ pub async fn insert_history_entry( Ok(()) } -/// Apply a batch of parsed TokenToken states to the database. +/// Apply a batch of parsed TokenToken states plus the batch's teardown probes. /// -/// Mirrors [`crate::db::cauldron::pool::update_pool_history`]: a swap is linked -/// to its pool by the spent main UTXO; a creation (no known parent) opens a new -/// pool; a withdrawal flags the pool closed. The queue defers states whose -/// parent is created later in the same block. +/// A state is linked to its pool by the spent main UTXO; an unknown (or +/// absent) parent opens a new pool — the parser only attributes candidate +/// spends, so an unknown outpoint means creation. The queue defers states +/// whose parent is created later in the same batch. +/// +/// After the states are applied, each probe's spent outpoints are checked +/// against recorded pool states: a known main UTXO spent by a blob-presenting +/// tx with no successor state is an LP teardown and flags the pool withdrawn. pub async fn update_tokentoken_pool_history( conn: &mut SqliteConnection, - pools: Vec, + pools: Vec, + teardown_probes: &[TeardownProbe], mtp_timestamp: Option, first_seen_timestamp: Option, ) -> Result<()> { - if pools.is_empty() { - return Ok(()); - } - let mut queue = VecDeque::from(pools); while let Some(current) = queue.pop_front() { - let (creation_utxo, is_new) = - match get_pool_by_main_utxo(&mut *conn, ¤t.spent_main_utxo_hash).await? { + let (creation_utxo, is_new) = match current.spent_main_utxo_hash { + None => (current.new_main_utxo_hash, true), + Some(parent) => match get_pool_by_main_utxo(&mut *conn, &parent).await? { Some(c) => (c, false), None => { - let has_parent = queue - .iter() - .any(|p| Some(current.spent_main_utxo_hash) == p.new_main_utxo_hash); + let has_parent = queue.iter().any(|p| parent == p.new_main_utxo_hash); if has_parent { queue.push_back(current); continue; } - match current.new_main_utxo_hash { - Some(utxo) if !current.is_withdrawn => (utxo, true), - // A withdrawal of a pool we never indexed, or a malformed - // state — nothing to update. - _ => continue, - } + // Unknown candidate parent: the parser over-attributes + // (any adjacent-outpoint input pair), so this is a + // creation. + (current.new_main_utxo_hash, true) } - }; + }, + }; - if current.is_withdrawn { - info!("TokenToken pool {} withdrawn", creation_utxo); - flag_as_withdrawn(&mut *conn, &creation_utxo, ¤t).await?; - } else if is_new { + if is_new { info!( "TokenToken pool created in tx {}", - current.new_main_utxo_txid.expect("new pool txid") + current.new_main_utxo_txid ); insert_new_pool(&mut *conn, ¤t).await?; insert_history_entry( @@ -291,23 +345,50 @@ pub async fn update_tokentoken_pool_history( ) .await?; } else { - let (prev_a, prev_b) = - get_reserves_at(&mut *conn, ¤t.spent_main_utxo_hash).await?; - let reserve_a_delta = current.token_a_amount.unwrap_or(0) - prev_a; - let reserve_b_delta = current.token_b_amount.unwrap_or(0) - prev_b; + let (prev_a, prev_b) = get_reserves_at( + &mut *conn, + ¤t.spent_main_utxo_hash.expect("linked state"), + ) + .await?; insert_history_entry( &mut *conn, &creation_utxo, ¤t, mtp_timestamp, first_seen_timestamp, - reserve_a_delta, - reserve_b_delta, + current.token_a_amount - prev_a, + current.token_b_amount - prev_b, ) .await?; } } + // Teardown pass: a known pool main among a probe's spent outpoints, with + // no successor state recorded, closes the pool. + for probe in teardown_probes { + for spent in &probe.spent_utxo_hashes { + let r = sqlx::query( + "UPDATE tokentoken_pool SET withdrawn_in_txid = ?1 + WHERE withdrawn_in_txid IS NULL + AND creation_utxo IN ( + SELECT h.pool FROM tokentoken_pool_history_entry h + WHERE h.utxo = ?2 + AND NOT EXISTS ( + SELECT 1 FROM tokentoken_pool_history_entry s + WHERE s.spent_main_utxo = ?2 + ) + )", + ) + .bind(probe.txid.to_blob()) + .bind(spent.to_blob()) + .execute(&mut *conn) + .await?; + if r.rows_affected() > 0 { + info!("TokenToken pool torn down in tx {}", probe.txid); + } + } + } + Ok(()) } @@ -315,7 +396,7 @@ pub async fn update_tokentoken_pool_history( /// /// Order-independent w.r.t. how reorged blocks are replayed: deletes the /// block's history rows, drops pools left with no history (their creation was -/// undone), and reverts withdrawals made in the block. +/// undone), and reverts teardowns flagged in the block. pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) -> Result { let r1 = sqlx::query( "DELETE FROM tokentoken_pool_history_entry @@ -343,7 +424,7 @@ pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) Ok(r1.rows_affected() + r3.rows_affected() != 0) } -fn serialize_u64_as_string(value: &u64, serializer: S) -> Result +pub(crate) fn serialize_u64_as_string(value: &u64, serializer: S) -> Result where S: serde::Serializer, { @@ -357,13 +438,27 @@ pub struct ActiveTokenTokenPool { pub nft_owner: String, pub token_a_id: String, pub token_b_id: String, - /// Decimal string (base units) — avoids f64 precision loss for large token amounts. + /// RAW token A amount of the main UTXO (includes `owed_a`) as a decimal + /// string — the tradeable reserve is `reserve_a - owed_a`. #[serde(serialize_with = "serialize_u64_as_string")] pub reserve_a: u64, #[serde(serialize_with = "serialize_u64_as_string")] pub reserve_b: u64, - pub fee_rate: u32, + /// Accrued platform fee (token A), excluded from the tradeable reserve. + #[serde(serialize_with = "serialize_u64_as_string")] + pub owed_a: u64, + /// Platform's cut, parts-per-1_000_000 (mutable by a platform sweep). + pub platform_fee_rate: u32, + /// LP's cut, parts-per-100_000. + pub pool_fee_rate: u32, + /// Minimum combined (platform + pool) fee in token-A base units. pub min_fee: u64, + /// Virtual reserve offset, token A side (0 = plain constant product). + #[serde(serialize_with = "serialize_u64_as_string")] + pub virtual_x: u64, + /// Virtual reserve offset, token B side. + #[serde(serialize_with = "serialize_u64_as_string")] + pub virtual_y: u64, /// Satoshis locked in the main UTXO (preserved across swaps). pub main_sats: u64, /// Satoshis locked in the storage UTXO (preserved across swaps). @@ -372,10 +467,61 @@ pub struct ActiveTokenTokenPool { pub txid: String, /// Output index of the main UTXO (token A). pub main_vout: u32, - /// Output index of the storage UTXO (token B); also `otherTokenOutpointIndex`. + /// Output index of the storage UTXO (token B); always `main_vout + 1`. pub storage_vout: u32, } +const ACTIVE_POOL_SELECT: &str = + "SELECT p.creation_utxo, p.nft_owner, p.token_a_id, p.token_b_id, p.pool_fee_rate, + p.min_fee, p.virtual_x, p.virtual_y, + phe.reserve_a, phe.reserve_b, phe.owed_a, phe.platform_fee_rate, + phe.main_sats, phe.storage_sats, phe.txid, phe.tx_pos + FROM tokentoken_pool p + JOIN tokentoken_pool_history_entry phe ON phe.pool = p.creation_utxo + AND phe.sequence = ( + SELECT MAX(sequence) FROM tokentoken_pool_history_entry WHERE pool = p.creation_utxo + ) + WHERE p.withdrawn_in_txid IS NULL"; + +fn row_to_active_pool(row: &sqlx::sqlite::SqliteRow) -> Result { + let pool_blob: Vec = row.get(0); + let nft_owner_blob: Vec = row.get(1); + let token_a_blob: Vec = row.get(2); + let token_b_blob: Vec = row.get(3); + let pool_fee_rate: i64 = row.get(4); + let min_fee: i64 = row.get(5); + let virtual_x: i64 = row.get(6); + let virtual_y: i64 = row.get(7); + let reserve_a: i64 = row.get(8); + let reserve_b: i64 = row.get(9); + let owed_a: i64 = row.get(10); + let platform_fee_rate: i64 = row.get(11); + let main_sats: i64 = row.get(12); + let storage_sats: i64 = row.get(13); + let txid_blob: Vec = row.get(14); + let tx_pos: i64 = row.get(15); + + Ok(ActiveTokenTokenPool { + pool_id: blob_to_display_hex::(&pool_blob)?, + nft_owner: hex::encode(&nft_owner_blob), + token_a_id: blob_to_display_hex::(&token_a_blob)?, + token_b_id: blob_to_display_hex::(&token_b_blob)?, + reserve_a: reserve_a as u64, + reserve_b: reserve_b as u64, + owed_a: owed_a as u64, + platform_fee_rate: platform_fee_rate as u32, + pool_fee_rate: pool_fee_rate as u32, + min_fee: min_fee as u64, + virtual_x: virtual_x as u64, + virtual_y: virtual_y as u64, + main_sats: main_sats as u64, + storage_sats: storage_sats as u64, + txid: blob_to_display_hex::(&txid_blob)?, + main_vout: tx_pos as u32, + storage_vout: tx_pos as u32 + 1, + }) +} + /// Active pools containing both tokens, regardless of which is A (main) or B /// (storage). Returns each pool's latest state. pub async fn db_active_pools_for_pair( @@ -383,60 +529,18 @@ pub async fn db_active_pools_for_pair( token_a: &TokenID, token_b: &TokenID, ) -> Result> { - let a = token_a.to_blob(); - let b = token_b.to_blob(); - - let rows = sqlx::query( - "SELECT p.creation_utxo, p.nft_owner, p.token_a_id, p.token_b_id, p.fee_rate, p.min_fee, - phe.reserve_a, phe.reserve_b, phe.main_sats, phe.storage_sats, - phe.txid, phe.tx_pos, phe.storage_tx_pos - FROM tokentoken_pool p - JOIN tokentoken_pool_history_entry phe ON phe.pool = p.creation_utxo - AND phe.sequence = ( - SELECT MAX(sequence) FROM tokentoken_pool_history_entry WHERE pool = p.creation_utxo - ) - WHERE p.withdrawn_in_txid IS NULL + let sql = format!( + "{ACTIVE_POOL_SELECT} AND ((p.token_a_id = ?1 AND p.token_b_id = ?2) - OR (p.token_a_id = ?2 AND p.token_b_id = ?1))", - ) - .bind(&a) - .bind(&b) - .fetch_all(pool) - .await?; + OR (p.token_a_id = ?2 AND p.token_b_id = ?1))" + ); + let rows = sqlx::query(&sql) + .bind(token_a.to_blob()) + .bind(token_b.to_blob()) + .fetch_all(pool) + .await?; - let mut out = Vec::with_capacity(rows.len()); - for row in rows { - let pool_blob: Vec = row.get(0); - let nft_owner_blob: Vec = row.get(1); - let token_a_blob: Vec = row.get(2); - let token_b_blob: Vec = row.get(3); - let fee_rate: i64 = row.get(4); - let min_fee: i64 = row.get(5); - let reserve_a: i64 = row.get(6); - let reserve_b: i64 = row.get(7); - let main_sats: i64 = row.get(8); - let storage_sats: i64 = row.get(9); - let txid_blob: Vec = row.get(10); - let tx_pos: i64 = row.get(11); - let storage_tx_pos: i64 = row.get(12); - - out.push(ActiveTokenTokenPool { - pool_id: blob_to_display_hex::(&pool_blob)?, - nft_owner: hex::encode(&nft_owner_blob), - token_a_id: blob_to_display_hex::(&token_a_blob)?, - token_b_id: blob_to_display_hex::(&token_b_blob)?, - reserve_a: reserve_a as u64, - reserve_b: reserve_b as u64, - fee_rate: fee_rate as u32, - min_fee: min_fee as u64, - main_sats: main_sats as u64, - storage_sats: storage_sats as u64, - txid: blob_to_display_hex::(&txid_blob)?, - main_vout: tx_pos as u32, - storage_vout: storage_tx_pos as u32, - }); - } - Ok(out) + rows.iter().map(row_to_active_pool).collect() } /// Every active pool (`withdrawn_in_txid IS NULL`) at its latest state, with no @@ -444,53 +548,8 @@ pub async fn db_active_pools_for_pair( /// set of pools to show which token pairs have liquidity (without probing each /// candidate pair individually). pub async fn db_all_active_pools(pool: &SqlitePool) -> Result> { - let rows = sqlx::query( - "SELECT p.creation_utxo, p.nft_owner, p.token_a_id, p.token_b_id, p.fee_rate, p.min_fee, - phe.reserve_a, phe.reserve_b, phe.main_sats, phe.storage_sats, - phe.txid, phe.tx_pos, phe.storage_tx_pos - FROM tokentoken_pool p - JOIN tokentoken_pool_history_entry phe ON phe.pool = p.creation_utxo - AND phe.sequence = ( - SELECT MAX(sequence) FROM tokentoken_pool_history_entry WHERE pool = p.creation_utxo - ) - WHERE p.withdrawn_in_txid IS NULL", - ) - .fetch_all(pool) - .await?; - - let mut out = Vec::with_capacity(rows.len()); - for row in rows { - let pool_blob: Vec = row.get(0); - let nft_owner_blob: Vec = row.get(1); - let token_a_blob: Vec = row.get(2); - let token_b_blob: Vec = row.get(3); - let fee_rate: i64 = row.get(4); - let min_fee: i64 = row.get(5); - let reserve_a: i64 = row.get(6); - let reserve_b: i64 = row.get(7); - let main_sats: i64 = row.get(8); - let storage_sats: i64 = row.get(9); - let txid_blob: Vec = row.get(10); - let tx_pos: i64 = row.get(11); - let storage_tx_pos: i64 = row.get(12); - - out.push(ActiveTokenTokenPool { - pool_id: blob_to_display_hex::(&pool_blob)?, - nft_owner: hex::encode(&nft_owner_blob), - token_a_id: blob_to_display_hex::(&token_a_blob)?, - token_b_id: blob_to_display_hex::(&token_b_blob)?, - reserve_a: reserve_a as u64, - reserve_b: reserve_b as u64, - fee_rate: fee_rate as u32, - min_fee: min_fee as u64, - main_sats: main_sats as u64, - storage_sats: storage_sats as u64, - txid: blob_to_display_hex::(&txid_blob)?, - main_vout: tx_pos as u32, - storage_vout: storage_tx_pos as u32, - }); - } - Ok(out) + let rows = sqlx::query(ACTIVE_POOL_SELECT).fetch_all(pool).await?; + rows.iter().map(row_to_active_pool).collect() } /// Distinct token ids that appear in at least one active TokenToken pool. @@ -519,6 +578,7 @@ mod tests { use bitcoin_hashes::Hash; use bitcoincash::{BlockHash, Txid}; + use riftenlabs_defi::tokentoken_delegation::DelegationConfig; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use std::sync::atomic::{AtomicU64, Ordering as AOrdering}; @@ -544,6 +604,101 @@ mod tests { create_table(&pool).await; } + /// A database still holding the pre-delegation AMM tables (recognizable by + /// the old `fee_rate` column) must be reset to the new schema. + #[tokio::test] + async fn create_table_drops_pre_delegation_schema() { + let id = TT_TEST_COUNTER.fetch_add(1, AOrdering::SeqCst); + let uri = format!("file:tt_mig_{}?mode=memory&cache=shared", id); + let opts = SqliteConnectOptions::new() + .filename(&uri) + .foreign_keys(false); + let pool = SqlitePoolOptions::new().connect_with(opts).await.unwrap(); + + sqlx::query( + "CREATE TABLE tokentoken_pool ( + creation_utxo BLOB PRIMARY KEY, nft_owner BLOB NOT NULL, + token_a_id BLOB NOT NULL, token_b_id BLOB NOT NULL, + fee_rate INT NOT NULL, min_fee INT NOT NULL, withdrawn_in_txid BLOB + )", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query("CREATE TABLE tokentoken_pool_history_entry (utxo BLOB PRIMARY KEY)") + .execute(&pool) + .await + .unwrap(); + + create_table(&pool).await; + + // The new schema is in place. + let has: (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM pragma_table_info('tokentoken_pool') WHERE name = 'pool_fee_rate'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(has.0, 1, "outdated table must be replaced"); + } + + fn oph(b: u8) -> OutPointHash { + OutPointHash::from_byte_array([b; 32]) + } + fn tid(b: u8) -> TokenID { + TokenID::from_byte_array([b; 32]) + } + fn txid(b: u8) -> Txid { + Txid::from_byte_array([b; 32]) + } + + fn config(owed_a: u64) -> DelegationConfig { + DelegationConfig { + owed_a, + platform_fee_rate: 600, + pool_fee_rate: 300, + min_fee: 1, + virtual_x: 0, + virtual_y: 0, + nft_owner: [0x11; 32], + } + } + + /// A creation state: no attributed parent. + fn creation(main: u8, txid_b: u8, ra: i64, rb: i64) -> ParsedDelegationPool { + ParsedDelegationPool { + config: config(0), + spent_main_utxo_hash: None, + spent_storage_utxo_hash: None, + new_main_utxo_hash: oph(main), + new_main_utxo_txid: txid(txid_b), + new_main_utxo_n: 0, + new_storage_utxo_hash: oph(main.wrapping_add(100)), + token_a_id: tid(0xAA), + token_a_amount: ra, + token_b_id: tid(0xBB), + token_b_amount: rb, + main_sats: 800, + storage_sats: 800, + } + } + + /// A swap spending `prev_main`, producing `new_main` in tx `txid_b`. + fn swap(prev_main: u8, new_main: u8, txid_b: u8, ra: i64, rb: i64) -> ParsedDelegationPool { + let mut p = creation(new_main, txid_b, ra, rb); + p.spent_main_utxo_hash = Some(oph(prev_main)); + p.spent_storage_utxo_hash = Some(oph(prev_main.wrapping_add(100))); + p.config.owed_a = 1; + p + } + + async fn seed_tx(pool: &SqlitePool, txid_val: &Txid, blockhash: &BlockHash) { + let mut conn = pool.acquire().await.unwrap(); + insert_block_tx(&mut conn, txid_val, blockhash, 1_700_000_000) + .await + .unwrap(); + } + /// A state first seen in the mempool (first_seen only) and later confirmed /// in a block (mtp only) must keep the mempool first_seen timestamp. #[tokio::test] @@ -552,15 +707,15 @@ mod tests { let block = BlockHash::all_zeros(); let c = creation(1, 10, 1_000_000, 2_000_000); - seed_tx(&pool, &c, &block).await; + seed_tx(&pool, &c.new_main_utxo_txid, &block).await; let mut conn = pool.acquire().await.unwrap(); // Mempool: first_seen only. - update_tokentoken_pool_history(&mut conn, vec![c.clone()], None, Some(1_700_000_100)) + update_tokentoken_pool_history(&mut conn, vec![c.clone()], &[], None, Some(1_700_000_100)) .await .unwrap(); // Block confirmation of the same state: mtp only. - update_tokentoken_pool_history(&mut conn, vec![c], Some(1_700_000_500), None) + update_tokentoken_pool_history(&mut conn, vec![c], &[], Some(1_700_000_500), None) .await .unwrap(); @@ -579,71 +734,18 @@ mod tests { assert_eq!(effective, Some(1_700_000_100), "first_seen wins"); } - fn oph(b: u8) -> OutPointHash { - OutPointHash::from_byte_array([b; 32]) - } - fn tid(b: u8) -> TokenID { - TokenID::from_byte_array([b; 32]) - } - fn txid(b: u8) -> Txid { - Txid::from_byte_array([b; 32]) - } - - fn creation(main: u8, txid_b: u8, ra: i64, rb: i64) -> ParsedTokenToken { - ParsedTokenToken { - nft_owner: [0x11; 32], - fee_rate: 300, - min_fee: 1, - other_token_outpoint_index: 1, - is_withdrawn: false, - spent_main_utxo_hash: OutPointHash::all_zeros(), - spent_storage_utxo_hash: Some(OutPointHash::all_zeros()), - new_main_utxo_hash: Some(oph(main)), - new_main_utxo_txid: Some(txid(txid_b)), - new_main_utxo_n: Some(0), - new_storage_utxo_hash: Some(oph(main.wrapping_add(100))), - new_storage_utxo_txid: Some(txid(txid_b)), - new_storage_utxo_n: Some(1), - token_a_id: Some(tid(0xAA)), - token_b_id: Some(tid(0xBB)), - token_a_amount: Some(ra), - token_b_amount: Some(rb), - main_sats: Some(800), - storage_sats: Some(800), - } - } - - /// A swap spending `prev_main`, producing `new_main` in tx `txid_b`. - fn swap(prev_main: u8, new_main: u8, txid_b: u8, ra: i64, rb: i64) -> ParsedTokenToken { - let mut p = creation(new_main, txid_b, ra, rb); - p.spent_main_utxo_hash = oph(prev_main); - p - } - - async fn seed_tx(pool: &SqlitePool, t: &ParsedTokenToken, blockhash: &BlockHash) { - let mut conn = pool.acquire().await.unwrap(); - insert_block_tx( - &mut conn, - &t.new_main_utxo_txid.unwrap(), - blockhash, - 1_700_000_000, - ) - .await - .unwrap(); - } - #[tokio::test] async fn creation_then_swap_tracks_reserves_and_pair_query() { let pool = test_db().await; let block = BlockHash::all_zeros(); let c = creation(1, 10, 1_000_000, 2_000_000); - seed_tx(&pool, &c, &block).await; + seed_tx(&pool, &c.new_main_utxo_txid, &block).await; let s = swap(1, 2, 11, 1_010_000, 1_980_000); - seed_tx(&pool, &s, &block).await; + seed_tx(&pool, &s.new_main_utxo_txid, &block).await; let mut conn = pool.acquire().await.unwrap(); - update_tokentoken_pool_history(&mut conn, vec![c, s], Some(1_700_000_000), None) + update_tokentoken_pool_history(&mut conn, vec![c, s], &[], Some(1_700_000_000), None) .await .unwrap(); drop(conn); @@ -655,38 +757,82 @@ mod tests { assert_eq!(pools.len(), 1); assert_eq!(pools[0].reserve_a, 1_010_000); assert_eq!(pools[0].reserve_b, 1_980_000); - assert_eq!(pools[0].fee_rate, 300); + assert_eq!(pools[0].owed_a, 1, "owed from the latest state"); + assert_eq!(pools[0].platform_fee_rate, 600); + assert_eq!(pools[0].pool_fee_rate, 300); assert_eq!(pools[0].main_vout, 0); assert_eq!(pools[0].storage_vout, 1); + // Deltas recorded against the previous raw reserves. + let row = sqlx::query( + "SELECT reserve_a_delta, reserve_b_delta FROM tokentoken_pool_history_entry + WHERE utxo = ?", + ) + .bind(oph(2).to_blob()) + .fetch_one(&pool) + .await + .unwrap(); + let (da, db_): (i64, i64) = (row.get(0), row.get(1)); + assert_eq!((da, db_), (10_000, -20_000)); + let tokens = db_pair_tokens(&pool).await.unwrap(); assert!(tokens.contains(&tid(0xAA).to_string()) && tokens.contains(&tid(0xBB).to_string())); } + /// A same-batch parent created later in the vec must be resolved by the + /// deferral queue. #[tokio::test] - async fn withdraw_hides_pool_and_reorg_restores_it() { + async fn same_batch_out_of_order_states_link() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c = creation(1, 10, 1_000_000, 2_000_000); + let s = swap(1, 2, 11, 1_000_500, 1_999_050); + seed_tx(&pool, &c.new_main_utxo_txid, &block).await; + seed_tx(&pool, &s.new_main_utxo_txid, &block).await; + + let mut conn = pool.acquire().await.unwrap(); + // Swap first, creation second: the queue defers the swap. + update_tokentoken_pool_history(&mut conn, vec![s, c], &[], Some(1), None) + .await + .unwrap(); + drop(conn); + + let n: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM tokentoken_pool") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(n.0, 1, "one pool, not two"); + let pools = db_all_active_pools(&pool).await.unwrap(); + assert_eq!(pools[0].reserve_a, 1_000_500, "latest state wins"); + } + + /// A teardown is a blob-presenting tx spending a known main with no + /// successor: the probe flags the pool withdrawn; a reorg of that block + /// restores it; a reorg of the creation removes the pool entirely. + #[tokio::test] + async fn teardown_probe_hides_pool_and_reorg_restores_it() { let pool = test_db().await; let block_a = BlockHash::from_byte_array([0xA1; 32]); let block_b = BlockHash::from_byte_array([0xB2; 32]); let c = creation(1, 10, 1_000_000, 2_000_000); - seed_tx(&pool, &c, &block_a).await; - - // Withdrawal in a later block. - let mut w = creation(0, 20, 0, 0); - w.is_withdrawn = true; - w.spent_main_utxo_hash = oph(1); - w.new_main_utxo_hash = None; - w.new_storage_utxo_hash = None; - // carry the withdrawing txid (as the parser does) - w.new_main_utxo_txid = Some(txid(20)); - seed_tx(&pool, &w, &block_b).await; + seed_tx(&pool, &c.new_main_utxo_txid, &block_a).await; let mut conn = pool.acquire().await.unwrap(); - update_tokentoken_pool_history(&mut conn, vec![c], Some(1), None) + update_tokentoken_pool_history(&mut conn, vec![c], &[], Some(1), None) .await .unwrap(); - update_tokentoken_pool_history(&mut conn, vec![w], Some(2), None) + + // Teardown tx in a later block: no states, one probe spending main 1 + // (plus an unrelated outpoint that must not confuse the probe). + let teardown_txid = txid(20); + seed_tx(&pool, &teardown_txid, &block_b).await; + let probe = TeardownProbe { + txid: teardown_txid, + spent_utxo_hashes: vec![oph(0xEE), oph(1)], + }; + update_tokentoken_pool_history(&mut conn, vec![], &[probe], Some(2), None) .await .unwrap(); drop(conn); @@ -698,15 +844,15 @@ mod tests { .await .unwrap() .is_empty(), - "withdrawn pool should not be active" + "torn-down pool should not be active" ); - // Reorg the withdrawal block: pool becomes active again. + // Reorg the teardown block: pool becomes active again. delete_entries_for_block(&pool, &block_b).await.unwrap(); assert_eq!( db_active_pools_for_pair(&pool, &a, &b).await.unwrap().len(), 1, - "reorg of the withdrawal should restore the pool" + "reorg of the teardown should restore the pool" ); // Reorg the creation block: pool disappears entirely. @@ -717,4 +863,35 @@ mod tests { .is_empty()); assert!(db_pair_tokens(&pool).await.unwrap().is_empty()); } + + /// A probe outpoint consumed by a successor state in the SAME batch is a + /// swap, not a teardown. + #[tokio::test] + async fn swap_probe_is_not_a_teardown() { + let pool = test_db().await; + let block = BlockHash::all_zeros(); + + let c = creation(1, 10, 1_000_000, 2_000_000); + let s = swap(1, 2, 11, 1_000_500, 1_999_050); + seed_tx(&pool, &c.new_main_utxo_txid, &block).await; + seed_tx(&pool, &s.new_main_utxo_txid, &block).await; + + // The swap tx presents the blob, so it is probed too. + let probe = TeardownProbe { + txid: txid(11), + spent_utxo_hashes: vec![oph(0x22), oph(1), oph(101)], + }; + + let mut conn = pool.acquire().await.unwrap(); + update_tokentoken_pool_history(&mut conn, vec![c, s], &[probe], Some(1), None) + .await + .unwrap(); + drop(conn); + + assert_eq!( + db_all_active_pools(&pool).await.unwrap().len(), + 1, + "swapped pool must stay active" + ); + } } diff --git a/src/db/ido/mod.rs b/src/db/ido/mod.rs index 1ad4794..980b977 100644 --- a/src/db/ido/mod.rs +++ b/src/db/ido/mod.rs @@ -88,9 +88,7 @@ const MAINNET_IDO_PARAMS_NFT_CATEGORY: &[u8; 32] = &[0u8; 32]; // (OP_PUSH8 "CldIdo00" OP_DROP), found at the start of every ido // state machine redeem script. -pub const IDO_SIGNATURE: &[u8] = &[ - 0x08, 0x43, 0x6c, 0x64, 0x49, 0x64, 0x6f, 0x30, 0x30, 0x75, -]; +pub const IDO_SIGNATURE: &[u8] = &[0x08, 0x43, 0x6c, 0x64, 0x49, 0x64, 0x6f, 0x30, 0x30, 0x75]; static OFFERING_CONTRACT: LazyLock> = LazyLock::new(|| { hex::decode("c0ce01207f75c0519c637600ce8800cf517f755f84518867c0009d00d000d394765479a269765579950500e876481796005c7900a063577900a069587900a0695c79587995048033e1015a7995a169785d7995587995048033e1010400e1f5059596776e947b757c7800a0696854790087535e7900a06376608577687863760120857768005f7900a0635f795680547958807e77686e7e51d28851d15779517e8851d356799d5800cf557f77547f757e557958807e567958807e547958807e607956807e5f7960798277009c637859797eaa776776827701209d6802aa20012052797e60797eaa7e01877e51cd8854796351cc5779a26960798277009c6352d159798852d3009d52d252798853d100876453d101207f75597987916968c4549d6752d100876452d101207f75597987916968c4539d686752d15a798852d35779a269525152807e60797e52cd8860798277009c6353d159798853d3009d53d252798854d100876454d101207f7559798791696855d100876455d101207f75597987916968c4569d6753d100876453d101207f7559798791696854d100876454d101207f75597987916968c4559d686800cf517f77547f758100cc00c6527993a26900cd00c78800cf557f77547f758100cf557f75788b54807e00d28800ce00d1886d6d6d6d6d686d6d6d6d6d51").unwrap() @@ -1547,12 +1545,16 @@ fn parse_ido_preinit_tx_params( .discountAnnualRateNumerator < 0 { - invalid_ido_reasons.push(anyhow::anyhow!("preinit_parameters.offering.offer.discountAnnualRateNumerator < 0")); + invalid_ido_reasons.push(anyhow::anyhow!( + "preinit_parameters.offering.offer.discountAnnualRateNumerator < 0" + )); is_valid_ido = false; } if preinit_parameters.offering.offer.maxDiscountRateNumerator < 0 { - invalid_ido_reasons.push(anyhow::anyhow!("preinit_parameters.offering.offer.maxDiscountRateNumerator < 0")); + invalid_ido_reasons.push(anyhow::anyhow!( + "preinit_parameters.offering.offer.maxDiscountRateNumerator < 0" + )); is_valid_ido = false; } if preinit_parameters.offeredTokenAmount <= 0 { @@ -1588,14 +1590,16 @@ fn parse_ido_preinit_tx_params( )); is_valid_ido = false; } - if preinit_parameters.offering.platformFeeNumerator != ido_params.platformFee + if preinit_parameters.offering.platformFeeNumerator + != ido_params.platformFee { invalid_ido_reasons.push(anyhow::anyhow!( "offering.platformFeeNumerator != ido_params.platformFee" )); is_valid_ido = false; } - if preinit_parameters.offering.executionFee != ido_params.entryExecutionFee { + if preinit_parameters.offering.executionFee != ido_params.entryExecutionFee + { invalid_ido_reasons.push(anyhow::anyhow!( "offering.executionFee != ido_params.entryExecutionFee" )); @@ -1989,7 +1993,7 @@ fn parse_ido_preinit_tx( hex::encode(script.to_bytes()) )); } - let token_amount = Integer::from(token.amount); + let token_amount = Integer::from(token.amount.to_int()); if token_amount < requiredTokens { errors.push(anyhow::anyhow!("invalid offered token amount!")); } @@ -2321,9 +2325,7 @@ fn ido_add_tx(context: &IdoContext, tx: &Transaction) -> Result { permanentLiquidityShareNumerator: preinit_params .permanentLiquidityShareNumerator .clone(), - offeredTokenTotalSupply: preinit_params - .offeredTokenTotalSupply - .clone(), + offeredTokenTotalSupply: preinit_params.offeredTokenTotalSupply.clone(), offeredTokenAmount: preinit_params.offeredTokenAmount.clone(), offering: preinit_params.offering.clone(), orbPoolParamsCategory: preinit_params.orbPoolParamsCategory.clone(), @@ -2529,8 +2531,9 @@ fn ido_add_tx(context: &IdoContext, tx: &Transaction) -> Result { if xtoken_category != xTokenCategory { return Err(anyhow::anyhow!("output#2 token category != xTokenCategory")); } - let supply_amount = xtoken.amount as u64; - let demand_amount = second_output.token.as_ref().unwrap().amount as u64; + let supply_amount = xtoken.amount.to_int() as u64; + let demand_amount = + second_output.token.as_ref().unwrap().amount.to_int() as u64; let lockup_timeval = if has_lockup { u64::try_from(&decode_padded_vm_number(&entry_commitment[1..7])) .unwrap_or(0) @@ -2817,8 +2820,8 @@ fn ido_add_tx(context: &IdoContext, tx: &Transaction) -> Result { anyhow::anyhow!("permanent pool oToken output (output#1) is missing!") })?; Some(IdoPermanentPoolV0 { - xTokenAmount: Integer::from(x_leg.amount), - oTokenAmount: Integer::from(o_leg.amount), + xTokenAmount: Integer::from(x_leg.amount.to_int()), + oTokenAmount: Integer::from(o_leg.amount.to_int()), }) } None => None, @@ -2842,12 +2845,12 @@ fn ido_add_tx(context: &IdoContext, tx: &Transaction) -> Result { .output .get(3) .and_then(|output| output.token.as_ref()) - .map(|token| Integer::from(token.amount)) + .map(|token| Integer::from(token.amount.to_int())) .unwrap_or(Integer::ZERO); let platform_output = tx.output.get(5); let platform_xtoken_amount = platform_output .and_then(|output| output.token.as_ref()) - .map(|token| Integer::from(token.amount)) + .map(|token| Integer::from(token.amount.to_int())) .unwrap_or(Integer::ZERO); let platform_bch_payout = platform_output .map(|output| Integer::from(output.value.to_sat())) @@ -2862,7 +2865,7 @@ fn ido_add_tx(context: &IdoContext, tx: &Transaction) -> Result { refundAmount: collector_otoken_output .token .as_ref() - .map(|token| Integer::from(token.amount)) + .map(|token| Integer::from(token.amount.to_int())) .unwrap_or(Integer::ZERO), discountAmount: prev_state.discountAmount.clone(), collectorEarnedAmount: collector_xtoken_amount, @@ -3392,6 +3395,8 @@ mod tests { .copy_from_slice(&encode_padded_vm_number(&Integer::from(2000i64), 4).unwrap()); commitment[117..121] .copy_from_slice(&encode_padded_vm_number(&Integer::from(1000i64), 4).unwrap()); + commitment[121..125] + .copy_from_slice(&encode_padded_vm_number(&Integer::from(2000i64), 4).unwrap()); let parsed = parse_ido_params_nft_commitment(&commitment).expect("should parse"); assert_eq!(parsed.version, IDO_PARAMS_VERSION); assert_eq!(parsed.orbPoolParamsCategory, vec![0xAA; 32]); @@ -3400,6 +3405,8 @@ mod tests { assert_eq!(parsed.platformFee, Integer::from(5000i64)); assert_eq!(parsed.minExpireDuration, Integer::from(3600i64)); assert_eq!(parsed.maxExpireDuration, Integer::from(2_592_000i64)); + assert_eq!(parsed.minPlpAfterDiscount, Integer::from(1000i64)); + assert_eq!(parsed.minPlpShare, Integer::from(2000i64)); assert_eq!(parsed.entryExecutionFee, Integer::from(1000i64)); assert_eq!(parsed.createExecutionFee, Integer::from(2000i64)); // wrong size is rejected diff --git a/src/db/init.rs b/src/db/init.rs index 440b955..b9ec664 100644 --- a/src/db/init.rs +++ b/src/db/init.rs @@ -14,6 +14,7 @@ use super::DB; use crate::db::bcmr::prepare_tables as bcmr_prepare_tables; use crate::db::cauldron::config::check_db_version; use crate::db::cauldron::prepare_tables as cauldron_prepare_tables; +use crate::db::cauldron::tokenbch::create_table as tokenbch_prepare_tables; use crate::db::cauldron::tokentoken::create_table as tokentoken_prepare_tables; use crate::db::crc20::prepare_tables as crc20_prepare_tables; use crate::db::ido::prepare_tables as ido_prepare_tables; @@ -106,8 +107,9 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul } else { check_db_version(&cauldron_db_read).await?; } - // Always-run migration: safe on both new and existing cauldron.db + // Always-run migrations: safe on both new and existing cauldron.db tokentoken_prepare_tables(&cauldron_db_write).await; + tokenbch_prepare_tables(&cauldron_db_write).await; // Initialize BCMR database let (db_exists, bcmr_db_write, bcmr_db_read) = diff --git a/src/db/mod.rs b/src/db/mod.rs index 352b173..80358f0 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -13,6 +13,7 @@ pub mod ido; pub mod init; pub mod moria; pub mod oracle; +pub mod orbconstants; pub mod search; #[derive(Clone)] diff --git a/src/db/orbconstants.rs b/src/db/orbconstants.rs new file mode 100644 index 0000000..f2b7b9b --- /dev/null +++ b/src/db/orbconstants.rs @@ -0,0 +1,88 @@ +// Copyright (C) 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 + +//! Per-network ORB v0 deployment constants. +//! +//! Mirrors libriften's `packages/cauldron/src/orb/v0/{chipnet,mainnet}.ts` +//! deployment records. Values here are MONEY-SAFETY-critical: while a +//! network's value is the all-zero placeholder, the dependent indexer is +//! disabled outright (fail-closed) — nothing is indexed with assumed +//! parameters. + +use bitcoincash::Network; +use riftenlabs_defi::{tokenbch_delegation, tokentoken_delegation}; + +/// The delegation-pool platform NFTH (the platform-fee settlement destination +/// baked into every tokentoken/tokenbch pool's withdraw blob) on chipnet. +/// One value serves BOTH delegation contracts. +/// +/// From libriften `orb/v0/chipnet.ts` (`ORB_V0_CHIPNET.poolPlatformNfth`). +const CHIPNET_POOL_PLATFORM_NFTH: &[u8; 32] = &[ + 0x07, 0x2d, 0x5f, 0xbe, 0xcf, 0xa1, 0xbc, 0x37, 0xa1, 0x3f, 0x58, 0xb1, 0x88, 0xf5, 0x0a, 0x5b, + 0xc1, 0xdf, 0xb9, 0xab, 0x8a, 0xe0, 0xbf, 0x5d, 0x3d, 0x9a, 0x63, 0xf6, 0xf8, 0x1e, 0xfc, 0x0b, +]; + +// TODO:: set the pool platform nfth for mainnet once ORB v0 deploys there +// (libriften orb/v0/mainnet.ts is still all-zero too). +const MAINNET_POOL_PLATFORM_NFTH: &[u8; 32] = &[0u8; 32]; + +/// The delegation-pool platform NFTH for a network (all-zero = unconfigured). +pub fn pool_platform_nfth(network: Option) -> [u8; 32] { + match network { + Some(Network::Chipnet) => *CHIPNET_POOL_PLATFORM_NFTH, + _ => *MAINNET_POOL_PLATFORM_NFTH, + } +} + +/// True once a real (non-placeholder) value is configured. The build-time +/// `0xab…ab` contract placeholder counts as unconfigured too: pools built on +/// it must never be surfaced (libriften DESIGN.md "never fund a pool on the +/// placeholder"). +pub fn is_configured(value: &[u8; 32]) -> bool { + *value != [0u8; 32] && value != tokentoken_delegation::PLATFORM_NFTH_PLACEHOLDER +} + +/// The tokentoken delegation artifacts for a network's platform NFTH, or +/// `None` while the network is unconfigured (fail-closed: no delegation pool +/// is indexed until a real NFTH is set). +pub fn tokentoken_artifacts( + network: Option, +) -> Option { + let nfth = pool_platform_nfth(network); + is_configured(&nfth).then(|| tokentoken_delegation::DelegationArtifacts::new(nfth)) +} + +/// The tokenbch delegation artifacts for a network's platform NFTH, or `None` +/// while the network is unconfigured (fail-closed). +pub fn tokenbch_artifacts( + network: Option, +) -> Option { + let nfth = pool_platform_nfth(network); + is_configured(&nfth).then(|| tokenbch_delegation::DelegationArtifacts::new(nfth)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn chipnet_is_configured_mainnet_is_not() { + assert!(is_configured(&pool_platform_nfth(Some(Network::Chipnet)))); + assert!(!is_configured(&pool_platform_nfth(Some(Network::Bitcoin)))); + assert!(!is_configured(&pool_platform_nfth(None))); + // The contract build placeholder must never count as configured. + assert!(!is_configured( + tokentoken_delegation::PLATFORM_NFTH_PLACEHOLDER + )); + } + + #[test] + fn artifacts_follow_configuration() { + assert!(tokentoken_artifacts(Some(Network::Chipnet)).is_some()); + assert!(tokenbch_artifacts(Some(Network::Chipnet)).is_some()); + assert!(tokentoken_artifacts(Some(Network::Bitcoin)).is_none()); + assert!(tokenbch_artifacts(None).is_none()); + } +} diff --git a/src/electrum.rs b/src/electrum.rs index 0ed3853..b670572 100644 --- a/src/electrum.rs +++ b/src/electrum.rs @@ -7,19 +7,19 @@ use std::{collections::HashSet, str::FromStr}; use anyhow::{Context, Result}; use bitcoincash::{ - blockdata::block::Header as BlockHeader, consensus::deserialize, Transaction, Txid, + blockdata::block::Header as BlockHeader, consensus::deserialize, Network, Transaction, Txid, }; use electrum_client_netagnostic::{Client, ElectrumApi, Param}; use log::info; use riftenlabs_defi::{ cauldron::V2_CONTRACT_TEMPLATE, delphi::{v2::DELPHI_V2_REDEEM_SCRIPT_BODY, DELPHI_REDEEM_SCRIPT}, - tokentoken::{CONJURE_HINT_PREFIX, TOKENTOKEN_CONTRACT_CODE}, }; use serde_json::{json, Value}; use crate::bcmr::BCMR_PREFIX; use crate::db::ido::{IDO_PREINIT_ANNOUNCEMENT_SIGNATURE, IDO_SIGNATURE}; +use crate::db::orbconstants; /// Fetch blockchain tip from electrum server pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> { @@ -41,28 +41,39 @@ pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> { Ok((deserialize(&hex::decode(header)?)?, height as u64)) } -/// Fetch defi (cauldron + tokentoken), oracle, ido and bcmr mempool transactions -pub fn electrum_fetch_mempool( - client: &Client, -) -> Result<(HashSet, HashSet, HashSet, HashSet)> { +/// The (defi, oracle, ido, bcmr) mempool txid sets. +pub type MempoolTxSets = (HashSet, HashSet, HashSet, HashSet); + +/// Fetch defi (cauldron + delegation pools), oracle, ido and bcmr mempool +/// transactions +pub fn electrum_fetch_mempool(client: &Client, network: Option) -> Result { let cauldron_filter = json!({ "scriptsig": hex::encode(&V2_CONTRACT_TEMPLATE[(V2_CONTRACT_TEMPLATE.len() - 43)..]), // cauldron spends "scriptpubkey": hex::encode([0x6a /* op_return */, 0x06 /* push */, b'S', b'U', b'M', b'M', b'O', b'N']), // new pools (potentially) "operation": "union" }); - // The contract code is the constant tail of every tokentoken redeem - // script, pushed in full in spending scriptsigs; creations carry an - // `OP_RETURN "CONJURE"` hint output. - let conjure_hint_prefix: Vec = [0x6a /* op_return */, 0x07 /* push */] - .iter() - .chain(CONJURE_HINT_PREFIX) - .copied() - .collect(); - let tokentoken_filter = json!({ - "scriptsig": hex::encode(TOKENTOKEN_CONTRACT_CODE), // pool spends - "scriptpubkey": hex::encode(&conjure_hint_prefix), // new pools (potentially) - "operation": "union" + // Delegation pools (tokentoken + tokenbch). Every pool spend — swap, + // platform sweep or LP teardown — presents the combined logic blob in a + // carrier input's scriptsig, so the blob is the spend filter. Creations + // carry the (per-network-NFTH) constant locking: the whole thin-main + // script for tokentoken, the dispatcher tail of the pool locking for + // tokenbch (the filters match substrings). None while the network's + // platform NFTH is unconfigured — nothing delegation-related is fetched + // (fail-closed, matching the indexer). + let tokentoken_filter = orbconstants::tokentoken_artifacts(network).map(|a| { + json!({ + "scriptsig": hex::encode(a.combined_blob()), // pool spends (incl. teardowns) + "scriptpubkey": hex::encode(a.thin_main_locking()), // new pools + swap outputs + "operation": "union" + }) + }); + let tokenbch_filter = orbconstants::tokenbch_artifacts(network).map(|a| { + json!({ + "scriptsig": hex::encode(a.combined_blob()), // pool spends (incl. teardowns) + "scriptpubkey": hex::encode(a.thin_main_locking()), // pool lockings (constant tail) + "operation": "union" + }) }); // v1 oracle filter: matches the cashc 0.10.5 redeem-script template. @@ -116,10 +127,16 @@ pub fn electrum_fetch_mempool( .collect()) }; - // Cauldron and tokentoken txs share one set: update_mempool diffs it - // against the stored mempool (tx table) to decide adds and deletes. + // Cauldron, tokentoken and tokenbch txs share one set: update_mempool + // diffs it against the stored mempool (tx table) to decide adds and + // deletes. let mut defi_txs = fetch_txs(cauldron_filter)?; - defi_txs.extend(fetch_txs(tokentoken_filter)?); + if let Some(filter) = tokentoken_filter { + defi_txs.extend(fetch_txs(filter)?); + } + if let Some(filter) = tokenbch_filter { + defi_txs.extend(fetch_txs(filter)?); + } let mut oracle_txs = fetch_txs(oracle_v1_filter)?; oracle_txs.extend(fetch_txs(oracle_v2_filter)?); let ido_txs = fetch_txs(ido_filter)?; diff --git a/src/index.rs b/src/index.rs index 1fa66e0..0813c50 100644 --- a/src/index.rs +++ b/src/index.rs @@ -17,8 +17,8 @@ use bitcoincash::{ use electrum_client_netagnostic::{Client, ElectrumApi, Param}; use log::{debug, info, warn}; use riftenlabs_defi::cauldron::{parse_cauldrons_from_tx, ParsedContract}; +use riftenlabs_defi::chainutil::{compute_outpoint_hash, OutPointHash}; use riftenlabs_defi::moria::MoriaTokenIds; -use riftenlabs_defi::tokentoken::parse_tokentoken_from_tx; use crate::{ bcmr::index_bcmr, @@ -30,13 +30,14 @@ use crate::{ cauldron::{ config::{config_get, config_set}, header::{db_get_header, store_headers}, + tokentoken::TeardownProbe, tx::insert_block_tx, user::insert_user_action, utxo_funding::insert_utxo_funding, utxo_spending::insert_utxo_spending, }, oracle::index_oracle, - DB, + orbconstants, DB, }, electrum::{electrum_fetch_mempool, electrum_get_tip, electrum_get_tx}, signal::shutdown_requested, @@ -78,7 +79,7 @@ pub async fn update_mempool( let electrum_clone = electrum.clone(); let (cauldron_txs, oracle_txs, ido_txs, bcmr_txs) = tokio::task::spawn_blocking(move || { - electrum_fetch_mempool(&electrum_clone.lock().unwrap()) + electrum_fetch_mempool(&electrum_clone.lock().unwrap(), network) }) .await??; @@ -112,23 +113,52 @@ pub async fn update_mempool( let mut all_cauldrons = vec![]; let mut all_tokentokens = vec![]; + let mut all_tokenbchs = vec![]; + let mut tt_probes: Vec = vec![]; + let mut tb_probes: Vec = vec![]; let current_timestamp = time_now() as u64; + // Delegation-pool artifacts for this network's platform NFTH; None + // while unconfigured (fail-closed: nothing delegation-related indexed). + let tt_arts = orbconstants::tokentoken_artifacts(network); + let tb_arts = orbconstants::tokenbch_artifacts(network); + for btx in txs_to_add { let cauldrons: Vec = parse_cauldrons_from_tx(&btx); - let tokentokens = parse_tokentoken_from_tx(&btx); + let tokentokens = tt_arts + .as_ref() + .map(|a| a.parse_delegation_pools_from_tx(&btx)) + .unwrap_or_default(); + let tokenbchs = tb_arts + .as_ref() + .map(|a| a.parse_delegation_pools_from_tx(&btx)) + .unwrap_or_default(); + // LP teardowns produce no pool output: every blob-presenting tx is + // probed against the recorded pool set instead. + let tt_blob = tt_arts + .as_ref() + .is_some_and(|a| a.presents_delegation_blob(&btx)); + let tb_blob = tb_arts + .as_ref() + .is_some_and(|a| a.presents_delegation_blob(&btx)); - if cauldrons.is_empty() && tokentokens.is_empty() { + if cauldrons.is_empty() + && tokentokens.is_empty() + && tokenbchs.is_empty() + && !tt_blob + && !tb_blob + { info!("ignoring non-defi tx {}", btx.compute_txid()); continue; } let txid = btx.compute_txid(); db::cauldron::tx::insert_mempool_tx(&mut db_tx, &txid, current_timestamp).await?; info!( - "mempool add {} with {} cauldrons, {} tokentoken states", + "mempool add {} with {} cauldrons, {} tokentoken states, {} tokenbch states", txid, cauldrons.len(), - tokentokens.len() + tokentokens.len(), + tokenbchs.len() ); if !cauldrons.is_empty() { @@ -138,7 +168,27 @@ pub async fn update_mempool( all_cauldrons.extend(cauldrons); } + if tt_blob || tb_blob { + let spent: Vec = btx + .input + .iter() + .map(|i| compute_outpoint_hash(&i.previous_output.txid, i.previous_output.vout)) + .collect(); + if tt_blob { + tt_probes.push(TeardownProbe { + txid, + spent_utxo_hashes: spent.clone(), + }); + } + if tb_blob { + tb_probes.push(TeardownProbe { + txid, + spent_utxo_hashes: spent, + }); + } + } all_tokentokens.extend(tokentokens); + all_tokenbchs.extend(tokenbchs); } db::cauldron::pool::update_pool_history( @@ -151,6 +201,15 @@ pub async fn update_mempool( db::cauldron::tokentoken::update_tokentoken_pool_history( &mut db_tx, all_tokentokens, + &tt_probes, + None, + Some(current_timestamp), + ) + .await?; + db::cauldron::tokenbch::update_tokenbch_pool_history( + &mut db_tx, + all_tokenbchs, + &tb_probes, None, Some(current_timestamp), ) @@ -464,17 +523,41 @@ pub async fn index_blocks( let mut all_cauldrons = vec![]; let mut all_tokentokens = vec![]; + let mut all_tokenbchs = vec![]; + let mut tt_probes: Vec = vec![]; + let mut tb_probes: Vec = vec![]; + + // Delegation-pool artifacts for this network's platform NFTH; None + // while unconfigured (fail-closed: nothing delegation-related indexed). + let tt_arts = orbconstants::tokentoken_artifacts(network); + let tb_arts = orbconstants::tokenbch_artifacts(network); let sorted_txs = ttor_sorted_kahn(block.txdata); for tx in &sorted_txs { let cauldrons = parse_cauldrons_from_tx(tx); - // Note: parse_tokentoken_from_tx skips the pool-creation check when - // any input spends an existing pool, so a single tx that both swaps - // pool P and creates pool Q drops Q (upstream riftenlabs-defi - // limitation, shared with parse_cauldrons_from_tx). - let tokentokens = parse_tokentoken_from_tx(tx); - if cauldrons.is_empty() && tokentokens.is_empty() { + let tokentokens = tt_arts + .as_ref() + .map(|a| a.parse_delegation_pools_from_tx(tx)) + .unwrap_or_default(); + let tokenbchs = tb_arts + .as_ref() + .map(|a| a.parse_delegation_pools_from_tx(tx)) + .unwrap_or_default(); + // LP teardowns produce no pool output: every blob-presenting tx is + // probed against the recorded pool set instead. + let tt_blob = tt_arts + .as_ref() + .is_some_and(|a| a.presents_delegation_blob(tx)); + let tb_blob = tb_arts + .as_ref() + .is_some_and(|a| a.presents_delegation_blob(tx)); + if cauldrons.is_empty() + && tokentokens.is_empty() + && tokenbchs.is_empty() + && !tt_blob + && !tb_blob + { continue; } @@ -496,8 +579,28 @@ pub async fn index_blocks( total_cauldrons += cauldrons.len(); all_cauldrons.extend(cauldrons); } + if tt_blob || tb_blob { + let spent: Vec = tx + .input + .iter() + .map(|i| compute_outpoint_hash(&i.previous_output.txid, i.previous_output.vout)) + .collect(); + if tt_blob { + tt_probes.push(TeardownProbe { + txid, + spent_utxo_hashes: spent.clone(), + }); + } + if tb_blob { + tb_probes.push(TeardownProbe { + txid, + spent_utxo_hashes: spent, + }); + } + } all_tokentokens.extend(tokentokens); + all_tokenbchs.extend(tokenbchs); } // Figuring out initial utxo needs to be done on all cauldrons in a block. @@ -505,6 +608,15 @@ pub async fn index_blocks( db::cauldron::tokentoken::update_tokentoken_pool_history( &mut db_tx, all_tokentokens, + &tt_probes, + Some(mtp), + None, + ) + .await?; + db::cauldron::tokenbch::update_tokenbch_pool_history( + &mut db_tx, + all_tokenbchs, + &tb_probes, Some(mtp), None, ) diff --git a/src/main.rs b/src/main.rs index 47385f4..c932466 100644 --- a/src/main.rs +++ b/src/main.rs @@ -213,6 +213,7 @@ async fn start_program( db::cauldron::pool::initialize_seq(&db.cauldron_r).await; db::cauldron::tokentoken::initialize_seq(&db.cauldron_r).await; + db::cauldron::tokenbch::initialize_seq(&db.cauldron_r).await; info!("Loading block headers..."); let all_headers = load_all_headers(&db.cauldron_r).await.unwrap(); @@ -679,6 +680,14 @@ async fn launch() -> _ { rpc::tokentoken::list_tokens, ], ) + .mount( + "/tokenbch", + routes![ + rpc::tokenbch::list_active_pools, + rpc::tokenbch::list_all_pools, + rpc::tokenbch::list_tokens, + ], + ) .mount( "/ido", routes![ diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index 61a7ef7..29be036 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -23,6 +23,7 @@ pub mod oracle; pub mod pool; pub mod price; pub mod response; +pub mod tokenbch; pub mod tokens; pub mod tokentoken; pub mod tvl; diff --git a/src/rpc/tokenbch.rs b/src/rpc/tokenbch.rs new file mode 100644 index 0000000..18c0c35 --- /dev/null +++ b/src/rpc/tokenbch.rs @@ -0,0 +1,63 @@ +// Copyright (C) 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 + +//! RPC endpoints for token <-> native-BCH (`TokenBch`) delegation pools. + +use crate::{ + db::{ + cauldron::tokenbch::{db_active_pools_for_token, db_all_active_pools, db_pool_tokens}, + DB, + }, + rpc::err::{bad_request, db_error, ApiErrorCode, CachedApiResult}, + rpc::response::{cached_ok, CACHE_AGGREGATE, CACHE_NONE}, +}; +use bitcoincash::TokenID; +use rocket::{get, State}; +use serde_json::{json, Value}; + +/// List active TokenBch pools that trade the given token against native BCH. +/// +/// `token` is required (a token id in display hex). Each pool's response +/// carries the RAW reserves (`sats`, `token_amount`), the accrued `owed` and +/// its fee side (`fee_paid_in_token`) so the caller can derive the tradeable +/// reserves, plus the virtual offsets for quoting. +#[get("/pool/active?")] +pub async fn list_active_pools(token: Option<&str>, conn: &State) -> CachedApiResult { + let Some(token) = token else { + return Err(bad_request( + ApiErrorCode::MissingParameters, + "Provide token", + )); + }; + let token = token + .parse::() + .map_err(|e| bad_request(ApiErrorCode::InvalidTokenId, &format!("Invalid token: {e}")))?; + + let active = db_active_pools_for_token(&conn.cauldron_r, &token) + .await + .map_err(db_error)?; + + Ok(cached_ok(json!({ "active": active }), CACHE_NONE)) +} + +/// List the token ids that appear in at least one active TokenBch pool. +/// Drives the frontend token selector and the "no route" decision. +#[get("/tokens")] +pub async fn list_tokens(conn: &State) -> CachedApiResult { + let tokens = db_pool_tokens(&conn.cauldron_r).await.map_err(db_error)?; + Ok(cached_ok(json!({ "tokens": tokens }), CACHE_AGGREGATE)) +} + +/// List ALL active TokenBch pools, regardless of token — every pool at its +/// latest state. Drives the frontend pooled-pair selector, so it can present +/// the tokens that have BCH liquidity without probing each candidate +/// individually. +#[get("/pools")] +pub async fn list_all_pools(conn: &State) -> CachedApiResult { + let active = db_all_active_pools(&conn.cauldron_r) + .await + .map_err(db_error)?; + Ok(cached_ok(json!({ "active": active }), CACHE_NONE)) +}