enterprise-portal: init database schema and handler store (#63139)

Part of CORE-99

This PR scaffolds the database schema and code structure based on
[CORE-99
comment](https://linear.app/sourcegraph/issue/CORE-99/enterprise-portal-design-sams-user-to-subscription-rpcs#comment-8105ac31)
with some modifications. See inline comments for more elaborations.
- It uses GORM's ONLY for auto migration, just to kick things off, we
may migrate to file-based migration like we are planning for SAMS.
- It then uses the `*pgxpool.Pool` as the DB interface for executing
business logic queries.

Additionally, refactored `subscriptionsservice/v1.go` to use a `Store`
that provide single interface for accessing data(base), as we have been
doing in SAMS and SSC.

## Test plan

Enterprise Portal starts locally, and database is initialized:

![CleanShot 2024-06-06 at 17 02
42@2x](https://github.com/sourcegraph/sourcegraph/assets/2946214/f6cbc2bf-bd95-4691-9f87-b2450cc31e4d)

cmd/enterprise-portal/README.md

@@ -1,3 +1,20 @@
 # enterprise-portal
 
 **WIP** - refer to [RFC 885 Sourcegraph Enterprise Portal (go/enterprise-portal)](https://docs.google.com/document/d/1tiaW1IVKm_YSSYhH-z7Q8sv4HSO_YJ_Uu6eYDjX7uU4/edit#heading=h.tdaxc5h34u7q) for more details.
+
+There are some services that are expected to be running by the Enterprise Portal:
+
+- PostgreSQL with a database named `enterprise-portal`
+- Redis running on `localhost:6379`
+
+To start the Enterprise Portal, run:
+
+```zsh
+sg run enterprise-portal
+```
+
+To customize the PostgreSQL and Redis connection strings, customize the following environment variables in your `sg.config.overwrite.yaml` file:
+
+- `PGDSN`: PostgreSQL connection string
+- `REDIS_HOST`: Redis host
+- `REDIS_PORT`: Redis port

cmd/enterprise-portal/internal/database/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "database",
+    srcs = [
+        "database.go",
+        "migrate.go",
+        "permissions.go",
+        "subscriptions.go",
+    ],
+    importpath = "github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/database",
+    visibility = ["//cmd/enterprise-portal:__subpackages__"],
+    deps = [
+        "//internal/redislock",
+        "//lib/errors",
+        "//lib/managedservicesplatform/runtime",
+        "@com_github_jackc_pgx_v5//pgxpool",
+        "@com_github_redis_go_redis_v9//:go-redis",
+        "@com_github_sourcegraph_log//:log",
+        "@io_gorm_driver_postgres//:postgres",
+        "@io_gorm_gorm//:gorm",
+        "@io_gorm_gorm//logger",
+        "@io_gorm_plugin_opentelemetry//tracing",
+        "@io_opentelemetry_go_otel//:otel",
+        "@io_opentelemetry_go_otel//attribute",
+        "@io_opentelemetry_go_otel//codes",
+        "@io_opentelemetry_go_otel_trace//:trace",
+    ],
+)

cmd/enterprise-portal/internal/database/database.go

@@ -0,0 +1,44 @@
+package database
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/redis/go-redis/v9"
+	"github.com/sourcegraph/log"
+	"go.opentelemetry.io/otel"
+
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+	"github.com/sourcegraph/sourcegraph/lib/managedservicesplatform/runtime"
+)
+
+var databaseTracer = otel.Tracer("enterprise-portal/internal/database")
+
+// DB is the database handle for the storage layer.
+type DB struct {
+	db *pgxpool.Pool
+}
+
+// ⚠️ WARNING: This list is meant to be read-only.
+var allTables = []any{
+	&Subscription{},
+	&Permission{},
+}
+
+const databaseName = "enterprise-portal"
+
+// NewHandle returns a new database handle with the given configuration. It may
+// attempt to auto-migrate the database schema if the application version has
+// changed.
+func NewHandle(ctx context.Context, logger log.Logger, contract runtime.Contract, redisClient *redis.Client, currentVersion string) (*DB, error) {
+	err := maybeMigrate(ctx, logger, contract, redisClient, currentVersion)
+	if err != nil {
+		return nil, errors.Wrap(err, "maybe migrate")
+	}
+
+	pool, err := contract.PostgreSQL.GetConnectionPool(ctx, databaseName)
+	if err != nil {
+		return nil, errors.Wrap(err, "get connection pool")
+	}
+	return &DB{db: pool}, nil
+}

cmd/enterprise-portal/internal/database/migrate.go

@@ -0,0 +1,126 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/sourcegraph/log"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	gormlogger "gorm.io/gorm/logger"
+	"gorm.io/plugin/opentelemetry/tracing"
+
+	"github.com/sourcegraph/sourcegraph/internal/redislock"
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+	"github.com/sourcegraph/sourcegraph/lib/managedservicesplatform/runtime"
+)
+
+// maybeMigrate runs the auto-migration for the database when needed based on
+// the given version.
+func maybeMigrate(ctx context.Context, logger log.Logger, contract runtime.Contract, redisClient *redis.Client, currentVersion string) (err error) {
+	ctx, span := databaseTracer.Start(
+		ctx,
+		"database.maybeMigrate",
+		trace.WithAttributes(
+			attribute.String("currentVersion", currentVersion),
+		),
+	)
+	defer func() {
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+		}
+		span.End()
+	}()
+
+	sqlDB, err := contract.PostgreSQL.OpenDatabase(ctx, databaseName)
+	if err != nil {
+		return errors.Wrap(err, "open database")
+	}
+	defer func() {
+		err := sqlDB.Close()
+		if err != nil {
+			logger.Error("failed to close database for migration", log.Error(err))
+		}
+	}()
+
+	conn, err := gorm.Open(
+		postgres.New(postgres.Config{Conn: sqlDB}),
+		&gorm.Config{
+			SkipDefaultTransaction: true,
+			NowFunc: func() time.Time {
+				return time.Now().UTC().Truncate(time.Microsecond)
+			},
+		},
+	)
+	if err != nil {
+		return errors.Wrap(err, "open connection")
+	}
+
+	if err = conn.Use(tracing.NewPlugin()); err != nil {
+		return errors.Wrap(err, "initialize tracing plugin")
+	}
+
+	// We want to make sure only one instance of the server is doing auto-migration
+	// at a time.
+	return redislock.OnlyOne(
+		logger,
+		redisClient,
+		fmt.Sprintf("%s:auto-migrate", databaseName),
+		15*time.Second,
+		func() error {
+			versionKey := fmt.Sprintf("%s:db_version", databaseName)
+			span.AddEvent("lock.acquired")
+
+			if shouldSkipMigration(
+				redisClient.Get(context.Background(), versionKey).Val(),
+				currentVersion,
+			) {
+				logger.Info("skipped auto-migration",
+					log.String("database", databaseName),
+					log.String("currentVersion", currentVersion),
+				)
+				span.SetAttributes(attribute.Bool("skipped", true))
+				return nil
+			}
+
+			// Create a session that ignore debug logging.
+			sess := conn.Session(&gorm.Session{
+				Logger: gormlogger.Default.LogMode(gormlogger.Warn),
+			})
+			// Auto-migrate database table definitions.
+			for _, table := range allTables {
+				err := sess.AutoMigrate(table)
+				if err != nil {
+					return errors.Wrapf(err, "auto migrating table for %s", errors.Safe(fmt.Sprintf("%T", table)))
+				}
+			}
+
+			return redisClient.Set(context.Background(), versionKey, currentVersion, 0).Err()
+		},
+	)
+}
+
+// shouldSkipMigration returns true if the migration should be skipped.
+func shouldSkipMigration(previousVersion, currentVersion string) bool {
+	// Skip for PR-builds.
+	if strings.HasPrefix(currentVersion, "_candidate") {
+		return true
+	}
+
+	const releaseBuildVersionExample = "277307_2024-06-06_5.4-9185da3c3e42"
+	// We always run the full auto-migration if the version is not release-build like.
+	if len(currentVersion) < len(releaseBuildVersionExample) ||
+		len(previousVersion) < len(releaseBuildVersionExample) {
+		return false
+	}
+
+	// The release build version is sorted lexicographically, so we can compare it as a string.
+	return previousVersion >= currentVersion
+}

cmd/enterprise-portal/internal/database/permissions.go

@@ -0,0 +1,20 @@
+package database
+
+import (
+	"time"
+)
+
+// Permission is a Zanzibar-inspired permission record.
+type Permission struct {
+	// Namespace is the namespace of the permission, e.g. "cody_analytics".
+	Namespace string `gorm:"not null;index"`
+	// Subject is the subject of the permission, e.g. "User:<SAMS account ID>".
+	Subject string `gorm:"not null;index"`
+	// Object is the object of the permission, e.g. "Subscription:<subscription ID>".
+	Object string `gorm:"not null"`
+	// Relation is the relationship between the subject and the object, e.g.
+	// "customer_admin".
+	Relation string `gorm:"not null"`
+	// CommitTime is the time when the permission was committed.
+	CommitTime time.Time `gorm:"not null"`
+}

cmd/enterprise-portal/internal/database/subscriptions.go

@@ -0,0 +1,10 @@
+package database
+
+// Subscription is a product subscription record.
+type Subscription struct {
+	// ID is the prefixed UUID-format identifier for the subscription.
+	ID string `gorm:"primaryKey"`
+	// InstanceDomain is the instance domain associated with the subscription, e.g.
+	// "acme.sourcegraphcloud.com".
+	InstanceDomain string `gorm:"index"`
+}

cmd/enterprise-portal/internal/subscriptionsservice/BUILD.bazel

@@ -5,12 +5,14 @@ go_library(
     srcs = [
         "adapters.go",
         "v1.go",
+        "v1_store.go",
     ],
     importpath = "github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/subscriptionsservice",
     tags = [TAG_INFRA_CORESERVICES],
     visibility = ["//cmd/enterprise-portal:__subpackages__"],
     deps = [
         "//cmd/enterprise-portal/internal/connectutil",
+        "//cmd/enterprise-portal/internal/database",
         "//cmd/enterprise-portal/internal/dotcomdb",
         "//cmd/enterprise-portal/internal/samsm2m",
         "//internal/trace",

cmd/enterprise-portal/internal/subscriptionsservice/v1.go

@@ -21,15 +21,11 @@ import (
 
 const Name = subscriptionsv1connect.SubscriptionsServiceName
 
-type DotComDB interface {
-	ListEnterpriseSubscriptionLicenses(context.Context, []*subscriptionsv1.ListEnterpriseSubscriptionLicensesFilter, int) ([]*dotcomdb.LicenseAttributes, error)
-}
-
 func RegisterV1(
 	logger log.Logger,
 	mux *http.ServeMux,
 	samsClient samsm2m.TokenIntrospector,
-	dotcom DotComDB,
+	store StoreV1,
 	opts ...connect.HandlerOption,
 ) {
 	mux.Handle(
@@ -37,7 +33,7 @@ func RegisterV1(
 			&handlerV1{
 				logger:     logger.Scoped("subscriptions.v1"),
 				samsClient: samsClient,
-				dotcom:     dotcom,
+				store:      store,
 			},
 			opts...,
 		),
@@ -49,7 +45,7 @@ type handlerV1 struct {
 
 	logger     log.Logger
 	samsClient samsm2m.TokenIntrospector
-	dotcom     DotComDB
+	store      StoreV1
 }
 
 var _ subscriptionsv1connect.SubscriptionsServiceHandler = (*handlerV1)(nil)
@@ -57,7 +53,7 @@ var _ subscriptionsv1connect.SubscriptionsServiceHandler = (*handlerV1)(nil)
 func (s *handlerV1) ListEnterpriseSubscriptionLicenses(ctx context.Context, req *connect.Request[subscriptionsv1.ListEnterpriseSubscriptionLicensesRequest]) (*connect.Response[subscriptionsv1.ListEnterpriseSubscriptionLicensesResponse], error) {
 	logger := trace.Logger(ctx, s.logger)
 
-	// 🚨 SECURITY: Require approrpiate M2M scope.
+	// 🚨 SECURITY: Require appropriate M2M scope.
 	requiredScope := samsm2m.EnterprisePortalScope("subscription", scopes.ActionRead)
 	clientAttrs, err := samsm2m.RequireScope(ctx, logger, s.samsClient, requiredScope, req)
 	if err != nil {
@@ -97,7 +93,7 @@ func (s *handlerV1) ListEnterpriseSubscriptionLicenses(ctx context.Context, req
 		}
 	}
 
-	licenses, err := s.dotcom.ListEnterpriseSubscriptionLicenses(ctx, filters,
+	licenses, err := s.store.ListEnterpriseSubscriptionLicenses(ctx, filters,
 		// Provide page size to allow "active license" functionality, by only
 		// retrieving the most recently created result.
 		int(req.Msg.GetPageSize()))

cmd/enterprise-portal/internal/subscriptionsservice/v1_store.go

@@ -0,0 +1,38 @@
+package subscriptionsservice
+
+import (
+	"context"
+
+	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/database"
+	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/dotcomdb"
+	subscriptionsv1 "github.com/sourcegraph/sourcegraph/lib/enterpriseportal/subscriptions/v1"
+)
+
+// StoreV1 is the data layer carrier for subscriptions service v1. This interface
+// is meant to abstract away and limit the exposure of the underlying data layer
+// to the handler through a thin-wrapper.
+type StoreV1 interface {
+	ListEnterpriseSubscriptionLicenses(context.Context, []*subscriptionsv1.ListEnterpriseSubscriptionLicensesFilter, int) ([]*dotcomdb.LicenseAttributes, error)
+}
+
+type storeV1 struct {
+	db       *database.DB
+	dotcomDB *dotcomdb.Reader
+}
+
+type NewStoreV1Options struct {
+	DB       *database.DB
+	DotcomDB *dotcomdb.Reader
+}
+
+// NewStoreV1 returns a new StoreV1 using the given client and database handles.
+func NewStoreV1(opts NewStoreV1Options) StoreV1 {
+	return &storeV1{
+		db:       opts.DB,
+		dotcomDB: opts.DotcomDB,
+	}
+}
+
+func (s *storeV1) ListEnterpriseSubscriptionLicenses(ctx context.Context, filters []*subscriptionsv1.ListEnterpriseSubscriptionLicensesFilter, limit int) ([]*dotcomdb.LicenseAttributes, error) {
+	return s.dotcomDB.ListEnterpriseSubscriptionLicenses(ctx, filters, limit)
+}

cmd/enterprise-portal/service/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "config.go",
         "dotcomdb.go",
+        "redis.go",
         "service.go",
         "state.go",
     ],
@@ -13,6 +14,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//cmd/enterprise-portal/internal/codyaccessservice",
+        "//cmd/enterprise-portal/internal/database",
         "//cmd/enterprise-portal/internal/dotcomdb",
         "//cmd/enterprise-portal/internal/subscriptionsservice",
         "//internal/debugserver",
@@ -27,10 +29,13 @@ go_library(
         "@com_connectrpc_grpcreflect//:grpcreflect",
         "@com_connectrpc_otelconnect//:otelconnect",
         "@com_github_jackc_pgx_v5//pgxpool",
+        "@com_github_redis_go_redis_extra_redisotel_v9//:redisotel",
+        "@com_github_redis_go_redis_v9//:go-redis",
         "@com_github_sourcegraph_log//:log",
         "@com_github_sourcegraph_sourcegraph_accounts_sdk_go//:sourcegraph-accounts-sdk-go",
         "@com_github_sourcegraph_sourcegraph_accounts_sdk_go//scopes",
         "@io_opentelemetry_go_contrib_instrumentation_net_http_otelhttp//:otelhttp",
+        "@io_opentelemetry_go_otel//:otel",
         "@org_golang_x_net//http2",
         "@org_golang_x_net//http2/h2c",
     ],

cmd/enterprise-portal/service/redis.go

@@ -0,0 +1,63 @@
+package service
+
+import (
+	"context"
+	"net"
+	"os"
+
+	"github.com/redis/go-redis/extra/redisotel/v9"
+	"github.com/redis/go-redis/v9"
+	"go.opentelemetry.io/otel"
+
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+)
+
+// newRedisClient creates a new Redis client with tracing enabled.
+func newRedisClient(msp bool, endpoint *string) (*redis.Client, error) {
+	var redisOpts *redis.Options
+	if msp && endpoint != nil {
+		var err error
+		redisOpts, err = redis.ParseURL(*endpoint)
+		if err != nil {
+			return nil, errors.Wrap(err, "invalid endpoint")
+		}
+	} else {
+		// Local dev fallback
+		redisOpts = &redis.Options{Addr: os.ExpandEnv("$REDIS_HOST:$REDIS_PORT")}
+	}
+	redisClient := redis.NewClient(redisOpts)
+	redisClient.AddHook(&redisTracingWrappingHook{})
+	if err := redisotel.InstrumentTracing(redisClient); err != nil {
+		return nil, errors.Wrap(err, "instrument tracing")
+	}
+	return redisClient, nil
+}
+
+var redisTracer = otel.Tracer("enterprise-portal/service/redis")
+
+// redisTracingWrappingHook creates a parent trace span called "redis" to make
+// Redis spans look nicer in the UI, it should be added before the real Redis
+// OTEL tracing hook.
+type redisTracingWrappingHook struct{}
+
+func (h *redisTracingWrappingHook) DialHook(next redis.DialHook) redis.DialHook {
+	return func(ctx context.Context, network, addr string) (net.Conn, error) {
+		// The DialHook already has "redis." prefix, thus we do nothing.
+		return next(ctx, network, addr)
+	}
+}
+
+func (h *redisTracingWrappingHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		ctx, span := redisTracer.Start(ctx, "redis")
+		defer span.End()
+		return next(ctx, cmd)
+	}
+}
+
+func (h *redisTracingWrappingHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return func(ctx context.Context, cmds []redis.Cmder) error {
+		// The ProcessPipelineHook already has "redis." prefix, thus we do nothing.
+		return next(ctx, cmds)
+	}
+}

cmd/enterprise-portal/service/service.go

@@ -14,12 +14,12 @@ import (
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
 
-	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/codyaccessservice"
-	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/subscriptionsservice"
-
 	sams "github.com/sourcegraph/sourcegraph-accounts-sdk-go"
 	"github.com/sourcegraph/sourcegraph-accounts-sdk-go/scopes"
 
+	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/codyaccessservice"
+	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/database"
+	"github.com/sourcegraph/sourcegraph/cmd/enterprise-portal/internal/subscriptionsservice"
 	"github.com/sourcegraph/sourcegraph/internal/debugserver"
 	"github.com/sourcegraph/sourcegraph/internal/httpserver"
 	"github.com/sourcegraph/sourcegraph/internal/trace/policy"
@@ -41,9 +41,19 @@ func (Service) Initialize(ctx context.Context, logger log.Logger, contract runti
 	// We use Sourcegraph tracing code, so explicitly configure a trace policy
 	policy.SetTracePolicy(policy.TraceAll)
 
+	redisClient, err := newRedisClient(contract.MSP, contract.RedisEndpoint)
+	if err != nil {
+		return nil, errors.Wrap(err, "initialize Redis client")
+	}
+
+	dbHandle, err := database.NewHandle(ctx, logger, contract, redisClient, version.Version())
+	if err != nil {
+		return nil, errors.Wrap(err, "initialize database handle")
+	}
+
 	dotcomDB, err := newDotComDBConn(ctx, config)
 	if err != nil {
-		return nil, errors.Wrap(err, "newDotComDBConn")
+		return nil, errors.Wrap(err, "initialize dotcom database handle")
 	}
 
 	// Prepare SAMS client, so that we can enforce SAMS-based M2M authz/authn
@@ -85,8 +95,18 @@ func (Service) Initialize(ctx context.Context, logger log.Logger, contract runti
 	// Register connect endpoints
 	codyaccessservice.RegisterV1(logger, httpServer, samsClient.Tokens(), dotcomDB,
 		connect.WithInterceptors(otelConnctInterceptor))
-	subscriptionsservice.RegisterV1(logger, httpServer, samsClient.Tokens(), dotcomDB,
-		connect.WithInterceptors(otelConnctInterceptor))
+	subscriptionsservice.RegisterV1(
+		logger,
+		httpServer,
+		samsClient.Tokens(),
+		subscriptionsservice.NewStoreV1(
+			subscriptionsservice.NewStoreV1Options{
+				DB:       dbHandle,
+				DotcomDB: dotcomDB,
+			},
+		),
+		connect.WithInterceptors(otelConnctInterceptor),
+	)
 
 	// Optionally enable reflection handlers and a debug UI
 	listenAddr := fmt.Sprintf(":%d", contract.Port)
@@ -134,6 +154,8 @@ func (Service) Initialize(ctx context.Context, logger log.Logger, contract runti
 		background.CallbackRoutine{
 			StopFunc: func(ctx context.Context) error {
 				start := time.Now()
+				// NOTE: If we simply shut down, some in-fly requests may be dropped as the
+				// service exits, so we attempt to gracefully shutdown first.
 				dotcomDB.Close()
 				logger.Info("database connection pool closed", log.Duration("elapsed", time.Since(start)))
 				return nil

deps.bzl

@@ -3586,6 +3586,20 @@ def go_dependencies():
         sum = "h1:w3ciUoD19shMCRargcpm0cm91ytaBhDvuRpz1ODO/U8=",
         version = "v0.4.0",
     )
+    go_repository(
+        name = "com_github_jinzhu_inflection",
+        build_file_proto_mode = "disable_global",
+        importpath = "github.com/jinzhu/inflection",
+        sum = "h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=",
+        version = "v1.0.0",
+    )
+    go_repository(
+        name = "com_github_jinzhu_now",
+        build_file_proto_mode = "disable_global",
+        importpath = "github.com/jinzhu/now",
+        sum = "h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=",
+        version = "v1.1.5",
+    )
     go_repository(
         name = "com_github_jmespath_go_jmespath",
         build_file_proto_mode = "disable_global",
@@ -5099,6 +5113,20 @@ def go_dependencies():
         sum = "h1:ta7tUOvsPHVHGom5hKW5VXNc2xZIkfCKP8iaqOyYtUQ=",
         version = "v0.0.0-20150907023854-cb7f23ec59be",
     )
+    go_repository(
+        name = "com_github_redis_go_redis_extra_rediscmd_v9",
+        build_file_proto_mode = "disable_global",
+        importpath = "github.com/redis/go-redis/extra/rediscmd/v9",
+        sum = "h1:EaDatTxkdHG+U3Bk4EUr+DZ7fOGwTfezUiUJMaIcaho=",
+        version = "v9.0.5",
+    )
+    go_repository(
+        name = "com_github_redis_go_redis_extra_redisotel_v9",
+        build_file_proto_mode = "disable_global",
+        importpath = "github.com/redis/go-redis/extra/redisotel/v9",
+        sum = "h1:EfpWLLCyXw8PSM2/XNJLjI3Pb27yVE+gIAfeqp8LUCc=",
+        version = "v9.0.5",
+    )
     go_repository(
         name = "com_github_redis_go_redis_v9",
         build_file_proto_mode = "disable_global",
@@ -7432,6 +7460,34 @@ def go_dependencies():
         sum = "h1:4NOGyOwD5sUZ22PiWYKmfxqoeh72z6EhYjNosKGLmZg=",
         version = "v3.5.10",
     )
+    go_repository(
+        name = "io_gorm_driver_postgres",
+        build_file_proto_mode = "disable_global",
+        importpath = "gorm.io/driver/postgres",
+        sum = "h1:8ptbNJTDbEmhdr62uReG5BGkdQyeasu/FZHxI0IMGnM=",
+        version = "v1.5.7",
+    )
+    go_repository(
+        name = "io_gorm_driver_sqlite",
+        build_file_proto_mode = "disable_global",
+        importpath = "gorm.io/driver/sqlite",
+        sum = "h1:zKYbzRCpBrT1bNijRnxLDJWPjVfImGEn0lSnUY5gZ+c=",
+        version = "v1.5.0",
+    )
+    go_repository(
+        name = "io_gorm_gorm",
+        build_file_proto_mode = "disable_global",
+        importpath = "gorm.io/gorm",
+        sum = "h1:9DShaph9qhkIYw7QF91I/ynrr4cOO2PZra2PFD7Mfeg=",
+        version = "v1.25.7-0.20240204074919-46816ad31dde",
+    )
+    go_repository(
+        name = "io_gorm_plugin_opentelemetry",
+        build_file_proto_mode = "disable_global",
+        importpath = "gorm.io/plugin/opentelemetry",
+        sum = "h1:7p0ocWELjSSRI7NCKPW2mVe6h43YPini99sNJcbsTuc=",
+        version = "v0.1.4",
+    )
     go_repository(
         name = "io_k8s_api",
         build_file_proto_mode = "disable_global",
@@ -7807,6 +7863,20 @@ def go_dependencies():
         sum = "h1:jq9TW8u3so/bN+JPT166wjOI6/vQPF6Xe7nMNIltagk=",
         version = "v0.49.0",
     )
+    go_repository(
+        name = "io_opentelemetry_go_contrib_instrumentation_runtime",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/contrib/instrumentation/runtime",
+        sum = "h1:EbmAUG9hEAMXyfWEasIt2kmh/WmXUznUksChApTgBGc=",
+        version = "v0.42.0",
+    )
+    go_repository(
+        name = "io_opentelemetry_go_contrib_propagators_b3",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/contrib/propagators/b3",
+        sum = "h1:ImOVvHnku8jijXqkwCSyYKRDt2YrnGXD4BbhcpfbfJo=",
+        version = "v1.17.0",
+    )
     go_repository(
         name = "io_opentelemetry_go_contrib_propagators_jaeger",
         build_file_proto_mode = "disable_global",
@@ -7814,6 +7884,13 @@ def go_dependencies():
         sum = "h1:CKtIfwSgDvJmaWsZROcHzONZgmQdMYn9mVYWypOWT5o=",
         version = "v1.24.0",
     )
+    go_repository(
+        name = "io_opentelemetry_go_contrib_propagators_opencensus",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/contrib/propagators/opencensus",
+        sum = "h1:wQBFvTXNs/UcCiWEi0cadOGJ8LMxDSOWbNtldFS0VI4=",
+        version = "v0.42.0",
+    )
     go_repository(
         name = "io_opentelemetry_go_contrib_propagators_ot",
         build_file_proto_mode = "disable_global",
@@ -7828,6 +7905,13 @@ def go_dependencies():
         sum = "h1:0LAOdjNmQeSTzGBzduGe/rU4tZhMwL5rWgtp9Ku5Jfo=",
         version = "v1.24.0",
     )
+    go_repository(
+        name = "io_opentelemetry_go_otel_bridge_opencensus",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/otel/bridge/opencensus",
+        sum = "h1:YHivttTaDhbZIHuPlg1sWsy2P5gj57vzqPfkHItgbwQ=",
+        version = "v0.39.0",
+    )
     go_repository(
         name = "io_opentelemetry_go_otel_bridge_opentracing",
         build_file_proto_mode = "disable_global",
@@ -7846,8 +7930,22 @@ def go_dependencies():
         name = "io_opentelemetry_go_otel_exporters_otlp_internal_retry",
         build_file_proto_mode = "disable_global",
         importpath = "go.opentelemetry.io/otel/exporters/otlp/internal/retry",
-        sum = "h1:/fXHZHGvro6MVqV34fJzDhi7sHGpX3Ej/Qjmfn003ho=",
-        version = "v1.14.0",
+        sum = "h1:t4ZwRPU+emrcvM2e9DHd0Fsf0JTPVcbfa/BhTDF03d0=",
+        version = "v1.16.0",
+    )
+    go_repository(
+        name = "io_opentelemetry_go_otel_exporters_otlp_otlpmetric",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/otel/exporters/otlp/otlpmetric",
+        sum = "h1:f6BwB2OACc3FCbYVznctQ9V6KK7Vq6CjmYXJ7DeSs4E=",
+        version = "v0.39.0",
+    )
+    go_repository(
+        name = "io_opentelemetry_go_otel_exporters_otlp_otlpmetric_otlpmetricgrpc",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc",
+        sum = "h1:rm+Fizi7lTM2UefJ1TO347fSRcwmIsUAaZmYmIGBRAo=",
+        version = "v0.39.0",
     )
 
     # See https://github.com/open-telemetry/opentelemetry-go-contrib/issues/872
@@ -7895,6 +7993,13 @@ def go_dependencies():
         sum = "h1:I8WIFXR351FoLJYuloU4EgXbtNX2URfU/85pUPheIEQ=",
         version = "v0.46.0",
     )
+    go_repository(
+        name = "io_opentelemetry_go_otel_exporters_stdout_stdouttrace",
+        build_file_proto_mode = "disable_global",
+        importpath = "go.opentelemetry.io/otel/exporters/stdout/stdouttrace",
+        sum = "h1:2PunuO5SbkN5MhCbuHCd3tC6qrcaj+uDAkX/qBU5BAs=",
+        version = "v1.15.1",
+    )
     go_repository(
         name = "io_opentelemetry_go_otel_metric",
         build_file_proto_mode = "disable_global",

go.mod

@@ -286,6 +286,8 @@ require (
 	github.com/pkoukk/tiktoken-go v0.1.6
 	github.com/pkoukk/tiktoken-go-loader v0.0.1
 	github.com/prometheus/statsd_exporter v0.22.7
+	github.com/redis/go-redis/extra/redisotel/v9 v9.0.5
+	github.com/redis/go-redis/v9 v9.0.5
 	github.com/robert-nix/ansihtml v1.0.1
 	github.com/sourcegraph/cloud-api v0.0.0-20240501113836-ecd1d4cba9dd
 	github.com/sourcegraph/log/logr v0.0.0-20240425170707-431bcb6c8668
@@ -312,6 +314,9 @@ require (
 	go.opentelemetry.io/collector/config/configtls v0.92.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.46.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20240311132316-a219d84964c2
+	gorm.io/driver/postgres v1.5.7
+	gorm.io/gorm v1.25.7-0.20240204074919-46816ad31dde
+	gorm.io/plugin/opentelemetry v0.1.4
 	oss.terrastruct.com/d2 v0.6.5
 	sigs.k8s.io/controller-runtime v0.17.3
 )
@@ -386,6 +391,8 @@ require (
 	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/knadh/koanf/v2 v2.0.1 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/matryer/is v1.2.0 // indirect
@@ -400,6 +407,7 @@ require (
 	github.com/pjbgf/sha1cd v0.3.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect
 	github.com/prometheus/prometheus v0.40.5 // indirect
+	github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5 // indirect
 	github.com/rickb777/date v1.14.3 // indirect
 	github.com/rickb777/plural v1.2.2 // indirect
 	github.com/shirou/gopsutil/v3 v3.23.12 // indirect

go.sum

@@ -346,8 +346,10 @@ github.com/bradleyjkemp/cupaloy/v2 v2.6.0 h1:knToPYa2xtfg42U3I6punFEjaGFKWQRXJwj
 github.com/bradleyjkemp/cupaloy/v2 v2.6.0/go.mod h1:bm7JXdkRd4BHJk9HpwqAI8BoAY1lps46Enkdqw6aRX0=
 github.com/bshuster-repo/logrus-logstash-hook v1.0.0/go.mod h1:zsTqEiSzDgAa/8GZR7E1qaXrhYNDKBYy5/dWPTIflbk=
 github.com/bsm/ginkgo/v2 v2.5.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
+github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
 github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
 github.com/bsm/gomega v1.20.0/go.mod h1:JifAceMQ4crZIWYUKrlGcmbN3bqHogVTADMD2ATsbwk=
+github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
 github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bufbuild/buf v1.25.0 h1:HFxKrR8wFcZwrBInN50K/oJX/WOtPVq24rHb/ArjfBA=
 github.com/bufbuild/buf v1.25.0/go.mod h1:GCKZ5bAP6Ht4MF7KcfaGVgBEXGumwAz2hXjjLVxx8ZU=
@@ -1220,6 +1222,10 @@ github.com/jhump/protoreflect v1.11.0/go.mod h1:U7aMIjN0NWq9swDP7xDdoMfRHb35uiuT
 github.com/jhump/protoreflect v1.12.0/go.mod h1:JytZfP5d0r8pVNLZvai7U/MCuTWITgrI4tTg7puQFKI=
 github.com/jhump/protoreflect v1.15.1 h1:HUMERORf3I3ZdX05WaQ6MIpd/NJ434hTp5YiKgfCL6c=
 github.com/jhump/protoreflect v1.15.1/go.mod h1:jD/2GMKKE6OqX8qTjhADU1e6DShO+gavG9e0Q693nKo=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
@@ -1600,6 +1606,10 @@ github.com/qustavo/sqlhooks/v2 v2.1.0/go.mod h1:aMREyKo7fOKTwiLuWPsaHRXEmtqG4yRE
 github.com/rafaeljusto/redigomock/v3 v3.1.2 h1:B4Y0XJQiPjpwYmkH55aratKX1VfR+JRqzmDKyZbC99o=
 github.com/rafaeljusto/redigomock/v3 v3.1.2/go.mod h1:F9zPqz8rMriScZkPtUiLJoLruYcpGo/XXREpeyasREM=
 github.com/rainycape/unidecode v0.0.0-20150907023854-cb7f23ec59be/go.mod h1:MIDFMn7db1kT65GmV94GzpX9Qdi7N/pQlwb+AN8wh+Q=
+github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5 h1:EaDatTxkdHG+U3Bk4EUr+DZ7fOGwTfezUiUJMaIcaho=
+github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5/go.mod h1:fyalQWdtzDBECAQFBJuQe5bzQ02jGd5Qcbgb97Flm7U=
+github.com/redis/go-redis/extra/redisotel/v9 v9.0.5 h1:EfpWLLCyXw8PSM2/XNJLjI3Pb27yVE+gIAfeqp8LUCc=
+github.com/redis/go-redis/extra/redisotel/v9 v9.0.5/go.mod h1:WZjPDy7VNzn77AAfnAfVjZNvfJTYfPetfZk5yoSTLaQ=
 github.com/redis/go-redis/v9 v9.0.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps=
 github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o=
 github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
@@ -2663,6 +2673,14 @@ gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gorm.io/driver/postgres v1.5.7 h1:8ptbNJTDbEmhdr62uReG5BGkdQyeasu/FZHxI0IMGnM=
+gorm.io/driver/postgres v1.5.7/go.mod h1:3e019WlBaYI5o5LIdNV+LyxCMNtLOQETBXL2h4chKpA=
+gorm.io/driver/sqlite v1.5.0 h1:zKYbzRCpBrT1bNijRnxLDJWPjVfImGEn0lSnUY5gZ+c=
+gorm.io/driver/sqlite v1.5.0/go.mod h1:kDMDfntV9u/vuMmz8APHtHF0b4nyBB7sfCieC6G8k8I=
+gorm.io/gorm v1.25.7-0.20240204074919-46816ad31dde h1:9DShaph9qhkIYw7QF91I/ynrr4cOO2PZra2PFD7Mfeg=
+gorm.io/gorm v1.25.7-0.20240204074919-46816ad31dde/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8=
+gorm.io/plugin/opentelemetry v0.1.4 h1:7p0ocWELjSSRI7NCKPW2mVe6h43YPini99sNJcbsTuc=
+gorm.io/plugin/opentelemetry v0.1.4/go.mod h1:tndJHOdvPT0pyGhOb8E2209eXJCUxhC5UpKw7bGVWeI=
 gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=

internal/rcache/rcache.go

@@ -258,6 +258,8 @@ type TB interface {
 	Helper()
 }
 
+const TestAddr = "127.0.0.1:6379"
+
 // SetupForTest adjusts the globalPrefix and clears it out. You will have
 // conflicts if you do `t.Parallel()`
 func SetupForTest(t testing.TB) {
@@ -267,7 +269,7 @@ func SetupForTest(t testing.TB) {
 		MaxIdle:     3,
 		IdleTimeout: 240 * time.Second,
 		Dial: func() (redis.Conn, error) {
-			return redis.Dial("tcp", "127.0.0.1:6379")
+			return redis.Dial("tcp", TestAddr)
 		},
 		TestOnBorrow: func(c redis.Conn, t time.Time) error {
 			_, err := c.Do("PING")

internal/redislock/BUILD.bazel

@@ -3,23 +3,41 @@ load("//dev:go_defs.bzl", "go_test")
 
 go_library(
     name = "redislock",
-    srcs = ["redislock.go"],
+    srcs = [
+        "mutex.go",
+        "redislock.go",
+    ],
     importpath = "github.com/sourcegraph/sourcegraph/internal/redislock",
     visibility = ["//:__subpackages__"],
     deps = [
         "//internal/redispool",
+        "//lib/errors",
+        "@com_github_go_redsync_redsync_v4//:redsync",
+        "@com_github_go_redsync_redsync_v4//redis/goredis/v9:goredis",
         "@com_github_gomodule_redigo//redis",
         "@com_github_google_uuid//:uuid",
+        "@com_github_redis_go_redis_v9//:go-redis",
+        "@com_github_sourcegraph_log//:log",
     ],
 )
 
 go_test(
     name = "redislock_test",
-    srcs = ["redislock_test.go"],
+    srcs = [
+        "mutex_test.go",
+        "redislock_test.go",
+    ],
     embed = [":redislock"],
+    tags = [
+        # Test requires localhost database
+        "requires-network",
+    ],
     deps = [
+        "//internal/rcache",
         "//internal/redispool",
         "@com_github_derision_test_go_mockgen_v2//testutil/require",
+        "@com_github_redis_go_redis_v9//:go-redis",
+        "@com_github_sourcegraph_log//logtest",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
     ],

internal/redislock/mutex.go

@@ -0,0 +1,34 @@
+package redislock
+
+import (
+	"time"
+
+	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
+	"github.com/redis/go-redis/v9"
+	"github.com/sourcegraph/log"
+
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+)
+
+// OnlyOne runs the given function only if the lock can be acquired. If the lock
+// is already held, it will wait until the lock is released or returns an error
+// when the expiry time is reached.
+func OnlyOne(logger log.Logger, client *redis.Client, name string, expiry time.Duration, fn func() error) (err error) {
+	logger = logger.With(log.String("name", name))
+	logger.Debug("acquiring lock")
+	mutex := redsync.New(goredis.NewPool(client)).NewMutex(name, redsync.WithExpiry(expiry))
+	if err = mutex.Lock(); err != nil {
+		return errors.Wrap(err, "acquire lock")
+	}
+
+	defer func() {
+		logger.Debug("releasing lock")
+		_, unlockErr := mutex.Unlock()
+		// Only overwrite the error if there wasn't already an error.
+		if err == nil && unlockErr != nil {
+			err = errors.Wrap(unlockErr, "release lock")
+		}
+	}()
+	return fn()
+}

internal/redislock/mutex_test.go

@@ -0,0 +1,32 @@
+package redislock
+
+import (
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/sourcegraph/log/logtest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/sourcegraph/sourcegraph/internal/rcache"
+)
+
+func TestOnlyOne(t *testing.T) {
+	rcache.SetupForTest(t)
+
+	logger := logtest.NoOp(t)
+	client := redis.NewClient(&redis.Options{Addr: rcache.TestAddr})
+
+	// If the algorithm is correct, there should be no data race detected in tests,
+	// and the final count should be 10.
+	count := 0
+	for i := 0; i < 10; i++ {
+		err := OnlyOne(logger, client, "test", 3*time.Second, func() error {
+			count++
+			return nil
+		})
+		require.NoError(t, err)
+	}
+	assert.Equal(t, 10, count)
+}

lib/managedservicesplatform/runtime/contract/postgresql.go

@@ -49,7 +49,7 @@ func (c postgreSQLContract) OpenDatabase(ctx context.Context, database string) (
 		if err != nil {
 			return nil, err
 		}
-		return sql.Open("customdsn", stdlib.RegisterConnConfig(config.ConnConfig))
+		return sql.Open("pgx", stdlib.RegisterConnConfig(config.ConnConfig))
 	}
 	return cloudsql.Open(ctx, c.getCloudSQLConnConfig(database))
 }

sg.config.yaml

@@ -393,6 +393,7 @@ commands:
 
   enterprise-portal:
     cmd: |
+      export PGDSN="postgres://$PGUSER:$PGPASSWORD@$PGHOST:$PGPORT/{{ .Database }}?sslmode=$PGSSLMODE"
       # Connect to local development database, with the assumption that it will
       # have dotcom database tables.
       export DOTCOM_PGDSN_OVERRIDE="postgres://$PGUSER:$PGPASSWORD@$PGHOST:$PGPORT/$PGDATABASE?sslmode=$PGSSLMODE"
@@ -412,6 +413,8 @@ commands:
       DOTCOM_INCLUDE_PRODUCTION_LICENSES: 'true'
       # Used for authentication
       SAMS_URL: https://accounts.sgdev.org
+      REDIS_HOST: localhost
+      REDIS_PORT: 6379
     externalSecrets:
       ENTERPRISE_PORTAL_SAMS_CLIENT_ID:
         project: sourcegraph-local-dev