feat: include signer messages in Stacks predicate payloads

Cargo.toml

@@ -8,4 +8,4 @@ default-members = ["components/chainhook-cli", "components/chainhook-sdk"]
 resolver = "2"
 
 [patch.crates-io]
-stacks-codec = { git = "https://github.com/hirosystems/clarinet.git", rev = "3a2f9136abd85b265e538fbe51c808e9c09a06cb" }
+stacks-codec = { git = "https://github.com/hirosystems/clarinet.git", rev = "b0683675115562d719ed4b5245f620e0990030a0" }

components/chainhook-cli/src/scan/stacks.rs

@@ -339,6 +339,7 @@ pub async fn scan_stacks_chainstate_via_rocksdb_using_predicate(
             chainhook: predicate_spec,
             apply: hits_per_blocks,
             rollback: vec![],
+            events: vec![]
         };
         let res = match handle_stacks_hook_action(
             trigger,
@@ -533,6 +534,8 @@ pub async fn scan_stacks_chainstate_via_csv_using_predicate(
             chainhook: predicate_spec,
             apply: hits_per_blocks,
             rollback: vec![],
+            // TODO(rafaelcr): Consider StackerDB chunks that come from TSVs.
+            events: vec![]
         };
         match handle_stacks_hook_action(trigger, &proofs, &config.get_event_observer_config(), ctx)
         {

components/chainhook-cli/src/service/mod.rs

@@ -619,7 +619,7 @@ impl Service {
                             }
                             StacksChainEvent::ChainUpdatedWithMicroblocks(_)
                             | StacksChainEvent::ChainUpdatedWithMicroblocksReorg(_) => {},
-                            StacksChainEvent::ChainUpdatedWithStackerDbChunks(data) => {
+                            StacksChainEvent::ChainUpdatedWithStackerDbChunks(_) => {
                                 // TODO(rafaelcr): Send via HTTP payload.
                             },
                         };

components/chainhook-sdk/src/chainhooks/stacks/mod.rs

@@ -1,13 +1,15 @@
 use crate::observer::EventObserverConfig;
-use crate::utils::{AbstractStacksBlock, Context, MAX_BLOCK_HEIGHTS_ENTRIES};
+use crate::utils::{
+    AbstractStacksBlock, AbstractStacksNonConsensusEvent, Context, MAX_BLOCK_HEIGHTS_ENTRIES,
+};
 
+use super::types::validate_txid;
 use super::types::{
     append_error_context, BlockIdentifierIndexRule, ChainhookInstance, ExactMatchingRule,
     HookAction,
 };
-use super::types::validate_txid;
 use chainhook_types::{
-    BlockIdentifier, StacksChainEvent, StacksNetwork, StacksTransactionData,
+    BlockIdentifier, StacksChainEvent, StacksNetwork, StacksStackerDbChunk, StacksTransactionData,
     StacksTransactionEvent, StacksTransactionEventPayload, StacksTransactionKind,
     TransactionIdentifier,
 };
@@ -259,6 +261,8 @@ pub enum StacksPredicate {
     NftEvent(StacksNftEventBasedPredicate),
     StxEvent(StacksStxEventBasedPredicate),
     Txid(ExactMatchingRule),
+    #[cfg(feature = "stacks-signers")]
+    SignerMessage(StacksSignerMessagePredicate),
 }
 
 impl StacksPredicate {
@@ -307,11 +311,28 @@ impl StacksPredicate {
                     ));
                 }
             }
+            #[cfg(feature = "stacks-signers")]
+            StacksPredicate::SignerMessage(StacksSignerMessagePredicate::FromSignerPubKey(_)) => {
+                // TODO(rafaelcr): Validate pubkey format
+            }
+            #[cfg(feature = "stacks-signers")]
+            StacksPredicate::SignerMessage(_) => {}
         }
         Ok(())
     }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum StacksSignerMessagePredicate {
+    AfterTimestamp(u64),
+    FromSignerPubKey(String),
+}
+
+impl StacksSignerMessagePredicate {
+    // TODO(rafaelcr): Write validators
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub struct StacksContractCallBasedPredicate {
@@ -457,6 +478,7 @@ pub struct StacksTriggerChainhook<'a> {
     pub chainhook: &'a StacksChainhookInstance,
     pub apply: Vec<(Vec<&'a StacksTransactionData>, &'a dyn AbstractStacksBlock)>,
     pub rollback: Vec<(Vec<&'a StacksTransactionData>, &'a dyn AbstractStacksBlock)>,
+    pub events: Vec<&'a dyn AbstractStacksNonConsensusEvent>,
 }
 
 #[derive(Clone, Debug)]
@@ -484,17 +506,13 @@ pub struct StacksChainhookOccurrencePayload {
 }
 
 impl StacksChainhookOccurrencePayload {
-    pub fn from_trigger(
-        trigger: StacksTriggerChainhook<'_>,
-    ) -> StacksChainhookOccurrencePayload {
+    pub fn from_trigger(trigger: StacksTriggerChainhook<'_>) -> StacksChainhookOccurrencePayload {
         StacksChainhookOccurrencePayload {
             apply: trigger
                 .apply
                 .into_iter()
                 .map(|(transactions, block)| {
-                    let transactions = transactions
-                        .into_iter().cloned()
-                        .collect::<Vec<_>>();
+                    let transactions = transactions.into_iter().cloned().collect::<Vec<_>>();
                     StacksApplyTransactionPayload {
                         block_identifier: block.get_identifier().clone(),
                         transactions,
@@ -505,9 +523,7 @@ impl StacksChainhookOccurrencePayload {
                 .rollback
                 .into_iter()
                 .map(|(transactions, block)| {
-                    let transactions = transactions
-                        .into_iter().cloned()
-                        .collect::<Vec<_>>();
+                    let transactions = transactions.into_iter().cloned().collect::<Vec<_>>();
                     StacksRollbackTransactionPayload {
                         block_identifier: block.get_identifier().clone(),
                         transactions,
@@ -593,6 +609,7 @@ pub fn evaluate_stacks_chainhooks_on_chain_event<'a>(
                         chainhook,
                         apply,
                         rollback,
+                        events: vec![],
                     })
                 }
             }
@@ -621,6 +638,7 @@ pub fn evaluate_stacks_chainhooks_on_chain_event<'a>(
                         chainhook,
                         apply,
                         rollback,
+                        events: vec![],
                     })
                 }
             }
@@ -657,6 +675,7 @@ pub fn evaluate_stacks_chainhooks_on_chain_event<'a>(
                         chainhook,
                         apply,
                         rollback,
+                        events: vec![],
                     })
                 }
             }
@@ -718,13 +737,35 @@ pub fn evaluate_stacks_chainhooks_on_chain_event<'a>(
                         chainhook,
                         apply,
                         rollback,
+                        events: vec![],
                     })
                 }
             }
-        },
+        }
         StacksChainEvent::ChainUpdatedWithStackerDbChunks(data) => {
-            // TODO: Support predicates to send this data
-        },
+            for chainhook in active_chainhooks.iter() {
+                let mut events = vec![];
+
+                evaluated_predicates.insert(chainhook.uuid.as_str(), &data.received_at_block);
+                let mut chunks: Vec<&dyn AbstractStacksNonConsensusEvent> = vec![];
+                for chunk in data.chunks.iter() {
+                    chunks.push(chunk);
+                }
+                let (mut occurrences, mut expirations) =
+                    evaluate_stacks_predicate_on_stackerdb_chunks(chunks, chainhook, ctx);
+                events.append(&mut occurrences);
+                expired_predicates.append(&mut expirations);
+
+                if events.len() > 0 {
+                    triggered_predicates.push(StacksTriggerChainhook {
+                        chainhook,
+                        apply: vec![],
+                        rollback: vec![],
+                        events,
+                    });
+                }
+            }
+        }
     }
     (
         triggered_predicates,
@@ -795,7 +836,45 @@ pub fn evaluate_stacks_predicate_on_block<'a>(
         | StacksPredicate::StxEvent(_)
         | StacksPredicate::PrintEvent(_)
         | StacksPredicate::Txid(_) => unreachable!(),
+        #[cfg(feature = "stacks-signers")]
+        StacksPredicate::SignerMessage(_) => unreachable!(),
+    }
+}
+
+#[cfg(feature = "stacks-signers")]
+pub fn evaluate_stacks_predicate_on_stackerdb_chunks<'a>(
+    chunks: Vec<&'a dyn AbstractStacksNonConsensusEvent>,
+    chainhook: &'a StacksChainhookInstance,
+    _ctx: &Context,
+) -> (
+    Vec<&'a dyn AbstractStacksNonConsensusEvent>,
+    BTreeMap<&'a str, &'a BlockIdentifier>,
+) {
+    let mut occurrences = vec![];
+    let expired_predicates = BTreeMap::new();
+    for chunk in chunks {
+        match &chainhook.predicate {
+            StacksPredicate::SignerMessage(StacksSignerMessagePredicate::AfterTimestamp(
+                timestamp,
+            )) => {
+                if chunk.get_timestamp() >= *timestamp as i64 {
+                    occurrences.push(chunk);
+                }
+            }
+            StacksPredicate::SignerMessage(StacksSignerMessagePredicate::FromSignerPubKey(_)) => {
+                todo!()
+            }
+            StacksPredicate::BlockHeight(_)
+            | StacksPredicate::ContractDeployment(_)
+            | StacksPredicate::ContractCall(_)
+            | StacksPredicate::FtEvent(_)
+            | StacksPredicate::NftEvent(_)
+            | StacksPredicate::StxEvent(_)
+            | StacksPredicate::PrintEvent(_)
+            | StacksPredicate::Txid(_) => unreachable!(),
+        };
     }
+    (occurrences, expired_predicates)
 }
 
 pub fn evaluate_stacks_predicate_on_transaction<'a>(
@@ -952,7 +1031,9 @@ pub fn evaluate_stacks_predicate_on_transaction<'a>(
         }
         StacksPredicate::PrintEvent(expected_event) => {
             for event in transaction.metadata.receipt.events.iter() {
-                if let StacksTransactionEventPayload::SmartContractEvent(actual) = &event.event_payload {
+                if let StacksTransactionEventPayload::SmartContractEvent(actual) =
+                    &event.event_payload
+                {
                     if actual.topic == "print" {
                         match expected_event {
                             StacksPrintEventBasedPredicate::Contains {
@@ -1006,6 +1087,8 @@ pub fn evaluate_stacks_predicate_on_transaction<'a>(
             txid.eq(&transaction.transaction_identifier.hash)
         }
         StacksPredicate::BlockHeight(_) => unreachable!(),
+        #[cfg(feature = "stacks-signers")]
+        StacksPredicate::SignerMessage(_) => unreachable!(),
     }
 }
 
@@ -1238,7 +1321,7 @@ pub fn serialized_decoded_clarity_value(hex_value: &str, ctx: &Context) -> serde
         Ok(bytes) => bytes,
         _ => return json!(hex_value.to_string()),
     };
-    
+
     match ClarityValue::consensus_deserialize(&mut Cursor::new(&value_bytes)) {
         Ok(value) => serialize_to_json(&value),
         Err(e) => {

components/chainhook-sdk/src/chainhooks/tests/mod.rs

@@ -732,6 +732,7 @@ fn test_stacks_hook_action_noop() {
         chainhook: &chainhook,
         apply: vec![(apply_transactions, apply_blocks)],
         rollback: vec![(rollback_transactions, rollback_blocks)],
+        events: vec![]
     };
 
     let proofs = HashMap::new();
@@ -810,6 +811,7 @@ fn test_stacks_hook_action_file_append() {
         chainhook: &chainhook,
         apply,
         rollback: vec![(rollback_transactions, rollback_block)],
+        events: vec![]
     };
 
     let proofs = HashMap::new();

components/chainhook-sdk/src/indexer/mod.rs

@@ -175,17 +175,22 @@ impl Indexer {
         ctx: &Context,
     ) -> Result<Option<StacksChainEvent>, String> {
         use chainhook_types::StacksChainUpdatedWithStackerDbChunksData;
-
+        let Some(chain_tip) = self.stacks_blocks_pool.get_canonical_fork_chain_tip() else {
+            return Err("StackerDB chunk received with no canonical chain tip".to_string());
+        };
         let chunks = stacks::standardize_stacks_marshalled_stackerdb_chunks(
-            &self.config,
             marshalled_stackerdb_chunks,
             receipt_time,
-            &mut self.stacks_context,
+            chain_tip,
             ctx,
         )?;
         if chunks.len() > 0 {
             Ok(Some(StacksChainEvent::ChainUpdatedWithStackerDbChunks(
-                StacksChainUpdatedWithStackerDbChunksData { chunks },
+                StacksChainUpdatedWithStackerDbChunksData {
+                    chunks,
+                    received_at: receipt_time,
+                    received_at_block: chain_tip.clone(),
+                },
             )))
         } else {
             Ok(None)

components/chainhook-sdk/src/indexer/stacks/blocks_pool.rs

@@ -49,6 +49,15 @@ impl StacksBlockPool {
         }
     }
 
+    pub fn get_canonical_fork_chain_tip(&self) -> Option<&BlockIdentifier> {
+        match self.forks.get(&self.canonical_fork_id) {
+            Some(fork) => {
+                Some(fork.get_tip())
+            },
+            None => None,
+        }
+    }
+
     pub fn seed_block_pool(&mut self, blocks: Vec<StacksBlockData>, ctx: &Context) {
         ctx.try_log(|logger| {
             slog::info!(logger, "Seeding block pool with {} blocks", blocks.len())

components/chainhook-sdk/src/indexer/stacks/mod.rs

@@ -593,22 +593,22 @@ pub fn standardize_stacks_microblock_trail(
 
 #[cfg(feature = "stacks-signers")]
 pub fn standardize_stacks_marshalled_stackerdb_chunks(
-    _indexer_config: &IndexerConfig,
     marshalled_stackerdb_chunks: JsonValue,
     receipt_time: u64,
-    _chain_ctx: &mut StacksChainContext,
+    chain_tip: &BlockIdentifier,
     _ctx: &Context,
 ) -> Result<Vec<StacksStackerDbChunk>, String> {
     let mut stackerdb_chunks: NewStackerDbChunks =
         serde_json::from_value(marshalled_stackerdb_chunks)
             .map_err(|e| format!("unable to parse stackerdb chunks {e}"))?;
-    standardize_stacks_stackerdb_chunks(&mut stackerdb_chunks, receipt_time)
+    standardize_stacks_stackerdb_chunks(&mut stackerdb_chunks, receipt_time, chain_tip)
 }
 
 #[cfg(feature = "stacks-signers")]
 pub fn standardize_stacks_stackerdb_chunks(
     stackerdb_chunks: &NewStackerDbChunks,
     receipt_time: u64,
+    chain_tip: &BlockIdentifier,
 ) -> Result<Vec<StacksStackerDbChunk>, String> {
     use stacks_codec::codec::BlockResponse;
     use stacks_codec::codec::RejectCode;
@@ -705,7 +705,8 @@ pub fn standardize_stacks_stackerdb_chunks(
             sig: slot.sig.clone(),
             pubkey: get_signer_pubkey_from_stackerdb_chunk_slot(slot, &data_bytes)?,
             message,
-            receipt_time,
+            received_at: receipt_time,
+            received_at_block: chain_tip.clone()
         });
     }