Skip to main content

evm_oracle_state/
redstone.rs

1use std::{
2    borrow::Cow,
3    collections::{BTreeMap, BTreeSet},
4    sync::Arc,
5    time::{SystemTime, UNIX_EPOCH},
6};
7
8use alloy_network::Ethereum;
9use alloy_primitives::{Address, B256, I256, U256, keccak256};
10use alloy_rpc_types_eth::Filter;
11use alloy_sol_types::sol;
12use evm_fork_cache::{
13    StateUpdate, StateView,
14    cache::EvmCache,
15    reactive::{
16        ChainStatus, HandlerError, HandlerId, HandlerOutcome, HookSignal, InvalidationReason,
17        InvalidationRequest, LogInterest, ReactiveContext, ReactiveEffect, ReactiveHandler,
18        ReactiveInput, ReactiveInterest, ReportTag, RouteKeySpec, StateEffectQuality,
19    },
20    state_update::PurgeScope,
21};
22
23use crate::{
24    ANSWER_UPDATED_TOPIC, AdapterFuture, AssetId, Denomination, EvmCacheChainlinkReader, Feed,
25    FeedConfig, FeedId, FeedMetadata, FeedRegistration, FeedSource, ORACLE_SIGNAL_NAMESPACE,
26    OracleAdapterFeedSkip, OracleAdapterId, OracleAdapterPlugin, OracleAdapterSkipReason,
27    OracleDiscoveredFeed, OracleDiscoveryContext, OracleDiscoveryReport, OracleError,
28    OracleFeedStatus, OraclePriceUpdate, OracleSignalKind, OracleStorageSync, OracleValueSource,
29    OracleValueStatus, REDSTONE_VALUE_UPDATE_TOPIC, RedstoneValueUpdate, RoundData,
30    StalenessPolicy, decode_answer_updated, decode_redstone_value_update, state::classify_round,
31};
32
33sol! {
34    interface RedstonePriceFeedInterface {
35        function getDataFeedId() external view returns (bytes32);
36        function getPriceFeedAdapter() external view returns (address);
37    }
38}
39
40use RedstonePriceFeedInterface::{getDataFeedIdCall, getPriceFeedAdapterCall};
41
42const ADAPTER_ID: &str = "evm-oracle-state.redstone";
43const HANDLER_ID: &str = "evm-oracle-state.redstone";
44const REDSTONE_NO_ROUNDS_ROUND_ID: u64 = 1;
45const REDSTONE_MULTI_FEED_DATA_FEEDS_STORAGE_LOCATION: B256 =
46    alloy_primitives::b256!("5e9fb4cb0eb3c2583734d3394f30bb14b241acb9b3a034f7e7ba1a62db4370f1");
47const REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION: B256 =
48    alloy_primitives::b256!("4dd0c77efa6f6d590c97573d8c70b714546e7311202ff7c11c484cc841d91bfc");
49const REDSTONE_PRICE_FEEDS_LATEST_UPDATE_TIMESTAMPS_STORAGE_LOCATION: B256 =
50    alloy_primitives::b256!("3d01e4d77237ea0f771f1786da4d4ff757fcba6a92933aa53b1dcef2d6bd6fe2");
51const REDSTONE_PRICE_FEEDS_WITH_ROUNDS_ROUND_TIMESTAMPS_MAPPING_STORAGE_LOCATION: B256 =
52    alloy_primitives::b256!("207e00944d909d1224f0c253d58489121d736649f8393199f55eecf4f0cf3eb0");
53const REDSTONE_PRICE_FEEDS_WITH_ROUNDS_LATEST_ROUND_ID_STORAGE_LOCATION: B256 =
54    alloy_primitives::b256!("c68d7f1ee07d8668991a8951e720010c9d44c2f11c06b5cac61fbc4083263938");
55const REDSTONE_DATA_TIMESTAMP_BITS: usize = 48;
56const REDSTONE_BLOCK_TIMESTAMP_BITS: usize = 48;
57const REDSTONE_MULTI_FEED_VALUE_BITS: usize = 152;
58const REDSTONE_MULTI_FEED_BLOCK_TIMESTAMP_OFFSET_BITS: usize = REDSTONE_DATA_TIMESTAMP_BITS;
59const REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS: usize =
60    REDSTONE_DATA_TIMESTAMP_BITS + REDSTONE_BLOCK_TIMESTAMP_BITS;
61const REDSTONE_MULTI_FEED_IS_VALUE_BIGGER_OFFSET_BITS: usize =
62    REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS + REDSTONE_MULTI_FEED_VALUE_BITS;
63const REDSTONE_PRICE_FEEDS_BLOCK_TIMESTAMP_OFFSET_BITS: usize = 128;
64
65/// Declarative RedStone push-feed registration.
66#[derive(Clone, Debug, PartialEq, Eq)]
67pub struct RedstoneFeed {
68    price_feed: Address,
69    adapter: Option<Address>,
70    data_feed_id: Option<B256>,
71    feed_id: Option<FeedId>,
72    label: Option<String>,
73    base: Option<AssetId>,
74    quote: Option<Denomination>,
75    staleness: StalenessPolicy,
76}
77
78impl RedstoneFeed {
79    /// Register a RedStone price-feed wrapper and discover adapter/feed id from it.
80    pub fn new(price_feed: Address) -> Self {
81        Self {
82            price_feed,
83            adapter: None,
84            data_feed_id: None,
85            feed_id: None,
86            label: None,
87            base: None,
88            quote: None,
89            staleness: StalenessPolicy::default(),
90        }
91    }
92
93    /// Register a RedStone price-feed wrapper and discover adapter/feed id from it.
94    pub fn price_feed(price_feed: Address) -> Self {
95        Self::new(price_feed)
96    }
97
98    /// Register a RedStone price-feed wrapper with an explicit adapter and data feed id.
99    pub fn push(price_feed: Address, adapter: Address, data_feed_id: B256) -> Self {
100        Self::new(price_feed)
101            .adapter(adapter)
102            .data_feed_id(data_feed_id)
103    }
104
105    /// Return the user-facing price-feed wrapper.
106    pub fn price_feed_address(&self) -> Address {
107        self.price_feed
108    }
109
110    /// Set the RedStone adapter/event-source address.
111    pub fn adapter(mut self, adapter: Address) -> Self {
112        self.adapter = Some(adapter);
113        self
114    }
115
116    /// Set the RedStone data feed id.
117    pub fn data_feed_id(mut self, data_feed_id: B256) -> Self {
118        self.data_feed_id = Some(data_feed_id);
119        self
120    }
121
122    /// Set a stable feed id.
123    pub fn id(mut self, id: impl Into<String>) -> Self {
124        self.feed_id = Some(FeedId::new(id));
125        self
126    }
127
128    /// Set a stable feed id.
129    pub fn feed_id(mut self, id: FeedId) -> Self {
130        self.feed_id = Some(id);
131        self
132    }
133
134    /// Set a human-readable label.
135    pub fn label(mut self, label: impl Into<String>) -> Self {
136        self.label = Some(label.into());
137        self
138    }
139
140    /// Set the base asset label.
141    pub fn base(mut self, base: AssetId) -> Self {
142        self.base = Some(base);
143        self
144    }
145
146    /// Set the quote denomination label.
147    pub fn quote(mut self, quote: Denomination) -> Self {
148        self.quote = Some(quote);
149        self
150    }
151
152    /// Set a max-age staleness policy.
153    pub fn max_age_secs(mut self, max_age_secs: u64) -> Self {
154        self.staleness = StalenessPolicy::max_age(max_age_secs);
155        self
156    }
157
158    /// Set the full staleness policy.
159    pub fn staleness(mut self, staleness: StalenessPolicy) -> Self {
160        self.staleness = staleness;
161        self
162    }
163
164    fn feed_for_skip(&self) -> Feed {
165        let mut feed = Feed::proxy(self.price_feed);
166        if let Some(id) = self.feed_id.clone() {
167            feed = feed.feed_id(id);
168        }
169        if let Some(label) = &self.label {
170            feed = feed.label(label.clone());
171        }
172        if let Some(base) = &self.base {
173            feed = feed.base(base.clone());
174        }
175        if let Some(quote) = &self.quote {
176            feed = feed.quote(quote.clone());
177        }
178        feed.staleness(self.staleness)
179    }
180
181    fn config(&self) -> FeedConfig {
182        FeedConfig {
183            proxy: self.price_feed,
184            id: self.feed_id.clone(),
185            label: self.label.clone(),
186            base: self.base.clone().map(String::from),
187            quote: self.quote.clone().map(String::from),
188            staleness: self.staleness,
189        }
190    }
191}
192
193/// Cache-backed RedStone push oracle discovery adapter.
194#[derive(Clone, Debug, Default)]
195pub struct RedstoneOracleAdapter {
196    feeds: Vec<RedstoneFeed>,
197    now_timestamp: Option<u64>,
198}
199
200impl RedstoneOracleAdapter {
201    /// Create an empty RedStone adapter.
202    pub fn new() -> Self {
203        Self::default()
204    }
205
206    /// Create an adapter for one RedStone price-feed wrapper.
207    pub fn price_feed(price_feed: Address) -> Self {
208        Self::new().feed(RedstoneFeed::new(price_feed))
209    }
210
211    /// Add one RedStone feed.
212    pub fn feed(mut self, feed: RedstoneFeed) -> Self {
213        self.feeds.push(feed);
214        self
215    }
216
217    /// Add multiple RedStone feeds.
218    pub fn feeds(mut self, feeds: impl IntoIterator<Item = RedstoneFeed>) -> Self {
219        self.feeds.extend(feeds);
220        self
221    }
222
223    /// Set a fixed timestamp for deterministic registration status classification.
224    pub fn now_timestamp(mut self, now_timestamp: u64) -> Self {
225        self.now_timestamp = Some(now_timestamp);
226        self
227    }
228
229    fn timestamp(&self, fallback: Option<u64>) -> Result<u64, OracleError> {
230        if let Some(now_timestamp) = self.now_timestamp.or(fallback) {
231            return Ok(now_timestamp);
232        }
233        Ok(SystemTime::now()
234            .duration_since(UNIX_EPOCH)
235            .map_err(crate::error::clock_error)?
236            .as_secs())
237    }
238
239    fn discover_feeds(
240        &self,
241        cache: &mut EvmCache,
242        _now_timestamp: u64,
243    ) -> Result<OracleDiscoveryReport, OracleError> {
244        let mut report = OracleDiscoveryReport::new();
245        for feed in &self.feeds {
246            match self.discover_feed(cache, feed) {
247                Ok(discovered) => report = report.with_feed(discovered),
248                Err(error) => {
249                    report = report.with_skip(OracleAdapterFeedSkip {
250                        feed: feed.feed_for_skip(),
251                        proxy: feed.price_feed,
252                        reason: OracleAdapterSkipReason::UnsupportedRedstoneSource {
253                            error: error.to_string(),
254                        },
255                    });
256                }
257            }
258        }
259        Ok(report)
260    }
261
262    fn discover_feed(
263        &self,
264        cache: &mut EvmCache,
265        feed: &RedstoneFeed,
266    ) -> Result<OracleDiscoveredFeed, OracleError> {
267        let data_feed_id = match feed.data_feed_id {
268            Some(data_feed_id) => data_feed_id,
269            None => cache
270                .call_sol(feed.price_feed, getDataFeedIdCall {})
271                .map_err(provider_error)?,
272        };
273        let adapter = match feed.adapter {
274            Some(adapter) => adapter,
275            None => cache
276                .call_sol(feed.price_feed, getPriceFeedAdapterCall {})
277                .map_err(provider_error)?,
278        };
279
280        let reader = EvmCacheChainlinkReader::new(cache);
281        let metadata = FeedMetadata {
282            decimals: reader.read_decimals(feed.price_feed)?,
283            description: reader.read_description(feed.price_feed)?,
284            version: reader.read_version(feed.price_feed)?,
285        };
286        let round = reader.read_latest_round_data(feed.price_feed)?;
287        let config = feed.config();
288        let id = config
289            .id
290            .unwrap_or_else(|| derive_redstone_feed_id(config.label.as_deref(), feed.price_feed));
291        let registration = FeedRegistration {
292            id,
293            proxy: config.proxy,
294            label: config.label,
295            base: config.base,
296            quote: config.quote,
297            staleness: config.staleness,
298            current_aggregator: Some(adapter),
299            aggregator_layout: None,
300            metadata,
301            source: FeedSource::redstone_push(feed.price_feed, adapter, data_feed_id),
302            status: OracleFeedStatus::Ready,
303        };
304
305        Ok(OracleDiscoveredFeed::new(registration, round))
306    }
307}
308
309impl OracleAdapterPlugin for RedstoneOracleAdapter {
310    fn adapter_id(&self) -> OracleAdapterId {
311        OracleAdapterId::new(ADAPTER_ID)
312    }
313
314    fn discover<'a>(
315        &'a self,
316        ctx: OracleDiscoveryContext<'a>,
317    ) -> AdapterFuture<'a, OracleDiscoveryReport> {
318        Box::pin(async move {
319            let now_timestamp = self.timestamp(Some(ctx.now_timestamp))?;
320            self.discover_feeds(ctx.cache, now_timestamp)
321        })
322    }
323
324    fn reactive_handler(
325        &self,
326        registrations: Vec<FeedRegistration>,
327        _storage_sync: OracleStorageSync,
328    ) -> Arc<dyn ReactiveHandler<Ethereum>> {
329        Arc::new(RedstoneReactiveHandler::new(registrations))
330    }
331}
332
333/// Storage helpers for RedStone `MultiFeedAdapterWithoutRounds` push feeds.
334///
335/// The live RedStone multi-feed adapter stores each feed in a packed
336/// `DataFeedDetails` word keyed by `dataFeedId`. A direct event write is only
337/// emitted when that packed slot is already hot in the cache. If the value does
338/// not fit the inline `uint152` field, the overflow slot must also be hot.
339#[derive(Clone, Debug, Default)]
340pub struct RedstoneMultiFeedStorageAdapter;
341
342impl RedstoneMultiFeedStorageAdapter {
343    /// Return the storage slot for `dataFeeds[dataFeedId]`.
344    pub fn data_feed_details_slot(data_feed_id: B256) -> U256 {
345        keyed_slot(
346            data_feed_id,
347            REDSTONE_MULTI_FEED_DATA_FEEDS_STORAGE_LOCATION,
348        )
349    }
350
351    /// Return the overflow slot for `dataFeeds[dataFeedId].biggerValue`.
352    pub fn bigger_value_slot(data_feed_id: B256) -> U256 {
353        Self::data_feed_details_slot(data_feed_id) + U256::from(1_u8)
354    }
355
356    /// Pack RedStone `DataFeedDetails`.
357    ///
358    /// RedStone's `ValueUpdate` event exposes one timestamp. The storage layout
359    /// keeps both data-package and block timestamps, so the event timestamp is
360    /// used for the block timestamp and `timestamp * 1000` is used for the
361    /// millisecond data timestamp.
362    pub fn pack_data_feed_details_from_event(value: U256, updated_at: u64) -> Option<U256> {
363        let data_timestamp_ms = updated_at.checked_mul(1_000)?;
364        Self::pack_data_feed_details(value, data_timestamp_ms, updated_at)
365    }
366
367    /// Pack RedStone `DataFeedDetails` with explicit timestamps.
368    pub fn pack_data_feed_details(
369        value: U256,
370        data_timestamp_ms: u64,
371        block_timestamp: u64,
372    ) -> Option<U256> {
373        if data_timestamp_ms > uint_mask_u64(REDSTONE_DATA_TIMESTAMP_BITS)
374            || block_timestamp > uint_mask_u64(REDSTONE_BLOCK_TIMESTAMP_BITS)
375        {
376            return None;
377        }
378
379        let inline_value = value & uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS);
380        let is_value_bigger = if value > uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS) {
381            U256::from(1_u8)
382        } else {
383            U256::ZERO
384        };
385
386        Some(
387            U256::from(data_timestamp_ms)
388                | (U256::from(block_timestamp) << REDSTONE_MULTI_FEED_BLOCK_TIMESTAMP_OFFSET_BITS)
389                | (inline_value << REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS)
390                | (is_value_bigger << REDSTONE_MULTI_FEED_IS_VALUE_BIGGER_OFFSET_BITS),
391        )
392    }
393
394    fn state_updates_for_value_update(
395        adapter: Address,
396        event: &RedstoneValueUpdate,
397        state: &dyn StateView,
398    ) -> Option<Vec<StateUpdate>> {
399        let details_slot = Self::data_feed_details_slot(event.data_feed_id);
400        state.storage(adapter, details_slot)?;
401        let details = Self::pack_data_feed_details_from_event(event.value, event.updated_at)?;
402        let mut updates = vec![StateUpdate::slot(adapter, details_slot, details)];
403
404        if event.value > uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS) {
405            let bigger_value_slot = Self::bigger_value_slot(event.data_feed_id);
406            state.storage(adapter, bigger_value_slot)?;
407            updates.push(StateUpdate::slot(adapter, bigger_value_slot, event.value));
408        }
409
410        Some(updates)
411    }
412
413    fn state_updates_for_answer(
414        adapter: Address,
415        data_feed_id: B256,
416        value: U256,
417        updated_at: u64,
418        state: &dyn StateView,
419    ) -> Option<Vec<StateUpdate>> {
420        let event = RedstoneValueUpdate {
421            adapter,
422            data_feed_id,
423            value,
424            updated_at,
425            block_number: None,
426            log_index: None,
427            removed: false,
428        };
429        Self::state_updates_for_value_update(adapter, &event, state)
430    }
431}
432
433/// Storage helpers for RedStone `PriceFeedsAdapter*` push-feed layouts.
434///
435/// These layouts are less common for the current multi-feed deployments but are
436/// still useful for older RedStone adapters and merged price-feed wrappers.
437#[derive(Clone, Debug, Default)]
438pub struct RedstonePriceFeedsStorageAdapter;
439
440impl RedstonePriceFeedsStorageAdapter {
441    /// Return the storage slot for the no-rounds `values[dataFeedId]` mapping.
442    pub fn value_slot(data_feed_id: B256) -> U256 {
443        keyed_slot(
444            data_feed_id,
445            REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION,
446        )
447    }
448
449    /// Return the storage slot for a with-rounds `values[dataFeedId][roundId]` value.
450    pub fn round_value_slot(data_feed_id: B256, round_id: U256) -> U256 {
451        keyed_slot2(
452            data_feed_id,
453            round_id,
454            REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION,
455        )
456    }
457
458    /// Return the packed latest-update timestamps slot for no-rounds adapters.
459    pub fn latest_update_timestamps_slot() -> U256 {
460        U256::from_be_slice(
461            REDSTONE_PRICE_FEEDS_LATEST_UPDATE_TIMESTAMPS_STORAGE_LOCATION.as_slice(),
462        )
463    }
464
465    /// Return the slot storing the latest round id for with-rounds adapters.
466    pub fn latest_round_id_slot() -> U256 {
467        U256::from_be_slice(
468            REDSTONE_PRICE_FEEDS_WITH_ROUNDS_LATEST_ROUND_ID_STORAGE_LOCATION.as_slice(),
469        )
470    }
471
472    /// Return the with-rounds timestamp slot for `roundId`.
473    pub fn round_timestamp_slot(round_id: U256) -> U256 {
474        mapping_slot(
475            round_id,
476            U256::from_be_slice(
477                REDSTONE_PRICE_FEEDS_WITH_ROUNDS_ROUND_TIMESTAMPS_MAPPING_STORAGE_LOCATION
478                    .as_slice(),
479            ),
480        )
481    }
482
483    /// Pack no-rounds latest update timestamps.
484    pub fn pack_latest_update_timestamps_from_event(updated_at: u64) -> Option<U256> {
485        let data_timestamp_ms = updated_at.checked_mul(1_000)?;
486        Self::pack_latest_update_timestamps(data_timestamp_ms, updated_at)
487    }
488
489    /// Pack no-rounds latest update timestamps with explicit timestamps.
490    ///
491    /// Always returns `Some`: both packed fields are 128 bits wide, so every
492    /// `u64` input fits. The `Option` return is kept for signature stability
493    /// with the other packing helpers.
494    pub fn pack_latest_update_timestamps(
495        data_timestamp_ms: u64,
496        block_timestamp: u64,
497    ) -> Option<U256> {
498        Some(
499            (U256::from(data_timestamp_ms) << REDSTONE_PRICE_FEEDS_BLOCK_TIMESTAMP_OFFSET_BITS)
500                | U256::from(block_timestamp),
501        )
502    }
503
504    fn state_updates_for_value_update(
505        adapter: Address,
506        event: &RedstoneValueUpdate,
507        state: &dyn StateView,
508    ) -> Option<Vec<StateUpdate>> {
509        let value_slot = Self::value_slot(event.data_feed_id);
510        let timestamp_slot = Self::latest_update_timestamps_slot();
511        state.storage(adapter, value_slot)?;
512        state.storage(adapter, timestamp_slot)?;
513        let timestamps = Self::pack_latest_update_timestamps_from_event(event.updated_at)?;
514
515        Some(vec![
516            StateUpdate::slot(adapter, value_slot, event.value),
517            StateUpdate::slot(adapter, timestamp_slot, timestamps),
518        ])
519    }
520
521    fn state_updates_for_no_rounds_answer(
522        adapter: Address,
523        data_feed_id: B256,
524        value: U256,
525        updated_at: u64,
526        state: &dyn StateView,
527    ) -> Option<Vec<StateUpdate>> {
528        let event = RedstoneValueUpdate {
529            adapter,
530            data_feed_id,
531            value,
532            updated_at,
533            block_number: None,
534            log_index: None,
535            removed: false,
536        };
537        Self::state_updates_for_value_update(adapter, &event, state)
538    }
539
540    fn state_updates_for_round_answer(
541        adapter: Address,
542        data_feed_id: B256,
543        round_id: U256,
544        value: U256,
545        updated_at: u64,
546        state: &dyn StateView,
547    ) -> Option<Vec<StateUpdate>> {
548        let value_slot = Self::round_value_slot(data_feed_id, round_id);
549        let round_timestamp_slot = Self::round_timestamp_slot(round_id);
550        let latest_round_id_slot = Self::latest_round_id_slot();
551        state.storage(adapter, value_slot)?;
552        state.storage(adapter, round_timestamp_slot)?;
553        state.storage(adapter, latest_round_id_slot)?;
554
555        Some(vec![
556            StateUpdate::slot(adapter, value_slot, value),
557            StateUpdate::slot(adapter, round_timestamp_slot, U256::from(updated_at)),
558            StateUpdate::slot(adapter, latest_round_id_slot, round_id),
559        ])
560    }
561}
562
563#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
564struct RedstoneValueUpdateKey {
565    adapter: Address,
566    data_feed_id: B256,
567}
568
569/// Reactive-runtime bridge for RedStone push oracle feeds.
570#[derive(Clone, Debug)]
571pub struct RedstoneReactiveHandler {
572    registrations_by_value_update: BTreeMap<RedstoneValueUpdateKey, Vec<FeedRegistration>>,
573    registrations_by_answer_updated: BTreeMap<Address, Vec<FeedRegistration>>,
574}
575
576impl RedstoneReactiveHandler {
577    /// Create a handler from current RedStone feed registrations.
578    pub fn new(registrations: Vec<FeedRegistration>) -> Self {
579        let mut registrations_by_value_update: BTreeMap<
580            RedstoneValueUpdateKey,
581            Vec<FeedRegistration>,
582        > = BTreeMap::new();
583        let mut registrations_by_answer_updated: BTreeMap<Address, Vec<FeedRegistration>> =
584            BTreeMap::new();
585        let mut seen_value_updates = BTreeSet::new();
586        let mut seen_answer_updates = BTreeSet::new();
587
588        for registration in registrations {
589            let FeedSource::RedstonePush {
590                price_feed,
591                adapter,
592                data_feed_id,
593            } = registration.source
594            else {
595                continue;
596            };
597
598            let value_key = RedstoneValueUpdateKey {
599                adapter,
600                data_feed_id,
601            };
602            if seen_value_updates.insert((registration.id.clone(), value_key)) {
603                registrations_by_value_update
604                    .entry(value_key)
605                    .or_default()
606                    .push(registration.clone());
607            }
608
609            if seen_answer_updates.insert((registration.id.clone(), price_feed)) {
610                registrations_by_answer_updated
611                    .entry(price_feed)
612                    .or_default()
613                    .push(registration);
614            }
615        }
616
617        Self {
618            registrations_by_value_update,
619            registrations_by_answer_updated,
620        }
621    }
622
623    /// Stable handler id.
624    pub fn id(&self) -> HandlerId {
625        HandlerId::new(HANDLER_ID)
626    }
627
628    /// Interests for RedStone adapter and merged-feed events.
629    pub fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
630        let mut interests = Vec::new();
631        let mut value_adapters = BTreeSet::new();
632        for key in self.registrations_by_value_update.keys() {
633            if value_adapters.insert(key.adapter) {
634                interests.push(log_interest(key.adapter, REDSTONE_VALUE_UPDATE_TOPIC));
635            }
636        }
637        interests.extend(
638            self.registrations_by_answer_updated
639                .keys()
640                .copied()
641                .map(|price_feed| log_interest(price_feed, ANSWER_UPDATED_TOPIC)),
642        );
643        interests
644    }
645
646    fn handle_value_update(
647        &self,
648        ctx: &ReactiveContext,
649        log: &alloy_rpc_types_eth::Log,
650        state: &dyn StateView,
651    ) -> Result<HandlerOutcome, HandlerError> {
652        let event = decode_redstone_value_update(log).map_err(|error| {
653            HandlerError::new(format!("decode RedStone ValueUpdate failed: {error}"))
654        })?;
655        let key = RedstoneValueUpdateKey {
656            adapter: event.adapter,
657            data_feed_id: event.data_feed_id,
658        };
659        let Some(registrations) = self.registrations_by_value_update.get(&key) else {
660            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
661        };
662
663        let answer = redstone_value_to_i256(event.value)?;
664        Ok(self.outcome_for_registrations(
665            ctx,
666            log,
667            registrations,
668            event.adapter,
669            answer,
670            U256::from(REDSTONE_NO_ROUNDS_ROUND_ID),
671            event.updated_at,
672            event.block_number,
673            event.log_index,
674            event.removed,
675            Some(event),
676            state,
677        ))
678    }
679
680    fn handle_answer_updated(
681        &self,
682        ctx: &ReactiveContext,
683        log: &alloy_rpc_types_eth::Log,
684        state: &dyn StateView,
685    ) -> Result<HandlerOutcome, HandlerError> {
686        let price_feed = log.address();
687        let Some(registrations) = self.registrations_by_answer_updated.get(&price_feed) else {
688            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
689        };
690        let event = decode_answer_updated(log).map_err(|error| {
691            HandlerError::new(format!("decode RedStone AnswerUpdated failed: {error}"))
692        })?;
693
694        Ok(self.outcome_for_registrations(
695            ctx,
696            log,
697            registrations,
698            price_feed,
699            event.current,
700            event.round_id,
701            event.updated_at,
702            event.block_number,
703            event.log_index,
704            event.removed,
705            None,
706            state,
707        ))
708    }
709
710    #[allow(clippy::too_many_arguments)]
711    fn outcome_for_registrations(
712        &self,
713        ctx: &ReactiveContext,
714        log: &alloy_rpc_types_eth::Log,
715        registrations: &[FeedRegistration],
716        event_source: Address,
717        raw_answer: I256,
718        event_round_id: U256,
719        updated_at: u64,
720        block_number: Option<u64>,
721        log_index: Option<u64>,
722        removed: bool,
723        value_update: Option<RedstoneValueUpdate>,
724        state: &dyn StateView,
725    ) -> HandlerOutcome {
726        let block_number = block_number.or_else(|| ctx.block.as_ref().map(|block| block.number));
727        let block_hash = log
728            .block_hash
729            .or_else(|| ctx.block.as_ref().map(|block| block.hash));
730        let log_index = log_index.or(ctx.log_index);
731        let value_status = if removed {
732            OracleValueStatus::RequiresRepair
733        } else {
734            OracleValueStatus::EventPending
735        };
736        let mut effects = Vec::new();
737        let mut tags = Vec::new();
738        let direct_updates = if removed {
739            None
740        } else {
741            registrations.first().and_then(|registration| {
742                redstone_state_updates(
743                    registration,
744                    event_source,
745                    raw_answer,
746                    event_round_id,
747                    updated_at,
748                    value_update.as_ref(),
749                    state,
750                )
751            })
752        };
753        let has_direct_updates = direct_updates.is_some();
754        if let Some(updates) = direct_updates {
755            effects.extend(updates.into_iter().map(ReactiveEffect::StateUpdate));
756        }
757
758        for registration in registrations {
759            if !has_direct_updates {
760                append_redstone_invalidations(&mut effects, registration, event_source);
761            }
762            let hook_tags = redstone_hook_tags(registration, event_source, value_update.as_ref());
763            tags.extend(hook_tags.clone());
764            let normalized_answer = registration
765                .source
766                .normalize_answer_from_event(Some(event_source), raw_answer);
767            let round = RoundData {
768                round_id: event_round_id,
769                answer: normalized_answer,
770                started_at: updated_at,
771                updated_at,
772                answered_in_round: event_round_id,
773            };
774            let round_status = classify_round(&round, updated_at, &registration.staleness);
775            effects.push(ReactiveEffect::Hook(HookSignal {
776                namespace: Cow::Borrowed(ORACLE_SIGNAL_NAMESPACE),
777                kind: Cow::Borrowed(OracleSignalKind::PriceUpdate.as_str()),
778                labels: hook_tags,
779                payload: Some(Arc::new(OraclePriceUpdate {
780                    id: registration.id.clone(),
781                    proxy: registration.proxy,
782                    aggregator: event_source,
783                    label: registration.label.clone(),
784                    base: registration.base.clone(),
785                    quote: registration.quote.clone(),
786                    raw_answer: normalized_answer,
787                    decimals: registration.metadata.decimals,
788                    event_round_id,
789                    started_at: updated_at,
790                    updated_at,
791                    block_number,
792                    block_hash,
793                    log_index,
794                    round_status,
795                    value_status,
796                    source: OracleValueSource::Event,
797                })),
798            }));
799        }
800
801        HandlerOutcome {
802            effects,
803            quality: if has_direct_updates {
804                StateEffectQuality::ExactFromInput
805            } else {
806                StateEffectQuality::RequiresRepair
807            },
808            tags,
809        }
810    }
811}
812
813impl ReactiveHandler<Ethereum> for RedstoneReactiveHandler {
814    fn id(&self) -> HandlerId {
815        self.id()
816    }
817
818    fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
819        self.interests()
820    }
821
822    fn handle(
823        &self,
824        ctx: &ReactiveContext,
825        input: &ReactiveInput<Ethereum>,
826        state: &dyn StateView,
827    ) -> Result<HandlerOutcome, HandlerError> {
828        let ReactiveInput::Log(log) = input else {
829            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
830        };
831        if log.removed && !matches!(ctx.chain_status, ChainStatus::Reorged { .. }) {
832            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
833        }
834
835        match log.topics().first().copied() {
836            Some(REDSTONE_VALUE_UPDATE_TOPIC) => self.handle_value_update(ctx, log, state),
837            Some(ANSWER_UPDATED_TOPIC) => self.handle_answer_updated(ctx, log, state),
838            _ => Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)),
839        }
840    }
841}
842
843fn log_interest(address: Address, topic: B256) -> ReactiveInterest<Ethereum> {
844    ReactiveInterest::Logs(LogInterest {
845        provider_filter: Filter::new().address(address).event_signature(topic),
846        local_matcher: None,
847        route_key: Some(RouteKeySpec::EmitterAddress),
848    })
849}
850
851fn append_redstone_invalidations(
852    effects: &mut Vec<ReactiveEffect>,
853    registration: &FeedRegistration,
854    event_source: Address,
855) {
856    let mut addresses = BTreeSet::from([registration.proxy, event_source]);
857    if let FeedSource::RedstonePush { adapter, .. } = registration.source {
858        addresses.insert(adapter);
859    }
860    effects.extend(addresses.into_iter().map(|address| {
861        ReactiveEffect::Invalidate(InvalidationRequest {
862            scope: PurgeScope::AllStorage,
863            address,
864            reason: InvalidationReason::HandlerRequested,
865        })
866    }));
867}
868
869fn redstone_hook_tags(
870    registration: &FeedRegistration,
871    event_source: Address,
872    value_update: Option<&RedstoneValueUpdate>,
873) -> Vec<ReportTag> {
874    let mut tags = vec![
875        ReportTag::new("feed_id", registration.id.to_string()),
876        ReportTag::new("proxy", format!("{:?}", registration.proxy)),
877        ReportTag::new("aggregator", format!("{event_source:?}")),
878    ];
879    if let Some(value_update) = value_update {
880        tags.push(ReportTag::new(
881            "data_feed_id",
882            format!("{:?}", value_update.data_feed_id),
883        ));
884    }
885    tags
886}
887
888fn redstone_value_to_i256(value: U256) -> Result<I256, HandlerError> {
889    I256::try_from(value)
890        .map_err(|_| HandlerError::new(format!("RedStone value {value} does not fit int256")))
891}
892
893#[allow(clippy::too_many_arguments)]
894fn redstone_state_updates(
895    registration: &FeedRegistration,
896    event_source: Address,
897    raw_answer: I256,
898    event_round_id: U256,
899    updated_at: u64,
900    value_update: Option<&RedstoneValueUpdate>,
901    state: &dyn StateView,
902) -> Option<Vec<StateUpdate>> {
903    match value_update {
904        Some(event) => redstone_state_updates_for_value_update(registration, event, state),
905        None => redstone_state_updates_for_answer_updated(
906            registration,
907            event_source,
908            raw_answer,
909            event_round_id,
910            updated_at,
911            state,
912        ),
913    }
914}
915
916fn redstone_state_updates_for_value_update(
917    registration: &FeedRegistration,
918    event: &RedstoneValueUpdate,
919    state: &dyn StateView,
920) -> Option<Vec<StateUpdate>> {
921    let FeedSource::RedstonePush {
922        adapter,
923        data_feed_id,
924        ..
925    } = registration.source
926    else {
927        return None;
928    };
929    if event.adapter != adapter || event.data_feed_id != data_feed_id {
930        return None;
931    }
932
933    RedstoneMultiFeedStorageAdapter::state_updates_for_value_update(adapter, event, state).or_else(
934        || RedstonePriceFeedsStorageAdapter::state_updates_for_value_update(adapter, event, state),
935    )
936}
937
938fn redstone_state_updates_for_answer_updated(
939    registration: &FeedRegistration,
940    event_source: Address,
941    raw_answer: I256,
942    event_round_id: U256,
943    updated_at: u64,
944    state: &dyn StateView,
945) -> Option<Vec<StateUpdate>> {
946    let FeedSource::RedstonePush {
947        price_feed,
948        adapter,
949        data_feed_id,
950    } = registration.source
951    else {
952        return None;
953    };
954    if event_source != price_feed || adapter != price_feed {
955        return None;
956    }
957    let value = redstone_i256_to_u256(raw_answer)?;
958
959    if event_round_id == U256::from(REDSTONE_NO_ROUNDS_ROUND_ID) {
960        RedstoneMultiFeedStorageAdapter::state_updates_for_answer(
961            adapter,
962            data_feed_id,
963            value,
964            updated_at,
965            state,
966        )
967        .or_else(|| {
968            RedstonePriceFeedsStorageAdapter::state_updates_for_no_rounds_answer(
969                adapter,
970                data_feed_id,
971                value,
972                updated_at,
973                state,
974            )
975        })
976    } else {
977        RedstonePriceFeedsStorageAdapter::state_updates_for_round_answer(
978            adapter,
979            data_feed_id,
980            event_round_id,
981            value,
982            updated_at,
983            state,
984        )
985    }
986}
987
988fn redstone_i256_to_u256(value: I256) -> Option<U256> {
989    let raw = value.into_raw();
990    if raw >> 255 == U256::ZERO {
991        Some(raw)
992    } else {
993        None
994    }
995}
996
997fn keyed_slot(key: B256, base_slot: B256) -> U256 {
998    let mut preimage = [0_u8; 64];
999    preimage[..32].copy_from_slice(key.as_slice());
1000    preimage[32..].copy_from_slice(base_slot.as_slice());
1001    U256::from_be_slice(keccak256(preimage).as_slice())
1002}
1003
1004fn keyed_slot2(key0: B256, key1: U256, base_slot: B256) -> U256 {
1005    let mut preimage = [0_u8; 96];
1006    preimage[..32].copy_from_slice(key0.as_slice());
1007    preimage[32..64].copy_from_slice(&key1.to_be_bytes::<32>());
1008    preimage[64..].copy_from_slice(base_slot.as_slice());
1009    U256::from_be_slice(keccak256(preimage).as_slice())
1010}
1011
1012fn mapping_slot(key: U256, base_slot: U256) -> U256 {
1013    let mut preimage = [0_u8; 64];
1014    preimage[..32].copy_from_slice(&key.to_be_bytes::<32>());
1015    preimage[32..].copy_from_slice(&base_slot.to_be_bytes::<32>());
1016    U256::from_be_slice(keccak256(preimage).as_slice())
1017}
1018
1019fn uint_mask(bits: usize) -> U256 {
1020    debug_assert!(bits <= 256);
1021    match bits {
1022        0 => U256::ZERO,
1023        256 => U256::MAX,
1024        bits => (U256::from(1_u8) << bits) - U256::from(1_u8),
1025    }
1026}
1027
1028fn uint_mask_u64(bits: usize) -> u64 {
1029    // Callers mask sub-word fields only; a full-width (or wider) mask would
1030    // make range guards like `x > uint_mask_u64(bits)` silently inoperative.
1031    debug_assert!(bits < 64);
1032    (1_u64 << bits) - 1
1033}
1034
1035fn derive_redstone_feed_id(label: Option<&str>, price_feed: Address) -> FeedId {
1036    if let Some(label) = label {
1037        let normalized = label
1038            .chars()
1039            .filter_map(|ch| {
1040                if ch.is_ascii_alphanumeric() {
1041                    Some(ch.to_ascii_lowercase())
1042                } else if ch.is_ascii_whitespace() || matches!(ch, '/' | '_' | '-') {
1043                    Some('-')
1044                } else {
1045                    None
1046                }
1047            })
1048            .collect::<String>()
1049            .trim_matches('-')
1050            .to_string();
1051        if !normalized.is_empty() {
1052            return FeedId::new(normalized);
1053        }
1054    }
1055    FeedId::new(format!("redstone-{price_feed:?}"))
1056}
1057
1058fn provider_error(error: impl std::fmt::Debug) -> OracleError {
1059    OracleError::Provider(format!("{error:?}"))
1060}