razeri_plugin_api/
sinks.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//! Built-in event sinks, implemented against the same public traits a
6//! third-party sink uses — nothing here has access a plugin author lacks.
7//!
8//! Each sink implements [`EventSink`] and registers via
9//! [`inventory::submit!`] so it is auto-discovered by
10//! [`crate::PluginRegistryBuilder::with_auto_discovered()`].
11//!
12//! Use [`create_sink`] to construct a concrete sink from an
13//! [`EventSinkConfig`].
14
15// The JSONL sink writes files via `tokio::fs`, which is unavailable on
16// wasm32-unknown-unknown — it is a native-only sink (a browser has no
17// filesystem). Gated out of the wasm build; `create_sink` rejects the type there.
18#[cfg(not(target_arch = "wasm32"))]
19pub mod jsonl;
20pub mod stdout;
21pub mod webhook;
22
23use std::sync::Arc;
24
25use crate::error::EventSinkError;
26use crate::event::EventSink;
27
28/// Sink configuration entry — matches the agent YAML schema.
29#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
30pub struct EventSinkConfig {
31    /// Sink type: `"stdout"`, `"jsonl"`, or `"webhook"`.
32    #[serde(rename = "type")]
33    pub sink_type: String,
34    /// File path (required for `jsonl`).
35    #[serde(skip_serializing_if = "Option::is_none")]
36    pub path: Option<String>,
37    /// HTTP endpoint URL (required for `webhook`).
38    #[serde(skip_serializing_if = "Option::is_none")]
39    pub url: Option<String>,
40}
41
42/// Create a concrete [`EventSink`] from a config entry.
43///
44/// # Errors
45///
46/// Returns an error if the operation fails.
47pub async fn create_sink(config: &EventSinkConfig) -> Result<Arc<dyn EventSink>, EventSinkError> {
48    match config.sink_type.as_str() {
49        "stdout" => Ok(Arc::new(stdout::StdoutSink::new())),
50        #[cfg(not(target_arch = "wasm32"))]
51        "jsonl" => {
52            let path = config
53                .path
54                .as_deref()
55                .ok_or_else(|| EventSinkError::SinkError("jsonl sink requires 'path'".into()))?;
56            let sink = jsonl::JsonlSink::open(path).await?;
57            Ok(Arc::new(sink))
58        }
59        "webhook" => {
60            let url = config
61                .url
62                .as_deref()
63                .ok_or_else(|| EventSinkError::SinkError("webhook sink requires 'url'".into()))?;
64            Ok(Arc::new(webhook::WebhookSink::new(url)))
65        }
66        other => Err(EventSinkError::SinkError(format!("unknown sink type: {other}"))),
67    }
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73
74    #[tokio::test]
75    async fn create_stdout_sink_from_config() {
76        let config = EventSinkConfig { sink_type: "stdout".into(), path: None, url: None };
77        let sink = create_sink(&config).await.unwrap();
78        assert_eq!(sink.name(), "stdout");
79    }
80
81    #[tokio::test]
82    async fn create_jsonl_sink_from_config() {
83        let dir = tempfile::tempdir().unwrap();
84        let path = dir.path().join("test.jsonl");
85        let config = EventSinkConfig {
86            sink_type: "jsonl".into(),
87            path: Some(path.to_string_lossy().into()),
88            url: None,
89        };
90        let sink = create_sink(&config).await.unwrap();
91        assert_eq!(sink.name(), "jsonl");
92    }
93
94    #[tokio::test]
95    async fn create_webhook_sink_from_config() {
96        let config = EventSinkConfig {
97            sink_type: "webhook".into(),
98            path: None,
99            url: Some("https://example.com/events".into()),
100        };
101        let sink = create_sink(&config).await.unwrap();
102        assert_eq!(sink.name(), "webhook");
103    }
104
105    #[tokio::test]
106    async fn create_jsonl_sink_missing_path_returns_error() {
107        let config = EventSinkConfig { sink_type: "jsonl".into(), path: None, url: None };
108        let result = create_sink(&config).await;
109        assert!(result.is_err());
110    }
111
112    #[tokio::test]
113    async fn create_webhook_sink_missing_url_returns_error() {
114        let config = EventSinkConfig { sink_type: "webhook".into(), path: None, url: None };
115        let result = create_sink(&config).await;
116        assert!(result.is_err());
117    }
118
119    #[tokio::test]
120    async fn create_unknown_sink_returns_error() {
121        let config = EventSinkConfig { sink_type: "kafka".into(), path: None, url: None };
122        let result = create_sink(&config).await;
123        assert!(result.is_err());
124    }
125}
razeri_plugin_api/sinks.rs

razeri_plugin_api/
sinks.rs
