dex: work out the DEX LQT api

crates/core/component/dex/src/component/dex.rs

@@ -14,7 +14,7 @@ use penumbra_sdk_proto::{DomainType as _, StateReadProto, StateWriteProto};
 use tendermint::v0_37::abci;
 use tracing::instrument;
 
-use crate::state_key::block_scoped;
+use crate::state_key::{block_scoped,};
 use crate::{
     component::SwapDataRead, component::SwapDataWrite, event, genesis, state_key,
     BatchSwapOutputData, DexParameters, DirectedTradingPair, SwapExecution, TradingPair,

crates/core/component/dex/src/component/lqt.rs

@@ -0,0 +1,65 @@
+use crate::lp::position;
+use crate::state_key::lqt;
+use anyhow::Result;
+use async_trait::async_trait;
+use cnidarium::StateRead;
+use futures::StreamExt;
+use penumbra_sdk_asset::asset;
+use penumbra_sdk_num::Amount;
+use penumbra_sdk_proto::StateReadProto;
+use penumbra_sdk_sct::component::clock::EpochRead;
+use std::pin::Pin;
+
+/// Provides public read access to LQT data.
+#[async_trait]
+pub trait LqtRead: StateRead {
+    /// Returns the cumulative volume of staking token for a trading pair.
+    /// This is the sum of the outflows of the staking token from all positions in the pair.
+    ///
+    /// Default to zero if no volume is found.
+    async fn get_volume_for_pair(&self, asset: asset::Id) -> Amount {
+        let epoch = self.get_current_epoch().await.expect("epoch is always set");
+        let key = lqt::v1::pair::lookup::volume_by_pair(epoch.index, asset);
+        let value = self.nonverifiable_get(&key).await.unwrap_or_default();
+        value.unwrap_or_default()
+    }
+
+    /// Returns the cumulative volume of staking token for a given position id.
+    /// This is the sum of the outflows of the staking token from the position.
+    ///
+    /// Default to zero if no volume is found.
+    async fn get_volume_for_position(&self, position_id: &position::Id) -> Amount {
+        let epoch = self.get_current_epoch().await.expect("epoch is always set");
+        let key = lqt::v1::lp::lookup::volume_by_position(epoch.index, position_id);
+        let value = self.nonverifiable_get(&key).await.unwrap_or_default();
+        value.unwrap_or_default()
+    }
+
+    /// Returns a stream of position ids sorted by descending volume.
+    /// The volume is the sum of the outflows of the staking token from the position.
+    fn positions_by_volume_stream(
+        &self,
+        epoch_index: u64,
+        asset_id: asset::Id,
+    ) -> Result<
+        Pin<
+            Box<
+                dyn futures::Stream<Item = Result<(asset::Id, position::Id, Amount)>>
+                    + Send
+                    + 'static,
+            >,
+        >,
+    > {
+        let key = lqt::v1::lp::by_volume::prefix_with_asset(epoch_index, &asset_id);
+        Ok(self
+            .nonverifiable_prefix_raw(&key)
+            .map(|res| {
+                res.map(|(raw_entry, _)| {
+                    let (asset, volume, position_id) =
+                        lqt::v1::lp::by_volume::parse_key(&raw_entry).expect("TODO");
+                    (asset, position_id, volume)
+                })
+            })
+            .boxed())
+    }
+}

crates/core/component/dex/src/component/mod.rs

@@ -13,13 +13,15 @@ pub(crate) mod circuit_breaker;
 mod dex;
 mod eviction_manager;
 mod flow;
+mod lqt;
 mod position_manager;
 mod swap_manager;
 
 pub use dex::{Dex, StateReadExt, StateWriteExt};
 pub use position_manager::PositionManager;
 
 // Read data from the Dex component;
+pub use lqt::LqtRead;
 pub use position_manager::PositionRead;
 pub use swap_manager::SwapDataRead;
 

crates/core/component/dex/src/state_key.rs

@@ -121,26 +121,52 @@ pub fn aggregate_value() -> &'static str {
 
 pub mod lqt {
     pub mod v1 {
-        pub mod lp {
+        pub mod pair {
             pub mod lookup {
                 use penumbra_sdk_asset::asset;
 
-                pub(crate) fn _prefix(epoch_index: u64) -> String {
-                    format!("dex/lqt/v1/lp/lookup/{epoch_index:020}/")
+                pub(crate) fn prefix(epoch_index: u64) -> String {
+                    format!("dex/lqt/v1/pair/lookup/{epoch_index:020}/")
                 }
 
-                // A lookup index used to update cumulative volumes.
-                /// It maps an trading pair (staking token, asset) to the cumulative volume of outbound liquidity.
+                // A lookup index used to inspect aggregate outflows for a given pair.
+                /// It maps a trading pair (staking token, asset) to the cumulative volume of outbound liquidity.
                 ///
                 /// # Key Encoding
                 /// The lookup key is encoded as `prefix || asset`
                 /// # Value Encoding
                 /// The value is encoded as `BE(Amount)`
-                pub(crate) fn _volume_by_pair(epoch_index: u64, asset: asset::Id) -> [u8; 74] {
-                    let prefix_bytes = _prefix(epoch_index);
+                pub(crate) fn volume_by_pair(epoch_index: u64, asset: asset::Id) -> [u8; 76] {
+                    let prefix_bytes = prefix(epoch_index);
+                    let mut key = [0u8; 76];
+                    key[0..44].copy_from_slice(prefix_bytes.as_bytes());
+                    key[44..44 + 32].copy_from_slice(&asset.to_bytes());
+                    key
+                }
+            }
+        }
+
+        pub mod lp {
+            pub mod lookup {
+                pub(crate) fn prefix(epoch_index: u64) -> String {
+                    format!("dex/lqt/v1/lp/lookup/{epoch_index:020}/")
+                }
+
+                /// A lookup index used to update the `by_volume` index.
+                /// It maps a position id to the latest tally of outbound cumulative volume.
+                ///
+                /// # Key Encoding
+                /// The lookup key is encoded as `prefix || position_id`
+                /// # Value Encoding
+                /// The value is encoded as `BE(Amount)`
+                pub(crate) fn volume_by_position(
+                    epoch_index: u64,
+                    position_id: &crate::lp::position::Id,
+                ) -> [u8; 74] {
+                    let prefix_bytes = prefix(epoch_index);
                     let mut key = [0u8; 74];
                     key[0..42].copy_from_slice(prefix_bytes.as_bytes());
-                    key[42..42 + 32].copy_from_slice(&asset.to_bytes());
+                    key[42..42 + 32].copy_from_slice(&position_id.0);
                     key
                 }
             }
@@ -150,45 +176,61 @@ pub mod lqt {
                 use penumbra_sdk_asset::asset;
                 use penumbra_sdk_num::Amount;
 
+                use crate::lp::position;
+
                 pub fn prefix(epoch_index: u64) -> String {
                     format!("dex/lqt/v1/lp/by_volume/{epoch_index:020}/")
                 }
 
+                pub fn prefix_with_asset(epoch_index: u64, asset: &asset::Id) -> [u8; 74] {
+                    let prefix = prefix(epoch_index);
+                    let mut key = [0u8; 74];
+                    key[0..42].copy_from_slice(prefix.as_bytes());
+                    key[42..42 + 32].copy_from_slice(&asset.to_bytes());
+                    key
+                }
+
                 /// Tracks the cumulative volume of outbound liquidity for a given pair.
                 /// The pair is always connected by the staking token, which is the implicit numeraire.
                 ///
                 /// # Encoding
-                /// The full key is encoded as: `prefix || asset || BE(!volume)`
+                /// The full key is encoded as: `prefix || asset || BE(!volume) || position`
                 pub(crate) fn _key(
                     epoch_index: u64,
                     asset: &asset::Id,
                     volume: Amount,
-                ) -> [u8; 93] {
+                    position: &position::Id,
+                ) -> [u8; 125] {
                     let prefix_bytes = prefix(epoch_index);
-                    let mut key = [0u8; 93];
+                    let mut key = [0u8; 125];
                     key[0..45].copy_from_slice(prefix_bytes.as_bytes());
                     key[45..45 + 32].copy_from_slice(&asset.to_bytes());
                     key[45 + 32..45 + 32 + 16].copy_from_slice(&(!volume).to_be_bytes());
+                    key[45 + 32 + 16..45 + 32 + 16 + 32].copy_from_slice(&position.0);
                     key
                 }
 
-                /// Extract the cumulative amount of liquidity from a fully specified key.
+                /// Parse a raw key into its constituent parts.
                 ///
                 /// # Errors
-                /// This function will return an error if the key is not 72 bytes. Or, if the
-                /// key contains an invalid asset identifier.
-                pub(crate) fn _parse_key(key: &[u8]) -> Result<(asset::Id, Amount)> {
-                    ensure!(key.len() == 93, "key must be 93 bytes");
+                /// This function will return an error if the key is not 125 bytes. Or, if the
+                /// key contains an invalid asset or position identifier.
+                pub(crate) fn parse_key(key: &[u8]) -> Result<(asset::Id, Amount, position::Id)> {
+                    ensure!(key.len() == 125, "key must be 125 bytes");
 
-                    // skip the first 45 bytes of prefix
+                    // Skip the first 45 bytes, which is the prefix.
                     let raw_asset: [u8; 32] = key[45..45 + 32].try_into()?;
                     let asset: asset::Id = raw_asset.try_into()?;
 
                     let raw_amount: [u8; 16] = key[45 + 32..45 + 32 + 16].try_into()?;
                     let amount_complement = Amount::from_be_bytes(raw_amount);
                     let amount = !amount_complement;
 
-                    Ok((asset, amount))
+                    let raw_position_id: [u8; 32] =
+                        key[45 + 32 + 16..45 + 32 + 16 + 32].try_into()?;
+                    let position_id = position::Id(raw_position_id);
+
+                    Ok((asset, amount, position_id))
                 }
             }
         }

crates/core/component/dex/src/trading_pair.rs

@@ -102,6 +102,10 @@ impl TradingPair {
         self.asset_2
     }
 
+    pub fn contains(&self, asset_id: asset::Id) -> bool {
+        self.asset_1 == asset_id || self.asset_2 == asset_id
+    }
+
     /// Convert the trading pair to bytes.
     pub(crate) fn to_bytes(self) -> [u8; 64] {
         let mut result: [u8; 64] = [0; 64];