Merge commit '467ac65'

Author: TimVosch

.golangci.yml

@@ -3,5 +3,12 @@ version: "2"
 linters:
   enable:
     - sqlclosecheck
+  settings:
+    errcheck:
+      exclude-functions:
+        - fmt.Fprintf
+        - (io.ReadCloser).Close
+        - (*sql.Rows).Close
+        - (*sqlx.Rows).Close
 run:
   timeout: 30m

internal/dbutils/collect.go

@@ -0,0 +1,106 @@
+package dbutils
+
+import (
+	"database/sql"
+	"errors"
+)
+
+var ErrTooManyRows = errors.New("too many rows returned")
+
+// RowToFunc is a function that scans or otherwise converts row to a T.
+type RowToFunc[T any] func(row *sql.Rows) (T, error)
+
+// AppendRows iterates through rows, calling fn for each row, and appending the results into a slice of T.
+//
+// This function closes the rows automatically on return.
+func AppendRows[T any, S ~[]T](slice S, rows *sql.Rows, fn RowToFunc[T]) (S, error) {
+	defer func() {
+		_ = rows.Close()
+	}()
+
+	for rows.Next() {
+		value, err := fn(rows)
+		if err != nil {
+			return nil, err
+		}
+		slice = append(slice, value)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return slice, nil
+}
+
+// CollectRows iterates through rows, calling fn for each row, and collecting the results into a slice of T.
+//
+// This function closes the rows automatically on return.
+func CollectRows[T any](rows *sql.Rows, fn RowToFunc[T]) ([]T, error) {
+	return AppendRows([]T{}, rows, fn)
+}
+
+// CollectOneRow calls fn for the first row in rows and returns the result. If no rows are found returns an error where errors.Is(ErrNoRows) is true.
+// CollectOneRow is to CollectRows as QueryRow is to Query.
+//
+// This function closes the rows automatically on return.
+func CollectOneRow[T any](rows *sql.Rows, fn RowToFunc[T]) (T, error) {
+	defer func() {
+		_ = rows.Close()
+	}()
+
+	var value T
+	var err error
+
+	if !rows.Next() {
+		if err = rows.Err(); err != nil {
+			return value, err
+		}
+		return value, sql.ErrNoRows
+	}
+
+	value, err = fn(rows)
+	if err != nil {
+		return value, err
+	}
+
+	_ = rows.Close()
+	return value, rows.Err()
+}
+
+// CollectExactlyOneRow calls fn for the first row in rows and returns the result.
+//   - If no rows are found returns an error where errors.Is(ErrNoRows) is true.
+//   - If more than 1 row is found returns an error where errors.Is(ErrTooManyRows) is true.
+//
+// This function closes the rows automatically on return.
+func CollectExactlyOneRow[T any](rows *sql.Rows, fn RowToFunc[T]) (T, error) {
+	defer func() {
+		_ = rows.Close()
+	}()
+
+	var (
+		err   error
+		value T
+	)
+
+	if !rows.Next() {
+		if err = rows.Err(); err != nil {
+			return value, err
+		}
+
+		return value, sql.ErrNoRows
+	}
+
+	value, err = fn(rows)
+	if err != nil {
+		return value, err
+	}
+
+	if rows.Next() {
+		var zero T
+
+		return zero, ErrTooManyRows
+	}
+
+	return value, rows.Err()
+}

internal/pagination/pagination.go

@@ -11,6 +11,7 @@ import (
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/fxamacker/cbor/v2"
+	"github.com/samber/lo"
 )
 
 const (
@@ -89,10 +90,6 @@ type Cursor[T any] struct {
 func CreatePageT[T1 any, T2 any](data []T1, cursor Cursor[T2]) Page[T1] {
 	var cursorString string
 
-	// TODO: Small bug, at this point the data has already been Limited and the data will always be equal to the Limit or smaller
-	// this means that a cursor will be made if the set is exactly equal to the limit but in reality, there is no more data
-	// meaning a new cursor is sent that won't result in any data
-	// https://github.com/sensorbucket/SensorBucket/issues/82
 	if len(data) >= int(cursor.Limit) {
 		cursorString = EncodeCursor(cursor)
 	}
@@ -137,24 +134,54 @@ func multiColumnCompare(columns []whereCol) sq.Sqlizer {
 	if len(columns) == 0 {
 		return nil
 	}
-	clause := sq.Or{}
-	for i := 0; i < len(columns); i++ {
-		and := sq.And{}
-		for j := 0; j <= i; j++ {
-			col := columns[j]
-			if j == i {
-				if col.order == "ASC" {
-					and = append(and, sq.Gt{col.column: col.value})
-				} else {
-					and = append(and, sq.Lt{col.column: col.value})
-				}
-				continue
-			}
-			and = append(and, sq.Eq{col.column: col.value})
+
+	// Determine the comparison operator based on the first column's order.
+	// This function assumes all columns in the cursor are sorted in the same direction (all ASC or all DESC).
+	// If mixed orders are present, the simple tuple comparison `(col1, col2) OP (val1, val2)` is semantically incorrect.
+	op := ""
+	switch columns[0].order {
+	case "ASC":
+		op = ">"
+	case "DESC":
+		op = "<"
+	default:
+		// This panic indicates an internal inconsistency: `whereCol.order` should only be "ASC" or "DESC".
+		panic(
+			fmt.Sprintf(
+				"multiColumnCompare: invalid order type %q encountered for column %q. Expected 'ASC' or 'DESC'.",
+				columns[0].order,
+				columns[0].column,
+			),
+		)
+	}
+
+	// Validate that all columns have the same sorting order.
+	// This is a strict requirement for the simple tuple comparison syntax used here.
+	for i := 1; i < len(columns); i++ {
+		if columns[i].order != columns[0].order {
+			// This panic indicates a misconfiguration in the pagination tags or an invalid cursor.
+			// Mixed ASC/DESC orders require a more complex WHERE clause (e.g., `(A > X) OR (A = X AND B < Y)`).
+			panic(
+				"multiColumnCompare: mixed ASC/DESC orders detected in pagination columns. This function supports only consistent ordering for tuple comparison.",
+			)
 		}
-		clause = append(clause, and)
 	}
-	return clause
+
+	placeholders := make([]string, len(columns))
+	for i := range columns {
+		placeholders[i] = "?"
+	}
+	columnNames := lo.Map(columns, func(col whereCol, _ int) string { return col.column })
+	columnValues := lo.Map(columns, func(col whereCol, _ int) any { return col.value })
+
+	return sq.Expr(
+		fmt.Sprintf(
+			"(%s) %s (%s)",
+			strings.Join(columnNames, ", "),
+			op,
+			strings.Join(placeholders, ","),
+		),
+		columnValues...)
 }
 
 func columnAlias(name string) string {
@@ -166,19 +193,14 @@ func columnAlias(name string) string {
 	return "paginated_" + name
 }
 
-func ApplyNoLimit[T any](q sq.SelectBuilder, c Cursor[T]) (sq.SelectBuilder, error) {
-	c.Limit = 0
-	return Apply(q, c)
-}
-
 func Apply[T any](q sq.SelectBuilder, c Cursor[T]) (sq.SelectBuilder, error) {
 	if c.Limit > 0 {
 		q = q.Limit(c.Limit)
 	}
 	rt := reflect.TypeOf(c.Columns)
 	rv := reflect.ValueOf(c.Columns)
 	columns := []whereCol{}
-	for ix := 0; ix < rt.NumField(); ix++ {
+	for ix := range rt.NumField() {
 		rf := rt.Field(ix)
 		if !rf.IsExported() {
 			continue
@@ -191,12 +213,20 @@ func Apply[T any](q sq.SelectBuilder, c Cursor[T]) (sq.SelectBuilder, error) {
 
 		tagParts := strings.Split(tag, ",")
 		if len(tagParts) != 2 {
-			return q, fmt.Errorf("invalid pagination tag on struct %s, for field %s", rt.Name(), rf.Name)
+			return q, fmt.Errorf(
+				"invalid pagination tag on struct %s, for field %s",
+				rt.Name(),
+				rf.Name,
+			)
 		}
 
 		column, order := tagParts[0], strings.ToUpper(tagParts[1])
 		if order != "ASC" && order != "DESC" {
-			return q, fmt.Errorf("invalid order in pagination tag on struct %s, for field %s", rt.Name(), rf.Name)
+			return q, fmt.Errorf(
+				"invalid order in pagination tag on struct %s, for field %s",
+				rt.Name(),
+				rf.Name,
+			)
 		}
 		q = q.OrderBy(column + " " + order).Column(column + " AS " + columnAlias(column))
 

pkg/auth/permissions.go

@@ -79,8 +79,8 @@ var stringPermissionMap = lo.SliceToMap(allPermissions, func(item Permission) (s
 	return string(item), item
 })
 
-func (this Permissions) Fulfills(that Permissions) error {
-	_, missing := lo.Difference(this, that)
+func (have Permissions) Fulfills(want Permissions) error {
+	_, missing := lo.Difference(have, want)
 	if len(missing) > 0 {
 		return fmt.Errorf("missing: %v", missing)
 	}

pkg/mq/amqp_connection.go

@@ -57,7 +57,12 @@ func (c *AMQPConnection) Start() {
 		c.usersLock.Unlock()
 		if c.connection != nil {
 			c.state = AMQP_DISCONNECTED
-			c.connection.Close()
+			if err := c.connection.Close(); err != nil {
+				log.Printf(
+					"AMQPConnection close returned an error but we're treating this as closed anyways: %s\n",
+					err.Error(),
+				)
+			}
 		}
 		log.Println("AMQPConnection stopped")
 	}()
@@ -109,7 +114,12 @@ func (c *AMQPConnection) Start() {
 
 		// Disconnected, so close to be sure
 		log.Printf("AMQPConnection disconnected\n")
-		c.connection.Close()
+		if err := c.connection.Close(); err != nil {
+			log.Printf(
+				"AMQPConnection close returned an error but we're treating this as closed anyways: %s\n",
+				err.Error(),
+			)
+		}
 	}
 }
 

services/core/devices/infra/query_builder.go

@@ -2,11 +2,13 @@ package deviceinfra
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/jmoiron/sqlx"
 
+	"sensorbucket.nl/sensorbucket/internal/dbutils"
 	"sensorbucket.nl/sensorbucket/internal/pagination"
 	"sensorbucket.nl/sensorbucket/pkg/auth"
 	"sensorbucket.nl/sensorbucket/services/core/devices"
@@ -75,7 +77,10 @@ func (b deviceQueryBuilder) WithinRange(r devices.RangeFilter) deviceQueryBuilde
 	return b
 }
 
-func (b deviceQueryBuilder) Query(ctx context.Context, db *sqlx.DB) (*pagination.Page[devices.Device], error) {
+func (b deviceQueryBuilder) Query(
+	ctx context.Context,
+	db *sqlx.DB,
+) (*pagination.Page[devices.Device], error) {
 	if b.err != nil {
 		return nil, b.err
 	}
@@ -92,7 +97,10 @@ func (b deviceQueryBuilder) Query(ctx context.Context, db *sqlx.DB) (*pagination
 	}
 	if len(b.filters.Sensor) > 0 {
 		// Update query here
-		subQ, subArgs, err := sq.Select("DISTINCT sensor.device_id").From("sensors sensor").Where(sq.Eq{"sensor.id": b.filters.Sensor}).ToSql()
+		subQ, subArgs, err := sq.Select("DISTINCT sensor.device_id").
+			From("sensors sensor").
+			Where(sq.Eq{"sensor.id": b.filters.Sensor}).
+			ToSql()
 		if err != nil {
 			return nil, err
 		}
@@ -113,30 +121,31 @@ func (b deviceQueryBuilder) Query(ctx context.Context, db *sqlx.DB) (*pagination
 	if err != nil {
 		return nil, err
 	}
-	defer rows.Close()
-
-	deviceModels := []DeviceModel{}
-	for rows.Next() {
-		var model DeviceModel
-		err := rows.Scan(
-			&model.ID,
-			&model.Code,
-			&model.Description,
-			&model.TenantID,
-			&model.Properties,
-			&model.LocationDescription,
-			&model.Longitude,
-			&model.Latitude,
-			&model.Altitude,
-			&model.State,
-			&model.CreatedAt,
-			&b.cursor.Columns.CreatedAt,
-			&b.cursor.Columns.ID,
-		)
-		if err != nil {
-			return nil, err
-		}
-		deviceModels = append(deviceModels, model)
+
+	deviceModels, err := dbutils.CollectRows(
+		rows,
+		func(row *sql.Rows) (DeviceModel, error) {
+			var model DeviceModel
+			err := rows.Scan(
+				&model.ID,
+				&model.Code,
+				&model.Description,
+				&model.TenantID,
+				&model.Properties,
+				&model.LocationDescription,
+				&model.Longitude,
+				&model.Latitude,
+				&model.Altitude,
+				&model.State,
+				&model.CreatedAt,
+				&b.cursor.Columns.CreatedAt,
+				&b.cursor.Columns.ID,
+			)
+			return model, err
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("devices query builder collecting rows: %w", err)
 	}
 
 	ids := make([]int64, len(deviceModels))

services/core/featuresofinterest/feature_of_interest.go

@@ -29,7 +29,7 @@ type CreateFeatureOfInterestOpts struct {
 
 func NewFeatureOfInterest(opts CreateFeatureOfInterestOpts) (*FeatureOfInterest, error) {
 	if opts.Name == "" || opts.TenantID == 0 {
-		return nil, fmt.Errorf("In NewFeatureOfInterest: missing required Name or TenantID")
+		return nil, fmt.Errorf("in NewFeatureOfInterest: missing required Name or TenantID")
 	}
 
 	var foi FeatureOfInterest