v0.3.13-pre.005

crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
-// version: 21
+// version: 22
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,10 +10,11 @@
 //! This tranche owns the concrete Worker family identity, validated technical settings
 //! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also
 //! owns bounded source-neutral admission, common RAW canonicalization/assembly and backend-neutral
-//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API. `pre.004` keeps the bounded 1..32 caller-composed
-//! aggregate and adds a productive standard Solana `blockSubscribe` direct RAW source beside Yellowstone and Standard Logs while simultaneous multi-source
-//! activation remains gated until the dedicated supervisor tranche. Standard Logs and Yellowstone reference paths converge into one source-neutral hydration
-//! coordinator contract; qualified Standard Block Legacy/V0/V1 transactions enter the existing central admission path directly. Yellowstone
+//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API. `pre.005` keeps the bounded 1..32 caller-composed
+//! aggregate and adds a productive Helius `transactionSubscribe` + HTTP hydration source beside Yellowstone, Standard Logs and Standard Block while
+//! simultaneous multi-source activation remains gated until the dedicated supervisor tranche. Helius Full notifications, Standard Logs and Yellowstone
+//! reference paths converge into one source-neutral hydration coordinator contract; qualified Standard Block Legacy/V0/V1 transactions enter the existing
+//! central admission path directly. Yellowstone
 //! Transaction/TransactionStatus/Block and standard logs notifications become signature/slot references; BlockMeta/Slot remain continuity-only signals.
 //! Hydration is coalesced by network/signature/commitment under bounded in-flight and pending budgets. A bounded run-local processing frontier projects
 //! hydration pending, oldest pending slot and highest unblocked actually observed slot. The productive source also projects safe Transport reconnect/replay
@@ -52,6 +53,8 @@ pub use self::runtime::RawTransactionIngestTerminalFuture;
 pub use self::runtime::RawTransactionIngestWorker;
 /// Maximum number of logical live sources accepted by one runtime-resource aggregate.
 pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES;
+/// Validated Helius `transactionSubscribe` + HTTP hydration source contract owned by the continuous RAW transaction ingest Worker.
+pub use self::runtime_resources::RawTransactionIngestHeliusTransactionSource;
 /// Caller-composed bounded runtime resources for supported continuous RAW transaction live-source families.
 pub use self::runtime_resources::RawTransactionIngestRuntimeResources;
 /// Validated standard Solana `blockSubscribe` direct RAW source contract owned by the continuous RAW transaction ingest Worker.

crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs

@@ -1,11 +1,14 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 16
+// version: 17
 
 use sha2::Digest; // rust-rules: trait-import
 
 /// Maximum number of logical live sources accepted in one RAW transaction ingest runtime-resource aggregate.
 pub const MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES: usize = 32;
 
+const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.helius_transaction.filter.v1\0";
+const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_PROTOCOL: &str = "helius_ws_http";
+const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.helius_transaction_http.source_key.v1\0";
 const RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.live_source.source_key.v1\0";
 const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_ACQUISITION_METHOD: &str = "block_subscribe";
 const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_block.filter.v1\0";
@@ -26,6 +29,7 @@ enum RawTransactionIngestSourceFamily {
 }
 
 enum RawTransactionIngestLiveSource {
+    HeliusTransaction(crate::RawTransactionIngestHeliusTransactionSource),
     StandardBlock(crate::RawTransactionIngestStandardBlockSource),
     StandardLogs(crate::RawTransactionIngestStandardLogsSource),
     Yellowstone(crate::RawTransactionIngestYellowstoneSource),
@@ -34,6 +38,7 @@ enum RawTransactionIngestLiveSource {
 impl RawTransactionIngestLiveSource {
     fn network(&self) -> &ksp_store_lib::RawNetworkId {
         return match self {
+            Self::HeliusTransaction(source) => &source.network,
             Self::StandardBlock(source) => &source.network,
             Self::StandardLogs(source) => &source.network,
             Self::Yellowstone(source) => &source.network,
@@ -42,6 +47,7 @@ impl RawTransactionIngestLiveSource {
 
     fn source_key(&self) -> [u8; 32] {
         return match self {
+            Self::HeliusTransaction(source) => source.source_key,
             Self::StandardBlock(source) => source.source_key,
             Self::StandardLogs(source) => source.source_key,
             Self::Yellowstone(source) => source.source_key,
@@ -56,6 +62,7 @@ impl RawTransactionIngestLiveSource {
         processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
     ) -> ksp_core_lib::Result<()> {
         return match self {
+            Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
@@ -578,6 +585,276 @@ impl crate::RawTransactionIngestYellowstoneSource {
     }
 }
 
+/// Validated Helius `transactionSubscribe` plus HTTP hydration source owned by the continuous RAW transaction ingest Worker.
+///
+/// The caller provides one Transport-owned Helius LaserStream WebSocket endpoint, one Helius transaction filter, a Confirmed/Finalized commitment and one HTTP
+/// hydration pool/role. The Worker always requests Full/Base64 notifications with `maxSupportedTransactionVersion = 1` but projects only the provider
+/// signature/slot/index reference into the source-neutral hydration coordinator; the nested Helius payload is never copied into Worker state, while endpoint
+/// URLs and credentials remain encapsulated by Transport-owned settings and are never exposed by Worker APIs or diagnostics.
+pub struct RawTransactionIngestHeliusTransactionSource {
+    ws_endpoint: ksp_onchain_transport_lib::WsEndpointSettings,
+    filter: ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+    hydration_role: ksp_onchain_transport_lib::HttpRoleName,
+    network: ksp_store_lib::RawNetworkId,
+    route: RawTransactionIngestSourceRoute,
+    filter_fingerprint: [u8; 32],
+    source_key: [u8; 32],
+}
+
+impl crate::RawTransactionIngestHeliusTransactionSource {
+    /// Creates one validated Helius transaction source without opening WebSocket or HTTP network I/O.
+    pub fn new(
+        ws_endpoint: ksp_onchain_transport_lib::WsEndpointSettings,
+        filter: ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter,
+        commitment: ksp_onchain_transport_lib::SolanaCommitment,
+        http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+        hydration_role: ksp_onchain_transport_lib::HttpRoleName,
+    ) -> ksp_core_lib::Result<Self> {
+        let ws_settings = ksp_onchain_transport_lib::WsTransportSettings::new(std::vec![ws_endpoint.clone()]);
+        if ws_settings.validate().is_err() {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_endpoint_invalid"));
+        }
+        if ws_endpoint.protocol() != ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_protocol_invalid"));
+        }
+        match commitment {
+            ksp_onchain_transport_lib::SolanaCommitment::Confirmed | ksp_onchain_transport_lib::SolanaCommitment::Finalized => {},
+            ksp_onchain_transport_lib::SolanaCommitment::Processed => {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_commitment_invalid"));
+            },
+        }
+        let request = helius_transaction_request(filter.clone(), commitment);
+        if request.validate().is_err() {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_request_invalid"));
+        }
+        let network = match ksp_store_lib::RawNetworkId::new(ws_endpoint.cluster().as_str()) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_network_unrepresentable"));
+            },
+        };
+        let provider = match ksp_store_lib::RawProvenanceCode::new(ws_endpoint.provider().as_str()) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_provider_unrepresentable"));
+            },
+        };
+        let endpoint_id = match ksp_store_lib::RawProvenanceCode::new(ws_endpoint.name()) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.helius_transaction_endpoint_unrepresentable"));
+            },
+        };
+        let route = RawTransactionIngestSourceRoute { endpoint_id, provider };
+        let method = match ksp_onchain_transport_lib::find_http_rpc_method("getTransaction") {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")),
+        };
+        let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str(), &route, "hx");
+        let compatible_http_routes = match compatible_http_routes {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if compatible_http_routes == 0 {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_role_unsupported"));
+        }
+        let filter_fingerprint = helius_transaction_filter_fingerprint(&filter);
+        let source_key = helius_transaction_live_source_key(&network, &route, commitment, &filter_fingerprint);
+        return std::result::Result::Ok(Self {
+            ws_endpoint,
+            filter,
+            commitment,
+            http_pool,
+            hydration_role,
+            network,
+            route,
+            filter_fingerprint,
+            source_key,
+        });
+    }
+
+    fn hydration_context(&self) -> RawTransactionIngestHydrationContext {
+        return RawTransactionIngestHydrationContext {
+            commitment: self.commitment,
+            http_pool: self.http_pool.clone(),
+            hydration_role: self.hydration_role.clone(),
+            network: self.network.clone(),
+            protocol: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_PROTOCOL,
+            route: self.route.clone(),
+            route_prefix: "hx",
+            source_key_domain: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_SOURCE_KEY_DOMAIN,
+        };
+    }
+
+    /// Runs one productive Helius `transactionSubscribe` source until cooperative stop or one safe terminal source failure.
+    pub(crate) async fn run(
+        self,
+        settings: crate::RawTransactionIngestSettings,
+        mut stop_receiver: tokio::sync::watch::Receiver<bool>,
+        admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
+        processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+    ) -> ksp_core_lib::Result<()> {
+        let connected = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(());
+            }
+            result = ksp_onchain_transport_lib::HeliusLaserStreamWsSession::connect(self.ws_endpoint.clone()) => result,
+        };
+        let session = match connected {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
+        };
+        let request = helius_transaction_request(self.filter.clone(), self.commitment);
+        let subscribed = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                let _closed = session.close().await;
+                return std::result::Result::Ok(());
+            }
+            result = session.transaction_subscribe(&request) => result,
+        };
+        let mut subscription = match subscribed {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => {
+                let _closed = session.close().await;
+                return std::result::Result::Err(source_transport_error(error.code()));
+            },
+        };
+        let hydration = self.hydration_context();
+        let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings);
+        let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active);
+        let mut fault = std::option::Option::None;
+        loop {
+            if *stop_receiver.borrow() {
+                break;
+            }
+            if let std::result::Result::Err(error) = coordinator.start_hydrations(&hydration) {
+                fault = std::option::Option::Some(error);
+                break;
+            }
+            let can_receive = coordinator.can_receive();
+            if !can_receive && coordinator.tasks.is_empty() {
+                fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
+                break;
+            }
+            tokio::select! {
+                biased;
+                _ = stop_receiver.changed() => {
+                    break;
+                }
+                joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => {
+                    let joined = match joined {
+                        std::option::Option::Some(value) => value,
+                        std::option::Option::None => {
+                            fault = std::option::Option::Some(crate::runtime_error("source.hydration_join_missing"));
+                            break;
+                        },
+                    };
+                    let handled = coordinator
+                        .handle_joined(
+                            joined,
+                            &hydration,
+                            &settings,
+                            &admission_sender,
+                            &mut stop_receiver,
+                            &mut processing_frontier,
+                        )
+                        .await;
+                    match handled {
+                        std::result::Result::Ok(true) => {},
+                        std::result::Result::Ok(false) => break,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(error);
+                            break;
+                        },
+                    }
+                }
+                notification = subscription.recv(), if can_receive => {
+                    let notification = match notification {
+                        std::option::Option::Some(std::result::Result::Ok(value)) => value,
+                        std::option::Option::Some(std::result::Result::Err(error)) => {
+                            fault = std::option::Option::Some(source_transport_error(error.code()));
+                            break;
+                        },
+                        std::option::Option::None => {
+                            let error = match subscription.terminal_error_code() {
+                                std::option::Option::Some(code) => source_transport_error(code),
+                                std::option::Option::None => crate::runtime_error("source.helius_transaction_subscription_closed"),
+                            };
+                            fault = std::option::Option::Some(error);
+                            break;
+                        },
+                    };
+                    let received_at = match current_raw_timestamp() {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(error);
+                            break;
+                        },
+                    };
+                    let full = match notification {
+                        ksp_onchain_transport_lib::HeliusTransactionNotification::Full(value) => value,
+                        _ => {
+                            fault = std::option::Option::Some(crate::runtime_error("source.helius_transaction_notification_unqualified"));
+                            break;
+                        },
+                    };
+                    let signal = match project_helius_transaction_signal(&self, &full) {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(error);
+                            break;
+                        },
+                    };
+                    if let std::result::Result::Err(error) = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier) {
+                        fault = std::option::Option::Some(error);
+                        break;
+                    }
+                }
+            }
+        }
+        coordinator.abort_all(&mut processing_frontier).await;
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing);
+        let closed = session.close().await;
+        if let std::option::Option::Some(error) = fault {
+            processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+            return std::result::Result::Err(error);
+        }
+        return match closed {
+            std::result::Result::Ok(()) => {
+                processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closed);
+                std::result::Result::Ok(())
+            },
+            std::result::Result::Err(error) => {
+                processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+                std::result::Result::Err(source_transport_error(error.code()))
+            },
+        };
+    }
+}
+
+impl std::fmt::Debug for crate::RawTransactionIngestHeliusTransactionSource {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let http_snapshot = self.http_pool.snapshot();
+        return formatter
+            .debug_struct("RawTransactionIngestHeliusTransactionSource")
+            .field("ws_endpoint_name", &self.ws_endpoint.name())
+            .field("ws_provider", &self.ws_endpoint.provider().as_str())
+            .field("network", &self.network.as_str())
+            .field("commitment", &self.commitment)
+            .field("filter", &self.filter)
+            .field("filter_fingerprint_bytes", &self.filter_fingerprint.len())
+            .field("hydration_role", &self.hydration_role.as_str())
+            .field("http_endpoint_count", &http_snapshot.endpoint_count())
+            .field("source_key_bytes", &self.source_key.len())
+            .finish();
+    }
+}
+
 /// Validated standard Solana `blockSubscribe` direct RAW source owned by the continuous RAW transaction ingest Worker.
 ///
 /// The caller provides one Transport-owned standard WebSocket endpoint, one block filter and a Confirmed/Finalized commitment. The source requests Full/Base64
@@ -1055,8 +1332,8 @@ impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
 
 /// Caller-composed runtime resources accepted by the continuous RAW transaction ingest Worker.
 ///
-/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.004` supports Yellowstone, standard Solana logs and standard
-/// Solana block sources while deliberately keeping simultaneous multi-source supervision gated until the dedicated supervisor tranche.
+/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.005` supports Yellowstone, standard Solana logs/block and
+/// Helius transaction sources while deliberately keeping simultaneous multi-source supervision gated until the dedicated supervisor tranche.
 pub struct RawTransactionIngestRuntimeResources {
     sources: std::vec::Vec<RawTransactionIngestLiveSource>,
 }
@@ -1068,6 +1345,12 @@ impl crate::RawTransactionIngestRuntimeResources {
         return Self { sources: std::vec![RawTransactionIngestLiveSource::Yellowstone(yellowstone_source)] };
     }
 
+    /// Creates one runtime-resource aggregate from one Helius `transactionSubscribe` plus HTTP hydration source contract.
+    #[must_use]
+    pub fn from_helius_transaction_source(source: crate::RawTransactionIngestHeliusTransactionSource) -> Self {
+        return Self { sources: std::vec![RawTransactionIngestLiveSource::HeliusTransaction(source)] };
+    }
+
     /// Creates one runtime-resource aggregate from one standard Solana `blockSubscribe` direct RAW source contract.
     #[must_use]
     pub fn from_standard_block_source(source: crate::RawTransactionIngestStandardBlockSource) -> Self {
@@ -1107,6 +1390,27 @@ impl crate::RawTransactionIngestRuntimeResources {
         return std::result::Result::Ok(());
     }
 
+    /// Adds one validated Helius transaction source while preserving the global 1..32 bound, one-network invariant and unique logical source identity.
+    pub fn try_push_helius_transaction_source(&mut self, source: crate::RawTransactionIngestHeliusTransactionSource) -> ksp_core_lib::Result<()> {
+        if self.sources.len() >= crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_count_exceeded"));
+        }
+        let candidate = RawTransactionIngestLiveSource::HeliusTransaction(source);
+        let expected_network = match self.sources.first() {
+            std::option::Option::Some(source) => source.network(),
+            std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")),
+        };
+        if candidate.network() != expected_network {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_network_mismatch"));
+        }
+        let source_key = candidate.source_key();
+        if self.sources.iter().any(|source| return source.source_key() == source_key) {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.duplicate_source_identity"));
+        }
+        self.sources.push(candidate);
+        return std::result::Result::Ok(());
+    }
+
     /// Adds one validated standard Solana block source while preserving the global 1..32 bound, one-network invariant and unique logical source identity.
     pub fn try_push_standard_block_source(&mut self, source: crate::RawTransactionIngestStandardBlockSource) -> ksp_core_lib::Result<()> {
         if self.sources.len() >= crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
@@ -1218,6 +1522,93 @@ impl std::hash::Hasher for RawTransactionIngestSourceKeyHashWriter<'_> {
     }
 }
 
+fn helius_transaction_filter_fingerprint(filter: &ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter) -> [u8; 32] {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_FILTER_FINGERPRINT_DOMAIN);
+    hash_helius_optional_bool(&mut hasher, b"failed", filter.failed());
+    hash_helius_optional_bool(&mut hasher, b"vote", filter.vote());
+    hash_helius_optional_pubkey_list(&mut hasher, b"account_exclude", filter.account_exclude());
+    hash_helius_optional_pubkey_list(&mut hasher, b"account_include", filter.account_include());
+    hash_helius_optional_pubkey_list(&mut hasher, b"account_required", filter.account_required());
+    hash_live_source_key_component(&mut hasher, b"signature");
+    match filter.signature() {
+        std::option::Option::Some(value) => {
+            hash_live_source_key_component(&mut hasher, b"value");
+            hash_live_source_key_component(&mut hasher, value.as_bytes());
+        },
+        std::option::Option::None => hash_live_source_key_component(&mut hasher, b"omitted"),
+    }
+    hash_live_source_key_component(&mut hasher, b"token_accounts");
+    match filter.token_accounts() {
+        std::option::Option::Some(value) => {
+            hash_live_source_key_component(&mut hasher, b"value");
+            hash_live_source_key_component(&mut hasher, value.as_str().as_bytes());
+        },
+        std::option::Option::None => hash_live_source_key_component(&mut hasher, b"omitted"),
+    }
+    return hasher.finalize().into();
+}
+
+fn helius_transaction_live_source_key(
+    network: &ksp_store_lib::RawNetworkId,
+    route: &RawTransactionIngestSourceRoute,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    filter_fingerprint: &[u8; 32],
+) -> [u8; 32] {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN);
+    hash_live_source_key_component(&mut hasher, b"helius_transaction");
+    hash_live_source_key_component(&mut hasher, network.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, route.provider.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, route.endpoint_id.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, commitment.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, filter_fingerprint);
+    return hasher.finalize().into();
+}
+
+fn helius_transaction_request(
+    filter: ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_onchain_transport_lib::HeliusTransactionSubscribeRequest {
+    let options = ksp_onchain_transport_lib::HeliusTransactionSubscribeOptions::new(
+        std::option::Option::Some(commitment),
+        std::option::Option::Some(ksp_onchain_transport_lib::HeliusTransactionSubscribeEncoding::Base64),
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionDetails::Full),
+        std::option::Option::Some(false),
+        std::option::Option::Some(1),
+    );
+    return ksp_onchain_transport_lib::HeliusTransactionSubscribeRequest::new(filter, std::option::Option::Some(options));
+}
+
+fn hash_helius_optional_bool(hasher: &mut sha2::Sha256, label: &[u8], value: std::option::Option<bool>) {
+    hash_live_source_key_component(hasher, label);
+    match value {
+        std::option::Option::Some(true) => hash_live_source_key_component(hasher, b"true"),
+        std::option::Option::Some(false) => hash_live_source_key_component(hasher, b"false"),
+        std::option::Option::None => hash_live_source_key_component(hasher, b"omitted"),
+    }
+    return;
+}
+
+fn hash_helius_optional_pubkey_list(hasher: &mut sha2::Sha256, label: &[u8], values: std::option::Option<&[ksp_core_lib::Pubkey]>) {
+    hash_live_source_key_component(hasher, label);
+    let values = match values {
+        std::option::Option::Some(values) => values,
+        std::option::Option::None => {
+            hash_live_source_key_component(hasher, b"omitted");
+            return;
+        },
+    };
+    hash_live_source_key_component(hasher, b"value");
+    let mut normalized = values.iter().map(|value| return value.to_bytes()).collect::<std::vec::Vec<_>>();
+    normalized.sort_unstable();
+    hash_live_source_key_component(hasher, &(normalized.len() as u64).to_be_bytes());
+    for value in normalized {
+        hash_live_source_key_component(hasher, &value);
+    }
+    return;
+}
+
 fn standard_block_filter_fingerprint(filter: &ksp_onchain_transport_lib::SolanaBlockSubscribeFilter) -> [u8; 32] {
     let mut hasher = sha2::Sha256::new();
     hasher.update(RAW_TRANSACTION_INGEST_STANDARD_BLOCK_FILTER_FINGERPRINT_DOMAIN);
@@ -1666,6 +2057,51 @@ fn build_standard_block_provenance(
     );
 }
 
+trait RawTransactionIngestHeliusTransactionView {
+    fn signature(&self) -> &str;
+
+    fn slot(&self) -> u64;
+
+    fn transaction_index(&self) -> u64;
+}
+
+impl RawTransactionIngestHeliusTransactionView for ksp_onchain_transport_lib::HeliusFullTransactionNotification {
+    fn signature(&self) -> &str {
+        return ksp_onchain_transport_lib::HeliusFullTransactionNotification::signature(self);
+    }
+
+    fn slot(&self) -> u64 {
+        return ksp_onchain_transport_lib::HeliusFullTransactionNotification::slot(self);
+    }
+
+    fn transaction_index(&self) -> u64 {
+        return ksp_onchain_transport_lib::HeliusFullTransactionNotification::transaction_index(self);
+    }
+}
+
+fn project_helius_transaction_signal<T: RawTransactionIngestHeliusTransactionView>(
+    source: &crate::RawTransactionIngestHeliusTransactionSource,
+    response: &T,
+) -> ksp_core_lib::Result<RawTransactionIngestSourceSignal> {
+    let signature = ksp_raw_transaction_lib::parse_raw_transaction_signature(response.signature());
+    let signature = match signature {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.helius_transaction_signature_invalid")),
+    };
+    return std::result::Result::Ok(RawTransactionIngestSourceSignal {
+        created_at: std::option::Option::None,
+        family: RawTransactionIngestSourceFamily::Transaction,
+        matched_filter_count: 1,
+        matched_filter_fingerprint: source.filter_fingerprint,
+        matched_filter_id: std::option::Option::None,
+        network: source.network.clone(),
+        route: source.route.clone(),
+        signature,
+        slot: response.slot(),
+        transaction_index: std::option::Option::Some(response.transaction_index()),
+    });
+}
+
 trait RawTransactionIngestStandardLogsView {
     fn signature(&self) -> &str;