kasl_server/webhooks/
mod.rs

1//! Webhooks: the server says what it noticed to where a manager already is.
2//!
3//! Alerts made the server notice things on its own (ADR 0018), and then wait
4//! on a dashboard for somebody to open it. This module takes the same record
5//! outward - to a Slack or Mattermost channel, a Telegram chat, or a system of
6//! the operator's own - without replacing it: delivery is added to the record,
7//! and the row an alert wrote is what gets sent.
8//!
9//! Three decisions shape everything here (ADR 0019):
10//!
11//! * **Destinations live in the environment, not in the database.** A hook URL
12//!   and a bot token are working credentials to post as somebody, and the
13//!   database is what `kasl-server backup` writes to a file. They sit next to
14//!   the database password, in `KASL_WEBHOOK_<NAME>`, and everything else -
15//!   the delivery log, the screen, the privacy manifest - knows a destination
16//!   by its name alone.
17//! * **An event is queued in the transaction that made it true.** The alert
18//!   row and the delivery row commit together, or neither does. An in-memory
19//!   channel loses what was queued on every restart, and "the server restarted
20//!   at the moment the agent died" is exactly the night nobody hears about.
21//! * **Delivery is at least once, in order, per destination.** The dispatcher
22//!   retries with growing gaps and gives up in writing after about a day; a
23//!   destination that is failing holds its own later events back rather than
24//!   letting "resolved" overtake "raised" in somebody's channel. Every event
25//!   carries an id a receiver can deduplicate on, because a request that timed
26//!   out may still have arrived.
27
28mod destination;
29mod dispatch;
30mod render;
31mod sign;
32
33use axum::{
34    Json,
35    extract::{Path, State},
36    http::StatusCode,
37    response::IntoResponse,
38};
39use chrono::{DateTime, NaiveDate, Utc};
40use serde::{Deserialize, Serialize};
41use sha2::{Digest, Sha256};
42use sqlx::PgConnection;
43use uuid::Uuid;
44
45pub use destination::{Destination, EventKind, Kind, PREFIX};
46pub use dispatch::{Dispatched, client, dispatch_due, run_dispatcher};
47pub use sign::{hmac_sha256, signature};
48
49use crate::{alerts::AlertRule, app::AppState, audit, calendar::WorkdayKind, error::ApiError, login::CurrentUser};
50
51/// The payload's shape. A receiver of the `json` kind reads this first; a
52/// change to what a field means is a new number, not a quiet edit.
53pub const PAYLOAD_VERSION: u32 = 1;
54
55/// Every destination this installation sends to, and where its screens live.
56#[derive(Debug, Clone, Default)]
57pub struct Webhooks {
58    /// Sorted by name, so the screen and the manifest list them the same way
59    /// on every start.
60    destinations: Vec<Destination>,
61    /// The address people open the web UI at (`KASL_PUBLIC_URL`), without a
62    /// trailing slash. When set, a message links to the person it is about;
63    /// when not, it says what happened and leaves the finding to the reader.
64    public_url: Option<String>,
65}
66
67impl Webhooks {
68    /// Reads every `KASL_WEBHOOK_*` variable among `vars`.
69    ///
70    /// An error in any one stops the server from starting, and names the
71    /// variable without repeating its value.
72    pub fn from_vars(vars: impl IntoIterator<Item = (String, String)>, public_url: Option<String>) -> Result<Self, String> {
73        let mut destinations = vars
74            .into_iter()
75            .filter(|(key, _)| key.starts_with(PREFIX))
76            .map(|(key, value)| Destination::parse(&key, &value))
77            .collect::<Result<Vec<_>, _>>()?;
78        destinations.sort_by(|a, b| a.name.cmp(&b.name));
79
80        let public_url = match public_url.map(|url| url.trim().trim_end_matches('/').to_string()) {
81            Some(url) if url.is_empty() => None,
82            Some(url) if url.starts_with("https://") || url.starts_with("http://") => Some(url),
83            Some(_) => return Err("KASL_PUBLIC_URL is not an http(s) address".to_string()),
84            None => None,
85        };
86
87        Ok(Self { destinations, public_url })
88    }
89
90    /// A set built in code, for the tests.
91    pub fn new(destinations: Vec<Destination>, public_url: Option<&str>) -> Self {
92        let mut destinations = destinations;
93        destinations.sort_by(|a, b| a.name.cmp(&b.name));
94        Self {
95            destinations,
96            public_url: public_url.map(|url| url.trim_end_matches('/').to_string()),
97        }
98    }
99
100    pub fn destinations(&self) -> &[Destination] {
101        &self.destinations
102    }
103
104    pub fn get(&self, name: &str) -> Option<&Destination> {
105        self.destinations.iter().find(|destination| destination.name == name)
106    }
107
108    /// Whether any destination hears `event` - checked before assembling one,
109    /// so an installation with no webhooks pays nothing for them.
110    pub fn anyone_hears(&self, event: EventKind) -> bool {
111        self.destinations.iter().any(|destination| destination.hears(event))
112    }
113
114    fn link(&self, path: &str) -> Option<String> {
115        self.public_url.as_ref().map(|base| format!("{base}{path}"))
116    }
117}
118
119// The event ---------------------------------------------------------------------
120
121/// One thing that happened, as every destination is told it.
122///
123/// This is the `json` kind's body verbatim, and what the chat kinds render
124/// their text from - so a Slack message can never say something the payload
125/// does not.
126#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
127pub struct Event {
128    pub version: u32,
129    /// The same for every destination this event went to, and for every retry
130    /// to one. What a receiver deduplicates on.
131    pub id: Uuid,
132    pub event: EventKind,
133    pub occurred_at: DateTime<Utc>,
134    /// The server that said it, so a receiver fed by several can tell them
135    /// apart and a message can name the version that sent it.
136    pub server_version: String,
137    /// Where to look, when the installation knows its own address.
138    #[serde(default, skip_serializing_if = "Option::is_none")]
139    pub link: Option<String>,
140    /// Who it is about. Absent from a test.
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub person: Option<Person>,
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub alert: Option<AlertPayload>,
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub day: Option<DayPayload>,
147    /// Who answered an alert, by name - on `alert.acknowledged` only. So the
148    /// channel knows it has been looked at and two managers do not both
149    /// chase it.
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub by: Option<String>,
152}
153
154/// The person an event is about.
155#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, sqlx::FromRow)]
156pub struct Person {
157    pub id: Uuid,
158    pub name: String,
159    pub department: Option<String>,
160}
161
162/// An alert as it fired. The figures are the ones it fired on, never
163/// re-derived: a message already read must not change under its reader
164/// (ADR 0018).
165#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, sqlx::FromRow)]
166pub struct AlertPayload {
167    pub id: Uuid,
168    pub rule: AlertRule,
169    pub observed_seconds: i64,
170    pub against_seconds: Option<i64>,
171    pub subject_date: Option<NaiveDate>,
172    pub fired_at: DateTime<Utc>,
173}
174
175/// A day as it arrived finished.
176#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
177pub struct DayPayload {
178    pub date: NaiveDate,
179    pub kind: WorkdayKind,
180    pub started_at: DateTime<Utc>,
181    pub ended_at: DateTime<Utc>,
182    /// The span minus the pauses, in seconds.
183    pub worked_seconds: i64,
184}
185
186impl Event {
187    fn new(event: EventKind, id: Uuid, now: DateTime<Utc>) -> Self {
188        Self {
189            version: PAYLOAD_VERSION,
190            id,
191            event,
192            occurred_at: now,
193            server_version: env!("CARGO_PKG_VERSION").to_string(),
194            link: None,
195            person: None,
196            alert: None,
197            day: None,
198            by: None,
199        }
200    }
201
202    /// A step in an alert's life.
203    pub fn alert(event: EventKind, alert: AlertPayload, person: Person, by: Option<String>, webhooks: &Webhooks, now: DateTime<Utc>) -> Self {
204        Self {
205            // One id per alert per step: the sweep that raises it and a
206            // second sweep racing it produce the same event, which the unique
207            // key then refuses to queue twice.
208            link: webhooks.link(&format!("/team/{}", person.id)),
209            person: Some(person),
210            by,
211            ..Self::new(event, derived_id(event, &[alert.id.as_bytes()]), now)
212        }
213        .with_alert(alert)
214    }
215
216    fn with_alert(mut self, alert: AlertPayload) -> Self {
217        self.alert = Some(alert);
218        self
219    }
220
221    /// A day that has just arrived finished.
222    pub fn day_closed(workday_id: Uuid, day: DayPayload, person: Person, webhooks: &Webhooks, now: DateTime<Utc>) -> Self {
223        // Keyed on the day and its end, so the same close re-sent is the same
224        // event, and a day reopened and closed again later is a new one.
225        let id = derived_id(EventKind::DayClosed, &[workday_id.as_bytes(), day.ended_at.to_rfc3339().as_bytes()]);
226        Self {
227            link: webhooks.link(&format!("/team/{}", person.id)),
228            person: Some(person),
229            day: Some(day),
230            ..Self::new(EventKind::DayClosed, id, now)
231        }
232    }
233
234    /// What an administrator sends to see a channel work.
235    pub fn test(webhooks: &Webhooks, now: DateTime<Utc>) -> Self {
236        Self {
237            link: webhooks.link("/"),
238            ..Self::new(EventKind::Test, Uuid::new_v4(), now)
239        }
240    }
241}
242
243/// An id that is a function of what the event is about.
244///
245/// Deterministic rather than random, which is what makes queuing idempotent:
246/// `(event_id, destination)` is unique, so the same fact reached twice - two
247/// overlapping sweeps, a retried upload - is queued once.
248fn derived_id(event: EventKind, parts: &[&[u8]]) -> Uuid {
249    let mut hash = Sha256::new();
250    hash.update(event.name().as_bytes());
251    for part in parts {
252        hash.update([0u8]);
253        hash.update(part);
254    }
255    let digest = hash.finalize();
256    let mut bytes = [0u8; 16];
257    bytes.copy_from_slice(&digest[..16]);
258    uuid::Builder::from_custom_bytes(bytes).into_uuid()
259}
260
261/// Reads who an event is about.
262pub async fn person(conn: &mut PgConnection, user_id: Uuid) -> Result<Person, ApiError> {
263    Ok(sqlx::query_as(
264        "SELECT u.id, u.display_name AS name, d.name AS department
265         FROM users u LEFT JOIN departments d ON d.id = u.department_id
266         WHERE u.id = $1",
267    )
268    .bind(user_id)
269    .fetch_one(conn)
270    .await?)
271}
272
273/// Queues an event for every destination that hears it and may hear about
274/// this person. Returns how many rows were written.
275///
276/// Takes the caller's connection so it runs inside the caller's transaction:
277/// the fact and its announcement commit together or not at all.
278pub async fn enqueue(conn: &mut PgConnection, webhooks: &Webhooks, event: &Event) -> Result<u64, ApiError> {
279    let department = event.person.as_ref().and_then(|person| person.department.as_deref());
280    let mut queued = 0;
281    for destination in &webhooks.destinations {
282        if !destination.hears(event.event) {
283            continue;
284        }
285        // A destination for one department hears about nobody else - not
286        // about people with no department either. The same boundary every
287        // screen draws around a manager (ADR 0009).
288        if let Some(wanted) = &destination.department
289            && !department.is_some_and(|actual| actual.eq_ignore_ascii_case(wanted))
290        {
291            continue;
292        }
293        queued += enqueue_to(conn, destination, event).await?;
294    }
295    Ok(queued)
296}
297
298/// Queues an event for one destination, whatever it subscribes to.
299async fn enqueue_to(conn: &mut PgConnection, destination: &Destination, event: &Event) -> Result<u64, ApiError> {
300    let written = sqlx::query(
301        "INSERT INTO webhook_deliveries (event_id, destination, event, user_id, payload, created_at, next_attempt_at)
302         VALUES ($1, $2, $3, $4, $5, $6, $6)
303         ON CONFLICT (event_id, destination) DO NOTHING",
304    )
305    .bind(event.id)
306    .bind(&destination.name)
307    .bind(event.event)
308    .bind(event.person.as_ref().map(|person| person.id))
309    .bind(sqlx::types::Json(event))
310    .bind(event.occurred_at)
311    .execute(conn)
312    .await?;
313    Ok(written.rows_affected())
314}
315
316// The API -------------------------------------------------------------------------
317
318/// One destination, as the screen shows it.
319#[derive(Debug, Serialize)]
320pub struct DestinationView {
321    pub name: String,
322    pub kind: Kind,
323    /// The host or the chat - never the credential.
324    pub target: String,
325    pub events: Vec<EventKind>,
326    pub department: Option<String>,
327    /// Whether the department it names exists. `false` is a destination that
328    /// will hear nothing at all - a rename, or a typo in the environment -
329    /// and the one place that can say so is here.
330    pub department_exists: Option<bool>,
331    pub pending: i64,
332    pub delivered: i64,
333    pub abandoned: i64,
334    pub last_delivered_at: Option<DateTime<Utc>>,
335    pub last_error: Option<String>,
336    pub last_error_at: Option<DateTime<Utc>>,
337}
338
339/// One delivery, newest first on the screen.
340#[derive(Debug, Serialize, sqlx::FromRow)]
341pub struct DeliveryView {
342    pub id: Uuid,
343    pub event_id: Uuid,
344    pub destination: String,
345    pub event: EventKind,
346    pub person: Option<String>,
347    pub created_at: DateTime<Utc>,
348    pub attempts: i32,
349    pub next_attempt_at: DateTime<Utc>,
350    pub delivered_at: Option<DateTime<Utc>>,
351    pub abandoned_at: Option<DateTime<Utc>>,
352    pub last_status: Option<i32>,
353    pub last_error: Option<String>,
354}
355
356/// The screen: where events go, and how the last ones went.
357#[derive(Debug, Serialize)]
358pub struct Overview {
359    pub destinations: Vec<DestinationView>,
360    pub recent: Vec<DeliveryView>,
361    /// Whether messages can link back. Said so an operator who sees plain
362    /// text in the channel knows which setting would add the link.
363    pub links: bool,
364}
365
366#[derive(Debug, sqlx::FromRow)]
367struct Counts {
368    destination: String,
369    pending: i64,
370    delivered: i64,
371    abandoned: i64,
372    last_delivered_at: Option<DateTime<Utc>>,
373}
374
375/// Answers the destinations and the recent deliveries. Administrators only:
376/// the log names people and the screen names where their alerts go.
377pub async fn overview(State(state): State<AppState>, user: CurrentUser) -> Result<impl IntoResponse, ApiError> {
378    user.require_admin()?;
379
380    let counts: Vec<Counts> = sqlx::query_as(
381        "SELECT destination,
382                count(*) FILTER (WHERE delivered_at IS NULL AND abandoned_at IS NULL) AS pending,
383                count(*) FILTER (WHERE delivered_at IS NOT NULL) AS delivered,
384                count(*) FILTER (WHERE abandoned_at IS NOT NULL) AS abandoned,
385                max(delivered_at) AS last_delivered_at
386         FROM webhook_deliveries GROUP BY destination",
387    )
388    .fetch_all(&state.pool)
389    .await?;
390
391    let failures: Vec<(String, String, DateTime<Utc>)> = sqlx::query_as(
392        "SELECT DISTINCT ON (destination) destination, last_error, coalesce(abandoned_at, next_attempt_at)
393         FROM webhook_deliveries
394         WHERE last_error IS NOT NULL AND delivered_at IS NULL
395         ORDER BY destination, created_at DESC",
396    )
397    .fetch_all(&state.pool)
398    .await?;
399
400    let departments: Vec<String> = sqlx::query_scalar("SELECT name FROM departments").fetch_all(&state.pool).await?;
401
402    let destinations = state
403        .webhooks
404        .destinations
405        .iter()
406        .map(|destination| {
407            let count = counts.iter().find(|count| count.destination == destination.name);
408            let failure = failures.iter().find(|(name, ..)| *name == destination.name);
409            DestinationView {
410                name: destination.name.clone(),
411                kind: destination.kind,
412                target: destination.shown_target(),
413                events: destination.events.clone(),
414                department: destination.department.clone(),
415                department_exists: destination
416                    .department
417                    .as_ref()
418                    .map(|wanted| departments.iter().any(|name| name.eq_ignore_ascii_case(wanted))),
419                pending: count.map_or(0, |count| count.pending),
420                delivered: count.map_or(0, |count| count.delivered),
421                abandoned: count.map_or(0, |count| count.abandoned),
422                last_delivered_at: count.and_then(|count| count.last_delivered_at),
423                last_error: failure.map(|(_, error, _)| error.clone()),
424                last_error_at: failure.map(|(.., at)| *at),
425            }
426        })
427        .collect();
428
429    let recent: Vec<DeliveryView> = sqlx::query_as(
430        "SELECT w.id, w.event_id, w.destination, w.event, u.display_name AS person, w.created_at, w.attempts,
431                w.next_attempt_at, w.delivered_at, w.abandoned_at, w.last_status, w.last_error
432         FROM webhook_deliveries w LEFT JOIN users u ON u.id = w.user_id
433         ORDER BY w.created_at DESC, w.id
434         LIMIT 50",
435    )
436    .fetch_all(&state.pool)
437    .await?;
438
439    Ok(Json(Overview {
440        destinations,
441        recent,
442        links: state.webhooks.public_url.is_some(),
443    }))
444}
445
446/// Queues a test message for one destination. Administrators only.
447///
448/// Queued like any event rather than sent inline, so the test exercises the
449/// path real alerts take - the dispatcher, its retries, its log - and the
450/// screen shows the outcome in the same list.
451pub async fn send_test(State(state): State<AppState>, user: CurrentUser, Path(name): Path<String>) -> Result<impl IntoResponse, ApiError> {
452    user.require_admin()?;
453
454    let Some(destination) = state.webhooks.get(&name) else {
455        return Err(ApiError::new(StatusCode::NOT_FOUND, format!("no destination named `{name}` is configured")));
456    };
457
458    let event = Event::test(&state.webhooks, Utc::now());
459    let mut conn = state.pool.acquire().await?;
460    enqueue_to(&mut conn, destination, &event).await?;
461    drop(conn);
462
463    // Posting into a team's channel is something done in the installation's
464    // name, and the log is where "who sent that?" gets answered.
465    audit::Entry::new(audit::action::WEBHOOK_TESTED)
466        .by(user.user_id)
467        .by_email(&user.email)
468        .with(serde_json::json!({ "destination": destination.name, "event_id": event.id }))
469        .record(&state.pool)
470        .await;
471
472    Ok((
473        StatusCode::ACCEPTED,
474        Json(serde_json::json!({ "event_id": event.id, "destination": destination.name })),
475    ))
476}
477
478#[cfg(test)]
479mod tests {
480    use super::*;
481
482    fn slack(name: &str, options: &str) -> Destination {
483        Destination::parse(name, &format!("slack https://hooks.slack.com/services/T/B/X {options}")).unwrap()
484    }
485
486    #[test]
487    fn every_webhook_variable_is_read_and_nothing_else() {
488        let webhooks = Webhooks::from_vars(
489            [
490                ("KASL_WEBHOOK_ZED".to_string(), "slack https://hooks.slack.com/z".to_string()),
491                ("KASL_WEBHOOK_ALPHA".to_string(), "slack https://hooks.slack.com/a".to_string()),
492                ("KASL_AGENTS".to_string(), "a@b.c:token".to_string()),
493                ("PATH".to_string(), "/usr/bin".to_string()),
494            ],
495            Some("https://kasl.example.com/".to_string()),
496        )
497        .unwrap();
498        let names: Vec<&str> = webhooks.destinations().iter().map(|d| d.name.as_str()).collect();
499        assert_eq!(names, ["alpha", "zed"], "sorted, and only the webhook variables");
500        assert_eq!(webhooks.link("/team/1").as_deref(), Some("https://kasl.example.com/team/1"));
501    }
502
503    #[test]
504    fn one_bad_destination_stops_the_start() {
505        let error = Webhooks::from_vars(
506            [
507                ("KASL_WEBHOOK_GOOD".to_string(), "slack https://hooks.slack.com/a".to_string()),
508                ("KASL_WEBHOOK_BAD".to_string(), "slak https://hooks.slack.com/b".to_string()),
509            ],
510            None,
511        )
512        .unwrap_err();
513        assert!(error.contains("KASL_WEBHOOK_BAD"), "{error}");
514
515        let error = Webhooks::from_vars([], Some("kasl.example.com".to_string())).unwrap_err();
516        assert!(error.contains("KASL_PUBLIC_URL"), "{error}");
517    }
518
519    #[test]
520    fn the_same_fact_gets_the_same_id() {
521        let alert = Uuid::new_v4();
522        assert_eq!(
523            derived_id(EventKind::AlertRaised, &[alert.as_bytes()]),
524            derived_id(EventKind::AlertRaised, &[alert.as_bytes()])
525        );
526        assert_ne!(
527            derived_id(EventKind::AlertRaised, &[alert.as_bytes()]),
528            derived_id(EventKind::AlertResolved, &[alert.as_bytes()]),
529            "raising and resolving one alert are two events",
530        );
531    }
532
533    #[test]
534    fn anyone_hears_follows_the_subscriptions() {
535        let webhooks = Webhooks::new(vec![slack("KASL_WEBHOOK_A", "events=day.closed")], None);
536        assert!(webhooks.anyone_hears(EventKind::DayClosed));
537        assert!(!webhooks.anyone_hears(EventKind::AlertRaised));
538        assert!(!Webhooks::default().anyone_hears(EventKind::AlertRaised));
539    }
540}
kasl_server/webhooks/mod.rs

kasl_server/webhooks/
mod.rs
