terdut-server/internal/api/notifier.go

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"
)

// 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 != "" {
		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
}