fix(core): database clients pooling improvements (#1047)

Closes #681 

Per service with db connection:
- one client
- one schema
- one shared pool

Not supported (and documented in db package)
- multiple schemas per client
- multiple services sharing a pool (unless rolled up into one service
like policy)
- multiple databases per platform Postgres instance
- multiple Postgres servers per platform instance

service/pkg/db/db.go

@@ -75,6 +75,20 @@ type Config struct {
 	MigrationsFS     *embed.FS
 }
 
+/*
+A wrapper around a pgxpool.Pool and sql.DB reference.
+
+Each service should have a single instance of the Client to share a connection pool,
+schema (driven by the service namespace), and an embedded file system for migrations.
+
+The 'search_path' is set to the schema on connection to the database.
+
+If the database config 'runMigrations' is set to true, the client will run migrations on startup,
+once per namespace (as there should only be one embedded migrations FS per namespace).
+
+Multiple pools, schemas, or migrations per service are not supported. Multiple databases per
+PostgreSQL instance or multiple PostgreSQL servers per platform instance are not supported.
+*/
 type Client struct {
 	Pgx    PgxIface
 	config Config
@@ -102,6 +116,7 @@ func New(ctx context.Context, config Config, o ...OptsFunc) (*Client, error) {
 		return nil, fmt.Errorf("failed to parse pgx config: %w", err)
 	}
 
+	slog.Info("opening new database pool", slog.String("schema", config.Schema))
 	pool, err := pgxpool.NewWithConfig(ctx, dbConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create pgxpool: %w", err)
@@ -117,6 +132,13 @@ func New(ctx context.Context, config Config, o ...OptsFunc) (*Client, error) {
 		}
 	}
 
+	// Set the Client search_path to the schema
+	q := fmt.Sprintf("SET search_path TO %s", config.Schema)
+	if _, err := c.Pgx.Exec(ctx, q); err != nil {
+		return nil, fmt.Errorf("failed to SET search_path for db Client schema to [%s]: %w", config.Schema, err)
+	}
+
+	slog.Info("successfully set database client search_path", slog.String("schema", config.Schema))
 	return &c, nil
 }
 

service/pkg/db/db_migration.go

@@ -17,13 +17,6 @@ func migrationInit(ctx context.Context, c *Client, migrations *embed.FS) (*goose
 		return nil, 0, nil, fmt.Errorf("migrations are disabled")
 	}
 
-	// Set the schema
-	q := fmt.Sprintf("SET search_path TO %s", c.config.Schema)
-	if tag, err := c.Pgx.Exec(ctx, q); err != nil {
-		slog.Error("migration error", slog.String("query", q), slog.String("error", err.Error()), slog.Any("tag", tag))
-		return nil, 0, nil, fmt.Errorf("failed to SET search_path [%w]", err)
-	}
-
 	// Cast the pgxpool.Pool to a *sql.DB
 	pool, ok := c.Pgx.(*pgxpool.Pool)
 	if !ok || pool == nil {
@@ -42,7 +35,7 @@ func migrationInit(ctx context.Context, c *Client, migrations *embed.FS) (*goose
 	if e != nil {
 		return nil, 0, nil, errors.Join(fmt.Errorf("failed to get current version"), e)
 	}
-	slog.Info("migration db info ", slog.Any("current version", v))
+	slog.Info("migration db info", slog.Any("current version", v))
 
 	// Return the provider, version, and close function
 	return provider, v, func() {