razeri_plugin_api/
sinks.rs

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

razeri_plugin_api/
sinks.rs
