// Package store persists check results and incidents in SQLite. package store import ( "database/sql" "fmt" "sort" "time" _ "modernc.org/sqlite" ) type Store struct { db *sql.DB } // Check is one persisted probe result. type Check struct { TS int64 `json:"ts"` // unix ms OK bool `json:"ok"` LatencyMS float64 `json:"latency_ms"` Detail string `json:"detail,omitempty"` Err string `json:"error,omitempty"` } // Incident is one contiguous down period for a target. type Incident struct { ID int64 `json:"id"` TargetID string `json:"target_id"` StartedAt int64 `json:"started_at"` // unix ms EndedAt *int64 `json:"ended_at,omitempty"` // nil = ongoing Err string `json:"error,omitempty"` } // Bucket aggregates checks over a fixed window (e.g. one hour). type Bucket struct { Start int64 `json:"start"` // unix ms Total int `json:"total"` Failed int `json:"failed"` AvgLatencyMS float64 `json:"avg_latency_ms"` MaxLatencyMS float64 `json:"max_latency_ms"` } // LatencyStats summarises successful-check latency over a window. type LatencyStats struct { AvgMS float64 `json:"avg_ms"` P50MS float64 `json:"p50_ms"` P95MS float64 `json:"p95_ms"` MinMS float64 `json:"min_ms"` MaxMS float64 `json:"max_ms"` } func Open(path string) (*Store, error) { db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)") if err != nil { return nil, err } // modernc.org/sqlite is happiest with a single writer connection. db.SetMaxOpenConns(1) s := &Store{db: db} if err := s.migrate(); err != nil { db.Close() return nil, err } return s, nil } func (s *Store) migrate() error { _, err := s.db.Exec(` CREATE TABLE IF NOT EXISTS checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id TEXT NOT NULL, ts INTEGER NOT NULL, ok INTEGER NOT NULL, latency_ms REAL NOT NULL, detail TEXT NOT NULL DEFAULT '', error TEXT NOT NULL DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_checks_target_ts ON checks(target_id, ts); CREATE TABLE IF NOT EXISTS incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id TEXT NOT NULL, started_at INTEGER NOT NULL, ended_at INTEGER, error TEXT NOT NULL DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_incidents_target ON incidents(target_id, started_at); `) return err } func (s *Store) Close() error { return s.db.Close() } func (s *Store) InsertCheck(targetID string, c Check) error { _, err := s.db.Exec( `INSERT INTO checks (target_id, ts, ok, latency_ms, detail, error) VALUES (?, ?, ?, ?, ?, ?)`, targetID, c.TS, c.OK, c.LatencyMS, c.Detail, c.Err) return err } // History returns checks in [from, to], downsampled to at most maxPoints by // striding (keeps every Nth row, newest data intact). func (s *Store) History(targetID string, from, to int64, maxPoints int) ([]Check, error) { rows, err := s.db.Query( `SELECT ts, ok, latency_ms, detail, error FROM checks WHERE target_id = ? AND ts >= ? AND ts <= ? ORDER BY ts ASC`, targetID, from, to) if err != nil { return nil, err } defer rows.Close() var all []Check for rows.Next() { var c Check if err := rows.Scan(&c.TS, &c.OK, &c.LatencyMS, &c.Detail, &c.Err); err != nil { return nil, err } all = append(all, c) } if err := rows.Err(); err != nil { return nil, err } if maxPoints <= 0 || len(all) <= maxPoints { return all, nil } stride := (len(all) + maxPoints - 1) / maxPoints out := make([]Check, 0, maxPoints) for i := 0; i < len(all); i += stride { // Prefer surfacing a failure within the stride so incidents // survive downsampling. pick := all[i] for j := i; j < i+stride && j < len(all); j++ { if !all[j].OK { pick = all[j] break } } out = append(out, pick) } return out, nil } // Buckets aggregates checks in [from, to] into windows of bucketMS, // aligned to bucket boundaries. func (s *Store) Buckets(targetID string, from, to, bucketMS int64) ([]Bucket, error) { rows, err := s.db.Query( `SELECT (ts / ?) * ? AS bucket, COUNT(*), SUM(CASE WHEN ok = 0 THEN 1 ELSE 0 END), AVG(CASE WHEN ok = 1 THEN latency_ms END), MAX(CASE WHEN ok = 1 THEN latency_ms END) FROM checks WHERE target_id = ? AND ts >= ? AND ts <= ? GROUP BY bucket ORDER BY bucket ASC`, bucketMS, bucketMS, targetID, from, to) if err != nil { return nil, err } defer rows.Close() var out []Bucket for rows.Next() { var b Bucket var avg, max sql.NullFloat64 if err := rows.Scan(&b.Start, &b.Total, &b.Failed, &avg, &max); err != nil { return nil, err } b.AvgLatencyMS = avg.Float64 b.MaxLatencyMS = max.Float64 out = append(out, b) } return out, rows.Err() } // Uptime returns (ok, total) check counts in [from, now]. func (s *Store) Uptime(targetID string, from int64) (okCount, total int, err error) { err = s.db.QueryRow( `SELECT COALESCE(SUM(ok), 0), COUNT(*) FROM checks WHERE target_id = ? AND ts >= ?`, targetID, from).Scan(&okCount, &total) return } // Latency computes latency stats over successful checks in [from, now]. func (s *Store) Latency(targetID string, from int64) (LatencyStats, error) { rows, err := s.db.Query( `SELECT latency_ms FROM checks WHERE target_id = ? AND ts >= ? AND ok = 1 ORDER BY latency_ms ASC`, targetID, from) if err != nil { return LatencyStats{}, err } defer rows.Close() var vals []float64 var sum float64 for rows.Next() { var v float64 if err := rows.Scan(&v); err != nil { return LatencyStats{}, err } vals = append(vals, v) sum += v } if err := rows.Err(); err != nil { return LatencyStats{}, err } if len(vals) == 0 { return LatencyStats{}, nil } pct := func(p float64) float64 { idx := int(p * float64(len(vals)-1)) return vals[idx] } return LatencyStats{ AvgMS: sum / float64(len(vals)), P50MS: pct(0.50), P95MS: pct(0.95), MinMS: vals[0], MaxMS: vals[len(vals)-1], }, nil } func (s *Store) OpenIncident(targetID string, startedAt int64, errStr string) (int64, error) { res, err := s.db.Exec( `INSERT INTO incidents (target_id, started_at, error) VALUES (?, ?, ?)`, targetID, startedAt, errStr) if err != nil { return 0, err } return res.LastInsertId() } func (s *Store) CloseIncident(id, endedAt int64) error { _, err := s.db.Exec(`UPDATE incidents SET ended_at = ? WHERE id = ?`, endedAt, id) return err } // OngoingIncident returns the open incident for a target, or nil. func (s *Store) OngoingIncident(targetID string) (*Incident, error) { var inc Incident err := s.db.QueryRow( `SELECT id, target_id, started_at, error FROM incidents WHERE target_id = ? AND ended_at IS NULL ORDER BY started_at DESC LIMIT 1`, targetID).Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &inc.Err) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } return &inc, nil } // Incidents lists incidents overlapping [from, now], newest first, // ongoing ones always included and first. func (s *Store) Incidents(from int64) ([]Incident, error) { rows, err := s.db.Query( `SELECT id, target_id, started_at, ended_at, error FROM incidents WHERE ended_at IS NULL OR ended_at >= ? ORDER BY (ended_at IS NULL) DESC, started_at DESC LIMIT 200`, from) if err != nil { return nil, err } defer rows.Close() out := []Incident{} for rows.Next() { var inc Incident var ended sql.NullInt64 if err := rows.Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &ended, &inc.Err); err != nil { return nil, err } if ended.Valid { inc.EndedAt = &ended.Int64 } out = append(out, inc) } return out, rows.Err() } // IncidentsFor lists recent incidents for one target, newest first. func (s *Store) IncidentsFor(targetID string, limit int) ([]Incident, error) { rows, err := s.db.Query( `SELECT id, target_id, started_at, ended_at, error FROM incidents WHERE target_id = ? ORDER BY started_at DESC LIMIT ?`, targetID, limit) if err != nil { return nil, err } defer rows.Close() out := []Incident{} for rows.Next() { var inc Incident var ended sql.NullInt64 if err := rows.Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &ended, &inc.Err); err != nil { return nil, err } if ended.Valid { inc.EndedAt = &ended.Int64 } out = append(out, inc) } // Ongoing first, then newest. sort.SliceStable(out, func(i, j int) bool { return out[i].EndedAt == nil && out[j].EndedAt != nil }) return out, rows.Err() } // Prune deletes checks older than the retention window. func (s *Store) Prune(olderThan time.Time) error { if _, err := s.db.Exec(`DELETE FROM checks WHERE ts < ?`, olderThan.UnixMilli()); err != nil { return fmt.Errorf("prune checks: %w", err) } _, err := s.db.Exec(`DELETE FROM incidents WHERE ended_at IS NOT NULL AND ended_at < ?`, olderThan.UnixMilli()) return err }