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