chore: paralleltest

.golangci.yaml

@@ -33,7 +33,6 @@ linters:
     - godox # TODO
     - nestif # TODO
     - nlreturn # TODO
-    - paralleltest # TODO
     - protogetter # TODO
     - revive # TODO
     - testpackage # TODO

libs/common/auth/auth_test.go

@@ -7,6 +7,8 @@ import (
 )
 
 func TestIDTokenClaims_AsExpected(t *testing.T) {
+	t.Parallel()
+
 	tests := []struct {
 		name          string
 		claims        IDTokenClaims
@@ -95,6 +97,8 @@ func TestIDTokenClaims_AsExpected(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
 			err := tt.claims.AsExpected()
 			if tt.expectedError {
 				require.Error(t, err)

libs/common/go.mod

@@ -3,6 +3,7 @@ module common
 go 1.23
 
 replace (
+	hwdb => ../hwdb
 	hwlocale => ../hwlocale
 	hwutil => ../hwutil
 	telemetry => ../telemetry
@@ -18,6 +19,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.2.0
 	github.com/joho/godotenv v1.5.1
 	github.com/nicksnyder/go-i18n/v2 v2.4.1
+	github.com/pashagolub/pgxmock/v4 v4.3.0
 	github.com/prometheus/client_golang v1.20.5
 	github.com/rs/zerolog v1.33.0
 	github.com/stretchr/testify v1.10.0
@@ -33,6 +35,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576
 	google.golang.org/grpc v1.69.2
 	google.golang.org/protobuf v1.36.1
+	hwdb v0.0.0
 	hwlocale v0.0.0
 	hwutil v0.0.0
 	telemetry v0.0.0
@@ -51,6 +54,11 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0 // indirect
+	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa // 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.2 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
@@ -62,12 +70,14 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.55.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/vgarvardt/pgx-google-uuid/v5 v5.6.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0 // indirect
 	go.opentelemetry.io/otel/metric v1.33.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.4.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576 // indirect
 	gopkg.in/square/go-jose.v2 v2.6.0 // indirect

libs/common/go.sum

@@ -50,6 +50,16 @@ github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.2.0 h1:kQ0NI7W1B3HwiN5gAYtY+X
 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.2.0/go.mod h1:zrT2dxOAjNFPRGjTUe2Xmb4q4YdUwVvQFV6xiCSf+z0=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0 h1:TmHmbvxPmaegwhDubVz0lICL0J5Ka2vwTzhoePEXsGE=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0/go.mod h1:qztMSjm835F2bXf+5HKAPIS5qsmQDqZna/PgVt4rWtI=
+github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6hFJUX53drDT4UsSW3DEhKn0ifuHw=
+github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+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.2 h1:mLoDLV6sonKlvjIEsV56SkWNCnuNv531l94GaIzO+XI=
+github.com/jackc/pgx/v5 v5.7.2/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ=
+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/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
@@ -74,6 +84,8 @@ github.com/nicksnyder/go-i18n/v2 v2.4.1 h1:zwzjtX4uYyiaU02K5Ia3zSkpJZrByARkRB4V3
 github.com/nicksnyder/go-i18n/v2 v2.4.1/go.mod h1:++Pl70FR6Cki7hdzZRnEEqdc2dJt+SAGotyFg/SvZMk=
 github.com/openzipkin/zipkin-go v0.4.3 h1:9EGwpqkgnwdEIJ+Od7QVSEIH+ocmm5nPat0G7sjsSdg=
 github.com/openzipkin/zipkin-go v0.4.3/go.mod h1:M9wCJZFWCo2RiY+o1eBCEMe0Dp2S5LDHcMZmk3RmK7c=
+github.com/pashagolub/pgxmock/v4 v4.3.0 h1:DqT7fk0OCK6H0GvqtcMsLpv8cIwWqdxWgfZNLeHCb/s=
+github.com/pashagolub/pgxmock/v4 v4.3.0/go.mod h1:9VoVHXwS3XR/yPtKGzwQvwZX1kzGB9sM8SviDcHDa3A=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
@@ -94,9 +106,13 @@ github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
 github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/vgarvardt/pgx-google-uuid/v5 v5.6.0 h1:EhPtK0mgrgaTMXpegE69hvoSOVC1Ahk8+QJ9B8b+OdU=
+github.com/vgarvardt/pgx-google-uuid/v5 v5.6.0/go.mod h1:5LtFrNEkgzxHvXPO9eOvcXsSn9/KeKYgx9kjeI2oXQI=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.58.0 h1:PS8wXpbyaDJQ2VDHHncMe9Vct0Zn1fEjpsjrLxGJoSc=
@@ -131,6 +147,8 @@ golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
 golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

libs/common/grpc.go

@@ -2,11 +2,14 @@ package common
 
 import (
 	"context"
+	"hwdb"
 	"hwutil"
 	"net"
 	"telemetry"
 
-	auth "common/auth"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"common/auth"
 	"common/hwgrpc"
 
 	"github.com/dapr/dapr/pkg/proto/runtime/v1"
@@ -41,7 +44,7 @@ func StartNewGRPCServer(ctx context.Context, addr string, registerServerHook fun
 	}
 
 	// dapr/grpc service
-	service, ok := daprd.NewServiceWithListener(listener, DefaultServerOptions()...).(*daprd.Server)
+	service, ok := daprd.NewServiceWithListener(listener, DefaultServerOptions(ctx)...).(*daprd.Server)
 	if !ok {
 		log.Fatal().Msg("dapr service listener is not a *daprd.Server")
 	}
@@ -87,13 +90,13 @@ func StartNewGRPCServer(ctx context.Context, addr string, registerServerHook fun
 // DefaultUnaryInterceptors returns the slice of default interceptors for unary gRPC calls
 //
 //	chain := grpc.ChainUnaryInterceptor(common.DefaultUnaryInterceptors()...)
-func DefaultUnaryInterceptors(metrics *prometheusGrpcProvider.ServerMetrics) []grpc.UnaryServerInterceptor {
+func DefaultUnaryInterceptors(ctx context.Context, panicsRecovered prometheus.Counter) []grpc.UnaryServerInterceptor {
 	return []grpc.UnaryServerInterceptor{
-		metrics.UnaryServerInterceptor(),
-		hwgrpc.UnaryPanicRecoverInterceptor(),
+		hwgrpc.UnaryPanicRecoverInterceptor(panicsRecovered),
 		hwgrpc.UnaryLoggingInterceptor,
 		hwgrpc.UnaryErrorQualityControlInterceptor,
 		hwgrpc.UnaryLocaleInterceptor,
+		hwgrpc.UnaryDBInterceptor(hwdb.GetDB(ctx)),
 		defaultUnaryAuthInterceptor,
 		defaultUnaryOrganizationInterceptor,
 		hwgrpc.UnaryValidateInterceptor,
@@ -104,27 +107,46 @@ func DefaultUnaryInterceptors(metrics *prometheusGrpcProvider.ServerMetrics) []g
 // DefaultStreamInterceptors returns the slice of default interceptors for stream gRPC calls
 //
 //	chain := grpc.ChainStreamInterceptor(common.DefaultStreamInterceptors()...)
-func DefaultStreamInterceptors(metrics *prometheusGrpcProvider.ServerMetrics) []grpc.StreamServerInterceptor {
+func DefaultStreamInterceptors(ctx context.Context, panicsRecovered prometheus.Counter) []grpc.StreamServerInterceptor {
 	return []grpc.StreamServerInterceptor{
-		metrics.StreamServerInterceptor(),
-		hwgrpc.StreamPanicRecoverInterceptor(),
+		hwgrpc.StreamPanicRecoverInterceptor(panicsRecovered),
 		hwgrpc.StreamLoggingInterceptor,
 		hwgrpc.StreamErrorQualityControlInterceptor,
 		hwgrpc.StreamLocaleInterceptor,
+		hwgrpc.StreamDBInterceptor(hwdb.GetDB(ctx)),
 		defaultStreamAuthInterceptor,
 		defaultStreamOrganizationInterceptor,
 		hwgrpc.StreamValidateInterceptor,
 		hwgrpc.StreamSpanInterceptor,
 	}
 }
 
-func DefaultServerOptions() []grpc.ServerOption {
+// DefaultServerOptions respects telemetry.WithPrometheusRegistry and hwdb.WithDB context values.
+// These will then be propagated into request contexts, if provided.
+func DefaultServerOptions(ctx context.Context) []grpc.ServerOption {
+	// counters
+	panicsRecovered := hwgrpc.NewPanicsRecoveredCounter(ctx)
+
+	// default interceptors
+	unaryInterceptors := DefaultUnaryInterceptors(ctx, panicsRecovered)
+	streamInterceptors := DefaultStreamInterceptors(ctx, panicsRecovered)
+
 	// register new metrics collector with prometheus
-	metrics := prometheusGrpcProvider.NewServerMetrics()
-	telemetry.PrometheusRegistry().MustRegister(metrics)
+	promRegistry := telemetry.PrometheusRegistry(ctx)
+	if promRegistry != nil {
+		metrics := prometheusGrpcProvider.NewServerMetrics()
+		promRegistry.MustRegister(metrics)
+
+		// prepend metrics interceptor
+		unaryInterceptors = hwutil.Prepend(metrics.UnaryServerInterceptor(), unaryInterceptors)
+		streamInterceptors = hwutil.Prepend(metrics.StreamServerInterceptor(), streamInterceptors)
+	}
+
+	// create chains
+	unaryInterceptorChain := grpc.ChainUnaryInterceptor(unaryInterceptors...)
+	streamInterceptorChain := grpc.ChainStreamInterceptor(streamInterceptors...)
 
-	unaryInterceptorChain := grpc.ChainUnaryInterceptor(DefaultUnaryInterceptors(metrics)...)
-	streamInterceptorChain := grpc.ChainStreamInterceptor(DefaultStreamInterceptors(metrics)...)
+	// otel stats handler
 	statsHandler := grpc.StatsHandler(otelgrpc.NewServerHandler())
 
 	return []grpc.ServerOption{unaryInterceptorChain, streamInterceptorChain, statsHandler}

libs/common/hwgrpc/db_interceptor.go

@@ -0,0 +1,32 @@
+package hwgrpc
+
+import (
+	"context"
+	"hwdb"
+
+	"google.golang.org/grpc"
+)
+
+func UnaryDBInterceptor(db hwdb.DBTX) grpc.UnaryServerInterceptor {
+	return func(
+		ctx context.Context,
+		req any,
+		_ *grpc.UnaryServerInfo,
+		next grpc.UnaryHandler,
+	) (any, error) {
+		return next(hwdb.WithDB(ctx, db), req)
+	}
+}
+
+func StreamDBInterceptor(db hwdb.DBTX) grpc.StreamServerInterceptor {
+	return func(
+		req any,
+		stream grpc.ServerStream,
+		_ *grpc.StreamServerInfo,
+		next grpc.StreamHandler,
+	) error {
+		ctx := hwdb.WithDB(stream.Context(), db)
+		stream = WrapServerStream(stream, ctx)
+		return next(req, stream)
+	}
+}

libs/common/hwgrpc/db_interceptor_test.go

@@ -0,0 +1,118 @@
+package hwgrpc_test
+
+import (
+	"context"
+	"hwdb"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/pashagolub/pgxmock/v4"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"common/hwgrpc"
+)
+
+func TestDBInterceptor(t *testing.T) {
+	t.Parallel()
+
+	type testCase struct {
+		db      hwdb.DBTX
+		handler func(t *testing.T, ctx context.Context)
+	}
+
+	existingDB, err := pgxmock.NewPool()
+	if err != nil {
+		panic(err)
+	}
+
+	testMatrix := map[string]testCase{
+		"missing db": {
+			db: nil,
+			handler: func(t *testing.T, ctx context.Context) {
+				t.Helper()
+
+				// GetDB will return nil
+				require.Nil(t, hwdb.GetDB(ctx))
+
+				// MustGetDB will not work
+				require.Panics(t, func() {
+					hwdb.MustGetDB(ctx)
+				})
+			},
+		},
+		"existing db": {
+			db: existingDB,
+			handler: func(t *testing.T, ctx context.Context) {
+				t.Helper()
+
+				require.Equal(t, existingDB, hwdb.MustGetDB(ctx))
+			},
+		},
+	}
+
+	for name, test := range testMatrix {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			ctx := context.Background()
+
+			//
+			// Stream
+			//
+
+			streamInterceptor := hwgrpc.StreamDBInterceptor(test.db)
+			streamHandlerWasCalled := false
+
+			streamNext := func(srv any, stream grpc.ServerStream) error {
+				streamHandlerWasCalled = true
+				test.handler(t, stream.Context())
+				return nil
+			}
+
+			stream := &fakeServerStream{ctx: ctx}
+			err = streamInterceptor(ctx, stream, nil, streamNext)
+
+			// next handler is properly called
+			require.NoError(t, err)
+			require.True(t, streamHandlerWasCalled)
+
+			//
+			// Unary
+			//
+			unaryInterceptor := hwgrpc.UnaryDBInterceptor(test.db)
+
+			expValue := uuid.New()
+			unaryHandlerWasCalled := false
+
+			unaryNext := func(ctx context.Context, req any) (any, error) {
+				unaryHandlerWasCalled = true
+				test.handler(t, ctx)
+				return expValue, nil
+			}
+
+			value, err := unaryInterceptor(ctx, nil, nil, unaryNext)
+
+			// unaryNext handler is properly called
+			require.NoError(t, err)
+			require.True(t, unaryHandlerWasCalled)
+			require.Equal(t, expValue, value)
+		})
+	}
+}
+
+type fakeServerStream struct {
+	grpc.ServerStream
+	ctx context.Context //nolint:containedctx
+}
+
+func (f *fakeServerStream) Context() context.Context {
+	return f.ctx
+}
+
+func (f *fakeServerStream) SendMsg(_ any) error {
+	return nil
+}
+
+func (f *fakeServerStream) RecvMsg(_ any) error {
+	return nil
+}

libs/common/hwgrpc/hwgrpc_test.go

@@ -21,6 +21,7 @@ func arrayEq(t *testing.T, expected []language.Tag, parsed []language.Tag) {
 }
 
 func TestParseLocales(t *testing.T) {
+	t.Parallel()
 	const mdnExample = "fr-CH, fr;q=0.9, en;q=0.8, de;q=0.7, *;q=0.5"
 	expected := []language.Tag{language.MustParse("fr-CH"), language.French, language.English, language.German}
 	parsed, ok := hwgrpc.ParseLocales(mdnExample)
@@ -32,6 +33,7 @@ func TestParseLocales(t *testing.T) {
 }
 
 func TestParseLocalesReordered(t *testing.T) {
+	t.Parallel()
 	const mdnExample = "fr-CH, de;q=0.7, en;q=0.8, fr;q=0.9, *;q=0.5"
 	expected := []language.Tag{language.MustParse("fr-CH"), language.French, language.English, language.German}
 	parsed, ok := hwgrpc.ParseLocales(mdnExample)
@@ -43,6 +45,7 @@ func TestParseLocalesReordered(t *testing.T) {
 }
 
 func TestParseLocalesSimple(t *testing.T) {
+	t.Parallel()
 	const mdnExample = "de"
 	expected := []language.Tag{language.German}
 	parsed, ok := hwgrpc.ParseLocales(mdnExample)