razeri_plugin_api/sinks/
jsonl.rs

1//! JSONL file sink — appends events as newline-delimited JSON.
2
3use std::path::{Path, PathBuf};
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use tokio::io::AsyncWriteExt;
8use tokio::sync::Mutex;
9
10use crate::builder::EventSinkRegistration;
11use crate::error::EventSinkError;
12use crate::event::{EventSink, RazeriEvent};
13
14/// Appends each event as a single JSON line to a file.
15pub struct JsonlSink {
16    path: PathBuf,
17    file: Mutex<tokio::fs::File>,
18}
19
20impl JsonlSink {
21    /// Open or create the JSONL file at `path`.
22    ///
23    /// # Errors
24    ///
25    /// Returns an error if the operation fails.
26    pub async fn open(path: impl AsRef<Path>) -> Result<Self, EventSinkError> {
27        let path = path.as_ref().to_path_buf();
28        if let Some(parent) = path.parent()
29            && !parent.as_os_str().is_empty()
30        {
31            tokio::fs::create_dir_all(parent)
32                .await
33                .map_err(|e| EventSinkError::SinkError(format!("mkdir failed: {e}")))?;
34        }
35        let file = tokio::fs::OpenOptions::new()
36            .create(true)
37            .append(true)
38            .open(&path)
39            .await
40            .map_err(|e| EventSinkError::SinkError(format!("open failed: {e}")))?;
41        Ok(Self { path, file: Mutex::new(file) })
42    }
43
44    /// Return the file path this sink writes to.
45    pub fn path(&self) -> &Path {
46        &self.path
47    }
48}
49
50#[async_trait]
51impl EventSink for JsonlSink {
52    fn name(&self) -> &'static str {
53        "jsonl"
54    }
55
56    async fn on_event(&self, event: &RazeriEvent) -> Result<(), EventSinkError> {
57        let mut line = serde_json::to_string(event)
58            .map_err(|e| EventSinkError::SinkError(format!("serialization failed: {e}")))?;
59        line.push('\n');
60        let mut file = self.file.lock().await;
61        file.write_all(line.as_bytes())
62            .await
63            .map_err(|e| EventSinkError::SinkError(format!("write failed: {e}")))?;
64        Ok(())
65    }
66
67    async fn flush(&self) -> Result<(), EventSinkError> {
68        let mut file = self.file.lock().await;
69        file.flush().await.map_err(|e| EventSinkError::SinkError(format!("flush failed: {e}")))?;
70        Ok(())
71    }
72}
73
74// JSONL sink requires a runtime path so auto-discovery uses a default path.
75// Real usage constructs the sink via `JsonlSink::open()`.
76inventory::submit! {
77    EventSinkRegistration::new("jsonl", || {
78        // Default JSONL sink writes to a well-known path.
79        // The runtime must call `JsonlSink::open(path)` for custom paths.
80        Arc::new(DefaultJsonlSink)
81    })
82}
83
84/// Placeholder sink returned by auto-discovery. Real JSONL sinks are
85/// constructed via [`JsonlSink::open`] in the composition root.
86struct DefaultJsonlSink;
87
88#[async_trait]
89impl EventSink for DefaultJsonlSink {
90    fn name(&self) -> &'static str {
91        "jsonl"
92    }
93
94    async fn on_event(&self, _event: &RazeriEvent) -> Result<(), EventSinkError> {
95        // No-op: real JsonlSink must be created with a file path
96        Ok(())
97    }
98
99    async fn flush(&self) -> Result<(), EventSinkError> {
100        Ok(())
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use crate::event::{RazeriEvent, WorkflowContext};
108    use chrono::Utc;
109
110    fn test_workflow() -> WorkflowContext {
111        WorkflowContext::new("test-wf").unwrap()
112    }
113
114    #[tokio::test]
115    async fn jsonl_sink_writes_events_to_file() {
116        let dir = tempfile::tempdir().unwrap();
117        let path = dir.path().join("events.jsonl");
118
119        let sink = JsonlSink::open(&path).await.unwrap();
120        assert_eq!(sink.name(), "jsonl");
121
122        let event = RazeriEvent::design_validated(
123            "jsonl-test-1",
124            Utc::now(),
125            test_workflow(),
126            "test.yaml",
127            0,
128            0,
129        )
130        .unwrap();
131        sink.on_event(&event).await.unwrap();
132        sink.flush().await.unwrap();
133
134        let content = tokio::fs::read_to_string(&path).await.unwrap();
135        assert!(!content.is_empty());
136        assert!(content.ends_with('\n'));
137        // Verify it's valid JSON
138        let _: serde_json::Value = serde_json::from_str(content.trim()).unwrap();
139    }
140
141    #[tokio::test]
142    async fn jsonl_sink_appends_multiple_events() {
143        let dir = tempfile::tempdir().unwrap();
144        let path = dir.path().join("multi.jsonl");
145
146        let sink = JsonlSink::open(&path).await.unwrap();
147
148        for i in 0..3 {
149            let event = RazeriEvent::design_validated(
150                format!("jsonl-multi-{i}"),
151                Utc::now(),
152                test_workflow(),
153                "test.yaml",
154                0,
155                0,
156            )
157            .unwrap();
158            sink.on_event(&event).await.unwrap();
159        }
160        sink.flush().await.unwrap();
161
162        let content = tokio::fs::read_to_string(&path).await.unwrap();
163        let lines: Vec<&str> = content.trim().lines().collect();
164        assert_eq!(lines.len(), 3);
165    }
166
167    #[tokio::test]
168    async fn jsonl_sink_creates_parent_dirs() {
169        let dir = tempfile::tempdir().unwrap();
170        let path = dir.path().join("nested").join("dir").join("events.jsonl");
171
172        let sink = JsonlSink::open(&path).await.unwrap();
173        assert_eq!(sink.path(), path);
174
175        let event = RazeriEvent::design_validated(
176            "jsonl-dir-1",
177            Utc::now(),
178            test_workflow(),
179            "test.yaml",
180            0,
181            0,
182        )
183        .unwrap();
184        sink.on_event(&event).await.unwrap();
185        sink.flush().await.unwrap();
186
187        assert!(path.exists());
188    }
189}
razeri_plugin_api/sinks/jsonl.rs

razeri_plugin_api/sinks/
jsonl.rs
