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

Merge branch 'orb+ido' into 'master'

Browse source 
tokentoken + tokenbch + ido

See merge request riftenlabs/riftenlabs-indexer!97
This commit is contained in:
Hossein Zoda 2026-08-04 17:54:52 +00:00
commit 69bf18de03
16 changed files with 2776 additions and 702 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?
{

196
src/db/bcmr/mod.rs
View file

@ -7,6 +7,7 @@ use crate::bcmr::parsedbcmr::{FileMeta, ParsedBCMR, Token, Uris, SOURCE_ON_CHAIN
use crate::db::blob::{display_hex_to_blob, FromBlob, ToBlob};
use anyhow::*;
use bitcoin_hashes::Hash;
use bitcoincash::{BlockHash, TokenID, Txid};
use log::info;
use riftenlabs_defi::chainutil::OutPointHash;
@ -116,6 +117,23 @@ pub async fn prepare_tables(pool: &SqlitePool) {
.execute(pool)
.await
.expect("failed to create index");
ensure_indexes(pool).await;
}
/// Always-run migration: indexes for the txid and blockhash lookups used by
/// mempool indexing. Safe on both new and existing bcmr.db.
pub async fn ensure_indexes(pool: &SqlitePool) {
sqlx::query("CREATE INDEX IF NOT EXISTS idx_auth_chain_txid ON auth_chain_entry(txid)")
.execute(pool)
.await
.expect("failed to create index");
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_auth_chain_blockhash ON auth_chain_entry(blockhash)",
)
.execute(pool)
.await
.expect("failed to create index");
}
/// Delete all well-known BCMR entries for the given source.
@ -131,6 +149,51 @@ pub async fn delete_entries_for_well_known<'e>(
Ok(())
}
/// Whether any auth chain entry (confirmed or mempool) was indexed from this tx.
pub async fn has_indexed_tx(pool: &SqlitePool, txid: &Txid) -> Result<bool> {
let row: Option<(i64,)> =
sqlx::query_as("SELECT 1 FROM auth_chain_entry WHERE txid = ? LIMIT 1")
.bind(txid.to_blob())
.fetch_optional(pool)
.await?;
Ok(row.is_some())
}
/// Txids of auth chain entries indexed from the mempool (all-zeros blockhash sentinel).
pub async fn get_unconfirmed_txids(pool: &SqlitePool) -> Result<Vec<Txid>> {
let rows = sqlx::query("SELECT DISTINCT txid FROM auth_chain_entry WHERE blockhash = ?")
.bind(BlockHash::all_zeros().to_blob())
.fetch_all(pool)
.await?;
let mut txids = Vec::with_capacity(rows.len());
for r in rows {
let txid_blob: Vec<u8> = r.get(0);
txids.push(Txid::from_blob(&txid_blob).context("failed to decode txid blob")?);
}
Ok(txids)
}
/// Delete a mempool-indexed (all-zeros blockhash) auth chain entry. Entries the
/// tx confirmed into are untouched: they carry the real blockhash by then.
pub async fn delete_unconfirmed_tx(pool: &SqlitePool, txid: &Txid) -> Result<()> {
sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ? AND txid = ?")
.bind(BlockHash::all_zeros().to_blob())
.bind(txid.to_blob())
.execute(pool)
.await?;
Ok(())
}
/// Delete all mempool-indexed (all-zeros blockhash) auth chain entries.
pub async fn clear_mempool(pool: &SqlitePool) -> Result<usize> {
let r = sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ?")
.bind(BlockHash::all_zeros().to_blob())
.execute(pool)
.await?;
Ok(r.rows_affected() as usize)
}
pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) -> Result<bool> {
let r = sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ?")
.bind(blockhash.to_blob())
@ -149,10 +212,18 @@ pub async fn insert_authheader(
height: usize,
bcmr_data: Option<Vec<u8>>,
) -> Result<()> {
// Upsert rather than INSERT OR REPLACE: REPLACE deletes the existing row,
// which cascades into bcmr_data and would throw away already-downloaded
// metadata every time a mempool-indexed entry is re-indexed on confirmation.
sqlx::query(
"INSERT OR REPLACE INTO auth_chain_entry
"INSERT INTO auth_chain_entry
(utxo, blockhash, txid, token_id, height, bcmr_data)
VALUES (?, ?, ?, ?, ?, ?)",
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(token_id, utxo) DO UPDATE SET
blockhash = excluded.blockhash,
txid = excluded.txid,
height = excluded.height,
bcmr_data = excluded.bcmr_data",
)
.bind(utxo.to_blob())
.bind(blockhash.to_blob())
@ -506,6 +577,127 @@ pub async fn filter_tokens_with_bcmr(
mod tests {
use super::*;
#[tokio::test]
async fn confirming_mempool_entry_preserves_downloaded_bcmr_data() {
use sqlx::sqlite::SqliteConnectOptions;
use std::str::FromStr;
// foreign_keys on, as in production (db/init.rs): with INSERT OR
// REPLACE the confirmation re-insert would delete the row and cascade
// away bcmr_data; the upsert must keep it.
let opts = SqliteConnectOptions::from_str("sqlite::memory:")
.unwrap()
.foreign_keys(true);
let pool = SqlitePool::connect_with(opts).await.unwrap();
prepare_tables(&pool).await;
let token_id = "0101010101010101010101010101010101010101010101010101010101010101"
.parse::<TokenID>()
.unwrap();
let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
.parse::<Txid>()
.unwrap();
let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
.parse::<OutPointHash>()
.unwrap();
// Indexed from mempool: sentinel blockhash
insert_authheader(
&pool,
&utxo,
&BlockHash::all_zeros(),
&txid,
&token_id,
0,
Some(vec![0x6a]),
)
.await
.unwrap();
assert_eq!(get_unconfirmed_txids(&pool).await.unwrap(), vec![txid]);
// Downloader stores the metadata while the tx is still unconfirmed
let parsed = ParsedBCMR {
name: "Test".into(),
description: "".into(),
token: Token {
category: token_id.to_string(),
symbol: "TST".into(),
decimals: 0,
},
uris: Uris {
web: None,
icon: None,
},
filemeta: FileMeta {
expected_hash: None,
actual_hash: None,
source: SOURCE_ON_CHAIN.into(),
},
};
insert_bcmr_data(&pool, &token_id, &utxo, &parsed)
.await
.unwrap();
// Tx confirms: same entry re-indexed with the real blockhash
let real_blockhash = "1111111111111111111111111111111111111111111111111111111111111111"
.parse::<BlockHash>()
.unwrap();
insert_authheader(
&pool,
&utxo,
&real_blockhash,
&txid,
&token_id,
0,
Some(vec![0x6a]),
)
.await
.unwrap();
assert!(get_unconfirmed_txids(&pool).await.unwrap().is_empty());
let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM bcmr_data")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(count.0, 1, "downloaded bcmr_data must survive confirmation");
let bh: (Vec<u8>,) = sqlx::query_as("SELECT blockhash FROM auth_chain_entry")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(bh.0, real_blockhash.to_blob());
}
#[tokio::test]
async fn delete_unconfirmed_tx_leaves_confirmed_entries() {
let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
prepare_tables(&pool).await;
let token_id = "0101010101010101010101010101010101010101010101010101010101010101"
.parse::<TokenID>()
.unwrap();
let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
.parse::<Txid>()
.unwrap();
let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
.parse::<OutPointHash>()
.unwrap();
let real_blockhash = "1111111111111111111111111111111111111111111111111111111111111111"
.parse::<BlockHash>()
.unwrap();
insert_authheader(&pool, &utxo, &real_blockhash, &txid, &token_id, 0, None)
.await
.unwrap();
// A confirmed entry from the same tx must not be dropped by the
// mempool cleanup pass.
delete_unconfirmed_tx(&pool, &txid).await.unwrap();
assert!(has_indexed_tx(&pool, &txid).await.unwrap());
}
#[tokio::test]
async fn test_update_bcmr_failure_multiple_tokens_same_utxo_txid() {
let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();

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

@ -20,6 +20,7 @@ pub mod mempool;
pub mod ohlcv;
pub mod pool;
pub mod poolvisitor;
pub mod tokenbch;
pub mod tokenlist;
pub mod tokentoken;
pub mod tx;
@ -48,6 +49,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"
);
}
}

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

File diff suppressed because one or more lines are too long

9
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) =
@ -115,6 +117,8 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul
if !db_exists {
bcmr_prepare_tables(&bcmr_db_write).await;
}
// Always-run migration: safe on both new and existing bcmr.db
crate::db::bcmr::ensure_indexes(&bcmr_db_write).await;
// Initialize CRC20 database
let (db_exists, crc20_db_write, crc20_db_read) =
@ -142,6 +146,9 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul
create_db_pool(&db_path(db_dir, "ido.db"), read_slots.ido).await;
if !db_exists {
ido_prepare_tables(&ido_db_write).await;
crate::db::ido::set_db_version(&ido_db_write).await?;
} else {
crate::db::ido::check_db_version(&ido_db_read).await?;
}
Ok(DB {

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)]

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

@ -0,0 +1,113 @@
// 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 ORB IdoParams NFT category (display byte order, like a token id) on
/// chipnet. Every IDO preinit must include this NFT (input #1, preserved at
/// output #10); its commitment carries the economic parameters the announced
/// IDO parameters are validated against.
///
/// From libriften `orb/v0/chipnet.ts` (`ORB_V0_CHIPNET.idoParams.token`).
const CHIPNET_IDO_PARAMS_NFT_CATEGORY: &[u8; 32] = &[
0xc9, 0x16, 0x7d, 0x12, 0x39, 0x46, 0x27, 0xba, 0x28, 0x30, 0x70, 0x5f, 0xb2, 0xb0, 0xf0, 0x0b,
0x49, 0xeb, 0x28, 0x46, 0x9e, 0x65, 0xda, 0x53, 0x69, 0x12, 0xd0, 0x50, 0x75, 0x89, 0x08, 0x00,
];
// TODO:: set the ido params nft category for mainnet once ORB v0 deploys there
// (libriften orb/v0/mainnet.ts is still all-zero too).
const MAINNET_IDO_PARAMS_NFT_CATEGORY: &[u8; 32] = &[0u8; 32];
/// The ORB IdoParams NFT category for a network (all-zero = unconfigured; the
/// preinit's params-NFT check then never matches, so no IDO is admitted —
/// fail-closed).
pub fn ido_params_nft_category(network: Option<Network>) -> [u8; 32] {
match network {
Some(Network::Chipnet) => *CHIPNET_IDO_PARAMS_NFT_CATEGORY,
_ => *MAINNET_IDO_PARAMS_NFT_CATEGORY,
}
}
/// 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());
}
}

70
src/electrum.rs
View file

@ -7,18 +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)> {
@ -40,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) and oracle mempool transactions
pub fn electrum_fetch_mempool(
client: &Client,
) -> Result<(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.
@ -82,6 +94,13 @@ pub fn electrum_fetch_mempool(
"operation": "union"
});
// BCMR registrations carry an `OP_RETURN "BCMR"` output (genesis or auth
// chain update). Auth chain transfers without a BCMR output are only
// picked up once confirmed.
let bcmr_filter = json!({
"scriptpubkey": hex::encode(BCMR_PREFIX),
});
let fetch_txs = |filter: Value| -> Result<HashSet<Txid>> {
let response = client.raw_call("mempool.get", [Param::Value(filter)])?;
@ -108,15 +127,22 @@ 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)?;
let bcmr_txs = fetch_txs(bcmr_filter)?;
Ok((defi_txs, oracle_txs, ido_txs))
Ok((defi_txs, oracle_txs, ido_txs, bcmr_txs))
}
/// Fetch blockchain tip from electrum server

181
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,
@ -77,8 +78,8 @@ pub async fn update_mempool(
db::cauldron::mempool::load_mempool(&db.cauldron_w).await?;
let electrum_clone = electrum.clone();
let (cauldron_txs, oracle_txs, ido_txs) = tokio::task::spawn_blocking(move || {
electrum_fetch_mempool(&electrum_clone.lock().unwrap())
let (cauldron_txs, oracle_txs, ido_txs, bcmr_txs) = tokio::task::spawn_blocking(move || {
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),
)
@ -218,6 +277,47 @@ pub async fn update_mempool(
let txs_to_add = ttor_sorted_kahn(txs_to_add);
db::ido::index_txs(network, &db.ido_w, &txs_to_add, None).await?;
// bcmr updates: index newly registered BCMRs without waiting for a confirmation
// Drop mempool-indexed entries whose tx left the mempool: either it
// confirmed (index_blocks already re-stamped the entry with the real
// blockhash, so it no longer matches the sentinel) or it was evicted or
// replaced and must not linger as a stale auth head.
for txid in db::bcmr::get_unconfirmed_txids(&db.bcmr_w).await? {
if !bcmr_txs.contains(&txid) {
db::bcmr::delete_unconfirmed_tx(&db.bcmr_w, &txid).await?;
}
}
// filter txs already indexed into the auth chain
let mut bcmr_to_add = Vec::new();
for txid in bcmr_txs {
if !db::bcmr::has_indexed_tx(&db.bcmr_w, &txid).await? {
bcmr_to_add.push(txid);
}
}
let bcmr_electrum = electrum.clone();
let txs_to_add: Vec<Transaction> = tokio::task::spawn_blocking(move || {
bcmr_to_add
.into_iter()
.filter_map(
|txid| match electrum_get_tx(&bcmr_electrum.lock().unwrap(), &txid) {
Ok(tx) => Some(tx),
Err(e) => {
info!("Failed to get mempool tx {txid}: {e}");
None
}
},
)
.collect()
})
.await?;
// an auth chain can have several unconfirmed txs in flight; index parents first
let txs_to_add = ttor_sorted_kahn(txs_to_add);
index_bcmr(&db.bcmr_w, &BlockHash::all_zeros(), txs_to_add).await?;
Ok(())
}
@ -423,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;
}
@ -455,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.
@ -464,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,
)

10
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();
@ -232,6 +233,7 @@ async fn start_program(
db::oracle::clear_mempool(&db.oracle_w).await.unwrap();
db::moria::clear_mempool(&db.moria_w).await.unwrap();
db::bcmr::clear_mempool(&db.bcmr_w).await.unwrap();
let indexing_in_progress_clone = indexing_in_progress.clone();
let ibd_state_clone = ibd_state.clone();
@ -678,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))
}