feat: Data transformation plugin

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["jetstreamer-firehose", "jetstreamer-plugin", "jetstreamer-utils"]
+members = ["jetstreamer-firehose", "jetstreamer-plugin", "jetstreamer-utils", "history-import-old-faithful"]
 
 [workspace.package]
 edition = "2024"
@@ -97,10 +97,18 @@ ctrlc = "3"
 indoc = "2"
 tempfile = "3"
 libc = "0"
+arrow = "56"
+aws-config = "1"
 aws-credential-types = "1"
 aws-sdk-s3 = "1"
 aws-smithy-types = "1"
 aws-types = "1"
+bincode_2 = { package = "bincode", version = "2", features = ["serde", "std", "alloc"] }
+clap = { version = "4", features = ["derive", "env"] }
+parquet = { version = "56", features = ["async"] }
+serde_with = { version = "3", features = ["base64"] }
+solana-account-decoder = "3"
+solana-transaction-error = "3"
 
 [dependencies]
 tokio.workspace = true

history-import-old-faithful/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "history-import-old-faithful"
+edition.workspace = true
+version.workspace = true
+
+[dependencies]
+arrow.workspace = true
+aws-config.workspace = true
+aws-sdk-s3.workspace = true
+bincode_2.workspace = true
+bs58.workspace = true
+clap.workspace = true
+clickhouse.workspace = true
+dashmap.workspace = true
+futures-util.workspace = true
+log.workspace = true
+parquet.workspace = true
+serde.workspace = true
+serde_with.workspace = true
+tempfile.workspace = true
+thiserror.workspace = true
+tokio = { workspace = true, features = ["full"] }
+zstd = { workspace = true, features = ["zstdmt"] }
+tikv-jemallocator = "0.6"
+
+jetstreamer = { path = ".." }
+jetstreamer-firehose.workspace = true
+jetstreamer-plugin.workspace = true
+
+solana-account-decoder.workspace = true
+solana-address.workspace = true
+solana-hash.workspace = true
+solana-reward-info.workspace = true
+solana-transaction.workspace = true
+solana-transaction-error.workspace = true
+solana-transaction-status.workspace = true

history-import-old-faithful/src/compat_bincode.rs

@@ -0,0 +1,20 @@
+//! Bincode helpers matching `nitro_utils::bincode` configuration.
+//!
+//! Uses bincode 2.x with serde compat, fixed-int encoding, and little-endian
+//! byte order — the exact same config as nitro-stream, so serialized bytes are
+//! identical.
+
+use bincode_2::config::Config;
+use serde::Serialize;
+
+pub type EncodeError = bincode_2::error::EncodeError;
+
+fn config() -> impl Config {
+    bincode_2::config::standard()
+        .with_fixed_int_encoding()
+        .with_little_endian()
+}
+
+pub fn serialize<T: Serialize>(value: &T) -> Result<Vec<u8>, EncodeError> {
+    bincode_2::serde::encode_to_vec(value, config())
+}

history-import-old-faithful/src/main.rs

@@ -0,0 +1,82 @@
+mod compat_bincode;
+mod plugin;
+mod types;
+mod writer;
+
+use clap::Parser;
+use jetstreamer::JetstreamerRunner;
+use std::path::PathBuf;
+use tikv_jemallocator::Jemalloc;
+
+use crate::plugin::ParquetExportPlugin;
+
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
+
+#[derive(Parser)]
+#[command(about = "Export Old Faithful block data to partitioned parquet files on S3")]
+struct Args {
+    /// First slot to process (inclusive).
+    #[arg(long, env)]
+    start_slot: u64,
+
+    /// Last slot to process (exclusive).
+    #[arg(long, env)]
+    end_slot: u64,
+
+    /// S3 bucket for parquet uploads.
+    #[arg(long, env)]
+    s3_bucket: String,
+
+    /// S3 key prefix (e.g. "data/v1").
+    #[arg(long, env, default_value = "")]
+    s3_prefix: String,
+
+    /// Number of firehose ingestion threads.
+    #[arg(long, env = "JETSTREAMER_THREADS", default_value = "4")]
+    threads: usize,
+
+    /// Optional local root directory where partition files are persisted.
+    /// Files are written under this root using the same key layout as S3.
+    #[arg(long, env)]
+    data_path: Option<PathBuf>,
+
+    /// Probe S3 once at startup and skip partitions that already exist.
+    #[arg(
+        long,
+        env = "HISTORY_IMPORT_SKIP_EXISTING_PARTITIONS",
+        default_value_t = true
+    )]
+    skip_existing_partitions: bool,
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = Args::parse();
+
+    assert!(
+        args.start_slot < args.end_slot,
+        "start_slot must be less than end_slot"
+    );
+
+    let temp_dir = tempfile::tempdir()?;
+
+    let plugin = ParquetExportPlugin::new(
+        args.s3_bucket,
+        args.s3_prefix,
+        temp_dir.path().to_path_buf(),
+        args.data_path,
+        args.start_slot,
+        args.end_slot,
+        args.threads,
+        args.skip_existing_partitions,
+    );
+
+    JetstreamerRunner::new()
+        .with_log_level("info")
+        .with_plugin(Box::new(plugin))
+        .with_threads(args.threads)
+        .with_slot_range_bounds(args.start_slot, args.end_slot)
+        .run()?;
+
+    Ok(())
+}