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

quicknode_sdk/streams/
stream.rs
