feat: introduce a Substrate source (#913)

Author: mduthey

Cargo.lock

@@ -22,6 +22,7 @@ mithril = ["mithril-client"]
 hydra = ["tungstenite", "tokio-tungstenite", "futures-util", "bytes"]
 eth = ["futures-util", "alloy"]
 btc = ["bitcoind-async-client", "corepc-types"]
+substrate = ["futures-util", "subxt"]
 # elasticsearch = auto feature flag
 # kafka = auto feature flag
 
@@ -87,6 +88,7 @@ zmq = { version = "0.10.0", optional = true }
 bitcoind-async-client = { version = "0.6.0", optional = true }
 corepc-types = { version = "0.10.1", optional = true }
 alloy = { version = "1.0.38", features = ["provider-ws"], optional = true }
+subxt = { version = "0.44.0", optional = true }
 
 [dev-dependencies]
 goldenfile = "1.7.3"

Cargo.toml

@@ -42,7 +42,8 @@ The data pipeline is implemented by the [Gasket](https://github.com/construkts/g
 You can run:
 - `oura watch cardano <socket>` to print Cardano transaction data into the terminal from the tip of a local or remote node
 - `oura watch bitcoin <rpc_host>` to watch Bitcoin blockchain events
-- `oura watch ethereum <url>` to watch Ethereum blockchain events via WebSocket.
+- `oura watch ethereum <url>` to watch Ethereum blockchain events via WebSocket
+- `oura watch substrate <url>` to watch Substrate blockchain events from any Substrate-based network
 
 These commands are useful as debugging tools for developers or if you're just curious to see what's going on in the network (for example, to see airdrops as they happen or oracles posting new information).
 

README.md

@@ -27,7 +27,8 @@ The data pipeline makes heavy use (maybe a bit too much) of multi-threading and
 You can run: 
 - `oura watch cardano <socket>` to print Cardano transaction data into the terminal from the tip of a local or remote node
 - `oura watch bitcoin <rpc_host>` to watch Bitcoin blockchain events
-- `oura watch ethereum <url>` to watch Ethereum blockchain events via WebSocket.
+- `oura watch ethereum <url>` to watch Ethereum blockchain events via WebSocket
+- `oura watch substrate <url>` to watch Substrate blockchain events from any Substrate-based network
 
 These commands are useful as debugging tools for developers or if you're just curious to see what's going on in the network (for example, to see airdrops as they happen or oracles posting new information).
 

docs/v3/index.mdx

@@ -0,0 +1,163 @@
+---
+title: Substrate RPC
+---
+
+The Substrate RPC source connects to a Substrate-based blockchain node via WebSocket RPC to subscribe to new blocks and retrieve complete block data. It works with any Substrate-based network including Polkadot, Kusama, and parachains.
+
+## Configuration
+
+The following snippet shows an example of how to set up a Substrate WebSocket source:
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "ws://localhost:9944"
+```
+
+### Section `source`:
+
+- `type`: this field must be set to the literal value `substrate-rpc`.
+- `url`: the WebSocket URL of the Substrate RPC endpoint (required). Must use `ws://` or `wss://` protocol.
+
+## Supported Networks
+
+The Substrate source works with any Substrate-based blockchain, including:
+
+- **Polkadot** - The main relay chain
+- **Kusama** - Polkadot's canary network
+- **Westend** - Polkadot's testnet
+- **Parachains** - Any parachain connected to Polkadot or Kusama
+- **Solo chains** - Independent Substrate-based blockchains
+- **Local development nodes** - For testing and development
+
+## Examples
+
+### Local Substrate Node
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "ws://localhost:9944"
+```
+
+### Polkadot Mainnet via OnFinality
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "wss://polkadot.api.onfinality.io/public-ws"
+```
+
+### Kusama Network
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "wss://kusama-rpc.polkadot.io"
+```
+
+### Westend Testnet
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "wss://westend-rpc.polkadot.io"
+```
+
+### Astar Parachain
+
+```toml
+[source]
+type = "substrate-rpc"
+url = "wss://rpc.astar.network"
+```
+
+## How It Works
+
+The Substrate source operates using Subxt library for Substrate interaction:
+
+1. **Connection**: Connects to the Substrate node via WebSocket using the Subxt client.
+
+2. **Subscription**: Subscribes to the best block stream to receive notifications when new blocks are finalized.
+
+3. **Block Processing**: For each new block:
+   - Retrieves the complete block header information
+   - Parses all extrinsics (transactions) in the block
+   - Extracts signature information for signed extrinsics
+   - Calculates block and extrinsic hashes
+
+4. **Event Generation**: Converts the block data into Oura's internal `ChainEvent::Apply` format for processing by filters and sinks.
+
+## Output Events
+
+The Substrate source produces `ChainEvent::Apply` events containing parsed Substrate block data. Each event includes:
+
+### Block Header
+- Block number and hash
+- Parent block hash
+- State root and extrinsics root
+- Block timestamp and other metadata
+
+### Extrinsics
+For each extrinsic in the block:
+- Extrinsic index and hash
+- Raw bytes (hex-encoded)
+- Signature information (for signed extrinsics):
+  - Signer address
+  - Signature data
+  - Signed extensions metadata
+
+### Additional Information
+- Total extrinsics count
+- Block processing metrics
+- Chain tip information
+
+## Data Structure
+
+The parsed block follows this structure:
+
+```rust
+ParsedBlock {
+    header: BlockHeader {
+        number: u64,
+        hash: String,
+        parent_hash: String,
+        state_root: String,
+        extrinsics_root: String,
+    },
+    extrinsics: Vec<Extrinsic>,
+    extrinsics_count: usize,
+}
+
+Extrinsic {
+    index: u32,
+    hash: String,
+    bytes: Option<String>,
+    signature: Option<SignatureInfo>,
+}
+
+SignatureInfo {
+    address: String,
+    signature: String,
+    extra: Option<String>,
+}
+```
+
+## Substrate Runtime Compatibility
+
+The source is designed to work with any Substrate runtime without requiring specific runtime metadata knowledge. It focuses on the core block structure that is consistent across all Substrate chains:
+
+- Block headers follow the standard Substrate format
+- Extrinsics use the SCALE codec for encoding
+- Signature extraction works with standard Substrate transaction formats
+
+This makes the source compatible with both current and future Substrate networks without requiring updates for new runtime versions.
+
+## Performance Considerations
+
+- The source subscribes to the best blocks only, ensuring reliable block ordering
+- Block processing is done asynchronously to maintain good throughput
+- Extrinsic parsing includes detailed signature extraction which may add some processing overhead
+- WebSocket connections are automatically managed and reconnected if needed
+
+The events follow the same pattern as other Oura sources, allowing them to be processed by standard filters and sinks in the pipeline.

docs/v3/sources/substrate.mdx

@@ -2,7 +2,7 @@
 title: Watch
 ---
 
-The `watch` mode provides a quick way to tail the latest events from blockchain networks. It supports Cardano, Bitcoin, and Ethereum blockchains, each with their own subcommands and connection methods.
+The `watch` mode provides a quick way to tail the latest events from blockchain networks. It supports Cardano, Bitcoin, Ethereum, and Substrate blockchains, each with their own subcommands and connection methods.
 
 The output is colorized by type of event and dynamically truncated to fit the width of the terminal. The speed of the output lines is throttled to facilitate visual inspection of each event, otherwise, all events for a block would be output simultaneously.
 
@@ -61,6 +61,21 @@ oura watch ethereum [OPTIONS] <url>
 - `--throttle`: milliseconds to wait between output lines (for easier reading).
 - `--wrap`: indicates that long output text should break and continue in the following line.
 
+### Substrate
+
+To watch Substrate blockchain events:
+
+```
+oura watch substrate [OPTIONS] <url>
+```
+
+- `<url>`: the Substrate WebSocket RPC URL to connect to.
+
+#### Substrate Options
+
+- `--throttle`: milliseconds to wait between output lines (for easier reading).
+- `--wrap`: indicates that long output text should break and continue in the following line.
+
 ## Examples
 
 ### Cardano Examples
@@ -148,3 +163,35 @@ oura watch ethereum ws://localhost:8546 --throttle 1000 --wrap
 ```sh
 oura watch ethereum wss://ethereum-sepolia-rpc.publicnode.com
 ```
+
+### Substrate Examples
+
+#### Watch Live Data From A Local Polkadot Node
+
+```sh
+oura watch substrate ws://localhost:9944
+```
+
+#### Watch Polkadot Mainnet via OnFinality
+
+```sh
+oura watch substrate wss://polkadot.api.onfinality.io/public-ws
+```
+
+#### Watch Substrate With Output Throttling
+
+```sh
+oura watch substrate ws://localhost:9944 --throttle 1000 --wrap
+```
+
+#### Watch Kusama Network
+
+```sh
+oura watch substrate wss://kusama-rpc.polkadot.io
+```
+
+#### Watch Westend Testnet
+
+```sh
+oura watch substrate wss://westend-rpc.polkadot.io
+```

docs/v3/usage/watch.mdx

@@ -0,0 +1,15 @@
+[source]
+type = "substrate-rpc"
+url = "wss://polkadot-rpc.publicnode.com"
+
+[intersect]
+type = "Tip"
+
+[sink]
+# type = "Stdout"
+type = "FileRotate"
+output_path = "/tmp/oura-substrate/mainnet"
+output_format = "JSONL"
+max_bytes_per_file = 1_000_000
+max_total_files = 10
+compress_files = false

examples/substrate/daemon.toml

@@ -6,6 +6,9 @@ use oura::watch::btc;
 #[cfg(feature = "eth")]
 use oura::watch::eth;
 
+#[cfg(feature = "substrate")]
+use oura::watch::substrate;
+
 use clap::{Parser, Subcommand};
 use oura::framework::Error;
 
@@ -18,6 +21,9 @@ pub fn run(args: &Args) -> Result<(), Error> {
 
         #[cfg(feature = "eth")]
         WatchCommand::Ethereum(eth_args) => eth::run(eth_args),
+
+        #[cfg(feature = "substrate")]
+        WatchCommand::Substrate(substrate_args) => substrate::run(substrate_args),
     }
 }
 
@@ -40,4 +46,8 @@ pub enum WatchCommand {
     /// Watch Ethereum blockchain
     #[cfg(feature = "eth")]
     Ethereum(eth::Args),
+
+    /// Watch Substrate blockchain
+    #[cfg(feature = "substrate")]
+    Substrate(substrate::Args),
 }

src/bin/oura/watch.rs

@@ -23,6 +23,8 @@ pub mod bitcoin;
 
 #[cfg(feature = "eth")]
 pub mod ethereum;
+
+#[cfg(feature = "substrate")]
 pub mod substrate;
 
 pub mod errors;
@@ -93,6 +95,8 @@ pub enum Record {
 
     #[cfg(feature = "btc")]
     Bitcoin(bitcoin::Record),
+
+    #[cfg(feature = "substrate")]
     Substrate(substrate::Record),
 }
 impl From<Record> for JsonValue {
@@ -104,6 +108,7 @@ impl From<Record> for JsonValue {
             Record::Ethereum(record) => record.into(),
             #[cfg(feature = "btc")]
             Record::Bitcoin(record) => record.into(),
+            #[cfg(feature = "substrate")]
             Record::Substrate(record) => record.into(),
         }
     }