quicknode_sdk/streams/
stream.rs

1#[cfg(feature = "rust")]
2use bon::Builder;
3#[cfg(feature = "node")]
4use napi_derive::napi;
5#[cfg(feature = "python")]
6use pyo3::{pyclass, pymethods};
7#[cfg(feature = "python")]
8use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
9use serde::{Deserialize, Deserializer, Serialize};
10
11fn deserialize_as_json_string<'de, D>(deserializer: D) -> Result<String, D::Error>
12where
13    D: Deserializer<'de>,
14{
15    let value = serde_json::Value::deserialize(deserializer)?;
16    serde_json::to_string(&value).map_err(serde::de::Error::custom)
17}
18
19// ── Enums ──────────────────────────────────────────────────────────────────
20
21/// Geographic region where a stream runs.
22#[cfg_attr(feature = "node", napi(string_enum))]
23#[cfg_attr(not(feature = "node"), derive(Clone))]
24#[derive(Debug, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum StreamRegion {
27    UsaEast,
28    EuropeCentral,
29    AsiaEast,
30}
31
32/// Type of on-chain data a stream delivers (blocks, transactions, logs, etc.).
33#[cfg_attr(feature = "node", napi(string_enum))]
34#[cfg_attr(not(feature = "node"), derive(Clone))]
35#[derive(Debug, Serialize, Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum StreamDataset {
38    Block,
39    BlockWithReceipts,
40    Transactions,
41    Logs,
42    Receipts,
43    TraceBlocks,
44    DebugTraces,
45    BlockWithReceiptsDebugTrace,
46    BlockWithReceiptsTraceBlock,
47    BlobSidecars,
48    ProgramsWithLogs,
49    Ledger,
50    Events,
51    Orders,
52    Trades,
53    BookUpdates,
54    Twap,
55    WriterActions,
56}
57
58/// Destination kind a stream delivers to (webhook, S3, Postgres, etc.).
59#[cfg_attr(feature = "node", napi(string_enum))]
60#[cfg_attr(not(feature = "node"), derive(Clone))]
61#[derive(Debug, Serialize, Deserialize)]
62#[serde(rename_all = "snake_case")]
63pub enum StreamDestination {
64    Webhook,
65    S3,
66    Azure,
67    Postgres,
68    Kafka,
69}
70
71/// Language a stream's filter function is written in.
72#[cfg_attr(feature = "node", napi(string_enum))]
73#[cfg_attr(not(feature = "node"), derive(Clone))]
74#[derive(Debug, Serialize, Deserialize)]
75#[serde(rename_all = "snake_case")]
76pub enum FilterLanguage {
77    Javascript,
78    Go,
79    Wasm,
80}
81
82/// Where stream metadata is included in delivered payloads.
83#[cfg_attr(feature = "node", napi(string_enum))]
84#[cfg_attr(not(feature = "node"), derive(Clone))]
85#[derive(Debug, Serialize, Deserialize)]
86#[serde(rename_all = "snake_case")]
87pub enum StreamMetadataLocation {
88    Body,
89    Header,
90    None,
91}
92
93/// Billing product type the stream is associated with.
94#[cfg_attr(feature = "node", napi(string_enum))]
95#[cfg_attr(not(feature = "node"), derive(Clone))]
96#[derive(Debug, Serialize, Deserialize)]
97#[serde(rename_all = "snake_case")]
98pub enum ProductType {
99    Stream,
100    Webhook,
101}
102
103/// Operational state of a stream.
104#[cfg_attr(feature = "node", napi(string_enum))]
105#[cfg_attr(not(feature = "node"), derive(Clone))]
106#[derive(Debug, Serialize, Deserialize)]
107#[serde(rename_all = "snake_case")]
108pub enum StreamStatus {
109    Active,
110    Paused,
111    Terminated,
112    Completed,
113    Blocked,
114}
115
116// ── Destination Attribute Structs ──────────────────────────────────────────
117//
118// Each struct corresponds to one StreamDestination variant. Set exactly one
119// on CreateStreamParams — see that struct's documentation for details.
120
121/// Configuration for delivering stream batches to an HTTP webhook endpoint.
122#[cfg_attr(feature = "python", gen_stub_pyclass)]
123#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
124#[cfg_attr(feature = "node", napi(object))]
125#[derive(Debug, Clone, Serialize, Deserialize)]
126pub struct WebhookAttributes {
127    /// Destination URL that receives batched stream payloads.
128    pub url: String,
129    /// Maximum number of retry attempts for a failed delivery. Must be in the range 1–10.
130    pub max_retry: i32,
131    /// Seconds to wait between retry attempts.
132    pub retry_interval_sec: i32,
133    /// Timeout in seconds for each POST request.
134    pub post_timeout_sec: i32,
135    /// Optional token included with each request so the receiver can verify authenticity. When supplied, must be at least 32 bytes (256 bits).
136    #[serde(skip_serializing_if = "Option::is_none")]
137    pub security_token: Option<String>,
138    /// Compression applied to the payload (e.g. `none`, `gzip`). When omitted the server defaults to no compression.
139    #[serde(skip_serializing_if = "Option::is_none")]
140    pub compression: Option<String>,
141}
142
143#[cfg(feature = "python")]
144#[gen_stub_pymethods]
145#[pymethods]
146impl WebhookAttributes {
147    #[new]
148    #[pyo3(signature = (url, max_retry, retry_interval_sec, post_timeout_sec, compression=None, security_token=None))]
149    pub fn new(
150        url: String,
151        max_retry: i32,
152        retry_interval_sec: i32,
153        post_timeout_sec: i32,
154        compression: Option<String>,
155        security_token: Option<String>,
156    ) -> Self {
157        Self {
158            url,
159            max_retry,
160            retry_interval_sec,
161            post_timeout_sec,
162            security_token,
163            compression,
164        }
165    }
166}
167
168/// Configuration for delivering stream batches to an S3-compatible object store.
169#[cfg_attr(feature = "python", gen_stub_pyclass)]
170#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
171#[cfg_attr(feature = "node", napi(object))]
172#[derive(Debug, Clone, Serialize, Deserialize)]
173pub struct S3Attributes {
174    /// S3 service endpoint (e.g. `s3.amazonaws.com`).
175    pub endpoint: String,
176    /// Access key used to authenticate with the S3 endpoint.
177    pub access_key: String,
178    /// Secret key used to authenticate with the S3 endpoint.
179    pub secret_key: String,
180    /// Target bucket name.
181    pub bucket: String,
182    /// Key prefix prepended to each written object.
183    pub object_prefix: String,
184    /// Compression applied to written objects (e.g. `none`, `gzip`).
185    pub compression: String,
186    /// File format/extension for written objects (e.g. `.json`).
187    pub file_type: String,
188    /// Maximum number of retry attempts for a failed write.
189    pub max_retry: i32,
190    /// Seconds to wait between retry attempts.
191    pub retry_interval_sec: i32,
192    /// Whether to use TLS when connecting to the endpoint.
193    #[serde(skip_serializing_if = "Option::is_none")]
194    pub use_ssl: Option<bool>,
195}
196
197#[cfg(feature = "python")]
198#[gen_stub_pymethods]
199#[pymethods]
200impl S3Attributes {
201    #[new]
202    #[allow(clippy::too_many_arguments)]
203    #[pyo3(signature = (endpoint, access_key, secret_key, bucket, object_prefix, compression, file_type, max_retry, retry_interval_sec, use_ssl=None))]
204    pub fn new(
205        endpoint: String,
206        access_key: String,
207        secret_key: String,
208        bucket: String,
209        object_prefix: String,
210        compression: String,
211        file_type: String,
212        max_retry: i32,
213        retry_interval_sec: i32,
214        use_ssl: Option<bool>,
215    ) -> Self {
216        Self {
217            endpoint,
218            access_key,
219            secret_key,
220            bucket,
221            object_prefix,
222            compression,
223            file_type,
224            max_retry,
225            retry_interval_sec,
226            use_ssl,
227        }
228    }
229}
230
231/// Configuration for delivering stream batches to Azure Blob Storage.
232#[cfg_attr(feature = "python", gen_stub_pyclass)]
233#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
234#[cfg_attr(feature = "node", napi(object))]
235#[derive(Debug, Clone, Serialize, Deserialize)]
236pub struct AzureAttributes {
237    /// Azure storage account name.
238    pub storage_account: String,
239    /// SAS token used to authorize writes.
240    pub sas_token: String,
241    /// Container that receives written blobs.
242    pub container: String,
243    /// Compression applied to written blobs (e.g. `none`, `gzip`).
244    pub compression: String,
245    /// File format/extension for written blobs (e.g. `.json`).
246    pub file_type: String,
247    /// Maximum number of retry attempts for a failed write.
248    pub max_retry: i32,
249    /// Seconds to wait between retry attempts.
250    pub retry_interval_sec: i32,
251    /// Optional name prefix prepended to each written blob.
252    #[serde(skip_serializing_if = "Option::is_none")]
253    pub blob_prefix: Option<String>,
254}
255
256#[cfg(feature = "python")]
257#[gen_stub_pymethods]
258#[pymethods]
259impl AzureAttributes {
260    #[new]
261    #[allow(clippy::too_many_arguments)]
262    #[pyo3(signature = (storage_account, sas_token, container, compression, file_type, max_retry, retry_interval_sec, blob_prefix=None))]
263    pub fn new(
264        storage_account: String,
265        sas_token: String,
266        container: String,
267        compression: String,
268        file_type: String,
269        max_retry: i32,
270        retry_interval_sec: i32,
271        blob_prefix: Option<String>,
272    ) -> Self {
273        Self {
274            storage_account,
275            sas_token,
276            container,
277            compression,
278            file_type,
279            max_retry,
280            retry_interval_sec,
281            blob_prefix,
282        }
283    }
284}
285
286/// Configuration for delivering stream batches to a PostgreSQL database.
287#[cfg_attr(feature = "python", gen_stub_pyclass)]
288#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
289#[cfg_attr(feature = "node", napi(object))]
290#[derive(Debug, Clone, Serialize, Deserialize)]
291pub struct PostgresAttributes {
292    /// Database host.
293    pub host: String,
294    /// Database port.
295    pub port: i32,
296    /// Database name.
297    pub database: String,
298    /// Username used to authenticate.
299    pub username: String,
300    /// Password used to authenticate.
301    pub password: String,
302    /// Destination table for inserted rows.
303    pub table_name: String,
304    /// Postgres SSL mode. The Quicknode API accepts only `disable` or `require`.
305    pub sslmode: String,
306    /// Maximum number of retry attempts for a failed write.
307    pub max_retry: i32,
308    /// Seconds to wait between retry attempts.
309    pub retry_interval_sec: i32,
310}
311
312#[cfg(feature = "python")]
313#[gen_stub_pymethods]
314#[pymethods]
315impl PostgresAttributes {
316    #[new]
317    #[allow(clippy::too_many_arguments)]
318    pub fn new(
319        host: String,
320        port: i32,
321        database: String,
322        username: String,
323        password: String,
324        table_name: String,
325        sslmode: String,
326        max_retry: i32,
327        retry_interval_sec: i32,
328    ) -> Self {
329        Self {
330            host,
331            port,
332            database,
333            username,
334            password,
335            table_name,
336            sslmode,
337            max_retry,
338            retry_interval_sec,
339        }
340    }
341}
342
343/// Configuration for delivering stream batches to a Kafka topic.
344#[cfg_attr(feature = "python", gen_stub_pyclass)]
345#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
346#[cfg_attr(feature = "node", napi(object))]
347#[derive(Debug, Clone, Serialize, Deserialize)]
348pub struct KafkaAttributes {
349    /// Comma-separated list of Kafka broker addresses (host:port).
350    pub bootstrap_servers: String,
351    /// Destination topic.
352    pub topic_name: String,
353    /// Compression codec applied to produced messages (e.g. `none`, `gzip`).
354    pub compression_type: String,
355    /// Maximum number of messages grouped per produce request.
356    pub batch_size: i32,
357    /// Milliseconds the producer waits to batch additional messages.
358    pub linger_ms: i32,
359    /// Maximum size in bytes of a single Kafka message (`max_message_bytes`).
360    pub max_message_bytes: i32,
361    /// Request timeout in seconds.
362    pub timeout_sec: i32,
363    /// Maximum number of retry attempts for a failed produce.
364    pub max_retry: i32,
365    /// Seconds to wait between retry attempts.
366    pub retry_interval_sec: i32,
367    /// Optional SASL username.
368    #[serde(skip_serializing_if = "Option::is_none")]
369    pub username: Option<String>,
370    /// Optional SASL password.
371    #[serde(skip_serializing_if = "Option::is_none")]
372    pub password: Option<String>,
373    /// Optional security protocol (e.g. `SASL_SSL`).
374    #[serde(skip_serializing_if = "Option::is_none")]
375    pub protocol: Option<String>,
376    /// Optional SASL mechanism (e.g. `PLAIN`, `SCRAM-SHA-256`).
377    #[serde(skip_serializing_if = "Option::is_none")]
378    pub mechanisms: Option<String>,
379}
380
381#[cfg(feature = "python")]
382#[gen_stub_pymethods]
383#[pymethods]
384impl KafkaAttributes {
385    #[new]
386    #[pyo3(signature = (bootstrap_servers, topic_name, compression_type, batch_size, linger_ms, max_message_bytes, timeout_sec, max_retry, retry_interval_sec, username=None, password=None, protocol=None, mechanisms=None))]
387    #[allow(clippy::too_many_arguments)]
388    pub fn new(
389        bootstrap_servers: String,
390        topic_name: String,
391        compression_type: String,
392        batch_size: i32,
393        linger_ms: i32,
394        max_message_bytes: i32,
395        timeout_sec: i32,
396        max_retry: i32,
397        retry_interval_sec: i32,
398        username: Option<String>,
399        password: Option<String>,
400        protocol: Option<String>,
401        mechanisms: Option<String>,
402    ) -> Self {
403        Self {
404            bootstrap_servers,
405            topic_name,
406            compression_type,
407            batch_size,
408            linger_ms,
409            max_message_bytes,
410            timeout_sec,
411            max_retry,
412            retry_interval_sec,
413            username,
414            password,
415            protocol,
416            mechanisms,
417        }
418    }
419}
420
421// ── Address Book Config ────────────────────────────────────────────────────
422
423/// Links a stream's filter to an address book so JSON paths resolve against its
424/// managed address set.
425#[cfg_attr(feature = "python", gen_stub_pyclass)]
426#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
427#[cfg_attr(feature = "node", napi(object))]
428#[derive(Debug, Clone, Serialize, Deserialize)]
429pub struct AddressBookConfig {
430    /// Identifier of the address book to use.
431    pub address_book_id: String,
432    /// Optional JSON path that resolves to an object whose fields are matched against the book.
433    #[serde(skip_serializing_if = "Option::is_none")]
434    pub objects_filter_path: Option<String>,
435    /// JSON paths whose resolved values are matched against the book's addresses.
436    pub elements_filter_paths: Vec<String>,
437}
438
439#[cfg(feature = "python")]
440#[gen_stub_pymethods]
441#[pymethods]
442impl AddressBookConfig {
443    #[new]
444    #[pyo3(signature = (address_book_id, elements_filter_paths, objects_filter_path=None))]
445    pub fn new(
446        address_book_id: String,
447        elements_filter_paths: Vec<String>,
448        objects_filter_path: Option<String>,
449    ) -> Self {
450        Self {
451            address_book_id,
452            objects_filter_path,
453            elements_filter_paths,
454        }
455    }
456}
457
458// ── Destination Attributes ─────────────────────────────────────────────────
459
460/// Destination-specific configuration for a stream. Exactly one variant
461/// selects where and how batches are delivered.
462// Pure-Rust discriminated union; no #[pyclass] / #[napi(object)] because PyO3
463// and napi-rs cannot represent enum-with-data. Each language binding crate
464// wraps this type for its own FFI surface.
465// The serde tag/content pair matches the API wire format when flattened into
466// a request/response struct.
467#[derive(Debug, Clone, Serialize, Deserialize)]
468#[serde(
469    tag = "destination",
470    content = "destination_attributes",
471    rename_all = "snake_case"
472)]
473pub enum DestinationAttributes {
474    /// HTTP webhook endpoint that receives batches in real time.
475    Webhook(WebhookAttributes),
476    /// S3-compatible object storage for archival or batch processing.
477    S3(S3Attributes),
478    /// Azure Blob Storage destination.
479    Azure(AzureAttributes),
480    /// PostgreSQL database destination.
481    Postgres(PostgresAttributes),
482    /// Kafka topic destination.
483    Kafka(KafkaAttributes),
484}
485
486impl DestinationAttributes {
487    pub fn tag(&self) -> StreamDestination {
488        match self {
489            Self::Webhook(_) => StreamDestination::Webhook,
490            Self::S3(_) => StreamDestination::S3,
491            Self::Azure(_) => StreamDestination::Azure,
492            Self::Postgres(_) => StreamDestination::Postgres,
493            Self::Kafka(_) => StreamDestination::Kafka,
494        }
495    }
496}
497
498// ── Request (public-facing) ────────────────────────────────────────────────
499
500/// Parameters for creating a new stream.
501#[cfg_attr(feature = "rust", derive(Builder))]
502#[derive(Debug, Clone, Serialize, Deserialize)]
503pub struct CreateStreamParams {
504    /// Human-readable label identifying the stream.
505    pub name: String,
506    /// Geographic region where the stream runs.
507    pub region: StreamRegion,
508    /// Blockchain network to stream from (e.g. `ethereum-mainnet`).
509    pub network: String,
510    /// Type of on-chain data to stream.
511    pub dataset: StreamDataset,
512    /// Block number to begin streaming from.
513    pub start_range: i64,
514    /// Block number to stop streaming at; `-1` for continuous operation.
515    pub end_range: i64,
516    /// Destination-specific configuration (webhook URL, S3 bucket, DB credentials, etc.).
517    // Flattening the enum's tag/content produces { destination, destination_attributes }.
518    #[serde(flatten)]
519    pub destination_attributes: DestinationAttributes,
520    /// Billing plan associated with the stream. Optional; the server applies the account default when omitted.
521    #[serde(skip_serializing_if = "Option::is_none")]
522    pub plan: Option<String>,
523    /// Buffer size used by the stream fetcher before delivery. Optional; the server applies its default when omitted.
524    #[serde(skip_serializing_if = "Option::is_none")]
525    pub threshold_fetch_buffer: Option<i64>,
526    /// Number of blocks grouped together per delivered batch. Required by the API.
527    pub dataset_batch_size: i64,
528    /// Upper bound on batch size when elastic batching is enabled.
529    #[serde(skip_serializing_if = "Option::is_none")]
530    pub max_batch_size: Option<i64>,
531    /// Maximum number of buffered blocks waiting to be processed.
532    #[serde(skip_serializing_if = "Option::is_none")]
533    pub max_buffer_range_size: Option<i64>,
534    /// Maximum number of worker threads processing buffered batches.
535    #[serde(skip_serializing_if = "Option::is_none")]
536    pub max_buffer_processing_workers: Option<i64>,
537    /// Number of blocks to stay behind the chain tip to reduce exposure to reorgs.
538    #[serde(skip_serializing_if = "Option::is_none")]
539    pub keep_distance_from_tip: Option<i64>,
540    /// Base64-encoded filter function applied to each batch before delivery.
541    #[serde(skip_serializing_if = "Option::is_none")]
542    pub filter_function: Option<String>,
543    /// Language the filter function is written in.
544    #[serde(skip_serializing_if = "Option::is_none")]
545    pub filter_language: Option<FilterLanguage>,
546    /// Optional address book to evaluate the filter against.
547    #[serde(skip_serializing_if = "Option::is_none")]
548    pub address_book_config: Option<AddressBookConfig>,
549    /// Where to include stream metadata in delivered payloads.
550    #[serde(skip_serializing_if = "Option::is_none")]
551    pub include_stream_metadata: Option<StreamMetadataLocation>,
552    /// Billing product type the stream is associated with.
553    #[serde(skip_serializing_if = "Option::is_none")]
554    pub product_type: Option<ProductType>,
555    /// Initial stream state (`active` or `paused`). Defaults to `active` when omitted.
556    #[serde(skip_serializing_if = "Option::is_none")]
557    pub status: Option<StreamStatus>,
558    /// Email address that receives stream termination or failure alerts.
559    #[serde(skip_serializing_if = "Option::is_none")]
560    pub notification_email: Option<String>,
561    /// Minimum charge cap applied to the stream's billing.
562    #[serde(skip_serializing_if = "Option::is_none")]
563    pub charge_min_cap: Option<i32>,
564    /// Flag (0 or 1) enabling automatic re-streaming of blocks affected by chain reorganizations.
565    #[serde(skip_serializing_if = "Option::is_none")]
566    pub fix_block_reorgs: Option<i32>,
567    /// When enabled, batch size is reduced toward 1 as the stream catches up to the chain tip. Required by the API.
568    pub elastic_batch_enabled: bool,
569    /// Additional destinations that receive the same batches alongside the primary.
570    // Not flattened: each element serializes as its own {destination, destination_attributes} pair.
571    #[serde(skip_serializing_if = "Option::is_none")]
572    pub extra_destinations: Option<Vec<DestinationAttributes>>,
573}
574
575// ── Response ───────────────────────────────────────────────────────────────
576
577/// A stream's full configuration and current state, as returned by the API.
578#[derive(Debug, Clone, Serialize, Deserialize)]
579pub struct Stream {
580    /// Unique stream identifier.
581    pub id: String,
582    /// Human-readable stream name.
583    pub name: String,
584    /// Current operational state (e.g. `active`, `paused`).
585    pub status: String,
586    /// Timestamp when the stream was created.
587    pub created_at: String,
588    /// Timestamp of the most recent modification.
589    pub updated_at: String,
590    /// Sequence number tracking stream progress.
591    pub sequence: i64,
592    /// Blockchain network the stream is reading from.
593    pub network: String,
594    /// Dataset being streamed.
595    pub dataset: String,
596    /// Geographic region where the stream runs.
597    pub region: String,
598    /// Starting block for the stream.
599    pub start_range: i64,
600    /// Ending block for the stream; `-1` indicates continuous operation.
601    pub end_range: i64,
602    /// Billing plan associated with the stream.
603    #[serde(skip_serializing_if = "Option::is_none")]
604    pub plan: Option<String>,
605    /// Buffer size used by the stream fetcher before delivery.
606    #[serde(skip_serializing_if = "Option::is_none")]
607    pub threshold_fetch_buffer: Option<i64>,
608    /// Number of blocks grouped together per delivered batch.
609    #[serde(skip_serializing_if = "Option::is_none")]
610    pub dataset_batch_size: Option<i64>,
611    /// Upper bound on batch size when elastic batching is enabled.
612    #[serde(skip_serializing_if = "Option::is_none")]
613    pub max_batch_size: Option<i64>,
614    /// Maximum number of buffered blocks waiting to be processed.
615    #[serde(skip_serializing_if = "Option::is_none")]
616    pub max_buffer_range_size: Option<i64>,
617    /// Maximum number of worker threads processing buffered batches.
618    #[serde(skip_serializing_if = "Option::is_none")]
619    pub max_buffer_processing_workers: Option<i64>,
620    /// Number of blocks the stream stays behind the chain tip.
621    #[serde(skip_serializing_if = "Option::is_none")]
622    pub keep_distance_from_tip: Option<i64>,
623    /// Base64-encoded filter function applied to each batch.
624    #[serde(skip_serializing_if = "Option::is_none")]
625    pub filter_function: Option<String>,
626    /// Language the filter function is written in.
627    #[serde(skip_serializing_if = "Option::is_none")]
628    pub filter_language: Option<String>,
629    /// Where stream metadata is included in delivered payloads.
630    #[serde(skip_serializing_if = "Option::is_none")]
631    pub include_stream_metadata: Option<String>,
632    /// Billing product type the stream is associated with.
633    #[serde(skip_serializing_if = "Option::is_none")]
634    pub product_type: Option<String>,
635    /// Email address notified of stream termination or failure.
636    #[serde(skip_serializing_if = "Option::is_none")]
637    pub notification_email: Option<String>,
638    /// Whether chain-reorg handling is enabled (0 or 1).
639    #[serde(skip_serializing_if = "Option::is_none")]
640    pub fix_block_reorgs: Option<i32>,
641    /// Most recent block hash processed by the stream.
642    #[serde(skip_serializing_if = "Option::is_none")]
643    pub current_hash: Option<String>,
644    /// Destination-specific configuration (present on single-stream responses).
645    // Optional because partial responses (e.g. list) may omit the destination pair.
646    #[serde(flatten, default, skip_serializing_if = "Option::is_none")]
647    pub destination_attributes: Option<DestinationAttributes>,
648    /// Whether elastic batching is active.
649    #[serde(skip_serializing_if = "Option::is_none")]
650    pub elastic_batch_enabled: Option<bool>,
651    /// Quicknode account ID that owns the stream.
652    #[serde(skip_serializing_if = "Option::is_none")]
653    pub qn_account_id: Option<String>,
654    /// Minimum charge cap applied to the stream's billing.
655    #[serde(skip_serializing_if = "Option::is_none")]
656    pub charge_min_cap: Option<i32>,
657    /// Free-text memo attached to the stream.
658    #[serde(skip_serializing_if = "Option::is_none")]
659    pub memo: Option<String>,
660    /// Address book linked to the stream's filter, if any.
661    #[serde(skip_serializing_if = "Option::is_none")]
662    pub address_book_config: Option<AddressBookConfig>,
663    /// Additional destinations receiving the same batches alongside the primary.
664    #[serde(default, skip_serializing_if = "Option::is_none")]
665    pub extra_destinations: Option<Vec<DestinationAttributes>>,
666}
667
668// ── New Request/Response Types ─────────────────────────────────────────────
669
670/// Pagination metadata returned alongside a paginated result set.
671#[cfg_attr(feature = "python", gen_stub_pyclass)]
672#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
673#[cfg_attr(feature = "node", napi(object))]
674#[derive(Debug, Clone, Serialize, Deserialize)]
675pub struct PageInfo {
676    /// Page size used for this response.
677    pub limit: i64,
678    /// Starting index of this page within the full result set.
679    pub offset: i64,
680    /// Total number of items matching the query across all pages.
681    pub total: i64,
682}
683
684/// Paginated response from `list_streams`.
685#[derive(Debug, Clone, Serialize, Deserialize)]
686pub struct ListStreamsResponse {
687    /// Streams on the current page.
688    pub data: Vec<Stream>,
689    /// Pagination metadata for the response.
690    #[serde(rename = "pageInfo")]
691    pub page_info: PageInfo,
692}
693
694/// Parameters for `list_streams`.
695#[cfg_attr(feature = "node", napi(object))]
696#[cfg_attr(not(feature = "node"), derive(Clone))]
697#[derive(Debug, Default, Serialize, Deserialize)]
698pub struct ListStreamsParams {
699    /// Filter results by stream type.
700    #[serde(skip_serializing_if = "Option::is_none")]
701    pub stream_type: Option<String>,
702    /// Starting index into the result set; defaults to 0.
703    #[serde(skip_serializing_if = "Option::is_none")]
704    pub offset: Option<i64>,
705    /// Maximum number of streams returned.
706    #[serde(skip_serializing_if = "Option::is_none")]
707    pub limit: Option<i64>,
708    /// Field to sort results by.
709    #[serde(skip_serializing_if = "Option::is_none")]
710    pub order_by: Option<String>,
711    /// Sort direction (`asc` or `desc`).
712    #[serde(skip_serializing_if = "Option::is_none")]
713    pub order_direction: Option<String>,
714}
715
716/// Parameters for `update_stream`. Only fields that are set are modified;
717/// omitted fields leave the current value unchanged.
718#[derive(Debug, Default, Clone, Serialize, Deserialize)]
719pub struct UpdateStreamParams {
720    /// New human-readable name.
721    #[serde(skip_serializing_if = "Option::is_none")]
722    pub name: Option<String>,
723    /// New region.
724    #[serde(skip_serializing_if = "Option::is_none")]
725    pub region: Option<StreamRegion>,
726    /// New blockchain network.
727    #[serde(skip_serializing_if = "Option::is_none")]
728    pub network: Option<String>,
729    /// New dataset.
730    #[serde(skip_serializing_if = "Option::is_none")]
731    pub dataset: Option<StreamDataset>,
732    /// New start block.
733    #[serde(skip_serializing_if = "Option::is_none")]
734    pub start_range: Option<i64>,
735    /// New end block; `-1` for continuous operation.
736    #[serde(skip_serializing_if = "Option::is_none")]
737    pub end_range: Option<i64>,
738    /// New primary destination configuration.
739    // Flattening Option<enum> omits the keys entirely when None.
740    #[serde(flatten, skip_serializing_if = "Option::is_none")]
741    pub destination_attributes: Option<DestinationAttributes>,
742    /// New billing plan.
743    #[serde(skip_serializing_if = "Option::is_none")]
744    pub plan: Option<String>,
745    /// New fetcher buffer threshold.
746    #[serde(skip_serializing_if = "Option::is_none")]
747    pub threshold_fetch_buffer: Option<i64>,
748    /// New batch size.
749    #[serde(skip_serializing_if = "Option::is_none")]
750    pub dataset_batch_size: Option<i64>,
751    /// New upper bound on elastic batch size.
752    #[serde(skip_serializing_if = "Option::is_none")]
753    pub max_batch_size: Option<i64>,
754    /// New maximum buffered block range.
755    #[serde(skip_serializing_if = "Option::is_none")]
756    pub max_buffer_range_size: Option<i64>,
757    /// New maximum number of buffer-processing workers.
758    #[serde(skip_serializing_if = "Option::is_none")]
759    pub max_buffer_processing_workers: Option<i64>,
760    /// New distance from the chain tip.
761    #[serde(skip_serializing_if = "Option::is_none")]
762    pub keep_distance_from_tip: Option<i64>,
763    /// New base64-encoded filter function.
764    #[serde(skip_serializing_if = "Option::is_none")]
765    pub filter_function: Option<String>,
766    /// New filter function language.
767    #[serde(skip_serializing_if = "Option::is_none")]
768    pub filter_language: Option<FilterLanguage>,
769    /// New address book configuration.
770    #[serde(skip_serializing_if = "Option::is_none")]
771    pub address_book_config: Option<AddressBookConfig>,
772    /// New stream-metadata location.
773    #[serde(skip_serializing_if = "Option::is_none")]
774    pub include_stream_metadata: Option<StreamMetadataLocation>,
775    /// New notification email.
776    #[serde(skip_serializing_if = "Option::is_none")]
777    pub notification_email: Option<String>,
778    /// New minimum charge cap.
779    #[serde(skip_serializing_if = "Option::is_none")]
780    pub charge_min_cap: Option<i32>,
781    /// New reorg-handling flag (0 or 1).
782    #[serde(skip_serializing_if = "Option::is_none")]
783    pub fix_block_reorgs: Option<i32>,
784    /// Whether elastic batching is enabled.
785    #[serde(skip_serializing_if = "Option::is_none")]
786    pub elastic_batch_enabled: Option<bool>,
787    /// New operational state.
788    #[serde(skip_serializing_if = "Option::is_none")]
789    pub status: Option<StreamStatus>,
790    /// Free-text memo to attach to the stream.
791    #[serde(skip_serializing_if = "Option::is_none")]
792    pub memo: Option<String>,
793    /// New set of extra destinations.
794    #[serde(skip_serializing_if = "Option::is_none")]
795    pub extra_destinations: Option<Vec<DestinationAttributes>>,
796}
797
798/// Parameters for `test_filter`.
799#[cfg_attr(feature = "node", napi(object))]
800#[cfg_attr(not(feature = "node"), derive(Clone))]
801#[derive(Debug, Serialize, Deserialize)]
802pub struct TestFilterParams {
803    /// Blockchain network to run the test against (e.g. `ethereum-mainnet`).
804    pub network: String,
805    /// Dataset the filter operates on.
806    pub dataset: StreamDataset,
807    /// Specific block number to feed into the filter for the test.
808    pub block: String,
809    /// Base64-encoded filter function to evaluate.
810    #[serde(skip_serializing_if = "Option::is_none")]
811    pub filter_function: Option<String>,
812    /// Language the filter function is written in.
813    #[serde(skip_serializing_if = "Option::is_none")]
814    pub filter_language: Option<FilterLanguage>,
815    /// Address book linked to the filter, if any.
816    #[serde(skip_serializing_if = "Option::is_none")]
817    pub address_book_config: Option<AddressBookConfig>,
818}
819
820/// Result of a `test_filter` call.
821#[cfg_attr(feature = "python", gen_stub_pyclass)]
822#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
823#[cfg_attr(feature = "node", napi(object))]
824#[derive(Debug, Clone, Serialize, Deserialize)]
825pub struct TestFilterResponse {
826    /// Filter output as a JSON string. Shape depends on the dataset and the user's filter function.
827    #[serde(deserialize_with = "deserialize_as_json_string")]
828    pub result: String,
829    /// Log lines emitted by the filter function during evaluation.
830    pub logs: Vec<String>,
831}
832
833/// Result of `get_enabled_count`.
834#[cfg_attr(feature = "python", gen_stub_pyclass)]
835#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
836#[cfg_attr(feature = "node", napi(object))]
837#[derive(Debug, Clone, Serialize, Deserialize)]
838pub struct EnabledCountResponse {
839    /// Total count of currently enabled streams.
840    pub total: i64,
841}
842
843#[cfg(test)]
844#[allow(clippy::unwrap_used)]
845mod destination_attributes_tests {
846    use super::*;
847
848    #[test]
849    fn webhook_roundtrip() {
850        let attrs = DestinationAttributes::Webhook(WebhookAttributes {
851            url: "https://x.example/hook".to_string(),
852            max_retry: 3,
853            retry_interval_sec: 5,
854            post_timeout_sec: 10,
855            compression: Some("none".to_string()),
856            security_token: None,
857        });
858        let json = serde_json::to_string(&attrs).unwrap();
859        assert!(json.contains(r#""destination":"webhook""#));
860        assert!(json.contains(r#""url":"https://x.example/hook""#));
861        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
862        assert!(matches!(parsed, DestinationAttributes::Webhook(_)));
863        assert!(matches!(parsed.tag(), StreamDestination::Webhook));
864    }
865
866    #[test]
867    fn s3_roundtrip() {
868        let attrs = DestinationAttributes::S3(S3Attributes {
869            endpoint: "s3.amazonaws.com".to_string(),
870            access_key: "AK".to_string(),
871            secret_key: "SK".to_string(),
872            bucket: "b".to_string(),
873            object_prefix: "p".to_string(),
874            compression: "none".to_string(),
875            file_type: "json".to_string(),
876            max_retry: 3,
877            retry_interval_sec: 5,
878            use_ssl: Some(true),
879        });
880        let json = serde_json::to_string(&attrs).unwrap();
881        assert!(json.contains(r#""destination":"s3""#));
882        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
883        assert!(matches!(parsed, DestinationAttributes::S3(_)));
884    }
885
886    #[test]
887    fn azure_roundtrip() {
888        let attrs = DestinationAttributes::Azure(AzureAttributes {
889            storage_account: "acct".to_string(),
890            sas_token: "tok".to_string(),
891            container: "c".to_string(),
892            compression: "none".to_string(),
893            file_type: "json".to_string(),
894            max_retry: 3,
895            retry_interval_sec: 5,
896            blob_prefix: None,
897        });
898        let json = serde_json::to_string(&attrs).unwrap();
899        assert!(json.contains(r#""destination":"azure""#));
900        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
901        assert!(matches!(parsed, DestinationAttributes::Azure(_)));
902    }
903
904    #[test]
905    fn postgres_roundtrip() {
906        let attrs = DestinationAttributes::Postgres(PostgresAttributes {
907            host: "h".to_string(),
908            port: 5432,
909            database: "db".to_string(),
910            username: "u".to_string(),
911            password: "p".to_string(),
912            table_name: "t".to_string(),
913            sslmode: "disable".to_string(),
914            max_retry: 3,
915            retry_interval_sec: 5,
916        });
917        let json = serde_json::to_string(&attrs).unwrap();
918        assert!(json.contains(r#""destination":"postgres""#));
919        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
920        assert!(matches!(parsed, DestinationAttributes::Postgres(_)));
921    }
922
923    #[test]
924    fn kafka_roundtrip() {
925        let attrs = DestinationAttributes::Kafka(KafkaAttributes {
926            bootstrap_servers: "host:9092".to_string(),
927            topic_name: "t".to_string(),
928            compression_type: "gzip".to_string(),
929            batch_size: 100,
930            linger_ms: 10,
931            max_message_bytes: 1024,
932            timeout_sec: 30,
933            max_retry: 3,
934            retry_interval_sec: 5,
935            username: None,
936            password: None,
937            protocol: None,
938            mechanisms: None,
939        });
940        let json = serde_json::to_string(&attrs).unwrap();
941        assert!(json.contains(r#""destination":"kafka""#));
942        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
943        assert!(matches!(parsed, DestinationAttributes::Kafka(_)));
944    }
945}
quicknode_sdk/streams/stream.rs

quicknode_sdk/streams/
stream.rs
