razeri_plugin_api/sinks/
jsonl.rs

1// Copyright (C) 2026 Gabriel Shea
2// Licensed under either of the Apache License, Version 2.0 or the MIT
3// license, at your option.
4
5//! JSONL file sink — appends events as newline-delimited JSON.
6//!
7//! **Registered only when asked for**: it needs a file, so no registry holds
8//! one until a program opens it with [`JsonlSink::open`] and registers it, or
9//! builds it with [`create_sink`](crate::sinks::create_sink) from
10//! `{ type: jsonl, path: ... }`. The file mover's event log is one.
11
12use std::path::{Path, PathBuf};
13
14use async_trait::async_trait;
15use tokio::io::AsyncWriteExt;
16use tokio::sync::Mutex;
17
18use crate::error::EventSinkError;
19use crate::event::{EventSink, RazeriEvent};
20
21/// Appends each event as a single JSON line to a file.
22pub struct JsonlSink {
23    path: PathBuf,
24    file: Mutex<tokio::fs::File>,
25}
26
27impl JsonlSink {
28    /// Open or create the JSONL file at `path`.
29    ///
30    /// # Errors
31    ///
32    /// Returns an error if the operation fails.
33    pub async fn open(path: impl AsRef<Path>) -> Result<Self, EventSinkError> {
34        let path = path.as_ref().to_path_buf();
35        if let Some(parent) = path.parent()
36            && !parent.as_os_str().is_empty()
37        {
38            tokio::fs::create_dir_all(parent)
39                .await
40                .map_err(|e| EventSinkError::SinkError(format!("mkdir failed: {e}")))?;
41        }
42        let file = tokio::fs::OpenOptions::new()
43            .create(true)
44            .append(true)
45            .open(&path)
46            .await
47            .map_err(|e| EventSinkError::SinkError(format!("open failed: {e}")))?;
48        Ok(Self { path, file: Mutex::new(file) })
49    }
50
51    /// Return the file path this sink writes to.
52    pub fn path(&self) -> &Path {
53        &self.path
54    }
55}
56
57#[async_trait]
58impl EventSink for JsonlSink {
59    fn name(&self) -> &'static str {
60        "jsonl"
61    }
62
63    async fn on_event(&self, event: &RazeriEvent) -> Result<(), EventSinkError> {
64        let mut line = serde_json::to_string(event)
65            .map_err(|e| EventSinkError::SinkError(format!("serialization failed: {e}")))?;
66        line.push('\n');
67        let mut file = self.file.lock().await;
68        file.write_all(line.as_bytes())
69            .await
70            .map_err(|e| EventSinkError::SinkError(format!("write failed: {e}")))?;
71        Ok(())
72    }
73
74    async fn flush(&self) -> Result<(), EventSinkError> {
75        let mut file = self.file.lock().await;
76        file.flush().await.map_err(|e| EventSinkError::SinkError(format!("flush failed: {e}")))?;
77        Ok(())
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84    use crate::event::{RazeriEvent, WorkflowContext};
85    use chrono::Utc;
86
87    fn test_workflow() -> WorkflowContext {
88        WorkflowContext::new("test-wf").unwrap()
89    }
90
91    #[tokio::test]
92    async fn jsonl_sink_writes_events_to_file() {
93        let dir = tempfile::tempdir().unwrap();
94        let path = dir.path().join("events.jsonl");
95
96        let sink = JsonlSink::open(&path).await.unwrap();
97        assert_eq!(sink.name(), "jsonl");
98
99        let event = RazeriEvent::design_validated(
100            "jsonl-test-1",
101            Utc::now(),
102            test_workflow(),
103            "test.yaml",
104            0,
105            0,
106        )
107        .unwrap();
108        sink.on_event(&event).await.unwrap();
109        sink.flush().await.unwrap();
110
111        let content = tokio::fs::read_to_string(&path).await.unwrap();
112        assert!(!content.is_empty());
113        assert!(content.ends_with('\n'));
114        // Verify it's valid JSON
115        let _: serde_json::Value = serde_json::from_str(content.trim()).unwrap();
116    }
117
118    #[tokio::test]
119    async fn jsonl_sink_appends_multiple_events() {
120        let dir = tempfile::tempdir().unwrap();
121        let path = dir.path().join("multi.jsonl");
122
123        let sink = JsonlSink::open(&path).await.unwrap();
124
125        for i in 0..3 {
126            let event = RazeriEvent::design_validated(
127                format!("jsonl-multi-{i}"),
128                Utc::now(),
129                test_workflow(),
130                "test.yaml",
131                0,
132                0,
133            )
134            .unwrap();
135            sink.on_event(&event).await.unwrap();
136        }
137        sink.flush().await.unwrap();
138
139        let content = tokio::fs::read_to_string(&path).await.unwrap();
140        let lines: Vec<&str> = content.trim().lines().collect();
141        assert_eq!(lines.len(), 3);
142    }
143
144    #[tokio::test]
145    async fn jsonl_sink_creates_parent_dirs() {
146        let dir = tempfile::tempdir().unwrap();
147        let path = dir.path().join("nested").join("dir").join("events.jsonl");
148
149        let sink = JsonlSink::open(&path).await.unwrap();
150        assert_eq!(sink.path(), path);
151
152        let event = RazeriEvent::design_validated(
153            "jsonl-dir-1",
154            Utc::now(),
155            test_workflow(),
156            "test.yaml",
157            0,
158            0,
159        )
160        .unwrap();
161        sink.on_event(&event).await.unwrap();
162        sink.flush().await.unwrap();
163
164        assert!(path.exists());
165    }
166}
razeri_plugin_api/sinks/jsonl.rs

razeri_plugin_api/sinks/
jsonl.rs
