opssignal_core/
sink.rs

1use crate::delivery::{DeliveryResult, ProviderOutcome};
2use crate::signal::{Signal, SignalError};
3use rusqlite::Connection;
4use std::sync::Mutex;
5
6/// A sink is a write-only side effect of the pipeline — not a queryable
7/// API the SDK exposes. Querying, dashboards, and aggregation are
8/// explicitly out of scope here; that's the hosted control plane's job,
9/// reading from the same store OSS users already populate.
10///
11/// Each signal is written twice: `record` before delivery starts, with
12/// status `pending`, and `complete` once the outcome is known. A signal
13/// whose process exits mid-delivery therefore stays visible as pending
14/// instead of vanishing.
15pub trait SignalSink: Send + Sync {
16    /// Stores a signal that is about to be delivered.
17    fn record(&self, signal: &Signal) -> Result<(), SignalError>;
18
19    /// Stores the delivery outcome, including `result.deliveries`. Must
20    /// also work when `record` failed or never ran for this signal, so
21    /// the outcome is never lost.
22    fn complete(&self, signal: &Signal, result: &DeliveryResult) -> Result<(), SignalError>;
23
24    /// Stores one provider's outcome while delivery is still running:
25    /// `pending` before its first attempt, then `delivered` or `failed`.
26    /// A fan-out cut short by a timeout or an exit then still shows
27    /// which providers already have the signal. Sinks that only keep
28    /// the overall outcome can leave this as is.
29    fn provider_outcome(
30        &self,
31        _signal: &Signal,
32        _outcome: &ProviderOutcome,
33    ) -> Result<(), SignalError> {
34        Ok(())
35    }
36}
37
38/// Zero-config default sink. Always available, no setup required, so
39/// "storage on by default" is actually true out of the box.
40pub struct SqliteSink {
41    conn: Mutex<Connection>,
42}
43
44impl SqliteSink {
45    pub fn open(path: &str) -> Result<Self, SignalError> {
46        let conn = Connection::open(path).map_err(|e| SignalError::Sink(e.to_string()))?;
47        conn.execute(
48            "CREATE TABLE IF NOT EXISTS signals (
49                id TEXT PRIMARY KEY,
50                source TEXT NOT NULL,
51                event_type TEXT NOT NULL,
52                severity TEXT NOT NULL,
53                title TEXT NOT NULL,
54                environment TEXT,
55                dedup_key TEXT,
56                delivery_status TEXT NOT NULL,
57                provider TEXT,
58                error TEXT,
59                payload_json TEXT NOT NULL,
60                timestamp TEXT NOT NULL
61            )",
62            [],
63        )
64        .map_err(|e| SignalError::Sink(e.to_string()))?;
65        conn.execute(
66            "CREATE INDEX IF NOT EXISTS idx_signals_dedup ON signals(dedup_key)",
67            [],
68        )
69        .map_err(|e| SignalError::Sink(e.to_string()))?;
70        // One row per (signal, provider). Added in beta.2; an older
71        // database gets the table on its next open.
72        conn.execute(
73            "CREATE TABLE IF NOT EXISTS deliveries (
74                signal_id TEXT NOT NULL,
75                provider TEXT NOT NULL,
76                status TEXT NOT NULL,
77                error TEXT,
78                attempts INTEGER NOT NULL,
79                updated_at TEXT NOT NULL,
80                PRIMARY KEY (signal_id, provider)
81            )",
82            [],
83        )
84        .map_err(|e| SignalError::Sink(e.to_string()))?;
85        Ok(SqliteSink {
86            conn: Mutex::new(conn),
87        })
88    }
89}
90
91impl SqliteSink {
92    fn insert(
93        &self,
94        signal: &Signal,
95        result: &DeliveryResult,
96        on_conflict: &str,
97    ) -> Result<(), SignalError> {
98        let conn = self.lock()?;
99        let payload =
100            serde_json::to_string(signal).map_err(|e| SignalError::Sink(e.to_string()))?;
101        conn.execute(
102            &format!(
103                "INSERT INTO signals (id, source, event_type, severity, title, environment, dedup_key, delivery_status, provider, error, payload_json, timestamp)
104                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) {on_conflict}"
105            ),
106            rusqlite::params![
107                signal.id.to_string(),
108                signal.source,
109                signal.event_type,
110                signal.severity.to_string(),
111                signal.title,
112                signal.environment,
113                signal.dedup_key,
114                result.status.to_string(),
115                result.provider.clone(),
116                result.error.clone(),
117                payload,
118                signal.timestamp.to_rfc3339(),
119            ],
120        )
121        .map_err(|e| SignalError::Sink(e.to_string()))?;
122        Ok(())
123    }
124
125    fn lock(&self) -> Result<std::sync::MutexGuard<'_, Connection>, SignalError> {
126        self.conn
127            .lock()
128            .map_err(|_| SignalError::Sink("sqlite connection mutex poisoned".to_string()))
129    }
130}
131
132fn upsert_outcome(
133    conn: &Connection,
134    signal: &Signal,
135    outcome: &ProviderOutcome,
136) -> Result<(), SignalError> {
137    conn.execute(
138        "INSERT INTO deliveries (signal_id, provider, status, error, attempts, updated_at)
139         VALUES (?1, ?2, ?3, ?4, ?5, ?6)
140         ON CONFLICT(signal_id, provider) DO UPDATE SET status = excluded.status, \
141         error = excluded.error, attempts = excluded.attempts, updated_at = excluded.updated_at",
142        rusqlite::params![
143            signal.id.to_string(),
144            outcome.provider,
145            outcome.status.to_string(),
146            outcome.error,
147            outcome.attempts,
148            chrono::Utc::now().to_rfc3339(),
149        ],
150    )
151    .map_err(|e| SignalError::Sink(e.to_string()))?;
152    Ok(())
153}
154
155impl SignalSink for SqliteSink {
156    fn record(&self, signal: &Signal) -> Result<(), SignalError> {
157        self.insert(signal, &DeliveryResult::pending(), "")
158    }
159
160    fn complete(&self, signal: &Signal, result: &DeliveryResult) -> Result<(), SignalError> {
161        self.insert(
162            signal,
163            result,
164            "ON CONFLICT(id) DO UPDATE SET delivery_status = excluded.delivery_status, \
165             provider = excluded.provider, error = excluded.error",
166        )?;
167        let conn = self.lock()?;
168        for outcome in &result.deliveries {
169            upsert_outcome(&conn, signal, outcome)?;
170        }
171        Ok(())
172    }
173
174    fn provider_outcome(
175        &self,
176        signal: &Signal,
177        outcome: &ProviderOutcome,
178    ) -> Result<(), SignalError> {
179        upsert_outcome(&*self.lock()?, signal, outcome)
180    }
181}
182
183// NOTE for v0.3 roadmap: PostgresSink and S3Sink implement the same
184// trait. Routing/retry logic never needs to change when a new sink is
185// added — that boundary is the whole point.
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190    use crate::delivery::DeliveryStatus;
191    use crate::signal::{Severity, SignalInput};
192
193    fn temp_db() -> String {
194        std::env::temp_dir()
195            .join(format!("opssignal-sink-{}.db", uuid::Uuid::new_v4()))
196            .to_string_lossy()
197            .into_owned()
198    }
199
200    fn signal() -> Signal {
201        Signal::from_input(SignalInput {
202            source: "airflow".into(),
203            event_type: "task_failed".into(),
204            title: "dbt model failed".into(),
205            severity: Severity::Error,
206            environment: Some("prod".into()),
207            metadata: Some(serde_json::json!({"dag": "nightly"})),
208            ..Default::default()
209        })
210    }
211
212    fn failed() -> DeliveryResult {
213        DeliveryResult {
214            status: DeliveryStatus::Failed,
215            provider: None,
216            error: Some("slack webhook returned status 500".into()),
217            attempts: 3,
218            deliveries: vec![],
219            suppressed: vec![],
220        }
221    }
222
223    fn columns(db: &str) -> Vec<Option<String>> {
224        Connection::open(db)
225            .expect("reopen")
226            .query_row(
227                "SELECT id, source, event_type, severity, title, environment, dedup_key, \
228                 delivery_status, provider, error, payload_json, timestamp FROM signals",
229                [],
230                |r| (0..12).map(|i| r.get(i)).collect(),
231            )
232            .expect("one row")
233    }
234
235    #[test]
236    fn complete_stores_every_column() {
237        let db = temp_db();
238        let sink = SqliteSink::open(&db).expect("open");
239        let signal = signal();
240        sink.complete(&signal, &failed()).expect("complete");
241
242        let row = columns(&db);
243        let col = |i: usize| row[i].as_deref();
244
245        assert_eq!(col(0), Some(signal.id.to_string().as_str()));
246        assert_eq!(col(1), Some("airflow"));
247        assert_eq!(col(2), Some("task_failed"));
248        assert_eq!(col(3), Some("error"));
249        assert_eq!(col(4), Some("dbt model failed"));
250        assert_eq!(col(5), Some("prod"));
251        assert_eq!(col(6), signal.dedup_key.as_deref());
252        assert_eq!(col(7), Some("failed"));
253        assert_eq!(col(8), None);
254        assert_eq!(col(9), Some("slack webhook returned status 500"));
255        let payload: serde_json::Value =
256            serde_json::from_str(col(10).expect("payload")).expect("payload is JSON");
257        assert_eq!(payload["metadata"]["dag"], "nightly");
258        assert_eq!(col(11), Some(signal.timestamp.to_rfc3339().as_str()));
259        let _ = std::fs::remove_file(&db);
260    }
261
262    #[test]
263    fn record_stores_the_signal_as_pending() {
264        let db = temp_db();
265        let sink = SqliteSink::open(&db).expect("open");
266        sink.record(&signal()).expect("record");
267
268        let row = columns(&db);
269        assert_eq!(row[7].as_deref(), Some("pending"));
270        assert_eq!(row[8], None);
271        assert_eq!(row[9], None);
272        let _ = std::fs::remove_file(&db);
273    }
274
275    #[test]
276    fn complete_updates_the_recorded_row_in_place() {
277        let db = temp_db();
278        let sink = SqliteSink::open(&db).expect("open");
279        let signal = signal();
280        sink.record(&signal).expect("record");
281        let delivered = DeliveryResult {
282            status: DeliveryStatus::Delivered,
283            provider: Some("slack".into()),
284            error: None,
285            attempts: 1,
286            deliveries: vec![],
287            suppressed: vec![],
288        };
289        sink.complete(&signal, &delivered).expect("complete");
290
291        let conn = Connection::open(&db).expect("reopen");
292        let count: i64 = conn
293            .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get(0))
294            .expect("count");
295        assert_eq!(count, 1);
296        let row = columns(&db);
297        assert_eq!(row[7].as_deref(), Some("delivered"));
298        assert_eq!(row[8].as_deref(), Some("slack"));
299        let _ = std::fs::remove_file(&db);
300    }
301
302    #[test]
303    fn reopening_an_existing_database_keeps_its_rows() {
304        let db = temp_db();
305        SqliteSink::open(&db)
306            .expect("open")
307            .record(&signal())
308            .expect("record");
309        let sink = SqliteSink::open(&db).expect("reopen is idempotent");
310        sink.record(&signal()).expect("second record");
311
312        let count: i64 = Connection::open(&db)
313            .expect("reopen")
314            .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get(0))
315            .expect("count");
316        assert_eq!(count, 2);
317        let _ = std::fs::remove_file(&db);
318    }
319
320    #[test]
321    fn recording_the_same_signal_twice_is_a_sink_error() {
322        let db = temp_db();
323        let sink = SqliteSink::open(&db).expect("open");
324        let signal = signal();
325        sink.record(&signal).expect("first record");
326
327        let err = sink
328            .record(&signal)
329            .expect_err("duplicate id violates the primary key");
330        assert!(matches!(err, SignalError::Sink(_)));
331        let _ = std::fs::remove_file(&db);
332    }
333
334    #[test]
335    fn unopenable_path_is_a_sink_error() {
336        let err = SqliteSink::open("/nonexistent-opssignal-dir/sub/signals.db")
337            .err()
338            .expect("missing parent directory must fail");
339        assert!(matches!(err, SignalError::Sink(_)));
340    }
341
342    fn outcome(provider: &str, status: DeliveryStatus, attempts: u32) -> ProviderOutcome {
343        ProviderOutcome {
344            provider: provider.into(),
345            status,
346            error: (status == DeliveryStatus::Failed).then(|| format!("{provider} is down")),
347            attempts,
348        }
349    }
350
351    fn delivery_rows(db: &str) -> Vec<(String, String, Option<String>, u32)> {
352        let conn = Connection::open(db).expect("reopen");
353        let mut stmt = conn
354            .prepare("SELECT provider, status, error, attempts FROM deliveries ORDER BY provider")
355            .expect("prepare");
356        stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
357            .expect("query")
358            .map(|r| r.expect("row"))
359            .collect()
360    }
361
362    #[test]
363    fn provider_outcomes_move_from_pending_to_their_result() {
364        let db = temp_db();
365        let sink = SqliteSink::open(&db).expect("open");
366        let signal = signal();
367        sink.record(&signal).expect("record");
368        for p in ["pagerduty", "slack"] {
369            sink.provider_outcome(&signal, &outcome(p, DeliveryStatus::Pending, 0))
370                .expect("pending");
371        }
372        assert_eq!(
373            delivery_rows(&db),
374            vec![
375                ("pagerduty".into(), "pending".into(), None, 0),
376                ("slack".into(), "pending".into(), None, 0),
377            ]
378        );
379
380        sink.provider_outcome(&signal, &outcome("slack", DeliveryStatus::Delivered, 1))
381            .expect("slack done");
382        sink.complete(
383            &signal,
384            &DeliveryResult {
385                status: DeliveryStatus::Partial,
386                provider: Some("slack".into()),
387                error: Some("pagerduty: pagerduty is down".into()),
388                attempts: 3,
389                deliveries: vec![
390                    outcome("slack", DeliveryStatus::Delivered, 1),
391                    outcome("pagerduty", DeliveryStatus::Failed, 3),
392                ],
393                suppressed: vec![],
394            },
395        )
396        .expect("complete");
397
398        assert_eq!(
399            delivery_rows(&db),
400            vec![
401                (
402                    "pagerduty".into(),
403                    "failed".into(),
404                    Some("pagerduty is down".into()),
405                    3
406                ),
407                ("slack".into(), "delivered".into(), None, 1),
408            ]
409        );
410        let row = columns(&db);
411        assert_eq!(row[7].as_deref(), Some("partial"));
412        assert_eq!(row[8].as_deref(), Some("slack"));
413        let _ = std::fs::remove_file(&db);
414    }
415
416    #[test]
417    fn opening_a_database_from_before_deliveries_adds_the_table() {
418        let db = temp_db();
419        Connection::open(&db)
420            .expect("create")
421            .execute_batch(
422                "CREATE TABLE signals (id TEXT PRIMARY KEY, source TEXT NOT NULL, \
423                 event_type TEXT NOT NULL, severity TEXT NOT NULL, title TEXT NOT NULL, \
424                 environment TEXT, dedup_key TEXT, delivery_status TEXT NOT NULL, \
425                 provider TEXT, error TEXT, payload_json TEXT NOT NULL, timestamp TEXT NOT NULL);",
426            )
427            .expect("old schema");
428        let sink = SqliteSink::open(&db).expect("open old database");
429        let signal = signal();
430        sink.provider_outcome(&signal, &outcome("slack", DeliveryStatus::Delivered, 1))
431            .expect("write to the new table");
432        assert_eq!(delivery_rows(&db).len(), 1);
433        let _ = std::fs::remove_file(&db);
434    }
435}
opssignal_core/sink.rs

opssignal_core/
sink.rs
