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    Clickhouse,
69    Snowflake,
70    Mysql,
71    Mongo,
72    Kafka,
73    Redis,
74}
75
76/// Language a stream's filter function is written in.
77#[cfg_attr(feature = "node", napi(string_enum))]
78#[cfg_attr(not(feature = "node"), derive(Clone))]
79#[derive(Debug, Serialize, Deserialize)]
80#[serde(rename_all = "snake_case")]
81pub enum FilterLanguage {
82    Javascript,
83    Go,
84    Wasm,
85}
86
87/// Where stream metadata is included in delivered payloads.
88#[cfg_attr(feature = "node", napi(string_enum))]
89#[cfg_attr(not(feature = "node"), derive(Clone))]
90#[derive(Debug, Serialize, Deserialize)]
91#[serde(rename_all = "snake_case")]
92pub enum StreamMetadataLocation {
93    Body,
94    Header,
95    None,
96}
97
98/// Billing product type the stream is associated with.
99#[cfg_attr(feature = "node", napi(string_enum))]
100#[cfg_attr(not(feature = "node"), derive(Clone))]
101#[derive(Debug, Serialize, Deserialize)]
102#[serde(rename_all = "snake_case")]
103pub enum ProductType {
104    Stream,
105    Webhook,
106}
107
108/// Operational state of a stream.
109#[cfg_attr(feature = "node", napi(string_enum))]
110#[cfg_attr(not(feature = "node"), derive(Clone))]
111#[derive(Debug, Serialize, Deserialize)]
112#[serde(rename_all = "snake_case")]
113pub enum StreamStatus {
114    Active,
115    Paused,
116    Terminated,
117    Completed,
118    Blocked,
119}
120
121// ── Destination Attribute Structs ──────────────────────────────────────────
122//
123// Each struct corresponds to one StreamDestination variant. Set exactly one
124// on CreateStreamParams — see that struct's documentation for details.
125
126/// Configuration for delivering stream batches to an HTTP webhook endpoint.
127#[cfg_attr(feature = "python", gen_stub_pyclass)]
128#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
129#[cfg_attr(feature = "node", napi(object))]
130#[derive(Debug, Clone, Serialize, Deserialize)]
131pub struct WebhookAttributes {
132    /// Destination URL that receives batched stream payloads.
133    pub url: String,
134    /// Maximum number of retry attempts for a failed delivery.
135    pub max_retry: i32,
136    /// Seconds to wait between retry attempts.
137    pub retry_interval_sec: i32,
138    /// Timeout in seconds for each POST request.
139    pub post_timeout_sec: i32,
140    /// Optional token included with each request so the receiver can verify authenticity.
141    #[serde(skip_serializing_if = "Option::is_none")]
142    pub security_token: Option<String>,
143    /// Compression applied to the payload (e.g. `none`, `gzip`).
144    pub compression: String,
145}
146
147#[cfg(feature = "python")]
148#[gen_stub_pymethods]
149#[pymethods]
150impl WebhookAttributes {
151    #[new]
152    #[pyo3(signature = (url, max_retry, retry_interval_sec, post_timeout_sec, compression, security_token=None))]
153    pub fn new(
154        url: String,
155        max_retry: i32,
156        retry_interval_sec: i32,
157        post_timeout_sec: i32,
158        compression: String,
159        security_token: Option<String>,
160    ) -> Self {
161        Self {
162            url,
163            max_retry,
164            retry_interval_sec,
165            post_timeout_sec,
166            security_token,
167            compression,
168        }
169    }
170}
171
172/// Configuration for delivering stream batches to an S3-compatible object store.
173#[cfg_attr(feature = "python", gen_stub_pyclass)]
174#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
175#[cfg_attr(feature = "node", napi(object))]
176#[derive(Debug, Clone, Serialize, Deserialize)]
177pub struct S3Attributes {
178    /// S3 service endpoint (e.g. `s3.amazonaws.com`).
179    pub endpoint: String,
180    /// Access key used to authenticate with the S3 endpoint.
181    pub access_key: String,
182    /// Secret key used to authenticate with the S3 endpoint.
183    pub secret_key: String,
184    /// Target bucket name.
185    pub bucket: String,
186    /// Key prefix prepended to each written object.
187    pub object_prefix: String,
188    /// Compression applied to written objects (e.g. `none`, `gzip`).
189    pub compression: String,
190    /// File format/extension for written objects (e.g. `.json`).
191    pub file_type: String,
192    /// Maximum number of retry attempts for a failed write.
193    pub max_retry: i32,
194    /// Seconds to wait between retry attempts.
195    pub retry_interval_sec: i32,
196    /// Whether to use TLS when connecting to the endpoint.
197    #[serde(skip_serializing_if = "Option::is_none")]
198    pub use_ssl: Option<bool>,
199}
200
201#[cfg(feature = "python")]
202#[gen_stub_pymethods]
203#[pymethods]
204impl S3Attributes {
205    #[new]
206    #[allow(clippy::too_many_arguments)]
207    #[pyo3(signature = (endpoint, access_key, secret_key, bucket, object_prefix, compression, file_type, max_retry, retry_interval_sec, use_ssl=None))]
208    pub fn new(
209        endpoint: String,
210        access_key: String,
211        secret_key: String,
212        bucket: String,
213        object_prefix: String,
214        compression: String,
215        file_type: String,
216        max_retry: i32,
217        retry_interval_sec: i32,
218        use_ssl: Option<bool>,
219    ) -> Self {
220        Self {
221            endpoint,
222            access_key,
223            secret_key,
224            bucket,
225            object_prefix,
226            compression,
227            file_type,
228            max_retry,
229            retry_interval_sec,
230            use_ssl,
231        }
232    }
233}
234
235/// Configuration for delivering stream batches to Azure Blob Storage.
236#[cfg_attr(feature = "python", gen_stub_pyclass)]
237#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
238#[cfg_attr(feature = "node", napi(object))]
239#[derive(Debug, Clone, Serialize, Deserialize)]
240pub struct AzureAttributes {
241    /// Azure storage account name.
242    pub storage_account: String,
243    /// SAS token used to authorize writes.
244    pub sas_token: String,
245    /// Container that receives written blobs.
246    pub container: String,
247    /// Compression applied to written blobs (e.g. `none`, `gzip`).
248    pub compression: String,
249    /// File format/extension for written blobs (e.g. `.json`).
250    pub file_type: String,
251    /// Maximum number of retry attempts for a failed write.
252    pub max_retry: i32,
253    /// Seconds to wait between retry attempts.
254    pub retry_interval_sec: i32,
255    /// Optional name prefix prepended to each written blob.
256    #[serde(skip_serializing_if = "Option::is_none")]
257    pub blob_prefix: Option<String>,
258}
259
260#[cfg(feature = "python")]
261#[gen_stub_pymethods]
262#[pymethods]
263impl AzureAttributes {
264    #[new]
265    #[allow(clippy::too_many_arguments)]
266    #[pyo3(signature = (storage_account, sas_token, container, compression, file_type, max_retry, retry_interval_sec, blob_prefix=None))]
267    pub fn new(
268        storage_account: String,
269        sas_token: String,
270        container: String,
271        compression: String,
272        file_type: String,
273        max_retry: i32,
274        retry_interval_sec: i32,
275        blob_prefix: Option<String>,
276    ) -> Self {
277        Self {
278            storage_account,
279            sas_token,
280            container,
281            compression,
282            file_type,
283            max_retry,
284            retry_interval_sec,
285            blob_prefix,
286        }
287    }
288}
289
290/// Configuration for delivering stream batches to a PostgreSQL database.
291#[cfg_attr(feature = "python", gen_stub_pyclass)]
292#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
293#[cfg_attr(feature = "node", napi(object))]
294#[derive(Debug, Clone, Serialize, Deserialize)]
295pub struct PostgresAttributes {
296    /// Database host.
297    pub host: String,
298    /// Database port.
299    pub port: i32,
300    /// Database name.
301    pub database: String,
302    /// Username used to authenticate.
303    pub username: String,
304    /// Password used to authenticate.
305    pub password: String,
306    /// Destination table for inserted rows.
307    pub table_name: String,
308    /// Postgres SSL mode (e.g. `disable`, `require`, `verify-full`).
309    pub sslmode: String,
310    /// Maximum number of retry attempts for a failed write.
311    pub max_retry: i32,
312    /// Seconds to wait between retry attempts.
313    pub retry_interval_sec: i32,
314}
315
316#[cfg(feature = "python")]
317#[gen_stub_pymethods]
318#[pymethods]
319impl PostgresAttributes {
320    #[new]
321    #[allow(clippy::too_many_arguments)]
322    pub fn new(
323        host: String,
324        port: i32,
325        database: String,
326        username: String,
327        password: String,
328        table_name: String,
329        sslmode: String,
330        max_retry: i32,
331        retry_interval_sec: i32,
332    ) -> Self {
333        Self {
334            host,
335            port,
336            database,
337            username,
338            password,
339            table_name,
340            sslmode,
341            max_retry,
342            retry_interval_sec,
343        }
344    }
345}
346
347/// Configuration for delivering stream batches to a MySQL database.
348#[cfg_attr(feature = "python", gen_stub_pyclass)]
349#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
350#[cfg_attr(feature = "node", napi(object))]
351#[derive(Debug, Clone, Serialize, Deserialize)]
352pub struct MysqlAttributes {
353    /// Database host.
354    pub host: String,
355    /// Database port.
356    pub port: i32,
357    /// Database name.
358    pub database: String,
359    /// Username used to authenticate.
360    pub username: String,
361    /// Password used to authenticate.
362    pub password: String,
363    /// Destination table for inserted rows.
364    pub table_name: String,
365    /// Maximum number of retry attempts for a failed write.
366    pub max_retry: i32,
367    /// Seconds to wait between retry attempts.
368    pub retry_interval_sec: i32,
369}
370
371#[cfg(feature = "python")]
372#[gen_stub_pymethods]
373#[pymethods]
374impl MysqlAttributes {
375    #[new]
376    #[allow(clippy::too_many_arguments)]
377    pub fn new(
378        host: String,
379        port: i32,
380        database: String,
381        username: String,
382        password: String,
383        table_name: String,
384        max_retry: i32,
385        retry_interval_sec: i32,
386    ) -> Self {
387        Self {
388            host,
389            port,
390            database,
391            username,
392            password,
393            table_name,
394            max_retry,
395            retry_interval_sec,
396        }
397    }
398}
399
400/// Configuration for delivering stream batches to a MongoDB database.
401#[cfg_attr(feature = "python", gen_stub_pyclass)]
402#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
403#[cfg_attr(feature = "node", napi(object))]
404#[derive(Debug, Clone, Serialize, Deserialize)]
405pub struct MongoAttributes {
406    /// Database host (connection string or hostname).
407    pub host: String,
408    /// Database name.
409    pub database: String,
410    /// Username used to authenticate.
411    pub username: String,
412    /// Password used to authenticate.
413    pub password: String,
414    /// Destination collection for inserted documents.
415    pub collection_name: String,
416    /// Maximum number of retry attempts for a failed write.
417    pub max_retry: i32,
418    /// Seconds to wait between retry attempts.
419    pub retry_interval_sec: i32,
420}
421
422#[cfg(feature = "python")]
423#[gen_stub_pymethods]
424#[pymethods]
425impl MongoAttributes {
426    #[new]
427    pub fn new(
428        host: String,
429        database: String,
430        username: String,
431        password: String,
432        collection_name: String,
433        max_retry: i32,
434        retry_interval_sec: i32,
435    ) -> Self {
436        Self {
437            host,
438            database,
439            username,
440            password,
441            collection_name,
442            max_retry,
443            retry_interval_sec,
444        }
445    }
446}
447
448/// Configuration for delivering stream batches to a ClickHouse cluster.
449#[cfg_attr(feature = "python", gen_stub_pyclass)]
450#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
451#[cfg_attr(feature = "node", napi(object))]
452#[derive(Debug, Clone, Serialize, Deserialize)]
453pub struct ClickhouseAttributes {
454    /// Comma-separated list of ClickHouse hosts.
455    pub hosts: String,
456    /// Database name.
457    pub database: String,
458    /// Username used to authenticate.
459    pub username: String,
460    /// Password used to authenticate.
461    pub password: String,
462    /// Destination table for inserted rows.
463    pub table_name: String,
464    /// Default table engine options applied when a table is created.
465    pub default_table_engine_opts: String,
466    /// Default index granularity for created tables.
467    pub default_granularity: i32,
468    /// Default compression codec for created tables.
469    pub default_compression: String,
470    /// Default secondary index type for created tables.
471    pub default_index_type: String,
472    /// Maximum number of retry attempts for a failed write.
473    pub max_retry: i32,
474    /// Seconds to wait between retry attempts.
475    pub retry_interval_sec: i32,
476    /// Disable datetime precision for older ClickHouse versions that don't support it.
477    #[serde(skip_serializing_if = "Option::is_none")]
478    pub disable_datetime_precision: Option<bool>,
479    /// Enable when the target ClickHouse server does not support `RENAME COLUMN`.
480    #[serde(skip_serializing_if = "Option::is_none")]
481    pub dont_support_rename_column: Option<bool>,
482    /// Enable when the target ClickHouse server does not support empty default values.
483    #[serde(skip_serializing_if = "Option::is_none")]
484    pub dont_support_empty_default_value: Option<bool>,
485    /// Skip writing version metadata during initialization.
486    #[serde(skip_serializing_if = "Option::is_none")]
487    pub skip_initialize_with_version: Option<bool>,
488}
489
490#[cfg(feature = "python")]
491#[gen_stub_pymethods]
492#[pymethods]
493impl ClickhouseAttributes {
494    #[new]
495    #[pyo3(signature = (hosts, database, username, password, table_name, default_table_engine_opts, default_granularity, default_compression, default_index_type, max_retry, retry_interval_sec, disable_datetime_precision=None, dont_support_rename_column=None, dont_support_empty_default_value=None, skip_initialize_with_version=None))]
496    #[allow(clippy::too_many_arguments)]
497    pub fn new(
498        hosts: String,
499        database: String,
500        username: String,
501        password: String,
502        table_name: String,
503        default_table_engine_opts: String,
504        default_granularity: i32,
505        default_compression: String,
506        default_index_type: String,
507        max_retry: i32,
508        retry_interval_sec: i32,
509        disable_datetime_precision: Option<bool>,
510        dont_support_rename_column: Option<bool>,
511        dont_support_empty_default_value: Option<bool>,
512        skip_initialize_with_version: Option<bool>,
513    ) -> Self {
514        Self {
515            hosts,
516            database,
517            username,
518            password,
519            table_name,
520            default_table_engine_opts,
521            default_granularity,
522            default_compression,
523            default_index_type,
524            max_retry,
525            retry_interval_sec,
526            disable_datetime_precision,
527            dont_support_rename_column,
528            dont_support_empty_default_value,
529            skip_initialize_with_version,
530        }
531    }
532}
533
534/// Configuration for delivering stream batches to a Snowflake data warehouse.
535#[cfg_attr(feature = "python", gen_stub_pyclass)]
536#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
537#[cfg_attr(feature = "node", napi(object))]
538#[derive(Debug, Clone, Serialize, Deserialize)]
539pub struct SnowflakeAttributes {
540    /// Snowflake account identifier.
541    pub account: String,
542    /// Snowflake host.
543    pub host: String,
544    /// Snowflake port.
545    pub port: i32,
546    /// Connection protocol (e.g. `https`).
547    pub protocol: String,
548    /// Database name.
549    pub database: String,
550    /// Schema within the database.
551    pub schema: String,
552    /// Warehouse used to run inserts.
553    pub warehouse: String,
554    /// Username used to authenticate.
555    pub username: String,
556    /// Password used to authenticate.
557    pub password: String,
558    /// Maximum number of retry attempts for a failed write.
559    pub max_retry: i32,
560    /// Seconds to wait between retry attempts.
561    pub retry_interval_sec: i32,
562    /// Optional destination table for inserted rows.
563    #[serde(skip_serializing_if = "Option::is_none")]
564    pub table_name: Option<String>,
565}
566
567#[cfg(feature = "python")]
568#[gen_stub_pymethods]
569#[pymethods]
570impl SnowflakeAttributes {
571    #[new]
572    #[pyo3(signature = (account, host, port, protocol, database, schema, warehouse, username, password, max_retry, retry_interval_sec, table_name=None))]
573    #[allow(clippy::too_many_arguments)]
574    pub fn new(
575        account: String,
576        host: String,
577        port: i32,
578        protocol: String,
579        database: String,
580        schema: String,
581        warehouse: String,
582        username: String,
583        password: String,
584        max_retry: i32,
585        retry_interval_sec: i32,
586        table_name: Option<String>,
587    ) -> Self {
588        Self {
589            account,
590            host,
591            port,
592            protocol,
593            database,
594            schema,
595            warehouse,
596            username,
597            password,
598            max_retry,
599            retry_interval_sec,
600            table_name,
601        }
602    }
603}
604
605/// Configuration for delivering stream batches to a Kafka topic.
606#[cfg_attr(feature = "python", gen_stub_pyclass)]
607#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
608#[cfg_attr(feature = "node", napi(object))]
609#[derive(Debug, Clone, Serialize, Deserialize)]
610pub struct KafkaAttributes {
611    /// Comma-separated list of Kafka broker addresses (host:port).
612    pub bootstrap_servers: String,
613    /// Destination topic.
614    pub topic_name: String,
615    /// Compression codec applied to produced messages (e.g. `none`, `gzip`).
616    pub compression_type: String,
617    /// Maximum number of messages grouped per produce request.
618    pub batch_size: i32,
619    /// Milliseconds the producer waits to batch additional messages.
620    pub linger_ms: i32,
621    /// Maximum request size in bytes.
622    pub max_request_size: i32,
623    /// Request timeout in seconds.
624    pub timeout_sec: i32,
625    /// Maximum number of retry attempts for a failed produce.
626    pub max_retry: i32,
627    /// Seconds to wait between retry attempts.
628    pub retry_interval_sec: i32,
629    /// Optional SASL username.
630    #[serde(skip_serializing_if = "Option::is_none")]
631    pub username: Option<String>,
632    /// Optional SASL password.
633    #[serde(skip_serializing_if = "Option::is_none")]
634    pub password: Option<String>,
635    /// Optional security protocol (e.g. `SASL_SSL`).
636    #[serde(skip_serializing_if = "Option::is_none")]
637    pub protocol: Option<String>,
638    /// Optional SASL mechanism (e.g. `PLAIN`, `SCRAM-SHA-256`).
639    #[serde(skip_serializing_if = "Option::is_none")]
640    pub mechanisms: Option<String>,
641}
642
643#[cfg(feature = "python")]
644#[gen_stub_pymethods]
645#[pymethods]
646impl KafkaAttributes {
647    #[new]
648    #[pyo3(signature = (bootstrap_servers, topic_name, compression_type, batch_size, linger_ms, max_request_size, timeout_sec, max_retry, retry_interval_sec, username=None, password=None, protocol=None, mechanisms=None))]
649    #[allow(clippy::too_many_arguments)]
650    pub fn new(
651        bootstrap_servers: String,
652        topic_name: String,
653        compression_type: String,
654        batch_size: i32,
655        linger_ms: i32,
656        max_request_size: i32,
657        timeout_sec: i32,
658        max_retry: i32,
659        retry_interval_sec: i32,
660        username: Option<String>,
661        password: Option<String>,
662        protocol: Option<String>,
663        mechanisms: Option<String>,
664    ) -> Self {
665        Self {
666            bootstrap_servers,
667            topic_name,
668            compression_type,
669            batch_size,
670            linger_ms,
671            max_request_size,
672            timeout_sec,
673            max_retry,
674            retry_interval_sec,
675            username,
676            password,
677            protocol,
678            mechanisms,
679        }
680    }
681}
682
683/// Configuration for delivering stream batches to a Redis instance.
684#[cfg_attr(feature = "python", gen_stub_pyclass)]
685#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
686#[cfg_attr(feature = "node", napi(object))]
687#[derive(Debug, Clone, Serialize, Deserialize)]
688pub struct RedisAttributes {
689    /// Redis host.
690    pub host: String,
691    /// Redis port.
692    pub port: i32,
693    /// Redis logical database index.
694    pub database: i32,
695    /// Username used to authenticate.
696    pub username: String,
697    /// Password used to authenticate.
698    pub password: String,
699    /// Redis key that receives written payloads.
700    pub key_name: String,
701    /// Maximum number of retry attempts for a failed write.
702    pub max_retry: i32,
703    /// Seconds to wait between retry attempts.
704    pub retry_interval_sec: i32,
705    /// Whether to connect over TLS.
706    #[serde(skip_serializing_if = "Option::is_none")]
707    pub tls: Option<bool>,
708}
709
710#[cfg(feature = "python")]
711#[gen_stub_pymethods]
712#[pymethods]
713impl RedisAttributes {
714    #[new]
715    #[pyo3(signature = (host, port, database, username, password, key_name, max_retry, retry_interval_sec, tls=None))]
716    #[allow(clippy::too_many_arguments)]
717    pub fn new(
718        host: String,
719        port: i32,
720        database: i32,
721        username: String,
722        password: String,
723        key_name: String,
724        max_retry: i32,
725        retry_interval_sec: i32,
726        tls: Option<bool>,
727    ) -> Self {
728        Self {
729            host,
730            port,
731            database,
732            username,
733            password,
734            key_name,
735            max_retry,
736            retry_interval_sec,
737            tls,
738        }
739    }
740}
741
742// ── Address Book Config ────────────────────────────────────────────────────
743
744/// Links a stream's filter to an address book so JSON paths resolve against its
745/// managed address set.
746#[cfg_attr(feature = "python", gen_stub_pyclass)]
747#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
748#[cfg_attr(feature = "node", napi(object))]
749#[derive(Debug, Clone, Serialize, Deserialize)]
750pub struct AddressBookConfig {
751    /// Identifier of the address book to use.
752    pub address_book_id: String,
753    /// Optional JSON path that resolves to an object whose fields are matched against the book.
754    #[serde(skip_serializing_if = "Option::is_none")]
755    pub objects_filter_path: Option<String>,
756    /// JSON paths whose resolved values are matched against the book's addresses.
757    pub elements_filter_paths: Vec<String>,
758}
759
760#[cfg(feature = "python")]
761#[gen_stub_pymethods]
762#[pymethods]
763impl AddressBookConfig {
764    #[new]
765    #[pyo3(signature = (address_book_id, elements_filter_paths, objects_filter_path=None))]
766    pub fn new(
767        address_book_id: String,
768        elements_filter_paths: Vec<String>,
769        objects_filter_path: Option<String>,
770    ) -> Self {
771        Self {
772            address_book_id,
773            objects_filter_path,
774            elements_filter_paths,
775        }
776    }
777}
778
779// ── Destination Attributes ─────────────────────────────────────────────────
780
781/// Destination-specific configuration for a stream. Exactly one variant
782/// selects where and how batches are delivered.
783// Pure-Rust discriminated union; no #[pyclass] / #[napi(object)] because PyO3
784// and napi-rs cannot represent enum-with-data. Each language binding crate
785// wraps this type for its own FFI surface.
786// The serde tag/content pair matches the API wire format when flattened into
787// a request/response struct.
788#[derive(Debug, Clone, Serialize, Deserialize)]
789#[serde(
790    tag = "destination",
791    content = "destination_attributes",
792    rename_all = "snake_case"
793)]
794pub enum DestinationAttributes {
795    /// HTTP webhook endpoint that receives batches in real time.
796    Webhook(WebhookAttributes),
797    /// S3-compatible object storage for archival or batch processing.
798    S3(S3Attributes),
799    /// Azure Blob Storage destination.
800    Azure(AzureAttributes),
801    /// PostgreSQL database destination.
802    Postgres(PostgresAttributes),
803    /// MySQL database destination.
804    Mysql(MysqlAttributes),
805    /// MongoDB database destination.
806    Mongo(MongoAttributes),
807    /// ClickHouse analytics database destination.
808    Clickhouse(ClickhouseAttributes),
809    /// Snowflake data warehouse destination.
810    Snowflake(SnowflakeAttributes),
811    /// Kafka topic destination.
812    Kafka(KafkaAttributes),
813    /// Redis in-memory data store destination.
814    Redis(RedisAttributes),
815}
816
817impl DestinationAttributes {
818    pub fn tag(&self) -> StreamDestination {
819        match self {
820            Self::Webhook(_) => StreamDestination::Webhook,
821            Self::S3(_) => StreamDestination::S3,
822            Self::Azure(_) => StreamDestination::Azure,
823            Self::Postgres(_) => StreamDestination::Postgres,
824            Self::Mysql(_) => StreamDestination::Mysql,
825            Self::Mongo(_) => StreamDestination::Mongo,
826            Self::Clickhouse(_) => StreamDestination::Clickhouse,
827            Self::Snowflake(_) => StreamDestination::Snowflake,
828            Self::Kafka(_) => StreamDestination::Kafka,
829            Self::Redis(_) => StreamDestination::Redis,
830        }
831    }
832}
833
834// ── Request (public-facing) ────────────────────────────────────────────────
835
836/// Parameters for creating a new stream.
837#[cfg_attr(feature = "rust", derive(Builder))]
838#[derive(Debug, Clone, Serialize, Deserialize)]
839pub struct CreateStreamParams {
840    /// Human-readable label identifying the stream.
841    pub name: String,
842    /// Geographic region where the stream runs.
843    pub region: StreamRegion,
844    /// Blockchain network to stream from (e.g. `ethereum-mainnet`).
845    pub network: String,
846    /// Type of on-chain data to stream.
847    pub dataset: StreamDataset,
848    /// Block number to begin streaming from.
849    pub start_range: i64,
850    /// Block number to stop streaming at; `-1` for continuous operation.
851    pub end_range: i64,
852    /// Destination-specific configuration (webhook URL, S3 bucket, DB credentials, etc.).
853    // Flattening the enum's tag/content produces { destination, destination_attributes }.
854    #[serde(flatten)]
855    pub destination_attributes: DestinationAttributes,
856    /// Billing plan associated with the stream.
857    pub plan: String,
858    /// Buffer size used by the stream fetcher before delivery.
859    pub threshold_fetch_buffer: i64,
860    /// Number of blocks grouped together per delivered batch.
861    #[serde(skip_serializing_if = "Option::is_none")]
862    pub dataset_batch_size: Option<i64>,
863    /// Upper bound on batch size when elastic batching is enabled.
864    #[serde(skip_serializing_if = "Option::is_none")]
865    pub max_batch_size: Option<i64>,
866    /// Maximum number of buffered blocks waiting to be processed.
867    #[serde(skip_serializing_if = "Option::is_none")]
868    pub max_buffer_range_size: Option<i64>,
869    /// Maximum number of worker threads processing buffered batches.
870    #[serde(skip_serializing_if = "Option::is_none")]
871    pub max_buffer_processing_workers: Option<i64>,
872    /// Number of blocks to stay behind the chain tip to reduce exposure to reorgs.
873    #[serde(skip_serializing_if = "Option::is_none")]
874    pub keep_distance_from_tip: Option<i64>,
875    /// Base64-encoded filter function applied to each batch before delivery.
876    #[serde(skip_serializing_if = "Option::is_none")]
877    pub filter_function: Option<String>,
878    /// Language the filter function is written in.
879    #[serde(skip_serializing_if = "Option::is_none")]
880    pub filter_language: Option<FilterLanguage>,
881    /// Optional address book to evaluate the filter against.
882    #[serde(skip_serializing_if = "Option::is_none")]
883    pub address_book_config: Option<AddressBookConfig>,
884    /// Where to include stream metadata in delivered payloads.
885    #[serde(skip_serializing_if = "Option::is_none")]
886    pub include_stream_metadata: Option<StreamMetadataLocation>,
887    /// Billing product type the stream is associated with.
888    #[serde(skip_serializing_if = "Option::is_none")]
889    pub product_type: Option<ProductType>,
890    /// Initial stream state (`active` or `paused`). Defaults to `active` when omitted.
891    #[serde(skip_serializing_if = "Option::is_none")]
892    pub status: Option<StreamStatus>,
893    /// Email address that receives stream termination or failure alerts.
894    #[serde(skip_serializing_if = "Option::is_none")]
895    pub notification_email: Option<String>,
896    /// Minimum charge cap applied to the stream's billing.
897    #[serde(skip_serializing_if = "Option::is_none")]
898    pub charge_min_cap: Option<i32>,
899    /// Flag (0 or 1) enabling automatic re-streaming of blocks affected by chain reorganizations.
900    #[serde(skip_serializing_if = "Option::is_none")]
901    pub fix_block_reorgs: Option<i32>,
902    /// When enabled, batch size is reduced toward 1 as the stream catches up to the chain tip.
903    #[serde(skip_serializing_if = "Option::is_none")]
904    pub elastic_batch_enabled: Option<bool>,
905    /// Additional destinations that receive the same batches alongside the primary.
906    // Not flattened: each element serializes as its own {destination, destination_attributes} pair.
907    #[serde(skip_serializing_if = "Option::is_none")]
908    pub extra_destinations: Option<Vec<DestinationAttributes>>,
909}
910
911// ── Response ───────────────────────────────────────────────────────────────
912
913/// A stream's full configuration and current state, as returned by the API.
914#[derive(Debug, Clone, Serialize, Deserialize)]
915pub struct Stream {
916    /// Unique stream identifier.
917    pub id: String,
918    /// Human-readable stream name.
919    pub name: String,
920    /// Current operational state (e.g. `active`, `paused`).
921    pub status: String,
922    /// Timestamp when the stream was created.
923    pub created_at: String,
924    /// Timestamp of the most recent modification.
925    pub updated_at: String,
926    /// Sequence number tracking stream progress.
927    pub sequence: i64,
928    /// Blockchain network the stream is reading from.
929    pub network: String,
930    /// Dataset being streamed.
931    pub dataset: String,
932    /// Geographic region where the stream runs.
933    pub region: String,
934    /// Starting block for the stream.
935    pub start_range: i64,
936    /// Ending block for the stream; `-1` indicates continuous operation.
937    pub end_range: i64,
938    /// Billing plan associated with the stream.
939    #[serde(skip_serializing_if = "Option::is_none")]
940    pub plan: Option<String>,
941    /// Buffer size used by the stream fetcher before delivery.
942    #[serde(skip_serializing_if = "Option::is_none")]
943    pub threshold_fetch_buffer: Option<i64>,
944    /// Number of blocks grouped together per delivered batch.
945    #[serde(skip_serializing_if = "Option::is_none")]
946    pub dataset_batch_size: Option<i64>,
947    /// Upper bound on batch size when elastic batching is enabled.
948    #[serde(skip_serializing_if = "Option::is_none")]
949    pub max_batch_size: Option<i64>,
950    /// Maximum number of buffered blocks waiting to be processed.
951    #[serde(skip_serializing_if = "Option::is_none")]
952    pub max_buffer_range_size: Option<i64>,
953    /// Maximum number of worker threads processing buffered batches.
954    #[serde(skip_serializing_if = "Option::is_none")]
955    pub max_buffer_processing_workers: Option<i64>,
956    /// Number of blocks the stream stays behind the chain tip.
957    #[serde(skip_serializing_if = "Option::is_none")]
958    pub keep_distance_from_tip: Option<i64>,
959    /// Base64-encoded filter function applied to each batch.
960    #[serde(skip_serializing_if = "Option::is_none")]
961    pub filter_function: Option<String>,
962    /// Language the filter function is written in.
963    #[serde(skip_serializing_if = "Option::is_none")]
964    pub filter_language: Option<String>,
965    /// Where stream metadata is included in delivered payloads.
966    #[serde(skip_serializing_if = "Option::is_none")]
967    pub include_stream_metadata: Option<String>,
968    /// Billing product type the stream is associated with.
969    #[serde(skip_serializing_if = "Option::is_none")]
970    pub product_type: Option<String>,
971    /// Email address notified of stream termination or failure.
972    #[serde(skip_serializing_if = "Option::is_none")]
973    pub notification_email: Option<String>,
974    /// Whether chain-reorg handling is enabled (0 or 1).
975    #[serde(skip_serializing_if = "Option::is_none")]
976    pub fix_block_reorgs: Option<i32>,
977    /// Most recent block hash processed by the stream.
978    #[serde(skip_serializing_if = "Option::is_none")]
979    pub current_hash: Option<String>,
980    /// Destination-specific configuration (present on single-stream responses).
981    // Optional because partial responses (e.g. list) may omit the destination pair.
982    #[serde(flatten, default, skip_serializing_if = "Option::is_none")]
983    pub destination_attributes: Option<DestinationAttributes>,
984    /// Whether elastic batching is active.
985    #[serde(skip_serializing_if = "Option::is_none")]
986    pub elastic_batch_enabled: Option<bool>,
987    /// Quicknode account ID that owns the stream.
988    #[serde(skip_serializing_if = "Option::is_none")]
989    pub qn_account_id: Option<String>,
990    /// Minimum charge cap applied to the stream's billing.
991    #[serde(skip_serializing_if = "Option::is_none")]
992    pub charge_min_cap: Option<i32>,
993    /// Free-text memo attached to the stream.
994    #[serde(skip_serializing_if = "Option::is_none")]
995    pub memo: Option<String>,
996    /// Address book linked to the stream's filter, if any.
997    #[serde(skip_serializing_if = "Option::is_none")]
998    pub address_book_config: Option<AddressBookConfig>,
999    /// Additional destinations receiving the same batches alongside the primary.
1000    #[serde(default, skip_serializing_if = "Option::is_none")]
1001    pub extra_destinations: Option<Vec<DestinationAttributes>>,
1002}
1003
1004// ── New Request/Response Types ─────────────────────────────────────────────
1005
1006/// Pagination metadata returned alongside a paginated result set.
1007#[cfg_attr(feature = "python", gen_stub_pyclass)]
1008#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
1009#[cfg_attr(feature = "node", napi(object))]
1010#[derive(Debug, Clone, Serialize, Deserialize)]
1011pub struct PageInfo {
1012    /// Page size used for this response.
1013    pub limit: i64,
1014    /// Starting index of this page within the full result set.
1015    pub offset: i64,
1016    /// Total number of items matching the query across all pages.
1017    pub total: i64,
1018}
1019
1020/// Paginated response from `list_streams`.
1021#[derive(Debug, Clone, Serialize, Deserialize)]
1022pub struct ListStreamsResponse {
1023    /// Streams on the current page.
1024    pub data: Vec<Stream>,
1025    /// Pagination metadata for the response.
1026    #[serde(rename = "pageInfo")]
1027    pub page_info: PageInfo,
1028}
1029
1030/// Parameters for `list_streams`.
1031#[cfg_attr(feature = "node", napi(object))]
1032#[cfg_attr(not(feature = "node"), derive(Clone))]
1033#[derive(Debug, Default, Serialize, Deserialize)]
1034pub struct ListStreamsParams {
1035    /// Filter results by stream type.
1036    #[serde(skip_serializing_if = "Option::is_none")]
1037    pub stream_type: Option<String>,
1038    /// Starting index into the result set; defaults to 0.
1039    #[serde(skip_serializing_if = "Option::is_none")]
1040    pub offset: Option<i64>,
1041    /// Maximum number of streams returned.
1042    #[serde(skip_serializing_if = "Option::is_none")]
1043    pub limit: Option<i64>,
1044    /// Field to sort results by.
1045    #[serde(skip_serializing_if = "Option::is_none")]
1046    pub order_by: Option<String>,
1047    /// Sort direction (`asc` or `desc`).
1048    #[serde(skip_serializing_if = "Option::is_none")]
1049    pub order_direction: Option<String>,
1050}
1051
1052/// Parameters for `update_stream`. Only fields that are set are modified;
1053/// omitted fields leave the current value unchanged.
1054#[derive(Debug, Default, Clone, Serialize, Deserialize)]
1055pub struct UpdateStreamParams {
1056    /// New human-readable name.
1057    #[serde(skip_serializing_if = "Option::is_none")]
1058    pub name: Option<String>,
1059    /// New region.
1060    #[serde(skip_serializing_if = "Option::is_none")]
1061    pub region: Option<StreamRegion>,
1062    /// New blockchain network.
1063    #[serde(skip_serializing_if = "Option::is_none")]
1064    pub network: Option<String>,
1065    /// New dataset.
1066    #[serde(skip_serializing_if = "Option::is_none")]
1067    pub dataset: Option<StreamDataset>,
1068    /// New start block.
1069    #[serde(skip_serializing_if = "Option::is_none")]
1070    pub start_range: Option<i64>,
1071    /// New end block; `-1` for continuous operation.
1072    #[serde(skip_serializing_if = "Option::is_none")]
1073    pub end_range: Option<i64>,
1074    /// New primary destination configuration.
1075    // Flattening Option<enum> omits the keys entirely when None.
1076    #[serde(flatten, skip_serializing_if = "Option::is_none")]
1077    pub destination_attributes: Option<DestinationAttributes>,
1078    /// New billing plan.
1079    #[serde(skip_serializing_if = "Option::is_none")]
1080    pub plan: Option<String>,
1081    /// New fetcher buffer threshold.
1082    #[serde(skip_serializing_if = "Option::is_none")]
1083    pub threshold_fetch_buffer: Option<i64>,
1084    /// New batch size.
1085    #[serde(skip_serializing_if = "Option::is_none")]
1086    pub dataset_batch_size: Option<i64>,
1087    /// New upper bound on elastic batch size.
1088    #[serde(skip_serializing_if = "Option::is_none")]
1089    pub max_batch_size: Option<i64>,
1090    /// New maximum buffered block range.
1091    #[serde(skip_serializing_if = "Option::is_none")]
1092    pub max_buffer_range_size: Option<i64>,
1093    /// New maximum number of buffer-processing workers.
1094    #[serde(skip_serializing_if = "Option::is_none")]
1095    pub max_buffer_processing_workers: Option<i64>,
1096    /// New distance from the chain tip.
1097    #[serde(skip_serializing_if = "Option::is_none")]
1098    pub keep_distance_from_tip: Option<i64>,
1099    /// New base64-encoded filter function.
1100    #[serde(skip_serializing_if = "Option::is_none")]
1101    pub filter_function: Option<String>,
1102    /// New filter function language.
1103    #[serde(skip_serializing_if = "Option::is_none")]
1104    pub filter_language: Option<FilterLanguage>,
1105    /// New address book configuration.
1106    #[serde(skip_serializing_if = "Option::is_none")]
1107    pub address_book_config: Option<AddressBookConfig>,
1108    /// New stream-metadata location.
1109    #[serde(skip_serializing_if = "Option::is_none")]
1110    pub include_stream_metadata: Option<StreamMetadataLocation>,
1111    /// New notification email.
1112    #[serde(skip_serializing_if = "Option::is_none")]
1113    pub notification_email: Option<String>,
1114    /// New minimum charge cap.
1115    #[serde(skip_serializing_if = "Option::is_none")]
1116    pub charge_min_cap: Option<i32>,
1117    /// New reorg-handling flag (0 or 1).
1118    #[serde(skip_serializing_if = "Option::is_none")]
1119    pub fix_block_reorgs: Option<i32>,
1120    /// Whether elastic batching is enabled.
1121    #[serde(skip_serializing_if = "Option::is_none")]
1122    pub elastic_batch_enabled: Option<bool>,
1123    /// New operational state.
1124    #[serde(skip_serializing_if = "Option::is_none")]
1125    pub status: Option<StreamStatus>,
1126    /// Free-text memo to attach to the stream.
1127    #[serde(skip_serializing_if = "Option::is_none")]
1128    pub memo: Option<String>,
1129    /// New set of extra destinations.
1130    #[serde(skip_serializing_if = "Option::is_none")]
1131    pub extra_destinations: Option<Vec<DestinationAttributes>>,
1132}
1133
1134/// Parameters for `test_filter`.
1135#[cfg_attr(feature = "node", napi(object))]
1136#[cfg_attr(not(feature = "node"), derive(Clone))]
1137#[derive(Debug, Serialize, Deserialize)]
1138pub struct TestFilterParams {
1139    /// Blockchain network to run the test against (e.g. `ethereum-mainnet`).
1140    pub network: String,
1141    /// Dataset the filter operates on.
1142    pub dataset: StreamDataset,
1143    /// Specific block number to feed into the filter for the test.
1144    pub block: String,
1145    /// Base64-encoded filter function to evaluate.
1146    #[serde(skip_serializing_if = "Option::is_none")]
1147    pub filter_function: Option<String>,
1148    /// Language the filter function is written in.
1149    #[serde(skip_serializing_if = "Option::is_none")]
1150    pub filter_language: Option<FilterLanguage>,
1151    /// Address book linked to the filter, if any.
1152    #[serde(skip_serializing_if = "Option::is_none")]
1153    pub address_book_config: Option<AddressBookConfig>,
1154}
1155
1156/// Result of a `test_filter` call.
1157#[cfg_attr(feature = "python", gen_stub_pyclass)]
1158#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
1159#[cfg_attr(feature = "node", napi(object))]
1160#[derive(Debug, Clone, Serialize, Deserialize)]
1161pub struct TestFilterResponse {
1162    /// Filter output as a JSON string. Shape depends on the dataset and the user's filter function.
1163    #[serde(deserialize_with = "deserialize_as_json_string")]
1164    pub result: String,
1165    /// Log lines emitted by the filter function during evaluation.
1166    pub logs: Vec<String>,
1167}
1168
1169/// Result of `get_enabled_count`.
1170#[cfg_attr(feature = "python", gen_stub_pyclass)]
1171#[cfg_attr(feature = "python", pyclass(get_all, set_all))]
1172#[cfg_attr(feature = "node", napi(object))]
1173#[derive(Debug, Clone, Serialize, Deserialize)]
1174pub struct EnabledCountResponse {
1175    /// Total count of currently enabled streams.
1176    pub total: i64,
1177}
1178
1179#[cfg(test)]
1180#[allow(clippy::unwrap_used)]
1181mod destination_attributes_tests {
1182    use super::*;
1183
1184    #[test]
1185    fn webhook_roundtrip() {
1186        let attrs = DestinationAttributes::Webhook(WebhookAttributes {
1187            url: "https://x.example/hook".to_string(),
1188            max_retry: 3,
1189            retry_interval_sec: 5,
1190            post_timeout_sec: 10,
1191            compression: "none".to_string(),
1192            security_token: None,
1193        });
1194        let json = serde_json::to_string(&attrs).unwrap();
1195        assert!(json.contains(r#""destination":"webhook""#));
1196        assert!(json.contains(r#""url":"https://x.example/hook""#));
1197        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1198        assert!(matches!(parsed, DestinationAttributes::Webhook(_)));
1199        assert!(matches!(parsed.tag(), StreamDestination::Webhook));
1200    }
1201
1202    #[test]
1203    fn s3_roundtrip() {
1204        let attrs = DestinationAttributes::S3(S3Attributes {
1205            endpoint: "s3.amazonaws.com".to_string(),
1206            access_key: "AK".to_string(),
1207            secret_key: "SK".to_string(),
1208            bucket: "b".to_string(),
1209            object_prefix: "p".to_string(),
1210            compression: "none".to_string(),
1211            file_type: "json".to_string(),
1212            max_retry: 3,
1213            retry_interval_sec: 5,
1214            use_ssl: Some(true),
1215        });
1216        let json = serde_json::to_string(&attrs).unwrap();
1217        assert!(json.contains(r#""destination":"s3""#));
1218        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1219        assert!(matches!(parsed, DestinationAttributes::S3(_)));
1220    }
1221
1222    #[test]
1223    fn azure_roundtrip() {
1224        let attrs = DestinationAttributes::Azure(AzureAttributes {
1225            storage_account: "acct".to_string(),
1226            sas_token: "tok".to_string(),
1227            container: "c".to_string(),
1228            compression: "none".to_string(),
1229            file_type: "json".to_string(),
1230            max_retry: 3,
1231            retry_interval_sec: 5,
1232            blob_prefix: None,
1233        });
1234        let json = serde_json::to_string(&attrs).unwrap();
1235        assert!(json.contains(r#""destination":"azure""#));
1236        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1237        assert!(matches!(parsed, DestinationAttributes::Azure(_)));
1238    }
1239
1240    #[test]
1241    fn postgres_roundtrip() {
1242        let attrs = DestinationAttributes::Postgres(PostgresAttributes {
1243            host: "h".to_string(),
1244            port: 5432,
1245            database: "db".to_string(),
1246            username: "u".to_string(),
1247            password: "p".to_string(),
1248            table_name: "t".to_string(),
1249            sslmode: "disable".to_string(),
1250            max_retry: 3,
1251            retry_interval_sec: 5,
1252        });
1253        let json = serde_json::to_string(&attrs).unwrap();
1254        assert!(json.contains(r#""destination":"postgres""#));
1255        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1256        assert!(matches!(parsed, DestinationAttributes::Postgres(_)));
1257    }
1258
1259    #[test]
1260    fn mysql_roundtrip() {
1261        let attrs = DestinationAttributes::Mysql(MysqlAttributes {
1262            host: "h".to_string(),
1263            port: 3306,
1264            database: "db".to_string(),
1265            username: "u".to_string(),
1266            password: "p".to_string(),
1267            table_name: "t".to_string(),
1268            max_retry: 3,
1269            retry_interval_sec: 5,
1270        });
1271        let json = serde_json::to_string(&attrs).unwrap();
1272        assert!(json.contains(r#""destination":"mysql""#));
1273        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1274        assert!(matches!(parsed, DestinationAttributes::Mysql(_)));
1275    }
1276
1277    #[test]
1278    fn mongo_roundtrip() {
1279        let attrs = DestinationAttributes::Mongo(MongoAttributes {
1280            host: "h".to_string(),
1281            database: "db".to_string(),
1282            username: "u".to_string(),
1283            password: "p".to_string(),
1284            collection_name: "c".to_string(),
1285            max_retry: 3,
1286            retry_interval_sec: 5,
1287        });
1288        let json = serde_json::to_string(&attrs).unwrap();
1289        assert!(json.contains(r#""destination":"mongo""#));
1290        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1291        assert!(matches!(parsed, DestinationAttributes::Mongo(_)));
1292    }
1293
1294    #[test]
1295    fn clickhouse_roundtrip() {
1296        let attrs = DestinationAttributes::Clickhouse(ClickhouseAttributes {
1297            hosts: "h".to_string(),
1298            database: "db".to_string(),
1299            username: "u".to_string(),
1300            password: "p".to_string(),
1301            table_name: "t".to_string(),
1302            default_table_engine_opts: "()".to_string(),
1303            default_granularity: 8192,
1304            default_compression: "lz4".to_string(),
1305            default_index_type: "minmax".to_string(),
1306            max_retry: 3,
1307            retry_interval_sec: 5,
1308            disable_datetime_precision: None,
1309            dont_support_rename_column: None,
1310            dont_support_empty_default_value: None,
1311            skip_initialize_with_version: None,
1312        });
1313        let json = serde_json::to_string(&attrs).unwrap();
1314        assert!(json.contains(r#""destination":"clickhouse""#));
1315        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1316        assert!(matches!(parsed, DestinationAttributes::Clickhouse(_)));
1317    }
1318
1319    #[test]
1320    fn snowflake_roundtrip() {
1321        let attrs = DestinationAttributes::Snowflake(SnowflakeAttributes {
1322            account: "acct".to_string(),
1323            host: "h".to_string(),
1324            port: 443,
1325            protocol: "https".to_string(),
1326            database: "db".to_string(),
1327            schema: "s".to_string(),
1328            warehouse: "w".to_string(),
1329            username: "u".to_string(),
1330            password: "p".to_string(),
1331            max_retry: 3,
1332            retry_interval_sec: 5,
1333            table_name: Some("t".to_string()),
1334        });
1335        let json = serde_json::to_string(&attrs).unwrap();
1336        assert!(json.contains(r#""destination":"snowflake""#));
1337        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1338        assert!(matches!(parsed, DestinationAttributes::Snowflake(_)));
1339    }
1340
1341    #[test]
1342    fn kafka_roundtrip() {
1343        let attrs = DestinationAttributes::Kafka(KafkaAttributes {
1344            bootstrap_servers: "host:9092".to_string(),
1345            topic_name: "t".to_string(),
1346            compression_type: "gzip".to_string(),
1347            batch_size: 100,
1348            linger_ms: 10,
1349            max_request_size: 1024,
1350            timeout_sec: 30,
1351            max_retry: 3,
1352            retry_interval_sec: 5,
1353            username: None,
1354            password: None,
1355            protocol: None,
1356            mechanisms: None,
1357        });
1358        let json = serde_json::to_string(&attrs).unwrap();
1359        assert!(json.contains(r#""destination":"kafka""#));
1360        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1361        assert!(matches!(parsed, DestinationAttributes::Kafka(_)));
1362    }
1363
1364    #[test]
1365    fn redis_roundtrip() {
1366        let attrs = DestinationAttributes::Redis(RedisAttributes {
1367            host: "h".to_string(),
1368            port: 6379,
1369            database: 0,
1370            username: "u".to_string(),
1371            password: "p".to_string(),
1372            key_name: "k".to_string(),
1373            max_retry: 3,
1374            retry_interval_sec: 5,
1375            tls: Some(false),
1376        });
1377        let json = serde_json::to_string(&attrs).unwrap();
1378        assert!(json.contains(r#""destination":"redis""#));
1379        let parsed: DestinationAttributes = serde_json::from_str(&json).unwrap();
1380        assert!(matches!(parsed, DestinationAttributes::Redis(_)));
1381    }
1382}
quicknode_sdk/streams/stream.rs

quicknode_sdk/streams/
stream.rs
