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

faucet_cli/
registry.rs
