// file: crates/ksp-offchain-transport-lib/src/market_price_service.rs // version: 3 //! Generic market-price refresh service owning provider dispatch and availability transitions. const MARKET_PRICE_RATE_LIMIT_FALLBACK_MILLIS: u64 = 1_000; const MARKET_PRICE_REFRESH_MAX_PROVIDERS: usize = 64; /// Provider-specific runtime setup consumed once by [`crate::MarketPriceService`]. /// /// This enum is intended for composition layers such as Config. Runtime consumers use the generic service methods and never need to branch on provider kinds. #[non_exhaustive] pub enum MarketPriceProviderSetup { /// Birdeye Standard setup. Birdeye(crate::MarketPriceBirdeyeSettings), /// Coinbase Exchange setup. CoinbaseExchange(crate::MarketPriceCoinbaseExchangeSettings), /// CoinGecko setup. CoinGecko(crate::MarketPriceCoinGeckoSettings), /// CoinMarketCap setup. CoinMarketCap(crate::MarketPriceCoinMarketCapSettings), /// CoinPaprika setup. CoinPaprika(crate::MarketPriceCoinPaprikaSettings), /// DexScreener setup bound to one explicit Solana pair. DexScreener(crate::MarketPriceDexScreenerSettings), /// Jupiter Price V3 setup. Jupiter(crate::MarketPriceJupiterSettings), /// Kraken Spot setup. Kraken(crate::MarketPriceKrakenSettings), } /// Generic result of one explicit market-price refresh attempt or eligibility projection. #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] pub struct MarketPriceRefreshOutcome { observation: std::option::Option, state: crate::MarketPriceProviderState, } impl crate::MarketPriceRefreshOutcome { /// Returns the successful normalized observation when this refresh produced one. #[must_use] pub fn observation(&self) -> std::option::Option<&crate::MarketPriceObservation> { return self.observation.as_ref(); } /// Returns the provider identifier without exposing provider-specific runtime types. #[must_use] pub const fn provider_id(&self) -> &crate::MarketPriceProviderId { return self.state.provider_id(); } /// Reports whether this refresh produced a new observation. #[must_use] pub fn refreshed(&self) -> bool { return self.observation.is_some(); } /// Returns the resulting provider-neutral runtime state. #[must_use] pub const fn state(&self) -> &crate::MarketPriceProviderState { return &self.state; } fn from_state(state: crate::MarketPriceProviderState) -> Self { return Self { observation: std::option::Option::None, state }; } fn from_observation(observation: crate::MarketPriceObservation, state: crate::MarketPriceProviderState) -> Self { return Self { observation: std::option::Option::Some(observation), state }; } } /// Provider-agnostic SOL/USD refresh service owning all configured provider adapters. pub struct MarketPriceService { providers: std::vec::Vec, registry: std::sync::Mutex, } impl crate::MarketPriceService { /// Builds the service from provider-specific setup supplied by a composition layer. /// /// Provider identifiers must be unique. Initial registry state is `Ready` for enabled providers and `Disabled` for disabled providers. pub fn new(setups: std::vec::Vec) -> ksp_core_lib::Result { if setups.len() > MARKET_PRICE_REFRESH_MAX_PROVIDERS { return std::result::Result::Err(refresh_error("provider_count")); } let mut providers = std::vec::Vec::with_capacity(setups.len()); let mut entries = std::vec::Vec::with_capacity(setups.len()); for setup in setups { let runtime = match MarketPriceProviderRuntime::new(setup) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let availability = if runtime.enabled() { crate::MarketPriceProviderAvailability::Ready } else { crate::MarketPriceProviderAvailability::Disabled }; entries.push(crate::MarketPriceProviderRegistryEntry::new(runtime.descriptor().clone(), availability)); providers.push(runtime); } let registry = match crate::MarketPriceProviderRegistry::new(entries) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; providers.sort_by(|left, right| return left.provider_id().cmp(right.provider_id())); ksp_logging_lib::debug!( target: crate::TRACING_TARGET, provider_count = providers.len(), "created generic market-price refresh service" ); return std::result::Result::Ok(Self { providers, registry: std::sync::Mutex::new(registry) }); } /// Returns a detached provider-neutral registry snapshot suitable for HID projection. #[must_use] pub fn registry(&self) -> crate::MarketPriceProviderRegistry { let guard = lock_registry(&self.registry); return guard.clone(); } /// Refreshes one provider by generic provider identifier. /// /// Non-eligible states are returned without network dispatch. Provider transport/application failures are normalized into availability and returned as an /// outcome rather than forcing consumers to parse provider-specific errors. pub async fn refresh(&self, provider_id: &crate::MarketPriceProviderId) -> ksp_core_lib::Result { let index = match self.provider_index(provider_id) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(provider_not_found_error(provider_id)), }; let now = current_timestamp_or_zero(); let state = match self.state_snapshot(provider_id) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(provider_not_found_error(provider_id)), }; if !state.availability().is_refresh_eligible_at(now) { return std::result::Result::Ok(crate::MarketPriceRefreshOutcome::from_state(state)); } let result = self.providers[index].fetch_sol_usd().await; return match result { std::result::Result::Ok(observation) => { let state = self.update_availability(provider_id, crate::MarketPriceProviderAvailability::Ready); std::result::Result::Ok(crate::MarketPriceRefreshOutcome::from_observation(observation, state)) }, std::result::Result::Err(error) => { let availability = availability_from_error(&error, self.providers[index].descriptor().auth_mode()); ksp_logging_lib::debug!( target: crate::TRACING_TARGET, provider_id = provider_id.as_str(), error_domain = error.code().domain(), error_code = error.code().code(), "classified market-price refresh failure" ); let state = self.update_availability(provider_id, availability); std::result::Result::Ok(crate::MarketPriceRefreshOutcome::from_state(state)) }, }; } /// Refreshes a caller-selected provider set in deterministic request order without fallback or consensus. /// /// The service never sleeps to wait for a local rate limit. Duplicate or unknown identifiers are rejected before any provider request is attempted. pub async fn refresh_many(&self, provider_ids: &[crate::MarketPriceProviderId]) -> ksp_core_lib::Result> { if provider_ids.len() > MARKET_PRICE_REFRESH_MAX_PROVIDERS { return std::result::Result::Err(refresh_error("provider_count")); } if let std::result::Result::Err(error) = validate_requested_provider_ids(self, provider_ids) { return std::result::Result::Err(error); } let mut outcomes = std::vec::Vec::with_capacity(provider_ids.len()); for provider_id in provider_ids { let outcome = match self.refresh(provider_id).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; outcomes.push(outcome); } return std::result::Result::Ok(outcomes); } /// Refreshes every currently eligible configured provider in stable provider-id order. /// /// Disabled, authentication-unavailable, misconfigured, quota-unavailable and not-yet-expired cooldown states are projected without network dispatch. pub async fn refresh_all(&self) -> ksp_core_lib::Result> { let provider_ids = self.providers.iter().map(|provider| return provider.provider_id().clone()).collect::>(); return self.refresh_many(provider_ids.as_slice()).await; } fn provider_index(&self, provider_id: &crate::MarketPriceProviderId) -> std::option::Option { let result = self.providers.binary_search_by(|provider| return provider.provider_id().cmp(provider_id)); return match result { std::result::Result::Ok(value) => std::option::Option::Some(value), std::result::Result::Err(_) => std::option::Option::None, }; } fn state_snapshot(&self, provider_id: &crate::MarketPriceProviderId) -> std::option::Option { let guard = lock_registry(&self.registry); return guard.state(provider_id).cloned(); } fn update_availability( &self, provider_id: &crate::MarketPriceProviderId, availability: crate::MarketPriceProviderAvailability, ) -> crate::MarketPriceProviderState { let mut guard = lock_registry(&self.registry); if guard.set_availability(provider_id, availability) && let std::option::Option::Some(state) = guard.state(provider_id) { return state.clone(); } return crate::MarketPriceProviderState::new(provider_id.clone(), availability); } } impl std::fmt::Debug for crate::MarketPriceService { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.debug_struct("MarketPriceService").field("registry", &self.registry()).finish_non_exhaustive(); } } enum MarketPriceProviderRuntime { Birdeye(crate::MarketPriceBirdeyeProvider), CoinbaseExchange(crate::MarketPriceCoinbaseExchangeProvider), CoinGecko(crate::MarketPriceCoinGeckoProvider), CoinMarketCap(crate::MarketPriceCoinMarketCapProvider), CoinPaprika(crate::MarketPriceCoinPaprikaProvider), DexScreener(crate::MarketPriceDexScreenerProvider), Jupiter(crate::MarketPriceJupiterProvider), Kraken(crate::MarketPriceKrakenProvider), } impl MarketPriceProviderRuntime { fn new(setup: crate::MarketPriceProviderSetup) -> ksp_core_lib::Result { return match setup { crate::MarketPriceProviderSetup::Birdeye(settings) => crate::MarketPriceBirdeyeProvider::new(settings).map(Self::Birdeye), crate::MarketPriceProviderSetup::CoinbaseExchange(settings) => { crate::MarketPriceCoinbaseExchangeProvider::new(settings).map(Self::CoinbaseExchange) }, crate::MarketPriceProviderSetup::CoinGecko(settings) => crate::MarketPriceCoinGeckoProvider::new(settings).map(Self::CoinGecko), crate::MarketPriceProviderSetup::CoinMarketCap(settings) => crate::MarketPriceCoinMarketCapProvider::new(settings).map(Self::CoinMarketCap), crate::MarketPriceProviderSetup::CoinPaprika(settings) => crate::MarketPriceCoinPaprikaProvider::new(settings).map(Self::CoinPaprika), crate::MarketPriceProviderSetup::DexScreener(settings) => crate::MarketPriceDexScreenerProvider::new(settings).map(Self::DexScreener), crate::MarketPriceProviderSetup::Jupiter(settings) => crate::MarketPriceJupiterProvider::new(settings).map(Self::Jupiter), crate::MarketPriceProviderSetup::Kraken(settings) => crate::MarketPriceKrakenProvider::new(settings).map(Self::Kraken), }; } fn descriptor(&self) -> &crate::MarketPriceProviderDescriptor { return match self { Self::Birdeye(provider) => provider.descriptor(), Self::CoinbaseExchange(provider) => provider.descriptor(), Self::CoinGecko(provider) => provider.descriptor(), Self::CoinMarketCap(provider) => provider.descriptor(), Self::CoinPaprika(provider) => provider.descriptor(), Self::DexScreener(provider) => provider.descriptor(), Self::Jupiter(provider) => provider.descriptor(), Self::Kraken(provider) => provider.descriptor(), }; } fn enabled(&self) -> bool { return match self { Self::Birdeye(provider) => provider.settings().common().enabled(), Self::CoinbaseExchange(provider) => provider.settings().common().enabled(), Self::CoinGecko(provider) => provider.settings().common().enabled(), Self::CoinMarketCap(provider) => provider.settings().common().enabled(), Self::CoinPaprika(provider) => provider.settings().common().enabled(), Self::DexScreener(provider) => provider.settings().common().enabled(), Self::Jupiter(provider) => provider.settings().common().enabled(), Self::Kraken(provider) => provider.settings().common().enabled(), }; } async fn fetch_sol_usd(&self) -> ksp_core_lib::Result { return match self { Self::Birdeye(provider) => provider.fetch_sol_usd().await, Self::CoinbaseExchange(provider) => provider.fetch_sol_usd().await, Self::CoinGecko(provider) => provider.fetch_sol_usd().await, Self::CoinMarketCap(provider) => provider.fetch_sol_usd().await, Self::CoinPaprika(provider) => provider.fetch_sol_usd().await, Self::DexScreener(provider) => provider.fetch_sol_usd().await, Self::Jupiter(provider) => provider.fetch_sol_usd().await, Self::Kraken(provider) => provider.fetch_sol_usd().await, }; } fn provider_id(&self) -> &crate::MarketPriceProviderId { return self.descriptor().id(); } } fn availability_from_error(error: &ksp_core_lib::Error, auth_mode: crate::MarketPriceProviderAuthMode) -> crate::MarketPriceProviderAvailability { let code = error.code(); if code == crate::ERROR_CODE_MARKET_PRICE_PROVIDER_DISABLED { return crate::MarketPriceProviderAvailability::Disabled; } if code == crate::ERROR_CODE_HTTP_ACCESS_DENIED { return match auth_mode { crate::MarketPriceProviderAuthMode::OptionalApiKey | crate::MarketPriceProviderAuthMode::RequiredApiKey => { crate::MarketPriceProviderAvailability::AuthenticationUnavailable }, crate::MarketPriceProviderAuthMode::None => crate::MarketPriceProviderAvailability::TemporarilyUnavailable { retry_at: std::option::Option::None }, }; } if code == crate::ERROR_CODE_HTTP_ADMISSION_DEFERRED { let delay = context_u64(error, "retry_after_millis").unwrap_or(MARKET_PRICE_RATE_LIMIT_FALLBACK_MILLIS); return crate::MarketPriceProviderAvailability::CoolingDown { retry_at: timestamp_after_millis(delay) }; } if code == crate::ERROR_CODE_HTTP_RATE_LIMITED { let delay = context_u64(error, "retry_after_seconds") .and_then(|seconds| return seconds.checked_mul(1_000)) .unwrap_or(MARKET_PRICE_RATE_LIMIT_FALLBACK_MILLIS); return crate::MarketPriceProviderAvailability::CoolingDown { retry_at: timestamp_after_millis(delay) }; } if code == crate::ERROR_CODE_HTTP_TEMPORARY_FAILURE { let retry_at = context_u64(error, "retry_after_seconds").and_then(|seconds| return seconds.checked_mul(1_000)).map(timestamp_after_millis); return crate::MarketPriceProviderAvailability::TemporarilyUnavailable { retry_at }; } if code == crate::ERROR_CODE_HTTP_CLIENT_BUILD_FAILED || code == crate::ERROR_CODE_HTTP_RATE_LIMIT_INVALID || code == crate::ERROR_CODE_HTTP_REQUEST_INVALID || code == crate::ERROR_CODE_HTTP_SETTINGS_INVALID || code == crate::ERROR_CODE_MARKET_PRICE_PROVIDER_SETTINGS_INVALID { return crate::MarketPriceProviderAvailability::Misconfigured; } return crate::MarketPriceProviderAvailability::TemporarilyUnavailable { retry_at: std::option::Option::None }; } fn context_u64(error: &ksp_core_lib::Error, key: &'static str) -> std::option::Option { for context in error.context() { if context.key() == key { return match context.value().parse::() { std::result::Result::Ok(value) => std::option::Option::Some(value), std::result::Result::Err(_) => std::option::Option::None, }; } } return std::option::Option::None; } fn current_timestamp_or_zero() -> crate::MarketPriceTimestamp { return match crate::current_timestamp() { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => crate::MarketPriceTimestamp::from_unix_millis(0), }; } fn lock_registry(registry: &std::sync::Mutex) -> std::sync::MutexGuard<'_, crate::MarketPriceProviderRegistry> { return match registry.lock() { std::result::Result::Ok(value) => value, std::result::Result::Err(poisoned) => poisoned.into_inner(), }; } fn provider_not_found_error(provider_id: &crate::MarketPriceProviderId) -> ksp_core_lib::Error { return ksp_core_lib::Error::new(crate::ERROR_CODE_MARKET_PRICE_PROVIDER_NOT_FOUND, "Market-price provider is not configured") .with_context("provider_id", provider_id.as_str()); } fn refresh_error(field: &'static str) -> ksp_core_lib::Error { return ksp_core_lib::Error::new(crate::ERROR_CODE_MARKET_PRICE_REFRESH_INVALID, "Market-price refresh request is invalid").with_context("field", field); } fn timestamp_after_millis(delay_millis: u64) -> crate::MarketPriceTimestamp { let now = current_timestamp_or_zero().unix_millis(); return crate::MarketPriceTimestamp::from_unix_millis(now.saturating_add(delay_millis)); } fn validate_requested_provider_ids(service: &crate::MarketPriceService, provider_ids: &[crate::MarketPriceProviderId]) -> ksp_core_lib::Result<()> { let mut seen = std::collections::BTreeSet::new(); for provider_id in provider_ids { if service.provider_index(provider_id).is_none() { return std::result::Result::Err(provider_not_found_error(provider_id)); } if !seen.insert(provider_id) { return std::result::Result::Err(refresh_error("provider_id")); } } return std::result::Result::Ok(()); } #[cfg(test)] #[path = "../unit_tests/market_price_service.rs"] mod tests;