claude
4f516657da
A rule over a key set that only exists at read time cannot declare its keys at wiring time. Kuma has one monitor per service and the names live in the gauge, so the rule declares a prefix and the gatherer resolves the family per tick. ServiceDownRule now fires on any monitor reading down, names it through DownServices, and is edge-triggered: a service that stays down is one nudge, not one per tick with cooldown as the only brake.
460 lines
16 KiB
Go
460 lines
16 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/kami/maven/internal/calendar"
|
|
)
|
|
|
|
// WriteFact appends a fact row. confidence must be 1.0 for taps and (0,1) for
|
|
// inferences; the caller is responsible for that provenance discipline.
|
|
//
|
|
// If voidsID is Valid, this row voids (cancels) the referenced fact. The
|
|
// caller must have verified voidsID points at an existing fact — we check it
|
|
// here too and refuse to write a dangling voids pointer (the audit trail must
|
|
// stay coherent).
|
|
func (s *Store) WriteFact(ctx context.Context, ts time.Time, kind FactKind, key, value, source string, confidence float64, voidsID sql.NullInt64) (int64, error) {
|
|
if confidence <= 0.0 || confidence > 1.0 {
|
|
return 0, fmt.Errorf("%w: %f", ErrConfidence, confidence)
|
|
}
|
|
if voidsID.Valid {
|
|
var ok int64
|
|
err := s.db.QueryRowContext(ctx, "SELECT 1 FROM facts WHERE id = ?", voidsID.Int64).Scan(&ok)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return 0, fmt.Errorf("%w: id=%d", ErrVoidsMissing, voidsID.Int64)
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("voids lookup: %w", err)
|
|
}
|
|
}
|
|
res, err := s.db.ExecContext(ctx,
|
|
`INSERT INTO facts (ts, kind, key, value, source, confidence, voids_id) VALUES (?,?,?,?,?,?,?)`,
|
|
ts.UnixMilli(), string(kind), key, value, source, confidence, voidsID)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("write fact: %w", err)
|
|
}
|
|
id, err := res.LastInsertId()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("last insert id: %w", err)
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
// FactRecallText is the text a fact is indexed under and read back as (#493).
|
|
//
|
|
// It used to be the utterance that wrote the fact, so recall of ANY
|
|
// voice-tapped fact answered with the sentence he said instead of the value
|
|
// stored: `go_version = 1.20` was indexed as "какая последняя версия языка
|
|
// Go?", and that question is what came back. The poisoned rows made the defect
|
|
// visible; the shape was wrong for legitimate facts too.
|
|
//
|
|
// The key is spoken with its underscores dropped, because a key is written for
|
|
// the store and this string is read out loud.
|
|
func FactRecallText(key, value string) string {
|
|
spoken := strings.TrimSpace(strings.ReplaceAll(key, "_", " "))
|
|
v := strings.TrimSpace(DecodeFactValue(value))
|
|
switch {
|
|
case v == "":
|
|
return spoken
|
|
case spoken == "":
|
|
return v
|
|
}
|
|
return spoken + " — " + v
|
|
}
|
|
|
|
// DecodeFactValue unwraps a stored value for reading. The column holds raw json
|
|
// when the writer serialized one (SetValue, CorrectValue) and a plain string
|
|
// when it did not (a voice tap), so a reader that wants the text handles both.
|
|
func DecodeFactValue(value string) string {
|
|
var s string
|
|
if err := json.Unmarshal([]byte(value), &s); err == nil {
|
|
return s
|
|
}
|
|
return value
|
|
}
|
|
|
|
// LatestFact returns the latest non-voided fact for key, or ErrNoFact.
|
|
// "Non-voided" = no later row has voids_id pointing at it. We resolve this by
|
|
// taking the newest row whose id is not referenced by any voids_id.
|
|
func (s *Store) LatestFact(ctx context.Context, key string) (Fact, error) {
|
|
row := s.db.QueryRowContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts
|
|
WHERE key = ?
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY ts DESC, id DESC
|
|
LIMIT 1`, key)
|
|
return scanFact(row)
|
|
}
|
|
|
|
// RecentFacts — the newest n facts across all keys, for the monitoring dash.
|
|
// Includes voided rows (the audit trail is the point: you want to SEE a
|
|
// correction, not have it hidden). Newest first.
|
|
func (s *Store) RecentFacts(ctx context.Context, n int) ([]Fact, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts ORDER BY ts DESC, id DESC LIMIT ?`, n)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("recent facts: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []Fact
|
|
for rows.Next() {
|
|
f, err := scanFact(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// RecentActiveFactsByKind — the newest n facts of one kind, newest first, with
|
|
// the retracted ones left out. "Active" means two exclusions: a row some later
|
|
// row voids, and the void marker VoidLatestFact writes to retract it. A
|
|
// correction still counts, because a correction is a value he stands behind.
|
|
//
|
|
// It exists because the facts table is shared and the noisy writers are not the
|
|
// interesting ones. A caller that reads n recent rows and then keeps the self
|
|
// ones has a window whose real length is set by how often the pollers write:
|
|
// one WireGuard peer alone rehandshakes every couple of minutes, which is
|
|
// enough env rows to reduce a 2000-row window to under three days. Filtering in
|
|
// SQL makes the bound mean what the caller thinks it means.
|
|
//
|
|
// Voided rows are excluded here and included by RecentFacts on purpose. The
|
|
// dash shows the audit trail, because you want to SEE a correction. A reader
|
|
// that asks what he usually does must not count something he explicitly took
|
|
// back.
|
|
func (s *Store) RecentActiveFactsByKind(ctx context.Context, kind FactKind, n int) ([]Fact, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts
|
|
WHERE kind = ?
|
|
AND NOT (voids_id IS NOT NULL AND value = '"voided"')
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY ts DESC, id DESC LIMIT ?`, string(kind), n)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("recent facts by kind: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []Fact
|
|
for rows.Next() {
|
|
f, err := scanFact(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// CalendarEvents returns calendar facts whose key date falls within [from, to).
|
|
// Calendar event keys have the format calendar_event_YYYYMMDD_<summary>.
|
|
//
|
|
// Every calendar source is included, not just the personal CalDAV poll: the work
|
|
// calendar arrives as ambient:notif notifications (Vikunja #126) and belongs in
|
|
// the same answer. The source stays on each Fact, along with its confidence, so
|
|
// the caller can hedge a reading it did not get from a calendar server —
|
|
// filtering by source here would have thrown that judgement away.
|
|
//
|
|
// One row per event, not one per write. The facts table is append-only, so a
|
|
// standup moved from 14:00 to 16:00 leaves two rows under the same key, and the
|
|
// day plan used to recite both as if the owner had two meetings. Voided rows
|
|
// are excluded, the latest row wins within a source, and the best-evidenced
|
|
// source wins across them — a calendar read beats the notification relay that
|
|
// guessed at the same meeting.
|
|
func (s *Store) CalendarEvents(ctx context.Context, from, to time.Time) ([]Fact, error) {
|
|
prefixFrom := calendar.KeyPrefixForDay(from)
|
|
prefixTo := calendar.KeyPrefixForDay(to)
|
|
sources := calendar.Sources()
|
|
args := make([]any, 0, len(sources)+2)
|
|
for _, src := range sources {
|
|
args = append(args, src)
|
|
}
|
|
args = append(args, prefixFrom, prefixTo)
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts
|
|
WHERE source IN (`+placeholders(len(sources))+`)
|
|
AND key >= ? AND key < ?
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY key, ts, id`, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("calendar events: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []Fact
|
|
for rows.Next() {
|
|
f, err := scanFact(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return latestPerCalendarKey(out), nil
|
|
}
|
|
|
|
// latestPerCalendarKey reduces the append-only rows for one day to one row per
|
|
// event key. Input must be ordered by key then oldest-first, so the last row
|
|
// seen for a key and source is that source's current value.
|
|
func latestPerCalendarKey(in []Fact) []Fact {
|
|
type slot struct {
|
|
bySource map[string]Fact
|
|
order []string
|
|
}
|
|
var keys []string
|
|
byKey := map[string]*slot{}
|
|
for _, f := range in {
|
|
s, ok := byKey[f.Key]
|
|
if !ok {
|
|
s = &slot{bySource: map[string]Fact{}}
|
|
byKey[f.Key] = s
|
|
keys = append(keys, f.Key)
|
|
}
|
|
if _, seen := s.bySource[f.Source]; !seen {
|
|
s.order = append(s.order, f.Source)
|
|
}
|
|
s.bySource[f.Source] = f
|
|
}
|
|
out := make([]Fact, 0, len(keys))
|
|
for _, k := range keys {
|
|
s := byKey[k]
|
|
best := s.bySource[s.order[0]]
|
|
for _, src := range s.order[1:] {
|
|
if s.bySource[src].Confidence > best.Confidence {
|
|
best = s.bySource[src]
|
|
}
|
|
}
|
|
out = append(out, best)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// LatestFactBySource — provenance-scoped. A rule on `service_down` trusts only
|
|
// source=poll:healthcheck; a compromised poller can't forge a trigger. Use this
|
|
// from rules, not LatestFact, whenever the rule's source contract matters.
|
|
func (s *Store) LatestFactBySource(ctx context.Context, key, source string) (Fact, error) {
|
|
row := s.db.QueryRowContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts
|
|
WHERE key = ? AND source = ?
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY ts DESC, id DESC
|
|
LIMIT 1`, key, source)
|
|
return scanFact(row)
|
|
}
|
|
|
|
// LatestFactsByPrefix — the latest non-voided fact for every key that starts
|
|
// with prefix, newest-per-key, ordered by key.
|
|
//
|
|
// The loop's gatherer loads the keys its rules declare, which works while the
|
|
// key set is static. Kuma's monitors are not: one fact per monitor means the
|
|
// keys are only known once the gauge is read, so the rule declares the prefix
|
|
// and this read resolves it per tick. `_` and `%` are escaped — a monitor name
|
|
// is user text and must not act as a LIKE wildcard.
|
|
func (s *Store) LatestFactsByPrefix(ctx context.Context, prefix string) ([]Fact, error) {
|
|
esc := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(prefix)
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, ts, kind, key, value, source, confidence, voids_id
|
|
FROM facts f
|
|
WHERE key LIKE ? ESCAPE '\'
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
AND id = (SELECT id FROM facts g
|
|
WHERE g.key = f.key
|
|
AND g.id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY g.ts DESC, g.id DESC LIMIT 1)
|
|
ORDER BY key`, esc+"%")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("facts by prefix %q: %w", prefix, err)
|
|
}
|
|
defer rows.Close()
|
|
var out []Fact
|
|
for rows.Next() {
|
|
f, err := scanFact(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// Since returns how long ago the latest non-voided fact for key landed, or
|
|
// (0, ErrNoFact). Implements the `since(key)==null → don't fire` guard from
|
|
// the spec — silence on no-data is "shuts up when uncertain".
|
|
func (s *Store) Since(ctx context.Context, key string, now time.Time) (time.Duration, error) {
|
|
f, err := s.LatestFact(ctx, key)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if now.Before(f.Ts) {
|
|
return 0, nil
|
|
}
|
|
return now.Sub(f.Ts), nil
|
|
}
|
|
|
|
// SetValue is a convenience for writing a structured (json) value at confidence 1.0
|
|
// from a tap. Self-taps (water, meal, shower) land here. Caller supplies source
|
|
// like "tap:water"; we serialize the value.
|
|
func (s *Store) SetValue(ctx context.Context, kind FactKind, key, source string, value any, ts time.Time) (int64, error) {
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("marshal value: %w", err)
|
|
}
|
|
return s.WriteFact(ctx, ts, kind, key, string(raw), source, 1.0, sql.NullInt64{})
|
|
}
|
|
|
|
// CorrectValue voids the latest non-voided fact for key and writes a replacement
|
|
// in one transaction. Use this for "you corrected a bad fact" feedback — keeps
|
|
// the audit trail, supersedes the wrong value.
|
|
func (s *Store) CorrectValue(ctx context.Context, key, source string, value any, ts time.Time) (newID int64, err error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
var oldID int64
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT id FROM facts
|
|
WHERE key = ?
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY ts DESC, id DESC LIMIT 1`, key).Scan(&oldID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// nothing to correct; write as a plain new fact instead — no voiding needed.
|
|
} else if err != nil {
|
|
return 0, fmt.Errorf("correct: find old: %w", err)
|
|
}
|
|
raw, merr := json.Marshal(value)
|
|
if merr != nil {
|
|
return 0, fmt.Errorf("marshal value: %w", merr)
|
|
}
|
|
var voids sql.NullInt64
|
|
if oldID != 0 {
|
|
voids = sql.NullInt64{Int64: oldID, Valid: true}
|
|
}
|
|
res, err := tx.ExecContext(ctx,
|
|
`INSERT INTO facts (ts, kind, key, value, source, confidence, voids_id) VALUES (?,?,?,?,?,?,?)`,
|
|
ts.UnixMilli(), string(KindSelf), key, string(raw), source, 1.0, voids)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("write corrected: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return 0, err
|
|
}
|
|
newID, err = res.LastInsertId()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("last insert id: %w", err)
|
|
}
|
|
// The same repair a void needs, for the same reason (#493). A correction
|
|
// supersedes the value, and the vector still holds the old one, so recall
|
|
// kept answering with the value he had just corrected. Dropping it costs
|
|
// the key its recall vector until the fact is tapped again: this layer has
|
|
// no embedder, and a missing vector loses a question while a stale one
|
|
// answers it wrongly.
|
|
//
|
|
// Best-effort: the corrected row is committed, and a correction that lands
|
|
// beats one that fails on cleanup.
|
|
if n, derr := s.VectorMemory().DeletePrefix(ctx, "fact:"+key+":"); derr != nil {
|
|
log.Printf("store: correct %q: memory vectors survive: %v", key, derr)
|
|
} else if n > 0 {
|
|
log.Printf("store: correct %q: dropped %d superseded memory vector(s)", key, n)
|
|
}
|
|
return newID, nil
|
|
}
|
|
|
|
// VoidLatestFact voids the latest non-voided fact for key. It writes a new
|
|
// fact with voids_id pointing at the old one, keeping the audit trail intact.
|
|
// Returns the voided fact's ID and the new void-marker fact's ID.
|
|
// If no fact exists for the key, returns ErrNoFact.
|
|
func (s *Store) VoidLatestFact(ctx context.Context, key, source string, ts time.Time) (oldID, newID int64, err error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT id FROM facts
|
|
WHERE key = ?
|
|
AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL)
|
|
ORDER BY ts DESC, id DESC LIMIT 1`, key).Scan(&oldID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return 0, 0, ErrNoFact
|
|
}
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("void: find latest: %w", err)
|
|
}
|
|
res, err := tx.ExecContext(ctx,
|
|
`INSERT INTO facts (ts, kind, key, value, source, confidence, voids_id) VALUES (?,?,?,?,?,?,?)`,
|
|
ts.UnixMilli(), string(KindSelf), key, `"voided"`, source, 1.0,
|
|
sql.NullInt64{Int64: oldID, Valid: true})
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("void: write void-marker: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return 0, 0, err
|
|
}
|
|
newID, err = res.LastInsertId()
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("void: last insert id: %w", err)
|
|
}
|
|
// The other half of the repair (#470). A fact reaches recall through a
|
|
// vector keyed `fact:<key>:<unix>`, holding the utterance that wrote it.
|
|
// Voiding the row alone left that vector answering questions, so revert
|
|
// reported success on a box that stayed broken. Deleting every vector for
|
|
// the key covers the earlier rows too: their values are superseded, and a
|
|
// superseded value has no business claiming a turn.
|
|
//
|
|
// Best-effort by design: the audit trail is already committed, and a fact
|
|
// that is voided but still recallable is better than a void that failed.
|
|
if n, derr := s.VectorMemory().DeletePrefix(ctx, "fact:"+key+":"); derr != nil {
|
|
log.Printf("store: void %q: memory vectors survive: %v", key, derr)
|
|
} else if n > 0 {
|
|
log.Printf("store: void %q: dropped %d memory vector(s)", key, n)
|
|
}
|
|
return oldID, newID, nil
|
|
}
|
|
|
|
type rowScanner interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func scanFact(r rowScanner) (Fact, error) {
|
|
var f Fact
|
|
var tsMilli int64
|
|
var kind string
|
|
var voids sql.NullInt64
|
|
if err := r.Scan(&f.ID, &tsMilli, &kind, &f.Key, &f.Value, &f.Source, &f.Confidence, &voids); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Fact{}, ErrNoFact
|
|
}
|
|
return Fact{}, err
|
|
}
|
|
f.Ts = time.UnixMilli(tsMilli).UTC()
|
|
f.Kind = FactKind(kind)
|
|
f.VoidsID = voids
|
|
return f, nil
|
|
}
|
|
|
|
// placeholders renders n comma-separated SQL bind markers.
|
|
func placeholders(n int) string {
|
|
return strings.TrimSuffix(strings.Repeat("?,", n), ",")
|
|
}
|