From ebb3d4d36671d77287a39b08923b6a02846c2adc Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 23:19:55 +0330 Subject: [PATCH] Import source: internal/store/store Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality. --- internal/store/store.go | 320 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 320 insertions(+) create mode 100644 internal/store/store.go diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..81461d1 --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,320 @@ +// 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 +}