diff --git a/README.md b/README.md index 04bc8c8..71bd32d 100644 --- a/README.md +++ b/README.md @@ -85,10 +85,15 @@ hostname is reachable outside a trusted network. | `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file | | `TERDUT_ARCHIVE_AFTER` | `168h` (7d) | How long a resolved alert or incident stays in the default list before being auto-archived | | `TERDUT_STALE_AFTER` | `6h` | How long a firing alert may go without a refreshing webhook before it is treated as resolved — **must exceed your Alertmanager `repeat_interval`** | +| `TERDUT_NTFY_URL` | — | ntfy server to publish push notifications to. Empty disables notifications entirely | +| `TERDUT_NTFY_TOKEN` | — | Bearer token for an access-controlled ntfy | +| `TERDUT_NTFY_FALLBACK_TOPIC` | — | Topic used when nobody is on call | +| `TERDUT_PUBLIC_URL` | — | Base URL a phone uses to reach this server, for the link and Acknowledge button inside a notification | +| `TERDUT_NOTIFY_REPEAT` | `15m` | How long an incident may sit unacknowledged before it is paged again. `0` notifies once and never repeats | Durations use Go syntax (`30m`, `12h`, `168h`). An unparseable value falls back to the default. -In the Helm chart the two sweeper durations are set via `sweeper.staleAfter` and `sweeper.archiveAfter`. +In the Helm chart the two sweeper durations are set via `sweeper.staleAfter` and `sweeper.archiveAfter`, and notifications via the `notify.*` values. --- @@ -170,6 +175,44 @@ A new incident is assigned to whoever holds today's schedule entry at the moment it opens (`GET /api/schedule/current`). If nobody is scheduled it opens unassigned. Reassign with `POST /api/incidents/{id}/assign`. +### Push notifications + +With `TERDUT_NTFY_URL` set, an incident that opens is pushed to the on-call +person's phone through [ntfy](https://ntfy.sh). Set each user's topic with +`PUT /api/users/{id}/notify`; a user with no topic falls back to +`TERDUT_NTFY_FALLBACK_TOPIC`, as does an incident that opens with nobody on call. +If neither yields a topic, nothing is queued. + +Three things get pushed: + +- **triggered** — an incident opened. Priority follows severity (`critical` maps + to ntfy's max priority, the one that overrides the phone's quiet settings). +- **reminder** — the incident is still `triggered` after `TERDUT_NOTIFY_REPEAT`. + Repeats until somebody acts. Acknowledging, snoozing, resolving or archiving + all stop it — snooze is the mute button. +- **resolved** — every alert under the incident stopped firing. Only sent to + whoever was paged in the first place, and only for the automatic cascade: + resolving by hand pushes nothing, since the person who did it already knows. + +Notifications carry an **Acknowledge** button that acknowledges the incident +without opening anything. It POSTs to `/api/notify/ack/{token}`, an +unauthenticated route authorised by the 256-bit single-use token in its path — +minted fresh per notification, scoped to one incident and one action, and valid +for 24 hours. A real API key is never put in a notification, because the message +is stored on the ntfy server and cached on the device. + +Two consequences worth planning for: + +- `/api/notify/ack/{token}` **must stay publicly reachable**, or the button will + not work when the responder is off your network. +- Notifications sent to the fallback topic carry **no** Acknowledge button. The + topic is shared, and a button on it would let any subscriber acknowledge as + somebody else. + +Delivery is a queue, not an inline call: the webhook writes a row and a +background notifier sends it within 30 seconds, retrying with exponential +backoff up to 8 attempts. Nothing about ingestion blocks on ntfy being reachable. + ### Stale alert expiry A resolved webhook is the only signal that an alert has stopped firing, so a @@ -209,6 +252,7 @@ Authorization: Bearer | `GET` | `/api/users` | List users | | `POST` | `/api/users` | Create user `{"username","email"}` | | `DELETE` | `/api/users/{id}` | Delete user (cascades to keys) | +| `PUT` | `/api/users/{id}/notify` | Set push notification target `{"ntfy_topic"}` — empty string clears it | | `POST` | `/api/users/{id}/api-keys` | Issue API key `{"name"}` — key shown once | | `DELETE` | `/api/users/{id}/api-keys/{keyID}` | Revoke API key | @@ -218,6 +262,12 @@ Authorization: Bearer |---|---|---| | `POST` | `/api/alertmanager/webhook` | Alertmanager v4 webhook receiver (no auth) | +### Notifications + +| Method | Path | Description | +|---|---|---| +| `POST` | `/api/notify/ack/{token}` | Acknowledge an incident from a push notification's Acknowledge button. No auth: the single-use token in the path is the credential. Must stay publicly reachable | + ### Incidents | Method | Path | Description | diff --git a/charts/terdut-server/Chart.yaml b/charts/terdut-server/Chart.yaml index 0b8ad33..8cb38e6 100644 --- a/charts/terdut-server/Chart.yaml +++ b/charts/terdut-server/Chart.yaml @@ -2,5 +2,5 @@ apiVersion: v2 name: terdut-server description: A Helm chart for Terminal Duty — on-call alert management server type: application -version: 0.5.0 +version: 0.6.0 appVersion: "latest" diff --git a/charts/terdut-server/templates/deployment.yaml b/charts/terdut-server/templates/deployment.yaml index e51d6fb..566e21f 100644 --- a/charts/terdut-server/templates/deployment.yaml +++ b/charts/terdut-server/templates/deployment.yaml @@ -33,6 +33,23 @@ spec: value: "{{ .Values.sweeper.staleAfter }}" - name: TERDUT_ARCHIVE_AFTER value: "{{ .Values.sweeper.archiveAfter }}" + {{- if .Values.notify.ntfyUrl }} + - name: TERDUT_NTFY_URL + value: "{{ .Values.notify.ntfyUrl }}" + - name: TERDUT_NTFY_FALLBACK_TOPIC + value: "{{ .Values.notify.fallbackTopic }}" + - name: TERDUT_NOTIFY_REPEAT + value: "{{ .Values.notify.repeatEvery }}" + - name: TERDUT_PUBLIC_URL + value: "{{ .Values.notify.publicUrl | default (printf "https://%s" .Values.networking.hostname) }}" + {{- if .Values.notify.tokenSecret.name }} + - name: TERDUT_NTFY_TOKEN + valueFrom: + secretKeyRef: + name: {{ .Values.notify.tokenSecret.name }} + key: {{ .Values.notify.tokenSecret.key }} + {{- end }} + {{- end }} volumeMounts: - name: data mountPath: /data diff --git a/charts/terdut-server/values.yaml b/charts/terdut-server/values.yaml index 35ab810..7087923 100644 --- a/charts/terdut-server/values.yaml +++ b/charts/terdut-server/values.yaml @@ -26,6 +26,30 @@ sweeper: # How long a resolved alert stays in the default list before auto-archiving. archiveAfter: 168h +notify: + # ntfy server that push notifications are published to, e.g. + # http://ntfy.ntfy.svc.cluster.local. Empty disables notifications entirely. + ntfyUrl: "" + # Topic used when nobody is on call today. Notifications sent here carry no + # Acknowledge button: the topic is shared, so there is no user to attribute an + # acknowledgement to. Leave empty to send nothing when the schedule is unset. + fallbackTopic: "" + # How long an incident may sit unacknowledged before it is paged again. + # Set to 0 to notify once and never repeat. + repeatEvery: 15m + # Base URL a phone uses to reach this server, for the link and the Acknowledge + # button inside a notification. Defaults to https://. + # + # The Acknowledge button is a POST to /api/notify/ack/{token} from the + # responder's phone, so that path has to stay publicly reachable — it is + # authorised by the single-use token in the URL, not by network placement. + publicUrl: "" + # Optional bearer token for an access-controlled ntfy, read from an existing + # Secret. Leave name empty for an open ntfy. + tokenSecret: + name: "" + key: token + bootstrap: enabled: true username: admin diff --git a/cmd/terdut/main.go b/cmd/terdut/main.go index 2b6f9be..d0c7223 100644 --- a/cmd/terdut/main.go +++ b/cmd/terdut/main.go @@ -28,7 +28,15 @@ func main() { log.Fatalf("migrate: %v", err) } - router := api.NewRouter(database) + notify := api.NotifyConfig{ + BaseURL: cfg.NtfyURL, + Token: cfg.NtfyToken, + FallbackTopic: cfg.NtfyFallbackTopic, + PublicURL: cfg.PublicURL, + RepeatEvery: cfg.NotifyRepeat, + } + + router := api.NewRouter(database, notify) srv := &http.Server{ Addr: cfg.Addr, @@ -42,6 +50,7 @@ func main() { defer stop() go api.StartArchiver(ctx, database, cfg.ArchiveAfter, cfg.StaleAfter) + go api.StartNotifier(ctx, database, notify) go func() { log.Printf("terdut-server %s listening on %s", version, cfg.Addr) diff --git a/internal/api/alertmanager.go b/internal/api/alertmanager.go index 462707a..9729fd9 100644 --- a/internal/api/alertmanager.go +++ b/internal/api/alertmanager.go @@ -58,7 +58,7 @@ type ingested struct { justResolved bool } -func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc { +func handleAlertmanagerWebhook(db *sql.DB, notify NotifyConfig) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var payload amPayload if err := decodeJSON(r, &payload); err != nil { @@ -69,7 +69,7 @@ func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc { // Alertmanager retries anything that is not 2xx, and a retry of a payload // we failed to store is more useful than an error it cannot act on — so // failures are logged, not surfaced. - if err := ingest(r.Context(), db, payload); err != nil { + if err := ingest(r.Context(), db, notify, payload); err != nil { log.Printf("webhook ingest (group %q): %v", payload.GroupKey, err) } @@ -80,7 +80,7 @@ func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc { // ingest stores a payload's alerts and reconciles the incident for its group. // The whole payload is one transaction: an incident that opened but whose alerts // failed to link would be a work item nobody could act on. -func ingest(ctx context.Context, db *sql.DB, payload amPayload) error { +func ingest(ctx context.Context, db *sql.DB, notify NotifyConfig, payload amPayload) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return err @@ -96,7 +96,7 @@ func ingest(ctx context.Context, db *sql.DB, payload amPayload) error { // resolution cascade are recomputed once per incident at the end. touched := map[int64]bool{} - incidentID, err := incidentForGroup(ctx, tx, payload, accepted) + incidentID, err := incidentForGroup(ctx, tx, notify, payload, accepted) if err != nil { return err } @@ -258,7 +258,7 @@ func upsertAlerts(ctx context.Context, tx *sql.Tx, alerts []amAlert) ([]ingested // something actually started firing. Without that, a manually resolved incident // would reappear on the next repeat_interval re-send of an alert that never // stopped, and manual resolution would be meaningless. -func incidentForGroup(ctx context.Context, tx *sql.Tx, payload amPayload, accepted []ingested) (int64, error) { +func incidentForGroup(ctx context.Context, tx *sql.Tx, notify NotifyConfig, payload amPayload, accepted []ingested) (int64, error) { var firstName string anyFiring, anyNew := false, false for _, a := range accepted { @@ -297,12 +297,12 @@ func incidentForGroup(ctx context.Context, tx *sql.Tx, payload amPayload, accept if !anyNew { return 0, nil } - return openIncident(ctx, tx, groupKey, payload.GroupLabels, firstName) + return openIncident(ctx, tx, notify, groupKey, payload.GroupLabels, firstName) } // openIncident creates an incident for a group and assigns it to whoever is on // call today, which is the point at which the schedule stops being decorative. -func openIncident(ctx context.Context, tx *sql.Tx, groupKey string, groupLabels map[string]string, fallbackName string) (int64, error) { +func openIncident(ctx context.Context, tx *sql.Tx, notify NotifyConfig, groupKey string, groupLabels map[string]string, fallbackName string) (int64, error) { onCall, err := currentOnCall(ctx, tx) if err != nil { return 0, err @@ -335,6 +335,13 @@ func openIncident(ctx context.Context, tx *sql.Tx, groupKey string, groupLabels return 0, err } } + + // Queue the page, but do not send it here: this runs inside the webhook's + // transaction on a single-connection pool, so an HTTP call would hold up + // every other request. The notifier picks the row up within a tick. + if err := enqueueOpened(ctx, tx, notify, id, onCall); err != nil { + return 0, err + } return id, nil } diff --git a/internal/api/api_test.go b/internal/api/api_test.go index 730d3f0..601c5db 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -20,12 +20,20 @@ import ( // tests can age rows directly — the sweeper's inputs are wall-clock timestamps. type ts struct { *httptest.Server - key string - db *sql.DB + key string + db *sql.DB + notify api.NotifyConfig } -func newTS(t *testing.T) *ts { +// newTS builds a server over a fresh in-memory database. Notifications are off +// unless a NotifyConfig is passed, so tests that predate them are unaffected. +func newTS(t *testing.T, notify ...api.NotifyConfig) *ts { t.Helper() + var cfg api.NotifyConfig + if len(notify) > 0 { + cfg = notify[0] + } + database, err := db.Open(":memory:") if err != nil { t.Fatalf("open db: %v", err) @@ -33,7 +41,7 @@ func newTS(t *testing.T) *ts { if err := db.Migrate(database); err != nil { t.Fatalf("migrate: %v", err) } - srv := httptest.NewServer(api.NewRouter(database)) + srv := httptest.NewServer(api.NewRouter(database, cfg)) t.Cleanup(func() { srv.Close(); database.Close() }) body, _ := json.Marshal(map[string]string{"username": "admin", "email": "admin@test.com"}) @@ -49,7 +57,7 @@ func newTS(t *testing.T) *ts { json.NewDecoder(resp.Body).Decode(&result) key := result["api_key"].(map[string]any)["key"].(string) - return &ts{Server: srv, key: key, db: database} + return &ts{Server: srv, key: key, db: database, notify: cfg} } // exec runs a statement against the test database. diff --git a/internal/api/archiver.go b/internal/api/archiver.go index da87ede..bc2fb75 100644 --- a/internal/api/archiver.go +++ b/internal/api/archiver.go @@ -44,6 +44,7 @@ func Sweep(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Durati resolveSettledIncidents(ctx, db) archiveResolved(ctx, db, archiveAfter) archiveResolvedIncidents(ctx, db, archiveAfter) + purgeAckTokens(ctx, db) } // expireStale resolves firing alerts that Alertmanager has stopped refreshing. diff --git a/internal/api/incident_store.go b/internal/api/incident_store.go index b74f3dc..bcc69aa 100644 --- a/internal/api/incident_store.go +++ b/internal/api/incident_store.go @@ -223,7 +223,32 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e if n == 0 { return false, nil } - return true, logEvent(ctx, q, incidentID, evResolved, nil, nil, nil) + if err := logEvent(ctx, q, incidentID, evResolved, nil, nil, nil); err != nil { + return false, err + } + // The all-clear goes only to whoever was paged in the first place, which + // enqueueResolved works out from the incident's own notification history. + // Manual resolution sends nothing: the person who closed it already knows. + return true, enqueueResolved(ctx, q, incidentID) +} + +// acknowledgeIncident records that userID has picked an incident up, and reports +// whether it changed anything — an already-resolved incident is left alone. +// Shared by the authenticated handler and the Acknowledge button in a push +// notification, so both write the same state and the same timeline entry. +func acknowledgeIncident(ctx context.Context, q querier, incidentID, userID int64) (bool, error) { + res, err := q.ExecContext(ctx, ` + UPDATE incidents + SET status = 'acknowledged', acknowledged_by = ?, acknowledged_at = ? + WHERE id = ? AND resolved_at IS NULL`, + userID, time.Now().Unix(), incidentID) + if err != nil { + return false, err + } + if n, _ := res.RowsAffected(); n == 0 { + return false, nil + } + return true, logEvent(ctx, q, incidentID, evAcknowledged, &userID, nil, nil) } // openIncidentForAlert returns the open incident an alert currently belongs to, diff --git a/internal/api/incidents.go b/internal/api/incidents.go index f1953c8..1b73edb 100644 --- a/internal/api/incidents.go +++ b/internal/api/incidents.go @@ -189,13 +189,16 @@ func handleIncidentAcknowledge(db *sql.DB) http.HandlerFunc { return } user, _ := userFromContext(r.Context()) - if !updateOpenIncident(w, r, db, id, - `UPDATE incidents SET status = 'acknowledged', acknowledged_by = ?, acknowledged_at = ? - WHERE id = ? AND resolved_at IS NULL`, user.ID, time.Now().Unix(), id) { + acked, err := acknowledgeIncident(r.Context(), db, id, user.ID) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) return } - if err := logEvent(r.Context(), db, id, evAcknowledged, &user.ID, nil, nil); err != nil { - respond(w, http.StatusInternalServerError, errResp("internal error")) + if !acked { + if !incidentExists(w, r, db, id) { + return + } + respond(w, http.StatusConflict, errResp("incident is resolved")) return } respondIncident(w, r, db, id) diff --git a/internal/api/notifier.go b/internal/api/notifier.go new file mode 100644 index 0000000..348e625 --- /dev/null +++ b/internal/api/notifier.go @@ -0,0 +1,532 @@ +package api + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "log" + "net/http" + "strings" + "time" + + "github.com/yeniklas/terdut-server/internal/models" +) + +const ( + // notifyInterval is how often the notifier looks for work. The archiver's + // 15 minute tick is far too coarse for something that has to wake a person. + notifyInterval = 30 * time.Second + + // notifyRetryBase and notifyRetryMax bound the delivery backoff. ntfy being + // briefly unreachable should not lose the page. + notifyRetryBase = 30 * time.Second + notifyRetryMax = 15 * time.Minute + + // notifyMaxAttempts stops a permanently undeliverable row from being retried + // forever. It keeps last_error so the reason survives. + notifyMaxAttempts = 8 + + // notifyBatch caps one delivery pass, so a large backlog cannot hold the + // single database connection for an unbounded stretch. + notifyBatch = 100 + + // ackTokenTTL is how long the Acknowledge button in a notification keeps + // working. Past this the notification is stale enough that the responder + // should look at the incident rather than blind-acknowledge it. + ackTokenTTL = 24 * time.Hour +) + +// Notification kinds, recording why a push was sent. +const ( + notifyTriggered = "triggered" + notifyReminder = "reminder" + notifyResolved = "resolved" +) + +// NotifyConfig is everything the notifier needs to reach ntfy and to build URLs +// a phone can follow back to this server. +type NotifyConfig struct { + // BaseURL is the ntfy server. Empty disables notifications entirely: no + // goroutine, and nothing is ever enqueued. + BaseURL string + + // Token is an optional bearer token for an access-controlled ntfy. + Token string + + // FallbackTopic receives incidents that open with nobody on call. Those + // notifications carry no Acknowledge button — there is no user to attribute + // the acknowledgement to, and putting one on a shared topic would let any + // subscriber acknowledge as somebody else. + FallbackTopic string + + // PublicURL is the base URL a phone uses to reach this server, for the + // notification's click target and its Acknowledge action. Without it a + // notification is informational only. + PublicURL string + + // RepeatEvery is how long an incident may sit unacknowledged before it is + // notified again. Zero disables reminders. + RepeatEvery time.Duration +} + +// enabled reports whether notifications are configured at all. +func (c NotifyConfig) enabled() bool { return c.BaseURL != "" } + +// notifyClient is shared: a page is small and infrequent, and the timeout is +// what keeps a hung ntfy from stalling the delivery pass. +var notifyClient = &http.Client{Timeout: 10 * time.Second} + +// StartNotifier delivers queued notifications until ctx is cancelled, starting +// with an immediate pass so a restart flushes whatever the last one left behind. +func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) { + if !cfg.enabled() { + log.Print("notifier: disabled (no ntfy URL configured)") + return + } + log.Printf("notifier: publishing to %s", cfg.BaseURL) + + ticker := time.NewTicker(notifyInterval) + defer ticker.Stop() + + NotifySweep(ctx, db, cfg) + for { + select { + case <-ticker.C: + NotifySweep(ctx, db, cfg) + case <-ctx.Done(): + return + } + } +} + +// NotifySweep runs a single pass: queue reminders for incidents nobody has +// picked up, then deliver everything that is due. Reminders are queued first so +// a freshly due one goes out in the same pass rather than a tick later. +// Exported so tests can drive a pass without waiting on the ticker. +func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) { + enqueueReminders(ctx, db, cfg) + deliverPending(ctx, db, cfg) +} + +// enqueueReminders re-notifies incidents that are still sitting untouched. +// +// The stop conditions are the incident states that already mean "somebody has +// this": acknowledged, snoozed, resolved, archived. Snooze in particular is the +// mute button — a deliberate "not now" that should not keep buzzing — which is +// why there is no separate reminder cap. +// +// The previous notification must have actually been sent before another is +// queued, so an ntfy outage produces a retry backlog rather than a reminder +// backlog that all lands at once when it comes back. +func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) { + if cfg.RepeatEvery <= 0 { + return + } + now := time.Now() + + type due struct { + incidentID int64 + userID *int64 + topic string + } + + rows, err := db.QueryContext(ctx, ` + SELECT n.incident_id, n.user_id, n.topic + FROM notifications n + JOIN incidents i ON i.id = n.incident_id + WHERE n.id = (SELECT MAX(id) FROM notifications WHERE incident_id = n.incident_id) + AND n.sent_at IS NOT NULL + AND n.created_at <= ? + AND i.resolved_at IS NULL + AND i.archived_at IS NULL + AND i.status = 'triggered' + AND (i.snoozed_until IS NULL OR i.snoozed_until <= ?)`, + now.Add(-cfg.RepeatEvery).Unix(), now.Unix()) + if err != nil { + log.Printf("notifier: find reminders: %v", err) + return + } + + // Collected before inserting: the pool is limited to a single connection, so + // an open cursor would block the writes behind it. + var pending []due + for rows.Next() { + var d due + if err := rows.Scan(&d.incidentID, &d.userID, &d.topic); err != nil { + rows.Close() + log.Printf("notifier: scan reminder: %v", err) + return + } + pending = append(pending, d) + } + if err := rows.Err(); err != nil { + rows.Close() + log.Printf("notifier: find reminders: %v", err) + return + } + rows.Close() + + for _, d := range pending { + if err := enqueueNotification(ctx, db, d.incidentID, d.userID, d.topic, notifyReminder); err != nil { + log.Printf("notifier: queue reminder for incident %d: %v", d.incidentID, err) + } + } + if len(pending) > 0 { + log.Printf("notifier: queued %d reminder(s)", len(pending)) + } +} + +// outboxRow is one queued notification, read before any HTTP happens. +type outboxRow struct { + id int64 + incidentID int64 + userID *int64 + topic string + kind string + attempts int +} + +// deliverPending sends everything that is due and records the outcome. +func deliverPending(ctx context.Context, db *sql.DB, cfg NotifyConfig) { + batch, err := pendingNotifications(ctx, db) + if err != nil { + log.Printf("notifier: find pending: %v", err) + return + } + + sent := 0 + for _, n := range batch { + if err := deliver(ctx, db, cfg, n); err != nil { + log.Printf("notifier: deliver %d (incident %d): %v", n.id, n.incidentID, err) + markFailed(ctx, db, n, err) + continue + } + if _, err := db.ExecContext(ctx, + "UPDATE notifications SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?", + time.Now().Unix(), n.id); err != nil { + log.Printf("notifier: mark sent %d: %v", n.id, err) + } + sent++ + } + if sent > 0 { + log.Printf("notifier: delivered %d notification(s)", sent) + } +} + +// pendingNotifications reads the due rows and closes the cursor before the +// caller writes, for the same single-connection reason as staleAlertIDs. +func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error) { + rows, err := db.QueryContext(ctx, ` + SELECT id, incident_id, user_id, topic, kind, attempts + FROM notifications + WHERE sent_at IS NULL + AND send_after <= ? + AND attempts < ? + ORDER BY id + LIMIT ?`, time.Now().Unix(), notifyMaxAttempts, notifyBatch) + if err != nil { + return nil, err + } + defer rows.Close() + + var batch []outboxRow + for rows.Next() { + var n outboxRow + if err := rows.Scan(&n.id, &n.incidentID, &n.userID, &n.topic, &n.kind, &n.attempts); err != nil { + return nil, err + } + batch = append(batch, n) + } + return batch, rows.Err() +} + +// markFailed bumps the attempt count and pushes the row out to its next retry. +func markFailed(ctx context.Context, db *sql.DB, n outboxRow, cause error) { + next := time.Now().Add(retryDelay(n.attempts)).Unix() + if _, err := db.ExecContext(ctx, + "UPDATE notifications SET attempts = attempts + 1, send_after = ?, last_error = ? WHERE id = ?", + next, cause.Error(), n.id); err != nil { + log.Printf("notifier: mark failed %d: %v", n.id, err) + } +} + +// retryDelay doubles the wait per attempt, up to notifyRetryMax. +func retryDelay(attempts int) time.Duration { + d := notifyRetryBase << attempts + if d > notifyRetryMax || d <= 0 { + return notifyRetryMax + } + return d +} + +// deliver renders one notification against the incident's *current* state and +// publishes it. Rendering happens here rather than at enqueue time so a message +// that waited in the queue while its incident escalated goes out at the +// severity the incident has now. +func deliver(ctx context.Context, db *sql.DB, cfg NotifyConfig, n outboxRow) error { + inc, err := fetchIncident(ctx, db, n.incidentID) + if err != nil { + return fmt.Errorf("load incident: %w", err) + } + + var firing int + if err := db.QueryRowContext(ctx, ` + SELECT COUNT(*) + FROM incident_alerts ia + JOIN alerts a ON a.id = ia.alert_id + WHERE ia.incident_id = ? AND a.status = 'firing'`, n.incidentID).Scan(&firing); err != nil { + return fmt.Errorf("count firing: %w", err) + } + + msg := renderNotification(inc, n, firing, cfg) + + // An Acknowledge button needs both a user to attribute the acknowledgement + // to and a URL the phone can reach. Minted per delivery, so every push + // carries its own short-lived token rather than reusing one. + if n.kind != notifyResolved && n.userID != nil && cfg.PublicURL != "" { + raw, err := issueAckToken(ctx, db, n.incidentID, *n.userID) + if err != nil { + return fmt.Errorf("issue ack token: %w", err) + } + msg.Actions = append(msg.Actions, ntfyAction{ + Action: "http", + Label: "Acknowledge", + URL: strings.TrimSuffix(cfg.PublicURL, "/") + "/api/notify/ack/" + raw, + Method: "POST", + Clear: true, + }) + } + + return publish(ctx, cfg, msg) +} + +// ntfyMessage is ntfy's JSON publish format. Using it rather than the X-Actions +// header avoids that header's comma and quote escaping rules, which are easy to +// break with a title that happens to contain a comma. +type ntfyMessage struct { + Topic string `json:"topic"` + Title string `json:"title,omitempty"` + Message string `json:"message"` + Priority int `json:"priority,omitempty"` + Tags []string `json:"tags,omitempty"` + Click string `json:"click,omitempty"` + Actions []ntfyAction `json:"actions,omitempty"` +} + +type ntfyAction struct { + Action string `json:"action"` + Label string `json:"label"` + URL string `json:"url"` + Method string `json:"method,omitempty"` + Clear bool `json:"clear,omitempty"` +} + +// renderNotification builds the message body for one queued notification. +func renderNotification(inc models.Incident, n outboxRow, firing int, cfg NotifyConfig) ntfyMessage { + msg := ntfyMessage{Topic: n.topic} + + if cfg.PublicURL != "" { + msg.Click = fmt.Sprintf("%s/api/incidents/%d", + strings.TrimSuffix(cfg.PublicURL, "/"), inc.ID) + } + + switch n.kind { + case notifyResolved: + msg.Title = "Resolved: " + inc.Title + msg.Message = "All alerts stopped firing after " + + humanDuration(time.Since(inc.TriggeredAt)) + msg.Priority = ntfyPriorityLow + msg.Tags = []string{"white_check_mark"} + return msg + + case notifyReminder: + msg.Title = "Still unacknowledged: " + inc.Title + default: + msg.Title = inc.Title + } + + severity := derefString(inc.Severity) + + parts := []string{fmt.Sprintf("%d alert%s firing", firing, plural(firing))} + if severity != "" { + parts = append(parts, "severity "+severity) + } + if assignee := derefString(inc.AssignedToUser); assignee != "" { + parts = append(parts, "on call: "+assignee) + } + if n.kind == notifyReminder { + parts = append(parts, "open "+humanDuration(time.Since(inc.TriggeredAt))) + } + + msg.Message = strings.Join(parts, " · ") + msg.Priority = ntfyPriority(severity) + msg.Tags = []string{severityTag(severity)} + return msg +} + +// ntfy's priority scale. Max is the one that overrides the phone's quiet +// settings, which is the whole point of paging on critical. +const ( + ntfyPriorityLow = 2 + ntfyPriorityDefault = 3 + ntfyPriorityHigh = 4 + ntfyPriorityMax = 5 +) + +// ntfyPriority maps an incident's severity onto ntfy's scale, following the +// same ordering severityRank uses. An unrecognised severity gets the default +// rather than being silenced. +func ntfyPriority(severity string) int { + switch severityRank(severity) { + case 4: + return ntfyPriorityMax + case 3: + return ntfyPriorityHigh + case 2: + return ntfyPriorityDefault + case 1: + return ntfyPriorityLow + default: + return ntfyPriorityDefault + } +} + +func severityTag(severity string) string { + switch severityRank(severity) { + case 4: + return "rotating_light" + case 3: + return "red_circle" + case 2: + return "warning" + case 1: + return "information_source" + default: + return "bell" + } +} + +// publish POSTs one message to ntfy. +func publish(ctx context.Context, cfg NotifyConfig, msg ntfyMessage) error { + body, err := json.Marshal(msg) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + strings.TrimSuffix(cfg.BaseURL, "/"), bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if cfg.Token != "" { + req.Header.Set("Authorization", "Bearer "+cfg.Token) + } + + resp, err := notifyClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("ntfy returned %s", resp.Status) + } + return nil +} + +// enqueueNotification adds one row to the outbox, due immediately. +func enqueueNotification(ctx context.Context, q querier, incidentID int64, userID *int64, topic, kind string) error { + now := time.Now().Unix() + _, err := q.ExecContext(ctx, ` + INSERT INTO notifications (incident_id, user_id, topic, kind, created_at, send_after) + VALUES (?, ?, ?, ?, ?, ?)`, incidentID, userID, topic, kind, now, now) + return err +} + +// notifyTarget decides where a newly opened incident's notification goes. +// +// The on-call user's own topic when they have one, otherwise the fallback +// topic with no user attached. Deliberately not "the fallback topic, attributed +// to the on-call user": the fallback is shared, and an Acknowledge button on a +// shared topic would let any subscriber acknowledge as somebody else. +func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int64) (topic string, userID *int64) { + if onCall != nil { + var t *string + err := q.QueryRowContext(ctx, + "SELECT ntfy_topic FROM users WHERE id = ?", *onCall).Scan(&t) + if err == nil && t != nil && *t != "" { + return *t, onCall + } + } + return cfg.FallbackTopic, nil +} + +// enqueueOpened queues the notification for a freshly opened incident. It is the +// only enqueue point that has to resolve a topic from scratch; every later +// notification for the incident reuses what this one chose. +func enqueueOpened(ctx context.Context, q querier, cfg NotifyConfig, incidentID int64, onCall *int64) error { + if !cfg.enabled() { + return nil + } + topic, userID := notifyTarget(ctx, q, cfg, onCall) + if topic == "" { + // Nobody on call has a topic and there is no fallback: there is nowhere + // to send this, and queueing it would only accumulate undeliverable rows. + return nil + } + return enqueueNotification(ctx, q, incidentID, userID, topic, notifyTriggered) +} + +// enqueueResolved queues the all-clear, reusing the topic the incident's last +// notification went to. That needs no configuration to reach this function, and +// it gives the right rule for free: you only hear that something resolved if you +// were told it started. +func enqueueResolved(ctx context.Context, q querier, incidentID int64) error { + var topic string + var userID *int64 + err := q.QueryRowContext(ctx, ` + SELECT topic, user_id FROM notifications + WHERE incident_id = ? ORDER BY id DESC LIMIT 1`, incidentID).Scan(&topic, &userID) + if err == sql.ErrNoRows { + return nil + } + if err != nil { + return err + } + return enqueueNotification(ctx, q, incidentID, userID, topic, notifyResolved) +} + +// humanDuration renders an age the way a person reads it at 3am: coarse, and +// never more than two units. +func humanDuration(d time.Duration) string { + if d < time.Minute { + return "less than a minute" + } + if d < time.Hour { + return fmt.Sprintf("%dm", int(d.Minutes())) + } + h := int(d.Hours()) + m := int(d.Minutes()) - h*60 + if m == 0 { + return fmt.Sprintf("%dh", h) + } + return fmt.Sprintf("%dh%dm", h, m) +} + +func plural(n int) string { + if n == 1 { + return "" + } + return "s" +} + +// derefString reads a nullable text column as a plain string. +func derefString(s *string) string { + if s == nil { + return "" + } + return *s +} diff --git a/internal/api/notify_ack.go b/internal/api/notify_ack.go new file mode 100644 index 0000000..f624138 --- /dev/null +++ b/internal/api/notify_ack.go @@ -0,0 +1,98 @@ +package api + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "log" + "net/http" + "time" + + "github.com/go-chi/chi/v5" +) + +// issueAckToken mints the secret behind one notification's Acknowledge button +// and returns the raw value to embed in its URL. Only the hash is stored, the +// same way api_keys works. +// +// A fresh token per delivery rather than one per incident: the raw value only +// exists for as long as it takes to build the message, so there is nothing to +// look up and reuse later, and a reminder that supersedes an earlier page +// carries its own credential. +func issueAckToken(ctx context.Context, q querier, incidentID, userID int64) (string, error) { + raw, hash, err := randomToken() + if err != nil { + return "", err + } + now := time.Now() + if _, err := q.ExecContext(ctx, ` + INSERT INTO incident_ack_tokens (token_hash, incident_id, user_id, created_at, expires_at) + VALUES (?, ?, ?, ?, ?)`, + hash, incidentID, userID, now.Unix(), now.Add(ackTokenTTL).Unix()); err != nil { + return "", err + } + return raw, nil +} + +// handleNotifyAck acknowledges an incident from the Acknowledge button in a +// push notification. +// +// It is deliberately outside AuthMiddleware: the caller is a phone acting on a +// notification, not a client holding an API key. What stands in for the key is +// the token in the path — 256 bits of entropy, valid for one incident, one +// action, and one day. It must stay publicly reachable for the button to work +// when the responder is off the cluster network. +func handleNotifyAck(db *sql.DB) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + h := sha256.Sum256([]byte(chi.URLParam(r, "token"))) + hash := hex.EncodeToString(h[:]) + + var incidentID, userID int64 + err := db.QueryRowContext(r.Context(), ` + SELECT incident_id, user_id FROM incident_ack_tokens + WHERE token_hash = ? AND expires_at > ?`, + hash, time.Now().Unix()).Scan(&incidentID, &userID) + if err != nil { + // Unknown and expired get the same answer, so the endpoint cannot be + // used to probe which tokens once existed. + respond(w, http.StatusNotFound, errResp("invalid or expired token")) + return + } + + acked, err := acknowledgeIncident(r.Context(), db, incidentID, userID) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + if !acked { + // The incident closed between the page and the tap. Nothing to do, + // and nothing the responder did wrong — report the state, not an error, + // so ntfy shows a success toast rather than a failure. + respond(w, http.StatusOK, map[string]any{ + "incident_id": incidentID, + "status": "resolved", + }) + return + } + respond(w, http.StatusOK, map[string]any{ + "incident_id": incidentID, + "status": "acknowledged", + }) + } +} + +// purgeAckTokens drops tokens whose notifications are long past. Nothing else +// deletes them: incidents are archived rather than removed, so the cascade never +// fires in practice. +func purgeAckTokens(ctx context.Context, db *sql.DB) { + res, err := db.ExecContext(ctx, + "DELETE FROM incident_ack_tokens WHERE expires_at < ?", time.Now().Unix()) + if err != nil { + log.Printf("sweeper: purge ack tokens: %v", err) + return + } + if n, _ := res.RowsAffected(); n > 0 { + log.Printf("sweeper: purged %d expired ack token(s)", n) + } +} diff --git a/internal/api/notify_test.go b/internal/api/notify_test.go new file mode 100644 index 0000000..cf7ab02 --- /dev/null +++ b/internal/api/notify_test.go @@ -0,0 +1,580 @@ +package api_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/yeniklas/terdut-server/internal/api" +) + +// --------------------------------------------------------------------------- +// Fake ntfy +// --------------------------------------------------------------------------- + +// pushed is one message the fake ntfy received, in ntfy's JSON publish shape. +type pushed struct { + Topic string `json:"topic"` + Title string `json:"title"` + Message string `json:"message"` + Priority int `json:"priority"` + Tags []string `json:"tags"` + Click string `json:"click"` + Actions []struct { + Action string `json:"action"` + Label string `json:"label"` + URL string `json:"url"` + Method string `json:"method"` + Clear bool `json:"clear"` + } `json:"actions"` +} + +// fakeNtfy records what the notifier published. status controls the reply, so a +// test can make delivery fail. +type fakeNtfy struct { + *httptest.Server + mu sync.Mutex + got []pushed + status int +} + +func newFakeNtfy(t *testing.T) *fakeNtfy { + t.Helper() + f := &fakeNtfy{status: http.StatusOK} + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var msg pushed + if err := json.NewDecoder(r.Body).Decode(&msg); err != nil { + http.Error(w, "bad json", http.StatusBadRequest) + return + } + f.mu.Lock() + f.got = append(f.got, msg) + status := f.status + f.mu.Unlock() + w.WriteHeader(status) + })) + t.Cleanup(f.Close) + return f +} + +func (f *fakeNtfy) messages() []pushed { + f.mu.Lock() + defer f.mu.Unlock() + return append([]pushed(nil), f.got...) +} + +func (f *fakeNtfy) failWith(status int) { + f.mu.Lock() + defer f.mu.Unlock() + f.status = status +} + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +// notifyTS builds a server with notifications enabled, the admin on call today, +// and a topic on the admin — the setup every delivery test needs. +func notifyTS(t *testing.T, cfg api.NotifyConfig) (*ts, *fakeNtfy) { + t.Helper() + f := newFakeNtfy(t) + cfg.BaseURL = f.URL + s := newTS(t, cfg) + + putOnCall(t, s, 1) + setTopic(t, s, 1, "terdut-admin") + return s, f +} + +func putOnCall(t *testing.T, s *ts, userID int) { + t.Helper() + today := time.Now().UTC().Format("2006-01-02") + resp := s.req(t, http.MethodPost, "/api/schedule", + map[string]any{"user_id": userID, "dates": []string{today}}) + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + t.Fatalf("schedule assignment returned %d", resp.StatusCode) + } +} + +func setTopic(t *testing.T, s *ts, userID int, topic string) { + t.Helper() + resp := s.req(t, http.MethodPut, + fmt.Sprintf("/api/users/%d/notify", userID), map[string]any{"ntfy_topic": topic}) + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("set notify topic returned %d", resp.StatusCode) + } +} + +func (s *ts) sweepNotify(t *testing.T) { + t.Helper() + api.NotifySweep(context.Background(), s.db, s.notify) +} + +// countNotifications reports how many outbox rows exist, optionally of one kind. +func (s *ts) countNotifications(t *testing.T, kind string) int { + t.Helper() + var n int + query := "SELECT COUNT(*) FROM notifications" + args := []any{} + if kind != "" { + query += " WHERE kind = ?" + args = append(args, kind) + } + if err := s.db.QueryRow(query, args...).Scan(&n); err != nil { + t.Fatalf("count notifications: %v", err) + } + return n +} + +// fireCritical posts a single critical alert, which opens one incident. +func fireCritical(t *testing.T, s *ts) { + t.Helper() + postWebhook(t, s, []map[string]any{ + amAlert("fp-notify", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, + map[string]string{"severity": "critical"}), + }, "{}:{alertname=\"DiskFull\"}") +} + +// --------------------------------------------------------------------------- +// Delivery +// --------------------------------------------------------------------------- + +func TestNotify_TriggeredIncidentPagesOnCall(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + + if got := s.countNotifications(t, "triggered"); got != 1 { + t.Fatalf("expected 1 queued notification, got %d", got) + } + s.sweepNotify(t) + + msgs := f.messages() + if len(msgs) != 1 { + t.Fatalf("expected 1 push, got %d", len(msgs)) + } + m := msgs[0] + + if m.Topic != "terdut-admin" { + t.Errorf("expected the on-call user's topic, got %q", m.Topic) + } + if m.Priority != 5 { + t.Errorf("expected max priority for a critical incident, got %d", m.Priority) + } + if !strings.Contains(m.Title, "DiskFull") { + t.Errorf("expected the incident title in %q", m.Title) + } + if !strings.Contains(m.Message, "severity critical") { + t.Errorf("expected the severity in %q", m.Message) + } + if m.Click != "https://terdut.example.com/api/incidents/1" { + t.Errorf("unexpected click target %q", m.Click) + } + if len(m.Actions) != 1 || m.Actions[0].Label != "Acknowledge" { + t.Fatalf("expected an Acknowledge action, got %+v", m.Actions) + } + if m.Actions[0].Method != http.MethodPost { + t.Errorf("expected the action to POST, got %q", m.Actions[0].Method) + } +} + +// A delivered row must not be delivered again on the next pass. +func TestNotify_DeliveredOnlyOnce(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + s.sweepNotify(t) + + if got := len(f.messages()); got != 1 { + t.Errorf("expected 1 push across two passes, got %d", got) + } +} + +// --------------------------------------------------------------------------- +// Acknowledging from the notification +// --------------------------------------------------------------------------- + +func TestNotify_AckButtonAcknowledgesIncident(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + + ackURL := f.messages()[0].Actions[0].URL + // The action URL is built for the public hostname; point it at the test + // server, which is the same handler. + path := ackURL[strings.Index(ackURL, "/api/notify/ack/"):] + + resp, err := http.Post(s.URL+path, "application/json", nil) + if err != nil { + t.Fatalf("ack: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected 200 from the ack button, got %d", resp.StatusCode) + } + + inc := getIncident(t, s, 1) + if inc["status"] != "acknowledged" { + t.Errorf("expected the incident acknowledged, got %v", inc["status"]) + } + if inc["acknowledged_by"] != "admin" { + t.Errorf("expected the ack attributed to the token's user, got %v", inc["acknowledged_by"]) + } + + // The timeline must record it the same way the authenticated route would. + events := timeline(t, s, 1) + if !contains(eventTypes(events), "acknowledged") { + t.Errorf("expected an acknowledged event, got %v", eventTypes(events)) + } + for _, e := range events { + if e["type"] == "acknowledged" && e["username"] != "admin" { + t.Errorf("expected the acknowledged event attributed to admin, got %v", e["username"]) + } + } +} + +func TestNotify_AckRejectsUnknownToken(t *testing.T) { + s, _ := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + fireCritical(t, s) + + resp, err := http.Post(s.URL+"/api/notify/ack/deadbeef", "application/json", nil) + if err != nil { + t.Fatalf("ack: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Errorf("expected 404 for an unknown token, got %d", resp.StatusCode) + } + if inc := getIncident(t, s, 1); inc["status"] != "triggered" { + t.Errorf("expected the incident untouched, got %v", inc["status"]) + } +} + +func TestNotify_AckRejectsExpiredToken(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + + ackURL := f.messages()[0].Actions[0].URL + path := ackURL[strings.Index(ackURL, "/api/notify/ack/"):] + + // Age the token past its TTL. The token's inputs are wall-clock timestamps, + // so this is the same trick the sweeper tests use. + s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix()) + + resp, err := http.Post(s.URL+path, "application/json", nil) + if err != nil { + t.Fatalf("ack: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Errorf("expected 404 for an expired token, got %d", resp.StatusCode) + } + if inc := getIncident(t, s, 1); inc["status"] != "triggered" { + t.Errorf("expected the incident untouched, got %v", inc["status"]) + } +} + +// The sweeper is what stops expired tokens accumulating forever. +func TestNotify_SweepPurgesExpiredAckTokens(t *testing.T) { + s, _ := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix()) + + api.Sweep(context.Background(), s.db, 168*time.Hour, 6*time.Hour) + + var n int + if err := s.db.QueryRow("SELECT COUNT(*) FROM incident_ack_tokens").Scan(&n); err != nil { + t.Fatalf("count tokens: %v", err) + } + if n != 0 { + t.Errorf("expected expired tokens purged, %d left", n) + } +} + +// --------------------------------------------------------------------------- +// Reminders +// --------------------------------------------------------------------------- + +// ageNotifications backdates every sent notification so the next pass sees the +// reminder as due. +func (s *ts) ageNotifications(t *testing.T, by time.Duration) { + t.Helper() + s.exec(t, "UPDATE notifications SET created_at = ? WHERE sent_at IS NOT NULL", + time.Now().Add(-by).Unix()) +} + +func TestNotify_UnacknowledgedIncidentIsRenotified(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{ + PublicURL: "https://terdut.example.com", + RepeatEvery: 15 * time.Minute, + }) + + fireCritical(t, s) + s.sweepNotify(t) + s.ageNotifications(t, 20*time.Minute) + s.sweepNotify(t) + + msgs := f.messages() + if len(msgs) != 2 { + t.Fatalf("expected a reminder push, got %d message(s)", len(msgs)) + } + if !strings.Contains(msgs[1].Title, "Still unacknowledged") { + t.Errorf("expected the reminder to say so, got %q", msgs[1].Title) + } + if msgs[1].Topic != "terdut-admin" { + t.Errorf("expected the reminder on the same topic, got %q", msgs[1].Topic) + } + // Each page carries its own credential. + if len(msgs[1].Actions) != 1 || msgs[1].Actions[0].URL == msgs[0].Actions[0].URL { + t.Errorf("expected the reminder to carry a fresh ack token") + } +} + +func TestNotify_AcknowledgedIncidentStopsReminders(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{ + PublicURL: "https://terdut.example.com", + RepeatEvery: 15 * time.Minute, + }) + + fireCritical(t, s) + s.sweepNotify(t) + + resp := s.req(t, http.MethodPost, "/api/incidents/1/acknowledge", nil) + resp.Body.Close() + + s.ageNotifications(t, 20*time.Minute) + s.sweepNotify(t) + + if got := len(f.messages()); got != 1 { + t.Errorf("expected no reminder once acknowledged, got %d message(s)", got) + } +} + +// Snooze is the deliberate "not now", and it is what mutes the pager. +func TestNotify_SnoozedIncidentStopsReminders(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{ + PublicURL: "https://terdut.example.com", + RepeatEvery: 15 * time.Minute, + }) + + fireCritical(t, s) + s.sweepNotify(t) + + resp := s.req(t, http.MethodPost, "/api/incidents/1/snooze", map[string]any{"duration": "1h"}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("snooze returned %d", resp.StatusCode) + } + resp.Body.Close() + + s.ageNotifications(t, 20*time.Minute) + s.sweepNotify(t) + + if got := len(f.messages()); got != 1 { + t.Errorf("expected no reminder while snoozed, got %d message(s)", got) + } +} + +func TestNotify_ZeroRepeatDisablesReminders(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + s.ageNotifications(t, 24*time.Hour) + s.sweepNotify(t) + + if got := len(f.messages()); got != 1 { + t.Errorf("expected reminders off, got %d message(s)", got) + } +} + +// --------------------------------------------------------------------------- +// Resolution +// --------------------------------------------------------------------------- + +func TestNotify_ResolvedIncidentSendsAllClear(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + + postWebhook(t, s, []map[string]any{ + amAlert("fp-notify", "DiskFull", "resolved", "2026-05-20T10:00:00Z", + "2026-05-20T11:00:00Z", map[string]string{"severity": "critical"}), + }, "{}:{alertname=\"DiskFull\"}") + s.sweepNotify(t) + + msgs := f.messages() + if len(msgs) != 2 { + t.Fatalf("expected an all-clear push, got %d message(s)", len(msgs)) + } + if !strings.HasPrefix(msgs[1].Title, "Resolved:") { + t.Errorf("expected a resolved title, got %q", msgs[1].Title) + } + if msgs[1].Priority != 2 { + t.Errorf("expected the all-clear at low priority, got %d", msgs[1].Priority) + } + // Nothing to acknowledge on a closed incident. + if len(msgs[1].Actions) != 0 { + t.Errorf("expected no actions on the all-clear, got %+v", msgs[1].Actions) + } +} + +// Closing an incident by hand sends nothing: the person who did it knows. +func TestNotify_ManualResolveSendsNothing(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + + fireCritical(t, s) + s.sweepNotify(t) + + resp := s.req(t, http.MethodPost, "/api/incidents/1/resolve", nil) + resp.Body.Close() + s.sweepNotify(t) + + if got := len(f.messages()); got != 1 { + t.Errorf("expected no push for a manual resolve, got %d message(s)", got) + } +} + +// --------------------------------------------------------------------------- +// Routing and configuration +// --------------------------------------------------------------------------- + +// With nobody on call the page goes to the shared fallback, and carries no +// Acknowledge button — there is no user to attribute the acknowledgement to. +func TestNotify_FallbackTopicHasNoAckButton(t *testing.T) { + f := newFakeNtfy(t) + s := newTS(t, api.NotifyConfig{ + BaseURL: f.URL, + FallbackTopic: "terdut-oncall", + PublicURL: "https://terdut.example.com", + }) + + fireCritical(t, s) + s.sweepNotify(t) + + msgs := f.messages() + if len(msgs) != 1 { + t.Fatalf("expected 1 push, got %d", len(msgs)) + } + if msgs[0].Topic != "terdut-oncall" { + t.Errorf("expected the fallback topic, got %q", msgs[0].Topic) + } + if len(msgs[0].Actions) != 0 { + t.Errorf("expected no ack button on a shared topic, got %+v", msgs[0].Actions) + } +} + +// Nobody on call and no fallback means there is nowhere to send: queueing would +// only pile up rows that can never be delivered. +func TestNotify_NoTargetQueuesNothing(t *testing.T) { + f := newFakeNtfy(t) + s := newTS(t, api.NotifyConfig{BaseURL: f.URL}) + + fireCritical(t, s) + + if got := s.countNotifications(t, ""); got != 0 { + t.Errorf("expected nothing queued without a target, got %d", got) + } + s.sweepNotify(t) + if got := len(f.messages()); got != 0 { + t.Errorf("expected no push, got %d", got) + } +} + +// The zero NotifyConfig is what every pre-existing test runs under. +func TestNotify_DisabledQueuesNothing(t *testing.T) { + s := newTS(t) + putOnCall(t, s, 1) + setTopic(t, s, 1, "terdut-admin") + + fireCritical(t, s) + + if got := s.countNotifications(t, ""); got != 0 { + t.Errorf("expected nothing queued with notifications off, got %d", got) + } +} + +// --------------------------------------------------------------------------- +// Retries +// --------------------------------------------------------------------------- + +func TestNotify_FailedDeliveryRetriesWithBackoff(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"}) + f.failWith(http.StatusInternalServerError) + + fireCritical(t, s) + s.sweepNotify(t) + + var attempts int + var sentAt *int64 + var sendAfter int64 + var lastError *string + if err := s.db.QueryRow( + "SELECT attempts, sent_at, send_after, last_error FROM notifications WHERE id = 1"). + Scan(&attempts, &sentAt, &sendAfter, &lastError); err != nil { + t.Fatalf("read notification: %v", err) + } + if attempts != 1 { + t.Errorf("expected 1 attempt recorded, got %d", attempts) + } + if sentAt != nil { + t.Errorf("expected the row unsent, got sent_at %v", *sentAt) + } + if sendAfter <= time.Now().Unix() { + t.Errorf("expected the retry pushed into the future, got %d", sendAfter) + } + if lastError == nil || !strings.Contains(*lastError, "500") { + t.Errorf("expected the failure recorded, got %v", lastError) + } + + // Backing off means the next pass leaves it alone until it is due. + s.sweepNotify(t) + if got := len(f.messages()); got != 1 { + t.Errorf("expected no immediate retry, got %d attempt(s)", got) + } + + // Once due and once ntfy recovers, it goes out. + f.failWith(http.StatusOK) + s.exec(t, "UPDATE notifications SET send_after = ? WHERE id = 1", time.Now().Add(-time.Second).Unix()) + s.sweepNotify(t) + + if err := s.db.QueryRow("SELECT sent_at FROM notifications WHERE id = 1").Scan(&sentAt); err != nil { + t.Fatalf("read notification: %v", err) + } + if sentAt == nil { + t.Error("expected the retry to succeed once ntfy recovered") + } +} + +// An ntfy outage must not produce a reminder backlog that all lands at once +// when it comes back: the previous page has to have been sent first. +func TestNotify_UnsentNotificationBlocksReminders(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{ + PublicURL: "https://terdut.example.com", + RepeatEvery: 15 * time.Minute, + }) + f.failWith(http.StatusInternalServerError) + + fireCritical(t, s) + s.sweepNotify(t) + s.exec(t, "UPDATE notifications SET created_at = ?", time.Now().Add(-time.Hour).Unix()) + s.sweepNotify(t) + + if got := s.countNotifications(t, "reminder"); got != 0 { + t.Errorf("expected no reminders queued behind an undelivered page, got %d", got) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index 7aa658e..d2783ff 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -8,7 +8,10 @@ import ( "github.com/go-chi/chi/v5/middleware" ) -func NewRouter(db *sql.DB) http.Handler { +// NewRouter builds the HTTP surface. notify is passed through to the webhook, +// the only handler that has to decide where a new incident's page goes; a zero +// value disables notifications. +func NewRouter(db *sql.DB, notify NotifyConfig) http.Handler { r := chi.NewRouter() r.Use(middleware.Logger) r.Use(middleware.Recoverer) @@ -17,9 +20,13 @@ func NewRouter(db *sql.DB) http.Handler { respond(w, http.StatusOK, map[string]string{"status": "ok"}) }) - // Unauthenticated: bootstrap and Alertmanager webhook receiver. + // Unauthenticated: bootstrap, the Alertmanager webhook receiver, and the + // Acknowledge button in a push notification. The last one is authorised by + // the single-use token in its path rather than an API key, and has to stay + // reachable from outside the cluster for the button to work. r.Post("/api/bootstrap", handleBootstrap(db)) - r.Post("/api/alertmanager/webhook", handleAlertmanagerWebhook(db)) + r.Post("/api/alertmanager/webhook", handleAlertmanagerWebhook(db, notify)) + r.Post("/api/notify/ack/{token}", handleNotifyAck(db)) // All other /api routes require a valid API key. r.Group(func(r chi.Router) { @@ -28,6 +35,7 @@ func NewRouter(db *sql.DB) http.Handler { r.Get("/api/users", handleListUsers(db)) r.Post("/api/users", handleCreateUser(db)) r.Delete("/api/users/{id}", handleDeleteUser(db)) + r.Put("/api/users/{id}/notify", handleSetNotifyTarget(db)) r.Post("/api/users/{id}/api-keys", handleCreateAPIKey(db)) r.Delete("/api/users/{id}/api-keys/{keyID}", handleDeleteAPIKey(db)) diff --git a/internal/api/users.go b/internal/api/users.go index 5117ee2..5d9dee5 100644 --- a/internal/api/users.go +++ b/internal/api/users.go @@ -48,7 +48,7 @@ func handleBootstrap(db *sql.DB) http.HandlerFunc { } userID, _ := res.LastInsertId() - raw, hash, err := newAPIKey() + raw, hash, err := randomToken() if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -70,7 +70,7 @@ func handleBootstrap(db *sql.DB) http.HandlerFunc { func handleListUsers(db *sql.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { rows, err := db.QueryContext(r.Context(), - "SELECT id, username, email, created_at FROM users ORDER BY id") + "SELECT id, username, email, created_at, ntfy_topic FROM users ORDER BY id") if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -81,7 +81,7 @@ func handleListUsers(db *sql.DB) http.HandlerFunc { for rows.Next() { var u models.User var ts int64 - if err := rows.Scan(&u.ID, &u.Username, &u.Email, &ts); err != nil { + if err := rows.Scan(&u.ID, &u.Username, &u.Email, &ts, &u.NtfyTopic); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } @@ -123,6 +123,50 @@ func handleCreateUser(db *sql.DB) http.HandlerFunc { } } +// handleSetNotifyTarget points a user's push notifications at an ntfy topic, or +// clears it with an empty string. The topic is a shared secret with the ntfy +// server — anyone who knows it can publish to it — so pick an unguessable one +// unless your ntfy enforces access control. +func handleSetNotifyTarget(db *sql.DB) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + respond(w, http.StatusBadRequest, errResp("invalid user id")) + return + } + var req struct { + NtfyTopic string `json:"ntfy_topic"` + } + if err := decodeJSON(r, &req); err != nil { + respond(w, http.StatusBadRequest, errResp("invalid request body")) + return + } + + var topic *string + if t := strings.TrimSpace(req.NtfyTopic); t != "" { + topic = &t + } + + res, err := db.ExecContext(r.Context(), + "UPDATE users SET ntfy_topic = ? WHERE id = ?", topic, id) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + if n, _ := res.RowsAffected(); n == 0 { + respond(w, http.StatusNotFound, errResp("user not found")) + return + } + + user, err := fetchUser(r.Context(), db, id) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + respond(w, http.StatusOK, user) + } +} + func handleDeleteUser(db *sql.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) @@ -170,7 +214,7 @@ func handleCreateAPIKey(db *sql.DB) http.HandlerFunc { return } - raw, hash, err := newAPIKey() + raw, hash, err := randomToken() if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -215,8 +259,9 @@ func handleDeleteAPIKey(db *sql.DB) http.HandlerFunc { } } -// newAPIKey generates a random 32-byte key encoded as hex, plus its SHA-256 hash for storage. -func newAPIKey() (raw, hash string, err error) { +// randomToken generates a random 32-byte secret encoded as hex, plus its SHA-256 +// hash for storage. Used for API keys and for notification acknowledge tokens. +func randomToken() (raw, hash string, err error) { b := make([]byte, 32) if _, err = rand.Read(b); err != nil { return @@ -230,8 +275,9 @@ func newAPIKey() (raw, hash string, err error) { func fetchUser(ctx context.Context, db *sql.DB, id int64) (models.User, error) { var u models.User var ts int64 - err := db.QueryRowContext(ctx, "SELECT id, username, email, created_at FROM users WHERE id = ?", id). - Scan(&u.ID, &u.Username, &u.Email, &ts) + err := db.QueryRowContext(ctx, + "SELECT id, username, email, created_at, ntfy_topic FROM users WHERE id = ?", id). + Scan(&u.ID, &u.Username, &u.Email, &ts, &u.NtfyTopic) if err != nil { return u, err } diff --git a/internal/config/config.go b/internal/config/config.go index f88297f..954d0a8 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -14,6 +14,25 @@ type Config struct { // before the sweeper treats it as resolved. It must exceed Alertmanager's // repeat_interval (default 4h), which is what refreshes the alert. StaleAfter time.Duration + + // NtfyURL is the ntfy server push notifications are published to. Empty + // disables notifications entirely. + NtfyURL string + + // NtfyToken is an optional bearer token for an access-controlled ntfy. + NtfyToken string + + // NtfyFallbackTopic receives incidents that open with nobody on call. + NtfyFallbackTopic string + + // PublicURL is the base URL a phone uses to reach this server, used for the + // link and the Acknowledge button inside a notification. Without it + // notifications carry neither. + PublicURL string + + // NotifyRepeat is how long an incident may sit unacknowledged before it is + // notified again. Zero disables reminders. + NotifyRepeat time.Duration } func Load() Config { @@ -25,17 +44,28 @@ func Load() Config { if dbPath == "" { dbPath = "terdut.db" } - archiveAfter := 7 * 24 * time.Hour - if s := os.Getenv("TERDUT_ARCHIVE_AFTER"); s != "" { - if d, err := time.ParseDuration(s); err == nil { - archiveAfter = d - } + return Config{ + Addr: addr, + DBPath: dbPath, + ArchiveAfter: duration("TERDUT_ARCHIVE_AFTER", 7*24*time.Hour), + StaleAfter: duration("TERDUT_STALE_AFTER", 6*time.Hour), + + NtfyURL: os.Getenv("TERDUT_NTFY_URL"), + NtfyToken: os.Getenv("TERDUT_NTFY_TOKEN"), + NtfyFallbackTopic: os.Getenv("TERDUT_NTFY_FALLBACK_TOPIC"), + PublicURL: os.Getenv("TERDUT_PUBLIC_URL"), + NotifyRepeat: duration("TERDUT_NOTIFY_REPEAT", 15*time.Minute), } - staleAfter := 6 * time.Hour - if s := os.Getenv("TERDUT_STALE_AFTER"); s != "" { - if d, err := time.ParseDuration(s); err == nil { - staleAfter = d - } - } - return Config{Addr: addr, DBPath: dbPath, ArchiveAfter: archiveAfter, StaleAfter: staleAfter} +} + +// duration reads a time.ParseDuration-formatted env var. An unset or +// unparseable value falls back to def rather than failing startup: a typo in one +// tuning knob should not take the server down. +func duration(env string, def time.Duration) time.Duration { + if s := os.Getenv(env); s != "" { + if d, err := time.ParseDuration(s); err == nil { + return d + } + } + return def } diff --git a/internal/db/migrations/009_notifications.sql b/internal/db/migrations/009_notifications.sql new file mode 100644 index 0000000..22e97ce --- /dev/null +++ b/internal/db/migrations/009_notifications.sql @@ -0,0 +1,44 @@ +-- Adds push notification delivery, so an incident reaches the person on call +-- instead of waiting to be discovered. +-- +-- Delivery is an outbox rather than an inline HTTP call: the pool is limited to +-- a single connection, so a POST made while holding the webhook's transaction +-- would stall every other request behind it. The webhook inserts a row; the +-- notifier goroutine delivers it. + +CREATE TABLE notifications ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + -- Nullable: a notification sent to the fallback topic belongs to nobody, + -- because nobody was on call when the incident opened. + user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, + topic TEXT NOT NULL, -- resolved at enqueue: who was on call then + kind TEXT NOT NULL CHECK(kind IN ('triggered', 'reminder', 'resolved')), + created_at INTEGER NOT NULL, + send_after INTEGER NOT NULL, -- retry backoff watermark + attempts INTEGER NOT NULL DEFAULT 0, + sent_at INTEGER, + last_error TEXT -- kept after the last attempt, for debugging +); + +-- The delivery loop's only query: what is due and still unsent. +CREATE INDEX notifications_pending_idx ON notifications(send_after) WHERE sent_at IS NULL; +-- Reminders and resolved notices both look up an incident's newest row. +CREATE INDEX notifications_incident_idx ON notifications(incident_id, id DESC); + +-- A notification body is stored on the ntfy server and cached on the device, so +-- a real API key must never appear in one. Each delivery mints its own token +-- instead: one incident, one action, one day. +CREATE TABLE incident_ack_tokens ( + token_hash TEXT PRIMARY KEY, -- SHA-256 of the raw token, as with api_keys + incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL +); + +CREATE INDEX incident_ack_tokens_expires_idx ON incident_ack_tokens(expires_at); + +-- Where this user's notifications go. NULL means they get none; incidents +-- assigned to them fall back to the configured fallback topic. +ALTER TABLE users ADD COLUMN ntfy_topic TEXT; diff --git a/internal/models/user.go b/internal/models/user.go index 421306a..39428e1 100644 --- a/internal/models/user.go +++ b/internal/models/user.go @@ -7,6 +7,11 @@ type User struct { Username string `json:"username"` Email string `json:"email"` CreatedAt time.Time `json:"created_at"` + + // NtfyTopic is where this user's push notifications go. Nil means they get + // none of their own; incidents assigned to them fall back to the configured + // fallback topic instead. + NtfyTopic *string `json:"ntfy_topic,omitempty"` } type APIKey struct {