8e7aa0d451
The sev4 repeat path reads the nudges table, so ending an alarm means writing an ending there. 'resolved' is the daemon closing it because the condition cleared, which is neither 'acted' nor 'ignored'. ResolvePendingTelegram is rule-scoped and accepts only the two endings the daemon may write. OldestPendingTelegram backs the age cap and scans into a NullInt64, because MIN over an empty set is one NULL row, not zero rows. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
302 lines
11 KiB
Go
302 lines
11 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// Nudge — one proactive send, with deferred outcome. outcomes are: pending |
|
||
// acted | snoozed | ignored. outcome_ts set when resolved. the outcome column
|
||
// IS the restraint-memory signal — no separate table for the feedback loop.
|
||
type Nudge struct {
|
||
ID int64
|
||
Ts time.Time // sent ts
|
||
Rule string
|
||
Channel string
|
||
Message string
|
||
Outcome string
|
||
OutcomeTs sql.NullInt64
|
||
}
|
||
|
||
const (
|
||
NudgePending = "pending"
|
||
NudgeActed = "acted"
|
||
NudgeSnoozed = "snoozed"
|
||
NudgeIgnored = "ignored"
|
||
|
||
// NudgeResolved — the thing it was about stopped being true, and nobody
|
||
// answered. Distinct from acted, which means he did something, and from
|
||
// ignored, which means he chose not to (Vikunja #535).
|
||
//
|
||
// It exists because a repeating alarm needs an ending that is not a lie.
|
||
// Marking a cleared service_down "acted" would credit him with a response
|
||
// he never made and would teach the cooldown tuner to nudge harder; leaving
|
||
// it pending is what made the alarm ring for two hours after the service
|
||
// came back. This outcome is written by the daemon, never by a person, and
|
||
// it is deliberately invisible to the feedback loop — see RecentOutcomes.
|
||
NudgeResolved = "resolved"
|
||
)
|
||
|
||
// SnoozeDuration — how long one `snoozed` outcome keeps its rule quiet.
|
||
//
|
||
// The nudges table records THAT a snooze happened and when, never for how
|
||
// long: nothing upstream can supply a length. ResolveNudge takes only
|
||
// (id, outcome, ts), and so do the IPC method and the web/telegram callers
|
||
// behind it. So a fixed default it is, rather than a new column no writer
|
||
// could fill.
|
||
//
|
||
// Two hours: longer than every rule's base cooldown (15–60m) so a snooze
|
||
// actually buys quiet instead of being swallowed by the cooldown, and short
|
||
// enough that a snooze the operator forgets about clears the same day. A
|
||
// snooze can never outlive this window, so Maven cannot go quiet forever.
|
||
const SnoozeDuration = 2 * time.Hour
|
||
|
||
var (
|
||
ErrNudgeNotFound = errors.New("store: nudge not found")
|
||
ErrNudgeOutcome = errors.New("store: nudge already resolved")
|
||
)
|
||
|
||
// RecordNudge inserts a pending nudge (sent). returns the id. the loop writes
|
||
// one row per proactive send per tick (one nudge per tick, max severity).
|
||
func (s *Store) RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error) {
|
||
res, err := s.db.ExecContext(ctx,
|
||
`INSERT INTO nudges (ts, rule, channel, message, outcome) VALUES (?,?,?,?, 'pending')`,
|
||
ts.UnixMilli(), rule, channel, message)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("record nudge: %w", err)
|
||
}
|
||
id, err := res.LastInsertId()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("record nudge: last insert id: %w", err)
|
||
}
|
||
return id, nil
|
||
}
|
||
|
||
// ResolveNudge sets the outcome of a still-pending nudge. acted | snoozed |
|
||
// ignored — caller decides by user response (or lack of it). idempotency is
|
||
// rejected here: a nudge can only be resolved once, by design — re-resolving
|
||
// would silently corrupt the feedback signal (the table IS the learning input).
|
||
func (s *Store) ResolveNudge(ctx context.Context, id int64, outcome string, ts time.Time) error {
|
||
switch outcome {
|
||
case NudgeActed, NudgeSnoozed, NudgeIgnored:
|
||
default:
|
||
return fmt.Errorf("store: unknown outcome %q", outcome)
|
||
}
|
||
res, err := s.db.ExecContext(ctx,
|
||
`UPDATE nudges SET outcome = ?, outcome_ts = ? WHERE id = ? AND outcome = 'pending'`,
|
||
outcome, ts.UnixMilli(), id)
|
||
if err != nil {
|
||
return fmt.Errorf("resolve nudge: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("resolve nudge: rows affected: %w", err)
|
||
}
|
||
if n == 0 {
|
||
// either no such row, or it was already resolved — distinguish so callers
|
||
// can tell a bug from a race.
|
||
var cur string
|
||
err := s.db.QueryRowContext(ctx, "SELECT outcome FROM nudges WHERE id = ?", id).Scan(&cur)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNudgeNotFound
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return fmt.Errorf("%w: currently %s", ErrNudgeOutcome, cur)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RecentOutcomes returns the last N outcomes (in reverse chronological order)
|
||
// for a given rule — the feedback loop's only input. used to compute
|
||
// ignored_rate → cooldown sizing. dead simple at mvp: a ratio over last N.
|
||
func (s *Store) RecentOutcomes(ctx context.Context, rule string, n int) ([]string, error) {
|
||
// 'resolved' is excluded alongside 'pending' (Vikunja #535). The tuner reads
|
||
// this as "how often does he answer me", and a resolution the daemon wrote
|
||
// is not him answering. Counting it would dilute both rates toward zero and
|
||
// make a service that fixes itself look like a rule he neither acts on nor
|
||
// ignores, which is a fact about the world and not feedback about her.
|
||
rows, err := s.db.QueryContext(ctx,
|
||
`SELECT outcome FROM nudges
|
||
WHERE rule = ? AND outcome NOT IN ('pending', 'resolved')
|
||
ORDER BY ts DESC, id DESC LIMIT ?`, rule, n)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("recent outcomes: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []string
|
||
for rows.Next() {
|
||
var o string
|
||
if err := rows.Scan(&o); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, o)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// UnackedTelegramRules returns rule names that have at least one still-pending
|
||
// (un-acked) nudge sent over the telegram channel. the dispatcher's
|
||
// RepeatUnacked re-sends these each tick until MarkAcked.
|
||
//
|
||
// telegram is the sev4-away channel by routing-table construction
|
||
// (ChannelsFor sends sev4 away → telegram, nothing else to telegram), so
|
||
// `channel='telegram' AND outcome='pending'` already implies sev4 ops-hard —
|
||
// no severity column on the nudges table, and none needed: the routing table
|
||
// is the authority. grouped by rule so the repeat stream is one-per-rule
|
||
// (the ack key IS the rule name).
|
||
func (s *Store) UnackedTelegramRules(ctx context.Context) ([]string, error) {
|
||
rows, err := s.db.QueryContext(ctx,
|
||
`SELECT rule FROM nudges
|
||
WHERE channel = 'telegram' AND outcome = 'pending'
|
||
GROUP BY rule ORDER BY rule`)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("unacked telegram rules: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []string
|
||
for rows.Next() {
|
||
var r string
|
||
if err := rows.Scan(&r); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, r)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ResolvePendingTelegram closes every still-pending telegram nudge for a rule
|
||
// and reports how many it closed. This is what stops a repeating alarm
|
||
// (Vikunja #535).
|
||
//
|
||
// Rule-scoped and not id-scoped, deliberately: the repeat key IS the rule name,
|
||
// so a rule with several pending rows repeats once per tick for all of them and
|
||
// must go quiet for all of them at once. Closing one id would leave the alarm
|
||
// ringing on the others.
|
||
//
|
||
// The outcome is a parameter rather than hardcoded, because "the condition
|
||
// cleared" and "nobody could ever answer this" are different endings and
|
||
// /notifications should not show them as the same one.
|
||
func (s *Store) ResolvePendingTelegram(ctx context.Context, rule, outcome string, ts time.Time) (int64, error) {
|
||
switch outcome {
|
||
case NudgeResolved, NudgeIgnored:
|
||
default:
|
||
return 0, fmt.Errorf("store: %q is not an ending the daemon may write", outcome)
|
||
}
|
||
res, err := s.db.ExecContext(ctx,
|
||
`UPDATE nudges SET outcome = ?, outcome_ts = ?
|
||
WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`,
|
||
outcome, ts.UnixMilli(), rule)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("resolve pending telegram %s: %w", rule, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("resolve pending telegram %s: rows affected: %w", rule, err)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// OldestPendingTelegram returns when the oldest still-pending telegram nudge
|
||
// for a rule was sent. Used for the repeat age cap: an alarm nobody has
|
||
// answered in hours is not one more repeat away from being answered.
|
||
//
|
||
// Oldest and not newest, because the age that matters is how long the alarm has
|
||
// been ringing, not when it last rang — the repeat itself does not create new
|
||
// rows, but a re-fire of the rule does, and the alarm is the whole run.
|
||
func (s *Store) OldestPendingTelegram(ctx context.Context, rule string) (time.Time, error) {
|
||
// MIN over an empty set is one row holding NULL, not zero rows, so this
|
||
// scans into a NullInt64 and never sees sql.ErrNoRows.
|
||
var tsMilli sql.NullInt64
|
||
err := s.db.QueryRowContext(ctx,
|
||
`SELECT MIN(ts) FROM nudges
|
||
WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`, rule).Scan(&tsMilli)
|
||
if err != nil {
|
||
return time.Time{}, fmt.Errorf("oldest pending telegram %s: %w", rule, err)
|
||
}
|
||
if !tsMilli.Valid {
|
||
return time.Time{}, ErrNudgeNotFound
|
||
}
|
||
return time.UnixMilli(tsMilli.Int64).UTC(), nil
|
||
}
|
||
|
||
// SnoozedUntil — per rule, when its most recent snooze runs out. This is the
|
||
// read behind the gate's snooze check: the `snoozed` outcome already in the
|
||
// nudges table IS the restraint memory, so there is no snooze table.
|
||
//
|
||
// Rules with no live snooze are absent from the map, which is what the gate
|
||
// wants (a missing key means "not snoozed"). Expired snoozes are filtered out
|
||
// in SQL, so an old snooze can never come back as a silent forever-mute.
|
||
//
|
||
// Called every tick (~60s). One indexed lookup over the snoozed rows only.
|
||
func (s *Store) SnoozedUntil(ctx context.Context, now time.Time) (map[string]time.Time, error) {
|
||
cutoff := now.Add(-SnoozeDuration).UnixMilli()
|
||
rows, err := s.db.QueryContext(ctx,
|
||
`SELECT rule, MAX(outcome_ts) FROM nudges
|
||
WHERE outcome = 'snoozed' AND outcome_ts > ?
|
||
GROUP BY rule`, cutoff)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("snoozed until: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
out := make(map[string]time.Time)
|
||
for rows.Next() {
|
||
var rule string
|
||
var tsMilli int64
|
||
if err := rows.Scan(&rule, &tsMilli); err != nil {
|
||
return nil, err
|
||
}
|
||
out[rule] = time.UnixMilli(tsMilli).UTC().Add(SnoozeDuration)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// RecentNudges — the newest n nudges across all rules, with outcomes, for the
|
||
// monitoring dash. Newest first.
|
||
func (s *Store) RecentNudges(ctx context.Context, n int) ([]Nudge, error) {
|
||
rows, err := s.db.QueryContext(ctx,
|
||
`SELECT id, ts, rule, channel, message, outcome, outcome_ts
|
||
FROM nudges ORDER BY ts DESC, id DESC LIMIT ?`, n)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("recent nudges: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []Nudge
|
||
for rows.Next() {
|
||
var ng Nudge
|
||
var tsMilli int64
|
||
if err := rows.Scan(&ng.ID, &tsMilli, &ng.Rule, &ng.Channel, &ng.Message, &ng.Outcome, &ng.OutcomeTs); err != nil {
|
||
return nil, err
|
||
}
|
||
ng.Ts = time.UnixMilli(tsMilli).UTC()
|
||
out = append(out, ng)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// LastNudge — newest nudge for a rule regardless of outcome. used by the gate
|
||
// for cooldown enforcement. returns ErrNudgeNotFound if the rule has never fired.
|
||
func (s *Store) LastNudge(ctx context.Context, rule string) (Nudge, error) {
|
||
row := s.db.QueryRowContext(ctx,
|
||
`SELECT id, ts, rule, channel, message, outcome, outcome_ts
|
||
FROM nudges WHERE rule = ? ORDER BY ts DESC, id DESC LIMIT 1`, rule)
|
||
var n Nudge
|
||
var tsMilli int64
|
||
var outcomeTs sql.NullInt64
|
||
if err := row.Scan(&n.ID, &tsMilli, &n.Rule, &n.Channel, &n.Message, &n.Outcome, &outcomeTs); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return Nudge{}, ErrNudgeNotFound
|
||
}
|
||
return Nudge{}, err
|
||
}
|
||
n.Ts = time.UnixMilli(tsMilli).UTC()
|
||
if outcomeTs.Valid {
|
||
n.OutcomeTs = outcomeTs
|
||
}
|
||
return n, nil
|
||
}
|