diff --git a/src/db/cauldron/tokenlist/token_utils.rs b/src/db/cauldron/tokenlist/token_utils.rs index 8794a57..030868d 100644 --- a/src/db/cauldron/tokenlist/token_utils.rs +++ b/src/db/cauldron/tokenlist/token_utils.rs @@ -11,6 +11,7 @@ use sqlx::SqlitePool; use crate::db::blob::display_hex_to_blob; use crate::db::cauldron::pool::{get_injections_between, get_pool_period_snapshot}; use crate::db::oracle::get_closest; +use crate::db::oracle::oracle_cash::get_oracle_cash_closest; use crate::rpc::apy::apyaggregator::APYAggregator; use crate::rpc::apy::poolperiod::split_at_injections; use malachite::base::num::arithmetic::traits::FloorSqrt; @@ -54,15 +55,31 @@ pub fn dec_to_f64_bounded(d: Decimal) -> f64 { } pub const SATS_PER_BCH: i64 = 100_000_000; -pub const ORACLE_SCALE: i64 = 1_000_000; +/// On-chain Delphi oracle scale: prices are stored in cents +/// (e.g. $384.24 → 38424). Same unit as the oracles.cash feed. +pub const ORACLE_SCALE: i64 = 100; +/// Timestamps before this value are looked up in the on-chain Delphi indexed table. +/// Timestamps on or after are looked up in the oracle_cash table (oracles.cash feed). +/// Value: 2026-04-23 00:00:00 UTC (last reliable Delphi oracle update). +const ORACLE_CUTOFF_TS: i64 = 1776902400; pub async fn usd_per_bch_at_or_before(oracle_pool: &SqlitePool, ts: i64) -> Decimal { - match get_closest(oracle_pool, &None, ts).await { - Ok(Some(e)) => { - Decimal::from_i64(e.oracle_price).unwrap_or(Decimal::ZERO) - / Decimal::from_i64(ORACLE_SCALE).unwrap() + if ts < ORACLE_CUTOFF_TS { + match get_closest(oracle_pool, &None, ts).await { + Ok(Some(e)) => { + Decimal::from_i64(e.oracle_price).unwrap_or(Decimal::ZERO) + / Decimal::from_i64(ORACLE_SCALE).unwrap() + } + _ => Decimal::ZERO, + } + } else { + // oracles.cash prices are in cents; divide by 100 to get USD/BCH + match get_oracle_cash_closest(oracle_pool, ts).await { + Ok(Some(e)) => { + Decimal::from_i64(e.oracle_price).unwrap_or(Decimal::ZERO) / Decimal::from(100) + } + _ => Decimal::ZERO, } - _ => Decimal::ZERO, } } diff --git a/src/db/init.rs b/src/db/init.rs index 0a6a47f..6a0c798 100644 --- a/src/db/init.rs +++ b/src/db/init.rs @@ -16,6 +16,7 @@ use crate::db::cauldron::config::check_db_version; use crate::db::cauldron::prepare_tables as cauldron_prepare_tables; use crate::db::crc20::prepare_tables as crc20_prepare_tables; use crate::db::moria::prepare_tables as moria_prepare_tables; +use crate::db::oracle::oracle_cash::prepare_oracle_cash_tables; use crate::db::oracle::prepare_tables as oracle_prepare_tables; /// Create read and write database pools for a given database path @@ -124,6 +125,8 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul if !db_exists { oracle_prepare_tables(&oracle_db_write).await; } + // Always-run migration: safe on both new and existing oracle.db + prepare_oracle_cash_tables(&oracle_db_write).await; // Initialize moria lending database let (db_exists, moria_db_write, moria_db_read) = diff --git a/src/db/oracle/mod.rs b/src/db/oracle/mod.rs index 181c00c..774558b 100644 --- a/src/db/oracle/mod.rs +++ b/src/db/oracle/mod.rs @@ -3,6 +3,8 @@ // 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 +pub mod oracle_cash; + use anyhow::{bail, Result}; use bitcoin_hashes::{hex::ToHex, Hash}; use bitcoincash::{BlockHash, TokenID, Transaction, Txid}; diff --git a/src/db/oracle/oracle_cash.rs b/src/db/oracle/oracle_cash.rs new file mode 100644 index 0000000..a517930 --- /dev/null +++ b/src/db/oracle/oracle_cash.rs @@ -0,0 +1,160 @@ +// Copyright (C) 2024-2026 Whiterun LLC +// +// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. +// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html + +use anyhow::{bail, Result}; +use sqlx::{Row, SqlitePool}; + +#[derive(Debug, Clone, serde::Serialize)] +pub struct OracleCashPrice { + pub oracle_timestamp: i64, + /// Price in cents (same convention as the on-chain Delphi oracle) + pub oracle_price: i64, + pub message_sequence: Option, +} + +/// Creates the oracle_cash_price table if it doesn't already exist. +/// Safe to call on both new and existing oracle.db instances. +pub async fn prepare_oracle_cash_tables(pool: &SqlitePool) { + sqlx::query( + "CREATE TABLE IF NOT EXISTS oracle_cash_price ( + oracle_timestamp INTEGER NOT NULL PRIMARY KEY, + oracle_price INTEGER NOT NULL, + message_sequence INTEGER + )", + ) + .execute(pool) + .await + .expect("failed to create oracle_cash_price table"); + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_oracle_cash_seq + ON oracle_cash_price(message_sequence) + WHERE message_sequence IS NOT NULL", + ) + .execute(pool) + .await + .expect("failed to create oracle_cash_seq index"); +} + +pub async fn upsert_oracle_cash_price( + pool: &SqlitePool, + oracle_timestamp: i64, + oracle_price: i64, + message_sequence: Option, +) -> Result<()> { + sqlx::query( + "INSERT OR REPLACE INTO oracle_cash_price + (oracle_timestamp, oracle_price, message_sequence) + VALUES (?, ?, ?)", + ) + .bind(oracle_timestamp) + .bind(oracle_price) + .bind(message_sequence) + .execute(pool) + .await + .map_err(|e| anyhow::anyhow!("failed to upsert oracle_cash_price: {}", e))?; + Ok(()) +} + +/// Returns the earliest oracle_timestamp stored, or None if the table is empty. +pub async fn get_oracle_cash_min_timestamp(pool: &SqlitePool) -> Result> { + let row: (Option,) = sqlx::query_as("SELECT MIN(oracle_timestamp) FROM oracle_cash_price") + .fetch_one(pool) + .await?; + Ok(row.0) +} + +pub async fn get_oracle_cash_closest( + pool: &SqlitePool, + timestamp: i64, +) -> Result> { + let row = sqlx::query( + "SELECT oracle_timestamp, oracle_price, message_sequence + FROM oracle_cash_price + WHERE oracle_timestamp <= ? + ORDER BY oracle_timestamp DESC + LIMIT 1", + ) + .bind(timestamp) + .fetch_optional(pool) + .await?; + + Ok(row.map(|r| OracleCashPrice { + oracle_timestamp: r.get(0), + oracle_price: r.get(1), + message_sequence: r.get(2), + })) +} + +pub async fn get_oracle_cash_range( + pool: &SqlitePool, + start: i64, + end: i64, +) -> Result> { + let rows = sqlx::query( + "SELECT oracle_timestamp, oracle_price, message_sequence + FROM oracle_cash_price + WHERE oracle_timestamp BETWEEN ? AND ? + ORDER BY oracle_timestamp ASC", + ) + .bind(start) + .bind(end) + .fetch_all(pool) + .await?; + + Ok(rows + .iter() + .map(|r| OracleCashPrice { + oracle_timestamp: r.get(0), + oracle_price: r.get(1), + message_sequence: r.get(2), + }) + .collect()) +} + +pub async fn get_oracle_cash_range_with_step( + pool: &SqlitePool, + start: i64, + end: i64, + step: i64, +) -> Result> { + let total_intervals = (end - start) / step; + const MAX_INTERVALS: i64 = 10_000; + if total_intervals > MAX_INTERVALS { + bail!( + "Too many intervals ({} > {})", + total_intervals, + MAX_INTERVALS + ); + } + + let all = get_oracle_cash_range(pool, start, end).await?; + + let mut buckets: Vec = Vec::new(); + let mut next_threshold = start + step; + let mut last_entry: Option = None; + + for entry in all { + while entry.oracle_timestamp >= next_threshold { + if let Some(e) = last_entry.take() { + buckets.push(e); + } + next_threshold += step; + } + last_entry = Some(entry); + } + + if let Some(e) = last_entry { + if buckets + .last() + .map(|b| b.oracle_timestamp != e.oracle_timestamp) + .unwrap_or(true) + { + buckets.push(e); + } + } + + Ok(buckets) +} diff --git a/src/db/search/mod.rs b/src/db/search/mod.rs index 31a78d6..3d0ff17 100644 --- a/src/db/search/mod.rs +++ b/src/db/search/mod.rs @@ -163,7 +163,7 @@ pub async fn search_tokens_by_volume( ] .concat(); - combined_tokens.sort_by(|a, b| a.0.to_lowercase().cmp(&b.0.to_lowercase())); + combined_tokens.sort_by_key(|a| a.0.to_lowercase()); combined_tokens.dedup_by(|a, b| { if a.0.eq_ignore_ascii_case(&b.0) { @@ -179,7 +179,7 @@ pub async fn search_tokens_by_volume( let mut result = token_volume(cauldron_pool, combined_tokens).await?; - result.sort_by(|a, b| b.3.cmp(&a.3)); + result.sort_by_key(|b| std::cmp::Reverse(b.3)); Ok(result) } diff --git a/src/main.rs b/src/main.rs index 7cba654..5072105 100644 --- a/src/main.rs +++ b/src/main.rs @@ -79,6 +79,7 @@ mod db; mod def; mod electrum; mod index; +mod oracle_cash; mod rpc; mod signal; mod timeutil; @@ -118,6 +119,7 @@ async fn start_program( BCMRDownloader, WellKnownDownloader, CRC20Fetcher, + oracle_cash::OracleCashFetcher, Arc, Arc, // indexing_in_progress )> { @@ -352,11 +354,15 @@ async fn start_program( spawn_token_metrics_updater(db.clone(), indexing_in_progress.clone()); + let mut oracle_cash_fetcher = oracle_cash::OracleCashFetcher::new(); + oracle_cash_fetcher.start(db.oracle_w.clone(), db.oracle_r.clone()); + Ok(( db, bcmrdownloader, wellknowndownloader, crc20fetcher, + oracle_cash_fetcher, ibd_state, indexing_in_progress, )) @@ -383,6 +389,7 @@ async fn launch() -> _ { bcmrdownloader, wellknowndownloader, crc20fetcher, + oracle_cash_fetcher, ibd_state, indexing_in_progress, ) = match start_program(config).await { @@ -570,10 +577,11 @@ async fn launch() -> _ { .manage(dbpool) .manage(ibd_state) .manage(ohlcv_state) - // give rocket ownership of downloader to ensure thread isn't dropped + // give rocket ownership of downloaders/fetchers to ensure threads aren't dropped .manage(bcmrdownloader) .manage(wellknowndownloader) .manage(crc20fetcher) + .manage(oracle_cash_fetcher) .mount( "/cauldron/", routes![ @@ -611,7 +619,9 @@ async fn launch() -> _ { routes![ rpc::oracle::oracle_get_closest, rpc::oracle::oracle_get_range, - rpc::oracle::oracle_get_history + rpc::oracle::oracle_get_history, + rpc::oracle::oracle_cash_closest, + rpc::oracle::oracle_cash_history, ], ) .mount( diff --git a/src/oracle_cash.rs b/src/oracle_cash.rs new file mode 100644 index 0000000..05c83e9 --- /dev/null +++ b/src/oracle_cash.rs @@ -0,0 +1,230 @@ +// 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 + +//! Background task that polls oracles.cash and keeps oracle_cash_price up-to-date. +//! This is a parallel feed alongside the on-chain Delphi oracle; both write to +//! oracle.db but to separate tables. + +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::Duration, +}; + +use log::{info, warn}; +use serde::Deserialize; +use sqlx::SqlitePool; +use tokio::task::JoinHandle; + +use crate::db::oracle::oracle_cash::{get_oracle_cash_min_timestamp, upsert_oracle_cash_price}; +use crate::timeutil::time_now; + +const ORACLES_CASH_URL: &str = "https://oracles.generalprotocols.com"; +/// General Protocols BCH/USD oracle public key +const BCH_USD_ORACLE_PUBKEY: &str = + "02d09db08af1ff4e8453919cc866a4be427d7bfe18f2c05e5444c196fcf6fd2818"; + +const POLL_INTERVAL: Duration = Duration::from_secs(300); // 5 minutes +const REQUEST_TIMEOUT: Duration = Duration::from_secs(20); +const BACKFILL_AGGREGATION: i64 = 3600; // 1-hour buckets +/// Backfill oracle_cash data starting from this timestamp. +/// Everything before this point is served from the on-chain Delphi indexed table, +/// so oracle_cash only needs to cover from here onward. +/// Value: 2026-04-23 00:00:00 UTC (last reliable Delphi oracle update). +const ORACLE_CUTOFF_TS: i64 = 1776902400; + +// ── API response types ──────────────────────────────────────────────────────── + +#[derive(Deserialize, Debug)] +struct MessageMetrics { + #[serde(rename = "maxMessageSequence")] + max_message_sequence: i64, + /// Price in cents (same unit as on-chain Delphi: 38424 = $384.24) + #[serde(rename = "currentPrice")] + current_price: i64, + #[serde(rename = "maxMessageTimestamp")] + max_message_timestamp: i64, +} + +#[derive(Deserialize, Debug)] +struct OracleEntry { + #[serde(rename = "publicKey")] + public_key: String, + #[serde(rename = "messageMetrics")] + message_metrics: Option, +} + +#[derive(Deserialize, Debug)] +struct OraclesResponse { + oracles: Vec, +} + +#[derive(Deserialize, Debug)] +struct PriceGraphPoint { + #[serde(rename = "averageTimestamp")] + average_timestamp: f64, + /// Average price in cents over the aggregation window + #[serde(rename = "averagePrice")] + average_price: f64, +} + +#[derive(Deserialize, Debug)] +struct PriceGraphResponse { + #[serde(rename = "priceGraphPoints")] + price_graph_points: Vec, +} + +// ── Fetch helpers ───────────────────────────────────────────────────────────── + +/// Returns (timestamp_secs, price_cents, message_sequence) +async fn fetch_current_price(client: &reqwest::Client) -> anyhow::Result<(i64, i64, i64)> { + let url = format!("{}/api/v1/oracles", ORACLES_CASH_URL); + let resp: OraclesResponse = client + .get(&url) + .timeout(REQUEST_TIMEOUT) + .send() + .await? + .error_for_status()? + .json() + .await?; + + let oracle = resp + .oracles + .into_iter() + .find(|o| o.public_key == BCH_USD_ORACLE_PUBKEY) + .ok_or_else(|| anyhow::anyhow!("BCH/USD oracle not found in oracles.cash response"))?; + + let m = oracle + .message_metrics + .ok_or_else(|| anyhow::anyhow!("No messageMetrics for BCH/USD oracle"))?; + + Ok(( + m.max_message_timestamp, + m.current_price, + m.max_message_sequence, + )) +} + +/// Fetches hourly price history from ORACLE_CUTOFF_TS to now and inserts into the DB. +async fn backfill_history(client: &reqwest::Client, pool: &SqlitePool) -> anyhow::Result<()> { + let now = time_now(); + let min_ts = ORACLE_CUTOFF_TS; + + let url = format!( + "{}/api/v2/priceGraphPoints?publicKey={}&minMessageTimestamp={}&maxMessageTimestamp={}&aggregationPeriod={}", + ORACLES_CASH_URL, BCH_USD_ORACLE_PUBKEY, min_ts, now, BACKFILL_AGGREGATION + ); + + let resp: PriceGraphResponse = client + .get(&url) + .timeout(Duration::from_secs(30)) + .send() + .await? + .error_for_status()? + .json() + .await?; + + let count = resp.price_graph_points.len(); + for point in resp.price_graph_points { + let ts = point.average_timestamp as i64; + let price = point.average_price as i64; + // No real sequence available for aggregated points + upsert_oracle_cash_price(pool, ts, price, None).await?; + } + + info!("oracle_cash: backfilled {count} hourly price points"); + Ok(()) +} + +// ── Fetcher ─────────────────────────────────────────────────────────────────── + +pub struct OracleCashFetcher { + keep_running: Arc, + task: Option>, +} + +impl OracleCashFetcher { + pub fn new() -> Self { + Self { + keep_running: Arc::new(AtomicBool::new(true)), + task: None, + } + } + + pub fn start(&mut self, oracle_w: SqlitePool, oracle_r: SqlitePool) { + let keep_running = self.keep_running.clone(); + + self.task = Some(tokio::spawn(async move { + let client = reqwest::Client::new(); + + // Backfill if the table is empty or if the earliest entry is after + // ORACLE_CUTOFF_TS (meaning we're missing data from when the Delphi oracle + // stopped being updated). + let needs_backfill = match get_oracle_cash_min_timestamp(&oracle_r).await { + Ok(None) => { + info!("oracle_cash: table empty, will backfill from ORACLE_CUTOFF_TS"); + true + } + Ok(Some(min_ts)) => { + if min_ts > ORACLE_CUTOFF_TS { + info!( + "oracle_cash: earliest entry {min_ts} is after cutoff {ORACLE_CUTOFF_TS} — will backfill" + ); + true + } else { + info!( + "oracle_cash: earliest entry {min_ts} covers cutoff, skipping backfill" + ); + false + } + } + Err(e) => { + warn!("oracle_cash: could not check earliest timestamp: {e}"); + false + } + }; + + if needs_backfill { + if let Err(e) = backfill_history(&client, &oracle_w).await { + warn!("oracle_cash: backfill failed: {e}"); + } + } + + loop { + if !keep_running.load(Ordering::Relaxed) { + info!("oracle_cash: exiting fetch task"); + return; + } + + match fetch_current_price(&client).await { + Ok((ts, price_cents, seq)) => { + match upsert_oracle_cash_price(&oracle_w, ts, price_cents, Some(seq)).await + { + Ok(()) => info!( + "oracle_cash: stored price={price_cents} cents ts={ts} seq={seq}" + ), + Err(e) => warn!("oracle_cash: failed to store price: {e}"), + } + } + Err(e) => warn!("oracle_cash: failed to fetch current price: {e}"), + } + + tokio::time::sleep(POLL_INTERVAL).await; + } + })); + } +} + +impl Drop for OracleCashFetcher { + fn drop(&mut self) { + self.keep_running.store(false, Ordering::SeqCst); + if let Some(task) = self.task.take() { + task.abort(); + info!("oracle_cash fetch task aborted"); + } + } +} diff --git a/src/rpc/oracle.rs b/src/rpc/oracle.rs index cdd3425..37050c8 100644 --- a/src/rpc/oracle.rs +++ b/src/rpc/oracle.rs @@ -3,6 +3,9 @@ // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html +use crate::db::oracle::oracle_cash::{ + get_oracle_cash_closest, get_oracle_cash_range, get_oracle_cash_range_with_step, +}; use crate::db::oracle::{get_closest, get_range, get_range_with_step}; use crate::db::DB; use crate::rpc::err::{bad_request, db_error, ApiErrorCode, CachedApiResult}; @@ -177,3 +180,83 @@ pub async fn oracle_get_history( CACHE_AGGREGATE, )) } + +/// Get the closest BCH/USD price from the oracles.cash feed for a given timestamp. +/// +/// Status: Stable +/// +/// - timestamp: Unix timestamp in seconds (optional, defaults to now) +/// +/// Returns `null` if no data is available. +/// +/// `oracle_price` is in **cents**. Divide by 100 to get USD. +#[get("/cash/closest?")] +pub async fn oracle_cash_closest( + timestamp: Option, + db: &State, +) -> CachedApiResult { + let ts = timestamp.unwrap_or_else(time_now); + let entry = get_oracle_cash_closest(&db.oracle_r, ts) + .await + .map_err(db_error)?; + Ok(cached_ok( + entry.map_or(serde_json::Value::Null, |e| { + serde_json::to_value(e).unwrap() + }), + CACHE_AGGREGATE, + )) +} + +/// Get historical BCH/USD prices from the oracles.cash feed. +/// +/// Status: Stable +/// +/// - start: Unix timestamp for start of period +/// - end: Unix timestamp for end of period (optional, defaults to now) +/// - stepsize: Seconds per interval (optional) +/// +/// `oracle_price` values are in **cents**. +/// +/// **Response Example:** +/// ```json +/// [ +/// { "oracle_timestamp": 1709468902, "oracle_price": 38424, "message_sequence": 12345 }, +/// { "oracle_timestamp": 1709472502, "oracle_price": 38501, "message_sequence": null } +/// ] +/// ``` +#[get("/cash/history?&&")] +pub async fn oracle_cash_history( + start: Option, + end: Option, + stepsize: Option, + db: &State, +) -> CachedApiResult { + let current_timestamp = time_now(); + let start_ts = start.unwrap_or(current_timestamp - 30 * 24 * 3600); + let end_ts = end.unwrap_or(current_timestamp); + + let entries = if let Some(step) = stepsize { + get_oracle_cash_range_with_step(&db.oracle_r, start_ts, end_ts, step) + .await + .map_err(|e| { + bad_request( + ApiErrorCode::InvalidParameters, + &format!("Query error: {e}"), + ) + })? + } else { + get_oracle_cash_range(&db.oracle_r, start_ts, end_ts) + .await + .map_err(|e| { + bad_request( + ApiErrorCode::InvalidParameters, + &format!("Query error: {e}"), + ) + })? + }; + + Ok(cached_ok( + serde_json::to_value(entries).unwrap(), + CACHE_AGGREGATE, + )) +}