feat: Using async icq for osmosis price feeder

Cargo.toml

@@ -3,7 +3,6 @@ members = [
   "packages/*",
   "contracts/provider/*",
   "contracts/consumer/*",
-  "contracts/osmosis-price-provider",
 ]
 resolver = "2"
 
@@ -37,6 +36,7 @@ cw-storage-plus = "1.2.0"
 cw-utils = "1.0.3"
 cw2 = "1.1.2"
 osmosis-std = "0.20.1"
+prost = { version = "0.11.0",  default-features = false, features = ["prost-derive"] }
 schemars = "0.8.17"
 serde = { version = "1.0.199", default-features = false, features = ["derive"] }
 thiserror = "1.0.59"

contracts/consumer/band-price-feed/src/contract.rs

@@ -67,6 +67,7 @@ impl RemotePriceFeedContract {
         prepare_gas: Uint64,
         execute_gas: Uint64,
         minimum_sources: u8,
+        epoch_in_secs: u64,
         price_info_ttl_in_secs: u64,
     ) -> Result<Response, ContractError> {
         nonpayable(&ctx.info)?;
@@ -86,6 +87,7 @@ impl RemotePriceFeedContract {
                 minimum_sources,
             },
         )?;
+        self.scheduler.init(&mut ctx.deps, epoch_in_secs)?;
         self.price_keeper
             .init(&mut ctx.deps, price_info_ttl_in_secs)?;
         Ok(Response::new())
@@ -204,6 +206,7 @@ mod tests {
                 Uint64::new(200000),
                 1,
                 60,
+                60,
             )
             .unwrap();
     }

contracts/consumer/band-price-feed/src/error.rs

@@ -1,5 +1,6 @@
 use cosmwasm_std::StdError;
 use cw_utils::PaymentError;
+use mesh_apis::ibc::VersionError;
 use thiserror::Error;
 
 use mesh_price_feed::PriceKeeperError;
@@ -16,6 +17,9 @@ pub enum ContractError {
     #[error("{0}")]
     Payment(#[from] PaymentError),
 
+    #[error("{0}")]
+    IbcVersion(#[from] VersionError),
+
     #[error("{0}")]
     PriceKeeper(#[from] PriceKeeperError),
 

contracts/consumer/band-price-feed/src/ibc.rs

@@ -8,7 +8,7 @@ use cosmwasm_std::{
     StdError, Uint128,
 };
 use cw_band::{OracleResponsePacketData, Output, ResolveStatus};
-use mesh_apis::ibc::{ack_fail, ack_success, PriceFeedAck};
+use mesh_apis::ibc::{ack_fail, ack_success, validate_channel_order, PriceFeedAck};
 use obi::OBIDecode;
 
 use crate::contract::RemotePriceFeedContract;
@@ -28,10 +28,12 @@ pub fn ibc_channel_open(
     if contract.channel.may_load(deps.storage)?.is_some() {
         return Err(ContractError::IbcChannelAlreadyOpen);
     }
-    // ensure we are called with OpenInit
     let channel = msg.channel();
     let counterparty_version = msg.counterparty_version();
 
+    // verify the ordering is correct
+    validate_channel_order(&channel.order)?;
+
     if channel.version != IBC_APP_VERSION {
         return Err(ContractError::InvalidIbcVersion {
             version: channel.version.clone(),
@@ -83,9 +85,8 @@ pub fn ibc_channel_connect(
             });
         }
     }
-    if channel.order != IbcOrder::Unordered {
-        return Err(ContractError::OnlyUnorderedChannel {});
-    }
+
+    validate_channel_order(&channel.order)?;
 
     // Version negotiation over, we can only store the channel
     let contract = RemotePriceFeedContract::new();

contracts/consumer/osmosis-price-feed/Cargo.toml

@@ -28,6 +28,8 @@ cw-storage-plus  = { workspace = true }
 cw2              = { workspace = true }
 cw-utils         = { workspace = true }
 
+osmosis-std      = { workspace = true }
+
 schemars         = { workspace = true }
 serde            = { workspace = true }
 thiserror        = { workspace = true }

contracts/consumer/osmosis-price-feed/src/contract.rs

@@ -1,21 +1,27 @@
+use std::vec;
+
 use cosmwasm_std::{Decimal, DepsMut, Env, IbcChannel, Response, Timestamp};
 use cw2::set_contract_version;
 use cw_storage_plus::Item;
 use cw_utils::nonpayable;
-use sylvia::types::{InstantiateCtx, QueryCtx, SudoCtx};
+use mesh_apis::ibc::{encode_request, ibc_query_packet, ArithmeticTwapToNowRequest, CosmosQuery};
+use osmosis_std::shim::Timestamp as OsmosisTimestamp;
+use osmosis_std::types::tendermint::abci::RequestQuery;
+use sylvia::types::{ExecCtx, InstantiateCtx, QueryCtx, SudoCtx};
 use sylvia::{contract, schemars};
 
 use mesh_apis::price_feed_api::{self, PriceFeedApi, PriceResponse};
 
 use crate::error::ContractError;
-use crate::ibc::{make_ibc_packet, AUTH_ENDPOINT};
-use crate::msg::AuthorizedEndpoint;
+use crate::ibc::make_ibc_packet;
 use crate::state::TradingPair;
 use mesh_price_feed::{Action, PriceKeeper, Scheduler};
 
 pub const CONTRACT_NAME: &str = env!("CARGO_PKG_NAME");
 pub const CONTRACT_VERSION: &str = env!("CARGO_PKG_VERSION");
 
+pub const OSMOSIS_QUERY_TWAP_PATH: &str = "/osmosis.twap.v1beta1.Query/ArithmeticTwapToNow";
+
 pub struct RemotePriceFeedContract {
     pub channel: Item<'static, IbcChannel>,
     pub trading_pair: Item<'static, TradingPair>,
@@ -53,7 +59,6 @@ impl RemotePriceFeedContract {
         &self,
         mut ctx: InstantiateCtx,
         trading_pair: TradingPair,
-        auth_endpoint: AuthorizedEndpoint,
         epoch_in_secs: u64,
         price_info_ttl_in_secs: u64,
     ) -> Result<Response, ContractError> {
@@ -65,12 +70,15 @@ impl RemotePriceFeedContract {
         self.price_keeper
             .init(&mut ctx.deps, price_info_ttl_in_secs)?;
         self.scheduler.init(&mut ctx.deps, epoch_in_secs)?;
-
-        AUTH_ENDPOINT.save(ctx.deps.storage, &auth_endpoint)?;
-
         Ok(Response::new())
     }
 
+    #[sv::msg(exec)]
+    pub fn request(&self, ctx: ExecCtx) -> Result<Response, ContractError> {
+        let ExecCtx { deps, env, info: _ } = ctx;
+        query_twap(deps, &env)
+    }
+
     pub(crate) fn update_twap(
         &self,
         deps: DepsMut,
@@ -116,19 +124,39 @@ pub fn query_twap(deps: DepsMut, env: &Env) -> Result<Response, ContractError> {
         .may_load(deps.storage)?
         .ok_or(ContractError::IbcChannelNotOpen)?;
 
-    let packet = mesh_apis::ibc::RemotePriceFeedPacket::QueryTwap {
+    let request = ArithmeticTwapToNowRequest {
         pool_id,
         base_asset,
         quote_asset,
+        start_time: Some(OsmosisTimestamp {
+            seconds: env.block.time.seconds() as i64,
+            nanos: 0,
+        }),
+    };
+    let packet = CosmosQuery {
+        requests: vec![RequestQuery {
+            path: OSMOSIS_QUERY_TWAP_PATH.to_string(),
+            data: encode_request(&request),
+            height: 0,
+            prove: false,
+        }],
     };
-    let msg = make_ibc_packet(&env.block.time, channel, packet)?;
+
+    let msg = make_ibc_packet(&env.block.time, channel, ibc_query_packet(packet))?;
 
     Ok(Response::new().add_message(msg))
 }
 
 #[cfg(test)]
 mod tests {
-    use cosmwasm_std::testing::{mock_dependencies, mock_env, mock_info};
+    use cosmwasm_std::{
+        from_json,
+        testing::{mock_dependencies, mock_env, mock_info},
+        Binary,
+    };
+    use mesh_apis::ibc::{
+        decode_response, AcknowledgementResult, CosmosResponse, InterchainQueryPacketAck,
+    };
 
     use super::*;
 
@@ -144,10 +172,6 @@ mod tests {
             base_asset: "base".to_string(),
             quote_asset: "quote".to_string(),
         };
-        let auth_endpoint = AuthorizedEndpoint {
-            connection_id: "connection".to_string(),
-            port_id: "port".to_string(),
-        };
 
         contract
             .instantiate(
@@ -157,10 +181,29 @@ mod tests {
                     info,
                 },
                 trading_pair,
-                auth_endpoint,
                 10,
                 50,
             )
             .unwrap();
     }
+
+    #[test]
+    fn json_binary() {
+        let resp = Binary::from_base64("eyJyZXN1bHQiOiJleUprWVhSaElqb2lRMmhqTmtaUmIxUk5WRUYzVFVSQmQwMUVRWGROUkVGM1RVUkJkMDFFUVhkTlFUMDlJbjA9In0=").unwrap();
+
+        let ack_result: AcknowledgementResult = from_json(resp).unwrap();
+        assert_eq!(
+            ack_result.result.to_string(),
+            String::from("eyJkYXRhIjoiQ2hjNkZRb1RNVEF3TURBd01EQXdNREF3TURBd01EQXdNQT09In0=")
+        );
+
+        let packet_ack: InterchainQueryPacketAck = from_json(&ack_result.result).unwrap();
+        assert_eq!(
+            packet_ack.data.to_string(),
+            String::from("Chc6FQoTMTAwMDAwMDAwMDAwMDAwMDAwMA==")
+        );
+
+        let response: CosmosResponse = decode_response(&packet_ack.data).unwrap();
+        assert_eq!(response.responses.len(), 1);
+    }
 }

contracts/consumer/osmosis-price-feed/src/error.rs

@@ -25,6 +25,18 @@ pub enum ContractError {
     #[error("Invalid authorized endpoint: {0}")]
     InvalidEndpoint(String),
 
+    #[error("Only supports channel with ibc version icq-1, got {version}")]
+    InvalidIbcVersion { version: String },
+
+    #[error("invalid ibc packet, result should only contains 1 ResponseQuery")]
+    InvalidResponseQuery,
+
+    #[error("failed to send interchain query")]
+    InvalidResponseQueryCode,
+
+    #[error("twap data is empty")]
+    EmptyTwap,
+
     #[error("Contract doesn't have an open IBC channel")]
     IbcChannelNotOpen,