storage: add speedtests table + Save/List/Delete + retention

Adds a Store-method API for the RecordSpeedTest RPC (Task 8) and the retention sweep (Task 14): - SaveSpeedTest / ListRecentSpeedTests / DeleteSpeedTestsBefore - SpeedTest struct (no json tags; that decision deferred to Task 10) Schema lives in migrations/002_speedtests.sql to match the existing embed.FS pattern (rather than inline in Go). The migrate() loop now iterates every *.sql in lexical order, so future migrations drop in without code changes. Both files use CREATE ... IF NOT EXISTS so the existing 001 migration remains idempotent on already-initialised DBs. CHECK constraints enforce node IN ('fl','pl1') and non-negative metric values at the DB level. country has no DB CHECK — validation belongs to the gRPC handler in Task 8. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-09 18:30:17 +05:00
parent ff85f21f5d
commit d1692842b3
4 changed files with 181 additions and 3 deletions
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS speedtests (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp INTEGER NOT NULL,
+    node TEXT NOT NULL CHECK(node IN ('fl', 'pl1')),
+    country TEXT NOT NULL,
+    dl_mbps REAL NOT NULL CHECK(dl_mbps >= 0),
+    ul_mbps REAL NOT NULL CHECK(ul_mbps >= 0),
+    ping_ms REAL NOT NULL CHECK(ping_ms >= 0),
+    jitter_ms REAL NOT NULL CHECK(jitter_ms >= 0)
+);
+
+CREATE INDEX IF NOT EXISTS idx_speedtests_ts ON speedtests(timestamp);
+CREATE INDEX IF NOT EXISTS idx_speedtests_node ON speedtests(node);
@@ -0,0 +1,65 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+)
+
+// SpeedTest is a single client-side speedtest result reported via the
+// RecordSpeedTest RPC. Timestamp is unix seconds. Node is 'fl' or 'pl1'
+// (DB-level CHECK). Country is ISO-3166-1 alpha-2 (validated at the
+// gRPC handler, not at the DB).
+type SpeedTest struct {
+	ID        int64
+	Timestamp int64
+	Node      string
+	Country   string
+	DlMbps    float64
+	UlMbps    float64
+	PingMs    float64
+	JitterMs  float64
+}
+
+// SaveSpeedTest inserts a speedtest row and returns the new id.
+func (s *Store) SaveSpeedTest(ctx context.Context, st SpeedTest) (int64, error) {
+	res, err := s.db.ExecContext(ctx, `
+		INSERT INTO speedtests(timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms)
+		VALUES(?, ?, ?, ?, ?, ?, ?)`,
+		st.Timestamp, st.Node, st.Country, st.DlMbps, st.UlMbps, st.PingMs, st.JitterMs)
+	if err != nil {
+		return 0, fmt.Errorf("insert speedtest: %w", err)
+	}
+	return res.LastInsertId()
+}
+
+// ListRecentSpeedTests returns up to limit speedtests ordered by id DESC
+// (newest first).
+func (s *Store) ListRecentSpeedTests(ctx context.Context, limit int) ([]SpeedTest, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT id, timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms
+		FROM speedtests ORDER BY id DESC LIMIT ?`, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []SpeedTest{}
+	for rows.Next() {
+		var st SpeedTest
+		if err := rows.Scan(&st.ID, &st.Timestamp, &st.Node, &st.Country,
+			&st.DlMbps, &st.UlMbps, &st.PingMs, &st.JitterMs); err != nil {
+			return nil, err
+		}
+		out = append(out, st)
+	}
+	return out, rows.Err()
+}
+
+// DeleteSpeedTestsBefore deletes rows with timestamp < ts and returns the
+// number of rows removed. Used by the retention sweep (Task 14).
+func (s *Store) DeleteSpeedTestsBefore(ctx context.Context, ts int64) (int64, error) {
+	res, err := s.db.ExecContext(ctx, `DELETE FROM speedtests WHERE timestamp < ?`, ts)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
@@ -0,0 +1,80 @@
+package storage
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// newSpeedTestDB mirrors the open pattern from sqlite_test.go since the
+// storage package has no dedicated test-DB helper — every test calls
+// Open(t.TempDir()/...) directly.
+func newSpeedTestDB(t *testing.T) *Store {
+	t.Helper()
+	dir := t.TempDir()
+	st, err := Open(filepath.Join(dir, "test.db"))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+func TestSpeedTest_SaveAndList(t *testing.T) {
+	db := newSpeedTestDB(t)
+	ctx := context.Background()
+
+	now := time.Now().Unix()
+	id, err := db.SaveSpeedTest(ctx, SpeedTest{
+		Timestamp: now, Node: "fl", Country: "RU",
+		DlMbps: 100.5, UlMbps: 20.1, PingMs: 35.2, JitterMs: 2.4,
+	})
+	if err != nil {
+		t.Fatalf("SaveSpeedTest: %v", err)
+	}
+	if id <= 0 {
+		t.Fatalf("expected positive id, got %d", id)
+	}
+
+	rows, err := db.ListRecentSpeedTests(ctx, 10)
+	if err != nil {
+		t.Fatalf("ListRecentSpeedTests: %v", err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("expected 1 row, got %d", len(rows))
+	}
+	if rows[0].Country != "RU" || rows[0].Node != "fl" {
+		t.Errorf("row mismatch: %+v", rows[0])
+	}
+}
+
+func TestSpeedTest_RetentionDelete(t *testing.T) {
+	db := newSpeedTestDB(t)
+	ctx := context.Background()
+
+	old := time.Now().Add(-31 * 24 * time.Hour).Unix()
+	fresh := time.Now().Unix()
+
+	for _, ts := range []int64{old, fresh} {
+		if _, err := db.SaveSpeedTest(ctx, SpeedTest{
+			Timestamp: ts, Node: "fl", Country: "RU",
+			DlMbps: 1, UlMbps: 1, PingMs: 1, JitterMs: 1,
+		}); err != nil {
+			t.Fatalf("save: %v", err)
+		}
+	}
+
+	deleted, err := db.DeleteSpeedTestsBefore(ctx, time.Now().Add(-30*24*time.Hour).Unix())
+	if err != nil {
+		t.Fatalf("DeleteSpeedTestsBefore: %v", err)
+	}
+	if deleted != 1 {
+		t.Errorf("expected 1 deleted row, got %d", deleted)
+	}
+
+	rows, _ := db.ListRecentSpeedTests(ctx, 10)
+	if len(rows) != 1 {
+		t.Errorf("expected 1 remaining row, got %d", len(rows))
+	}
+}
@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"embed"
 	"fmt"
+	"sort"
 	"time"

 	"github.com/bergabruh/stats-gateway/internal/probes"
@@ -42,13 +43,32 @@ func Open(path string) (*Store, error) {
 // Close closes the underlying database.
 func (s *Store) Close() error { return s.db.Close() }

+// migrate applies every SQL file under migrations/ in lexical filename order.
+// Each file is executed wholesale (statements are CREATE ... IF NOT EXISTS,
+// so re-applying on an existing DB is a no-op).
 func (s *Store) migrate() error {
-	body, err := migrations.ReadFile("migrations/001_init.sql")
+	entries, err := migrations.ReadDir("migrations")
 	if err != nil {
 		return err
 	}
-	_, err = s.db.Exec(string(body))
-	return err
+	names := make([]string, 0, len(entries))
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+		names = append(names, e.Name())
+	}
+	sort.Strings(names)
+	for _, name := range names {
+		body, err := migrations.ReadFile("migrations/" + name)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", name, err)
+		}
+		if _, err := s.db.Exec(string(body)); err != nil {
+			return fmt.Errorf("apply migration %s: %w", name, err)
+		}
+	}
+	return nil
 }

 // InsertSample stores a probe result for the given node.