razeri_plugin_api/sinks/

jsonl.rs

// Copyright (C) 2026 Gabriel Shea
// Licensed under either of the Apache License, Version 2.0 or the MIT
// license, at your option.

//! JSONL file sink — appends events as newline-delimited JSON.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use async_trait::async_trait;
use tokio::io::AsyncWriteExt;
use tokio::sync::Mutex;

use crate::builder::EventSinkRegistration;
use crate::error::EventSinkError;
use crate::event::{EventSink, RazeriEvent};

/// Appends each event as a single JSON line to a file.
pub struct JsonlSink {
    path: PathBuf,
    file: Mutex<tokio::fs::File>,
}

impl JsonlSink {
    /// Open or create the JSONL file at `path`.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub async fn open(path: impl AsRef<Path>) -> Result<Self, EventSinkError> {
        let path = path.as_ref().to_path_buf();
        if let Some(parent) = path.parent()
            && !parent.as_os_str().is_empty()
        {
            tokio::fs::create_dir_all(parent)
                .await
                .map_err(|e| EventSinkError::SinkError(format!("mkdir failed: {e}")))?;
        }
        let file = tokio::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .await
            .map_err(|e| EventSinkError::SinkError(format!("open failed: {e}")))?;
        Ok(Self { path, file: Mutex::new(file) })
    }

    /// Return the file path this sink writes to.
    pub fn path(&self) -> &Path {
        &self.path
    }
}

#[async_trait]
impl EventSink for JsonlSink {
    fn name(&self) -> &'static str {
        "jsonl"
    }

    async fn on_event(&self, event: &RazeriEvent) -> Result<(), EventSinkError> {
        let mut line = serde_json::to_string(event)
            .map_err(|e| EventSinkError::SinkError(format!("serialization failed: {e}")))?;
        line.push('\n');
        let mut file = self.file.lock().await;
        file.write_all(line.as_bytes())
            .await
            .map_err(|e| EventSinkError::SinkError(format!("write failed: {e}")))?;
        Ok(())
    }

    async fn flush(&self) -> Result<(), EventSinkError> {
        let mut file = self.file.lock().await;
        file.flush().await.map_err(|e| EventSinkError::SinkError(format!("flush failed: {e}")))?;
        Ok(())
    }
}

// JSONL sink requires a runtime path so auto-discovery uses a default path.
// Real usage constructs the sink via `JsonlSink::open()`.
inventory::submit! {
    EventSinkRegistration::new("jsonl", || {
        // Default JSONL sink writes to a well-known path.
        // The runtime must call `JsonlSink::open(path)` for custom paths.
        Arc::new(DefaultJsonlSink)
    })
}

/// Placeholder sink returned by auto-discovery. Real JSONL sinks are
/// constructed via [`JsonlSink::open`] in the composition root.
struct DefaultJsonlSink;

#[async_trait]
impl EventSink for DefaultJsonlSink {
    fn name(&self) -> &'static str {
        "jsonl"
    }

    async fn on_event(&self, _event: &RazeriEvent) -> Result<(), EventSinkError> {
        // No-op: real JsonlSink must be created with a file path
        Ok(())
    }

    async fn flush(&self) -> Result<(), EventSinkError> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{RazeriEvent, WorkflowContext};
    use chrono::Utc;

    fn test_workflow() -> WorkflowContext {
        WorkflowContext::new("test-wf").unwrap()
    }

    #[tokio::test]
    async fn jsonl_sink_writes_events_to_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("events.jsonl");

        let sink = JsonlSink::open(&path).await.unwrap();
        assert_eq!(sink.name(), "jsonl");

        let event = RazeriEvent::design_validated(
            "jsonl-test-1",
            Utc::now(),
            test_workflow(),
            "test.yaml",
            0,
            0,
        )
        .unwrap();
        sink.on_event(&event).await.unwrap();
        sink.flush().await.unwrap();

        let content = tokio::fs::read_to_string(&path).await.unwrap();
        assert!(!content.is_empty());
        assert!(content.ends_with('\n'));
        // Verify it's valid JSON
        let _: serde_json::Value = serde_json::from_str(content.trim()).unwrap();
    }

    #[tokio::test]
    async fn jsonl_sink_appends_multiple_events() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("multi.jsonl");

        let sink = JsonlSink::open(&path).await.unwrap();

        for i in 0..3 {
            let event = RazeriEvent::design_validated(
                format!("jsonl-multi-{i}"),
                Utc::now(),
                test_workflow(),
                "test.yaml",
                0,
                0,
            )
            .unwrap();
            sink.on_event(&event).await.unwrap();
        }
        sink.flush().await.unwrap();

        let content = tokio::fs::read_to_string(&path).await.unwrap();
        let lines: Vec<&str> = content.trim().lines().collect();
        assert_eq!(lines.len(), 3);
    }

    #[tokio::test]
    async fn jsonl_sink_creates_parent_dirs() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested").join("dir").join("events.jsonl");

        let sink = JsonlSink::open(&path).await.unwrap();
        assert_eq!(sink.path(), path);

        let event = RazeriEvent::design_validated(
            "jsonl-dir-1",
            Utc::now(),
            test_workflow(),
            "test.yaml",
            0,
            0,
        )
        .unwrap();
        sink.on_event(&event).await.unwrap();
        sink.flush().await.unwrap();

        assert!(path.exists());
    }
}