BergaBruh
e6f1f63b8d
- Skip rows with null/malformed timestamps instead of silently re-stamping with time.Now() (data corruption hidden behind a fallback). Drops the time import. - Replace ad-hoc parentDir() with stdlib filepath.Dir(). - Add t.Fatal error checks to TestExtractRows_SkipsBadExtra (CREATE TABLE / INSERT) and switch sql.Open to checked form. - New TestExtractRows_SkipsBadTimestamp covers the null-ts path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
108 lines
2.6 KiB
Go
108 lines
2.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
|
|
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
|
|
}
|
|
|
|
if !ts.Valid {
|
|
log.Printf("skip id=%d: null timestamp", id)
|
|
continue
|
|
}
|
|
var tsUnix int64
|
|
if _, err := fmt.Sscanf(ts.String, "%d", &tsUnix); err != nil || tsUnix == 0 {
|
|
log.Printf("skip id=%d: bad timestamp %q", id, ts.String)
|
|
continue
|
|
}
|
|
|
|
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
|
|
}
|