Page the on-call person when an incident opens
An incident opened, got assigned to whoever held today's schedule entry,
and then sat there silently until somebody thought to look. The schedule
and the incident model were both built; nothing reached the person
holding the pager.
Notifications go out through ntfy, over plain HTTP with no new
dependencies. Delivery is an outbox rather than an inline 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 and a notifier goroutine sends it within a tick,
retrying with exponential backoff.
Only opening an incident has to resolve a topic from scratch. Reminders
and all-clears reuse whatever that first notification chose, which keeps
configuration out of resolveIfSettled and gives the right rule for free:
you only hear that something resolved if you were told it started.
Each push carries an Acknowledge button, because the useful thing to do
at 3am is stop the pager without unlocking anything. It POSTs to an
unauthenticated /api/notify/ack/{token} — a notification body lives on
the ntfy server and in the device cache, so a real API key must never
appear in one. The token is minted per delivery, scoped to one incident
and one action, and expires in a day.
Reminders repeat until the incident stops being untouched. The stop
conditions are the states that already mean somebody has it: acknowledged,
snoozed, resolved, archived. Snooze is the mute button, so there is no
separate reminder cap.
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.
This commit is contained in:
Niklas Ye
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user