diff --git a/internal/storage/migrations/002_speedtests.sql b/internal/storage/migrations/002_speedtests.sql new file mode 100644 index 0000000..d6bc57f --- /dev/null +++ b/internal/storage/migrations/002_speedtests.sql @@ -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); diff --git a/internal/storage/speedtests.go b/internal/storage/speedtests.go new file mode 100644 index 0000000..fee90bc --- /dev/null +++ b/internal/storage/speedtests.go @@ -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() +} diff --git a/internal/storage/speedtests_test.go b/internal/storage/speedtests_test.go new file mode 100644 index 0000000..2cbd684 --- /dev/null +++ b/internal/storage/speedtests_test.go @@ -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)) + } +} diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index fc4eb90..de0d498 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -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.