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

kasl_server/webhooks/
mod.rs
