terdut-server/internal/api/notifier.go

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
}