BitcoinCash/riftenlabs-indexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tokentoken/tokenbch: index delegation pools, replace the pre-delegation AMM

Browse source 
Replace the old P2SH32/CONJURE TokenToken AMM indexer (never in
production; its tables are dropped by an always-run migration) with the
delegation-model indexers from riftenlabs-defi 0.4.5:

- tokentoken: two co-created bare-P2S UTXOs (thin main + storage
  sibling). tokenbch (new): single-UTXO token<->BCH pool with the
  runtime feePaidInToken fee side. Both admitted purely by the constant
  locking derived from the per-network delegation-pool platform NFTH.
- src/db/orbconstants.rs: ONE platform NFTH serves both contracts —
  chipnet pins the deployed ORB v0 value (libriften orb/v0/chipnet.ts),
  mainnet is the all-zero placeholder: while unconfigured NOTHING
  delegation-related is parsed, fetched or indexed (fail-closed).
- Schema: immutable config (nftOwner, tokens, poolFeeRate, minFee,
  virtualX/Y, tokenbch feePaidInToken) on the pool row; per-state
  history carries raw reserves, owed, platformFeeRate (both mutable by
  platform sweeps) and deltas. History rows now FK tx(txid) ON DELETE
  CASCADE, so evicted mempool txs clean up after themselves (the old
  tokentoken indexer leaked those).
- LP teardowns carry no pool output: ingestion probes every
  blob-presenting tx's spent outpoints against the recorded pool set
  and flags a known pool UTXO spent without successor as withdrawn
  (withdrawn_in_txid FK ON DELETE SET NULL reverts an evicted teardown;
  reorg undo reverts a confirmed one).
- electrum mempool filters: the combined logic blob (scriptsig; every
  spend incl. teardowns) + the constant thin-main locking (scriptpubkey;
  creations), per contract, skipped while unconfigured.
- RPC: /tokentoken responses gain owed_a, platform/pool fee rates,
  minFee and virtual offsets; new /tokenbch (pool/active?token, pools,
  tokens) mirrors it for BCH pairs.
- deps: bitcoincash 0.32.4 (token.amount -> .to_int() in the ido
  indexer), riftenlabs-defi 0.4.5. NOTE: 0.4.5 (rust-riftenlabs-defi
  7c65f66) is not on crates.io yet — until it is published, build with
  a local [patch.crates-io] pointing at the sibling checkout (the
  committed Cargo.lock carries the path-patched entry). Also fixes a
  latent test bug (ido_params_nft_commitment_roundtrip never wrote
  createExecutionFee after the 127-byte layout shift).

Verified: cargo build, test (232 passing), clippy (no new warnings)
and fmt all clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Hossein Zoda 2026-08-04 15:30:55 +00:00
parent f39ce11bf2
commit dcebaf2648
15 changed files with 1571 additions and 338 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
Cargo.lock generated
View file

@ -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",
]

4
Cargo.toml
View file

@ -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"

8
src/chain.rs
View file

@ -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?
{

2
src/db/cauldron/mod.rs
View file

@ -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);",

736
src/db/cauldron/tokenbch.rs Normal file
View file

@ -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 <config[63]> OP_DROP <dispatcher>`). 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<Option<OutPointHash>> {
let row: Option<(Vec<u8>,)> =
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<u64>,
first_seen_timestamp: Option<u64>,
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<ParsedTokenBchPool>,
teardown_probes: &[TeardownProbe],
mtp_timestamp: Option<u64>,
first_seen_timestamp: Option<u64>,
) -> 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, &current).await?;
insert_history_entry(
&mut *conn,
&current.new_utxo_hash,
&current,
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,
&current,
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<bool> {
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<ActiveTokenBchPool> {
let pool_blob: Vec<u8> = row.get(0);
let nft_owner_blob: Vec<u8> = row.get(1);
let token_blob: Vec<u8> = 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<u8> = row.get(12);
let tx_pos: i64 = row.get(13);
Ok(ActiveTokenBchPool {
pool_id: blob_to_display_hex::<OutPointHash>(&pool_blob)?,
nft_owner: hex::encode(&nft_owner_blob),
token_id: blob_to_display_hex::<TokenID>(&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>(&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<Vec<ActiveTokenBchPool>> {
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<Vec<ActiveTokenBchPool>> {
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<Vec<String>> {
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<u8> = row.get(0);
out.push(blob_to_display_hex::<TokenID>(&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<i64> = row.get(0);
let first_seen: Option<i64> = row.get(1);
let effective: Option<i64> = 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"
);
}
}

733
src/db/cauldron/tokentoken.rs
View file

@ -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<OutPointHash>,
}
/// 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<u64>,
first_seen_timestamp: Option<u64>,
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<ParsedTokenToken>,
pools: Vec<ParsedDelegationPool>,
teardown_probes: &[TeardownProbe],
mtp_timestamp: Option<u64>,
first_seen_timestamp: Option<u64>,
) -> 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, &current.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, &current).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, &current).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, &current.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,
&current.spent_main_utxo_hash.expect("linked state"),
)
.await?;
insert_history_entry(
&mut *conn,
&creation_utxo,
&current,
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<bool> {
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<S>(value: &u64, serializer: S) -> Result<S::Ok, S::Error>
pub(crate) fn serialize_u64_as_string<S>(value: &u64, serializer: S) -> Result<S::Ok, S::Error>
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<ActiveTokenTokenPool> {
let pool_blob: Vec<u8> = row.get(0);
let nft_owner_blob: Vec<u8> = row.get(1);
let token_a_blob: Vec<u8> = row.get(2);
let token_b_blob: Vec<u8> = 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<u8> = row.get(14);
let tx_pos: i64 = row.get(15);
Ok(ActiveTokenTokenPool {
pool_id: blob_to_display_hex::<OutPointHash>(&pool_blob)?,
nft_owner: hex::encode(&nft_owner_blob),
token_a_id: blob_to_display_hex::<TokenID>(&token_a_blob)?,
token_b_id: blob_to_display_hex::<TokenID>(&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>(&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<Vec<ActiveTokenTokenPool>> {
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<u8> = row.get(0);
let nft_owner_blob: Vec<u8> = row.get(1);
let token_a_blob: Vec<u8> = row.get(2);
let token_b_blob: Vec<u8> = 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<u8> = 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::<OutPointHash>(&pool_blob)?,
nft_owner: hex::encode(&nft_owner_blob),
token_a_id: blob_to_display_hex::<TokenID>(&token_a_blob)?,
token_b_id: blob_to_display_hex::<TokenID>(&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::<bitcoincash::Txid>(&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<Vec<ActiveTokenTokenPool>> {
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<u8> = row.get(0);
let nft_owner_blob: Vec<u8> = row.get(1);
let token_a_blob: Vec<u8> = row.get(2);
let token_b_blob: Vec<u8> = 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<u8> = 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::<OutPointHash>(&pool_blob)?,
nft_owner: hex::encode(&nft_owner_blob),
token_a_id: blob_to_display_hex::<TokenID>(&token_a_blob)?,
token_b_id: blob_to_display_hex::<TokenID>(&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::<bitcoincash::Txid>(&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"
);
}
}

43
src/db/ido/mod.rs
View file

@ -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<Vec<u8>> = 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<IdoAddResult> {
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<IdoAddResult> {
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<IdoAddResult> {
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<IdoAddResult> {
.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<IdoAddResult> {
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

4
src/db/init.rs
View file

@ -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) =

1
src/db/mod.rs
View file

@ -13,6 +13,7 @@ pub mod ido;
pub mod init;
pub mod moria;
pub mod oracle;
pub mod orbconstants;
pub mod search;
#[derive(Clone)]

88
src/db/orbconstants.rs Normal file
View file

@ -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<Network>) -> [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<Network>,
) -> Option<tokentoken_delegation::DelegationArtifacts> {
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<Network>,
) -> Option<tokenbch_delegation::DelegationArtifacts> {
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());
}
}

59
src/electrum.rs
View file

@ -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<Txid>, HashSet<Txid>, HashSet<Txid>, HashSet<Txid>)> {
/// The (defi, oracle, ido, bcmr) mempool txid sets.
pub type MempoolTxSets = (HashSet<Txid>, HashSet<Txid>, HashSet<Txid>, HashSet<Txid>);
/// Fetch defi (cauldron + delegation pools), oracle, ido and bcmr mempool
/// transactions
pub fn electrum_fetch_mempool(client: &Client, network: Option<Network>) -> Result<MempoolTxSets> {
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<u8> = [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)?;

138
src/index.rs
View file

@ -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<TeardownProbe> = vec![];
let mut tb_probes: Vec<TeardownProbe> = 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<ParsedContract> = 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<OutPointHash> = 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<TeardownProbe> = vec![];
let mut tb_probes: Vec<TeardownProbe> = 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<OutPointHash> = 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,
)

9
src/main.rs
View file

@ -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![

1
src/rpc/mod.rs
View file

@ -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;

63
src/rpc/tokenbch.rs Normal file
View file

@ -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?<token>")]
pub async fn list_active_pools(token: Option<&str>, conn: &State<DB>) -> CachedApiResult<Value> {
let Some(token) = token else {
return Err(bad_request(
ApiErrorCode::MissingParameters,
"Provide token",
));
};
let token = token
.parse::<TokenID>()
.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<DB>) -> CachedApiResult<Value> {
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<DB>) -> CachedApiResult<Value> {
let active = db_all_active_pools(&conn.cauldron_r)
.await
.map_err(db_error)?;
Ok(cached_ok(json!({ "active": active }), CACHE_NONE))
}