faucet_sink_jsonl/
config.rs

1//! JSON Lines sink configuration.
2
3use std::path::PathBuf;
4
5use faucet_core::DEFAULT_BATCH_SIZE;
6use schemars::JsonSchema;
7use serde::{Deserialize, Serialize};
8
9/// Configuration for the JSON Lines file sink.
10#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
11pub struct JsonlSinkConfig {
12    /// Path to the output file.
13    pub path: PathBuf,
14    /// Whether to append to an existing file (default: false, truncates).
15    #[serde(default)]
16    pub append: bool,
17    /// Whether to pretty-print each JSON record (default: false, compact).
18    #[serde(default)]
19    pub pretty: bool,
20    /// Records per upstream [`StreamPage`](faucet_core::StreamPage). The JSONL
21    /// sink writes records to disk one at a time through a buffered async
22    /// writer, so this field has **no behavioural impact** at the sink — it is
23    /// exposed purely for config parity across every sink in the workspace.
24    /// Defaults to [`DEFAULT_BATCH_SIZE`].
25    ///
26    /// `batch_size = 0` (the "no batching" sentinel) and any positive value
27    /// produce byte-for-byte identical output for this sink: each record is
28    /// serialised and appended individually regardless of how upstream chunked
29    /// the page.
30    #[serde(default = "default_batch_size")]
31    pub batch_size: usize,
32    /// Compression codec for the output file. Defaults to
33    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
34    /// `.gz` / `.zst` suffix selects gzip / zstd, anything else writes
35    /// uncompressed. Requires the crate-local `compression` feature.
36    #[cfg(feature = "compression")]
37    #[serde(default)]
38    pub compression: faucet_core::CompressionConfig,
39    /// Encryption at rest (#207): seal every record line with AES-256-GCM and
40    /// write it base64-encoded (one sealed record per line, append-safe).
41    /// The natural fit for a file-backed DLQ holding sensitive failed rows;
42    /// `faucet dlq inspect/replay/discard` decrypt transparently given the
43    /// key. Mutually exclusive with `compression`. Requires the crate-local
44    /// `encryption` feature.
45    #[cfg(feature = "encryption")]
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub encryption: Option<faucet_core::EncryptionSpec>,
48}
49
50fn default_batch_size() -> usize {
51    DEFAULT_BATCH_SIZE
52}
53
54impl JsonlSinkConfig {
55    /// Create a new config with the required file path and sensible defaults.
56    pub fn new(path: impl Into<PathBuf>) -> Self {
57        Self {
58            path: path.into(),
59            append: false,
60            pretty: false,
61            batch_size: DEFAULT_BATCH_SIZE,
62            #[cfg(feature = "compression")]
63            compression: faucet_core::CompressionConfig::Auto,
64            #[cfg(feature = "encryption")]
65            encryption: None,
66        }
67    }
68
69    /// Append to an existing file instead of truncating.
70    pub fn append(mut self, append: bool) -> Self {
71        self.append = append;
72        self
73    }
74
75    /// Pretty-print each JSON record (one record still per logical entry,
76    /// but with indentation). Note: this breaks strict JSONL format.
77    pub fn pretty(mut self, pretty: bool) -> Self {
78        self.pretty = pretty;
79        self
80    }
81
82    /// Set the compression codec. Available only with the `compression` feature.
83    #[cfg(feature = "compression")]
84    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
85        self.compression = c;
86        self
87    }
88
89    /// Seal every record line at rest (#207). Available only with the
90    /// `encryption` feature; mutually exclusive with `compression`.
91    #[cfg(feature = "encryption")]
92    pub fn encryption(mut self, e: faucet_core::EncryptionSpec) -> Self {
93        self.encryption = Some(e);
94        self
95    }
96
97    /// Set the per-page record count hint reported alongside other sink
98    /// configs.
99    ///
100    /// This sink writes per-record through a buffered writer, so the value is
101    /// observably a no-op: `0` (the "no batching" sentinel) and any positive
102    /// value produce the same on-disk output. Present for symmetry with sinks
103    /// whose `batch_size` does drive I/O sizing (e.g. SQL multi-row inserts,
104    /// BigQuery streaming inserts).
105    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
106        self.batch_size = batch_size;
107        self
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    #[test]
116    fn default_config() {
117        let config = JsonlSinkConfig::new("/tmp/out.jsonl");
118        assert_eq!(config.path, PathBuf::from("/tmp/out.jsonl"));
119        assert!(!config.append);
120        assert!(!config.pretty);
121    }
122
123    #[test]
124    fn builder_methods() {
125        let config = JsonlSinkConfig::new("/tmp/out.jsonl")
126            .append(true)
127            .pretty(true);
128        assert!(config.append);
129        assert!(config.pretty);
130    }
131
132    #[test]
133    fn batch_size_defaults_to_default_batch_size() {
134        let config = JsonlSinkConfig::new("/tmp/out.jsonl");
135        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
136    }
137
138    #[test]
139    fn with_batch_size_overrides_default() {
140        let config = JsonlSinkConfig::new("/tmp/out.jsonl").with_batch_size(250);
141        assert_eq!(config.batch_size, 250);
142    }
143
144    #[test]
145    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
146        let config = JsonlSinkConfig::new("/tmp/out.jsonl").with_batch_size(0);
147        assert_eq!(config.batch_size, 0);
148        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
149    }
150
151    #[test]
152    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
153        let config =
154            JsonlSinkConfig::new("/tmp/out.jsonl").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
155        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
156    }
157
158    #[test]
159    fn batch_size_deserializes_from_json() {
160        let json = r#"{
161            "path": "/tmp/out.jsonl",
162            "append": false,
163            "pretty": false,
164            "batch_size": 500
165        }"#;
166        let config: JsonlSinkConfig = serde_json::from_str(json).unwrap();
167        assert_eq!(config.batch_size, 500);
168    }
169
170    #[test]
171    fn batch_size_defaults_when_missing_in_json() {
172        let json = r#"{"path": "/tmp/out.jsonl"}"#;
173        let config: JsonlSinkConfig = serde_json::from_str(json).unwrap();
174        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
175    }
176}
faucet_sink_jsonl/config.rs

faucet_sink_jsonl/
config.rs
