Refactor consensus_param_updates + use array for block/events attrs

blocks/cosmos/src/consensus_param_updates.rs

@@ -8,41 +8,18 @@ pub fn collect_consensus_params(block: &Block, timestamp: &BlockTimestamp) -> Ve
     let mut vec: Vec<ConsensusParamUpdate> = vec![];
 
     if let Some(consensus_params) = &block.consensus_param_updates {
-        let mut json = serde_json::json!({});
-
-        if let Some(block_params) = &consensus_params.block {
-            json["block"] = serde_json::json!({
-                "max_bytes": block_params.max_bytes,
-                "max_gas": block_params.max_gas,
-            });
-        }
-
-        if let Some(evidence_params) = &consensus_params.evidence {
-            json["evidence"] = serde_json::json!({
-                "max_age_num_blocks": evidence_params.max_age_num_blocks,
-                "max_age_duration": evidence_params.max_age_duration.as_ref().map(|d| d.to_string()),
-                "max_bytes": evidence_params.max_bytes
-            });
-        }
-
-        if let Some(validator_params) = &consensus_params.validator {
-            json["validator"] = serde_json::json!({
-                "pub_key_types": validator_params.pub_key_types
-            });
-        }
-
-        if let Some(version_params) = &consensus_params.version {
-            json["version"] = serde_json::json!({
-                "app_version": version_params.app
-            });
-        }
-
         vec.push(ConsensusParamUpdate {
             block_time: Some(timestamp.time),
             block_number: timestamp.number,
             block_date: timestamp.date.clone(),
             block_hash: Hex::encode(&block.hash),
-            json: json.to_string(),
+            block_max_bytes: consensus_params.block.as_ref().map(|b| b.max_bytes),
+            block_max_gas: consensus_params.block.as_ref().map(|b| b.max_gas),
+            evidence_max_age_num_blocks: consensus_params.evidence.as_ref().map(|e| e.max_age_num_blocks),
+            evidence_max_age_duration: consensus_params.evidence.as_ref().and_then(|e| e.max_age_duration.as_ref().map(|d| d.to_string())),
+            evidence_max_bytes: consensus_params.evidence.as_ref().map(|e| e.max_bytes),
+            validator_pub_key_types: consensus_params.validator.as_ref().map(|v| v.pub_key_types.clone()).unwrap_or_default(),
+            app_version: consensus_params.version.as_ref().map(|v| v.app),
         });
     }
 

blocks/cosmos/src/pb/pinax.cosmos.rs

@@ -112,8 +112,8 @@ pub struct TransactionEvent {
     #[prost(string, tag="7")]
     pub r#type: ::prost::alloc::string::String,
     /// Should be Array(Tuple(Text, Text)) when supported by sink-files
-    #[prost(string, tag="8")]
-    pub attributes: ::prost::alloc::string::String,
+    #[prost(string, repeated, tag="8")]
+    pub attributes: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -130,8 +130,9 @@ pub struct BlockEvent {
     pub index: u32,
     #[prost(string, tag="6")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(string, tag="7")]
-    pub attributes: ::prost::alloc::string::String,
+    /// Should be Array(Tuple(Text, Text)) when supported by sink-files
+    #[prost(string, repeated, tag="7")]
+    pub attributes: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -188,8 +189,20 @@ pub struct ConsensusParamUpdate {
     pub block_date: ::prost::alloc::string::String,
     #[prost(string, tag="4")]
     pub block_hash: ::prost::alloc::string::String,
-    #[prost(string, tag="5")]
-    pub json: ::prost::alloc::string::String,
+    #[prost(int64, optional, tag="5")]
+    pub block_max_bytes: ::core::option::Option<i64>,
+    #[prost(int64, optional, tag="6")]
+    pub block_max_gas: ::core::option::Option<i64>,
+    #[prost(int64, optional, tag="7")]
+    pub evidence_max_age_num_blocks: ::core::option::Option<i64>,
+    #[prost(string, optional, tag="8")]
+    pub evidence_max_age_duration: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(int64, optional, tag="9")]
+    pub evidence_max_bytes: ::core::option::Option<i64>,
+    #[prost(string, repeated, tag="10")]
+    pub validator_pub_key_types: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    #[prost(uint64, optional, tag="11")]
+    pub app_version: ::core::option::Option<u64>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

blocks/cosmos/src/tx_and_block_events.rs

@@ -3,7 +3,7 @@ use substreams_cosmos::{pb::TxResults, Block};
 
 use crate::{
     pb::pinax::cosmos::{BlockEvent, TransactionEvent},
-    utils::build_attributes_array_string,
+    utils::build_attributes_array,
 };
 
 pub fn collect_block_events(block: &Block, timestamp: &BlockTimestamp) -> Vec<BlockEvent> {
@@ -17,7 +17,7 @@ pub fn collect_block_events(block: &Block, timestamp: &BlockTimestamp) -> Vec<Bl
             block_hash: timestamp.hash.clone(),
             index: index as u32,
             r#type: event.r#type.clone(),
-            attributes: build_attributes_array_string(&event.attributes),
+            attributes: build_attributes_array(&event.attributes),
         });
     }
 
@@ -36,7 +36,7 @@ pub fn collect_transaction_events(tx_result: &TxResults, tx_hash: &str, timestam
             index: event_index as u32,
             tx_hash: tx_hash.to_string(),
             r#type: event.r#type.clone(),
-            attributes: build_attributes_array_string(&event.attributes),
+            attributes: build_attributes_array(&event.attributes),
         });
     }
 

blocks/cosmos/src/utils.rs

@@ -10,8 +10,9 @@ pub fn compute_tx_hash(tx_as_bytes: &[u8]) -> String {
     return Hex::encode(tx_hash);
 }
 
-// Builds a string in the format of an array of tuples (key, value)
-pub fn build_attributes_array_string(attributes: &[EventAttribute]) -> String {
-    let tuples: Vec<(&str, &str)> = attributes.iter().map(|attr| (attr.key.as_str(), attr.value.as_str())).collect();
-    serde_json::to_string(&tuples).unwrap_or_default()
+pub fn build_attributes_array(attributes: &[EventAttribute]) -> Vec<String> {
+    attributes
+        .iter()
+        .map(|attr| format!("\"{key}\":\"{value}\"", key = attr.key, value = serde_json::to_string(&attr.value).unwrap_or_default()))
+        .collect()
 }

blocks/cosmos/substreams.yaml

@@ -25,4 +25,4 @@ modules:
       - source: sf.substreams.v1.Clock
       - source: sf.cosmos.type.v2.Block
     output:
-      type: proto:cosmos.Events
+      type: proto:pinax.cosmos.Events

proto/cosmos.proto

@@ -63,7 +63,7 @@ message TransactionEvent {
   uint32 index = 6;
   string type = 7;
   // Should be Array(Tuple(Text, Text)) when supported by sink-files
-  string attributes = 8;
+  repeated string attributes = 8;
 }
 
 message BlockEvent {
@@ -73,7 +73,8 @@ message BlockEvent {
   string block_hash = 4;
   uint32 index = 5;
   string type = 6;
-  string attributes = 7;
+  // Should be Array(Tuple(Text, Text)) when supported by sink-files
+  repeated string attributes = 7;
 }
 
 
@@ -106,7 +107,13 @@ message ConsensusParamUpdate {
   uint64 block_number = 2;
   string block_date = 3;
   string block_hash = 4;
-  string json = 5;
+  optional int64 block_max_bytes = 5;
+  optional int64 block_max_gas = 6;
+  optional int64 evidence_max_age_num_blocks = 7;
+  optional string evidence_max_age_duration = 8;
+  optional int64 evidence_max_bytes = 9;
+  repeated string validator_pub_key_types = 10;
+  optional uint64 app_version = 11;
 }
 
 message TransactionMessage {