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-redis")]
113        "redis" => {
114            let cfg = decode::<faucet_source_redis::RedisSourceConfig>("source", "redis", config)?;
115            Ok(Box::new(faucet_source_redis::RedisSource::new(cfg)?))
116        }
117        #[cfg(feature = "source-webhook")]
118        "webhook" => {
119            let cfg =
120                decode::<faucet_source_webhook::WebhookSourceConfig>("source", "webhook", config)?;
121            Ok(Box::new(faucet_source_webhook::WebhookSource::new(cfg)))
122        }
123        #[cfg(feature = "source-websocket")]
124        "websocket" => {
125            let cfg = decode::<faucet_source_websocket::WebsocketSourceConfig>(
126                "source",
127                "websocket",
128                config,
129            )?;
130            let mut s = faucet_source_websocket::WebsocketSource::new(cfg)?;
131            if let Some(name) = &auth_ref {
132                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
133            }
134            Ok(Box::new(s))
135        }
136        #[cfg(feature = "source-csv")]
137        "csv" => {
138            let cfg = decode::<faucet_source_csv::CsvSourceConfig>("source", "csv", config)?;
139            Ok(Box::new(faucet_source_csv::CsvSource::new(cfg)))
140        }
141        #[cfg(feature = "source-elasticsearch")]
142        "elasticsearch" => {
143            let cfg = decode::<faucet_source_elasticsearch::ElasticsearchSourceConfig>(
144                "source",
145                "elasticsearch",
146                config,
147            )?;
148            let mut s = faucet_source_elasticsearch::ElasticsearchSource::new(cfg)?;
149            if let Some(name) = &auth_ref {
150                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
151            }
152            Ok(Box::new(s))
153        }
154        #[cfg(feature = "source-kafka")]
155        "kafka" => {
156            let cfg = decode::<faucet_source_kafka::KafkaSourceConfig>("source", "kafka", config)?;
157            Ok(Box::new(faucet_source_kafka::KafkaSource::new(cfg).await?))
158        }
159        #[cfg(feature = "source-parquet")]
160        "parquet" => {
161            let cfg =
162                decode::<faucet_source_parquet::ParquetSourceConfig>("source", "parquet", config)?;
163            Ok(Box::new(
164                faucet_source_parquet::ParquetSource::new(cfg).await?,
165            ))
166        }
167        #[cfg(feature = "source-gcs")]
168        "gcs" => {
169            let cfg = decode::<faucet_source_gcs::GcsSourceConfig>("source", "gcs", config)?;
170            Ok(Box::new(faucet_source_gcs::GcsSource::new(cfg).await?))
171        }
172        #[cfg(feature = "source-bigquery")]
173        "bigquery" => {
174            let cfg = decode::<faucet_source_bigquery::BigQuerySourceConfig>(
175                "source", "bigquery", config,
176            )?;
177            Ok(Box::new(
178                faucet_source_bigquery::BigQuerySource::new(cfg).await?,
179            ))
180        }
181        #[cfg(feature = "source-snowflake")]
182        "snowflake" => {
183            let cfg = decode::<faucet_source_snowflake::SnowflakeSourceConfig>(
184                "source",
185                "snowflake",
186                config,
187            )?;
188            let mut s = faucet_source_snowflake::SnowflakeSource::new(cfg)?;
189            if let Some(name) = &auth_ref {
190                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
191            }
192            Ok(Box::new(s))
193        }
194        other => Err(unknown(other, "source", source_kinds())),
195    }
196}
197
198/// Build a [`Sink`] trait object from a `(kind, config)` pair. When the config
199/// carries `auth: { ref: <name> }`, the named provider is resolved from `auth`
200/// (the catalog) and injected into the connector.
201pub async fn build_sink(kind: &str, config: Value, auth: &AuthCatalog) -> CliResult<Box<dyn Sink>> {
202    let auth_ref = auth_catalog::auth_ref(&config);
203    match kind {
204        #[cfg(feature = "sink-bigquery")]
205        "bigquery" => {
206            let cfg =
207                decode::<faucet_sink_bigquery::BigQuerySinkConfig>("sink", "bigquery", config)?;
208            Ok(Box::new(
209                faucet_sink_bigquery::BigQuerySink::new(cfg).await?,
210            ))
211        }
212        #[cfg(feature = "sink-postgres")]
213        "postgres" => {
214            let cfg =
215                decode::<faucet_sink_postgres::PostgresSinkConfig>("sink", "postgres", config)?;
216            Ok(Box::new(
217                faucet_sink_postgres::PostgresSink::new(cfg).await?,
218            ))
219        }
220        #[cfg(feature = "sink-jsonl")]
221        "jsonl" => {
222            let cfg = decode::<faucet_sink_jsonl::JsonlSinkConfig>("sink", "jsonl", config)?;
223            Ok(Box::new(faucet_sink_jsonl::JsonlSink::new(cfg)))
224        }
225        #[cfg(feature = "sink-snowflake")]
226        "snowflake" => {
227            let cfg =
228                decode::<faucet_sink_snowflake::SnowflakeSinkConfig>("sink", "snowflake", config)?;
229            let mut s = faucet_sink_snowflake::SnowflakeSink::new(cfg)?;
230            if let Some(name) = &auth_ref {
231                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
232            }
233            Ok(Box::new(s))
234        }
235        #[cfg(feature = "sink-mysql")]
236        "mysql" => {
237            let cfg = decode::<faucet_sink_mysql::MysqlSinkConfig>("sink", "mysql", config)?;
238            Ok(Box::new(faucet_sink_mysql::MysqlSink::new(cfg).await?))
239        }
240        #[cfg(feature = "sink-mssql")]
241        "mssql" => {
242            let cfg = decode::<faucet_sink_mssql::MssqlSinkConfig>("sink", "mssql", config)?;
243            Ok(Box::new(faucet_sink_mssql::MssqlSink::new(cfg).await?))
244        }
245        #[cfg(feature = "sink-sqlite")]
246        "sqlite" => {
247            let cfg = decode::<faucet_sink_sqlite::SqliteSinkConfig>("sink", "sqlite", config)?;
248            Ok(Box::new(faucet_sink_sqlite::SqliteSink::new(cfg).await?))
249        }
250        #[cfg(feature = "sink-s3")]
251        "s3" => {
252            let cfg = decode::<faucet_sink_s3::S3SinkConfig>("sink", "s3", config)?;
253            Ok(Box::new(faucet_sink_s3::S3Sink::new(cfg).await?))
254        }
255        #[cfg(feature = "sink-mongodb")]
256        "mongodb" => {
257            let cfg = decode::<faucet_sink_mongodb::MongoSinkConfig>("sink", "mongodb", config)?;
258            Ok(Box::new(faucet_sink_mongodb::MongoSink::new(cfg).await?))
259        }
260        #[cfg(feature = "sink-redis")]
261        "redis" => {
262            let cfg = decode::<faucet_sink_redis::RedisSinkConfig>("sink", "redis", config)?;
263            Ok(Box::new(faucet_sink_redis::RedisSink::new(cfg).await?))
264        }
265        #[cfg(feature = "sink-csv")]
266        "csv" => {
267            let cfg = decode::<faucet_sink_csv::CsvSinkConfig>("sink", "csv", config)?;
268            Ok(Box::new(faucet_sink_csv::CsvSink::new(cfg)))
269        }
270        #[cfg(feature = "sink-elasticsearch")]
271        "elasticsearch" => {
272            let cfg = decode::<faucet_sink_elasticsearch::ElasticsearchSinkConfig>(
273                "sink",
274                "elasticsearch",
275                config,
276            )?;
277            let mut s = faucet_sink_elasticsearch::ElasticsearchSink::new(cfg)?;
278            if let Some(name) = &auth_ref {
279                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
280            }
281            Ok(Box::new(s))
282        }
283        #[cfg(feature = "sink-kafka")]
284        "kafka" => {
285            let cfg = decode::<faucet_sink_kafka::KafkaSinkConfig>("sink", "kafka", config)?;
286            Ok(Box::new(faucet_sink_kafka::KafkaSink::new(cfg).await?))
287        }
288        #[cfg(feature = "sink-http")]
289        "http" => {
290            let cfg = decode::<faucet_sink_http::HttpSinkConfig>("sink", "http", config)?;
291            let mut s = faucet_sink_http::HttpSink::new(cfg);
292            if let Some(name) = &auth_ref {
293                s = s.with_auth_provider(auth_catalog::resolve(auth, name)?);
294            }
295            Ok(Box::new(s))
296        }
297        #[cfg(feature = "sink-stdout")]
298        "stdout" => {
299            let cfg = decode::<faucet_sink_stdout::StdoutSinkConfig>("sink", "stdout", config)?;
300            Ok(Box::new(faucet_sink_stdout::StdoutSink::new(cfg)))
301        }
302        #[cfg(feature = "sink-parquet")]
303        "parquet" => {
304            let cfg = decode::<faucet_sink_parquet::ParquetSinkConfig>("sink", "parquet", config)?;
305            Ok(Box::new(faucet_sink_parquet::ParquetSink::new(cfg).await?))
306        }
307        #[cfg(feature = "sink-gcs")]
308        "gcs" => {
309            let cfg = decode::<faucet_sink_gcs::GcsSinkConfig>("sink", "gcs", config)?;
310            Ok(Box::new(faucet_sink_gcs::GcsSink::new(cfg).await?))
311        }
312        other => Err(unknown(other, "sink", sink_kinds())),
313    }
314}
315
316/// Return the JSON Schema for the named source's config struct.
317pub fn source_schema(kind: &str) -> CliResult<Value> {
318    match kind {
319        #[cfg(feature = "source-rest")]
320        "rest" => Ok(schema::<faucet_source_rest::RestStreamConfig>()),
321        #[cfg(feature = "source-graphql")]
322        "graphql" => Ok(schema::<faucet_source_graphql::GraphqlStreamConfig>()),
323        #[cfg(feature = "source-xml")]
324        "xml" => Ok(schema::<faucet_source_xml::XmlStreamConfig>()),
325        #[cfg(feature = "source-grpc")]
326        "grpc" => Ok(schema::<faucet_source_grpc::GrpcStreamConfig>()),
327        #[cfg(feature = "source-postgres")]
328        "postgres" => Ok(schema::<faucet_source_postgres::PostgresSourceConfig>()),
329        #[cfg(feature = "source-postgres-cdc")]
330        "postgres-cdc" => Ok(schema::<faucet_source_postgres_cdc::PostgresCdcSourceConfig>()),
331        #[cfg(feature = "source-mysql")]
332        "mysql" => Ok(schema::<faucet_source_mysql::MysqlSourceConfig>()),
333        #[cfg(feature = "source-mssql")]
334        "mssql" => Ok(schema::<faucet_source_mssql::MssqlSourceConfig>()),
335        #[cfg(feature = "source-sqlite")]
336        "sqlite" => Ok(schema::<faucet_source_sqlite::SqliteSourceConfig>()),
337        #[cfg(feature = "source-s3")]
338        "s3" => Ok(schema::<faucet_source_s3::S3SourceConfig>()),
339        #[cfg(feature = "source-mongodb")]
340        "mongodb" => Ok(schema::<faucet_source_mongodb::MongoSourceConfig>()),
341        #[cfg(feature = "source-redis")]
342        "redis" => Ok(schema::<faucet_source_redis::RedisSourceConfig>()),
343        #[cfg(feature = "source-webhook")]
344        "webhook" => Ok(schema::<faucet_source_webhook::WebhookSourceConfig>()),
345        #[cfg(feature = "source-websocket")]
346        "websocket" => Ok(schema::<faucet_source_websocket::WebsocketSourceConfig>()),
347        #[cfg(feature = "source-csv")]
348        "csv" => Ok(schema::<faucet_source_csv::CsvSourceConfig>()),
349        #[cfg(feature = "source-elasticsearch")]
350        "elasticsearch" => Ok(schema::<
351            faucet_source_elasticsearch::ElasticsearchSourceConfig,
352        >()),
353        #[cfg(feature = "source-kafka")]
354        "kafka" => Ok(schema::<faucet_source_kafka::KafkaSourceConfig>()),
355        #[cfg(feature = "source-parquet")]
356        "parquet" => Ok(schema::<faucet_source_parquet::ParquetSourceConfig>()),
357        #[cfg(feature = "source-gcs")]
358        "gcs" => Ok(schema::<faucet_source_gcs::GcsSourceConfig>()),
359        #[cfg(feature = "source-bigquery")]
360        "bigquery" => Ok(schema::<faucet_source_bigquery::BigQuerySourceConfig>()),
361        #[cfg(feature = "source-snowflake")]
362        "snowflake" => Ok(schema::<faucet_source_snowflake::SnowflakeSourceConfig>()),
363        other => Err(unknown(other, "source", source_kinds())),
364    }
365}
366
367/// Check if a source kind is registered (not unknown or disabled by feature gate).
368pub fn source_exists(kind: &str) -> bool {
369    source_schema(kind).is_ok()
370}
371
372/// Check if a sink kind is registered (not unknown or disabled by feature gate).
373pub fn sink_exists(kind: &str) -> bool {
374    sink_schema(kind).is_ok()
375}
376
377/// Return the JSON Schema for the named sink's config struct.
378pub fn sink_schema(kind: &str) -> CliResult<Value> {
379    match kind {
380        #[cfg(feature = "sink-bigquery")]
381        "bigquery" => Ok(schema::<faucet_sink_bigquery::BigQuerySinkConfig>()),
382        #[cfg(feature = "sink-postgres")]
383        "postgres" => Ok(schema::<faucet_sink_postgres::PostgresSinkConfig>()),
384        #[cfg(feature = "sink-jsonl")]
385        "jsonl" => Ok(schema::<faucet_sink_jsonl::JsonlSinkConfig>()),
386        #[cfg(feature = "sink-snowflake")]
387        "snowflake" => Ok(schema::<faucet_sink_snowflake::SnowflakeSinkConfig>()),
388        #[cfg(feature = "sink-mysql")]
389        "mysql" => Ok(schema::<faucet_sink_mysql::MysqlSinkConfig>()),
390        #[cfg(feature = "sink-mssql")]
391        "mssql" => Ok(schema::<faucet_sink_mssql::MssqlSinkConfig>()),
392        #[cfg(feature = "sink-sqlite")]
393        "sqlite" => Ok(schema::<faucet_sink_sqlite::SqliteSinkConfig>()),
394        #[cfg(feature = "sink-s3")]
395        "s3" => Ok(schema::<faucet_sink_s3::S3SinkConfig>()),
396        #[cfg(feature = "sink-mongodb")]
397        "mongodb" => Ok(schema::<faucet_sink_mongodb::MongoSinkConfig>()),
398        #[cfg(feature = "sink-redis")]
399        "redis" => Ok(schema::<faucet_sink_redis::RedisSinkConfig>()),
400        #[cfg(feature = "sink-csv")]
401        "csv" => Ok(schema::<faucet_sink_csv::CsvSinkConfig>()),
402        #[cfg(feature = "sink-elasticsearch")]
403        "elasticsearch" => Ok(schema::<faucet_sink_elasticsearch::ElasticsearchSinkConfig>()),
404        #[cfg(feature = "sink-kafka")]
405        "kafka" => Ok(schema::<faucet_sink_kafka::KafkaSinkConfig>()),
406        #[cfg(feature = "sink-http")]
407        "http" => Ok(schema::<faucet_sink_http::HttpSinkConfig>()),
408        #[cfg(feature = "sink-stdout")]
409        "stdout" => Ok(schema::<faucet_sink_stdout::StdoutSinkConfig>()),
410        #[cfg(feature = "sink-parquet")]
411        "parquet" => Ok(schema::<faucet_sink_parquet::ParquetSinkConfig>()),
412        #[cfg(feature = "sink-gcs")]
413        "gcs" => Ok(schema::<faucet_sink_gcs::GcsSinkConfig>()),
414        other => Err(unknown(other, "sink", sink_kinds())),
415    }
416}
417
418/// One-line summary of every compiled-in source connector. Used by `faucet list`.
419#[allow(clippy::vec_init_then_push)]
420pub fn source_descriptions() -> Vec<(&'static str, &'static str)> {
421    let mut v: Vec<(&'static str, &'static str)> = Vec::new();
422    #[cfg(feature = "source-rest")]
423    v.push(("rest", "REST API source with pagination, auth, transforms"));
424    #[cfg(feature = "source-graphql")]
425    v.push(("graphql", "GraphQL API source with cursor pagination"));
426    #[cfg(feature = "source-xml")]
427    v.push(("xml", "XML / SOAP API source with XML→JSON conversion"));
428    #[cfg(feature = "source-grpc")]
429    v.push(("grpc", "gRPC source with dynamic protobuf"));
430    #[cfg(feature = "source-postgres")]
431    v.push(("postgres", "PostgreSQL query source"));
432    #[cfg(feature = "source-postgres-cdc")]
433    v.push((
434        "postgres-cdc",
435        "PostgreSQL CDC source (logical replication)",
436    ));
437    #[cfg(feature = "source-mysql")]
438    v.push(("mysql", "MySQL query source"));
439    #[cfg(feature = "source-mssql")]
440    v.push(("mssql", "Microsoft SQL Server query source"));
441    #[cfg(feature = "source-sqlite")]
442    v.push(("sqlite", "SQLite query source"));
443    #[cfg(feature = "source-s3")]
444    v.push(("s3", "AWS S3 object source"));
445    #[cfg(feature = "source-mongodb")]
446    v.push(("mongodb", "MongoDB query source"));
447    #[cfg(feature = "source-redis")]
448    v.push(("redis", "Redis (streams, lists, keys) source"));
449    #[cfg(feature = "source-webhook")]
450    v.push(("webhook", "Webhook HTTP receiver source"));
451    #[cfg(feature = "source-websocket")]
452    v.push((
453        "websocket",
454        "WebSocket streaming source — connects, subscribes, streams each message as a record",
455    ));
456    #[cfg(feature = "source-csv")]
457    v.push(("csv", "CSV file source"));
458    #[cfg(feature = "source-elasticsearch")]
459    v.push(("elasticsearch", "Elasticsearch search / scroll source"));
460    #[cfg(feature = "source-kafka")]
461    v.push(("kafka", "Apache Kafka consumer (rdkafka). Subscribes to topics and drains messages with idle/max-messages termination."));
462    #[cfg(feature = "source-parquet")]
463    v.push(("parquet", "Apache Parquet file source (local path, glob, or S3). Streams record batches via the Arrow async reader."));
464    #[cfg(feature = "source-gcs")]
465    v.push((
466        "gcs",
467        "Google Cloud Storage source — JSONL, JSON array, or raw text",
468    ));
469    #[cfg(feature = "source-bigquery")]
470    v.push((
471        "bigquery",
472        "Google BigQuery query source (jobs.query + jobs.getQueryResults)",
473    ));
474    #[cfg(feature = "source-snowflake")]
475    v.push((
476        "snowflake",
477        "Snowflake query source (SQL REST API with partition paging)",
478    ));
479    v
480}
481
482/// One-line summary of every compiled-in sink connector. Used by `faucet list`.
483#[allow(clippy::vec_init_then_push)]
484pub fn sink_descriptions() -> Vec<(&'static str, &'static str)> {
485    let mut v: Vec<(&'static str, &'static str)> = Vec::new();
486    #[cfg(feature = "sink-bigquery")]
487    v.push(("bigquery", "Google BigQuery streaming-insert sink"));
488    #[cfg(feature = "sink-postgres")]
489    v.push(("postgres", "PostgreSQL sink (JSONB or auto-mapped columns)"));
490    #[cfg(feature = "sink-jsonl")]
491    v.push(("jsonl", "JSON Lines file sink"));
492    #[cfg(feature = "sink-snowflake")]
493    v.push(("snowflake", "Snowflake SQL REST API sink"));
494    #[cfg(feature = "sink-mysql")]
495    v.push(("mysql", "MySQL sink"));
496    #[cfg(feature = "sink-mssql")]
497    v.push((
498        "mssql",
499        "Microsoft SQL Server sink (auto-mapped columns or JSON column)",
500    ));
501    #[cfg(feature = "sink-sqlite")]
502    v.push(("sqlite", "SQLite sink"));
503    #[cfg(feature = "sink-s3")]
504    v.push(("s3", "AWS S3 object sink"));
505    #[cfg(feature = "sink-mongodb")]
506    v.push(("mongodb", "MongoDB insert sink"));
507    #[cfg(feature = "sink-redis")]
508    v.push(("redis", "Redis (streams, lists, key-value) sink"));
509    #[cfg(feature = "sink-csv")]
510    v.push(("csv", "CSV file sink"));
511    #[cfg(feature = "sink-elasticsearch")]
512    v.push(("elasticsearch", "Elasticsearch bulk index sink"));
513    #[cfg(feature = "sink-kafka")]
514    v.push(("kafka", "Apache Kafka producer (rdkafka). FuturesUnordered batched sends with QueueFull retry; supports fixed or per-record topic routing."));
515    #[cfg(feature = "sink-http")]
516    v.push(("http", "HTTP POST sink (individual or array batch)"));
517    #[cfg(feature = "sink-stdout")]
518    v.push(("stdout", "Stdout / stderr sink (JSON Lines, pretty, TSV)"));
519    #[cfg(feature = "sink-parquet")]
520    v.push(("parquet", "Apache Parquet file sink (local path or S3). Schema-inferred, configurable compression, row/byte rollover."));
521    #[cfg(feature = "sink-gcs")]
522    v.push(("gcs", "Google Cloud Storage sink — JSONL files"));
523    v
524}
525
526/// Names of every compiled-in source connector.
527pub fn source_kinds() -> Vec<&'static str> {
528    source_descriptions().into_iter().map(|(k, _)| k).collect()
529}
530
531/// Names of every compiled-in sink connector.
532pub fn sink_kinds() -> Vec<&'static str> {
533    sink_descriptions().into_iter().map(|(k, _)| k).collect()
534}
535
536fn decode<T: DeserializeOwned>(kind: &'static str, name: &str, config: Value) -> CliResult<T> {
537    serde_json::from_value(config).map_err(|e| CliError::InvalidConnectorConfig {
538        kind,
539        name: name.to_owned(),
540        message: scrub_config_error(&e.to_string()),
541    })
542}
543
544/// Sanitise a serde deserialization error before it reaches stderr/logs.
545///
546/// serde_json's `invalid type:` errors echo the offending value as a
547/// double-quoted literal — which can be a secret injected via
548/// `${secret:...}` / `${env:...}`. Replace every double-quoted run with a
549/// placeholder (field/type names use backticks and are preserved for
550/// diagnostics) and cap the length so a huge value can't flood the log
551/// (#78/#38). Note: `${secret:}` is currently an `${env:}` alias with no
552/// at-rest redaction — this only scrubs error *output*.
553fn scrub_config_error(msg: &str) -> String {
554    const MAX_CHARS: usize = 200;
555    let mut out = String::with_capacity(msg.len());
556    let mut in_quote = false;
557    for c in msg.chars() {
558        if c == '"' {
559            if !in_quote {
560                out.push_str("\"<redacted>\"");
561            }
562            in_quote = !in_quote;
563            continue;
564        }
565        if !in_quote {
566            out.push(c);
567        }
568    }
569    if out.chars().count() > MAX_CHARS {
570        let truncated: String = out.chars().take(MAX_CHARS).collect();
571        return format!("{truncated}…");
572    }
573    out
574}
575
576fn schema<T: faucet_core::JsonSchema>() -> Value {
577    serde_json::to_value(faucet_core::schema_for!(T))
578        .unwrap_or_else(|_| serde_json::json!({"type": "object"}))
579}
580
581fn unknown(name: &str, kind: &'static str, available: Vec<&'static str>) -> CliError {
582    CliError::UnknownConnector {
583        kind,
584        name: name.to_owned(),
585        available: if available.is_empty() {
586            "(none — rebuild faucet-cli with the relevant feature enabled)".to_owned()
587        } else {
588            available.join(", ")
589        },
590    }
591}
592
593#[cfg(test)]
594mod tests {
595    use super::*;
596
597    #[cfg(feature = "source-rest")]
598    #[test]
599    fn rest_source_appears_in_listings() {
600        assert!(source_kinds().contains(&"rest"));
601    }
602
603    #[cfg(feature = "sink-jsonl")]
604    #[test]
605    fn jsonl_sink_appears_in_listings() {
606        assert!(sink_kinds().contains(&"jsonl"));
607    }
608
609    #[tokio::test]
610    async fn unknown_source_kind_errors() {
611        let err = build_source("nope", serde_json::json!({}), &AuthCatalog::new())
612            .await
613            .err()
614            .expect("should fail");
615        match err {
616            CliError::UnknownConnector { kind, name, .. } => {
617                assert_eq!(kind, "source");
618                assert_eq!(name, "nope");
619            }
620            other => panic!("expected UnknownConnector, got {other:?}"),
621        }
622    }
623
624    #[tokio::test]
625    async fn unknown_sink_kind_errors() {
626        let err = build_sink("nope", serde_json::json!({}), &AuthCatalog::new())
627            .await
628            .err()
629            .expect("should fail");
630        assert!(matches!(
631            err,
632            CliError::UnknownConnector { kind: "sink", .. }
633        ));
634    }
635
636    #[cfg(feature = "source-rest")]
637    #[test]
638    fn rest_schema_is_object() {
639        let s = source_schema("rest").unwrap();
640        assert!(s.is_object());
641    }
642
643    #[cfg(feature = "sink-jsonl")]
644    #[test]
645    fn jsonl_schema_is_object() {
646        let s = sink_schema("jsonl").unwrap();
647        assert!(s.is_object());
648    }
649
650    #[test]
651    fn scrub_config_error_redacts_quoted_values() {
652        // A serde "invalid type" error echoes the offending value in double
653        // quotes — must be redacted so a secret can't reach the log (#78/#38).
654        let msg =
655            r#"invalid type: string "sk-super-secret-123", expected a sequence at line 1 column 9"#;
656        let scrubbed = scrub_config_error(msg);
657        assert!(!scrubbed.contains("sk-super-secret-123"), "{scrubbed}");
658        assert!(scrubbed.contains("<redacted>"), "{scrubbed}");
659        // Structural context outside the quotes is preserved.
660        assert!(scrubbed.contains("invalid type"), "{scrubbed}");
661        assert!(scrubbed.contains("expected a sequence"), "{scrubbed}");
662    }
663
664    #[test]
665    fn scrub_config_error_truncates_long_messages() {
666        let msg = "x".repeat(500);
667        let scrubbed = scrub_config_error(&msg);
668        assert!(
669            scrubbed.chars().count() <= 201,
670            "len {}",
671            scrubbed.chars().count()
672        );
673        assert!(scrubbed.ends_with('…'));
674    }
675}
faucet_cli/registry.rs

faucet_cli/
registry.rs
