crates: snapshot

crates/brk_query/src/impl/price.rs

@@ -1,20 +1,68 @@
-use brk_error::Result;
+use std::sync::Arc;
+
+use brk_computer::prices::Vecs as PricesVecs;
+use brk_error::{Error, Result};
+use brk_oracle::{Config, Oracle, cents_to_bin};
 use brk_types::{
     Dollars, ExchangeRates, HistoricalPrice, HistoricalPriceEntry, Hour4, INDEX_EPOCH, Timestamp,
 };
-use vecdb::ReadableVec;
+use vecdb::{AnyVec, ReadableVec, VecIndex};
 
 use crate::Query;
 
 impl Query {
     pub fn live_price(&self) -> Result<Dollars> {
-        let mut oracle = self.computer().prices.live_oracle(self.indexer())?;
+        let base = self.cached_oracle()?;
+        Ok(match self.mempool() {
+            Some(mempool) => {
+                let mut oracle = (*base).clone();
+                oracle.process_histogram(&mempool.live_histogram());
+                oracle.price_dollars()
+            }
+            None => base.price_dollars(),
+        })
+    }
 
-        if let Some(mempool) = self.mempool() {
-            mempool.process_live_outputs(|iter| oracle.process_outputs(iter));
+    /// Oracle warmed by the last `window_size` committed blocks, seeded from
+    /// the last committed price. Cached per tip height; rebuilt on advance or
+    /// reorg. Reads are capped at `safe_lengths` so concurrent indexer writes
+    /// stay invisible.
+    fn cached_oracle(&self) -> Result<Arc<Oracle>> {
+        let safe_lengths = self.safe_lengths();
+        let height = safe_lengths.height;
+
+        if let Some(oracle) = self
+            .0
+            .live_oracle
+            .read()
+            .unwrap()
+            .as_ref()
+            .filter(|(h, _)| *h == height)
+            .map(|(_, o)| o.clone())
+        {
+            return Ok(oracle);
         }
 
-        Ok(oracle.price_dollars())
+        let cents_height = &self.computer().prices.spot.cents.height;
+        let last_cents = cents_height
+            .len()
+            .checked_sub(1)
+            .and_then(|i| cents_height.collect_one_at(i))
+            .ok_or_else(|| Error::NotFound("oracle prices not yet computed".to_string()))?;
+
+        let config = Config::default();
+        let seed_bin = cents_to_bin(last_cents.inner() as f64);
+        let tip = height.to_usize();
+        let warmup_range = tip.saturating_sub(config.window_size)..tip;
+        let oracle = Arc::new(Oracle::from_checkpoint(seed_bin, config, |o| {
+            PricesVecs::feed_blocks(o, self.indexer(), warmup_range, Some(&safe_lengths));
+        }));
+
+        let mut cache = self.0.live_oracle.write().unwrap();
+        if cache.as_ref().is_none_or(|(h, _)| *h != height) {
+            *cache = Some((height, oracle.clone()));
+        }
+        Ok(oracle)
     }
 
     pub fn historical_price(&self, timestamp: Option<Timestamp>) -> Result<HistoricalPrice> {