feat(sink): update SqlDb sink to support cbor blocks and transactions

Author: Mercurial

examples/postgresql/cursor

@@ -0,0 +1,42 @@
+[
+  [
+    60485013,
+    "043018656b71c25c2a0338fe7000a53dedc704c5688e0c90a6093f08178a8b36"
+  ],
+  [
+    60485009,
+    "80669929ff82d23e5dd097d1554541bc678d53322df6c7bfec13c62645c68737"
+  ],
+  [
+    60485003,
+    "0f2f548b4360e3553923cc0fd96b9e10ec5142bc43f93e0a0a5509be7fd4aa20"
+  ],
+  [
+    60484990,
+    "4fc80c10ec326b1d1229f3284c8b5d73c442524a95da91727f1f5684b8e58484"
+  ],
+  [
+    60484982,
+    "f6c8a7b2b69b47da827ff255cf92e282493c70e64587ab233426b5b604939ca5"
+  ],
+  [
+    60484942,
+    "15989755ba2c52d302b197771c076491ea15422a83f08f66c90e25ed1a73bb78"
+  ],
+  [
+    60484936,
+    "d6e485277c441e6868fd742c6a3e57a4ca1de30b4e8c619624c054352f0b83a2"
+  ],
+  [
+    60484917,
+    "40844b925abd2004c70694eebd6da02a6ec4c3eca51aba32314364d52679a75b"
+  ],
+  [
+    60484876,
+    "43ae8400494cff3fdde99a7ca780d5b4a216116b44f4b39cc4f24f735eb70570"
+  ],
+  [
+    60484869,
+    "4595260047133e4da97d2939c4edf6087621e6b72380c2179b35d6df35520ab4"
+  ]
+]

examples/postgresql/daemon.toml

@@ -1,17 +1,27 @@
 [source]
 type = "N2N"
-peers = ["relays-new.cardano-mainnet.iohk.io:3001"]
+peers = ["preprod-node.play.dev.cardano.org:3001"]
+
+[chain]
+type = "preprod"
 
 [intersect]
 type = "Point"
-value = [114220807, "01822b10afde3d09bd5e72758857b669ddadcdaad6776a3dc8ee902c3ace1d7e"]
+value = [60484568, "874858059dc6c871105b80b38238cf84a91e617b2afde53904407cb2a3299e9d"]
 
 [[filters]]
-type = "SplitBlock"
+type = "EmitCbor"
 
 [sink]
 type = "SqlDb"
-connection = "postgres://postgres:example@localhost:5432/postgres"
-apply_template = "INSERT INTO txs (slot, cbor) VALUES ('{{point.slot}}', decode('{{record.hex}}', 'hex'));"
-undo_template = "DELETE FROM txs WHERE slot = {{point.slot}}"
-reset_template = "DELETE FROM txs WHERE slot > {{point.slot}}"
+connection = "postgres://postgres:test1234@localhost:15432/oura_test"
+apply_cbor_block_template = "INSERT INTO blocks (slot, cbor) VALUES ('{{point.slot}}', decode('{{record.hex}}', 'hex'));"
+undo_cbor_block_template = "DELETE FROM blocks WHERE slot = {{point.slot}}"
+apply_cbor_tx_template = "INSERT INTO txs (slot, cbor) VALUES ('{{point.slot}}', decode('{{record.hex}}', 'hex'));"
+undo_cbor_tx_template = "DELETE FROM txs WHERE slot = {{point.slot}}"
+reset_cbor_block_template = "DELETE FROM blocks WHERE slot > {{point.slot}};"
+reset_cbor_tx_template = "DELETE FROM txs WHERE slot > {{point.slot}};"
+
+[cursor]
+type = "File"
+path = "cursor"

examples/postgresql/init.sql

@@ -1,6 +1,17 @@
+-- Table for storing CBOR blocks
+CREATE TABLE blocks (
+    slot INTEGER NOT NULL,
+    cbor BYTEA
+);
+
+-- Index for the blocks table
+CREATE INDEX idx_blocks_slot ON blocks(slot);
+
+-- Table for storing CBOR transactions
 CREATE TABLE txs (
     slot INTEGER NOT NULL,
     cbor BYTEA
 );
 
-CREATE INDEX idx_txs_slot ON txs(slot);
+-- Index for the txs table
+CREATE INDEX idx_txs_slot ON txs(slot);

src/sinks/sql_db.rs

@@ -46,15 +46,24 @@ impl gasket::framework::Worker<Stage> for Worker {
         let template = match unit {
             ChainEvent::Apply(p, r) => {
                 let data = hbs_data(p.clone(), Some(r.clone()));
-                stage.templates.render("apply", &data)
+                match r {
+                    Record::CborBlock(_) => stage.templates.render("apply_cbor_block", &data),
+                    Record::CborTx(_) => stage.templates.render("apply_cbor_tx", &data),
+                    _ => stage.templates.render("apply", &data),
+                }
             }
             ChainEvent::Undo(p, r) => {
                 let data = hbs_data(p.clone(), Some(r.clone()));
-                stage.templates.render("undo", &data)
+                match r {
+                    Record::CborBlock(_) => stage.templates.render("undo_cbor_block", &data),
+                    Record::CborTx(_) => stage.templates.render("undo_cbor_tx", &data),
+                    _ => stage.templates.render("undo", &data),
+                }
             }
             ChainEvent::Reset(p) => {
                 let data = hbs_data(p.clone(), None);
-                stage.templates.render("reset", &data)
+                stage.templates.render("reset_cbor_block", &data).ok();
+                stage.templates.render("reset_cbor_tx", &data)
             }
         };
 
@@ -91,9 +100,12 @@ pub struct Stage {
 pub struct Config {
     /// eg: sqlite::memory:
     pub connection: String,
-    pub apply_template: String,
-    pub undo_template: String,
-    pub reset_template: String,
+    pub apply_cbor_block_template: String,
+    pub undo_cbor_block_template: String,
+    pub apply_cbor_tx_template: String,
+    pub undo_cbor_tx_template: String,
+    pub reset_cbor_block_template: String,
+    pub reset_cbor_tx_template: String,
 }
 
 impl Config {
@@ -103,15 +115,27 @@ impl Config {
         let mut templates = handlebars::Handlebars::new();
 
         templates
-            .register_template_string("apply", &self.apply_template)
+            .register_template_string("apply_cbor_block", &self.apply_cbor_block_template)
             .map_err(Error::config)?;
 
         templates
-            .register_template_string("undo", &self.undo_template)
+            .register_template_string("undo_cbor_block", &self.undo_cbor_block_template)
             .map_err(Error::config)?;
 
         templates
-            .register_template_string("reset", &self.reset_template)
+            .register_template_string("apply_cbor_tx", &self.apply_cbor_tx_template)
+            .map_err(Error::config)?;
+
+        templates
+            .register_template_string("undo_cbor_tx", &self.undo_cbor_tx_template)
+            .map_err(Error::config)?;
+
+        templates
+            .register_template_string("reset_cbor_block", &self.reset_cbor_block_template)
+            .map_err(Error::config)?;
+
+        templates
+            .register_template_string("reset_cbor_tx", &self.reset_cbor_tx_template)
             .map_err(Error::config)?;
 
         let stage = Stage {