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

faucet_cli/
registry.rs
