Use sqlx Instead of String Formatting in TimescaleDB Provider

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/influxdata/influxdb-client-go/v2 v2.14.0
 	github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf
+	github.com/jmoiron/sqlx v1.4.0
 	github.com/lib/pq v1.10.9
 	github.com/mattn/go-sqlite3 v1.14.28
 	github.com/pelletier/go-toml/v2 v2.2.3

go.sum

@@ -1,3 +1,5 @@
+filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
+filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
@@ -14,6 +16,8 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
+github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
@@ -26,13 +30,16 @@ github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjw
 github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI=
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf h1:7JTmneyiNEwVBOHSjoMxiWAqB992atOeepeFYegn5RU=
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
+github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o=
+github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY=
 github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/mattn/go-sqlite3 v1.14.28 h1:ThEiQrnbtumT+QMknw63Befp/ce/nUPgBPMlRFEum7A=
 github.com/mattn/go-sqlite3 v1.14.28/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/oapi-codegen/runtime v1.1.1 h1:EXLHh0DXIJnWhdRPN2w4MXAzFyE4CskzhNLUmtpMYro=

pkg/timescaledb/convert.go

@@ -2,24 +2,24 @@ package timescaledb
 
 import (
 	"context"
-	"database/sql"
 	"encoding/json"
 	"fmt"
 	"strconv"
 	stdtime "time"
 
+	"github.com/jmoiron/sqlx"
 	"github.com/nasa/hermes/pkg/pb"
 )
 
 const (
 	insertEventDefSQL = `INSERT INTO eventDefs (id, component, name, severity, args)
-		VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`
+		VALUES (:id, :component, :name, :severity, :args) ON CONFLICT DO NOTHING`
 	insertEventSQL = `INSERT INTO events (eventDefId, time, timeSclk, message, source, args, ert)
-		VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING`
+		VALUES (:eventDefId, :time, :timeSclk, :message, :source, :args, :ert) ON CONFLICT DO NOTHING`
 	insertTelemetryDefSQL = `INSERT INTO telemetryDefs (id, name, component)
-		VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`
+		VALUES (:id, :name, :component) ON CONFLICT DO NOTHING`
 	insertTelemetrySQL = `INSERT INTO telemetry (time, telemetryDefId, timeSclk, source, labels, key, valueType, integral, floating, boolval, string, bytes, ert)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT DO NOTHING`
+		VALUES (:time, :telemetryDefId, :timeSclk, :source, :labels, :key, :valueType, :integral, :floating, :boolval, :string, :bytes, :ert) ON CONFLICT DO NOTHING`
 )
 
 func valuesToAnys(values []*pb.Value) ([]any, error) {
@@ -34,7 +34,7 @@ func valuesToAnys(values []*pb.Value) ([]any, error) {
 	return valueAnys, nil
 }
 
-func InsertEvent(ctx context.Context, db *sql.DB, msg *pb.SourcedEvent) error {
+func InsertEvent(ctx context.Context, db *sqlx.DB, msg *pb.SourcedEvent) error {
 	event := msg.GetEvent()
 
 	eventArgsArray, err := valuesToAnys(event.GetArgs())
@@ -52,30 +52,39 @@ func InsertEvent(ctx context.Context, db *sql.DB, msg *pb.SourcedEvent) error {
 		return fmt.Errorf("failed to marshal def args: %w", err)
 	}
 
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTxx(ctx, nil)
 	if err != nil {
 		return fmt.Errorf("failed to begin transaction: %w", err)
 	}
 	defer tx.Rollback()
 
 	ref := event.GetRef()
-	if _, err := tx.ExecContext(ctx, insertEventDefSQL,
-		ref.GetId(), ref.GetComponent(), ref.GetName(), ref.GetSeverity(), string(defArgs),
-	); err != nil {
+	if _, err := tx.NamedExecContext(ctx, insertEventDefSQL, map[string]any{
+		"id":        ref.GetId(),
+		"component": ref.GetComponent(),
+		"name":      ref.GetName(),
+		"severity":  ref.GetSeverity(),
+		"args":      string(defArgs),
+	}); err != nil {
 		return fmt.Errorf("failed to insert event def: %w", err)
 	}
 
-	if _, err := tx.ExecContext(ctx, insertEventSQL,
-		ref.GetId(), event.GetTime().GetUnix().AsTime(), event.GetTime().GetSclk(),
-		event.GetMessage(), msg.GetSource(), string(eventArgs), stdtime.Now(),
-	); err != nil {
+	if _, err := tx.NamedExecContext(ctx, insertEventSQL, map[string]any{
+		"eventDefId": ref.GetId(),
+		"time":       event.GetTime().GetUnix().AsTime(),
+		"timeSclk":   event.GetTime().GetSclk(),
+		"message":    event.GetMessage(),
+		"source":     msg.GetSource(),
+		"args":       string(eventArgs),
+		"ert":        stdtime.Now(),
+	}); err != nil {
 		return fmt.Errorf("failed to insert event: %w", err)
 	}
 
 	return tx.Commit()
 }
 
-func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry) error {
+func InsertTelemetry(ctx context.Context, db *sqlx.DB, msg *pb.SourcedTelemetry) error {
 	tlm := msg.GetTelemetry()
 	def := tlm.GetRef()
 
@@ -84,15 +93,17 @@ func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry)
 		return fmt.Errorf("failed to marshal telemetry labels: %w", err)
 	}
 
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTxx(ctx, nil)
 	if err != nil {
 		return fmt.Errorf("failed to begin transaction: %w", err)
 	}
 	defer tx.Rollback()
 
-	if _, err := tx.ExecContext(ctx, insertTelemetryDefSQL,
-		def.GetId(), def.GetName(), def.GetComponent(),
-	); err != nil {
+	if _, err := tx.NamedExecContext(ctx, insertTelemetryDefSQL, map[string]any{
+		"id":        def.GetId(),
+		"name":      def.GetName(),
+		"component": def.GetComponent(),
+	}); err != nil {
 		return fmt.Errorf("failed to insert telemetry def: %w", err)
 	}
 
@@ -103,7 +114,7 @@ func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry)
 	return tx.Commit()
 }
 
-func insertValue(ctx context.Context, tx *sql.Tx, time *pb.Time, telemetryDefId int32, source string, labels string, path string, value *pb.Value) error {
+func insertValue(ctx context.Context, tx *sqlx.Tx, time *pb.Time, telemetryDefId int32, source string, labels string, path string, value *pb.Value) error {
 	var (
 		valueType          string
 		integral, floating any
@@ -151,11 +162,20 @@ func insertValue(ctx context.Context, tx *sql.Tx, time *pb.Time, telemetryDefId
 		bytes = valueTy.R.Value
 	}
 
-	now := stdtime.Now()
-	_, err := tx.ExecContext(ctx, insertTelemetrySQL,
-		time.GetUnix().AsTime(), telemetryDefId, time.GetSclk(),
-		source, labels, path, valueType,
-		integral, floating, boolval, str, bytes, now,
-	)
+	_, err := tx.NamedExecContext(ctx, insertTelemetrySQL, map[string]any{
+		"time":           time.GetUnix().AsTime(),
+		"telemetryDefId": telemetryDefId,
+		"timeSclk":       time.GetSclk(),
+		"source":         source,
+		"labels":         labels,
+		"key":            path,
+		"valueType":      valueType,
+		"integral":       integral,
+		"floating":       floating,
+		"boolval":        boolval,
+		"string":         str,
+		"bytes":          bytes,
+		"ert":            stdtime.Now(),
+	})
 	return err
 }

pkg/timescaledb/timescaledb.go

@@ -6,8 +6,7 @@ import (
 
 	_ "embed"
 
-	"database/sql"
-
+	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
 	"github.com/nasa/hermes/pkg/host"
 	"github.com/nasa/hermes/pkg/pb"
@@ -54,7 +53,7 @@ func (t *timescaleDbProvider) Start(
 		dsn,
 	)
 
-	db, err := sql.Open("postgres", dsn)
+	db, err := sqlx.Open("postgres", dsn)
 	if err != nil {
 		return fmt.Errorf("failed to open timescaledb connection: %w", err)
 	}