feat: Add common data aggregations to events (#13)

- add count of transactions to blocks - add total coin output to transactions - add input count to transactions - add output count to transactions - update terminal formatting output accordingly - update example ES mappings accordingly
txpipe · Dec 26, 2021 · 50052c3 · 50052c3
1 parent 5d00437
commit 50052c3
Show file tree

Hide file tree

Showing 4 changed files with 66 additions and 15 deletions.
diff --git a/src/framework.rs b/src/framework.rs
@@ -63,12 +63,16 @@ pub enum EventData {
     Block {
         body_size: usize,
         issuer_vkey: String,
+        tx_count: usize,
     },
     Transaction {
         fee: u64,
         ttl: Option<u64>,
         validity_interval_start: Option<u64>,
         network_id: Option<u32>,
+        input_count: usize,
+        output_count: usize,
+        total_output: u64,
     },
     TxInput {
         tx_id: String,

diff --git a/src/mapping.rs b/src/mapping.rs
@@ -210,14 +210,18 @@ impl EventSource for AuxiliaryData {
     }
 }
 
+fn get_tx_output_coin_value(amount: &Value) -> u64 {
+    match amount {
+        Value::Coin(x) => *x,
+        Value::Multiasset(x, _) => *x,
+    }
+}
+
 impl EventSource for TransactionOutput {
     fn write_events(&self, writer: &EventWriter) -> Result<(), Error> {
         writer.append(EventData::TxOutput {
             address: self.address.try_to_bech32("addr")?,
-            amount: match self.amount {
-                Value::Coin(x) => x,
-                Value::Multiasset(x, _) => x,
-            },
+            amount: get_tx_output_coin_value(&self.amount),
         })?;
 
         if let Value::Multiasset(_, assets) = &self.amount {
@@ -299,6 +303,9 @@ impl EventSource for TransactionBody {
         let mut ttl = None;
         let mut validity_interval_start = None;
         let mut network_id = None;
+        let mut total_output = 0;
+        let mut output_count = 0;
+        let mut input_count = 0;
 
         for component in self.iter() {
             match component {
@@ -313,6 +320,17 @@ impl EventSource for TransactionBody {
                 }
                 TransactionBodyComponent::NetworkId(NetworkId::One) => network_id = Some(1),
                 TransactionBodyComponent::NetworkId(NetworkId::Two) => network_id = Some(2),
+                TransactionBodyComponent::Outputs(outputs) => {
+                    output_count = outputs.len();
+
+                    total_output = outputs
+                        .iter()
+                        .map(|o| get_tx_output_coin_value(&o.amount))
+                        .fold(0u64, |prev, x| prev + x);
+                },
+                TransactionBodyComponent::Inputs(inputs) => {
+                    input_count = inputs.len();
+                }
                 // TODO
                 // TransactionBodyComponent::ScriptDataHash(_) => todo!(),
                 // TransactionBodyComponent::RequiredSigners(_) => todo!(),
@@ -326,6 +344,9 @@ impl EventSource for TransactionBody {
             ttl,
             validity_interval_start,
             network_id,
+            total_output,
+            output_count,
+            input_count,
         })?;
 
         // write sub-events
@@ -349,6 +370,7 @@ impl EventSource for Block {
         writer.append(EventData::Block {
             body_size: self.header.header_body.block_body_size as usize,
             issuer_vkey: self.header.header_body.issuer_vkey.to_hex(),
+            tx_count: self.transaction_bodies.len(),
         })?;
 
         for (idx, tx) in self.transaction_bodies.iter().enumerate() {

diff --git a/src/sinks/terminal/format.rs b/src/sinks/terminal/format.rs
@@ -18,25 +18,35 @@ impl LogLine {
             EventData::Block {
                 body_size,
                 issuer_vkey,
-            } => LogLine {
-                prefix: "BLOCK",
-                color: Color::Magenta,
-                content: format!(
-                    "{{ hash: {}, body size: {}, issues vkey: {}, timestamp: {} }}",
+                tx_count,
+                ..
+            } => {
+                LogLine {
+                    prefix: "BLOCK",
+                    color: Color::Magenta,
+                    content: format!(
+                    "{{ hash: {}, body size: {}, tx_count: {}, issuer vkey: {}, timestamp: {} }}",
                     &source.context.block_hash.as_ref().unwrap_or(&"".to_string()),
                     body_size,
+                    tx_count,
                     issuer_vkey,
                     source.context.timestamp.unwrap_or_default(),
                 ),
-                source,
-                max_width,
-            },
-            EventData::Transaction { fee, ttl, .. } => LogLine {
+                    source,
+                    max_width,
+                }
+            }
+            EventData::Transaction {
+                total_output,
+                fee,
+                ttl,
+                ..
+            } => LogLine {
                 prefix: "TX",
                 color: Color::DarkBlue,
                 content: format!(
-                    "{{ fee: {}, hash: {:?}, ttl: {:?} }}",
-                    fee, &source.context.tx_hash, ttl
+                    "{{ total_output: {}, fee: {}, hash: {:?}, ttl: {:?} }}",
+                    total_output, fee, &source.context.tx_hash, ttl
                 ),
                 source,
                 max_width,

diff --git a/testdrive/elasticsearch/index-template.json b/testdrive/elasticsearch/index-template.json
@@ -7,6 +7,9 @@
                 },
                 "block": {
                     "properties": {
+                        "tx_count": {
+                            "type": "long"
+                        },
                         "body_size": {
                             "type": "long"
                         },
@@ -30,6 +33,9 @@
                         "block_number": {
                             "type": "long"
                         },
+                        "block_hash": {
+                            "type": "keyword"
+                        },
                         "input_idx": {
                             "type": "long"
                         },
@@ -149,6 +155,15 @@
                         },
                         "validity_interval_start": {
                             "type": "long"
+                        },
+                        "input_count": {
+                            "type": "long"
+                        },
+                        "output_count": {
+                            "type": "long"
+                        },
+                        "total_output": {
+                            "type": "long"
                         }
                     }
                 },