refactor: scaffold multi-chain pipeline

Cargo.toml

@@ -20,6 +20,7 @@ zeromq = ["zmq"]
 u5c = ["utxorpc", "futures"]
 mithril = ["mithril-client"]
 hydra = ["tungstenite", "tokio-tungstenite", "futures-util", "bytes"]
+eth = ["futures-util"]
 # elasticsearch = auto feature flag
 # kafka = auto feature flag
 
@@ -83,6 +84,9 @@ futures-util = { version = "0.3", optional = true }
 bytes = { version = "1.7.2", optional = true }
 zmq = { version = "0.10.0", optional = true }
 
+# TODO(p): add feature
+alloy = { version = "1.0.30", features = ["full"] }
+
 [dev-dependencies]
 goldenfile = "1.7.3"
 tempfile = "3.4"

src/bin/oura/dump.rs

@@ -2,7 +2,7 @@ use clap::{Parser, ValueEnum};
 use oura::{
     daemon::{run_daemon, ConfigRoot},
     filters,
-    framework::{ChainConfig, Error, IntersectConfig},
+    framework::{cardano, Chain, Error, IntersectConfig},
     sinks, sources,
 };
 use tracing::{info, Level};
@@ -19,7 +19,7 @@ pub fn run(args: &Args) -> Result<(), Error> {
         .with(env_filter)
         .init();
 
-    let chain = args.magic.clone().unwrap_or_default().into();
+    let chain = Chain::Cardano(args.magic.clone().unwrap_or_default().into());
     let intersect = parse_since(args.since.clone())?;
     let bearer = args.bearer.clone().unwrap_or_default();
     let source = match bearer {
@@ -95,7 +95,7 @@ pub struct Args {
     bearer: Option<Bearer>,
 
     #[arg(long)]
-    magic: Option<Chain>,
+    magic: Option<DumpChain>,
 
     /// point in the chain to start reading from, expects format `slot,hex-hash`
     #[arg(long)]
@@ -115,20 +115,21 @@ enum Bearer {
 }
 
 #[derive(ValueEnum, Clone, Default)]
-enum Chain {
+enum DumpChain {
     #[default]
     Mainnet,
     Testnet,
     PreProd,
     Preview,
 }
-impl From<Chain> for ChainConfig {
-    fn from(value: Chain) -> Self {
+// TODO(p): add support multchain
+impl From<DumpChain> for cardano::ChainConfig {
+    fn from(value: DumpChain) -> Self {
         match value {
-            Chain::Mainnet => ChainConfig::Mainnet,
-            Chain::Testnet => ChainConfig::Testnet,
-            Chain::PreProd => ChainConfig::PreProd,
-            Chain::Preview => ChainConfig::Preview,
+            DumpChain::Mainnet => cardano::ChainConfig::Mainnet,
+            DumpChain::Testnet => cardano::ChainConfig::Testnet,
+            DumpChain::PreProd => cardano::ChainConfig::PreProd,
+            DumpChain::Preview => cardano::ChainConfig::Preview,
         }
     }
 }

src/bin/oura/watch.rs

@@ -2,7 +2,7 @@ use clap::{Parser, ValueEnum};
 use oura::{
     daemon::{run_daemon, ConfigRoot},
     filters,
-    framework::{ChainConfig, Error, IntersectConfig},
+    framework::{cardano, Chain, Error, IntersectConfig},
     sinks, sources,
 };
 use tracing::{info, Level};
@@ -19,7 +19,7 @@ pub fn run(args: &Args) -> Result<(), Error> {
         .with(env_filter)
         .init();
 
-    let chain = args.magic.clone().unwrap_or_default().into();
+    let chain = Chain::Cardano(args.magic.clone().unwrap_or_default().into());
     let intersect = parse_since(args.since.clone())?;
     let bearer = args.bearer.clone().unwrap_or_default();
     let source = match bearer {
@@ -92,7 +92,7 @@ pub struct Args {
     bearer: Option<Bearer>,
 
     #[arg(long)]
-    magic: Option<Chain>,
+    magic: Option<WatchChain>,
 
     /// point in the chain to start reading from, expects format `slot,hex-hash`
     #[arg(long)]
@@ -116,20 +116,20 @@ pub enum Bearer {
 }
 
 #[derive(ValueEnum, Clone, Default)]
-pub enum Chain {
+pub enum WatchChain {
     #[default]
     Mainnet,
     Testnet,
     PreProd,
     Preview,
 }
-impl From<Chain> for ChainConfig {
-    fn from(value: Chain) -> Self {
+impl From<WatchChain> for cardano::ChainConfig {
+    fn from(value: WatchChain) -> Self {
         match value {
-            Chain::Mainnet => ChainConfig::Mainnet,
-            Chain::Testnet => ChainConfig::Testnet,
-            Chain::PreProd => ChainConfig::PreProd,
-            Chain::Preview => ChainConfig::Preview,
+            WatchChain::Mainnet => cardano::ChainConfig::Mainnet,
+            WatchChain::Testnet => cardano::ChainConfig::Testnet,
+            WatchChain::PreProd => cardano::ChainConfig::PreProd,
+            WatchChain::Preview => cardano::ChainConfig::Preview,
         }
     }
 }

src/daemon/mod.rs

@@ -16,7 +16,7 @@ pub struct ConfigRoot {
     pub sink: sinks::Config,
     pub intersect: IntersectConfig,
     pub finalize: Option<FinalizeConfig>,
-    pub chain: Option<ChainConfig>,
+    pub chain: Option<Chain>,
     pub retries: Option<gasket::retries::Policy>,
     pub cursor: Option<cursor::Config>,
     pub metrics: Option<MetricsConfig>,

src/filters/into_json.rs

@@ -26,7 +26,11 @@ impl From<&Stage> for Worker {
 }
 
 gasket::impl_mapper!(|_worker: Worker, stage: Stage, unit: ChainEvent| => {
-    let out = unit.clone().map_record(|r| Record::GenericJson(JsonValue::from(r)));
+    let out = unit.clone().map_record(|r| match r {
+        Record::Cardano(record) => Record::GenericJson(JsonValue::from(record)),
+        x => x,
+    });
+
     stage.ops_count.inc(1);
     out
 });

src/filters/legacy_v1/cip15.rs

@@ -4,7 +4,7 @@ use serde_json::Value as JsonValue;
 use pallas::ledger::primitives::alonzo::Metadatum;
 
 use super::EventWriter;
-use crate::framework::legacy_v1::*;
+use crate::framework::cardano::legacy_v1::*;
 
 fn extract_json_property<'a>(json: &'a JsonValue, key: &str) -> Option<&'a JsonValue> {
     json.as_object().and_then(|x| x.get(key))

src/filters/legacy_v1/cip25.rs

@@ -3,7 +3,7 @@ use pallas::ledger::primitives::alonzo::Metadatum;
 use serde_json::Value as JsonValue;
 use tracing::warn;
 
-use crate::framework::legacy_v1::CIP25AssetRecord;
+use crate::framework::cardano::legacy_v1::CIP25AssetRecord;
 
 use super::EventWriter;
 

src/filters/legacy_v1/crawl.rs

@@ -3,7 +3,7 @@ use pallas::ledger::primitives::babbage::MintedDatumOption;
 use pallas::ledger::traverse::{MultiEraBlock, MultiEraInput, MultiEraOutput, MultiEraTx};
 use pallas::network::miniprotocols::Point;
 
-use crate::framework::legacy_v1::*;
+use crate::framework::cardano::legacy_v1::*;
 use crate::framework::Error as OuraError;
 
 use super::EventWriter;

src/filters/legacy_v1/map.rs

@@ -21,7 +21,7 @@ use pallas::ledger::traverse::{
 use pallas::network::miniprotocols::Point;
 use pallas::{codec::utils::KeepRaw, crypto::hash::Hash};
 
-use crate::framework::legacy_v1::*;
+use crate::framework::cardano::legacy_v1::*;
 
 use super::EventWriter;
 

src/filters/legacy_v1/mod.rs

@@ -39,7 +39,7 @@ gasket::impl_splitter!(|_worker: Worker, stage: Stage, unit: ChainEvent| => {
     let mut buffer = Vec::new();
 
     match unit {
-        ChainEvent::Apply(point, Record::CborBlock(cbor)) => {
+        ChainEvent::Apply(point, Record::Cardano(cardano::Record::CborBlock(cbor))) => {
             let mut writer = EventWriter::new(
                 point.clone(),
                 &stage.output,
@@ -89,9 +89,13 @@ pub struct Config {
 
 impl Config {
     pub fn bootstrapper(self, ctx: &Context) -> Result<Stage, Error> {
+        let chain_config = match &ctx.chain {
+            Chain::Cardano(chain_config) => chain_config.clone(),
+        };
+
         let stage = Stage {
             config: self,
-            genesis: ctx.chain.clone().into(),
+            genesis: chain_config.into(),
             ops_count: Default::default(),
             input: Default::default(),
             output: Default::default(),

src/filters/legacy_v1/prelude.rs

@@ -1,4 +1,4 @@
-use crate::framework::legacy_v1::*;
+use crate::framework::cardano::legacy_v1::*;
 use crate::framework::*;
 
 use gasket::framework::WorkerError;
@@ -42,7 +42,10 @@ impl<'a> EventWriter<'a> {
             fingerprint: None,
         };
 
-        let msg = ChainEvent::Apply(self.point.clone(), Record::OuraV1Event(evt));
+        let msg = ChainEvent::Apply(
+            self.point.clone(),
+            Record::Cardano(cardano::Record::OuraV1Event(evt)),
+        );
         self.buffer.push(msg);
 
         Ok(())

src/filters/parse_cbor.rs

@@ -40,15 +40,15 @@ impl From<&Stage> for Worker {
 
 gasket::impl_mapper!(|_worker: Worker, stage: Stage, unit: ChainEvent| => {
     let output = unit.clone().try_map_record(|r| match r {
-        Record::CborBlock(cbor) => {
+        Record::Cardano(cardano::Record::CborBlock(cbor)) => {
             let block = trv::MultiEraBlock::decode(&cbor).or_panic()?;
             let block = stage.mapper.map_block(&block);
-            Ok(Record::ParsedBlock(block))
+            Ok(Record::Cardano(cardano::Record::ParsedBlock(block)))
         }
-        Record::CborTx(cbor) => {
+        Record::Cardano(cardano::Record::CborTx(cbor)) => {
             let tx = trv::MultiEraTx::decode(&cbor).or_panic()?;
             let tx = stage.mapper.map_tx(&tx);
-            Ok(Record::ParsedTx(tx))
+            Ok(Record::Cardano(cardano::Record::ParsedTx(tx)))
         }
         x => Ok(x),
     })?;

src/filters/select/eval/mod.rs

@@ -605,10 +605,16 @@ fn eval_block(block: &ParsedBlock, predicate: &Predicate) -> MatchOutcome {
 
 pub fn eval(record: &Record, predicate: &Predicate) -> MatchOutcome {
     match record {
-        Record::ParsedTx(x) => eval_tx(x, predicate),
-        Record::ParsedBlock(x) => eval_block(x, predicate),
+        Record::Cardano(record) => match record {
+            cardano::Record::ParsedTx(x) => eval_tx(x, predicate),
+            cardano::Record::ParsedBlock(x) => eval_block(x, predicate),
+            _ => {
+                warn!("The select filter is valid only for ParsedTx / ParsedBlock records");
+                MatchOutcome::Uncertain
+            }
+        },
         _ => {
-            warn!("The select filter is valid only with ParsedTx & ParsedBlock records");
+            warn!("The select filter is valid only for Cardano ParsedTx / ParsedBlock records");
             MatchOutcome::Uncertain
         }
     }

src/filters/split_block.rs

@@ -45,10 +45,10 @@ impl From<&Stage> for Worker {
 
 gasket::impl_splitter!(|_worker: Worker, stage: Stage, unit: ChainEvent| => {
     let output = unit.clone().try_map_record_to_many(|r| match r {
-        Record::CborBlock(cbor) => {
+        Record::Cardano(cardano::Record::CborBlock(cbor)) => {
             let out = map_block_to_tx(Cow::Borrowed(&cbor))?
                 .into_iter()
-                .map(|tx| Record::CborTx(tx.into()))
+                .map(|tx| Record::Cardano(cardano::Record::CborTx(tx.into())))
                 .collect();
 
             Ok(out)

src/filters/wasm_plugin.rs

@@ -5,6 +5,8 @@ use serde::Deserialize;
 
 use crate::framework::*;
 
+pub type CardanoRecord = cardano::Record;
+
 #[derive(Stage)]
 #[stage(name = "filter-wasm", unit = "ChainEvent", worker = "Worker")]
 pub struct Stage {
@@ -18,23 +20,20 @@ pub struct Stage {
 }
 
 impl Stage {
-    fn map_record(&mut self, r: Record) -> Result<Vec<Record>, Error> {
+    fn map_cardano_record(&mut self, r: CardanoRecord) -> Result<Vec<Record>, Error> {
         let extism::convert::Json::<serde_json::Value>(output) = match r {
-            Record::CborBlock(x) => self.plugin.call("map_cbor_block", x).unwrap(),
-            Record::CborTx(x) => self.plugin.call("map_cbor_tx", x).unwrap(),
-            Record::ParsedTx(x) => self
+            CardanoRecord::CborBlock(x) => self.plugin.call("map_cbor_block", x).unwrap(),
+            CardanoRecord::CborTx(x) => self.plugin.call("map_cbor_tx", x).unwrap(),
+            CardanoRecord::ParsedTx(x) => self
                 .plugin
                 .call("map_u5c_tx", extism::convert::Json(x))
                 .unwrap(),
-            Record::ParsedBlock(x) => self
+            CardanoRecord::ParsedBlock(x) => self
                 .plugin
                 .call("map_u5c_block", extism::convert::Json(x))
                 .unwrap(),
-            Record::GenericJson(x) => self
-                .plugin
-                .call("map_json", extism::convert::Json(x))
-                .unwrap(),
-            Record::OuraV1Event(x) => self
+
+            CardanoRecord::OuraV1Event(x) => self
                 .plugin
                 .call("map_json", extism::convert::Json(x))
                 .unwrap(),
@@ -48,6 +47,13 @@ impl Stage {
 
         Ok(output)
     }
+
+    fn map_record(&mut self, r: Record) -> Result<Vec<Record>, Error> {
+        match r {
+            Record::Cardano(x) => self.map_cardano_record(x),
+            x => Ok(vec![x]),
+        }
+    }
 }
 
 #[derive(Default)]

src/framework/bitcoin/mod.rs

@@ -0,0 +1,17 @@
+use serde_json::{json, Value as JsonValue};
+
+#[derive(Debug, Clone)]
+pub enum Record {
+    // Scaffold placeholder for now
+    ParsedBlock(()),
+    RawBlock(Vec<u8>),
+}
+
+impl From<Record> for JsonValue {
+    fn from(value: Record) -> Self {
+        match value {
+            Record::ParsedBlock(x) => json!(x),
+            Record::RawBlock(x) => json!({ "hex": hex::encode(x) }),
+        }
+    }
+}