faucet_cli/
registry.rs

1//! Feature-gated dispatch from a string `type` to a concrete connector.
2//!
3//! Every arm in this file is guarded by the matching `source-*` / `sink-*`
4//! Cargo feature so users can build a slim binary with just the connectors
5//! they need. The string keys here are the public contract of the CLI's
6//! `type:` field in YAML/JSON pipeline configs.
7
8use crate::auth_catalog::{self, AuthCatalog};
9use crate::error::{CliError, CliResult};
10use faucet_core::{Sink, Source};
11use serde::de::DeserializeOwned;
12use serde_json::Value;
13
14/// Build a [`Source`] trait object from a `(kind, config)` pair. When the
15/// config carries `auth: { ref: <name> }`, the named provider is resolved from
16/// `auth` (the catalog) and injected into the connector.
17pub async fn build_source(
18    kind: &str,
19    config: Value,
20    auth: &AuthCatalog,
21    retry_policy: Option<&faucet_core::RetryPolicy>,
22) -> CliResult<Box<dyn Source>> {
23    let auth_ref = auth_catalog::auth_ref(&config);
24    match kind {
25        #[cfg(feature = "source-rest")]
26        "rest" => {
27            let cfg = decode::<faucet_source_rest::RestStreamConfig>("source", "rest", config)?;
28            let mut s = faucet_source_rest::RestStream::new(cfg)?;
29            if let Some(name) = &auth_ref {
30                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
31            }
32            if let Some(rp) = retry_policy {
33                s = s.with_retry_policy(rp.clone());
34            }
35            Ok(Box::new(s))
36        }
37        #[cfg(feature = "source-graphql")]
38        "graphql" => {
39            let cfg =
40                decode::<faucet_source_graphql::GraphqlStreamConfig>("source", "graphql", config)?;
41            let mut s = faucet_source_graphql::GraphqlStream::new(cfg);
42            if let Some(name) = &auth_ref {
43                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
44            }
45            if let Some(rp) = retry_policy {
46                s = s.with_retry_policy(rp.clone());
47            }
48            Ok(Box::new(s))
49        }
50        #[cfg(feature = "source-xml")]
51        "xml" => {
52            let cfg = decode::<faucet_source_xml::XmlStreamConfig>("source", "xml", config)?;
53            let mut s = faucet_source_xml::XmlStream::new(cfg);
54            if let Some(name) = &auth_ref {
55                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
56            }
57            if let Some(rp) = retry_policy {
58                s = s.with_retry_policy(rp.clone());
59            }
60            Ok(Box::new(s))
61        }
62        #[cfg(feature = "source-grpc")]
63        "grpc" => {
64            let cfg = decode::<faucet_source_grpc::GrpcStreamConfig>("source", "grpc", config)?;
65            let mut s = faucet_source_grpc::GrpcStream::new(cfg)?;
66            if let Some(name) = &auth_ref {
67                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
68            }
69            Ok(Box::new(s))
70        }
71        #[cfg(feature = "source-postgres")]
72        "postgres" => {
73            let cfg = decode::<faucet_source_postgres::PostgresSourceConfig>(
74                "source", "postgres", config,
75            )?;
76            Ok(Box::new(
77                faucet_source_postgres::PostgresSource::new(cfg).await?,
78            ))
79        }
80        #[cfg(feature = "source-postgres-cdc")]
81        "postgres-cdc" => {
82            let cfg = decode::<faucet_source_postgres_cdc::PostgresCdcSourceConfig>(
83                "source",
84                "postgres-cdc",
85                config,
86            )?;
87            Ok(Box::new(
88                faucet_source_postgres_cdc::PostgresCdcSource::new(cfg).await?,
89            ))
90        }
91        #[cfg(feature = "source-mysql")]
92        "mysql" => {
93            let cfg = decode::<faucet_source_mysql::MysqlSourceConfig>("source", "mysql", config)?;
94            Ok(Box::new(faucet_source_mysql::MysqlSource::new(cfg).await?))
95        }
96        #[cfg(feature = "source-mssql")]
97        "mssql" => {
98            let cfg = decode::<faucet_source_mssql::MssqlSourceConfig>("source", "mssql", config)?;
99            Ok(Box::new(faucet_source_mssql::MssqlSource::new(cfg).await?))
100        }
101        #[cfg(feature = "source-sqlite")]
102        "sqlite" => {
103            let cfg =
104                decode::<faucet_source_sqlite::SqliteSourceConfig>("source", "sqlite", config)?;
105            Ok(Box::new(
106                faucet_source_sqlite::SqliteSource::new(cfg).await?,
107            ))
108        }
109        #[cfg(feature = "source-s3")]
110        "s3" => {
111            let cfg = decode::<faucet_source_s3::S3SourceConfig>("source", "s3", config)?;
112            Ok(Box::new(faucet_source_s3::S3Source::new(cfg).await?))
113        }
114        #[cfg(feature = "source-mongodb")]
115        "mongodb" => {
116            let cfg =
117                decode::<faucet_source_mongodb::MongoSourceConfig>("source", "mongodb", config)?;
118            Ok(Box::new(
119                faucet_source_mongodb::MongoSource::new(cfg).await?,
120            ))
121        }
122        #[cfg(feature = "source-mongodb-cdc")]
123        "mongodb-cdc" => {
124            let cfg = decode::<faucet_source_mongodb_cdc::MongoCdcSourceConfig>(
125                "source",
126                "mongodb-cdc",
127                config,
128            )?;
129            Ok(Box::new(
130                faucet_source_mongodb_cdc::MongoCdcSource::new(cfg).await?,
131            ))
132        }
133        #[cfg(feature = "source-mysql-cdc")]
134        "mysql-cdc" => {
135            let cfg = decode::<faucet_source_mysql_cdc::MysqlCdcSourceConfig>(
136                "source",
137                "mysql-cdc",
138                config,
139            )?;
140            Ok(Box::new(
141                faucet_source_mysql_cdc::MysqlCdcSource::new(cfg).await?,
142            ))
143        }
144        #[cfg(feature = "source-redis")]
145        "redis" => {
146            let cfg = decode::<faucet_source_redis::RedisSourceConfig>("source", "redis", config)?;
147            Ok(Box::new(faucet_source_redis::RedisSource::new(cfg)?))
148        }
149        #[cfg(feature = "source-webhook")]
150        "webhook" => {
151            let cfg =
152                decode::<faucet_source_webhook::WebhookSourceConfig>("source", "webhook", config)?;
153            Ok(Box::new(faucet_source_webhook::WebhookSource::new(cfg)))
154        }
155        #[cfg(feature = "source-websocket")]
156        "websocket" => {
157            let cfg = decode::<faucet_source_websocket::WebsocketSourceConfig>(
158                "source",
159                "websocket",
160                config,
161            )?;
162            let mut s = faucet_source_websocket::WebsocketSource::new(cfg)?;
163            if let Some(name) = &auth_ref {
164                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
165            }
166            Ok(Box::new(s))
167        }
168        #[cfg(feature = "source-csv")]
169        "csv" => {
170            let cfg = decode::<faucet_source_csv::CsvSourceConfig>("source", "csv", config)?;
171            Ok(Box::new(faucet_source_csv::CsvSource::new(cfg)))
172        }
173        #[cfg(feature = "source-elasticsearch")]
174        "elasticsearch" => {
175            let cfg = decode::<faucet_source_elasticsearch::ElasticsearchSourceConfig>(
176                "source",
177                "elasticsearch",
178                config,
179            )?;
180            let mut s = faucet_source_elasticsearch::ElasticsearchSource::new(cfg)?;
181            if let Some(name) = &auth_ref {
182                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
183            }
184            Ok(Box::new(s))
185        }
186        #[cfg(feature = "source-kafka")]
187        "kafka" => {
188            let cfg = decode::<faucet_source_kafka::KafkaSourceConfig>("source", "kafka", config)?;
189            Ok(Box::new(faucet_source_kafka::KafkaSource::new(cfg).await?))
190        }
191        #[cfg(feature = "source-parquet")]
192        "parquet" => {
193            let cfg =
194                decode::<faucet_source_parquet::ParquetSourceConfig>("source", "parquet", config)?;
195            Ok(Box::new(
196                faucet_source_parquet::ParquetSource::new(cfg).await?,
197            ))
198        }
199        #[cfg(feature = "source-gcs")]
200        "gcs" => {
201            let cfg = decode::<faucet_source_gcs::GcsSourceConfig>("source", "gcs", config)?;
202            Ok(Box::new(faucet_source_gcs::GcsSource::new(cfg).await?))
203        }
204        #[cfg(feature = "source-bigquery")]
205        "bigquery" => {
206            let cfg = decode::<faucet_source_bigquery::BigQuerySourceConfig>(
207                "source", "bigquery", config,
208            )?;
209            Ok(Box::new(
210                faucet_source_bigquery::BigQuerySource::new(cfg).await?,
211            ))
212        }
213        #[cfg(feature = "source-snowflake")]
214        "snowflake" => {
215            let cfg = decode::<faucet_source_snowflake::SnowflakeSourceConfig>(
216                "source",
217                "snowflake",
218                config,
219            )?;
220            let mut s = faucet_source_snowflake::SnowflakeSource::new(cfg)?;
221            if let Some(name) = &auth_ref {
222                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
223            }
224            Ok(Box::new(s))
225        }
226        other => Err(unknown(other, "source", source_kinds())),
227    }
228}
229
230/// Build a [`Sink`] trait object from a `(kind, config)` pair. When the config
231/// carries `auth: { ref: <name> }`, the named provider is resolved from `auth`
232/// (the catalog) and injected into the connector.
233pub async fn build_sink(kind: &str, config: Value, auth: &AuthCatalog) -> CliResult<Box<dyn Sink>> {
234    let auth_ref = auth_catalog::auth_ref(&config);
235    match kind {
236        #[cfg(feature = "sink-bigquery")]
237        "bigquery" => {
238            let cfg =
239                decode::<faucet_sink_bigquery::BigQuerySinkConfig>("sink", "bigquery", config)?;
240            Ok(Box::new(
241                faucet_sink_bigquery::BigQuerySink::new(cfg).await?,
242            ))
243        }
244        #[cfg(feature = "sink-iceberg")]
245        "iceberg" => {
246            let cfg = decode::<faucet_sink_iceberg::IcebergSinkConfig>("sink", "iceberg", config)?;
247            Ok(Box::new(faucet_sink_iceberg::IcebergSink::new(cfg).await?))
248        }
249        #[cfg(feature = "sink-postgres")]
250        "postgres" => {
251            let cfg =
252                decode::<faucet_sink_postgres::PostgresSinkConfig>("sink", "postgres", config)?;
253            Ok(Box::new(
254                faucet_sink_postgres::PostgresSink::new(cfg).await?,
255            ))
256        }
257        #[cfg(feature = "sink-jsonl")]
258        "jsonl" => {
259            let cfg = decode::<faucet_sink_jsonl::JsonlSinkConfig>("sink", "jsonl", config)?;
260            Ok(Box::new(faucet_sink_jsonl::JsonlSink::new(cfg)))
261        }
262        #[cfg(feature = "sink-snowflake")]
263        "snowflake" => {
264            let cfg =
265                decode::<faucet_sink_snowflake::SnowflakeSinkConfig>("sink", "snowflake", config)?;
266            let mut s = faucet_sink_snowflake::SnowflakeSink::new(cfg)?;
267            if let Some(name) = &auth_ref {
268                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
269            }
270            Ok(Box::new(s))
271        }
272        #[cfg(feature = "sink-mysql")]
273        "mysql" => {
274            let cfg = decode::<faucet_sink_mysql::MysqlSinkConfig>("sink", "mysql", config)?;
275            Ok(Box::new(faucet_sink_mysql::MysqlSink::new(cfg).await?))
276        }
277        #[cfg(feature = "sink-mssql")]
278        "mssql" => {
279            let cfg = decode::<faucet_sink_mssql::MssqlSinkConfig>("sink", "mssql", config)?;
280            Ok(Box::new(faucet_sink_mssql::MssqlSink::new(cfg).await?))
281        }
282        #[cfg(feature = "sink-sqlite")]
283        "sqlite" => {
284            let cfg = decode::<faucet_sink_sqlite::SqliteSinkConfig>("sink", "sqlite", config)?;
285            Ok(Box::new(faucet_sink_sqlite::SqliteSink::new(cfg).await?))
286        }
287        #[cfg(feature = "sink-s3")]
288        "s3" => {
289            let cfg = decode::<faucet_sink_s3::S3SinkConfig>("sink", "s3", config)?;
290            Ok(Box::new(faucet_sink_s3::S3Sink::new(cfg).await?))
291        }
292        #[cfg(feature = "sink-mongodb")]
293        "mongodb" => {
294            let cfg = decode::<faucet_sink_mongodb::MongoSinkConfig>("sink", "mongodb", config)?;
295            Ok(Box::new(faucet_sink_mongodb::MongoSink::new(cfg).await?))
296        }
297        #[cfg(feature = "sink-redis")]
298        "redis" => {
299            let cfg = decode::<faucet_sink_redis::RedisSinkConfig>("sink", "redis", config)?;
300            Ok(Box::new(faucet_sink_redis::RedisSink::new(cfg).await?))
301        }
302        #[cfg(feature = "sink-csv")]
303        "csv" => {
304            let cfg = decode::<faucet_sink_csv::CsvSinkConfig>("sink", "csv", config)?;
305            Ok(Box::new(faucet_sink_csv::CsvSink::new(cfg)))
306        }
307        #[cfg(feature = "sink-elasticsearch")]
308        "elasticsearch" => {
309            let cfg = decode::<faucet_sink_elasticsearch::ElasticsearchSinkConfig>(
310                "sink",
311                "elasticsearch",
312                config,
313            )?;
314            let mut s = faucet_sink_elasticsearch::ElasticsearchSink::new(cfg)?;
315            if let Some(name) = &auth_ref {
316                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
317            }
318            Ok(Box::new(s))
319        }
320        #[cfg(feature = "sink-kafka")]
321        "kafka" => {
322            let cfg = decode::<faucet_sink_kafka::KafkaSinkConfig>("sink", "kafka", config)?;
323            Ok(Box::new(faucet_sink_kafka::KafkaSink::new(cfg).await?))
324        }
325        #[cfg(feature = "sink-http")]
326        "http" => {
327            let cfg = decode::<faucet_sink_http::HttpSinkConfig>("sink", "http", config)?;
328            let mut s = faucet_sink_http::HttpSink::new(cfg);
329            if let Some(name) = &auth_ref {
330                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
331            }
332            Ok(Box::new(s))
333        }
334        #[cfg(feature = "sink-stdout")]
335        "stdout" => {
336            let cfg = decode::<faucet_sink_stdout::StdoutSinkConfig>("sink", "stdout", config)?;
337            Ok(Box::new(faucet_sink_stdout::StdoutSink::new(cfg)))
338        }
339        #[cfg(feature = "sink-parquet")]
340        "parquet" => {
341            let cfg = decode::<faucet_sink_parquet::ParquetSinkConfig>("sink", "parquet", config)?;
342            Ok(Box::new(faucet_sink_parquet::ParquetSink::new(cfg).await?))
343        }
344        #[cfg(feature = "sink-gcs")]
345        "gcs" => {
346            let cfg = decode::<faucet_sink_gcs::GcsSinkConfig>("sink", "gcs", config)?;
347            Ok(Box::new(faucet_sink_gcs::GcsSink::new(cfg).await?))
348        }
349        other => Err(unknown(other, "sink", sink_kinds())),
350    }
351}
352
353/// Source connector kinds that deterministically replay (exactly-once-capable).
354/// Mirrors `Source::supports_exactly_once` overrides — keep in sync when a new
355/// source opts in. The single source of truth for both the boolean gate and the
356/// human-readable list shown in error messages (F44).
357pub const EXACTLY_ONCE_SOURCE_KINDS: &[&str] = &["postgres-cdc", "mysql-cdc", "mongodb-cdc"];
358
359/// Sink connector kinds that can durably commit a token atomically with data.
360/// Mirrors `Sink::supports_idempotent_writes` overrides — keep in sync when a
361/// new sink opts in. Single source of truth for the gate + the error-message
362/// list (F44).
363pub const IDEMPOTENT_SINK_KINDS: &[&str] = &[
364    "sqlite", "postgres", "mysql", "mssql", "iceberg", "bigquery", "kafka",
365];
366
367/// Sink kinds that can apply additive/widening DDL via `Sink::evolve_schema`.
368/// Mirrors each sink's `supports_schema_evolution()` override. Iceberg is
369/// intentionally excluded — iceberg-rust 0.9.1 exposes no schema-evolution API (#255).
370pub const SCHEMA_EVOLUTION_SINK_KINDS: &[&str] = &[
371    "postgres",
372    "mysql",
373    "mssql",
374    "sqlite",
375    "bigquery",
376    "elasticsearch",
377];
378
379/// Sink kinds that support `write_mode: upsert|delete`. Mirrors each sink's
380/// `Sink::supported_write_modes()` override. Single source of truth for the gate
381/// + the error-message list (F44).
382pub const UPSERT_SINK_KINDS: &[&str] = &[
383    "postgres",
384    "sqlite",
385    "mysql",
386    "mssql",
387    "mongodb",
388    "elasticsearch",
389    "bigquery",
390];
391
392/// See [`EXACTLY_ONCE_SOURCE_KINDS`].
393pub fn source_supports_exactly_once(kind: &str) -> bool {
394    EXACTLY_ONCE_SOURCE_KINDS.contains(&kind)
395}
396
397/// See [`IDEMPOTENT_SINK_KINDS`].
398pub fn sink_supports_idempotent_writes(kind: &str) -> bool {
399    IDEMPOTENT_SINK_KINDS.contains(&kind)
400}
401
402/// See [`SCHEMA_EVOLUTION_SINK_KINDS`].
403pub fn sink_supports_schema_evolution(kind: &str) -> bool {
404    SCHEMA_EVOLUTION_SINK_KINDS.contains(&kind)
405}
406
407/// Write modes each sink kind supports. Kept in sync with each sink's
408/// `Sink::supported_write_modes()` override via [`UPSERT_SINK_KINDS`].
409pub fn sink_supported_write_modes(kind: &str) -> &'static [faucet_core::WriteMode] {
410    use faucet_core::WriteMode;
411    if UPSERT_SINK_KINDS.contains(&kind) {
412        &[WriteMode::Append, WriteMode::Upsert, WriteMode::Delete]
413    } else {
414        &[WriteMode::Append]
415    }
416}
417
418/// Return the JSON Schema for the named source's config struct.
419pub fn source_schema(kind: &str) -> CliResult<Value> {
420    match kind {
421        #[cfg(feature = "source-rest")]
422        "rest" => Ok(schema::<faucet_source_rest::RestStreamConfig>()),
423        #[cfg(feature = "source-graphql")]
424        "graphql" => Ok(schema::<faucet_source_graphql::GraphqlStreamConfig>()),
425        #[cfg(feature = "source-xml")]
426        "xml" => Ok(schema::<faucet_source_xml::XmlStreamConfig>()),
427        #[cfg(feature = "source-grpc")]
428        "grpc" => Ok(schema::<faucet_source_grpc::GrpcStreamConfig>()),
429        #[cfg(feature = "source-postgres")]
430        "postgres" => Ok(schema::<faucet_source_postgres::PostgresSourceConfig>()),
431        #[cfg(feature = "source-postgres-cdc")]
432        "postgres-cdc" => Ok(schema::<faucet_source_postgres_cdc::PostgresCdcSourceConfig>()),
433        #[cfg(feature = "source-mysql")]
434        "mysql" => Ok(schema::<faucet_source_mysql::MysqlSourceConfig>()),
435        #[cfg(feature = "source-mssql")]
436        "mssql" => Ok(schema::<faucet_source_mssql::MssqlSourceConfig>()),
437        #[cfg(feature = "source-sqlite")]
438        "sqlite" => Ok(schema::<faucet_source_sqlite::SqliteSourceConfig>()),
439        #[cfg(feature = "source-s3")]
440        "s3" => Ok(schema::<faucet_source_s3::S3SourceConfig>()),
441        #[cfg(feature = "source-mongodb")]
442        "mongodb" => Ok(schema::<faucet_source_mongodb::MongoSourceConfig>()),
443        #[cfg(feature = "source-mongodb-cdc")]
444        "mongodb-cdc" => Ok(schema::<faucet_source_mongodb_cdc::MongoCdcSourceConfig>()),
445        #[cfg(feature = "source-mysql-cdc")]
446        "mysql-cdc" => Ok(schema::<faucet_source_mysql_cdc::MysqlCdcSourceConfig>()),
447        #[cfg(feature = "source-redis")]
448        "redis" => Ok(schema::<faucet_source_redis::RedisSourceConfig>()),
449        #[cfg(feature = "source-webhook")]
450        "webhook" => Ok(schema::<faucet_source_webhook::WebhookSourceConfig>()),
451        #[cfg(feature = "source-websocket")]
452        "websocket" => Ok(schema::<faucet_source_websocket::WebsocketSourceConfig>()),
453        #[cfg(feature = "source-csv")]
454        "csv" => Ok(schema::<faucet_source_csv::CsvSourceConfig>()),
455        #[cfg(feature = "source-elasticsearch")]
456        "elasticsearch" => Ok(schema::<
457            faucet_source_elasticsearch::ElasticsearchSourceConfig,
458        >()),
459        #[cfg(feature = "source-kafka")]
460        "kafka" => Ok(schema::<faucet_source_kafka::KafkaSourceConfig>()),
461        #[cfg(feature = "source-parquet")]
462        "parquet" => Ok(schema::<faucet_source_parquet::ParquetSourceConfig>()),
463        #[cfg(feature = "source-gcs")]
464        "gcs" => Ok(schema::<faucet_source_gcs::GcsSourceConfig>()),
465        #[cfg(feature = "source-bigquery")]
466        "bigquery" => Ok(schema::<faucet_source_bigquery::BigQuerySourceConfig>()),
467        #[cfg(feature = "source-snowflake")]
468        "snowflake" => Ok(schema::<faucet_source_snowflake::SnowflakeSourceConfig>()),
469        other => Err(unknown(other, "source", source_kinds())),
470    }
471}
472
473/// Check if a source kind is registered (not unknown or disabled by feature gate).
474pub fn source_exists(kind: &str) -> bool {
475    source_schema(kind).is_ok()
476}
477
478/// Check if a sink kind is registered (not unknown or disabled by feature gate).
479pub fn sink_exists(kind: &str) -> bool {
480    sink_schema(kind).is_ok()
481}
482
483/// Return the JSON Schema for the named sink's config struct.
484pub fn sink_schema(kind: &str) -> CliResult<Value> {
485    match kind {
486        #[cfg(feature = "sink-bigquery")]
487        "bigquery" => Ok(schema::<faucet_sink_bigquery::BigQuerySinkConfig>()),
488        #[cfg(feature = "sink-iceberg")]
489        "iceberg" => Ok(schema::<faucet_sink_iceberg::IcebergSinkConfig>()),
490        #[cfg(feature = "sink-postgres")]
491        "postgres" => Ok(schema::<faucet_sink_postgres::PostgresSinkConfig>()),
492        #[cfg(feature = "sink-jsonl")]
493        "jsonl" => Ok(schema::<faucet_sink_jsonl::JsonlSinkConfig>()),
494        #[cfg(feature = "sink-snowflake")]
495        "snowflake" => Ok(schema::<faucet_sink_snowflake::SnowflakeSinkConfig>()),
496        #[cfg(feature = "sink-mysql")]
497        "mysql" => Ok(schema::<faucet_sink_mysql::MysqlSinkConfig>()),
498        #[cfg(feature = "sink-mssql")]
499        "mssql" => Ok(schema::<faucet_sink_mssql::MssqlSinkConfig>()),
500        #[cfg(feature = "sink-sqlite")]
501        "sqlite" => Ok(schema::<faucet_sink_sqlite::SqliteSinkConfig>()),
502        #[cfg(feature = "sink-s3")]
503        "s3" => Ok(schema::<faucet_sink_s3::S3SinkConfig>()),
504        #[cfg(feature = "sink-mongodb")]
505        "mongodb" => Ok(schema::<faucet_sink_mongodb::MongoSinkConfig>()),
506        #[cfg(feature = "sink-redis")]
507        "redis" => Ok(schema::<faucet_sink_redis::RedisSinkConfig>()),
508        #[cfg(feature = "sink-csv")]
509        "csv" => Ok(schema::<faucet_sink_csv::CsvSinkConfig>()),
510        #[cfg(feature = "sink-elasticsearch")]
511        "elasticsearch" => Ok(schema::<faucet_sink_elasticsearch::ElasticsearchSinkConfig>()),
512        #[cfg(feature = "sink-kafka")]
513        "kafka" => Ok(schema::<faucet_sink_kafka::KafkaSinkConfig>()),
514        #[cfg(feature = "sink-http")]
515        "http" => Ok(schema::<faucet_sink_http::HttpSinkConfig>()),
516        #[cfg(feature = "sink-stdout")]
517        "stdout" => Ok(schema::<faucet_sink_stdout::StdoutSinkConfig>()),
518        #[cfg(feature = "sink-parquet")]
519        "parquet" => Ok(schema::<faucet_sink_parquet::ParquetSinkConfig>()),
520        #[cfg(feature = "sink-gcs")]
521        "gcs" => Ok(schema::<faucet_sink_gcs::GcsSinkConfig>()),
522        other => Err(unknown(other, "sink", sink_kinds())),
523    }
524}
525
526/// One-line summary of every compiled-in source connector. Used by `faucet list`.
527#[allow(clippy::vec_init_then_push)]
528pub fn source_descriptions() -> Vec<(&'static str, &'static str)> {
529    let mut v: Vec<(&'static str, &'static str)> = Vec::new();
530    #[cfg(feature = "source-rest")]
531    v.push(("rest", "REST API source with pagination, auth, transforms"));
532    #[cfg(feature = "source-graphql")]
533    v.push(("graphql", "GraphQL API source with cursor pagination"));
534    #[cfg(feature = "source-xml")]
535    v.push(("xml", "XML / SOAP API source with XML→JSON conversion"));
536    #[cfg(feature = "source-grpc")]
537    v.push(("grpc", "gRPC source with dynamic protobuf"));
538    #[cfg(feature = "source-postgres")]
539    v.push(("postgres", "PostgreSQL query source"));
540    #[cfg(feature = "source-postgres-cdc")]
541    v.push((
542        "postgres-cdc",
543        "PostgreSQL CDC source (logical replication)",
544    ));
545    #[cfg(feature = "source-mysql")]
546    v.push(("mysql", "MySQL query source"));
547    #[cfg(feature = "source-mssql")]
548    v.push(("mssql", "Microsoft SQL Server query source"));
549    #[cfg(feature = "source-sqlite")]
550    v.push(("sqlite", "SQLite query source"));
551    #[cfg(feature = "source-s3")]
552    v.push(("s3", "AWS S3 object source"));
553    #[cfg(feature = "source-mongodb")]
554    v.push(("mongodb", "MongoDB query source"));
555    #[cfg(feature = "source-mongodb-cdc")]
556    v.push(("mongodb-cdc", "MongoDB CDC source (Change Streams)"));
557    #[cfg(feature = "source-mysql-cdc")]
558    v.push(("mysql-cdc", "MySQL CDC source (binlog replication)"));
559    #[cfg(feature = "source-redis")]
560    v.push(("redis", "Redis (streams, lists, keys) source"));
561    #[cfg(feature = "source-webhook")]
562    v.push(("webhook", "Webhook HTTP receiver source"));
563    #[cfg(feature = "source-websocket")]
564    v.push((
565        "websocket",
566        "WebSocket streaming source — connects, subscribes, streams each message as a record",
567    ));
568    #[cfg(feature = "source-csv")]
569    v.push(("csv", "CSV file source"));
570    #[cfg(feature = "source-elasticsearch")]
571    v.push(("elasticsearch", "Elasticsearch search / scroll source"));
572    #[cfg(feature = "source-kafka")]
573    v.push(("kafka", "Apache Kafka consumer (rdkafka). Subscribes to topics and drains messages with idle/max-messages termination."));
574    #[cfg(feature = "source-parquet")]
575    v.push(("parquet", "Apache Parquet file source (local path, glob, or S3). Streams record batches via the Arrow async reader."));
576    #[cfg(feature = "source-gcs")]
577    v.push((
578        "gcs",
579        "Google Cloud Storage source — JSONL, JSON array, or raw text",
580    ));
581    #[cfg(feature = "source-bigquery")]
582    v.push((
583        "bigquery",
584        "Google BigQuery query source (jobs.query + jobs.getQueryResults)",
585    ));
586    #[cfg(feature = "source-snowflake")]
587    v.push((
588        "snowflake",
589        "Snowflake query source (SQL REST API with partition paging)",
590    ));
591    v
592}
593
594/// One-line summary of every compiled-in sink connector. Used by `faucet list`.
595#[allow(clippy::vec_init_then_push)]
596pub fn sink_descriptions() -> Vec<(&'static str, &'static str)> {
597    let mut v: Vec<(&'static str, &'static str)> = Vec::new();
598    #[cfg(feature = "sink-bigquery")]
599    v.push(("bigquery", "Google BigQuery streaming-insert sink"));
600    #[cfg(feature = "sink-iceberg")]
601    v.push((
602        "iceberg",
603        "Apache Iceberg sink (append, REST/Glue/SQL/HMS catalogs)",
604    ));
605    #[cfg(feature = "sink-postgres")]
606    v.push(("postgres", "PostgreSQL sink (JSONB or auto-mapped columns)"));
607    #[cfg(feature = "sink-jsonl")]
608    v.push(("jsonl", "JSON Lines file sink"));
609    #[cfg(feature = "sink-snowflake")]
610    v.push(("snowflake", "Snowflake SQL REST API sink"));
611    #[cfg(feature = "sink-mysql")]
612    v.push(("mysql", "MySQL sink"));
613    #[cfg(feature = "sink-mssql")]
614    v.push((
615        "mssql",
616        "Microsoft SQL Server sink (auto-mapped columns or JSON column)",
617    ));
618    #[cfg(feature = "sink-sqlite")]
619    v.push(("sqlite", "SQLite sink"));
620    #[cfg(feature = "sink-s3")]
621    v.push(("s3", "AWS S3 object sink"));
622    #[cfg(feature = "sink-mongodb")]
623    v.push(("mongodb", "MongoDB insert sink"));
624    #[cfg(feature = "sink-redis")]
625    v.push(("redis", "Redis (streams, lists, key-value) sink"));
626    #[cfg(feature = "sink-csv")]
627    v.push(("csv", "CSV file sink"));
628    #[cfg(feature = "sink-elasticsearch")]
629    v.push(("elasticsearch", "Elasticsearch bulk index sink"));
630    #[cfg(feature = "sink-kafka")]
631    v.push(("kafka", "Apache Kafka producer (rdkafka). FuturesUnordered batched sends with QueueFull retry; supports fixed or per-record topic routing."));
632    #[cfg(feature = "sink-http")]
633    v.push(("http", "HTTP POST sink (individual or array batch)"));
634    #[cfg(feature = "sink-stdout")]
635    v.push(("stdout", "Stdout / stderr sink (JSON Lines, pretty, TSV)"));
636    #[cfg(feature = "sink-parquet")]
637    v.push(("parquet", "Apache Parquet file sink (local path or S3). Schema-inferred, configurable compression, row/byte rollover."));
638    #[cfg(feature = "sink-gcs")]
639    v.push(("gcs", "Google Cloud Storage sink — JSONL files"));
640    v
641}
642
643/// Names of every compiled-in source connector.
644pub fn source_kinds() -> Vec<&'static str> {
645    source_descriptions().into_iter().map(|(k, _)| k).collect()
646}
647
648/// Names of every compiled-in sink connector.
649pub fn sink_kinds() -> Vec<&'static str> {
650    sink_descriptions().into_iter().map(|(k, _)| k).collect()
651}
652
653fn decode<T: DeserializeOwned>(kind: &'static str, name: &str, config: Value) -> CliResult<T> {
654    serde_json::from_value(config).map_err(|e| CliError::InvalidConnectorConfig {
655        kind,
656        name: name.to_owned(),
657        message: scrub_config_error(&e.to_string()),
658    })
659}
660
661/// Sanitise a serde deserialization error before it reaches stderr/logs.
662///
663/// serde_json's `invalid type:` errors echo the offending value as a
664/// double-quoted literal — which can be a secret injected via
665/// `${secret:...}` / `${env:...}`. Replace every double-quoted run with a
666/// placeholder (field/type names use backticks and are preserved for
667/// diagnostics) and cap the length so a huge value can't flood the log
668/// (#78/#38). Note: `${secret:}` is currently an `${env:}` alias with no
669/// at-rest redaction — this only scrubs error *output*.
670fn scrub_config_error(msg: &str) -> String {
671    const MAX_CHARS: usize = 200;
672    let mut out = String::with_capacity(msg.len());
673    let mut in_quote = false;
674    for c in msg.chars() {
675        if c == '"' {
676            if !in_quote {
677                out.push_str("\"<redacted>\"");
678            }
679            in_quote = !in_quote;
680            continue;
681        }
682        if !in_quote {
683            out.push(c);
684        }
685    }
686    if out.chars().count() > MAX_CHARS {
687        let truncated: String = out.chars().take(MAX_CHARS).collect();
688        return format!("{truncated}…");
689    }
690    out
691}
692
693fn schema<T: faucet_core::JsonSchema>() -> Value {
694    serde_json::to_value(faucet_core::schema_for!(T))
695        .unwrap_or_else(|_| serde_json::json!({"type": "object"}))
696}
697
698fn unknown(name: &str, kind: &'static str, available: Vec<&'static str>) -> CliError {
699    CliError::UnknownConnector {
700        kind,
701        name: name.to_owned(),
702        available: if available.is_empty() {
703            "(none — rebuild faucet-cli with the relevant feature enabled)".to_owned()
704        } else {
705            available.join(", ")
706        },
707    }
708}
709
710#[cfg(test)]
711mod tests {
712    use super::*;
713
714    #[test]
715    fn capability_constants_match_their_predicates() {
716        // F44: the human-readable lists in error messages derive from these
717        // constants, which must stay in lockstep with the boolean gates. In
718        // particular the idempotent-sink list must include bigquery AND kafka,
719        // and the upsert-sink list must include bigquery — the values the old
720        // hand-maintained message strings had drifted away from.
721        for &k in EXACTLY_ONCE_SOURCE_KINDS {
722            assert!(
723                source_supports_exactly_once(k),
724                "{k} should be exactly-once"
725            );
726        }
727        for &k in IDEMPOTENT_SINK_KINDS {
728            assert!(
729                sink_supports_idempotent_writes(k),
730                "{k} should be idempotent"
731            );
732        }
733        for &k in UPSERT_SINK_KINDS {
734            use faucet_core::WriteMode;
735            assert!(
736                sink_supported_write_modes(k).contains(&WriteMode::Upsert),
737                "{k} should support upsert"
738            );
739        }
740        assert!(IDEMPOTENT_SINK_KINDS.contains(&"bigquery"));
741        assert!(IDEMPOTENT_SINK_KINDS.contains(&"kafka"));
742        assert!(UPSERT_SINK_KINDS.contains(&"bigquery"));
743    }
744
745    #[cfg(feature = "source-rest")]
746    #[test]
747    fn rest_source_appears_in_listings() {
748        assert!(source_kinds().contains(&"rest"));
749    }
750
751    #[cfg(feature = "sink-jsonl")]
752    #[test]
753    fn jsonl_sink_appears_in_listings() {
754        assert!(sink_kinds().contains(&"jsonl"));
755    }
756
757    #[tokio::test]
758    async fn unknown_source_kind_errors() {
759        let err = build_source("nope", serde_json::json!({}), &AuthCatalog::new(), None)
760            .await
761            .err()
762            .expect("should fail");
763        match err {
764            CliError::UnknownConnector { kind, name, .. } => {
765                assert_eq!(kind, "source");
766                assert_eq!(name, "nope");
767            }
768            other => panic!("expected UnknownConnector, got {other:?}"),
769        }
770    }
771
772    #[tokio::test]
773    async fn unknown_sink_kind_errors() {
774        let err = build_sink("nope", serde_json::json!({}), &AuthCatalog::new())
775            .await
776            .err()
777            .expect("should fail");
778        assert!(matches!(
779            err,
780            CliError::UnknownConnector { kind: "sink", .. }
781        ));
782    }
783
784    #[cfg(feature = "source-rest")]
785    #[test]
786    fn rest_schema_is_object() {
787        let s = source_schema("rest").unwrap();
788        assert!(s.is_object());
789    }
790
791    #[cfg(feature = "sink-jsonl")]
792    #[test]
793    fn jsonl_schema_is_object() {
794        let s = sink_schema("jsonl").unwrap();
795        assert!(s.is_object());
796    }
797
798    #[test]
799    fn scrub_config_error_redacts_quoted_values() {
800        // A serde "invalid type" error echoes the offending value in double
801        // quotes — must be redacted so a secret can't reach the log (#78/#38).
802        let msg =
803            r#"invalid type: string "sk-super-secret-123", expected a sequence at line 1 column 9"#;
804        let scrubbed = scrub_config_error(msg);
805        assert!(!scrubbed.contains("sk-super-secret-123"), "{scrubbed}");
806        assert!(scrubbed.contains("<redacted>"), "{scrubbed}");
807        // Structural context outside the quotes is preserved.
808        assert!(scrubbed.contains("invalid type"), "{scrubbed}");
809        assert!(scrubbed.contains("expected a sequence"), "{scrubbed}");
810    }
811
812    #[test]
813    fn scrub_config_error_truncates_long_messages() {
814        let msg = "x".repeat(500);
815        let scrubbed = scrub_config_error(&msg);
816        assert!(
817            scrubbed.chars().count() <= 201,
818            "len {}",
819            scrubbed.chars().count()
820        );
821        assert!(scrubbed.ends_with('…'));
822    }
823
824    // A `(kind, config)` pair that builds without performing any network/disk
825    // I/O — the CSV source's `new()` only stores config, so we can drive the
826    // real `build_source` dispatch arm and inspect the resulting trait object.
827    #[cfg(feature = "source-csv")]
828    #[tokio::test]
829    async fn build_source_csv_succeeds_without_io() {
830        let src = build_source(
831            "csv",
832            serde_json::json!({ "path": "/tmp/does-not-need-to-exist.csv" }),
833            &AuthCatalog::new(),
834            None,
835        )
836        .await
837        .expect("csv source should build without I/O");
838        // The CSV source uses the default `connector_name()` (stripped type
839        // name) rather than overriding it with a friendly label.
840        assert_eq!(src.connector_name(), "CsvSource");
841    }
842
843    // The JSONL sink's `new()` is also pure (it opens the file lazily on first
844    // write), so building it exercises the sink dispatch arm with no I/O.
845    #[cfg(feature = "sink-jsonl")]
846    #[tokio::test]
847    async fn build_sink_jsonl_succeeds_without_io() {
848        let sink = build_sink(
849            "jsonl",
850            serde_json::json!({ "path": "/tmp/does-not-need-to-exist.jsonl" }),
851            &AuthCatalog::new(),
852        )
853        .await
854        .expect("jsonl sink should build without I/O");
855        assert_eq!(sink.connector_name(), "jsonl");
856    }
857
858    // The stdout sink builds without any config fields and without I/O.
859    #[cfg(feature = "sink-stdout")]
860    #[tokio::test]
861    async fn build_sink_stdout_succeeds_without_io() {
862        let sink = build_sink("stdout", serde_json::json!({}), &AuthCatalog::new())
863            .await
864            .expect("stdout sink should build without I/O");
865        // The stdout sink uses the default `connector_name()` (stripped type
866        // name) rather than overriding it with a friendly label.
867        assert_eq!(sink.connector_name(), "StdoutSink");
868    }
869
870    // A malformed config for a known connector must surface as a typed
871    // `InvalidConnectorConfig` from the `decode` helper, not a panic.
872    #[cfg(feature = "source-csv")]
873    #[tokio::test]
874    async fn build_source_csv_invalid_config_errors() {
875        // `path` is a required String; supplying an integer is a type error.
876        // `Box<dyn Source>` is not `Debug`, so match the Result directly rather
877        // than using `expect_err`.
878        let res = build_source(
879            "csv",
880            serde_json::json!({ "path": 42 }),
881            &AuthCatalog::new(),
882            None,
883        )
884        .await;
885        match res {
886            Err(CliError::InvalidConnectorConfig { kind, name, .. }) => {
887                assert_eq!(kind, "source");
888                assert_eq!(name, "csv");
889            }
890            Ok(_) => panic!("expected InvalidConnectorConfig, got Ok"),
891            Err(other) => panic!("expected InvalidConnectorConfig, got {other:?}"),
892        }
893    }
894
895    // `source_schema` must return a JSON object that surfaces the connector's
896    // config fields (here: the required `path`).
897    #[cfg(feature = "source-csv")]
898    #[test]
899    fn source_schema_csv_exposes_path_property() {
900        let schema = source_schema("csv").expect("csv schema");
901        let props = schema
902            .get("properties")
903            .and_then(Value::as_object)
904            .expect("schema should have a properties object");
905        assert!(props.contains_key("path"), "schema props: {props:?}");
906    }
907
908    #[cfg(feature = "sink-jsonl")]
909    #[test]
910    fn sink_schema_jsonl_exposes_path_property() {
911        let schema = sink_schema("jsonl").expect("jsonl schema");
912        let props = schema
913            .get("properties")
914            .and_then(Value::as_object)
915            .expect("schema should have a properties object");
916        assert!(props.contains_key("path"), "schema props: {props:?}");
917    }
918
919    #[test]
920    fn unknown_source_schema_errors_with_available_list() {
921        let err = source_schema("definitely-not-a-source").expect_err("unknown source");
922        match err {
923            CliError::UnknownConnector {
924                kind,
925                name,
926                available,
927            } => {
928                assert_eq!(kind, "source");
929                assert_eq!(name, "definitely-not-a-source");
930                // Under `--all-features` the available list is non-empty.
931                assert!(!available.is_empty());
932            }
933            other => panic!("expected UnknownConnector, got {other:?}"),
934        }
935    }
936
937    #[test]
938    fn unknown_sink_schema_errors() {
939        let err = sink_schema("definitely-not-a-sink").expect_err("unknown sink");
940        assert!(matches!(
941            err,
942            CliError::UnknownConnector { kind: "sink", .. }
943        ));
944    }
945
946    #[cfg(feature = "source-csv")]
947    #[test]
948    fn source_exists_is_true_for_known_and_false_for_unknown() {
949        assert!(source_exists("csv"));
950        assert!(!source_exists("definitely-not-a-source"));
951    }
952
953    #[cfg(feature = "sink-jsonl")]
954    #[test]
955    fn sink_exists_is_true_for_known_and_false_for_unknown() {
956        assert!(sink_exists("jsonl"));
957        assert!(!sink_exists("definitely-not-a-sink"));
958    }
959
960    // Descriptions back `faucet list`: non-empty, with a one-line summary, and
961    // each name must resolve to a real schema (no orphan listing).
962    #[test]
963    fn source_descriptions_are_non_empty_and_consistent() {
964        let descs = source_descriptions();
965        assert!(!descs.is_empty());
966        for (name, summary) in &descs {
967            assert!(!name.is_empty(), "empty connector name");
968            assert!(!summary.is_empty(), "empty summary for {name}");
969            assert!(
970                source_schema(name).is_ok(),
971                "listed source `{name}` has no schema"
972            );
973        }
974    }
975
976    #[test]
977    fn sink_descriptions_are_non_empty_and_consistent() {
978        let descs = sink_descriptions();
979        assert!(!descs.is_empty());
980        for (name, summary) in &descs {
981            assert!(!name.is_empty(), "empty connector name");
982            assert!(!summary.is_empty(), "empty summary for {name}");
983            assert!(
984                sink_schema(name).is_ok(),
985                "listed sink `{name}` has no schema"
986            );
987        }
988    }
989
990    // `*_kinds()` is derived from `*_descriptions()`; under `--all-features`
991    // the canonical built-in connectors must be present.
992    #[cfg(all(feature = "source-csv", feature = "source-rest"))]
993    #[test]
994    fn source_kinds_contains_expected_builtins() {
995        let kinds = source_kinds();
996        assert!(kinds.contains(&"csv"));
997        assert!(kinds.contains(&"rest"));
998    }
999
1000    #[cfg(all(feature = "sink-jsonl", feature = "sink-stdout"))]
1001    #[test]
1002    fn sink_kinds_contains_expected_builtins() {
1003        let kinds = sink_kinds();
1004        assert!(kinds.contains(&"jsonl"));
1005        assert!(kinds.contains(&"stdout"));
1006    }
1007
1008    // Build a catalog holding one `static` bearer provider, then build a
1009    // connector whose config carries `auth: { ref: "tok" }` — exercising the
1010    // `with_auth_provider` injection branch in the dispatch arm.
1011    #[cfg(feature = "source-rest")]
1012    #[tokio::test]
1013    async fn build_source_injects_referenced_auth_provider() {
1014        let mut specs = std::collections::HashMap::new();
1015        specs.insert(
1016            "tok".to_string(),
1017            serde_json::json!({"type": "static", "config": {"token": "abc"}}),
1018        );
1019        let catalog = auth_catalog::build_auth_catalog(Some(&specs)).expect("catalog");
1020
1021        let src = build_source("rest", rest_config_with_auth_ref("tok"), &catalog, None)
1022            .await
1023            .expect("rest source with a resolvable auth ref should build");
1024        assert_eq!(src.connector_name(), "rest");
1025    }
1026
1027    // A minimal, fully-valid rest config (built from the real constructor so
1028    // every required field is present) carrying an `auth: { ref }` pointer.
1029    #[cfg(feature = "source-rest")]
1030    fn rest_config_with_auth_ref(name: &str) -> Value {
1031        let cfg = faucet_source_rest::RestStreamConfig::new("https://api.example.com", "/v1");
1032        let mut v = serde_json::to_value(cfg).expect("serialize rest config");
1033        v.as_object_mut()
1034            .unwrap()
1035            .insert("auth".to_string(), serde_json::json!({ "ref": name }));
1036        v
1037    }
1038
1039    // An `auth: { ref }` pointing at a name absent from the catalog must surface
1040    // as `UnknownAuthProvider`, not silently build without auth.
1041    #[cfg(feature = "source-rest")]
1042    #[tokio::test]
1043    async fn build_source_unknown_auth_ref_errors() {
1044        let res = build_source(
1045            "rest",
1046            rest_config_with_auth_ref("missing"),
1047            &AuthCatalog::new(),
1048            None,
1049        )
1050        .await;
1051        match res {
1052            Err(CliError::UnknownAuthProvider { name, .. }) => assert_eq!(name, "missing"),
1053            Ok(_) => panic!("expected UnknownAuthProvider, got Ok"),
1054            Err(other) => panic!("expected UnknownAuthProvider, got {other:?}"),
1055        }
1056    }
1057
1058    #[test]
1059    fn exactly_once_capability_allowlists() {
1060        assert!(source_supports_exactly_once("postgres-cdc"));
1061        assert!(source_supports_exactly_once("mysql-cdc"));
1062        assert!(source_supports_exactly_once("mongodb-cdc"));
1063        assert!(!source_supports_exactly_once("rest"));
1064
1065        assert!(sink_supports_idempotent_writes("postgres"));
1066        assert!(sink_supports_idempotent_writes("iceberg"));
1067        assert!(sink_supports_idempotent_writes("bigquery"));
1068        assert!(sink_supports_idempotent_writes("kafka"));
1069        assert!(!sink_supports_idempotent_writes("jsonl"));
1070    }
1071
1072    #[test]
1073    fn sink_supported_write_modes_allowlist() {
1074        use faucet_core::WriteMode;
1075        assert!(sink_supported_write_modes("postgres").contains(&WriteMode::Upsert));
1076        assert!(sink_supported_write_modes("elasticsearch").contains(&WriteMode::Delete));
1077        assert!(sink_supported_write_modes("bigquery").contains(&WriteMode::Upsert));
1078        // a sink without upsert support is append-only
1079        assert_eq!(sink_supported_write_modes("jsonl"), &[WriteMode::Append]);
1080        assert_eq!(sink_supported_write_modes("kafka"), &[WriteMode::Append]);
1081    }
1082
1083    #[test]
1084    fn sink_supports_schema_evolution_allowlist() {
1085        assert!(sink_supports_schema_evolution("postgres"));
1086        assert!(sink_supports_schema_evolution("mysql"));
1087        assert!(sink_supports_schema_evolution("mssql"));
1088        assert!(sink_supports_schema_evolution("sqlite"));
1089        assert!(sink_supports_schema_evolution("bigquery"));
1090        assert!(sink_supports_schema_evolution("elasticsearch"));
1091        assert!(!sink_supports_schema_evolution("iceberg"));
1092        assert!(!sink_supports_schema_evolution("jsonl"));
1093        assert!(!sink_supports_schema_evolution("kafka"));
1094    }
1095}
faucet_cli/registry.rs

faucet_cli/
registry.rs
