Share PostgreSQL with Substrate

contrib/cncf/technical-review.md

@@ -325,7 +325,7 @@ Default values can be found in [helm/kagent/values.yaml](https://github.com/kage
 **Additional Configurations:**
 For production use, configure:
 
-- External PostgreSQL connection (set `database.postgres.bundled.enabled=false` and set either `database.postgres.url` or `database.postgres.urlFile`)
+- External PostgreSQL connection (set `database.postgres.bundled.enabled=false` and configure `database.postgres.url` or `database.postgres.secretRef`)
 - LLM API keys via Secrets (`providers.openAI.apiKeySecretRef`)
 - TLS for external LLM connections (`modelConfig.tls`)
 - Resource limits based on workload (`agents.*.resources`)

go/core/cli/internal/commands/db/db.go

@@ -48,21 +48,21 @@ func NewDBCmd() *cobra.Command {
 // precedence, on: the DATABASE_VECTOR_ENABLED env var in the CLI's own
 // environment (explicit operator intent, works without a cluster), the
 // controller's configmap on the live cluster (the same value the server
-// reads), and finally the controller's default (enabled).
+// reads), and finally the controller's default (disabled).
 func migrationSources(namespace *string) dbmigrate.SourcesFunc {
 	return func(ctx context.Context) ([]migrations.Source, error) {
-		vectorEnabled := true
+		vectorEnabled := false
 		if v := os.Getenv(vectorEnabledKey); v != "" {
 			b, err := strconv.ParseBool(v)
 			if err != nil {
-				fmt.Fprintf(os.Stderr, "warning: invalid %s=%q; assuming true\n", vectorEnabledKey, v)
+				fmt.Fprintf(os.Stderr, "warning: invalid %s=%q; assuming false\n", vectorEnabledKey, v)
 			} else {
 				vectorEnabled = b
 			}
 		} else if b, ok := clusterVectorEnabled(ctx, *namespace); ok {
 			vectorEnabled = b
 		}
-		return migrations.BuiltinSources(vectorEnabled), nil
+		return migrations.BuiltinSourcesInSchema(vectorEnabled, kagentenv.DatabaseSchema.Get(), kagentenv.DatabaseVectorSchema.Get()), nil
 	}
 }
 

go/core/cli/internal/db/migrate/migrate.go

@@ -22,6 +22,7 @@ import (
 
 const (
 	dbURLEnv   = "POSTGRES_DATABASE_URL"
+	dbRoleEnv  = "POSTGRES_DATABASE_ROLE"
 	sourceFlag = "source"
 )
 
@@ -35,6 +36,7 @@ type SourcesFunc func(ctx context.Context) ([]migrations.Source, error)
 
 type commandState struct {
 	dbURL     string
+	dbRole    string
 	source    string
 	resolveFn SourcesFunc
 
@@ -93,9 +95,10 @@ func NewCommandFromFunc(fn SourcesFunc) *cobra.Command {
 		Use:   "migrate",
 		Short: "Apply, roll back, and inspect database migrations",
 		Long: `Apply, roll back, and inspect database migrations.
-The command reads POSTGRES_DATABASE_URL when --db-url is empty.`,
+The command reads POSTGRES_DATABASE_URL and POSTGRES_DATABASE_ROLE when their flags are empty.`,
 	}
 	command.PersistentFlags().StringVar(&state.dbURL, "db-url", "", "PostgreSQL connection URL")
+	command.PersistentFlags().StringVar(&state.dbRole, "db-role", "", "Stable PostgreSQL role to assume after authentication")
 	command.PersistentFlags().StringVar(&state.source, sourceFlag, "", "Migration source for down, goto, or version")
 	command.AddCommand(newUpCmd(state))
 	command.AddCommand(newDownCmd(state))
@@ -105,6 +108,13 @@ The command reads POSTGRES_DATABASE_URL when --db-url is empty.`,
 	return command
 }
 
+func (s *commandState) role() string {
+	if role := strings.TrimSpace(s.dbRole); role != "" {
+		return role
+	}
+	return strings.TrimSpace(os.Getenv(dbRoleEnv))
+}
+
 func (s *commandState) resolveDSN() (string, error) {
 	dsn := strings.TrimSpace(s.dbURL)
 	if dsn == "" {
@@ -204,7 +214,7 @@ func newUpCmd(state *commandState) *cobra.Command {
 			if len(sources) == 0 {
 				return errors.New("no migration sources are registered")
 			}
-			if err := migrations.RunUp(command.Context(), dsn, sources); err != nil {
+			if err := migrations.RunUpAsRole(command.Context(), dsn, state.role(), sources); err != nil {
 				return err
 			}
 			fmt.Fprintln(command.OutOrStdout(), "schema is up to date")
@@ -236,7 +246,7 @@ func newDownCmd(state *commandState) *cobra.Command {
 			if err != nil {
 				return err
 			}
-			return migrations.WithProvider(command.Context(), dsn, source, func(provider *goose.Provider) error {
+			return migrations.WithProviderAsRole(command.Context(), dsn, state.role(), source, func(provider *goose.Provider) error {
 				current, err := readVersion(command.Context(), provider)
 				if err != nil {
 					return err
@@ -316,7 +326,7 @@ func newStatusCmd(state *commandState) *cobra.Command {
 				if err != nil {
 					return err
 				}
-				err = migrations.WithProvider(command.Context(), dsn, source, func(provider *goose.Provider) error {
+				err = migrations.WithProviderAsRole(command.Context(), dsn, state.role(), source, func(provider *goose.Provider) error {
 					status, err := provider.Status(command.Context())
 					if err != nil {
 						return err
@@ -432,7 +442,7 @@ func newVersionCmd(state *commandState) *cobra.Command {
 				sources = sources[index : index+1]
 			}
 			for _, source := range sources {
-				err := migrations.WithProvider(command.Context(), dsn, source, func(provider *goose.Provider) error {
+				err := migrations.WithProviderAsRole(command.Context(), dsn, state.role(), source, func(provider *goose.Provider) error {
 					version, err := readVersion(command.Context(), provider)
 					if err != nil {
 						return err
@@ -486,7 +496,7 @@ func newGotoCmd(state *commandState) *cobra.Command {
 			if target != 0 && !slices.Contains(versions, target) {
 				return fmt.Errorf("version %d is not available. Valid versions are %s", target, formatVersionList(versions))
 			}
-			return migrations.WithProvider(command.Context(), dsn, source, func(provider *goose.Provider) error {
+			return migrations.WithProviderAsRole(command.Context(), dsn, state.role(), source, func(provider *goose.Provider) error {
 				current, err := readVersion(command.Context(), provider)
 				if err != nil {
 					return err

go/core/cli/internal/db/migrate/migrate_test.go

@@ -115,6 +115,16 @@ func TestResolveDSN(t *testing.T) {
 	}
 }
 
+func TestResolveRole(t *testing.T) {
+	t.Setenv(dbRoleEnv, "env_role")
+	if got := (&commandState{}).role(); got != "env_role" {
+		t.Fatalf("role() = %q, want env_role", got)
+	}
+	if got := (&commandState{dbRole: "flag_role"}).role(); got != "flag_role" {
+		t.Fatalf("role() = %q, want flag_role", got)
+	}
+}
+
 func TestResolveSource(t *testing.T) {
 	multi := testSources()
 	single := multi[:1]

go/core/cmd/controller/main.go

@@ -22,9 +22,12 @@ import (
 	"log/slog"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
+	"github.com/kagent-dev/kagent/go/core/internal/database"
 	"github.com/kagent-dev/kagent/go/core/pkg/app"
+	kagentenv "github.com/kagent-dev/kagent/go/core/pkg/env"
 )
 
 func main() {
@@ -35,6 +38,17 @@ func main() {
 	logger := slog.Default()
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
+	switch os.Getenv("KAGENT_DATABASE_BOOTSTRAP") {
+	case "", "false":
+	case "true":
+		if err := runDatabaseBootstrap(ctx); err != nil {
+			logger.ErrorContext(ctx, "database bootstrap failed", "error", err)
+			os.Exit(1)
+		}
+	default:
+		logger.ErrorContext(ctx, "invalid database bootstrap value")
+		os.Exit(1)
+	}
 
 	// No options: core's own controller runs with the default authenticator and
 	// authorizer. A library consumer supplies its own by calling app.Run directly.
@@ -43,3 +57,37 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+func runDatabaseBootstrap(ctx context.Context) error {
+	adminUsername, err := readRequiredFile("POSTGRES_ADMIN_USERNAME_FILE")
+	if err != nil {
+		return err
+	}
+	adminPassword, err := readRequiredFile("POSTGRES_ADMIN_PASSWORD_FILE")
+	if err != nil {
+		return err
+	}
+	return database.Bootstrap(ctx, database.BootstrapConfig{
+		EndpointSource: os.Getenv("POSTGRES_DATABASE_URL"),
+		AdminUsername:  adminUsername,
+		AdminPassword:  adminPassword,
+		Schema:         kagentenv.DatabaseSchema.Get(),
+		VectorEnabled:  kagentenv.DatabaseVectorEnabled.Get(),
+		VectorSchema:   kagentenv.DatabaseVectorSchema.Get(),
+	})
+}
+
+func readRequiredFile(envName string) (string, error) {
+	path := os.Getenv(envName)
+	if path == "" {
+		return "", fmt.Errorf("%s must name a credential file", envName)
+	}
+	value, err := os.ReadFile(path)
+	if err != nil {
+		return "", fmt.Errorf("read %s: %w", envName, err)
+	}
+	if value := strings.TrimSpace(string(value)); value != "" {
+		return value, nil
+	}
+	return "", fmt.Errorf("%s credential file is empty", envName)
+}

go/core/internal/database/bootstrap.go

@@ -0,0 +1,101 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
+)
+
+const (
+	OwnerRoleName = "kagent_owner"
+	UserName      = "kagent_user"
+	UserPassword  = "kagent"
+)
+
+// BootstrapConfig contains the first-install PostgreSQL credentials.
+// EndpointSource supplies the endpoint, database, and TLS configuration.
+type BootstrapConfig struct {
+	EndpointSource string
+	AdminUsername  string
+	AdminPassword  string
+	Schema         string
+	VectorEnabled  bool
+	VectorSchema   string
+}
+
+// Bootstrap creates the fixed Kagent identity and schema.
+// It does not change the password for an existing user.
+func Bootstrap(ctx context.Context, cfg BootstrapConfig) error {
+	if cfg.EndpointSource == "" {
+		return errors.New("PostgreSQL connection string must not be empty")
+	}
+	if cfg.Schema == "" {
+		return errors.New("PostgreSQL schema must not be empty")
+	}
+	for name, value := range map[string]string{
+		"administrator username": cfg.AdminUsername,
+		"administrator password": cfg.AdminPassword,
+	} {
+		if value == "" {
+			return fmt.Errorf("PostgreSQL %s must not be empty", name)
+		}
+	}
+
+	dsn, err := ResolveURL(cfg.EndpointSource)
+	if err != nil {
+		return err
+	}
+	// The application DSN may contain pool-only options that PostgreSQL cannot accept.
+	poolConfig, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		return errors.New("parse PostgreSQL bootstrap connection string: invalid value")
+	}
+	connConfig := poolConfig.ConnConfig
+	if connConfig.User != UserName {
+		return fmt.Errorf("PostgreSQL bootstrap connection string must contain the %q user", UserName)
+	}
+	if connConfig.Password != UserPassword {
+		return errors.New("PostgreSQL bootstrap connection string does not match the fixed development password")
+	}
+	connConfig.User = cfg.AdminUsername
+	connConfig.Password = cfg.AdminPassword
+	conn, err := pgx.ConnectConfig(ctx, connConfig)
+	if err != nil {
+		return fmt.Errorf("connect as PostgreSQL administrator: %w", err)
+	}
+	defer conn.Close(ctx) //nolint:errcheck // The transaction result decides success.
+
+	tx, err := conn.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("start PostgreSQL bootstrap transaction: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // Commit or the returned error decides the outcome.
+
+	for setting, value := range map[string]string{
+		"kagent.bootstrap_username":       UserName,
+		"kagent.bootstrap_password":       UserPassword,
+		"kagent.bootstrap_owner_role":     OwnerRoleName,
+		"kagent.bootstrap_schema":         cfg.Schema,
+		"kagent.bootstrap_vector_enabled": fmt.Sprint(cfg.VectorEnabled),
+		"kagent.bootstrap_vector_schema":  cfg.VectorSchema,
+	} {
+		if _, err := tx.Exec(ctx, `SELECT set_config($1, $2, true)`, setting, value); err != nil {
+			return fmt.Errorf("set PostgreSQL bootstrap parameter %q: %w", setting, err)
+		}
+	}
+	identitySQL, err := migrations.FS.ReadFile("identity/bootstrap.sql")
+	if err != nil {
+		return fmt.Errorf("read PostgreSQL identity SQL: %w", err)
+	}
+	if _, err := tx.Conn().PgConn().ExecParams(ctx, string(identitySQL), nil, nil, nil, nil).Close(); err != nil {
+		return fmt.Errorf("apply PostgreSQL identity SQL: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit PostgreSQL bootstrap: %w", err)
+	}
+	return nil
+}