diff --git a/Makefile b/Makefile index ec025a7..50a72d2 100644 --- a/Makefile +++ b/Makefile @@ -5,6 +5,7 @@ build: go build -o bin/stats-collector ./cmd/collector go build -o bin/stats-snapshot ./cmd/snapshot go build -o bin/stats-issue-token ./cmd/issue-token + CGO_ENABLED=0 go build -o bin/speedtest-sync ./cmd/speedtest-sync test: go test ./... diff --git a/cmd/speedtest-sync/main.go b/cmd/speedtest-sync/main.go new file mode 100644 index 0000000..a6429ba --- /dev/null +++ b/cmd/speedtest-sync/main.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "database/sql" + "flag" + "log" + "os" + "strconv" + "time" + + pb "github.com/bergabruh/stats-gateway/gen/stats/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + + _ "modernc.org/sqlite" +) + +func main() { + dbPath := flag.String("source-db", "/var/lib/librespeed-db/speedtest_telemetry.db", + "Path to LibreSpeed SQLite (read-only)") + statsAddr := flag.String("stats-addr", "stats.projectshitpost.fun:443", + "stats-gateway gRPC address") + cursorPath := flag.String("cursor", "/var/lib/speedtest-sync/cursor", + "File holding the last synced source ID") + tokenEnv := flag.String("token-env", "JWT_TOKEN", + "Environment variable with JWT token") + insecureFlag := flag.Bool("insecure", false, "Use insecure gRPC (testing only)") + flag.Parse() + + token := os.Getenv(*tokenEnv) + if token == "" { + log.Fatalf("env %s is empty", *tokenEnv) + } + + cursor := readCursor(*cursorPath) + + db, err := sql.Open("sqlite", "file:"+*dbPath+"?mode=ro") + if err != nil { + log.Fatalf("open source db: %v", err) + } + defer db.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + rows, err := extractRows(ctx, db, cursor) + if err != nil { + log.Fatalf("extract: %v", err) + } + if len(rows) == 0 { + log.Printf("no new rows since id=%d", cursor) + return + } + + var dialOpt grpc.DialOption + if *insecureFlag { + dialOpt = grpc.WithTransportCredentials(insecure.NewCredentials()) + } else { + dialOpt = grpc.WithTransportCredentials(credentials.NewTLS(nil)) + } + conn, err := grpc.NewClient(*statsAddr, dialOpt) + if err != nil { + log.Fatalf("grpc dial: %v", err) + } + defer conn.Close() + + client := pb.NewSpeedStatusClient(conn) + + authCtx := metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+token) + sent := 0 + for _, r := range rows { + if err := sendRow(authCtx, client, r); err != nil { + log.Printf("send id=%d failed: %v (stopping; cursor not advanced)", r.SourceID, err) + break + } + cursor = r.SourceID + sent++ + } + + if err := writeCursor(*cursorPath, cursor); err != nil { + log.Printf("WARN: cursor write failed: %v", err) + } + log.Printf("synced %d rows, cursor=%d", sent, cursor) +} + +func readCursor(path string) int64 { + b, err := os.ReadFile(path) + if err != nil { + return 0 + } + n, err := strconv.ParseInt(string(b), 10, 64) + if err != nil { + return 0 + } + return n +} + +func writeCursor(path string, id int64) error { + if err := os.MkdirAll(parentDir(path), 0o755); err != nil { + return err + } + return os.WriteFile(path, []byte(strconv.FormatInt(id, 10)), 0o600) +} + +func parentDir(p string) string { + for i := len(p) - 1; i >= 0; i-- { + if p[i] == '/' { + return p[:i] + } + } + return "." +} diff --git a/cmd/speedtest-sync/sync.go b/cmd/speedtest-sync/sync.go new file mode 100644 index 0000000..99aa298 --- /dev/null +++ b/cmd/speedtest-sync/sync.go @@ -0,0 +1,106 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log" + "time" + + pb "github.com/bergabruh/stats-gateway/gen/stats/v1" +) + +type extractedRow struct { + SourceID int64 + Timestamp int64 + Node string + Country string + DlMbps float64 + UlMbps float64 + PingMs float64 + JitterMs float64 +} + +// extractRows reads rows from the LibreSpeed source DB whose id > sinceID and +// returns a privacy-filtered projection. The SELECT intentionally touches only +// id, timestamp, extra, dl, ul, ping, jitter — never ip, ispinfo, ua, lang, log. +func extractRows(ctx context.Context, db *sql.DB, sinceID int64) ([]extractedRow, error) { + rows, err := db.QueryContext(ctx, ` + SELECT id, strftime('%s', timestamp), extra, dl, ul, ping, jitter + FROM speedtest_users + WHERE id > ? AND extra IS NOT NULL AND extra != '' + ORDER BY id ASC`, sinceID) + if err != nil { + return nil, fmt.Errorf("query source: %w", err) + } + defer rows.Close() + + out := []extractedRow{} + for rows.Next() { + var ( + id int64 + ts sql.NullString + extraJSON sql.NullString + dl, ul, ping, jitter sql.NullFloat64 + ) + if err := rows.Scan(&id, &ts, &extraJSON, &dl, &ul, &ping, &jitter); err != nil { + return nil, err + } + var meta struct { + Node string `json:"node"` + Country string `json:"country"` + } + if !extraJSON.Valid { + continue + } + if err := json.Unmarshal([]byte(extraJSON.String), &meta); err != nil { + log.Printf("skip id=%d: bad extra json: %v", id, err) + continue + } + if meta.Node != "fl" && meta.Node != "pl1" { + log.Printf("skip id=%d: bad node %q", id, meta.Node) + continue + } + + var tsUnix int64 + if ts.Valid { + fmt.Sscanf(ts.String, "%d", &tsUnix) + } + if tsUnix == 0 { + tsUnix = time.Now().Unix() + } + + out = append(out, extractedRow{ + SourceID: id, + Timestamp: tsUnix, + Node: meta.Node, + Country: nonEmpty(meta.Country, "XX"), + DlMbps: dl.Float64, + UlMbps: ul.Float64, + PingMs: ping.Float64, + JitterMs: jitter.Float64, + }) + } + return out, rows.Err() +} + +func nonEmpty(s, fallback string) string { + if s == "" { + return fallback + } + return s +} + +func sendRow(ctx context.Context, client pb.SpeedStatusClient, r extractedRow) error { + _, err := client.RecordSpeedTest(ctx, &pb.RecordSpeedTestRequest{ + Timestamp: r.Timestamp, + Node: r.Node, + Country: r.Country, + DlMbps: r.DlMbps, + UlMbps: r.UlMbps, + PingMs: r.PingMs, + JitterMs: r.JitterMs, + }) + return err +} diff --git a/cmd/speedtest-sync/sync_test.go b/cmd/speedtest-sync/sync_test.go new file mode 100644 index 0000000..8867365 --- /dev/null +++ b/cmd/speedtest-sync/sync_test.go @@ -0,0 +1,74 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "path/filepath" + "testing" + + _ "modernc.org/sqlite" +) + +func TestExtractRows_FiltersExtraJSON(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "test.db") + db, err := sql.Open("sqlite", dbPath) + if err != nil { + t.Fatal(err) + } + defer db.Close() + + _, err = db.Exec(` + CREATE TABLE speedtest_users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp DATETIME, ip TEXT, ispinfo TEXT, extra TEXT, + ua TEXT, lang TEXT, dl REAL, ul REAL, ping REAL, jitter REAL, log TEXT + )`) + if err != nil { + t.Fatal(err) + } + + extra, _ := json.Marshal(map[string]string{"node": "fl", "country": "RU"}) + _, err = db.Exec(`INSERT INTO speedtest_users(timestamp, ip, ispinfo, extra, dl, ul, ping, jitter) + VALUES(datetime('now'), '0.0.0.0', '', ?, 100.5, 20.1, 35.2, 2.4)`, string(extra)) + if err != nil { + t.Fatal(err) + } + + rows, err := extractRows(context.Background(), db, 0) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 { + t.Fatalf("expected 1 row, got %d", len(rows)) + } + if rows[0].Node != "fl" || rows[0].Country != "RU" { + t.Errorf("extract mismatch: %+v", rows[0]) + } + if rows[0].DlMbps != 100.5 { + t.Errorf("dl mismatch: %v", rows[0].DlMbps) + } +} + +func TestExtractRows_SkipsBadExtra(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "test.db") + db, _ := sql.Open("sqlite", dbPath) + defer db.Close() + + db.Exec(`CREATE TABLE speedtest_users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp DATETIME, ip TEXT, ispinfo TEXT, extra TEXT, + ua TEXT, lang TEXT, dl REAL, ul REAL, ping REAL, jitter REAL, log TEXT)`) + db.Exec(`INSERT INTO speedtest_users(timestamp, extra, dl, ul, ping, jitter) + VALUES(datetime('now'), 'not-json', 1, 1, 1, 1)`) + + rows, err := extractRows(context.Background(), db, 0) + if err != nil { + t.Fatal(err) + } + if len(rows) != 0 { + t.Errorf("expected 0 rows (bad extra skipped), got %d", len(rows)) + } +} diff --git a/go.mod b/go.mod index 3765fd2..2cb8e85 100644 --- a/go.mod +++ b/go.mod @@ -11,8 +11,17 @@ require ( ) require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect golang.org/x/net v0.51.0 // indirect golang.org/x/sys v0.42.0 // indirect golang.org/x/text v0.34.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect + modernc.org/libc v1.72.0 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.11.0 // indirect + modernc.org/sqlite v1.50.0 // indirect ) diff --git a/go.sum b/go.sum index 7673f52..c4effdb 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -10,8 +12,14 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-sqlite3 v1.14.44 h1:3VSe+xafpbzsLbdr2AWlAZk9yRHiBhTBakioXaCKTF8= github.com/mattn/go-sqlite3 v1.14.44/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= @@ -26,6 +34,7 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= @@ -44,3 +53,11 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/libc v1.72.0 h1:IEu559v9a0XWjw0DPoVKtXpO2qt5NVLAnFaBbjq+n8c= +modernc.org/libc v1.72.0/go.mod h1:tTU8DL8A+XLVkEY3x5E/tO7s2Q/q42EtnNWda/L5QhQ= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/sqlite v1.50.0 h1:eMowQSWLK0MeiQTdmz3lqoF5dqclujdlIKeJA11+7oM= +modernc.org/sqlite v1.50.0/go.mod h1:m0w8xhwYUVY3H6pSDwc3gkJ/irZT/0YEXwBlhaxQEew=