terdut-server/internal/api/escalation.go

package api

import (
	"context"
	"database/sql"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"
)

// evEscalated records a rung of the ladder on the incident's timeline: which
// level, and who it woke.
const evEscalated = "escalated"

// escalationPolicy is a team's ladder, loaded whole. It is small — a handful of
// levels with a few targets each — and every use needs all of it, so there is
// no point reading it a level at a time.
type escalationPolicy struct {
	teamID        int64
	repeatCount   int64
	fallbackTopic string
	levels        []escalationLevel
}

type escalationLevel struct {
	id       int64
	position int64
	timeout  time.Duration
	targets  []escalationTarget
}

type escalationTarget struct {
	kind   string // "user" or "oncall"
	userID *int64
}

// configured reports whether this team has anything to escalate through. A
// policy row with no levels is the same as no policy: the team gets the
// pre-escalation behaviour, which is reminders on the assignee's topic.
func (p *escalationPolicy) configured() bool { return p != nil && len(p.levels) > 0 }

// level returns the level at a 1-based position.
func (p *escalationPolicy) level(pos int64) (escalationLevel, bool) {
	for _, l := range p.levels {
		if l.position == pos {
			return l, true
		}
	}
	return escalationLevel{}, false
}

// loadEscalationPolicy reads one team's ladder. A team with no policy row
// returns nil, which every caller treats as "not configured" rather than as an
// error: most teams will never set one up.
func loadEscalationPolicy(ctx context.Context, q querier, teamID int64) (*escalationPolicy, error) {
	p := &escalationPolicy{teamID: teamID}
	err := q.QueryRowContext(ctx,
		"SELECT repeat_count, fallback_topic FROM escalation_policies WHERE team_id = $1",
		teamID).Scan(&p.repeatCount, &p.fallbackTopic)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}

	rows, err := q.QueryContext(ctx, `
		SELECT l.id, l.position, l.timeout_seconds, t.kind, t.user_id
		FROM   escalation_levels l
		LEFT   JOIN escalation_targets t ON t.level_id = l.id
		WHERE  l.team_id = $1
		ORDER  BY l.position, t.id`, teamID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	byPosition := map[int64]int{} // position -> index in p.levels
	for rows.Next() {
		var id, position, timeout int64
		var kind *string
		var userID *int64
		if err := rows.Scan(&id, &position, &timeout, &kind, &userID); err != nil {
			return nil, err
		}
		idx, seen := byPosition[position]
		if !seen {
			p.levels = append(p.levels, escalationLevel{
				id:       id,
				position: position,
				timeout:  time.Duration(timeout) * time.Second,
			})
			idx = len(p.levels) - 1
			byPosition[position] = idx
		}
		// LEFT JOIN: a level with no targets yet still produces a row, with a
		// NULL kind. It is a rung that pages nobody, which the API refuses to
		// store but an older row could still hold.
		if kind != nil {
			p.levels[idx].targets = append(p.levels[idx].targets,
				escalationTarget{kind: *kind, userID: userID})
		}
	}
	return p, rows.Err()
}

// escalate advances every incident whose current level has run out of time.
//
// Runs on the notifier's tick, beside the reminder pass, because it is the same
// question asked differently: reminders ask "has this been ignored long
// enough to say it again", escalation asks "long enough to say it to somebody
// else". Sharing the tick means one query cadence and one outbox.
func escalate(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
	rows, err := db.QueryContext(ctx, `
		SELECT i.id, i.team_id, i.escalation_level, i.escalation_level_at, i.escalation_round
		FROM   incidents i
		JOIN   escalation_policies p ON p.team_id = i.team_id
		WHERE  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 <= $1)
		  AND  i.escalation_level > 0`, time.Now().Unix())
	if err != nil {
		log.Printf("escalation: find due: %v", err)
		return
	}

	type pending struct {
		incidentID, teamID, level, round int64
		levelAt                          int64
	}
	var due []pending
	for rows.Next() {
		var p pending
		var levelAt *int64
		if err := rows.Scan(&p.incidentID, &p.teamID, &p.level, &levelAt, &p.round); err != nil {
			rows.Close()
			log.Printf("escalation: scan: %v", err)
			return
		}
		if levelAt == nil {
			continue
		}
		p.levelAt = *levelAt
		due = append(due, p)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		log.Printf("escalation: iterate: %v", err)
		return
	}

	now := time.Now()
	for _, d := range due {
		policy, err := loadEscalationPolicy(ctx, db, d.teamID)
		if err != nil {
			log.Printf("escalation: load policy for team %d: %v", d.teamID, err)
			continue
		}
		if !policy.configured() {
			continue
		}
		current, ok := policy.level(d.level)
		if !ok {
			continue
		}
		if now.Sub(time.Unix(d.levelAt, 0)) < current.timeout {
			continue
		}
		if err := advanceEscalation(ctx, db, cfg, policy, d.incidentID, d.level, d.round, now); err != nil {
			log.Printf("escalation: advance incident %d: %v", d.incidentID, err)
		}
	}
}

// advanceEscalation moves one incident to its next rung, or off the end of the
// ladder.
//
// The whole move is one transaction: the level, the page and the timeline entry
// are one event, and an incident recorded as being at level 3 that nobody at
// level 3 was told about is the worst of the possible half-states.
func advanceEscalation(ctx context.Context, db *sql.DB, cfg NotifyConfig, policy *escalationPolicy, incidentID, level, round int64, now time.Time) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback() //nolint:errcheck

	next := level + 1
	nextRound := round
	if _, ok := policy.level(next); !ok {
		// Off the end. Either start the chain again, or make the last call.
		if round < policy.repeatCount {
			next, nextRound = 1, round+1
		} else {
			if err := escalationExhausted(ctx, tx, policy, incidentID, now); err != nil {
				return err
			}
			return tx.Commit()
		}
	}

	target, ok := policy.level(next)
	if !ok {
		return nil
	}
	paged, err := pageLevel(ctx, tx, cfg, policy, incidentID, target)
	if err != nil {
		return err
	}

	if _, err := tx.ExecContext(ctx, `
		UPDATE incidents
		   SET escalation_level = $1, escalation_level_at = $2, escalation_round = $3
		 WHERE id = $4`, next, now.Unix(), nextRound, incidentID); err != nil {
		return err
	}

	detail := "level " + strconv.FormatInt(next, 10)
	if nextRound > round {
		detail += " (round " + strconv.FormatInt(nextRound+1, 10) + ")"
	}
	if len(paged) > 0 {
		detail += ": " + strings.Join(paged, ", ")
	} else {
		// Worth recording loudly: the rung exists, its turn came, and it woke
		// nobody. That is a policy that looks configured and is not.
		detail += ": nobody reachable"
	}
	if err := logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail); err != nil {
		return err
	}
	return tx.Commit()
}

// escalationExhausted is the end of the line: the fallback topic, once, and a
// timeline entry saying the ladder is finished. The incident stays triggered —
// escalation running out is not the same as somebody answering.
func escalationExhausted(ctx context.Context, tx *sql.Tx, policy *escalationPolicy, incidentID int64, now time.Time) error {
	detail := "escalation exhausted"
	if policy.fallbackTopic != "" {
		if err := enqueueNotification(ctx, tx, incidentID, nil, policy.fallbackTopic, notifyEscalated); err != nil {
			return err
		}
		detail += ": paged " + policy.fallbackTopic
	} else {
		detail += ": no fallback topic configured"
	}

	// Level 0 again, so the sweep stops considering it. The round counter is
	// left where it is, as the record of how far it got.
	if _, err := tx.ExecContext(ctx,
		"UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1",
		incidentID); err != nil {
		return err
	}
	return logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail)
}

// pageLevel notifies every target of one level and reports who was woken.
//
// Each target gets its own outbox row, so each gets its own Acknowledge token:
// the button in a notification must acknowledge as the person holding the
// phone, not as whoever was paged first.
func pageLevel(ctx context.Context, tx *sql.Tx, cfg NotifyConfig, policy *escalationPolicy, incidentID int64, level escalationLevel) ([]string, error) {
	var paged []string
	seen := map[int64]bool{}

	for _, t := range level.targets {
		userID := t.userID
		if t.kind == "oncall" {
			onCall, err := currentOnCall(ctx, tx, policy.teamID)
			if err != nil {
				return nil, err
			}
			if onCall == nil {
				continue
			}
			userID = onCall
		}
		if userID == nil || seen[*userID] {
			continue
		}
		seen[*userID] = true

		var topic *string
		var username string
		if err := tx.QueryRowContext(ctx,
			"SELECT ntfy_topic, username FROM users WHERE id = $1 AND disabled_at IS NULL",
			*userID).Scan(&topic, &username); err != nil {
			// A disabled or deleted account is not an error in the middle of an
			// escalation: it is a target that cannot be woken, and the next
			// level is the answer to that.
			continue
		}
		if topic == nil || *topic == "" {
			continue
		}
		if err := enqueueNotification(ctx, tx, incidentID, userID, *topic, notifyEscalated); err != nil {
			return nil, err
		}
		paged = append(paged, username)
	}
	return paged, nil
}

// startEscalation puts a newly opened incident on the first rung, when its team
// has a ladder. Called from openIncident, inside the same transaction, so an
// incident is never briefly open with no escalation clock running.
func startEscalation(ctx context.Context, q querier, incidentID, teamID int64) error {
	policy, err := loadEscalationPolicy(ctx, q, teamID)
	if err != nil || !policy.configured() {
		return err
	}
	_, err = q.ExecContext(ctx,
		"UPDATE incidents SET escalation_level = 1, escalation_level_at = $1 WHERE id = $2",
		time.Now().Unix(), incidentID)
	return err
}

// stopEscalation takes an incident off the ladder. Acknowledging or resolving
// is somebody saying "I have this", and continuing to wake people after that is
// the behaviour that teaches people to ignore the tool.
func stopEscalation(ctx context.Context, q querier, incidentID int64) error {
	_, err := q.ExecContext(ctx,
		"UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1",
		incidentID)
	return err
}

// handleGetEscalation returns a team's ladder.
func handleGetEscalation(db *sql.DB) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		teamID, ok := teamParam(w, r)
		if !ok {
			return
		}
		if !requireTeamMember(w, r, teamID) {
			return
		}

		policy, err := loadEscalationPolicy(r.Context(), db, teamID)
		if err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}
		respond(w, http.StatusOK, escalationResponse(policy, teamID))
	}
}

type escalationLevelJSON struct {
	Position       int64                  `json:"position"`
	TimeoutSeconds int64                  `json:"timeout_seconds"`
	Targets        []escalationTargetJSON `json:"targets"`
}

type escalationTargetJSON struct {
	Kind   string `json:"kind"`
	UserID *int64 `json:"user_id,omitempty"`
}

type escalationJSON struct {
	TeamID        int64                 `json:"team_id"`
	RepeatCount   int64                 `json:"repeat_count"`
	FallbackTopic string                `json:"fallback_topic"`
	Levels        []escalationLevelJSON `json:"levels"`
}

func escalationResponse(p *escalationPolicy, teamID int64) escalationJSON {
	out := escalationJSON{TeamID: teamID, Levels: []escalationLevelJSON{}}
	if p == nil {
		return out
	}
	out.RepeatCount = p.repeatCount
	out.FallbackTopic = p.fallbackTopic
	for _, l := range p.levels {
		level := escalationLevelJSON{
			Position:       l.position,
			TimeoutSeconds: int64(l.timeout.Seconds()),
			Targets:        []escalationTargetJSON{},
		}
		for _, t := range l.targets {
			level.Targets = append(level.Targets, escalationTargetJSON{Kind: t.kind, UserID: t.userID})
		}
		out.Levels = append(out.Levels, level)
	}
	return out
}

// handleSetEscalation replaces a team's ladder wholesale.
//
// Replace rather than patch: the levels are an order, and an API that edits one
// rung has to answer what happens to the numbering of the others. Sending the
// whole ladder makes the order the client's to decide and the server's to
// store, and makes an edit atomic — there is no moment where level 2 exists
// twice.
func handleSetEscalation(db *sql.DB) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		teamID, ok := teamParam(w, r)
		if !ok {
			return
		}
		if !requireTeamOwner(w, r, teamID) {
			return
		}

		var req escalationJSON
		if err := decodeJSON(r, &req); err != nil {
			respond(w, http.StatusBadRequest, errResp("invalid request body"))
			return
		}
		if req.RepeatCount < 0 || req.RepeatCount > 10 {
			respond(w, http.StatusBadRequest, errResp("repeat_count must be between 0 and 10"))
			return
		}
		for i, l := range req.Levels {
			if l.TimeoutSeconds <= 0 {
				respond(w, http.StatusBadRequest, errResp("every level needs a timeout"))
				return
			}
			if len(l.Targets) == 0 {
				// A rung that pages nobody is not a delay, it is a silence with
				// a number on it.
				respond(w, http.StatusBadRequest,
					errResp("level "+strconv.FormatInt(int64(i+1), 10)+" has no targets"))
				return
			}
			for _, t := range l.Targets {
				switch t.Kind {
				case "oncall":
					if t.UserID != nil {
						respond(w, http.StatusBadRequest, errResp("an oncall target takes no user_id"))
						return
					}
				case "user":
					if t.UserID == nil {
						respond(w, http.StatusBadRequest, errResp("a user target needs a user_id"))
						return
					}
				default:
					respond(w, http.StatusBadRequest, errResp("target kind must be user or oncall"))
					return
				}
			}
		}

		tx, err := db.BeginTx(r.Context(), nil)
		if err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}
		defer tx.Rollback() //nolint:errcheck

		if _, err := tx.ExecContext(r.Context(), `
			INSERT INTO escalation_policies (team_id, repeat_count, fallback_topic, updated_at)
			VALUES ($1, $2, $3, `+nowEpoch+`)
			ON CONFLICT (team_id) DO UPDATE SET
				repeat_count = excluded.repeat_count,
				fallback_topic = excluded.fallback_topic,
				updated_at = excluded.updated_at`,
			teamID, req.RepeatCount, req.FallbackTopic); err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}
		// The levels are replaced, not merged; the cascade takes the targets.
		if _, err := tx.ExecContext(r.Context(),
			"DELETE FROM escalation_levels WHERE team_id = $1", teamID); err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}

		for i, l := range req.Levels {
			var levelID int64
			if err := tx.QueryRowContext(r.Context(), `
				INSERT INTO escalation_levels (team_id, position, timeout_seconds)
				VALUES ($1, $2, $3) RETURNING id`,
				teamID, int64(i+1), l.TimeoutSeconds).Scan(&levelID); err != nil {
				respond(w, http.StatusInternalServerError, errResp("internal error"))
				return
			}
			for _, t := range l.Targets {
				if _, err := tx.ExecContext(r.Context(), `
					INSERT INTO escalation_targets (level_id, kind, user_id)
					VALUES ($1, $2, $3)`, levelID, t.Kind, t.UserID); err != nil {
					// The only foreign key here is the user.
					respond(w, http.StatusBadRequest, errResp("unknown user in targets"))
					return
				}
			}
		}

		if err := tx.Commit(); err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}

		policy, err := loadEscalationPolicy(r.Context(), db, teamID)
		if err != nil {
			respond(w, http.StatusInternalServerError, errResp("internal error"))
			return
		}
		respond(w, http.StatusOK, escalationResponse(policy, teamID))
	}
}