fix(chdb): return error on log_scores row scan failure

Previously a per-row scan error was logged and skipped with continue, so a
decode failure (e.g. the UInt64->int64 regression) produced a 200 response
with partial or empty data and no signal to the client. Fail the request
instead so the HTTP handlers return 500. Also check rows.Err() after iteration.

.woodpecker.yaml

@@ -58,7 +58,7 @@ steps:
     depends_on: [test]
 
   - name: generate-tags
-    image: ghcr.io/abh/woodpecker-docker-tags-plugin:sha-8a3bd7c
+    image: ghcr.io/abh/woodpecker-docker-tags-plugin:0.2.1-abh
     settings:
       tags: |
         branch

chdb/logscores.go

@@ -2,16 +2,62 @@ package chdb
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"strings"
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/jackc/pgx/v5/pgtype"
 	"go.ntppool.org/common/logger"
 	"go.ntppool.org/common/tracing"
 	"go.ntppool.org/data-api/ntpdb"
 )
 
+// scanLogScore reads one ClickHouse log_scores row into an ntpdb.LogScore.
+// The ClickHouse columns are unsigned (id UInt64, monitor_id/server_id UInt32,
+// rtt UInt32), but the PostgreSQL-generated model uses signed/pgtype fields, so
+// the row is scanned into locals matching the ClickHouse types and then converted.
+func scanLogScore(rows interface{ Scan(...any) error }) (ntpdb.LogScore, error) {
+	row := ntpdb.LogScore{}
+
+	var (
+		id        uint64
+		monitorID uint32
+		serverID  uint32
+		ts        time.Time
+		offset    sql.NullFloat64
+		rtt       sql.NullInt32
+		leap      uint8
+	)
+
+	if err := rows.Scan(
+		&id,
+		&monitorID,
+		&serverID,
+		&ts,
+		&row.Score,
+		&row.Step,
+		&offset,
+		&rtt,
+		&leap,
+		&row.Attributes.Warning,
+		&row.Attributes.Error,
+	); err != nil {
+		return row, err
+	}
+
+	row.ID = int64(id)
+	row.MonitorID = pgtype.Int8{Int64: int64(monitorID), Valid: true}
+	row.ServerID = int64(serverID)
+	row.Ts = pgtype.Timestamptz{Time: ts, Valid: true}
+	row.Offset = pgtype.Float8{Float64: offset.Float64, Valid: offset.Valid}
+	row.Rtt = pgtype.Int4{Int32: rtt.Int32, Valid: rtt.Valid}
+	row.Attributes.Leap = int8(leap)
+
+	return row, nil
+}
+
 func (d *ClickHouse) Logscores(ctx context.Context, serverID, monitorID int, since time.Time, limit int, fullHistory bool) ([]ntpdb.LogScore, error) {
 	log := logger.Setup()
 	ctx, span := tracing.Tracer().Start(ctx, "CH Logscores")
@@ -74,35 +120,21 @@ func (d *ClickHouse) Logscores(ctx context.Context, serverID, monitorID int, sin
 	rv := []ntpdb.LogScore{}
 
 	for rows.Next() {
-
+		row, err := scanLogScore(rows)
-		row := ntpdb.LogScore{}
+		if err != nil {
-
+			// Fail the whole request so the caller returns a 500 rather than
-		var leap uint8
+			// silently serving partial/empty data. If this turns into frequent
-
+			// user-facing errors, we could tolerate a few (or a small
-		if err := rows.Scan(
+			// percentage of) row errors per request when the rest is good data.
-			&row.ID,
+			log.ErrorContext(ctx, "could not parse row", "err", err)
-			&row.MonitorID,
+			return nil, fmt.Errorf("could not parse log_scores row: %w", err)
-			&row.ServerID,
-			&row.Ts,
-			&row.Score,
-			&row.Step,
-			&row.Offset,
-			&row.Rtt,
-			&leap,
-			&row.Attributes.Warning,
-			&row.Attributes.Error,
-		); err != nil {
-			log.Error("could not parse row", "err", err)
-			continue
 		}
-
-		row.Attributes.Leap = int8(leap)
-
 		rv = append(rv, row)
-
 	}
-
+	if err := rows.Err(); err != nil {
-	// log.InfoContext(ctx, "returning data", "rv", rv)
+		log.ErrorContext(ctx, "rows error", "err", err)
+		return nil, fmt.Errorf("database error")
+	}
 
 	return rv, nil
 }
@@ -138,7 +170,8 @@ func (d *ClickHouse) LogscoresTimeRange(ctx context.Context, serverID, monitorID
 		args = append(args, limit)
 	}
 
-	log.DebugContext(ctx, "clickhouse time range query",
+	log.DebugContext(
+		ctx, "clickhouse time range query",
 		"query", query,
 		"args", args,
 		"server_id", serverID,
@@ -181,31 +214,24 @@ func (d *ClickHouse) LogscoresTimeRange(ctx context.Context, serverID, monitorID
 	rv := []ntpdb.LogScore{}
 
 	for rows.Next() {
-		row := ntpdb.LogScore{}
+		row, err := scanLogScore(rows)
-		var leap uint8
+		if err != nil {
-
+			// Fail the whole request so the caller returns a 500 rather than
-		if err := rows.Scan(
+			// silently serving partial/empty data. If this turns into frequent
-			&row.ID,
+			// user-facing errors, we could tolerate a few (or a small
-			&row.MonitorID,
+			// percentage of) row errors per request when the rest is good data.
-			&row.ServerID,
+			log.ErrorContext(ctx, "could not parse row", "err", err)
-			&row.Ts,
+			return nil, fmt.Errorf("could not parse log_scores row: %w", err)
-			&row.Score,
-			&row.Step,
-			&row.Offset,
-			&row.Rtt,
-			&leap,
-			&row.Attributes.Warning,
-			&row.Attributes.Error,
-		); err != nil {
-			log.Error("could not parse row", "err", err)
-			continue
 		}
-
-		row.Attributes.Leap = int8(leap)
 		rv = append(rv, row)
 	}
+	if err := rows.Err(); err != nil {
+		log.ErrorContext(ctx, "rows error", "err", err)
+		return nil, fmt.Errorf("database error")
+	}
 
-	log.InfoContext(ctx, "time range query results",
+	log.InfoContext(
+		ctx, "time range query results",
 		"rows_returned", len(rv),
 		"server_id", serverID,
 		"monitor_id", monitorID,
@@ -220,7 +246,7 @@ func (d *ClickHouse) LogscoresTimeRange(ctx context.Context, serverID, monitorID
 				samples = append(samples, map[string]interface{}{
 					"id":           row.ID,
 					"monitor_id":   row.MonitorID,
-					"ts":           row.Ts.Format(time.RFC3339),
+					"ts":           row.Ts.Time.Format(time.RFC3339),
 					"score":        row.Score,
 					"rtt_valid":    row.Rtt.Valid,
 					"offset_valid": row.Offset.Valid,

go.mod

@@ -15,13 +15,13 @@ tool (
 require (
 	dario.cat/mergo v1.0.2
 	github.com/ClickHouse/clickhouse-go/v2 v2.40.3
-	github.com/go-sql-driver/mysql v1.9.3
 	github.com/hashicorp/go-retryablehttp v0.7.8
+	github.com/jackc/pgx/v5 v5.7.6
 	github.com/labstack/echo-contrib v0.17.4
 	github.com/labstack/echo/v4 v4.13.4
 	github.com/samber/slog-echo v1.17.2
 	github.com/spf13/cobra v1.10.1
-	go.ntppool.org/common v0.5.2
+	go.ntppool.org/common v0.6.3-0.20251129195245-283d3936f6d0
 	go.opentelemetry.io/contrib/instrumentation/github.com/labstack/echo/otelecho v0.63.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.63.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0
@@ -52,6 +52,7 @@ require (
 	github.com/go-faster/errors v0.7.1 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-sql-driver/mysql v1.9.3 // indirect
 	github.com/google/cel-go v0.25.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
@@ -62,7 +63,6 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
-	github.com/jackc/pgx/v5 v5.7.5 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
@@ -137,7 +137,7 @@ require (
 	golang.org/x/sys v0.36.0 // indirect
 	golang.org/x/text v0.29.0 // indirect
 	golang.org/x/time v0.13.0 // indirect
-	golang.org/x/tools v0.36.0 // indirect
+	golang.org/x/tools v0.37.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250922171735-9219d122eba9 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250922171735-9219d122eba9 // indirect
 	google.golang.org/grpc v1.75.1 // indirect

go.sum

@@ -92,8 +92,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.7.5 h1:JHGfMnQY+IEtGM63d+NGMjoRpysB2JBwDr5fsngwmJs=
+github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk=
-github.com/jackc/pgx/v5 v5.7.5/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
+github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
@@ -233,8 +233,8 @@ github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7Jul
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g=
-go.ntppool.org/common v0.5.2 h1:Ijlezhiqqs7TJYZTWwEwultLFxhNaXsh6DkaO53m/F4=
+go.ntppool.org/common v0.6.3-0.20251129195245-283d3936f6d0 h1:Vbs/RgrwfdA9ZzGAkhFRaU7ZSEl8D28pk95iYhjzvyA=
-go.ntppool.org/common v0.5.2/go.mod h1:e5ohROK9LdZZTI1neNiSlmgmWC23F779qzLvSi4JzyI=
+go.ntppool.org/common v0.6.3-0.20251129195245-283d3936f6d0/go.mod h1:Dkc2P5+aaCseC/cs0uD9elh4yTllqvyeZ1NNT/G/414=
 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/contrib/bridges/otelslog v0.13.0 h1:bwnLpizECbPr1RrQ27waeY2SPIPeccCx/xLuoYADZ9s=
@@ -363,8 +363,8 @@ golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.36.0 h1:kWS0uv/zsvHEle1LbV5LE8QujrxB3wfQyxHfhOk0Qkg=
+golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE=
-golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s=
+golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

logscores/history.go

@@ -2,9 +2,10 @@ package logscores
 
 import (
 	"context"
-	"database/sql"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"go.ntppool.org/common/logger"
 	"go.ntppool.org/common/tracing"
 	"go.ntppool.org/data-api/chdb"
@@ -19,12 +20,12 @@ type LogScoreHistory struct {
 	// MonitorIDs []uint32
 }
 
-func GetHistoryClickHouse(ctx context.Context, ch *chdb.ClickHouse, db *sql.DB, serverID, monitorID uint32, since time.Time, count int, fullHistory bool) (*LogScoreHistory, error) {
+func GetHistoryClickHouse(ctx context.Context, ch *chdb.ClickHouse, db *pgxpool.Pool, serverID, monitorID int64, since time.Time, count int, fullHistory bool) (*LogScoreHistory, error) {
 	log := logger.FromContext(ctx)
 	ctx, span := tracing.Tracer().Start(ctx, "logscores.GetHistoryClickHouse",
 		trace.WithAttributes(
-			attribute.Int("server", int(serverID)),
+			attribute.Int64("server", serverID),
-			attribute.Int("monitor", int(monitorID)),
+			attribute.Int64("monitor", monitorID),
 			attribute.Bool("full_history", fullHistory),
 		),
 	)
@@ -51,17 +52,17 @@ func GetHistoryClickHouse(ctx context.Context, ch *chdb.ClickHouse, db *sql.DB,
 	}, nil
 }
 
-func GetHistoryMySQL(ctx context.Context, db *sql.DB, serverID, monitorID uint32, since time.Time, count int) (*LogScoreHistory, error) {
+func GetHistoryPostgres(ctx context.Context, db *pgxpool.Pool, serverID, monitorID int64, since time.Time, count int) (*LogScoreHistory, error) {
 	log := logger.FromContext(ctx)
-	ctx, span := tracing.Tracer().Start(ctx, "logscores.GetHistoryMySQL")
+	ctx, span := tracing.Tracer().Start(ctx, "logscores.GetHistoryPostgres")
 	defer span.End()
 
 	span.SetAttributes(
-		attribute.Int("server", int(serverID)),
+		attribute.Int64("server", serverID),
-		attribute.Int("monitor", int(monitorID)),
+		attribute.Int64("monitor", monitorID),
 	)
 
-	log.Debug("GetHistoryMySQL", "server", serverID, "monitor", monitorID, "since", since, "count", count)
+	log.Debug("GetHistoryPostgres", "server", serverID, "monitor", monitorID, "since", since, "count", count)
 
 	q := ntpdb.NewWrappedQuerier(ntpdb.New(db))
 
@@ -69,13 +70,13 @@ func GetHistoryMySQL(ctx context.Context, db *sql.DB, serverID, monitorID uint32
 	var err error
 	if monitorID > 0 {
 		ls, err = q.GetServerLogScoresByMonitorID(ctx, ntpdb.GetServerLogScoresByMonitorIDParams{
-			ServerID:  serverID,
+			ServerID:  int64(serverID),
-			MonitorID: sql.NullInt32{Int32: int32(monitorID), Valid: true},
+			MonitorID: pgtype.Int8{Int64: int64(monitorID), Valid: true},
 			Limit:     int32(count),
 		})
 	} else {
 		ls, err = q.GetServerLogScores(ctx, ntpdb.GetServerLogScoresParams{
-			ServerID: serverID,
+			ServerID: int64(serverID),
 			Limit:    int32(count),
 		})
 	}
@@ -97,12 +98,12 @@ func GetHistoryMySQL(ctx context.Context, db *sql.DB, serverID, monitorID uint32
 
 func getMonitorNames(ctx context.Context, ls []ntpdb.LogScore, q ntpdb.QuerierTx) (map[int]string, error) {
 	monitors := map[int]string{}
-	monitorIDs := []uint32{}
+	monitorIDs := []int64{}
 	for _, l := range ls {
 		if !l.MonitorID.Valid {
 			continue
 		}
-		mID := uint32(l.MonitorID.Int32)
+		mID := l.MonitorID.Int64
 		if _, ok := monitors[int(mID)]; !ok {
 			monitors[int(mID)] = ""
 			monitorIDs = append(monitorIDs, mID)

ntpdb/db.go

@@ -6,14 +6,15 @@ package ntpdb
 
 import (
 	"context"
-	"database/sql"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 )
 
 type DBTX interface {
-	ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
+	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
-	PrepareContext(context.Context, string) (*sql.Stmt, error)
+	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
-	QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
+	QueryRow(context.Context, string, ...interface{}) pgx.Row
-	QueryRowContext(context.Context, string, ...interface{}) *sql.Row
 }
 
 func New(db DBTX) *Queries {
@@ -24,7 +25,7 @@ type Queries struct {
 	db DBTX
 }
 
-func (q *Queries) WithTx(tx *sql.Tx) *Queries {
+func (q *Queries) WithTx(tx pgx.Tx) *Queries {
 	return &Queries{
 		db: tx,
 	}

ntpdb/dbconn.go

@@ -1,85 +1,15 @@
 package ntpdb
 
+//go:generate go tool github.com/hexdigest/gowrap/cmd/gowrap gen -t ./opentelemetry.gowrap -g -i QuerierTx -p . -o otel.go
+
 import (
 	"context"
-	"database/sql"
-	"database/sql/driver"
-	"fmt"
-	"os"
-	"time"
 
-	"github.com/go-sql-driver/mysql"
+	"github.com/jackc/pgx/v5/pgxpool"
-	"go.ntppool.org/common/logger"
+	"go.ntppool.org/common/database/pgdb"
-	"gopkg.in/yaml.v3"
 )
 
-type Config struct {
+// OpenDB opens a PostgreSQL connection pool using the specified config file
-	MySQL DBConfig `yaml:"mysql"`
+func OpenDB(ctx context.Context, configFile string) (*pgxpool.Pool, error) {
-}
+	return pgdb.OpenPoolWithConfigFile(ctx, configFile)
-
-type DBConfig struct {
-	DSN  string `default:"" flag:"dsn" usage:"Database DSN"`
-	User string `default:"" flag:"user"`
-	Pass string `default:"" flag:"pass"`
-}
-
-func OpenDB(ctx context.Context, configFile string) (*sql.DB, error) {
-	log := logger.FromContext(ctx)
-
-	dbconn := sql.OpenDB(Driver{CreateConnectorFunc: createConnector(ctx, configFile)})
-
-	dbconn.SetConnMaxLifetime(time.Minute * 3)
-	dbconn.SetMaxOpenConns(8)
-	dbconn.SetMaxIdleConns(3)
-
-	err := dbconn.Ping()
-	if err != nil {
-		log.DebugContext(ctx, "could not connect to database: %s", "err", err)
-		return nil, err
-	}
-
-	return dbconn, nil
-}
-
-func createConnector(ctx context.Context, configFile string) CreateConnectorFunc {
-	log := logger.FromContext(ctx)
-	return func() (driver.Connector, error) {
-		log.DebugContext(ctx, "opening db config file", "filename", configFile)
-
-		dbFile, err := os.Open(configFile)
-		if err != nil {
-			return nil, err
-		}
-
-		dec := yaml.NewDecoder(dbFile)
-
-		cfg := Config{}
-
-		err = dec.Decode(&cfg)
-		if err != nil {
-			return nil, err
-		}
-
-		// log.Printf("db cfg: %+v", cfg)
-
-		dsn := cfg.MySQL.DSN
-		if len(dsn) == 0 {
-			return nil, fmt.Errorf("--database.dsn flag or DATABASE_DSN environment variable required")
-		}
-
-		dbcfg, err := mysql.ParseDSN(dsn)
-		if err != nil {
-			return nil, err
-		}
-
-		if user := cfg.MySQL.User; len(user) > 0 {
-			dbcfg.User = user
-		}
-
-		if pass := cfg.MySQL.Pass; len(pass) > 0 {
-			dbcfg.Passwd = pass
-		}
-
-		return mysql.NewConnector(dbcfg)
-	}
 }

ntpdb/dynamic_connect.go

@@ -1,33 +0,0 @@
-package ntpdb
-
-import (
-	"context"
-	"database/sql/driver"
-	"errors"
-	"fmt"
-)
-
-// from https://github.com/Boostport/dynamic-database-config
-
-type CreateConnectorFunc func() (driver.Connector, error)
-
-type Driver struct {
-	CreateConnectorFunc CreateConnectorFunc
-}
-
-func (d Driver) Driver() driver.Driver {
-	return d
-}
-
-func (d Driver) Connect(ctx context.Context) (driver.Conn, error) {
-	connector, err := d.CreateConnectorFunc()
-	if err != nil {
-		return nil, fmt.Errorf("error creating connector from function: %w", err)
-	}
-
-	return connector.Connect(ctx)
-}
-
-func (d Driver) Open(name string) (driver.Conn, error) {
-	return nil, errors.New("open is not supported")
-}

ntpdb/models.go

@@ -5,11 +5,10 @@
 package ntpdb
 
 import (
-	"database/sql"
 	"database/sql/driver"
 	"fmt"
-	"time"
 
+	"github.com/jackc/pgx/v5/pgtype"
 	"go.ntppool.org/common/types"
 )
 
@@ -271,73 +270,73 @@ func (ns NullZoneServerCountsIpVersion) Value() (driver.Value, error) {
 }
 
 type LogScore struct {
-	ID         uint64                   `db:"id" json:"id"`
+	ID         int64                    `db:"id" json:"id"`
-	MonitorID  sql.NullInt32            `db:"monitor_id" json:"monitor_id"`
+	MonitorID  pgtype.Int8              `db:"monitor_id" json:"monitor_id"`
-	ServerID   uint32                   `db:"server_id" json:"server_id"`
+	ServerID   int64                    `db:"server_id" json:"server_id"`
-	Ts         time.Time                `db:"ts" json:"ts"`
+	Ts         pgtype.Timestamptz       `db:"ts" json:"ts"`
 	Score      float64                  `db:"score" json:"score"`
 	Step       float64                  `db:"step" json:"step"`
-	Offset     sql.NullFloat64          `db:"offset" json:"offset"`
+	Offset     pgtype.Float8            `db:"offset" json:"offset"`
-	Rtt        sql.NullInt32            `db:"rtt" json:"rtt"`
+	Rtt        pgtype.Int4              `db:"rtt" json:"rtt"`
 	Attributes types.LogScoreAttributes `db:"attributes" json:"attributes"`
 }
 
 type Monitor struct {
-	ID            uint32                `db:"id" json:"id"`
+	ID            int64                 `db:"id" json:"id"`
-	IDToken       sql.NullString        `db:"id_token" json:"id_token"`
+	IDToken       pgtype.Text           `db:"id_token" json:"id_token"`
 	Type          MonitorsType          `db:"type" json:"type"`
-	UserID        sql.NullInt32         `db:"user_id" json:"user_id"`
+	UserID        pgtype.Int8           `db:"user_id" json:"user_id"`
-	AccountID     sql.NullInt32         `db:"account_id" json:"account_id"`
+	AccountID     pgtype.Int8           `db:"account_id" json:"account_id"`
 	Hostname      string                `db:"hostname" json:"hostname"`
 	Location      string                `db:"location" json:"location"`
-	Ip            sql.NullString        `db:"ip" json:"ip"`
+	Ip            pgtype.Text           `db:"ip" json:"ip"`
 	IpVersion     NullMonitorsIpVersion `db:"ip_version" json:"ip_version"`
-	TlsName       sql.NullString        `db:"tls_name" json:"tls_name"`
+	TlsName       pgtype.Text           `db:"tls_name" json:"tls_name"`
-	ApiKey        sql.NullString        `db:"api_key" json:"api_key"`
+	ApiKey        pgtype.Text           `db:"api_key" json:"api_key"`
 	Status        MonitorsStatus        `db:"status" json:"status"`
 	Config        string                `db:"config" json:"config"`
 	ClientVersion string                `db:"client_version" json:"client_version"`
-	LastSeen      sql.NullTime          `db:"last_seen" json:"last_seen"`
+	LastSeen      pgtype.Timestamptz    `db:"last_seen" json:"last_seen"`
-	LastSubmit    sql.NullTime          `db:"last_submit" json:"last_submit"`
+	LastSubmit    pgtype.Timestamptz    `db:"last_submit" json:"last_submit"`
-	CreatedOn     time.Time             `db:"created_on" json:"created_on"`
+	CreatedOn     pgtype.Timestamptz    `db:"created_on" json:"created_on"`
-	DeletedOn     sql.NullTime          `db:"deleted_on" json:"deleted_on"`
+	DeletedOn     pgtype.Timestamptz    `db:"deleted_on" json:"deleted_on"`
-	IsCurrent     sql.NullBool          `db:"is_current" json:"is_current"`
+	IsCurrent     pgtype.Bool           `db:"is_current" json:"is_current"`
 }
 
 type Server struct {
-	ID             uint32           `db:"id" json:"id"`
+	ID             int64              `db:"id" json:"id"`
 	Ip             string             `db:"ip" json:"ip"`
 	IpVersion      ServersIpVersion   `db:"ip_version" json:"ip_version"`
-	UserID         sql.NullInt32    `db:"user_id" json:"user_id"`
+	UserID         pgtype.Int8        `db:"user_id" json:"user_id"`
-	AccountID      sql.NullInt32    `db:"account_id" json:"account_id"`
+	AccountID      pgtype.Int8        `db:"account_id" json:"account_id"`
-	Hostname       sql.NullString   `db:"hostname" json:"hostname"`
+	Hostname       pgtype.Text        `db:"hostname" json:"hostname"`
-	Stratum        sql.NullInt16    `db:"stratum" json:"stratum"`
+	Stratum        pgtype.Int2        `db:"stratum" json:"stratum"`
-	InPool         uint8            `db:"in_pool" json:"in_pool"`
+	InPool         int16              `db:"in_pool" json:"in_pool"`
-	InServerList   uint8            `db:"in_server_list" json:"in_server_list"`
+	InServerList   int16              `db:"in_server_list" json:"in_server_list"`
-	Netspeed       uint32           `db:"netspeed" json:"netspeed"`
+	Netspeed       int64              `db:"netspeed" json:"netspeed"`
-	NetspeedTarget uint32           `db:"netspeed_target" json:"netspeed_target"`
+	NetspeedTarget int64              `db:"netspeed_target" json:"netspeed_target"`
-	CreatedOn      time.Time        `db:"created_on" json:"created_on"`
+	CreatedOn      pgtype.Timestamptz `db:"created_on" json:"created_on"`
-	UpdatedOn      time.Time        `db:"updated_on" json:"updated_on"`
+	UpdatedOn      pgtype.Timestamptz `db:"updated_on" json:"updated_on"`
-	ScoreTs        sql.NullTime     `db:"score_ts" json:"score_ts"`
+	ScoreTs        pgtype.Timestamptz `db:"score_ts" json:"score_ts"`
 	ScoreRaw       float64            `db:"score_raw" json:"score_raw"`
-	DeletionOn     sql.NullTime     `db:"deletion_on" json:"deletion_on"`
+	DeletionOn     pgtype.Date        `db:"deletion_on" json:"deletion_on"`
 	Flags          string             `db:"flags" json:"flags"`
 }
 
 type Zone struct {
-	ID          uint32         `db:"id" json:"id"`
+	ID          int64       `db:"id" json:"id"`
 	Name        string      `db:"name" json:"name"`
-	Description sql.NullString `db:"description" json:"description"`
+	Description pgtype.Text `db:"description" json:"description"`
-	ParentID    sql.NullInt32  `db:"parent_id" json:"parent_id"`
+	ParentID    pgtype.Int8 `db:"parent_id" json:"parent_id"`
 	Dns         bool        `db:"dns" json:"dns"`
 }
 
 type ZoneServerCount struct {
-	ID              uint32                    `db:"id" json:"id"`
+	ID              int64                     `db:"id" json:"id"`
-	ZoneID          uint32                    `db:"zone_id" json:"zone_id"`
+	ZoneID          int64                     `db:"zone_id" json:"zone_id"`
 	IpVersion       ZoneServerCountsIpVersion `db:"ip_version" json:"ip_version"`
-	Date            time.Time                 `db:"date" json:"date"`
+	Date            pgtype.Date               `db:"date" json:"date"`
-	CountActive     uint32                    `db:"count_active" json:"count_active"`
+	CountActive     int32                     `db:"count_active" json:"count_active"`
-	CountRegistered uint32                    `db:"count_registered" json:"count_registered"`
+	CountRegistered int32                     `db:"count_registered" json:"count_registered"`
 	NetspeedActive  int                       `db:"netspeed_active" json:"netspeed_active"`
 }

ntpdb/opentelemetry.gowrap

@@ -0,0 +1,55 @@
+import (
+    "context"
+
+    _codes "go.opentelemetry.io/otel/codes"
+    "go.opentelemetry.io/otel"
+    "go.opentelemetry.io/otel/attribute"
+)
+
+{{ $decorator := (or .Vars.DecoratorName (printf "%sWithTracing" .Interface.Name)) }}
+{{ $spanNameType := (or .Vars.SpanNamePrefix .Interface.Name) }}
+
+// {{$decorator}} implements {{.Interface.Name}} interface instrumented with open telemetry spans
+type {{$decorator}} struct {
+  {{.Interface.Type}}
+  _instance string
+  _spanDecorator func(span trace.Span, params, results map[string]interface{})
+}
+
+// New{{$decorator}} returns {{$decorator}}
+func New{{$decorator}} (base {{.Interface.Type}}, instance string, spanDecorator ...func(span trace.Span, params, results map[string]interface{})) {{$decorator}} {
+  d := {{$decorator}} {
+    {{.Interface.Name}}: base,
+    _instance: instance,
+  }
+
+  if len(spanDecorator) > 0 && spanDecorator[0] != nil {
+    d._spanDecorator = spanDecorator[0]
+  }
+
+  return d
+}
+
+{{range $method := .Interface.Methods}}
+  {{if $method.AcceptsContext}}
+    // {{$method.Name}} implements {{$.Interface.Name}}
+func (_d {{$decorator}}) {{$method.Declaration}} {
+  ctx, _span := otel.Tracer(_d._instance).Start(ctx, "{{$spanNameType}}.{{$method.Name}}")
+  defer func() {
+    if _d._spanDecorator != nil {
+      _d._spanDecorator(_span, {{$method.ParamsMap}}, {{$method.ResultsMap}})
+    }{{- if $method.ReturnsError}} else if err != nil {
+      _span.RecordError(err)
+      _span.SetStatus(_codes.Error, err.Error())
+      _span.SetAttributes(
+        attribute.String("event", "error"),
+        attribute.String("message", err.Error()),
+      )
+    }
+    {{end}}
+    _span.End()
+  }()
+  {{$method.Pass (printf "_d.%s." $.Interface.Name) }}
+}
+  {{end}}
+{{end}}

ntpdb/otel.go

@@ -1,5 +1,5 @@
 // Code generated by gowrap. DO NOT EDIT.
-// template: https://raw.githubusercontent.com/hexdigest/gowrap/6bd1bc023b4d2a619f30020924f258b8ff665a7a/templates/opentelemetry
+// template: opentelemetry.gowrap
 // gowrap: http://github.com/hexdigest/gowrap
 
 package ntpdb
@@ -7,10 +7,11 @@ package ntpdb
 import (
 	"context"
 
+	"go.opentelemetry.io/otel/trace"
+
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	_codes "go.opentelemetry.io/otel/codes"
-	"go.opentelemetry.io/otel/trace"
 )
 
 // QuerierTxWithTracing implements QuerierTx interface instrumented with open telemetry spans
@@ -104,7 +105,7 @@ func (_d QuerierTxWithTracing) GetMonitorByNameAndIPVersion(ctx context.Context,
 }
 
 // GetMonitorsByID implements QuerierTx
-func (_d QuerierTxWithTracing) GetMonitorsByID(ctx context.Context, monitorids []uint32) (ma1 []Monitor, err error) {
+func (_d QuerierTxWithTracing) GetMonitorsByID(ctx context.Context, monitorids []int64) (ma1 []Monitor, err error) {
 	ctx, _span := otel.Tracer(_d._instance).Start(ctx, "QuerierTx.GetMonitorsByID")
 	defer func() {
 		if _d._spanDecorator != nil {
@@ -128,7 +129,7 @@ func (_d QuerierTxWithTracing) GetMonitorsByID(ctx context.Context, monitorids [
 }
 
 // GetServerByID implements QuerierTx
-func (_d QuerierTxWithTracing) GetServerByID(ctx context.Context, id uint32) (s1 Server, err error) {
+func (_d QuerierTxWithTracing) GetServerByID(ctx context.Context, id int64) (s1 Server, err error) {
 	ctx, _span := otel.Tracer(_d._instance).Start(ctx, "QuerierTx.GetServerByID")
 	defer func() {
 		if _d._spanDecorator != nil {
@@ -224,14 +225,14 @@ func (_d QuerierTxWithTracing) GetServerLogScoresByMonitorID(ctx context.Context
 }
 
 // GetServerNetspeed implements QuerierTx
-func (_d QuerierTxWithTracing) GetServerNetspeed(ctx context.Context, ip string) (u1 uint32, err error) {
+func (_d QuerierTxWithTracing) GetServerNetspeed(ctx context.Context, ip string) (i1 int64, err error) {
 	ctx, _span := otel.Tracer(_d._instance).Start(ctx, "QuerierTx.GetServerNetspeed")
 	defer func() {
 		if _d._spanDecorator != nil {
 			_d._spanDecorator(_span, map[string]interface{}{
 				"ctx": ctx,
 				"ip":  ip}, map[string]interface{}{
-				"u1":  u1,
+				"i1":  i1,
 				"err": err})
 		} else if err != nil {
 			_span.RecordError(err)
@@ -296,7 +297,7 @@ func (_d QuerierTxWithTracing) GetZoneByName(ctx context.Context, name string) (
 }
 
 // GetZoneCounts implements QuerierTx
-func (_d QuerierTxWithTracing) GetZoneCounts(ctx context.Context, zoneID uint32) (za1 []ZoneServerCount, err error) {
+func (_d QuerierTxWithTracing) GetZoneCounts(ctx context.Context, zoneID int64) (za1 []ZoneServerCount, err error) {
 	ctx, _span := otel.Tracer(_d._instance).Start(ctx, "QuerierTx.GetZoneCounts")
 	defer func() {
 		if _d._spanDecorator != nil {

ntpdb/querier.go

@@ -10,15 +10,15 @@ import (
 
 type Querier interface {
 	GetMonitorByNameAndIPVersion(ctx context.Context, arg GetMonitorByNameAndIPVersionParams) (Monitor, error)
-	GetMonitorsByID(ctx context.Context, monitorids []uint32) ([]Monitor, error)
+	GetMonitorsByID(ctx context.Context, monitorids []int64) ([]Monitor, error)
-	GetServerByID(ctx context.Context, id uint32) (Server, error)
+	GetServerByID(ctx context.Context, id int64) (Server, error)
 	GetServerByIP(ctx context.Context, ip string) (Server, error)
 	GetServerLogScores(ctx context.Context, arg GetServerLogScoresParams) ([]LogScore, error)
 	GetServerLogScoresByMonitorID(ctx context.Context, arg GetServerLogScoresByMonitorIDParams) ([]LogScore, error)
-	GetServerNetspeed(ctx context.Context, ip string) (uint32, error)
+	GetServerNetspeed(ctx context.Context, ip string) (int64, error)
 	GetServerScores(ctx context.Context, arg GetServerScoresParams) ([]GetServerScoresRow, error)
 	GetZoneByName(ctx context.Context, name string) (Zone, error)
-	GetZoneCounts(ctx context.Context, zoneID uint32) ([]ZoneServerCount, error)
+	GetZoneCounts(ctx context.Context, zoneID int64) ([]ZoneServerCount, error)
 	GetZoneStatsData(ctx context.Context) ([]GetZoneStatsDataRow, error)
 	GetZoneStatsV2(ctx context.Context, ip string) ([]GetZoneStatsV2Row, error)
 }

ntpdb/query.sql.go

@@ -7,28 +7,27 @@ package ntpdb
 
 import (
 	"context"
-	"database/sql"
+
-	"strings"
+	"github.com/jackc/pgx/v5/pgtype"
-	"time"
 )
 
 const getMonitorByNameAndIPVersion = `-- name: GetMonitorByNameAndIPVersion :one
 select id, id_token, type, user_id, account_id, hostname, location, ip, ip_version, tls_name, api_key, status, config, client_version, last_seen, last_submit, created_on, deleted_on, is_current from monitors
 where
-  tls_name like ? AND
+  tls_name like $1 AND
-  (ip_version = ? OR (type = 'score' AND ip_version IS NULL)) AND
+  (ip_version = $2 OR (type = 'score' AND ip_version IS NULL)) AND
-  is_current = 1
+  is_current = true
   order by id
   limit 1
 `
 
 type GetMonitorByNameAndIPVersionParams struct {
-	TlsName   sql.NullString        `db:"tls_name" json:"tls_name"`
+	TlsName   pgtype.Text           `db:"tls_name" json:"tls_name"`
 	IpVersion NullMonitorsIpVersion `db:"ip_version" json:"ip_version"`
 }
 
 func (q *Queries) GetMonitorByNameAndIPVersion(ctx context.Context, arg GetMonitorByNameAndIPVersionParams) (Monitor, error) {
-	row := q.db.QueryRowContext(ctx, getMonitorByNameAndIPVersion, arg.TlsName, arg.IpVersion)
+	row := q.db.QueryRow(ctx, getMonitorByNameAndIPVersion, arg.TlsName, arg.IpVersion)
 	var i Monitor
 	err := row.Scan(
 		&i.ID,
@@ -56,21 +55,11 @@ func (q *Queries) GetMonitorByNameAndIPVersion(ctx context.Context, arg GetMonit
 
 const getMonitorsByID = `-- name: GetMonitorsByID :many
 select id, id_token, type, user_id, account_id, hostname, location, ip, ip_version, tls_name, api_key, status, config, client_version, last_seen, last_submit, created_on, deleted_on, is_current from monitors
-where id in (/*SLICE:MonitorIDs*/?)
+where id = ANY($1::bigint[])
 `
 
-func (q *Queries) GetMonitorsByID(ctx context.Context, monitorids []uint32) ([]Monitor, error) {
+func (q *Queries) GetMonitorsByID(ctx context.Context, monitorids []int64) ([]Monitor, error) {
-	query := getMonitorsByID
+	rows, err := q.db.Query(ctx, getMonitorsByID, monitorids)
-	var queryParams []interface{}
-	if len(monitorids) > 0 {
-		for _, v := range monitorids {
-			queryParams = append(queryParams, v)
-		}
-		query = strings.Replace(query, "/*SLICE:MonitorIDs*/?", strings.Repeat(",?", len(monitorids))[1:], 1)
-	} else {
-		query = strings.Replace(query, "/*SLICE:MonitorIDs*/?", "NULL", 1)
-	}
-	rows, err := q.db.QueryContext(ctx, query, queryParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -103,9 +92,6 @@ func (q *Queries) GetMonitorsByID(ctx context.Context, monitorids []uint32) ([]M
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -115,11 +101,11 @@ func (q *Queries) GetMonitorsByID(ctx context.Context, monitorids []uint32) ([]M
 const getServerByID = `-- name: GetServerByID :one
 select id, ip, ip_version, user_id, account_id, hostname, stratum, in_pool, in_server_list, netspeed, netspeed_target, created_on, updated_on, score_ts, score_raw, deletion_on, flags from servers
 where
-  id = ?
+  id = $1
 `
 
-func (q *Queries) GetServerByID(ctx context.Context, id uint32) (Server, error) {
+func (q *Queries) GetServerByID(ctx context.Context, id int64) (Server, error) {
-	row := q.db.QueryRowContext(ctx, getServerByID, id)
+	row := q.db.QueryRow(ctx, getServerByID, id)
 	var i Server
 	err := row.Scan(
 		&i.ID,
@@ -146,11 +132,11 @@ func (q *Queries) GetServerByID(ctx context.Context, id uint32) (Server, error)
 const getServerByIP = `-- name: GetServerByIP :one
 select id, ip, ip_version, user_id, account_id, hostname, stratum, in_pool, in_server_list, netspeed, netspeed_target, created_on, updated_on, score_ts, score_raw, deletion_on, flags from servers
 where
-  ip = ?
+  ip = $1
 `
 
 func (q *Queries) GetServerByIP(ctx context.Context, ip string) (Server, error) {
-	row := q.db.QueryRowContext(ctx, getServerByIP, ip)
+	row := q.db.QueryRow(ctx, getServerByIP, ip)
 	var i Server
 	err := row.Scan(
 		&i.ID,
@@ -175,20 +161,20 @@ func (q *Queries) GetServerByIP(ctx context.Context, ip string) (Server, error)
 }
 
 const getServerLogScores = `-- name: GetServerLogScores :many
-select id, monitor_id, server_id, ts, score, step, offset, rtt, attributes from log_scores
+select id, monitor_id, server_id, ts, score, step, "offset", rtt, attributes from log_scores
 where
-  server_id = ?
+  server_id = $1
   order by ts desc
-  limit ?
+  limit $2
 `
 
 type GetServerLogScoresParams struct {
-	ServerID uint32 `db:"server_id" json:"server_id"`
+	ServerID int64 `db:"server_id" json:"server_id"`
 	Limit    int32 `db:"limit" json:"limit"`
 }
 
 func (q *Queries) GetServerLogScores(ctx context.Context, arg GetServerLogScoresParams) ([]LogScore, error) {
-	rows, err := q.db.QueryContext(ctx, getServerLogScores, arg.ServerID, arg.Limit)
+	rows, err := q.db.Query(ctx, getServerLogScores, arg.ServerID, arg.Limit)
 	if err != nil {
 		return nil, err
 	}
@@ -211,9 +197,6 @@ func (q *Queries) GetServerLogScores(ctx context.Context, arg GetServerLogScores
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -221,22 +204,22 @@ func (q *Queries) GetServerLogScores(ctx context.Context, arg GetServerLogScores
 }
 
 const getServerLogScoresByMonitorID = `-- name: GetServerLogScoresByMonitorID :many
-select id, monitor_id, server_id, ts, score, step, offset, rtt, attributes from log_scores
+select id, monitor_id, server_id, ts, score, step, "offset", rtt, attributes from log_scores
 where
-  server_id = ? AND
+  server_id = $1 AND
-  monitor_id = ?
+  monitor_id = $2
   order by ts desc
-  limit ?
+  limit $3
 `
 
 type GetServerLogScoresByMonitorIDParams struct {
-	ServerID  uint32        `db:"server_id" json:"server_id"`
+	ServerID  int64       `db:"server_id" json:"server_id"`
-	MonitorID sql.NullInt32 `db:"monitor_id" json:"monitor_id"`
+	MonitorID pgtype.Int8 `db:"monitor_id" json:"monitor_id"`
 	Limit     int32       `db:"limit" json:"limit"`
 }
 
 func (q *Queries) GetServerLogScoresByMonitorID(ctx context.Context, arg GetServerLogScoresByMonitorIDParams) ([]LogScore, error) {
-	rows, err := q.db.QueryContext(ctx, getServerLogScoresByMonitorID, arg.ServerID, arg.MonitorID, arg.Limit)
+	rows, err := q.db.Query(ctx, getServerLogScoresByMonitorID, arg.ServerID, arg.MonitorID, arg.Limit)
 	if err != nil {
 		return nil, err
 	}
@@ -259,9 +242,6 @@ func (q *Queries) GetServerLogScoresByMonitorID(ctx context.Context, arg GetServ
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -269,12 +249,12 @@ func (q *Queries) GetServerLogScoresByMonitorID(ctx context.Context, arg GetServ
 }
 
 const getServerNetspeed = `-- name: GetServerNetspeed :one
-select netspeed from servers where ip = ?
+select netspeed from servers where ip = $1
 `
 
-func (q *Queries) GetServerNetspeed(ctx context.Context, ip string) (uint32, error) {
+func (q *Queries) GetServerNetspeed(ctx context.Context, ip string) (int64, error) {
-	row := q.db.QueryRowContext(ctx, getServerNetspeed, ip)
+	row := q.db.QueryRow(ctx, getServerNetspeed, ip)
-	var netspeed uint32
+	var netspeed int64
 	err := row.Scan(&netspeed)
 	return netspeed, err
 }
@@ -287,39 +267,28 @@ select
     inner join monitors m
       on (m.id=ss.monitor_id)
 where
-  server_id = ? AND
+  server_id = $1 AND
-  monitor_id in (/*SLICE:MonitorIDs*/?)
+  monitor_id = ANY($2::bigint[])
 `
 
 type GetServerScoresParams struct {
-	ServerID   uint32   `db:"server_id" json:"server_id"`
+	ServerID   int64   `db:"server_id" json:"server_id"`
-	MonitorIDs []uint32 `db:"MonitorIDs" json:"MonitorIDs"`
+	MonitorIDs []int64 `db:"MonitorIDs" json:"MonitorIDs"`
 }
 
 type GetServerScoresRow struct {
-	ID       uint32             `db:"id" json:"id"`
+	ID       int64              `db:"id" json:"id"`
 	Hostname string             `db:"hostname" json:"hostname"`
-	TlsName  sql.NullString     `db:"tls_name" json:"tls_name"`
+	TlsName  pgtype.Text        `db:"tls_name" json:"tls_name"`
 	Location string             `db:"location" json:"location"`
 	Type     MonitorsType       `db:"type" json:"type"`
 	ScoreRaw float64            `db:"score_raw" json:"score_raw"`
-	ScoreTs  sql.NullTime       `db:"score_ts" json:"score_ts"`
+	ScoreTs  pgtype.Timestamptz `db:"score_ts" json:"score_ts"`
 	Status   ServerScoresStatus `db:"status" json:"status"`
 }
 
 func (q *Queries) GetServerScores(ctx context.Context, arg GetServerScoresParams) ([]GetServerScoresRow, error) {
-	query := getServerScores
+	rows, err := q.db.Query(ctx, getServerScores, arg.ServerID, arg.MonitorIDs)
-	var queryParams []interface{}
-	queryParams = append(queryParams, arg.ServerID)
-	if len(arg.MonitorIDs) > 0 {
-		for _, v := range arg.MonitorIDs {
-			queryParams = append(queryParams, v)
-		}
-		query = strings.Replace(query, "/*SLICE:MonitorIDs*/?", strings.Repeat(",?", len(arg.MonitorIDs))[1:], 1)
-	} else {
-		query = strings.Replace(query, "/*SLICE:MonitorIDs*/?", "NULL", 1)
-	}
-	rows, err := q.db.QueryContext(ctx, query, queryParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -341,9 +310,6 @@ func (q *Queries) GetServerScores(ctx context.Context, arg GetServerScoresParams
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -353,11 +319,11 @@ func (q *Queries) GetServerScores(ctx context.Context, arg GetServerScoresParams
 const getZoneByName = `-- name: GetZoneByName :one
 select id, name, description, parent_id, dns from zones
 where
-  name = ?
+  name = $1
 `
 
 func (q *Queries) GetZoneByName(ctx context.Context, name string) (Zone, error) {
-	row := q.db.QueryRowContext(ctx, getZoneByName, name)
+	row := q.db.QueryRow(ctx, getZoneByName, name)
 	var i Zone
 	err := row.Scan(
 		&i.ID,
@@ -371,12 +337,12 @@ func (q *Queries) GetZoneByName(ctx context.Context, name string) (Zone, error)
 
 const getZoneCounts = `-- name: GetZoneCounts :many
 select id, zone_id, ip_version, date, count_active, count_registered, netspeed_active from zone_server_counts
-  where zone_id = ?
+  where zone_id = $1
   order by date
 `
 
-func (q *Queries) GetZoneCounts(ctx context.Context, zoneID uint32) ([]ZoneServerCount, error) {
+func (q *Queries) GetZoneCounts(ctx context.Context, zoneID int64) ([]ZoneServerCount, error) {
-	rows, err := q.db.QueryContext(ctx, getZoneCounts, zoneID)
+	rows, err := q.db.Query(ctx, getZoneCounts, zoneID)
 	if err != nil {
 		return nil, err
 	}
@@ -397,9 +363,6 @@ func (q *Queries) GetZoneCounts(ctx context.Context, zoneID uint32) ([]ZoneServe
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -408,7 +371,7 @@ func (q *Queries) GetZoneCounts(ctx context.Context, zoneID uint32) ([]ZoneServe
 
 const getZoneStatsData = `-- name: GetZoneStatsData :many
 SELECT zc.date, z.name, zc.ip_version, count_active, count_registered, netspeed_active
-FROM zone_server_counts zc USE INDEX (date_idx)
+FROM zone_server_counts zc
   INNER JOIN zones z
     ON(zc.zone_id=z.id)
   WHERE date IN (SELECT max(date) from zone_server_counts)
@@ -416,16 +379,16 @@ ORDER BY name
 `
 
 type GetZoneStatsDataRow struct {
-	Date            time.Time                 `db:"date" json:"date"`
+	Date            pgtype.Date               `db:"date" json:"date"`
 	Name            string                    `db:"name" json:"name"`
 	IpVersion       ZoneServerCountsIpVersion `db:"ip_version" json:"ip_version"`
-	CountActive     uint32                    `db:"count_active" json:"count_active"`
+	CountActive     int32                     `db:"count_active" json:"count_active"`
-	CountRegistered uint32                    `db:"count_registered" json:"count_registered"`
+	CountRegistered int32                     `db:"count_registered" json:"count_registered"`
 	NetspeedActive  int                       `db:"netspeed_active" json:"netspeed_active"`
 }
 
 func (q *Queries) GetZoneStatsData(ctx context.Context) ([]GetZoneStatsDataRow, error) {
-	rows, err := q.db.QueryContext(ctx, getZoneStatsData)
+	rows, err := q.db.Query(ctx, getZoneStatsData)
 	if err != nil {
 		return nil, err
 	}
@@ -445,9 +408,6 @@ func (q *Queries) GetZoneStatsData(ctx context.Context) ([]GetZoneStatsDataRow,
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
@@ -455,15 +415,14 @@ func (q *Queries) GetZoneStatsData(ctx context.Context) ([]GetZoneStatsDataRow,
 }
 
 const getZoneStatsV2 = `-- name: GetZoneStatsV2 :many
-select zone_name, netspeed_active+0 as netspeed_active FROM (
 SELECT
 	z.name as zone_name,
-	SUM(
+	CAST(SUM(
-		IF (deletion_on IS NULL AND score_raw > 10,
+		CASE WHEN deletion_on IS NULL AND score_raw > 10
-			netspeed,
+			THEN netspeed
-		  0
+			ELSE 0
-    )
+		END
-	) AS netspeed_active
+	) AS int) AS netspeed_active
 FROM
 	servers s
 	INNER JOIN server_zones sz ON (sz.server_id = s.id)
@@ -472,23 +431,22 @@ FROM
     select zone_id, s.ip_version
     from server_zones sz
       inner join servers s on (s.id=sz.server_id)
-    where s.ip=?
+    where s.ip=$1
   ) as srvz on (srvz.zone_id=z.id AND srvz.ip_version=s.ip_version)
 WHERE
 	(deletion_on IS NULL OR deletion_on > NOW())
 	AND in_pool = 1
 	AND netspeed > 0
-GROUP BY z.name)
+GROUP BY z.name
-AS server_netspeed
 `
 
 type GetZoneStatsV2Row struct {
 	ZoneName       string `db:"zone_name" json:"zone_name"`
-	NetspeedActive int    `db:"netspeed_active" json:"netspeed_active"`
+	NetspeedActive int32  `db:"netspeed_active" json:"netspeed_active"`
 }
 
 func (q *Queries) GetZoneStatsV2(ctx context.Context, ip string) ([]GetZoneStatsV2Row, error) {
-	rows, err := q.db.QueryContext(ctx, getZoneStatsV2, ip)
+	rows, err := q.db.Query(ctx, getZoneStatsV2, ip)
 	if err != nil {
 		return nil, err
 	}
@@ -501,9 +459,6 @@ func (q *Queries) GetZoneStatsV2(ctx context.Context, ip string) ([]GetZoneStats
 		}
 		items = append(items, i)
 	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}

ntpdb/tx.go

@@ -2,7 +2,11 @@ package ntpdb
 
 import (
 	"context"
-	"database/sql"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"go.ntppool.org/common/logger"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type QuerierTx interface {
@@ -11,14 +15,17 @@ type QuerierTx interface {
 	Begin(ctx context.Context) (QuerierTx, error)
 	Commit(ctx context.Context) error
 	Rollback(ctx context.Context) error
+
+	// Conn returns the connection used by this transaction
+	Conn() *pgx.Conn
 }
 
 type Beginner interface {
-	Begin(context.Context) (sql.Tx, error)
+	Begin(context.Context) (pgx.Tx, error)
 }
 
 type Tx interface {
-	Begin(context.Context) (sql.Tx, error)
+	Begin(context.Context) (pgx.Tx, error)
 	Commit(ctx context.Context) error
 	Rollback(ctx context.Context) error
 }
@@ -28,21 +35,33 @@ func (q *Queries) Begin(ctx context.Context) (QuerierTx, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Queries{db: &tx}, nil
+	return &Queries{db: tx}, nil
 }
 
 func (q *Queries) Commit(ctx context.Context) error {
 	tx, ok := q.db.(Tx)
 	if !ok {
-		return sql.ErrTxDone
+		// Commit called on Queries with dbpool, so treat as transaction already committed
+		return pgx.ErrTxClosed
 	}
 	return tx.Commit(ctx)
 }
 
+func (q *Queries) Conn() *pgx.Conn {
+	// pgx.Tx is an interface that has Conn() method
+	tx, ok := q.db.(pgx.Tx)
+	if !ok {
+		logger.Setup().Error("could not get connection from QuerierTx")
+		return nil
+	}
+	return tx.Conn()
+}
+
 func (q *Queries) Rollback(ctx context.Context) error {
 	tx, ok := q.db.(Tx)
 	if !ok {
-		return sql.ErrTxDone
+		// Rollback called on Queries with dbpool, so treat as transaction already committed
+		return pgx.ErrTxClosed
 	}
 	return tx.Rollback(ctx)
 }
@@ -62,3 +81,41 @@ func (wq *WrappedQuerier) Begin(ctx context.Context) (QuerierTx, error) {
 	}
 	return NewWrappedQuerier(q), nil
 }
+
+func (wq *WrappedQuerier) Conn() *pgx.Conn {
+	return wq.QuerierTxWithTracing.Conn()
+}
+
+// LogRollback logs and performs a rollback if the transaction is still active
+func LogRollback(ctx context.Context, tx QuerierTx) {
+	if !isInTransaction(tx) {
+		return
+	}
+
+	log := logger.FromContext(ctx)
+	log.WarnContext(ctx, "transaction rollback called on an active transaction")
+
+	// if caller ctx is done we still need rollback to happen
+	// so Rollback gets a fresh context with span copied over
+	rbCtx := context.Background()
+	if span := trace.SpanFromContext(ctx); span != nil {
+		rbCtx = trace.ContextWithSpan(rbCtx, span)
+	}
+	if err := tx.Rollback(rbCtx); err != nil && !errors.Is(err, pgx.ErrTxClosed) {
+		log.ErrorContext(ctx, "rollback failed", "err", err)
+	}
+}
+
+func isInTransaction(tx QuerierTx) bool {
+	if tx == nil {
+		return false
+	}
+
+	conn := tx.Conn()
+	if conn == nil {
+		return false
+	}
+
+	// 'I' means idle, so if it's not idle, we're in a transaction
+	return conn.PgConn().TxStatus() != 'I'
+}

query.sql

@@ -1,6 +1,6 @@
 -- name: GetZoneStatsData :many
 SELECT zc.date, z.name, zc.ip_version, count_active, count_registered, netspeed_active
-FROM zone_server_counts zc USE INDEX (date_idx)
+FROM zone_server_counts zc
   INNER JOIN zones z
     ON(zc.zone_id=z.id)
   WHERE date IN (SELECT max(date) from zone_server_counts)
@@ -8,18 +8,17 @@ ORDER BY name;
 
 
 -- name: GetServerNetspeed :one
-select netspeed from servers where ip = ?;
+select netspeed from servers where ip = $1;
 
 -- name: GetZoneStatsV2 :many
-select zone_name, netspeed_active+0 as netspeed_active FROM (
 SELECT
 	z.name as zone_name,
-	SUM(
+	CAST(SUM(
-		IF (deletion_on IS NULL AND score_raw > 10,
+		CASE WHEN deletion_on IS NULL AND score_raw > 10
-			netspeed,
+			THEN netspeed
-		  0
+			ELSE 0
-    )
+		END
-	) AS netspeed_active
+	) AS int) AS netspeed_active
 FROM
 	servers s
 	INNER JOIN server_zones sz ON (sz.server_id = s.id)
@@ -28,19 +27,18 @@ FROM
     select zone_id, s.ip_version
     from server_zones sz
       inner join servers s on (s.id=sz.server_id)
-    where s.ip=?
+    where s.ip=$1
   ) as srvz on (srvz.zone_id=z.id AND srvz.ip_version=s.ip_version)
 WHERE
 	(deletion_on IS NULL OR deletion_on > NOW())
 	AND in_pool = 1
 	AND netspeed > 0
-GROUP BY z.name)
+GROUP BY z.name;
-AS server_netspeed;
 
 -- name: GetServerByID :one
 select * from servers
 where
-  id = ?;
+  id = $1;
 
 -- name: GetServerByIP :one
 select * from servers
@@ -52,13 +50,13 @@ select * from monitors
 where
   tls_name like sqlc.arg('tls_name') AND
   (ip_version = sqlc.arg('ip_version') OR (type = 'score' AND ip_version IS NULL)) AND
-  is_current = 1
+  is_current = true
   order by id
   limit 1;
 
 -- name: GetMonitorsByID :many
 select * from monitors
-where id in (sqlc.slice('MonitorIDs'));
+where id = ANY(sqlc.arg('MonitorIDs')::bigint[]);
 
 -- name: GetServerScores :many
 select
@@ -68,23 +66,23 @@ select
     inner join monitors m
       on (m.id=ss.monitor_id)
 where
-  server_id = ? AND
+  server_id = $1 AND
-  monitor_id in (sqlc.slice('MonitorIDs'));
+  monitor_id = ANY(sqlc.arg('MonitorIDs')::bigint[]);
 
 -- name: GetServerLogScores :many
 select * from log_scores
 where
-  server_id = ?
+  server_id = $1
   order by ts desc
-  limit ?;
+  limit $2;
 
 -- name: GetServerLogScoresByMonitorID :many
 select * from log_scores
 where
-  server_id = ? AND
+  server_id = $1 AND
-  monitor_id = ?
+  monitor_id = $2
   order by ts desc
-  limit ?;
+  limit $3;
 
 -- name: GetZoneByName :one
 select * from zones
@@ -93,5 +91,5 @@ where
 
 -- name: GetZoneCounts :many
 select * from zone_server_counts
-  where zone_id = ?
+  where zone_id = $1
   order by date;

server/dnsanswers.go

@@ -1,11 +1,11 @@
 package server
 
 import (
-	"database/sql"
 	"errors"
 	"net/http"
 	"net/netip"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/labstack/echo/v4"
 	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/sync/errgroup"
@@ -56,7 +56,7 @@ func (srv *Server) dnsAnswers(c echo.Context) error {
 	queryGroup, ctx := errgroup.WithContext(ctx)
 
 	var zoneStats []ntpdb.GetZoneStatsV2Row
-	var serverNetspeed uint32
+	var serverNetspeed int64
 
 	queryGroup.Go(func() error {
 		var err error
@@ -64,7 +64,7 @@ func (srv *Server) dnsAnswers(c echo.Context) error {
 
 		serverNetspeed, err = q.GetServerNetspeed(ctx, ip.String())
 		if err != nil {
-			if !errors.Is(err, sql.ErrNoRows) {
+			if !errors.Is(err, pgx.ErrNoRows) {
 				log.Error("GetServerNetspeed", "err", err)
 			}
 			return err // this will return if the server doesn't exist
@@ -116,7 +116,7 @@ func (srv *Server) dnsAnswers(c echo.Context) error {
 
 	err = queryGroup.Wait()
 	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
+		if errors.Is(err, pgx.ErrNoRows) {
 			return c.String(http.StatusNotFound, "Not found")
 		}
 		log.Error("query error", "err", err)
@@ -130,7 +130,7 @@ func (srv *Server) dnsAnswers(c echo.Context) error {
 		if zn == "@" {
 			zn = ""
 		}
-		zoneTotals[zn] = z.NetspeedActive // binary.BigEndian.Uint64(...)
+		zoneTotals[zn] = int(z.NetspeedActive) // binary.BigEndian.Uint64(...)
 		// log.Info("zone netspeed", "cc", z.ZoneName, "speed", z.NetspeedActive)
 	}
 

server/functions.go

@@ -2,12 +2,12 @@ package server
 
 import (
 	"context"
-	"database/sql"
 	"errors"
 	"net/netip"
 	"strconv"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"go.ntppool.org/common/logger"
 	"go.ntppool.org/common/tracing"
 	"go.ntppool.org/data-api/ntpdb"
@@ -22,7 +22,7 @@ func (srv *Server) FindServer(ctx context.Context, serverID string) (ntpdb.Serve
 	var serverData ntpdb.Server
 	var dberr error
 	if id, err := strconv.Atoi(serverID); id > 0 && err == nil {
-		serverData, dberr = q.GetServerByID(ctx, uint32(id))
+		serverData, dberr = q.GetServerByID(ctx, int64(id))
 	} else {
 		ip, err := netip.ParseAddr(serverID)
 		if err != nil || !ip.IsValid() {
@@ -31,7 +31,7 @@ func (srv *Server) FindServer(ctx context.Context, serverID string) (ntpdb.Serve
 		serverData, dberr = q.GetServerByIP(ctx, ip.String())
 	}
 	if dberr != nil {
-		if !errors.Is(dberr, sql.ErrNoRows) {
+		if !errors.Is(dberr, pgx.ErrNoRows) {
 			log.Error("could not query server id", "err", dberr)
 			return serverData, dberr
 		}

server/grafana.go

@@ -195,7 +195,7 @@ func transformToGrafanaTableFormat(history *logscores.LogScoreHistory, monitors
 			skippedInvalidMonitors++
 			continue
 		}
-		monitorID := int(ls.MonitorID.Int32)
+		monitorID := int(ls.MonitorID.Int64)
 		monitorData[monitorID] = append(monitorData[monitorID], ls)
 	}
 
@@ -275,7 +275,7 @@ func transformToGrafanaTableFormat(history *logscores.LogScoreHistory, monitors
 		var values [][]interface{}
 		for _, ls := range logScores {
 			// Convert timestamp to milliseconds
-			timestampMs := ls.Ts.Unix() * 1000
+			timestampMs := ls.Ts.Time.Unix() * 1000
 
 			// Create row: [time, score, rtt, offset]
 			row := []interface{}{
@@ -382,7 +382,7 @@ func (srv *Server) scoresTimeRange(c echo.Context) error {
 		"time_range_duration", params.to.Sub(params.from).String(),
 	)
 
-	logScores, err := srv.ch.LogscoresTimeRange(ctx, int(server.ID), params.monitorID, params.from, params.to, params.maxDataPoints)
+	logScores, err := srv.ch.LogscoresTimeRange(ctx, int(server.ID), int(params.monitorID), params.from, params.to, params.maxDataPoints)
 	if err != nil {
 		log.ErrorContext(ctx, "clickhouse time range query", "err", err,
 			"server_id", server.ID,
@@ -397,8 +397,8 @@ func (srv *Server) scoresTimeRange(c echo.Context) error {
 	log.InfoContext(ctx, "clickhouse query results",
 		"server_id", server.ID,
 		"rows_returned", len(logScores),
-		"first_few_ids", func() []uint64 {
+		"first_few_ids", func() []int64 {
-			ids := make([]uint64, 0, 3)
+			ids := make([]int64, 0, 3)
 			for i, ls := range logScores {
 				if i >= 3 {
 					break
@@ -416,10 +416,10 @@ func (srv *Server) scoresTimeRange(c echo.Context) error {
 	}
 
 	// Get monitor names for the returned data
-	monitorIDs := []uint32{}
+	monitorIDs := []int64{}
 	for _, ls := range logScores {
 		if ls.MonitorID.Valid {
-			monitorID := uint32(ls.MonitorID.Int32)
+			monitorID := ls.MonitorID.Int64
 			if _, exists := history.Monitors[int(monitorID)]; !exists {
 				history.Monitors[int(monitorID)] = ""
 				monitorIDs = append(monitorIDs, monitorID)

server/history.go

@@ -3,7 +3,6 @@ package server
 import (
 	"bytes"
 	"context"
-	"database/sql"
 	"encoding/csv"
 	"errors"
 	"fmt"
@@ -15,6 +14,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/labstack/echo/v4"
 	"go.ntppool.org/common/logger"
 	"go.ntppool.org/common/tracing"
@@ -63,7 +64,7 @@ func paramHistoryMode(s string) historyMode {
 
 type historyParameters struct {
 	limit       int
-	monitorID   int
+	monitorID   int64
 	server      ntpdb.Server
 	since       time.Time
 	fullHistory bool
@@ -90,7 +91,7 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 
 	monitorParam := c.QueryParam("monitor")
 
-	var monitorID uint32
+	var monitorID int64
 	switch monitorParam {
 	case "":
 		name := "recentmedian.scores.ntp.dev"
@@ -101,7 +102,7 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 			ipVersion = ntpdb.NullMonitorsIpVersion{MonitorsIpVersion: ntpdb.MonitorsIpVersionV6, Valid: true}
 		}
 		monitor, err := q.GetMonitorByNameAndIPVersion(ctx, ntpdb.GetMonitorByNameAndIPVersionParams{
-			TlsName:   sql.NullString{Valid: true, String: name},
+			TlsName:   pgtype.Text{Valid: true, String: name},
 			IpVersion: ipVersion,
 		})
 		if err != nil {
@@ -111,9 +112,9 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 	case "*":
 		monitorID = 0 // don't filter on monitor ID
 	default:
-		mID, err := strconv.ParseUint(monitorParam, 10, 32)
+		mID, err := strconv.ParseInt(monitorParam, 10, 64)
 		if err == nil {
-			monitorID = uint32(mID)
+			monitorID = mID
 		} else {
 			// only accept the name prefix; no wildcards; trust the database
 			// to filter out any other crazy
@@ -129,11 +130,11 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 				ipVersion = ntpdb.NullMonitorsIpVersion{MonitorsIpVersion: ntpdb.MonitorsIpVersionV6, Valid: true}
 			}
 			monitor, err := q.GetMonitorByNameAndIPVersion(ctx, ntpdb.GetMonitorByNameAndIPVersionParams{
-				TlsName:   sql.NullString{Valid: true, String: monitorParam},
+				TlsName:   pgtype.Text{Valid: true, String: monitorParam},
 				IpVersion: ipVersion,
 			})
 			if err != nil {
-				if err == sql.ErrNoRows {
+				if errors.Is(err, pgx.ErrNoRows) {
 					return p, echo.NewHTTPError(http.StatusNotFound, "monitor not found").WithInternal(err)
 				}
 				log.WarnContext(ctx, "could not find monitor", "name", monitorParam, "ip_version", server.IpVersion, "err", err)
@@ -144,7 +145,7 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 		}
 	}
 
-	p.monitorID = int(monitorID)
+	p.monitorID = monitorID
 	log.DebugContext(ctx, "monitor param", "monitor", monitorID, "ip_version", server.IpVersion)
 
 	since, _ := strconv.ParseInt(c.QueryParam("since"), 10, 64) // defaults to 0 so don't care if it parses
@@ -170,8 +171,8 @@ func (srv *Server) getHistoryParameters(ctx context.Context, c echo.Context, ser
 	return p, nil
 }
 
-func (srv *Server) getHistoryMySQL(ctx context.Context, _ echo.Context, p historyParameters) (*logscores.LogScoreHistory, error) {
+func (srv *Server) getHistoryPostgres(ctx context.Context, _ echo.Context, p historyParameters) (*logscores.LogScoreHistory, error) {
-	ls, err := logscores.GetHistoryMySQL(ctx, srv.db, p.server.ID, uint32(p.monitorID), p.since, p.limit)
+	ls, err := logscores.GetHistoryPostgres(ctx, srv.db, p.server.ID, p.monitorID, p.since, p.limit)
 	return ls, err
 }
 
@@ -230,9 +231,9 @@ func (srv *Server) history(c echo.Context) error {
 	}
 
 	if sourceParam == "m" {
-		history, err = srv.getHistoryMySQL(ctx, c, p)
+		history, err = srv.getHistoryPostgres(ctx, c, p)
 	} else {
-		history, err = logscores.GetHistoryClickHouse(ctx, srv.ch, srv.db, p.server.ID, uint32(p.monitorID), p.since, p.limit, p.fullHistory)
+		history, err = logscores.GetHistoryClickHouse(ctx, srv.ch, srv.db, p.server.ID, p.monitorID, p.since, p.limit, p.fullHistory)
 	}
 	if err != nil {
 		var httpError *echo.HTTPError
@@ -276,7 +277,7 @@ func (srv *Server) historyJSON(ctx context.Context, c echo.Context, server ntpdb
 	}
 
 	type MonitorEntry struct {
-		ID     uint32   `json:"id"`
+		ID     int64    `json:"id"`
 		Name   string   `json:"name"`
 		Type   string   `json:"type"`
 		Ts     string   `json:"ts"`
@@ -297,9 +298,9 @@ func (srv *Server) historyJSON(ctx context.Context, c echo.Context, server ntpdb
 
 	// log.InfoContext(ctx, "monitor id list", "ids", history.MonitorIDs)
 
-	monitorIDs := []uint32{}
+	monitorIDs := []int64{}
 	for k := range history.Monitors {
-		monitorIDs = append(monitorIDs, uint32(k))
+		monitorIDs = append(monitorIDs, int64(k))
 	}
 
 	q := ntpdb.NewWrappedQuerier(ntpdb.New(srv.db))
@@ -318,12 +319,12 @@ func (srv *Server) historyJSON(ctx context.Context, c echo.Context, server ntpdb
 	// log.InfoContext(ctx, "got logScoreMonitors", "count", len(logScoreMonitors))
 
 	// Calculate average RTT per monitor
-	monitorRttSums := make(map[uint32]float64)
+	monitorRttSums := make(map[int64]float64)
-	monitorRttCounts := make(map[uint32]int)
+	monitorRttCounts := make(map[int64]int)
 
 	for _, ls := range history.LogScores {
 		if ls.MonitorID.Valid && ls.Rtt.Valid {
-			monitorID := uint32(ls.MonitorID.Int32)
+			monitorID := ls.MonitorID.Int64
 			monitorRttSums[monitorID] += float64(ls.Rtt.Int32) / 1000.0
 			monitorRttCounts[monitorID]++
 		}
@@ -362,8 +363,8 @@ func (srv *Server) historyJSON(ctx context.Context, c echo.Context, server ntpdb
 		x := float64(1000000000000)
 		score := math.Round(ls.Score*x) / x
 		res.History[i] = ScoresEntry{
-			TS:        ls.Ts.Unix(),
+			TS:        ls.Ts.Time.Unix(),
-			MonitorID: int(ls.MonitorID.Int32),
+			MonitorID: int(ls.MonitorID.Int64),
 			Step:      ls.Step,
 			Score:     score,
 		}
@@ -414,7 +415,7 @@ func (srv *Server) historyCSV(ctx context.Context, c echo.Context, history *logs
 		score := ff(l.Score)
 		var monName string
 		if l.MonitorID.Valid {
-			monName = history.Monitors[int(l.MonitorID.Int32)]
+			monName = history.Monitors[int(l.MonitorID.Int64)]
 		}
 		var leap string
 		if l.Attributes.Leap != 0 {
@@ -427,13 +428,13 @@ func (srv *Server) historyCSV(ctx context.Context, c echo.Context, history *logs
 		}
 
 		err := w.Write([]string{
-			strconv.Itoa(int(l.Ts.Unix())),
+			strconv.Itoa(int(l.Ts.Time.Unix())),
 			// l.Ts.Format(time.RFC3339),
-			l.Ts.Format("2006-01-02 15:04:05"),
+			l.Ts.Time.Format("2006-01-02 15:04:05"),
 			offset,
 			step,
 			score,
-			fmt.Sprintf("%d", l.MonitorID.Int32),
+			fmt.Sprintf("%d", l.MonitorID.Int64),
 			monName,
 			rtt,
 			leap,
@@ -464,7 +465,7 @@ func setHistoryCacheControl(c echo.Context, history *logscores.LogScoreHistory)
 	if len(history.LogScores) == 0 ||
 		// cache for longer if data hasn't updated for a while; or we didn't
 		// find any.
-		(time.Now().Add(-8 * time.Hour).After(history.LogScores[len(history.LogScores)-1].Ts)) {
+		(time.Now().Add(-8 * time.Hour).After(history.LogScores[len(history.LogScores)-1].Ts.Time)) {
 		hdr.Set("Cache-Control", "s-maxage=260,max-age=360")
 	} else {
 		if len(history.LogScores) == 1 {

server/server.go

@@ -2,17 +2,16 @@ package server
 
 import (
 	"context"
-	"database/sql"
 	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
 	"os"
-	"strconv"
 	"time"
 
 	"golang.org/x/sync/errgroup"
 
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/labstack/echo-contrib/echoprometheus"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
@@ -36,7 +35,7 @@ import (
 )
 
 type Server struct {
-	db     *sql.DB
+	db     *pgxpool.Pool
 	ch     *chdb.ClickHouse
 	config *config.Config
 
@@ -55,7 +54,7 @@ func NewServer(ctx context.Context, configFile string) (*Server, error) {
 	}
 	db, err := ntpdb.OpenDB(ctx, configFile)
 	if err != nil {
-		return nil, fmt.Errorf("mysql open: %w", err)
+		return nil, fmt.Errorf("postgres open: %w", err)
 	}
 
 	conf := config.New()
@@ -137,7 +136,8 @@ func (srv *Server) Run() error {
 	}))
 	e.Use(otelecho.Middleware("data-api"))
 
-	e.Use(slogecho.NewWithConfig(log,
+	e.Use(slogecho.NewWithConfig(
+		log,
 		slogecho.Config{
 			WithTraceID:      false, // done by logger already
 			DefaultLevel:     slog.LevelInfo,
@@ -182,11 +182,15 @@ func (srv *Server) Run() error {
 			"http://localhost", "http://localhost:5173", "http://localhost:5174", "http://localhost:8080",
 			"https://www.ntppool.org", "https://*.ntppool.org",
 			"https://web.beta.grundclock.com", "https://manage.beta.grundclock.com",
-			"https:/*.askdev.grundclock.com",
+			"https://*.askdev.grundclock.com",
 		},
 		AllowHeaders: []string{echo.HeaderOrigin, echo.HeaderContentType, echo.HeaderAccept},
 	}))
 
+	e.Pre(middleware.Rewrite(map[string]string{
+		"/api/data/*": "/api/$1",
+	}))
+
 	e.Use(middleware.RecoverWithConfig(middleware.RecoverConfig{
 		LogErrorFunc: func(c echo.Context, err error, stack []byte) error {
 			log.ErrorContext(c.Request().Context(), err.Error(), "stack", string(stack))
@@ -303,22 +307,9 @@ func healthHandler(srv *Server, log *slog.Logger) func(w http.ResponseWriter, re
 		defer cancel()
 		g, ctx := errgroup.WithContext(ctx)
 
-		stats := srv.db.Stats()
+		stats := srv.db.Stat()
-		if stats.OpenConnections > 3 {
+		if stats.TotalConns() > 3 {
-			log.InfoContext(ctx, "health requests", "url", req.URL.String(), "stats", stats)
+			log.InfoContext(ctx, "health requests", "url", req.URL.String(), "total_conns", stats.TotalConns())
-		}
-
-		if resetParam := req.URL.Query().Get("reset"); resetParam != "" {
-			reset, err := strconv.ParseBool(resetParam)
-			log.InfoContext(ctx, "db reset request", "err", err, "reset", reset)
-
-			if err == nil && reset {
-				// this feature was to debug some specific problem
-				log.InfoContext(ctx, "setting idle db conns to zero")
-				srv.db.SetConnMaxLifetime(30 * time.Second)
-				srv.db.SetMaxIdleConns(0)
-				srv.db.SetMaxIdleConns(4)
-			}
 		}
 
 		g.Go(func() error {
@@ -340,7 +331,7 @@ func healthHandler(srv *Server, log *slog.Logger) func(w http.ResponseWriter, re
 		})
 
 		g.Go(func() error {
-			err := srv.db.PingContext(ctx)
+			err := srv.db.Ping(ctx)
 			if err != nil {
 				log.WarnContext(ctx, "db ping", "err", err)
 				return err

server/zones.go

@@ -1,12 +1,12 @@
 package server
 
 import (
-	"database/sql"
 	"errors"
 	"net/http"
 	"strconv"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/labstack/echo/v4"
 	"go.ntppool.org/common/logger"
 	"go.ntppool.org/common/tracing"
@@ -27,7 +27,7 @@ func (srv *Server) zoneCounts(c echo.Context) error {
 
 	zone, err := q.GetZoneByName(ctx, c.Param("zone_name"))
 	if err != nil || zone.ID == 0 {
-		if errors.Is(err, sql.ErrNoRows) {
+		if errors.Is(err, pgx.ErrNoRows) {
 			return c.String(http.StatusNotFound, "Not found")
 		}
 		log.ErrorContext(ctx, "could not query for zone", "err", err)
@@ -37,7 +37,7 @@ func (srv *Server) zoneCounts(c echo.Context) error {
 
 	counts, err := q.GetZoneCounts(ctx, zone.ID)
 	if err != nil {
-		if !errors.Is(err, sql.ErrNoRows) {
+		if !errors.Is(err, pgx.ErrNoRows) {
 			log.ErrorContext(ctx, "get counts", "err", err)
 			span.RecordError(err)
 			return c.String(http.StatusInternalServerError, "internal error")
@@ -71,7 +71,7 @@ func (srv *Server) zoneCounts(c echo.Context) error {
 		count := 0
 		dates := map[int64]bool{}
 		for _, c := range counts {
-			ep := c.Date.Unix()
+			ep := c.Date.Time.Unix()
 			if _, ok := dates[ep]; !ok {
 				count++
 				dates[ep] = true
@@ -99,13 +99,13 @@ func (srv *Server) zoneCounts(c echo.Context) error {
 	lastSkip := int64(0)
 	skipThreshold := 0.5
 	for _, c := range counts {
-		cDate := c.Date.Unix()
+		cDate := c.Date.Time.Unix()
 		if (toSkip <= skipThreshold && cDate != lastSkip) ||
 			lastDate == cDate ||
 			mostRecentDate == cDate {
-			// log.Info("adding date", "date", c.Date.Format(time.DateOnly))
+			// log.Info("adding date", "date", c.Date.Time.Format(time.DateOnly))
 			rv.History = append(rv.History, historyEntry{
-				D:  c.Date.Format(time.DateOnly),
+				D:  c.Date.Time.Format(time.DateOnly),
 				Ts: int(cDate),
 				Ac: int(c.CountActive),
 				Rc: int(c.CountRegistered),

sqlc.yaml

@@ -2,16 +2,17 @@ version: "2"
 sql:
   - schema: "schema.sql"
     queries: "query.sql"
-    engine: "mysql"
+    engine: "postgresql"
+    strict_order_by: false
     gen:
       go:
         package: "ntpdb"
         out: "ntpdb"
+        sql_package: "pgx/v5"
         emit_json_tags: true
         emit_db_tags: true
         omit_unused_structs: true
         emit_interface: true
-        # emit_all_enum_values: true
         rename:
           servers.Ip: IP
         overrides: