115 lines
3.5 KiB
Go
115 lines
3.5 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"embed"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/bergabruh/stats-gateway/internal/probes"
|
|
_ "github.com/mattn/go-sqlite3"
|
|
)
|
|
|
|
//go:embed migrations/*.sql
|
|
var migrations embed.FS
|
|
|
|
// tsLayout is RFC3339 with fixed 9-digit nanosecond fraction. Lexical
|
|
// ordering of formatted strings preserves chronological order — unlike
|
|
// time.RFC3339Nano which strips trailing zeros (".1Z" lex-sorts after
|
|
// ".100000001Z" because 'Z' > '0'). SQLite TEXT columns are compared
|
|
// lexically, so this matters for QueryRange boundaries.
|
|
const tsLayout = "2006-01-02T15:04:05.000000000Z07:00"
|
|
|
|
// Store wraps a SQLite database for probe sample persistence.
|
|
type Store struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
// Open opens (or creates) the SQLite database at path and runs migrations.
|
|
func Open(path string) (*Store, error) {
|
|
db, err := sql.Open("sqlite3", path+"?_journal_mode=WAL&_busy_timeout=5000")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s := &Store{db: db}
|
|
if err := s.migrate(); err != nil {
|
|
return nil, fmt.Errorf("migrate: %w", err)
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// Close closes the underlying database.
|
|
func (s *Store) Close() error { return s.db.Close() }
|
|
|
|
func (s *Store) migrate() error {
|
|
body, err := migrations.ReadFile("migrations/001_init.sql")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = s.db.Exec(string(body))
|
|
return err
|
|
}
|
|
|
|
// InsertSample stores a probe result for the given node.
|
|
func (s *Store) InsertSample(ctx context.Context, node string, r *probes.Result) error {
|
|
_, err := s.db.ExecContext(ctx, `
|
|
INSERT INTO samples (node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
`, node, r.Timestamp.Format(tsLayout),
|
|
r.HTTPLatencyMs, r.TCPRttMs, r.DNSResolveMs, r.TLSHandshakeMs, r.ThroughputKbps)
|
|
return err
|
|
}
|
|
|
|
// QueryRecent returns up to limit samples for node ordered by ts DESC.
|
|
// If since is zero, no lower bound is applied. If node is empty, all nodes are returned.
|
|
func (s *Store) QueryRecent(ctx context.Context, node string, limit int, since time.Time) ([]probes.Result, error) {
|
|
q := `SELECT node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps
|
|
FROM samples WHERE 1=1`
|
|
args := []any{}
|
|
if node != "" {
|
|
q += ` AND node = ?`
|
|
args = append(args, node)
|
|
}
|
|
if !since.IsZero() {
|
|
q += ` AND ts >= ?`
|
|
args = append(args, since.Format(tsLayout))
|
|
}
|
|
q += ` ORDER BY ts DESC LIMIT ?`
|
|
args = append(args, limit)
|
|
|
|
return s.scanSamples(ctx, q, args...)
|
|
}
|
|
|
|
// QueryRange returns samples for [since, until). Used by Aggregate (Task 5).
|
|
func (s *Store) QueryRange(ctx context.Context, node string, since, until time.Time) ([]probes.Result, error) {
|
|
q := `SELECT node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps
|
|
FROM samples WHERE ts >= ? AND ts < ?`
|
|
args := []any{since.Format(tsLayout), until.Format(tsLayout)}
|
|
if node != "" {
|
|
q += ` AND node = ?`
|
|
args = append(args, node)
|
|
}
|
|
q += ` ORDER BY ts ASC`
|
|
return s.scanSamples(ctx, q, args...)
|
|
}
|
|
|
|
func (s *Store) scanSamples(ctx context.Context, q string, args ...any) ([]probes.Result, error) {
|
|
rows, err := s.db.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []probes.Result
|
|
for rows.Next() {
|
|
var ts string
|
|
var r probes.Result
|
|
if err := rows.Scan(&r.Node, &ts, &r.HTTPLatencyMs, &r.TCPRttMs, &r.DNSResolveMs, &r.TLSHandshakeMs, &r.ThroughputKbps); err != nil {
|
|
return nil, err
|
|
}
|
|
r.Timestamp, _ = time.Parse(tsLayout, ts)
|
|
out = append(out, r)
|
|
}
|
|
return out, rows.Err()
|
|
}
|