Implement an event for candlestick data (#4889)

## Describe your changes

This adds a new event `EventCandlestickData`, containing a directed
trading pair and a candlestick, allowing us to emit events for when
trading activity producing a stick happens on a given pair, at a given
height.

## Issue ticket number and link

Closes #4868

## Checklist before requesting a review

- [x] If this code contains consensus-breaking changes, I have added the
"consensus-breaking" label. Otherwise, I declare my belief that there
are not consensus-breaking changes, for the following reason:

  > event addition only

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

@@ -8,6 +8,7 @@ use penumbra_proto::{DomainType, StateReadProto, StateWriteProto};
 use penumbra_sct::component::clock::EpochRead as _;
 use tonic::async_trait;
 
+use crate::event::EventCandlestickData;
 use crate::{lp::position::Position, state_key::candlesticks, DirectedTradingPair, SwapExecution};
 
 use crate::CandlestickData;
@@ -210,6 +211,13 @@ pub trait Chandelier: StateWrite {
                 candlesticks::data::by_pair_and_height(&trading_pair, height).into(),
                 candlestick,
             );
+            self.record_proto(
+                EventCandlestickData {
+                    pair: *trading_pair,
+                    stick: candlestick,
+                }
+                .to_proto(),
+            )
         }
 
         Ok(())

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

@@ -5,12 +5,14 @@ use crate::{
     },
     swap::Swap,
     swap_claim::SwapClaim,
-    BatchSwapOutputData, DirectedTradingPair, SwapExecution,
+    BatchSwapOutputData, CandlestickData, DirectedTradingPair, SwapExecution,
 };
+use anyhow::{anyhow, Context};
+use prost::Name;
 
 use penumbra_asset::asset;
 use penumbra_num::Amount;
-use penumbra_proto::penumbra::core::component::dex::v1 as pb;
+use penumbra_proto::{penumbra::core::component::dex::v1 as pb, DomainType};
 
 pub fn swap(swap: &Swap) -> pb::EventSwap {
     pb::EventSwap {
@@ -135,3 +137,36 @@ pub fn vcb_debit(
         new_balance: Some(new_balance.into()),
     }
 }
+
+#[derive(Clone, Debug)]
+pub struct EventCandlestickData {
+    pub pair: DirectedTradingPair,
+    pub stick: CandlestickData,
+}
+
+impl TryFrom<pb::EventCandlestickData> for EventCandlestickData {
+    type Error = anyhow::Error;
+
+    fn try_from(value: pb::EventCandlestickData) -> Result<Self, Self::Error> {
+        fn inner(value: pb::EventCandlestickData) -> anyhow::Result<EventCandlestickData> {
+            Ok(EventCandlestickData {
+                pair: value.pair.ok_or(anyhow!("missing `pair`"))?.try_into()?,
+                stick: value.stick.ok_or(anyhow!("missing `stick`"))?.try_into()?,
+            })
+        }
+        inner(value).context(format!("parsing {}", pb::EventCandlestickData::NAME))
+    }
+}
+
+impl From<EventCandlestickData> for pb::EventCandlestickData {
+    fn from(value: EventCandlestickData) -> Self {
+        Self {
+            pair: Some(value.pair.into()),
+            stick: Some(value.stick.into()),
+        }
+    }
+}
+
+impl DomainType for EventCandlestickData {
+    type Proto = pb::EventCandlestickData;
+}

crates/proto/src/gen/penumbra.core.component.dex.v1.rs

@@ -1555,6 +1555,27 @@ impl ::prost::Name for EventValueCircuitBreakerDebit {
         ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
     }
 }
+/// Emitted whenever there's non-empty candlestick data for a particular pair.
+///
+/// Beware: if there's no activity on a given pair, there's no guarantee
+/// that a candlestick will be emitted.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EventCandlestickData {
+    /// The pair the candlestick is for.
+    #[prost(message, optional, tag = "1")]
+    pub pair: ::core::option::Option<DirectedTradingPair>,
+    /// The candlestick for this particular pair.
+    #[prost(message, optional, tag = "2")]
+    pub stick: ::core::option::Option<CandlestickData>,
+}
+impl ::prost::Name for EventCandlestickData {
+    const NAME: &'static str = "EventCandlestickData";
+    const PACKAGE: &'static str = "penumbra.core.component.dex.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
+    }
+}
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct DexParameters {

crates/proto/src/gen/penumbra.core.component.dex.v1.serde.rs

@@ -2225,6 +2225,118 @@ impl<'de> serde::Deserialize<'de> for EventBatchSwap {
         deserializer.deserialize_struct("penumbra.core.component.dex.v1.EventBatchSwap", FIELDS, GeneratedVisitor)
     }
 }
+impl serde::Serialize for EventCandlestickData {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeStruct;
+        let mut len = 0;
+        if self.pair.is_some() {
+            len += 1;
+        }
+        if self.stick.is_some() {
+            len += 1;
+        }
+        let mut struct_ser = serializer.serialize_struct("penumbra.core.component.dex.v1.EventCandlestickData", len)?;
+        if let Some(v) = self.pair.as_ref() {
+            struct_ser.serialize_field("pair", v)?;
+        }
+        if let Some(v) = self.stick.as_ref() {
+            struct_ser.serialize_field("stick", v)?;
+        }
+        struct_ser.end()
+    }
+}
+impl<'de> serde::Deserialize<'de> for EventCandlestickData {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &[
+            "pair",
+            "stick",
+        ];
+
+        #[allow(clippy::enum_variant_names)]
+        enum GeneratedField {
+            Pair,
+            Stick,
+            __SkipField__,
+        }
+        impl<'de> serde::Deserialize<'de> for GeneratedField {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                struct GeneratedVisitor;
+
+                impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+                    type Value = GeneratedField;
+
+                    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                        write!(formatter, "expected one of: {:?}", &FIELDS)
+                    }
+
+                    #[allow(unused_variables)]
+                    fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+                    where
+                        E: serde::de::Error,
+                    {
+                        match value {
+                            "pair" => Ok(GeneratedField::Pair),
+                            "stick" => Ok(GeneratedField::Stick),
+                            _ => Ok(GeneratedField::__SkipField__),
+                        }
+                    }
+                }
+                deserializer.deserialize_identifier(GeneratedVisitor)
+            }
+        }
+        struct GeneratedVisitor;
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = EventCandlestickData;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("struct penumbra.core.component.dex.v1.EventCandlestickData")
+            }
+
+            fn visit_map<V>(self, mut map_: V) -> std::result::Result<EventCandlestickData, V::Error>
+                where
+                    V: serde::de::MapAccess<'de>,
+            {
+                let mut pair__ = None;
+                let mut stick__ = None;
+                while let Some(k) = map_.next_key()? {
+                    match k {
+                        GeneratedField::Pair => {
+                            if pair__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("pair"));
+                            }
+                            pair__ = map_.next_value()?;
+                        }
+                        GeneratedField::Stick => {
+                            if stick__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("stick"));
+                            }
+                            stick__ = map_.next_value()?;
+                        }
+                        GeneratedField::__SkipField__ => {
+                            let _ = map_.next_value::<serde::de::IgnoredAny>()?;
+                        }
+                    }
+                }
+                Ok(EventCandlestickData {
+                    pair: pair__,
+                    stick: stick__,
+                })
+            }
+        }
+        deserializer.deserialize_struct("penumbra.core.component.dex.v1.EventCandlestickData", FIELDS, GeneratedVisitor)
+    }
+}
 impl serde::Serialize for EventPositionClose {
     #[allow(deprecated)]
     fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>

proto/penumbra/penumbra/core/component/dex/v1/dex.proto

@@ -737,6 +737,17 @@ message EventValueCircuitBreakerDebit {
   num.v1.Amount new_balance = 3;
 }
 
+// Emitted whenever there's non-empty candlestick data for a particular pair.
+//
+// Beware: if there's no activity on a given pair, there's no guarantee
+// that a candlestick will be emitted.
+message EventCandlestickData {
+  // The pair the candlestick is for.
+  DirectedTradingPair pair = 1;
+  // The candlestick for this particular pair.
+  CandlestickData stick = 2;
+}
+
 message DexParameters {
   // Whether or not the DEX is enabled.
   bool is_enabled = 1;