package api import ( "bytes" "context" "database/sql" "encoding/json" "fmt" "log" "net/http" "strings" "time" "git.ryuvia.com/niklas/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" ) // 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 ntfy accepted it would be a lie the timeline keeps. // // The topic is deliberately absent from both. It is a shared secret with the // ntfy server — anyone holding it can 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 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) } // Logged, not returned: the page has already gone out, and treating a // failed timeline write as a failed delivery would send it again. if err := logEvent(ctx, db, n.incidentID, eventNotified, n.userID, nil, &n.kind); 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, 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. // // 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 = ?, last_error = ? WHERE id = ?", 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", n.kind, cause) if err := logEvent(ctx, db, n.incidentID, eventNotifyFailed, n.userID, nil, &detail); err != nil { log.Printf("notifier: log failure of %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 != "" { // The web UI's page for the incident, so tapping the notification // opens something a browser can use. msg.Click = fmt.Sprintf("%s/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 }