package api import ( "context" "database/sql" "fmt" "log" "regexp" "strings" "time" "git.ryuvia.com/niklas/terdut-server/internal/models" "git.ryuvia.com/niklas/terdut-server/internal/notify" ) 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. A provider // 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 ) // notifierLockKey is the Postgres advisory lock the notifier takes for the // duration of each pass, so that running more than one replica does not // deliver (or double-deliver) the same notification from more than one of // them at once. Its value has no meaning beyond being distinct from // archiverLockKey. const notifierLockKey int64 = 7265_0002 // Notification kinds, recording why a push was sent. const ( notifyTriggered = notify.KindTriggered notifyReminder = notify.KindReminder notifyResolved = notify.KindResolved // notifyEscalated is a page that went out because nobody answered the last // one. Told apart from a reminder because it goes to somebody else. notifyEscalated = notify.KindEscalated ) // Timeline event types the notifier writes, so an incident's history says who // was paged and whether the page landed. Written from the delivery result // rather than at enqueue: a queued notification is an intention, and claiming // somebody was told before the provider accepted it would be a lie the timeline // keeps. // // The channel's address is deliberately absent from both; only its type is // named. An address can be a shared secret (an ntfy topic lets anyone holding it // publish to it) and the timeline is readable by every API key. const ( eventNotified = "notified" eventNotifyFailed = "notify_failed" ) // NotifyConfig is everything the notifier needs to reach people and to build // URLs they can follow back to this server. type NotifyConfig struct { // Providers are the ways this server can deliver a notification. Empty // disables notifications entirely: no goroutine, and nothing is ever enqueued. Providers *notify.Registry // Fallback receives incidents that open with nobody on call who can be // reached. Those notifications carry no Acknowledge button — there is no user // to attribute the acknowledgement to, and putting one on a shared target // would let any recipient acknowledge as somebody else. Fallback notify.Target // 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.Providers.Empty() } // StartNotifier delivers queued notifications until ctx is cancelled, starting // with an immediate pass so a restart flushes whatever the last one left behind. // // Each pass runs under notifierLockKey (see withAdvisoryLock), so that on more // than one replica only whichever instance's tick takes the lock first actually // delivers; the rest skip that tick rather than racing the same pass. func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) { if !cfg.enabled() { log.Print("notifier: disabled (no notification provider configured)") return } var types []string for _, p := range cfg.Providers.List() { types = append(types, p.Type()) } log.Printf("notifier: delivering through %s", strings.Join(types, ", ")) ticker := time.NewTicker(notifyInterval) defer ticker.Stop() sweep := func() { withAdvisoryLock(ctx, db, notifierLockKey, "notifier", func() { NotifySweep(ctx, db, cfg) }) } sweep() for { select { case <-ticker.C: sweep() 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) // Escalation before delivery, so a level that comes due on this tick is // paged on this tick rather than waiting for the next one. escalate(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 a provider outage produces a retry backlog rather than a reminder // backlog that all lands at once when it comes back. "Previous notification" // is the incident's latest batch: everything one fan-out queued shares a // created_at, and the batch counts as sent once any of its rows has landed. // // A channel the user has since disabled or removed is not reminded: that is the // off switch for a channel, and a reminder that ignored it would defeat it. func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) { // cfg.RepeatEvery is what the server started with; the settings table is // what it runs on. Read per tick, so an administrator lengthening the // interval at 02:00 is obeyed at 02:00 and not at the next restart. repeat := NewSettings(db).Duration(ctx, SettingNotifyRepeat, cfg.RepeatEvery) if repeat <= 0 { return } now := time.Now() rows, err := db.QueryContext(ctx, ` SELECT n.incident_id, n.user_id, n.channel_id, n.channel_type, n.address FROM notifications n JOIN incidents i ON i.id = n.incident_id LEFT JOIN notification_channels c ON c.id = n.channel_id WHERE n.created_at = (SELECT MAX(created_at) FROM notifications WHERE incident_id = n.incident_id) AND EXISTS (SELECT 1 FROM notifications s WHERE s.incident_id = n.incident_id AND s.created_at = n.created_at AND s.sent_at IS NOT NULL) AND n.created_at <= $1 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 <= $2) -- A shared fallback target has no user and no channel row; everything -- else must still point at an enabled channel. AND (n.user_id IS NULL OR c.enabled) -- A team with an escalation ladder gets escalation instead. Both -- would mean two pages for one silence, which is how people learn to -- mute a tool. AND NOT EXISTS ( SELECT 1 FROM escalation_levels el WHERE el.team_id = i.team_id) ORDER BY n.id`, now.Add(-repeat).Unix(), now.Unix()) if err != nil { log.Printf("notifier: find reminders: %v", err) return } // Collected before inserting, rather than written while walking the cursor: // the inserts below are what this query selects on, and a cursor reading its // own writes is a hazard whatever the pool size. byIncident := map[int64][]recipient{} var order []int64 for rows.Next() { var incidentID int64 var r recipient if err := rows.Scan(&incidentID, &r.userID, &r.channelID, &r.typ, &r.address); err != nil { rows.Close() log.Printf("notifier: scan reminder: %v", err) return } if _, seen := byIncident[incidentID]; !seen { order = append(order, incidentID) } byIncident[incidentID] = append(byIncident[incidentID], r) } if err := rows.Err(); err != nil { rows.Close() log.Printf("notifier: find reminders: %v", err) return } rows.Close() // One batch per incident, so the reminder is itself a batch the next one // can be measured from. for _, id := range order { if err := enqueueNotifications(ctx, db, id, byIncident[id], notifyReminder); err != nil { log.Printf("notifier: queue reminder for incident %d: %v", id, err) } } if len(order) > 0 { log.Printf("notifier: queued reminders for %d incident(s)", len(order)) } } // outboxRow is one queued notification, read before any HTTP happens. type outboxRow struct { id int64 incidentID int64 userID *int64 channel string // channel type; the provider that delivers it address 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 = $1, attempts = attempts + 1, last_error = NULL WHERE id = $2", time.Now().Unix(), n.id); err != nil { log.Printf("notifier: mark sent %d: %v", n.id, err) } // Logged, not returned: the page has already gone out, and treating a // failed timeline write as a failed delivery would send it again. detail := deliveryDetail(n) if err := logEvent(ctx, db, n.incidentID, eventNotified, n.userID, nil, nil, &detail); err != nil { log.Printf("notifier: log delivery of %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, channel_type, address, kind, attempts FROM notifications WHERE sent_at IS NULL AND send_after <= $1 AND attempts < $2 ORDER BY id LIMIT $3`, 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.channel, &n.address, &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. // // The attempt that exhausts the budget also writes a timeline event. Without it // a page that never landed leaves the incident's history identical to one that // did, which is the failure most worth seeing: nobody was told, and nothing // says so. 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 = $1, last_error = $2 WHERE id = $3", next, cause.Error(), n.id); err != nil { log.Printf("notifier: mark failed %d: %v", n.id, err) } if n.attempts+1 < notifyMaxAttempts { return } detail := fmt.Sprintf("%s: %s", deliveryDetail(n), cause) if err := logEvent(ctx, db, n.incidentID, eventNotifyFailed, n.userID, nil, nil, &detail); err != nil { log.Printf("notifier: log failure of %d: %v", n.id, err) } } // deliveryDetail is what a timeline event says about a delivery: the kind and // the channel type, "triggered via ntfy". Never the address. func deliveryDetail(n outboxRow) string { return n.kind + " via " + n.channel } // 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 // hands it to the provider its channel names. 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 { // Looked up first: a row for a provider this server no longer has can never // succeed, and there is no point rendering it. provider, ok := cfg.Providers.Get(n.channel) if !ok { return fmt.Errorf("no %q provider is configured", n.channel) } 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 = $1 AND a.status = 'firing'`, n.incidentID).Scan(&firing); err != nil { return fmt.Errorf("count firing: %w", err) } msg := renderNotification(inc, n, firing, cfg) // The page that opens an incident carries what fixed it last time, so the // person woken up starts from that. Best effort: a failed lookup must not // hold back the page itself. if n.kind == notifyTriggered { if sim, err := similarIncidents(ctx, db, n.incidentID, 1); err == nil && len(sim) > 0 && len(sim[0].ResolutionNotes) > 0 { notes := sim[0].ResolutionNotes msg.Body += "\nLast time: " + shorten(derefString(notes[len(notes)-1].Detail), 160) } } // An Acknowledge control needs both a user to attribute the acknowledgement // to and a URL the recipient can reach. if n.kind != notifyResolved && n.userID != nil && cfg.PublicURL != "" { base := strings.TrimSuffix(cfg.PublicURL, "/") if provider.Capabilities().ActionButtons { // Minted per delivery, so every push carries its own short-lived // token rather than reusing one. raw, err := issueAckToken(ctx, db, n.incidentID, *n.userID) if err != nil { return fmt.Errorf("issue ack token: %w", err) } msg.AckURL = base + "/api/notify/ack/" + raw } else { // No button to put it on, so the way to acknowledge travels as text. // The token URL is POST-only and a link to it would do nothing, and // minting one nobody can use is waste: this is the incident page, // which has the Acknowledge control. msg.Body += fmt.Sprintf("\nAcknowledge: %s/incidents/%d", base, n.incidentID) } } return provider.Send(ctx, msg) } // renderNotification builds the provider-neutral message for one queued // notification. func renderNotification(inc models.Incident, n outboxRow, firing int, cfg NotifyConfig) notify.Message { msg := notify.Message{Address: n.address, Kind: n.kind} if cfg.PublicURL != "" { // The web UI's page for the incident, so tapping the notification // opens something a browser can use. msg.ClickURL = fmt.Sprintf("%s/incidents/%d", strings.TrimSuffix(cfg.PublicURL, "/"), inc.ID) } title := pageTitle(inc) switch n.kind { case notifyResolved: msg.Title = "Resolved: " + title msg.Body = "All alerts stopped firing after " + humanDuration(time.Since(inc.TriggeredAt)) return msg case notifyReminder: msg.Title = "Still unacknowledged: " + title default: msg.Title = 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.Body = strings.Join(parts, " · ") msg.Level = severityRank(severity) return msg } // originLabel is the label that says where an alert came from, for a team with // several Kubernetes clusters behind it. It comes from Prometheus's // externalLabels and reaches an incident through Alertmanager's group_by; the // web UI reads the same label, and docs/incidents.md ("Several clusters, one team") // explains how to set it up. const originLabel = "cluster" // pageTitle is the incident's title for a notification. A phone's lock screen // cuts a long title off at the end, and the incident title puts the grouping // labels there, so the cluster would be the first thing lost. When the incident // has an origin it leads instead, "[prod-eu] PodRestarting (namespace=foo)", and // is dropped from the parenthesis so it is not said twice. A title that is not // in incidentTitle's "name (k=v, k=v)" shape keeps its text and gains the prefix. func pageTitle(inc models.Incident) string { origin := inc.GroupLabels[originLabel] if origin == "" { return inc.Title } return "[" + origin + "] " + titleWithoutLabel(inc.Title, originLabel, origin) } var titleShape = regexp.MustCompile(`(?s)^(.*?) \((.*)\)$`) // titleWithoutLabel removes "key=value" from the parenthesised tail of a title // built by incidentTitle, and the parentheses with it if nothing else is left. func titleWithoutLabel(title, key, value string) string { m := titleShape.FindStringSubmatch(title) if m == nil { return title } var rest []string for _, part := range strings.Split(m[2], ", ") { if part != key+"="+value { rest = append(rest, part) } } if len(rest) == 0 { return m[1] } return m[1] + " (" + strings.Join(rest, ", ") + ")" } // recipient is one place a notification goes: a user's channel, or the shared // fallback target. channelID and userID are nil for the fallback, which belongs // to nobody. A batch may hold several users (an escalation level with more than // one target), which is why the user is here and not on the batch. type recipient struct { userID *int64 channelID *int64 typ string address string } // enqueueNotifications queues one row per recipient for a single notification. // // The rows share one created_at. That is what makes them a batch: the reminder // and all-clear paths find "who was told last" as the rows carrying an // incident's newest created_at, so a fan-out to three channels is reminded and // resolved as one unit rather than as whichever row happened to land last. func enqueueNotifications(ctx context.Context, q querier, incidentID int64, recipients []recipient, kind string) error { now := time.Now().Unix() for _, r := range recipients { if _, err := q.ExecContext(ctx, ` INSERT INTO notifications (incident_id, user_id, channel_id, channel_type, address, kind, created_at, send_after) VALUES ($1, $2, $3, $4, $5, $6, $7, $7)`, incidentID, r.userID, r.channelID, r.typ, r.address, kind, now); err != nil { return err } } return nil } // userRecipients is the channels a notification to this user should go to: the // ones they have enabled, of a type this server can deliver. A channel whose // provider is not configured is skipped rather than queued, since it could only // fail eight times and then sit in the table. func userRecipients(ctx context.Context, q querier, cfg NotifyConfig, userID int64) ([]recipient, error) { uid := userID rows, err := q.QueryContext(ctx, ` SELECT id, type, address FROM notification_channels WHERE user_id = $1 AND enabled ORDER BY id`, userID) if err != nil { return nil, err } defer rows.Close() var out []recipient for rows.Next() { var id int64 var r recipient if err := rows.Scan(&id, &r.typ, &r.address); err != nil { return nil, err } if !cfg.Providers.Has(r.typ) { continue } r.channelID, r.userID = &id, &uid out = append(out, r) } return out, rows.Err() } // notifyTarget decides where a newly opened incident's notification goes. // // The on-call user's own channels when they have any, otherwise the fallback // target with no user attached. Deliberately not "the fallback, attributed to // the on-call user": the fallback is shared, and an Acknowledge button on a // shared target would let any recipient acknowledge as somebody else. func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int64) []recipient { if onCall != nil { if rs, err := userRecipients(ctx, q, cfg, *onCall); err == nil && len(rs) > 0 { return rs } } return fallbackRecipients(cfg, cfg.Fallback) } // fallbackRecipients turns a fallback target into the zero or one recipients it // stands for. A target whose provider this server does not have is nobody. func fallbackRecipients(cfg NotifyConfig, t notify.Target) []recipient { if t.Address == "" || !cfg.Providers.Has(t.Type) { return nil } return []recipient{{typ: t.Type, address: t.Address}} } // enqueueOpened queues the notification for a freshly opened incident. It is the // only enqueue point that has to resolve recipients 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 } recipients := notifyTarget(ctx, q, cfg, onCall) if len(recipients) == 0 { // Nobody on call has a channel and there is no fallback: there is nowhere // to send this, and queueing it would only accumulate undeliverable rows. return nil } return enqueueNotifications(ctx, q, incidentID, recipients, notifyTriggered) } // enqueueResolved queues the all-clear, reusing the recipients of the incident's // last batch. 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. A channel since disabled or removed is not told. func enqueueResolved(ctx context.Context, q querier, incidentID int64) error { rows, err := q.QueryContext(ctx, ` SELECT n.user_id, n.channel_id, n.channel_type, n.address FROM notifications n LEFT JOIN notification_channels c ON c.id = n.channel_id WHERE n.incident_id = $1 AND n.created_at = (SELECT MAX(created_at) FROM notifications WHERE incident_id = $1) AND (n.user_id IS NULL OR c.enabled) ORDER BY n.id`, incidentID) if err != nil { return err } var recipients []recipient for rows.Next() { var r recipient if err := rows.Scan(&r.userID, &r.channelID, &r.typ, &r.address); err != nil { rows.Close() return err } recipients = append(recipients, r) } err = rows.Err() rows.Close() if err != nil || len(recipients) == 0 { return err } return enqueueNotifications(ctx, q, incidentID, recipients, 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" } // shorten cuts s to at most n runes, marking the cut, and flattens newlines so // a multi-line note stays one line in a push. func shorten(s string, n int) string { s = strings.Join(strings.Fields(s), " ") r := []rune(s) if len(r) <= n { return s } return string(r[:n-1]) + "…" } // derefString reads a nullable text column as a plain string. func derefString(s *string) string { if s == nil { return "" } return *s }