diff --git a/.env.example b/.env.example index 9057c8e2..c6c9223f 100644 --- a/.env.example +++ b/.env.example @@ -290,6 +290,12 @@ DATABASE_MAX_OPEN_CONNS= DATABASE_MAX_IDLE_CONNS= DATABASE_VECTOR_MAX_CONNS= +## Postgres schema handling, only used when TENANT_ID is set (Supabase needs +## DATABASE_EXTENSIONS_SCHEMA=extensions; its pooler may need the other one). +## See backend/docs/config.md -> "Multi-Instance Deployment (TENANT_ID)". +DATABASE_EXTENSIONS_SCHEMA= +DATABASE_SEARCH_PATH_VIA_OPTIONS= + ## Graphiti knowledge graph settings ## Set GRAPHITI_ENABLED=true and GRAPHITI_URL=http://graphiti:8000 to enable embedded Graphiti GRAPHITI_ENABLED=false diff --git a/.vscode/launch.json b/.vscode/launch.json index 0098e147..10253c9d 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -141,9 +141,9 @@ "LLM_SERVER_CONFIG_PATH": "${workspaceFolder}/examples/configs/openrouter.provider.yml", "DATABASE_URL": "postgres://postgres:postgres@localhost:5432/pentagidb?sslmode=disable", // Langfuse (optional) uncomment to enable - "LANGFUSE_BASE_URL": "http://localhost:4000", + // "LANGFUSE_BASE_URL": "http://localhost:4000", // Observability (optional) uncomment to enable - "OTEL_HOST": "localhost:8148", + // "OTEL_HOST": "localhost:8148", }, "args": [ "-flow", "0", diff --git a/README.md b/README.md index b352fb0f..91c3e2a7 100644 --- a/README.md +++ b/README.md @@ -706,7 +706,7 @@ Set it when several PentAGI installations share external resources: one PostgreS | Area | Effect when `TENANT_ID=acme` | | ---- | ---------------------------- | -| PostgreSQL | The instance creates and works inside schema `acme` instead of `public`; extensions stay shared in `public` | +| PostgreSQL | The instance creates and works inside schema `acme` instead of `public`; extensions stay shared in `DATABASE_EXTENSIONS_SCHEMA` (default `public`, `extensions` on Supabase) | | Worker containers | `acme-pentagi-terminal-` instead of `pentagi-terminal-`; volumes and hostnames follow, and both carry a `pentagi.tenant` label | | Knowledge graph | Graphiti/Neo4j group ids become `acme-flow-` | | Auth | Cookie and API token keys are derived from `COOKIE_SIGNING_SALT` **plus** the tenant, and the session cookie is renamed | @@ -3018,6 +3018,22 @@ docker exec pgvector sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -c \ GROUP BY 1, 2, 3 ORDER BY count DESC;"' ``` +##### External PostgreSQL and schema handling + +`DATABASE_URL` may point at any PostgreSQL instance, not only the bundled `pgvector` container. Two extra knobs apply when — and only when — `TENANT_ID` is set, because that is when PentAGI creates its own schema and rewrites the connection's `search_path`: + +| Env var | Default | Purpose | +|---|---|---| +| `DATABASE_EXTENSIONS_SCHEMA` | `public` | Schema holding the shared `vector` and `pg_trgm` extensions that every tenant's `search_path` must reach | +| `DATABASE_SEARCH_PATH_VIA_OPTIONS` | `false` | Send the tenant `search_path` inside the `options` startup parameter instead of as a bare connection parameter | + +**Supabase (cloud or self-hosted)** needs both of them considered, and is the reason they exist: + +- Supabase installs its bundled extensions into an `extensions` schema instead of `public`, so set `DATABASE_EXTENSIONS_SCHEMA=extensions`. Without it, startup aborts with an error naming the schema where `vector` was actually found — no need to move a provider-managed extension with `ALTER EXTENSION`. +- Supabase's pooler (Supavisor) does not reliably forward a bare `search_path` connection parameter. Prefer a **direct** PostgreSQL connection: self-hosted, expose the `db` service port and bypass the `supavisor` service; cloud, use the "Direct connection" string (or the IPv4 add-on on IPv4-only networks). If the pooler cannot be bypassed, use its session mode and try `DATABASE_SEARCH_PATH_VIA_OPTIONS=true` — PentAGI verifies the effective schema on boot and refuses to start if it did not take effect, so a silent cross-tenant data mix-up is not possible. + +Both settings are managed by the installer under *Server Settings*, next to `TENANT_ID` — see [Running Several Instances](#running-several-instances-tenant_id) for that scenario, and [Multi-Instance Deployment](backend/docs/config.md#multi-instance-deployment-tenant_id) for the full matrix, including the PgBouncer recipe (`pool_mode = session`, `ignore_startup_parameters`, per-tenant `connect_query`). + #### Frontend Configuration Edit the configuration for `frontend` in `.vscode/launch.json` file: diff --git a/backend/cmd/etester/main.go b/backend/cmd/etester/main.go index 2cb53cb9..e7ab6553 100644 --- a/backend/cmd/etester/main.go +++ b/backend/cmd/etester/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "database/sql" "flag" "log" "os" @@ -10,12 +11,14 @@ import ( "time" "pentagi/pkg/config" + "pentagi/pkg/database" "pentagi/pkg/providers/embeddings" "pentagi/pkg/terminal" "pentagi/pkg/version" "github.com/jackc/pgx/v5/pgxpool" "github.com/joho/godotenv" + _ "github.com/lib/pq" "github.com/sirupsen/logrus" ) @@ -62,14 +65,32 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - // Initialize database connection pool + // Create this tenant's schema and repoint DATABASE_URL at it before any + // consumer reads the DSN. No-op when TENANT_ID is empty. + if err := database.EnsureTenantSchema(ctx, cfg); err != nil { + log.Fatalf("Tenant schema initialization failed: %v", err) + } + + // Verify search_path on a short-lived database/sql connection before the + // long-lived pgxpool is opened — same guard the main server uses. + verifyDB, err := sql.Open("postgres", cfg.DatabaseURL) + if err != nil { + log.Fatalf("Unable to open database for schema verification: %v", err) + } + verifyDB.SetMaxOpenConns(1) + if err := database.VerifySearchPath(ctx, verifyDB, cfg); err != nil { + verifyDB.Close() + log.Fatalf("Tenant schema verification failed: %v", err) + } + verifyDB.Close() + poolConfig, err := pgxpool.ParseConfig(cfg.DatabaseURL) if err != nil { log.Fatalf("Unable to parse database URL: %v", err) } - poolConfig.MaxConns = 10 - poolConfig.MinConns = 2 + poolConfig.MaxConns = min(int32(cfg.DBVectorMaxConns), 10) + poolConfig.MinConns = min(int32(cfg.DBMaxIdleConns), 2) poolConfig.MaxConnLifetime = time.Hour poolConfig.MaxConnIdleTime = 30 * time.Minute diff --git a/backend/cmd/ftester/main.go b/backend/cmd/ftester/main.go index 7830da32..76d56fca 100644 --- a/backend/cmd/ftester/main.go +++ b/backend/cmd/ftester/main.go @@ -83,16 +83,26 @@ func main() { _ = obs.Observer.Drain(drainCtx) }() - // Initialize database connection + // Create this tenant's schema and repoint DATABASE_URL at it before any + // consumer reads the DSN. No-op when TENANT_ID is empty. + if err := database.EnsureTenantSchema(ctx, cfg); err != nil { + log.Fatalf("Tenant schema initialization failed: %v", err) + } + db, err := sql.Open("postgres", cfg.DatabaseURL) if err != nil { log.Fatalf("Unable to open database: %v", err) } + defer db.Close() - db.SetMaxOpenConns(10) - db.SetMaxIdleConns(2) + db.SetMaxOpenConns(min(cfg.DBMaxOpenConns, 10)) + db.SetMaxIdleConns(min(cfg.DBMaxIdleConns, 2)) db.SetConnMaxLifetime(time.Hour) + if err := database.VerifySearchPath(ctx, db, cfg); err != nil { + log.Fatalf("Tenant schema verification failed: %v", err) + } + queries := database.New(db) terminal.PrintHeader("Function Tester (ftester)") diff --git a/backend/cmd/installer/processor/pg.go b/backend/cmd/installer/processor/pg.go index 7a86d577..f44b3abc 100644 --- a/backend/cmd/installer/processor/pg.go +++ b/backend/cmd/installer/processor/pg.go @@ -4,6 +4,11 @@ import ( "context" "database/sql" "fmt" + "strconv" + "strings" + + "pentagi/pkg/config" + "pentagi/pkg/database" _ "github.com/lib/pq" "golang.org/x/crypto/bcrypt" @@ -46,38 +51,69 @@ func (p *processor) performPasswordReset(ctx context.Context, newPassword string dbName = envVar.Value } - // create connection string - connStr := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", - PostgreSQLHost, PostgreSQLPort, dbUser, dbPassword, dbName) + cfg := &config.Config{ + DatabaseURL: fmt.Sprintf( + "host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", + PostgreSQLHost, PostgreSQLPort, dbUser, dbPassword, dbName, + ), + } - // open database connection - db, err := sql.Open("postgres", connStr) + if envVar, ok := p.state.GetVar("TENANT_ID"); ok { + cfg.TenantID = strings.TrimSpace(envVar.Value) + } + if err := cfg.ValidateTenantID(); err != nil { + return err + } + if envVar, ok := p.state.GetVar("DATABASE_EXTENSIONS_SCHEMA"); ok { + cfg.DatabaseExtensionsSchema = strings.TrimSpace(envVar.Value) + } + if envVar, ok := p.state.GetVar("DATABASE_SEARCH_PATH_VIA_OPTIONS"); ok && envVar.Value != "" { + viaOptions, err := strconv.ParseBool(envVar.Value) + if err != nil { + return fmt.Errorf("invalid DATABASE_SEARCH_PATH_VIA_OPTIONS %q: %w", envVar.Value, err) + } + cfg.DatabaseSearchPathViaOptions = viaOptions + } + + // Point the DSN at the tenant schema before opening a connection. Without + // this, an UPDATE on unqualified "users" would hit public.users while the + // running instance owns .users — a silent no-op or wrong-tenant write. + if err := database.RewriteDatabaseURLForTenant(cfg); err != nil { + return fmt.Errorf("failed to apply tenant search_path: %w", err) + } + + db, err := sql.Open("postgres", cfg.DatabaseURL) if err != nil { return fmt.Errorf("failed to connect to database: %w", err) } defer db.Close() - // test connection + db.SetMaxOpenConns(1) + if err := db.PingContext(ctx); err != nil { return fmt.Errorf("failed to ping database: %w", err) } - p.appendLog(fmt.Sprintf("Connected to PostgreSQL at %s:%s (database: %s)", PostgreSQLHost, PostgreSQLPort, dbName), ProductStackPentagi, state) + if err := database.VerifySearchPath(ctx, db, cfg); err != nil { + return fmt.Errorf("tenant schema verification failed: %w", err) + } + + p.appendLog(fmt.Sprintf( + "Connected to PostgreSQL at %s:%s (database: %s, schema: %s)", + PostgreSQLHost, PostgreSQLPort, dbName, cfg.SchemaName(), + ), ProductStackPentagi, state) - // hash the new password hashedPassword, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost) if err != nil { return fmt.Errorf("failed to hash password: %w", err) } - // update the admin user password and status query := `UPDATE users SET password = $1, status = 'active' WHERE mail = $2` result, err := db.ExecContext(ctx, query, string(hashedPassword), AdminEmail) if err != nil { return fmt.Errorf("failed to update password: %w", err) } - // check if any rows were affected rowsAffected, err := result.RowsAffected() if err != nil { return fmt.Errorf("failed to get rows affected: %w", err) diff --git a/backend/cmd/installer/wizard/controller/controller.go b/backend/cmd/installer/wizard/controller/controller.go index 03c3ce51..0c897e94 100644 --- a/backend/cmd/installer/wizard/controller/controller.go +++ b/backend/cmd/installer/wizard/controller/controller.go @@ -1996,21 +1996,23 @@ func (c *controller) ResetDockerConfig() *DockerConfig { // ServerSettingsConfig represents PentAGI server settings configuration type ServerSettingsConfig struct { // direct form field mappings using loader.EnvVar - TenantID loader.EnvVar // TENANT_ID - LicenseKey loader.EnvVar // LICENSE_KEY - PprofAddr loader.EnvVar // PPROF_ADDR - ListenIP loader.EnvVar // PENTAGI_LISTEN_IP - ListenPort loader.EnvVar // PENTAGI_LISTEN_PORT - PublicURL loader.EnvVar // PUBLIC_URL - CorsOrigins loader.EnvVar // CORS_ORIGINS - CookieSigningSalt loader.EnvVar // COOKIE_SIGNING_SALT - ProxyURL loader.EnvVar // PROXY_URL - HTTPClientTimeout loader.EnvVar // HTTP_CLIENT_TIMEOUT - TerminalToolTimeout loader.EnvVar // TERMINAL_TOOL_TIMEOUT - ExternalSSLCAPath loader.EnvVar // EXTERNAL_SSL_CA_PATH - ExternalSSLInsecure loader.EnvVar // EXTERNAL_SSL_INSECURE - SSLDir loader.EnvVar // PENTAGI_SSL_DIR - DataDir loader.EnvVar // PENTAGI_DATA_DIR + TenantID loader.EnvVar // TENANT_ID + LicenseKey loader.EnvVar // LICENSE_KEY + PprofAddr loader.EnvVar // PPROF_ADDR + ListenIP loader.EnvVar // PENTAGI_LISTEN_IP + ListenPort loader.EnvVar // PENTAGI_LISTEN_PORT + PublicURL loader.EnvVar // PUBLIC_URL + CorsOrigins loader.EnvVar // CORS_ORIGINS + CookieSigningSalt loader.EnvVar // COOKIE_SIGNING_SALT + ProxyURL loader.EnvVar // PROXY_URL + HTTPClientTimeout loader.EnvVar // HTTP_CLIENT_TIMEOUT + TerminalToolTimeout loader.EnvVar // TERMINAL_TOOL_TIMEOUT + ExternalSSLCAPath loader.EnvVar // EXTERNAL_SSL_CA_PATH + ExternalSSLInsecure loader.EnvVar // EXTERNAL_SSL_INSECURE + SSLDir loader.EnvVar // PENTAGI_SSL_DIR + DataDir loader.EnvVar // PENTAGI_DATA_DIR + DatabaseExtensionsSchema loader.EnvVar // DATABASE_EXTENSIONS_SCHEMA + DatabaseSearchPathViaOpt loader.EnvVar // DATABASE_SEARCH_PATH_VIA_OPTIONS // parsed credentials for proxy server (extracted from URLs) ProxyUsername string @@ -2035,20 +2037,24 @@ func (c *controller) GetServerSettingsConfig() *ServerSettingsConfig { "EXTERNAL_SSL_INSECURE", "PENTAGI_SSL_DIR", "PENTAGI_DATA_DIR", + "DATABASE_EXTENSIONS_SCHEMA", + "DATABASE_SEARCH_PATH_VIA_OPTIONS", }) defaults := map[string]string{ - "LICENSE_KEY": "", - "PPROF_ADDR": "", - "PENTAGI_LISTEN_IP": "127.0.0.1", - "PENTAGI_LISTEN_PORT": "8443", - "PUBLIC_URL": "https://localhost:8443", - "CORS_ORIGINS": "https://localhost:8443", - "PENTAGI_DATA_DIR": "pentagi-data", - "PENTAGI_SSL_DIR": "pentagi-ssl", - "HTTP_CLIENT_TIMEOUT": "600", - "TERMINAL_TOOL_TIMEOUT": "600", - "EXTERNAL_SSL_INSECURE": "false", + "LICENSE_KEY": "", + "PPROF_ADDR": "", + "PENTAGI_LISTEN_IP": "127.0.0.1", + "PENTAGI_LISTEN_PORT": "8443", + "PUBLIC_URL": "https://localhost:8443", + "CORS_ORIGINS": "https://localhost:8443", + "PENTAGI_DATA_DIR": "pentagi-data", + "PENTAGI_SSL_DIR": "pentagi-ssl", + "HTTP_CLIENT_TIMEOUT": "600", + "TERMINAL_TOOL_TIMEOUT": "600", + "EXTERNAL_SSL_INSECURE": "false", + "DATABASE_EXTENSIONS_SCHEMA": "public", + "DATABASE_SEARCH_PATH_VIA_OPTIONS": "false", } for varName, defaultValue := range defaults { @@ -2059,21 +2065,23 @@ func (c *controller) GetServerSettingsConfig() *ServerSettingsConfig { } cfg := &ServerSettingsConfig{ - TenantID: vars["TENANT_ID"], - LicenseKey: vars["LICENSE_KEY"], - PprofAddr: vars["PPROF_ADDR"], - ListenIP: vars["PENTAGI_LISTEN_IP"], - ListenPort: vars["PENTAGI_LISTEN_PORT"], - PublicURL: vars["PUBLIC_URL"], - CorsOrigins: vars["CORS_ORIGINS"], - CookieSigningSalt: vars["COOKIE_SIGNING_SALT"], - ProxyURL: vars["PROXY_URL"], - HTTPClientTimeout: vars["HTTP_CLIENT_TIMEOUT"], - TerminalToolTimeout: vars["TERMINAL_TOOL_TIMEOUT"], - ExternalSSLCAPath: vars["EXTERNAL_SSL_CA_PATH"], - ExternalSSLInsecure: vars["EXTERNAL_SSL_INSECURE"], - SSLDir: vars["PENTAGI_SSL_DIR"], - DataDir: vars["PENTAGI_DATA_DIR"], + TenantID: vars["TENANT_ID"], + LicenseKey: vars["LICENSE_KEY"], + PprofAddr: vars["PPROF_ADDR"], + ListenIP: vars["PENTAGI_LISTEN_IP"], + ListenPort: vars["PENTAGI_LISTEN_PORT"], + PublicURL: vars["PUBLIC_URL"], + CorsOrigins: vars["CORS_ORIGINS"], + CookieSigningSalt: vars["COOKIE_SIGNING_SALT"], + ProxyURL: vars["PROXY_URL"], + HTTPClientTimeout: vars["HTTP_CLIENT_TIMEOUT"], + TerminalToolTimeout: vars["TERMINAL_TOOL_TIMEOUT"], + ExternalSSLCAPath: vars["EXTERNAL_SSL_CA_PATH"], + ExternalSSLInsecure: vars["EXTERNAL_SSL_INSECURE"], + SSLDir: vars["PENTAGI_SSL_DIR"], + DataDir: vars["PENTAGI_DATA_DIR"], + DatabaseExtensionsSchema: vars["DATABASE_EXTENSIONS_SCHEMA"], + DatabaseSearchPathViaOpt: vars["DATABASE_SEARCH_PATH_VIA_OPTIONS"], } // split proxy URL into credentials + naked URL for UI @@ -2101,21 +2109,23 @@ func (c *controller) UpdateServerSettingsConfig(config *ServerSettingsConfig) er } updates := map[string]string{ - "TENANT_ID": config.TenantID.Value, - "LICENSE_KEY": config.LicenseKey.Value, - "PPROF_ADDR": config.PprofAddr.Value, - "PENTAGI_LISTEN_IP": config.ListenIP.Value, - "PENTAGI_LISTEN_PORT": config.ListenPort.Value, - "PUBLIC_URL": config.PublicURL.Value, - "CORS_ORIGINS": config.CorsOrigins.Value, - "COOKIE_SIGNING_SALT": config.CookieSigningSalt.Value, - "PROXY_URL": proxyURL, - "HTTP_CLIENT_TIMEOUT": config.HTTPClientTimeout.Value, - "TERMINAL_TOOL_TIMEOUT": config.TerminalToolTimeout.Value, - "EXTERNAL_SSL_CA_PATH": config.ExternalSSLCAPath.Value, - "EXTERNAL_SSL_INSECURE": config.ExternalSSLInsecure.Value, - "PENTAGI_SSL_DIR": config.SSLDir.Value, - "PENTAGI_DATA_DIR": config.DataDir.Value, + "TENANT_ID": config.TenantID.Value, + "LICENSE_KEY": config.LicenseKey.Value, + "PPROF_ADDR": config.PprofAddr.Value, + "PENTAGI_LISTEN_IP": config.ListenIP.Value, + "PENTAGI_LISTEN_PORT": config.ListenPort.Value, + "PUBLIC_URL": config.PublicURL.Value, + "CORS_ORIGINS": config.CorsOrigins.Value, + "COOKIE_SIGNING_SALT": config.CookieSigningSalt.Value, + "PROXY_URL": proxyURL, + "HTTP_CLIENT_TIMEOUT": config.HTTPClientTimeout.Value, + "TERMINAL_TOOL_TIMEOUT": config.TerminalToolTimeout.Value, + "EXTERNAL_SSL_CA_PATH": config.ExternalSSLCAPath.Value, + "EXTERNAL_SSL_INSECURE": config.ExternalSSLInsecure.Value, + "PENTAGI_SSL_DIR": config.SSLDir.Value, + "PENTAGI_DATA_DIR": config.DataDir.Value, + "DATABASE_EXTENSIONS_SCHEMA": config.DatabaseExtensionsSchema.Value, + "DATABASE_SEARCH_PATH_VIA_OPTIONS": config.DatabaseSearchPathViaOpt.Value, } if err := c.SetVars(updates); err != nil { @@ -2143,6 +2153,8 @@ func (c *controller) ResetServerSettingsConfig() *ServerSettingsConfig { "EXTERNAL_SSL_INSECURE", "PENTAGI_SSL_DIR", "PENTAGI_DATA_DIR", + "DATABASE_EXTENSIONS_SCHEMA", + "DATABASE_SEARCH_PATH_VIA_OPTIONS", } if err := c.ResetVars(vars); err != nil { @@ -2395,6 +2407,8 @@ func (c *controller) getVariableDescription(varName string) string { "PENTAGI_DOCKER_CERT_PATH": locale.EnvDesc_PENTAGI_DOCKER_CERT_PATH, "PENTAGI_LLM_SERVER_CONFIG_PATH": locale.EnvDesc_PENTAGI_LLM_SERVER_CONFIG_PATH, "PENTAGI_OLLAMA_SERVER_CONFIG_PATH": locale.EnvDesc_PENTAGI_OLLAMA_SERVER_CONFIG_PATH, + "DATABASE_EXTENSIONS_SCHEMA": locale.EnvDesc_DATABASE_EXTENSIONS_SCHEMA, + "DATABASE_SEARCH_PATH_VIA_OPTIONS": locale.EnvDesc_DATABASE_SEARCH_PATH_VIA_OPTIONS, "STATIC_DIR": locale.EnvDesc_STATIC_DIR, "STATIC_URL": locale.EnvDesc_STATIC_URL, @@ -2610,26 +2624,28 @@ var criticalVariables = map[string]bool{ "MAX_LIMITED_AGENT_TOOL_CALLS": true, "AGENT_PLANNING_STEP_ENABLED": true, - "TENANT_ID": true, - "LICENSE_KEY": true, - "PPROF_ADDR": true, - "PENTAGI_LISTEN_IP": true, - "PENTAGI_LISTEN_PORT": true, - "PUBLIC_URL": true, - "CORS_ORIGINS": true, - "COOKIE_SIGNING_SALT": true, - "PROXY_URL": true, - "EXTERNAL_SSL_CA_PATH": true, - "EXTERNAL_SSL_INSECURE": true, - "STATIC_DIR": true, - "STATIC_URL": true, - "SERVER_PORT": true, - "SERVER_HOST": true, - "SERVER_SSL_CRT": true, - "SERVER_SSL_KEY": true, - "SERVER_USE_SSL": true, - "PENTAGI_SSL_DIR": true, - "PENTAGI_DATA_DIR": true, + "TENANT_ID": true, + "LICENSE_KEY": true, + "PPROF_ADDR": true, + "PENTAGI_LISTEN_IP": true, + "PENTAGI_LISTEN_PORT": true, + "PUBLIC_URL": true, + "CORS_ORIGINS": true, + "COOKIE_SIGNING_SALT": true, + "PROXY_URL": true, + "EXTERNAL_SSL_CA_PATH": true, + "EXTERNAL_SSL_INSECURE": true, + "STATIC_DIR": true, + "STATIC_URL": true, + "SERVER_PORT": true, + "SERVER_HOST": true, + "SERVER_SSL_CRT": true, + "SERVER_SSL_KEY": true, + "SERVER_USE_SSL": true, + "PENTAGI_SSL_DIR": true, + "PENTAGI_DATA_DIR": true, + "DATABASE_EXTENSIONS_SCHEMA": true, + "DATABASE_SEARCH_PATH_VIA_OPTIONS": true, // scraper settings "SCRAPER_PUBLIC_URL": true, diff --git a/backend/cmd/installer/wizard/locale/locale.go b/backend/cmd/installer/wizard/locale/locale.go index 34bda841..d8ec1f78 100644 --- a/backend/cmd/installer/wizard/locale/locale.go +++ b/backend/cmd/installer/wizard/locale/locale.go @@ -1293,23 +1293,31 @@ Examples: ServerSettingsCookieSigningSalt = "Cookie Signing Salt" ServerSettingsCookieSigningSaltDesc = "Secret used to sign cookies (keep private)" + ServerSettingsDatabaseExtensionsSchema = "Database Extensions Schema" + ServerSettingsDatabaseExtensionsSchemaDesc = "Schema holding shared extensions when Tenant ID is set (e.g., public, extensions for Supabase)" + + ServerSettingsDatabaseSearchPathViaOptions = "Search Path via Options" + ServerSettingsDatabaseSearchPathViaOptionsDesc = "Send the tenant search_path inside the options startup parameter (needed by some poolers)" + // Hints for fields overview - ServerSettingsLicenseKeyHint = "License Key" - ServerSettingsTenantIDHint = "Tenant ID" - ServerSettingsPprofAddrHint = "pprof Address" - ServerSettingsHostHint = "Listen IP" - ServerSettingsPortHint = "Listen Port" - ServerSettingsPublicURLHint = "Public URL" - ServerSettingsCORSOriginsHint = "CORS Origins" - ServerSettingsProxyURLHint = "Proxy URL" - ServerSettingsProxyUsernameHint = "Proxy Username" - ServerSettingsProxyPasswordHint = "Proxy Password" - ServerSettingsHTTPClientTimeoutHint = "HTTP Timeout" - ServerSettingsTerminalToolTimeoutHint = "Terminal Timeout" - ServerSettingsExternalSSLCAPathHint = "Custom CA Path" - ServerSettingsExternalSSLInsecureHint = "Skip SSL Verification" - ServerSettingsSSLDirHint = "SSL Directory" - ServerSettingsDataDirHint = "Data Directory" + ServerSettingsLicenseKeyHint = "License Key" + ServerSettingsTenantIDHint = "Tenant ID" + ServerSettingsPprofAddrHint = "pprof Address" + ServerSettingsHostHint = "Listen IP" + ServerSettingsPortHint = "Listen Port" + ServerSettingsPublicURLHint = "Public URL" + ServerSettingsCORSOriginsHint = "CORS Origins" + ServerSettingsProxyURLHint = "Proxy URL" + ServerSettingsProxyUsernameHint = "Proxy Username" + ServerSettingsProxyPasswordHint = "Proxy Password" + ServerSettingsHTTPClientTimeoutHint = "HTTP Timeout" + ServerSettingsTerminalToolTimeoutHint = "Terminal Timeout" + ServerSettingsExternalSSLCAPathHint = "Custom CA Path" + ServerSettingsExternalSSLInsecureHint = "Skip SSL Verification" + ServerSettingsSSLDirHint = "SSL Directory" + ServerSettingsDataDirHint = "Data Directory" + ServerSettingsDatabaseExtensionsSchemaHint = "Extensions Schema" + ServerSettingsDatabaseSearchPathViaOptionsHint = "Search Path via Options" // Help texts per-field ServerSettingsGeneralHelp = `PentAGI exposes its web UI via Docker with configurable host and port. @@ -1399,6 +1407,20 @@ When enabled, all certificate validation is bypassed, making connections vulnera ServerSettingsDataDirHelp = `Host directory for persistent data. PentAGI stores agent artifacts under flow-N subdirectories, which map to /work inside worker containers.` ServerSettingsCookieSigningSaltHelp = `Secret salt used to sign cookies. Keep it private.` + + ServerSettingsDatabaseExtensionsSchemaHelp = `PostgreSQL schema that holds the shared extensions (vector, pg_trgm) every tenant must reach through its search_path. + +Only used when Tenant ID is set; leave empty for the default "public", which is where a stock PostgreSQL install keeps them. Set it when your database follows another convention — Supabase installs its extensions into "extensions", and startup aborts with a message naming the schema it found if this does not match. + +Examples: +• public +• extensions` + + ServerSettingsDatabaseSearchPathViaOptionsHelp = `Sends the tenant search_path as options=--search_path= instead of a bare search_path connection parameter. + +Only used when Tenant ID is set. Keep it false for a direct PostgreSQL connection. Enable it when connecting through a pooler that forwards the "options" startup parameter but drops an unrecognized bare search_path — reported to be the case for some versions of Supabase's Supavisor. It is not guaranteed to work: startup fails with a clear schema-mismatch error if the value never reaches the backend. + +Values: true, false` ) // Human-in-the-loop screen strings @@ -2410,6 +2432,8 @@ const ( EnvDesc_PUBLIC_URL = "PentAGI Public URL" EnvDesc_CORS_ORIGINS = "PentAGI CORS Origins" EnvDesc_COOKIE_SIGNING_SALT = "PentAGI Cookie Signing Salt" + EnvDesc_DATABASE_EXTENSIONS_SCHEMA = "PostgreSQL Extensions Schema" + EnvDesc_DATABASE_SEARCH_PATH_VIA_OPTIONS = "PostgreSQL Search Path via Options" EnvDesc_PROXY_URL = "HTTP/HTTPS Proxy URL" EnvDesc_HTTP_CLIENT_TIMEOUT = "HTTP Client Timeout (seconds)" EnvDesc_TERMINAL_TOOL_TIMEOUT = "Terminal Tool Timeout (seconds)" diff --git a/backend/cmd/installer/wizard/models/server_settings_form.go b/backend/cmd/installer/wizard/models/server_settings_form.go index f595aa46..2319f2d8 100644 --- a/backend/cmd/installer/wizard/models/server_settings_form.go +++ b/backend/cmd/installer/wizard/models/server_settings_form.go @@ -3,6 +3,7 @@ package models import ( "fmt" "net" + "regexp" "strconv" "strings" @@ -18,6 +19,11 @@ import ( "github.com/vxcontrol/cloud/sdk" ) +// schemaNameRegex keeps DATABASE_EXTENSIONS_SCHEMA within PostgreSQL's +// unquoted-identifier rules and 63-byte limit, so a typo surfaces here rather +// than as a failed CREATE EXTENSION on first boot. +var schemaNameRegex = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]{0,62}$`) + // ServerSettingsFormModel represents the PentAGI server settings configuration form type ServerSettingsFormModel struct { *BaseScreen @@ -162,6 +168,20 @@ func (m *ServerSettingsFormModel) BuildForm() tea.Cmd { true, )) + fields = append(fields, m.createTextField("database_extensions_schema", + locale.ServerSettingsDatabaseExtensionsSchema, + locale.ServerSettingsDatabaseExtensionsSchemaDesc, + config.DatabaseExtensionsSchema, + false, + )) + + fields = append(fields, m.createTextField("database_search_path_via_options", + locale.ServerSettingsDatabaseSearchPathViaOptions, + locale.ServerSettingsDatabaseSearchPathViaOptionsDesc, + config.DatabaseSearchPathViaOpt, + false, + )) + m.SetFormFields(fields) return nil } @@ -253,6 +273,26 @@ func (m *ServerSettingsFormModel) GetCurrentConfiguration() string { sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsTenantIDHint, tenantID)) } + // both only take effect with a tenant configured, so keep them out of the + // overview of a single-instance deployment + if cfg.TenantID.Value != "" { + if schema := cfg.DatabaseExtensionsSchema.Value; schema != "" { + schema = m.GetStyles().Info.Render(schema) + sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsDatabaseExtensionsSchemaHint, schema)) + } else if schema := cfg.DatabaseExtensionsSchema.Default; schema != "" { + schema = m.GetStyles().Muted.Render(schema) + sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsDatabaseExtensionsSchemaHint, schema)) + } + + if viaOptions := cfg.DatabaseSearchPathViaOpt.Value; viaOptions == "true" { + viaOptions = m.GetStyles().Info.Render("Enabled") + sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsDatabaseSearchPathViaOptionsHint, viaOptions)) + } else { + viaOptions = m.GetStyles().Muted.Render("Disabled") + sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsDatabaseSearchPathViaOptionsHint, viaOptions)) + } + } + if pprofAddr := cfg.PprofAddr.Value; pprofAddr != "" { pprofAddr = m.GetStyles().Info.Render(pprofAddr) sections = append(sections, fmt.Sprintf("• %s: %s", locale.ServerSettingsPprofAddrHint, pprofAddr)) @@ -416,6 +456,10 @@ func (m *ServerSettingsFormModel) GetHelpContent() string { sections = append(sections, locale.ServerSettingsDataDirHelp) case "pentagi_cookie_signing_salt": sections = append(sections, locale.ServerSettingsCookieSigningSaltHelp) + case "database_extensions_schema": + sections = append(sections, locale.ServerSettingsDatabaseExtensionsSchemaHelp) + case "database_search_path_via_options": + sections = append(sections, locale.ServerSettingsDatabaseSearchPathViaOptionsHelp) default: sections = append(sections, locale.ServerSettingsFormOverview) } @@ -429,21 +473,23 @@ func (m *ServerSettingsFormModel) HandleSave() error { fields := m.GetFormFields() newCfg := &controller.ServerSettingsConfig{ - TenantID: cfg.TenantID, - LicenseKey: cfg.LicenseKey, - PprofAddr: cfg.PprofAddr, - ListenIP: cfg.ListenIP, - ListenPort: cfg.ListenPort, - CorsOrigins: cfg.CorsOrigins, - CookieSigningSalt: cfg.CookieSigningSalt, - ProxyURL: cfg.ProxyURL, - HTTPClientTimeout: cfg.HTTPClientTimeout, - TerminalToolTimeout: cfg.TerminalToolTimeout, - ExternalSSLCAPath: cfg.ExternalSSLCAPath, - ExternalSSLInsecure: cfg.ExternalSSLInsecure, - SSLDir: cfg.SSLDir, - DataDir: cfg.DataDir, - PublicURL: cfg.PublicURL, + TenantID: cfg.TenantID, + LicenseKey: cfg.LicenseKey, + PprofAddr: cfg.PprofAddr, + ListenIP: cfg.ListenIP, + ListenPort: cfg.ListenPort, + CorsOrigins: cfg.CorsOrigins, + CookieSigningSalt: cfg.CookieSigningSalt, + ProxyURL: cfg.ProxyURL, + HTTPClientTimeout: cfg.HTTPClientTimeout, + TerminalToolTimeout: cfg.TerminalToolTimeout, + ExternalSSLCAPath: cfg.ExternalSSLCAPath, + ExternalSSLInsecure: cfg.ExternalSSLInsecure, + SSLDir: cfg.SSLDir, + DataDir: cfg.DataDir, + PublicURL: cfg.PublicURL, + DatabaseExtensionsSchema: cfg.DatabaseExtensionsSchema, + DatabaseSearchPathViaOpt: cfg.DatabaseSearchPathViaOpt, } for _, field := range fields { @@ -521,6 +567,16 @@ func (m *ServerSettingsFormModel) HandleSave() error { newCfg.DataDir.Value = value case "pentagi_cookie_signing_salt": newCfg.CookieSigningSalt.Value = value + case "database_extensions_schema": + if value != "" && !schemaNameRegex.MatchString(value) { + return fmt.Errorf("invalid extensions schema: must match %s", schemaNameRegex.String()) + } + newCfg.DatabaseExtensionsSchema.Value = value + case "database_search_path_via_options": + if value != "" && value != "true" && value != "false" { + return fmt.Errorf("invalid value for search path via options: must be 'true' or 'false'") + } + newCfg.DatabaseSearchPathViaOpt.Value = value } } diff --git a/backend/cmd/pentagi/main.go b/backend/cmd/pentagi/main.go index cfd6a40f..ecf3715a 100644 --- a/backend/cmd/pentagi/main.go +++ b/backend/cmd/pentagi/main.go @@ -96,7 +96,7 @@ func main() { // Create this tenant's schema and repoint DATABASE_URL at it before any // consumer reads the DSN. No-op when TENANT_ID is empty. - if err := ensureTenantSchema(ctx, cfg); err != nil { + if err := database.EnsureTenantSchema(ctx, cfg); err != nil { logrus.WithError(err).Fatal("Tenant schema initialization failed") } @@ -109,7 +109,7 @@ func main() { db.SetMaxIdleConns(cfg.DBMaxIdleConns) db.SetConnMaxLifetime(time.Hour) - if err := verifySearchPath(ctx, db, cfg); err != nil { + if err := database.VerifySearchPath(ctx, db, cfg); err != nil { logrus.WithError(err).Fatal("Tenant schema verification failed") } @@ -146,10 +146,16 @@ func main() { logrus.WithError(err).Fatal("Database dialect configuration failed") } + // goose's own queries are unqualified, so without this a fresh tenant + // schema silently inherits public's version table via search_path and + // skips its migrations. See pkg/database/tenant.go for the + // schema/search_path setup. + goose.SetTableName(cfg.SchemaName() + ".goose_db_version") + // Hold an advisory lock so simultaneous boots cannot execute the same // migration set concurrently; the initial migration uses bare CREATE TABLE, // so the loser would otherwise abort on "relation already exists". - if err := runMigrations(ctx, db, cfg, func(db *sql.DB) error { + if err := database.RunMigrations(ctx, db, cfg, func(db *sql.DB) error { return goose.Up(db, "sql") }); err != nil { // Fatal: continuing on a half-migrated schema and then serving traffic is diff --git a/backend/docs/config.md b/backend/docs/config.md index 5e82ff8f..329d9e87 100644 --- a/backend/docs/config.md +++ b/backend/docs/config.md @@ -11,9 +11,15 @@ This document serves as a comprehensive guide to the configuration system in Pen - [Still Server-Managed](#still-server-managed) - [General Settings](#general-settings) - [Multi-Instance Deployment (`TENANT_ID`)](#multi-instance-deployment-tenant_id) + - [What the application namespaces automatically](#what-the-application-namespaces-automatically) + - [What stays the operator's responsibility](#what-stays-the-operators-responsibility) + - [Deployment topologies](#deployment-topologies) + - [Extensions installed outside `public` (`DATABASE_EXTENSIONS_SCHEMA`)](#extensions-installed-outside-public-database_extensions_schema) + - [Multi-tenant PostgreSQL access through PgBouncer](#multi-tenant-postgresql-access-through-pgbouncer) + - [Multi-tenant PostgreSQL through Supabase's Supavisor pooler (`DATABASE_SEARCH_PATH_VIA_OPTIONS`)](#multi-tenant-postgresql-through-supabases-supavisor-pooler-database_search_path_via_options) - [Usage Details](#usage-details) - [Docker Settings](#docker-settings) - - [Worker Docker Access (DOCKER_INSIDE_*)](#worker-docker-access-docker_inside_) + - [Worker Docker Access (`DOCKER_INSIDE_*`)](#worker-docker-access-docker_inside_) - [Usage Details](#usage-details-1) - [Server Settings](#server-settings) - [Usage Details](#usage-details-2) @@ -136,6 +142,8 @@ These settings control basic application behavior and are foundational for the s | Option | Environment Variable | Default Value | Description | | ---------------- | --------------------------- | ---------------------------------------------------------------------------- | ------------------------------------------------------------------------ | | DatabaseURL | `DATABASE_URL` | `postgres://pentagiuser:pentagipass@pgvector:5432/pentagidb?sslmode=disable` | Connection string for the PostgreSQL database with pgvector extension | +| DatabaseExtensionsSchema | `DATABASE_EXTENSIONS_SCHEMA` | `public` | Schema every tenant's search_path must include for shared extensions (`vector`, `pg_trgm`) to resolve. Only used when `TenantID` is set. Override for databases that install extensions elsewhere by convention, e.g. Supabase uses `extensions`. See [Extensions installed outside `public`](#extensions-installed-outside-public-database_extensions_schema). | +| DatabaseSearchPathViaOptions | `DATABASE_SEARCH_PATH_VIA_OPTIONS` | `false` | Sends the tenant search_path as `options=--search_path=` instead of a bare `search_path` parameter. Only used when `TenantID` is set. For poolers that forward `options` but drop an unrecognized bare `search_path` (e.g. some Supabase Supavisor versions). See [Multi-tenant PostgreSQL through Supabase's Supavisor pooler](#multi-tenant-postgresql-through-supabases-supavisor-pooler-database_search_path_via_options). | | DBMaxOpenConns | `DATABASE_MAX_OPEN_CONNS` | `25` | Maximum open connections in the shared `sql.DB` pool (sqlc + GORM combined). See [database.md §Connection Pooling](database.md#connection-pooling). | | DBMaxIdleConns | `DATABASE_MAX_IDLE_CONNS` | `5` | Maximum idle connections kept open between requests | | DBVectorMaxConns | `DATABASE_VECTOR_MAX_CONNS` | `10` | Maximum connections in the shared `pgxpool` for all pgvector stores | @@ -171,7 +179,7 @@ Hyphens are excluded deliberately: they separate the tenant from the rest of a g | Area | Effect | | --- | --- | -| PostgreSQL | A schema named after the tenant is created on boot and `search_path` is set to `,public`; the DSN is rewritten once so sqlc, GORM, goose and the pgvector pool all follow. Extensions (`vector`, `pg_trgm`) stay shared in `public`. | +| PostgreSQL | A schema named after the tenant is created on boot and `search_path` is set to `,`; the DSN is rewritten once so sqlc, GORM, goose and the pgvector pool all follow. Extensions (`vector`, `pg_trgm`) stay shared in `DATABASE_EXTENSIONS_SCHEMA` (default `public`). | | Worker containers | Sandbox container names become `-pentagi-terminal-`; the per-flow volume and the container hostname derive from that name automatically. Both are labelled `pentagi.tenant`, so daemon-wide sweeps can filter by owner. | | Host ports | Per-flow sandbox ports are allocated from `DOCKER_PORTS_BASE` (default `28000`), giving each instance the window `[base, base+2000)`. | | Knowledge graph | Graphiti/Neo4j group ids become `-flow-`. The GraphQL API contract is unchanged — clients still send `flow-` and the server rebuilds the namespaced key internally. | @@ -188,7 +196,7 @@ Hyphens are excluded deliberately: they separate the tenant from the rest of a g | `DOCKER_NETWORK` | Instances may legitimately share one Docker network; the application never renames it. Set it explicitly if you want separate networks. | | Published ports | `PENTAGI_LISTEN_PORT`, `PGVECTOR_LISTEN_PORT`, `SCRAPER_LISTEN_PORT`, `PPROF_ADDR` and `DOCKER_PORTS_BASE` are host-level resources, not string namespaces. | | `INSTALLATION_ID` | Unique per installation, or left empty to be generated once and cached in `DATA_DIR`. | -| Database privileges | The configured user needs `CREATE SCHEMA`, and on first boot `CREATE EXTENSION` — unless an administrator pre-installed `vector` and `pg_trgm` into `public`. | +| Database privileges | The configured user needs `CREATE SCHEMA`, and on first boot `CREATE EXTENSION` — unless an administrator pre-installed `vector` and `pg_trgm` into `DATABASE_EXTENSIONS_SCHEMA` (default `public`). | The values actually in effect are written to the startup log under `Instance identity` (`tenant_id`, `data_dir`, `schema`, `installation_id`), which is the quickest way to confirm two instances are not sharing something they should not. @@ -205,6 +213,53 @@ The installer provisions **one** instance per server; it does not manage several **Upgrading an existing deployment.** Leave `TENANT_ID` empty. The instance keeps using `public` and its current data directory; no migration is required. Setting `TENANT_ID` on an existing installation points it at a **new, empty schema** — the data in `public` is not migrated and will appear to be gone. Do not point an existing `public` deployment at a `search_path` that lists another tenant's schema. +#### Extensions installed outside `public` (`DATABASE_EXTENSIONS_SCHEMA`) + +`ensureTenantSchema` requires `vector` and `pg_trgm` to already live in (or be creatable in) one schema common to every tenant, because it must be part of every tenant's `search_path`. That schema defaults to `public`, which is where a stock PostgreSQL/`docker-compose.yml` install keeps them. + +Managed providers do not always follow that convention. **Supabase** (cloud and self-hosted) installs its bundled extensions into a dedicated `extensions` schema instead, and its default roles get `extensions` added to their `search_path` for exactly that reason. Pointing `DATABASE_URL` at such a database with `TENANT_ID` set fails fast on boot: + +``` +Tenant schema initialization failed: extension "vector" is installed in schema "extensions", +but multi-tenant mode requires it in "public" so every tenant can reach it; either run +ALTER EXTENSION vector SET SCHEMA public, or set DATABASE_EXTENSIONS_SCHEMA=extensions +to match where it already lives +``` + +Set `DATABASE_EXTENSIONS_SCHEMA=extensions` (or whatever schema the error reports) instead of moving the extension with `ALTER EXTENSION ... SET SCHEMA` — the schema only needs to be part of the `search_path` PentAGI computes (`,`), moving a provider-managed extension out of its documented location is unnecessary and risks breaking whatever else that provider expects to find it there. + +#### Multi-tenant PostgreSQL access through PgBouncer + +`DATABASE_URL` can point at a PgBouncer instance instead of PostgreSQL directly, but three things have to be true, independent of each other: + +1. **`pool_mode = session` on the PgBouncer side.** PentAGI holds a `pg_advisory_lock`/`pg_advisory_unlock` pair on one dedicated connection across the whole tenant-bootstrap + migration sequence (`backend/pkg/database/tenant.go`), and `pgx` (used by the pgvector pool) caches server-side prepared statements by default. Both break silently under `transaction`/`statement` pooling, because PgBouncer is then free to hand the client a different backend connection between statements. This requirement is unrelated to tenancy — it applies even with `TENANT_ID` empty. +2. **`ignore_startup_parameters = search_path` in `pgbouncer.ini`.** With a tenant configured, PentAGI's own DSN already carries `?search_path=,` (see the table above). PgBouncer validates startup parameters from the client against a small built-in allowlist and rejects anything else with `unsupported startup parameter: search_path` unless it is explicitly ignored. Ignoring it does **not** apply the value — it only stops PgBouncer from rejecting the connection — so this step alone is not sufficient; see the next point. +3. **A `connect_query` per tenant in PgBouncer's `[databases]` section**, since PentAGI's own `search_path` startup parameter is ignored per point 2 above. `connect_query` runs on PgBouncer's own connection to PostgreSQL before any client statement, so it is not subject to the client-facing startup-parameter allowlist and works regardless of pool mode: + + ```ini + [databases] + pentagi_testing = host=pgvector port=5432 dbname=pentagidb pool_mode=session connect_query='SET search_path TO testing,public' + pentagi_acme = host=pgvector port=5432 dbname=pentagidb pool_mode=session connect_query='SET search_path TO acme,public' + ``` + + Point each instance's `DATABASE_URL` at its own virtual database name (`pentagi_testing`, `pentagi_acme`, ...) rather than the shared `pentagidb` — PgBouncer pools per `(user, dbname)` pair, so distinct virtual names are what keeps the tenants' pools, and therefore their `connect_query`, apart. + + (`track_extra_parameters = search_path` is PgBouncer's other mechanism for this, but it only works when PostgreSQL reports `search_path` changes back to the client, which requires PostgreSQL 18+ or Citus 12+ — not an option against the PostgreSQL 16/17 that ships in `docker-compose.yml`.) + +#### Multi-tenant PostgreSQL through Supabase's Supavisor pooler (`DATABASE_SEARCH_PATH_VIA_OPTIONS`) + +Supabase's shared/self-hosted pooler (Supavisor) is not PgBouncer and none of its `[databases]`/`connect_query` configuration exists for it, so the PgBouncer recipe above does not apply. Reports on whether Supavisor forwards a tenant's `search_path` at all are inconsistent — see [supabase/supavisor#206](https://github.com/supabase/supavisor/issues/206) — and depend on the Supavisor version (a parsing fix landed in [PR #768](https://github.com/supabase/supavisor/pull/768)). + +If bypassing the pooler entirely (connecting straight to the underlying PostgreSQL, or to a Supabase project's "Direct connection"/IPv4-add-on string) is not an option, set: + +``` +DATABASE_SEARCH_PATH_VIA_OPTIONS=true +``` + +This sends the tenant's search_path as `options=--search_path=,` instead of a bare `search_path=` parameter. Some poolers forward the `options` startup parameter through to the real backend while silently dropping an unrecognized bare `search_path` — this is exactly the workaround reported to work against some Supavisor versions. **It is not guaranteed** — verify it actually took effect by checking that the app starts (`verifySearchPath` fails fast with a clear error if it did not) rather than assuming success from the flag alone. + +This flag changes nothing for a direct PostgreSQL connection or a PgBouncer setup already following the recipe above; both accept `search_path` and `options` equally, so there is no reason to enable it outside a Supavisor-fronted deployment. + ### Usage Details - **DatabaseURL**: This is a critical setting used throughout the application for all database connections. It is used to: diff --git a/backend/docs/database.md b/backend/docs/database.md index 4a94b466..5455dbba 100644 --- a/backend/docs/database.md +++ b/backend/docs/database.md @@ -1,908 +1,356 @@ -# Database Package Documentation +# Database Layer ## Overview -The `database` package is a core component of PentAGI that provides a robust, type-safe interface for interacting with PostgreSQL database operations. Built on top of [sqlc](https://sqlc.dev/), this package automatically generates Go code from SQL queries, ensuring compile-time safety and eliminating the need for manual ORM mapping. +PentAGI stores application state in PostgreSQL and uses the `vector` extension for agent memory and knowledge search. The database layer combines: -PentAGI uses PostgreSQL with the [pgvector](https://github.com/pgvector/pgvector) extension to support vector embeddings for AI-powered semantic search and memory storage capabilities. +- **sqlc v1.27.0** for the main type-safe query API (`backend/pkg/database`); +- **GORM v1** (`github.com/jinzhu/gorm`) for HTTP server models and handlers (`backend/pkg/server/models`); +- **goose v3** for embedded, ordered schema migrations (`backend/migrations`); +- **lib/pq** for the shared `database/sql` pool used by sqlc and GORM; +- **pgxpool** for every `pgvector.Store` instance. -## Architecture +SQL queries live in `backend/sqlc/models`. Schema history lives in `backend/migrations/sql`. -### Database Technology Stack +This document describes how the product uses the database. It is not a substitute for the schema itself. Sources of truth, in descending order: -- **Database Engine**: PostgreSQL 15+ with pgvector extension -- **Code Generation**: sqlc for type-safe SQL-to-Go compilation -- **ORM Support**: GORM v1 for advanced operations and HTTP server handlers -- **Schema Management**: Database migrations located in `backend/migrations/` -- **Vector Operations**: pgvector extension for AI embeddings and semantic search +1. `backend/migrations/sql/*.sql` — schema and data migrations; +2. `backend/sqlc/models/*.sql` — application queries; +3. `backend/pkg/database/*.sql.go`, `models.go`, `querier.go` — generated Go API; +4. `backend/cmd/pentagi/main.go` and `backend/pkg/database/tenant.go` — connection, migration and tenant bootstrap behavior; +5. `backend/pkg/database/{database.go,converter,knowledge}` — helpers and higher-level database services. -### Entity Relationship Model +Environment variables, PgBouncer and Supavisor setup are documented in [config.md](config.md); this file only covers the product behavior those settings enable. -The database follows PentAGI's hierarchical data model for penetration testing workflows: +## Runtime Architecture -``` -Flow (Top-level workflow) -├── Task (Major testing phases) -│ └── SubTask (Specific agent assignments) -│ └── Action (Individual operations) -│ ├── Artifact (Output files/data) -│ └── Memory (Knowledge/observations) -└── Assistant (AI assistants for flows) - └── AssistantLog (Assistant interaction logs) +### Startup sequence + +`backend/cmd/pentagi/main.go` initializes PostgreSQL in this order: + +1. Load and validate configuration (including `TENANT_ID`). +2. If `TENANT_ID` is set, create the tenant schema, validate shared extension placement and rewrite `DATABASE_URL` with the tenant `search_path` (`database.EnsureTenantSchema`). +3. Open one `*sql.DB` through `lib/pq`. +4. Configure the shared `database/sql` connection pool (`DATABASE_MAX_OPEN_CONNS`, `DATABASE_MAX_IDLE_CONNS`, one-hour max lifetime). +5. Verify that `current_schema()` resolves to the configured tenant schema (`database.VerifySearchPath`). +6. Build sqlc `Queries` and GORM on the same `*sql.DB`. +7. Create one shared `pgxpool.Pool` for every pgvector store and attach it to `cfg.PgxPool`. +8. Configure goose with the schema-qualified version table and run embedded migrations under a PostgreSQL advisory lock (`database.RunMigrations`). +9. Start controllers and the API server. + +The process refuses to serve traffic after a tenant schema mismatch or a migration failure. The same tenant helpers are reused by utility binaries (`ftester`, `etester`) and by the installer's password-reset path when they talk to PostgreSQL. + +### Database clients and pools + +PentAGI opens two independent pools to the same PostgreSQL database: + +| Pool | Configuration | Default | Consumers | +|---|---|---:|---| +| `database/sql` (`lib/pq`) | `DATABASE_MAX_OPEN_CONNS` / `DATABASE_MAX_IDLE_CONNS` | `25` / `5` | sqlc and GORM | +| `pgxpool.Pool` | `DATABASE_VECTOR_MAX_CONNS` | `10` | agent memory and knowledge pgvector stores | + +GORM does not open another pool: + +```go +queries := database.New(db) +orm, err := database.NewGorm(db, cfg.Debug) ``` -Additional supporting entities include: -- **Container**: Docker containers for isolated execution -- **User**: System users with role-based access -- **MsgChain**: LLM conversation chains -- **ToolCall**: Function calls made by AI agents -- **Various Logs**: Comprehensive audit trail for all operations +During normal startup the shared pgx pool is stored in `cfg.PgxPool` and passed to vector stores with `pgvector.WithConn(cfg.PgxPool)`. Tool executors and the knowledge API therefore reuse the same pool. Their utility/test fallback uses `pgvector.WithConnectionURL(cfg.DatabaseURL)` only when `PgxPool` is nil. -## SQL Query Organization +### Connection budget -The database package is built on a comprehensive set of SQL queries organized by entity type in the `backend/sqlc/models/` directory. Each file contains CRUD operations and specialized queries for its respective entity. +With the defaults, one PentAGI process can use up to 35 PostgreSQL connections: 25 through `database/sql` and 10 through `pgxpool`. This is a hard application budget, not a prediction of steady-state usage. -### Query File Structure - -| File | Entity | Purpose | -| -------------------- | ------------ | ---------------------------------- | -| `flows.sql` | Flow | Top-level workflow management and analytics | -| `tasks.sql` | Task | Task lifecycle and status tracking | -| `subtasks.sql` | SubTask | Agent assignment and execution | -| `assistants.sql` | Assistant | AI assistant management | -| `containers.sql` | Container | Docker environment tracking | -| `users.sql` | User | User management and authentication | -| `roles.sql` | Role | Role-based access control | -| `prompts.sql` | Prompt | User-defined prompt templates | -| `providers.sql` | Provider | LLM provider configurations | -| `msgchains.sql` | MsgChain | LLM conversation chains and usage stats | -| `toolcalls.sql` | ToolCall | AI function call tracking and analytics | -| `screenshots.sql` | Screenshot | Visual artifacts storage | -| `analytics.sql` | Analytics | Flow execution time and hierarchy analytics | -| **Logging Entities** | | | -| `agentlogs.sql` | AgentLog | Inter-agent communication | -| `assistantlogs.sql` | AssistantLog | Human-assistant interactions | -| `msglogs.sql` | MsgLog | General message logging | -| `searchlogs.sql` | SearchLog | External search operations | -| `termlogs.sql` | TermLog | Terminal command execution | -| `vecstorelogs.sql` | VecStoreLog | Vector database operations | - -### Query Naming Conventions - -sqlc queries follow consistent naming patterns: +When several PentAGI instances share one database server, add both pool limits for every instance and leave capacity for PostgreSQL reserved connections, autovacuum, monitoring clients, platform services and administrative/migration sessions. ```sql --- CRUD Operations --- name: Create[Entity] :one --- name: Get[Entity] :one --- name: Get[Entities] :many --- name: Update[Entity] :one --- name: Delete[Entity] :exec/:one +SELECT name, setting +FROM pg_settings +WHERE name IN ('max_connections', 'superuser_reserved_connections'); --- Scoped Operations --- name: GetUser[Entity] :one --- name: GetUser[Entities] :many --- name: GetFlow[Entity] :one --- name: GetFlow[Entities] :many - --- Specialized Queries --- name: Get[Entity][Condition] :many --- name: Update[Entity][Field] :one +SELECT application_name, client_addr, state, count(*) +FROM pg_stat_activity +WHERE pid <> pg_backend_pid() +GROUP BY 1, 2, 3 +ORDER BY count(*) DESC; ``` -### Security and Multi-tenancy Patterns +The stock Compose stack uses `vxcontrol/pgvector:latest`. Do not assume a specific PostgreSQL major version from the image tag; inspect `SHOW server_version` on the deployed database. -Most queries implement user-scoped access through JOIN operations: +## Package Layout -```sql --- Example: User-scoped flow access --- name: GetUserFlow :one -SELECT f.* -FROM flows f -INNER JOIN users u ON f.user_id = u.id -WHERE f.id = $1 AND f.user_id = $2 AND f.deleted_at IS NULL; +| Path | Responsibility | +|---|---| +| `backend/migrations/migrations.go` | Embeds all goose migration SQL into the binary | +| `backend/migrations/sql/` | Authoritative ordered schema and data migration history | +| `backend/sqlc/sqlc.yml` | sqlc input, type overrides and output configuration | +| `backend/sqlc/models/` | Hand-written parameterized SQL queries | +| `backend/pkg/database/db.go` | Generated `DBTX`, `Queries`, `New` and `WithTx` | +| `backend/pkg/database/models.go` | Generated table models and PostgreSQL enum wrappers | +| `backend/pkg/database/querier.go` | Generated `Querier` interface covering all sqlc operations | +| `backend/pkg/database/*.sql.go` | Generated query implementations and result/parameter structs | +| `backend/pkg/database/database.go` | Null helpers, UTF-8 sanitization and shared GORM initialization | +| `backend/pkg/database/tenant.go` | Tenant schema bootstrap, DSN rewrite, search-path verification and advisory locks (shared by pentagi, ftester, etester, installer) | +| `backend/pkg/database/converter/` | Conversion from database rows to GraphQL models plus execution analytics calculations | +| `backend/pkg/database/knowledge/` | Knowledge-store business logic over sqlc, pgvector embeddings and GraphQL subscriptions | +| `backend/pkg/server/models/` | GORM v1 models used by REST/server services | +| `backend/cmd/pentagi/main.go` | Production pool creation, migration execution and dependency wiring | --- Example: Flow-scoped task access --- name: GetFlowTasks :many -SELECT t.* -FROM tasks t -INNER JOIN flows f ON t.flow_id = f.id -WHERE t.flow_id = $1 AND f.deleted_at IS NULL -ORDER BY t.created_at ASC; +sqlc and GORM are both active. Controllers and GraphQL paths primarily use generated sqlc queries; REST/server services continue to use GORM models for authentication, users, logs, settings, analytics, resources, flow files and related endpoints. Both clients share the same `*sql.DB`. GORM never owns schema creation. + +## Schema and Data Model + +### Workflow hierarchy + +```text +users + └── flows + ├── tasks + │ └── subtasks + ├── containers + ├── assistants + ├── msgchains + ├── toolcalls + ├── screenshots + └── operational logs ``` -### Soft Delete Implementation +There are no relational `actions`, `artifacts` or `memories` tables. Individual operations are represented by tool calls and specialized log tables. Agent memory and knowledge documents are stored in the LangChain pgvector tables. -Critical entities implement soft deletes to maintain audit trails: +### Table groups -```sql --- Soft delete operation --- name: DeleteFlow :one -UPDATE flows -SET deleted_at = CURRENT_TIMESTAMP -WHERE id = $1 -RETURNING *; +#### Identity and authorization --- All queries filter soft-deleted records -WHERE f.deleted_at IS NULL +| Table | Key fields / notes | +|---|---| +| `users` | Local/OAuth identity (`type`, `mail`, `hash`, `password`, `provider`), `status`, `role_id`, `password_change_required` | +| `roles` | Built-in application roles (seeded in the initial migration) | +| `privileges` | Per-role permission names used by REST and GraphQL authorization; grants evolve through later privilege migrations (not RLS) | +| `api_tokens` | `token_id`, `user_id`, `role_id`, `ttl`, `status`, soft deletion via `deleted_at` | +| `user_preferences` | One JSONB preferences document per user, including favorite-flow state | + +#### Workflow and interaction + +| Table | Key fields / notes | +|---|---| +| `flows` | Status, title, model, provider name/type, language, functions JSON, `tool_call_id_template`, optional `trace_id`, soft deletion | +| `tasks` | Status, title, input, result; owned by `flow_id` | +| `subtasks` | Status, title, description, result, persisted `context`; owned by `task_id` | +| `containers` | Type (`primary`/`secondary`), name, image, status, optional Docker `local_id`/`local_dir` | +| `assistants` | Flow-scoped interactive assistants with model/provider/functions, `use_agents`, optional `msgchain_id`, soft deletion | +| `msgchains` | LLM chain JSON plus usage (`usage_in`/`out`, cache, cost) and `duration_seconds` | +| `toolcalls` | `call_id`, name, args JSON, result, status, `duration_seconds` | +| `flow_templates` | User-owned reusable flow descriptions (`title`, `text`) | + +`flows` and `assistants` support soft deletion. Normal API deletion marks a flow's `deleted_at` and leaves child rows in place for audit/history; foreign-key cascades only run if the flow row is physically deleted. + +#### Configuration and user content + +| Table | Key fields / notes | +|---|---| +| `providers` | User-owned LLM provider configs (`type`, `name`, `config` JSON), soft deletion | +| `prompts` | User-owned prompt templates keyed by `PROMPT_TYPE` | +| `user_resources` | Uploaded file/directory metadata (`hash`, `name`, `path`, `size`, `is_dir`) | + +Flow and assistant rows contain model/provider selection and runtime function configuration. Prompt templates themselves are stored in `prompts`, not in current flow or assistant rows. + +#### Logs and artifacts + +| Table | Key fields / notes | +|---|---| +| `agentlogs` | Agent-to-agent delegation (`initiator`, `executor`, task/result text) | +| `assistantlogs` | Assistant messages with optional `thinking` and `result_format` | +| `msglogs` | General flow messages with optional `thinking` and `result_format` | +| `searchlogs` | Search engine calls (`engine`, query/result) | +| `termlogs` | Terminal stdin/stdout/stderr; **requires** `container_id` and `flow_id` | +| `vecstorelogs` | Vector-store ops (`action`, filter JSON, query/result) | +| `screenshots` | Screenshot metadata (`name`, `url`); **requires** `flow_id` | + +Several log/artifact tables carry nullable `task_id` and `subtask_id` in addition to a required `flow_id`, allowing flow-, task- and subtask-level retrieval. + +#### Vector knowledge and memory + +| Table | Purpose | +|---|---| +| `langchain_pg_collection` | Logical pgvector collections (`name`, `cmetadata`, `uuid`) | +| `langchain_pg_embedding` | Document text, vector embedding and JSON `cmetadata` | + +PentAGI uses the collection named `langchain`. Ownership and association are represented in `cmetadata` fields such as `user_id`, `flow_id`, `task_id`, `subtask_id`, `doc_type`, `question`, `description`, `guide_type`, `answer_type`, `code_lang`, chunk sizing and a `manual` flag. See `backend/pkg/database/knowledge`. + +The knowledge query API excludes `doc_type = 'memory'` from user-managed knowledge listings/searches. GraphQL and REST flow-deletion paths issue an explicit best-effort deletion of memory rows (`DeleteFlowMemoryDocuments`); this is application behavior, not a database trigger or foreign-key cascade. The knowledge migration intentionally does not drop the LangChain tables on downgrade because they may contain production data managed by the vector store. + +### Enums + +PostgreSQL enums are migrated explicitly and generated as Go string types in `models.go`. Current values: + +| Enum | Values | +|---|---| +| `FLOW_STATUS` / `TASK_STATUS` / `SUBTASK_STATUS` / `ASSISTANT_STATUS` | `created`, `running`, `waiting`, `finished`, `failed` | +| `CONTAINER_STATUS` | `starting`, `running`, `stopped`, `deleted`, `failed` | +| `CONTAINER_TYPE` | `primary`, `secondary` | +| `TOOLCALL_STATUS` | `received`, `running`, `finished`, `failed` | +| `TOKEN_STATUS` | `active`, `revoked` | +| `USER_STATUS` | `created`, `active`, `blocked` | +| `USER_TYPE` | `local`, `oauth` | +| `MSGCHAIN_TYPE` | `primary_agent`, `reporter`, `generator`, `refiner`, `reflector`, `enricher`, `adviser`, `coder`, `memorist`, `searcher`, `installer`, `pentester`, `summarizer`, `tool_call_fixer`, `assistant` | +| `MSGLOG_TYPE` | `answer`, `report`, `thoughts`, `browser`, `terminal`, `file`, `search`, `advice`, `ask`, `input`, `done` | +| `MSGLOG_RESULT_FORMAT` | `plain`, `markdown`, `terminal` | +| `TERMLOG_TYPE` | `stdin`, `stdout`, `stderr` | +| `VECSTORE_ACTION_TYPE` | `retrieve`, `store` | +| `PROVIDER_TYPE` | `openai`, `anthropic`, `gemini`, `bedrock`, `ollama`, `custom`, `deepseek`, `glm`, `kimi`, `qwen`, `minimax` | +| `SEARCHENGINE_TYPE` | `google`, `tavily`, `firecrawl`, `traversaal`, `browser`, `duckduckgo`, `perplexity`, `searxng`, `sploitus` | +| `PROMPT_TYPE` | Agent/system prompt keys from `primary_agent` through `task_assignment_wrapper` (full list in `models.go`) | + +Never add an enum value only in Go code. Add or replace the PostgreSQL enum in a goose migration, regenerate sqlc, and update backend validation where applicable. Provider and search-engine additions have additional project steps documented in `CLAUDE.md`. + +### Data lifecycle and integrity + +Primary keys use PostgreSQL identity columns. Foreign keys define ownership and use `ON DELETE CASCADE` for records that have no meaning without their parent, including user-owned settings/content and flow-owned execution data. Cascades apply only to physical deletion; ordinary flow, assistant, provider and API-token deletion paths use `deleted_at` where their schema supports soft deletion. + +The shared `update_modified_column()` trigger maintains `updated_at` for mutable entities such as flows, tasks, subtasks, containers, tool calls, message chains, assistants, providers, API tokens, preferences, templates and resources. Generated models represent database-default timestamps with `sql.NullTime`, so callers should not assume a non-null Go `time.Time` before insertion/returning. + +Important integrity constraints include unique user mail/hash values, one prompt type per user, one preferences row per user, active provider names per user, API-token identifiers, resource paths per user and non-empty template/resource text fields. JSON configuration and chain payloads use PostgreSQL `JSON`; preferences use `JSONB` with a GIN index; optional relationships use nullable SQL columns and generated `sql.Null*` wrappers. + +## Application Scoping and Deployment Tenancy + +PentAGI has two distinct isolation layers. They solve different problems and must not be confused. + +### User scoping inside one PentAGI instance + +Rows such as flows, providers, prompts, templates, resources and preferences carry `user_id` directly or are reached through a flow owned by a user. User-facing handlers select user-scoped sqlc methods such as `GetUserFlow` / `GetUserFlows`; admin paths may intentionally use unscoped variants such as `GetFlow` / `GetFlows`. + +This access control is implemented by application queries and privilege checks. PentAGI does not rely on PostgreSQL row-level security for its own tables. + +### Instance scoping with `TENANT_ID` + +`TENANT_ID` isolates independent PentAGI installations that share a PostgreSQL database. It creates one PostgreSQL schema per instance (`public` when empty, otherwise the tenant name). For a tenant, the effective search path is `,`. + +Tenant bootstrap (`backend/pkg/database/tenant.go`): + +- validates `TENANT_ID`; +- creates the tenant schema under the `pentagi-tenant-bootstrap` advisory lock; +- ensures `vector` and `pg_trgm` exist in the configured shared extension schema; +- refuses to move provider-managed extensions automatically; +- rewrites `DATABASE_URL` once, before sqlc, GORM, goose or pgxpool consume it; +- verifies `current_schema()` and aborts on mismatch. + +The goose version table is schema-qualified as `.goose_db_version`. This prevents a new tenant from reading `public.goose_db_version`, incorrectly deciding that migrations are already applied, and starting with an empty schema. + +For the complete multi-instance deployment contract (non-database resources, validation regex, PgBouncer `connect_query`, Supavisor `DATABASE_SEARCH_PATH_VIA_OPTIONS`, Supabase `DATABASE_EXTENSIONS_SCHEMA=extensions`), see [config.md](config.md#multi-instance-deployment-tenant_id). + +## Migrations + +Migration files are stored in `backend/migrations/sql` and embedded by `backend/migrations/migrations.go`: + +```go +//go:embed sql/*.sql +var EmbedMigrations embed.FS ``` -### Logging Query Patterns +At startup goose uses that filesystem and runs `goose.Up`. Migrations use goose `Up`/`Down` annotations and numeric filename prefixes. The current migration head is determined by the newest migration file; do not hard-code it in application logic. -Logging entities follow consistent patterns for audit trails: +PentAGI serializes: -```sql --- name: CreateAgentLog :one -INSERT INTO agentlogs ( - initiator, -- AI agent that initiated the action - executor, -- AI agent that executed the action - task, -- Description of the task - result, -- JSON result of the operation - flow_id, -- Associated flow - task_id, -- Associated task (nullable) - subtask_id -- Associated subtask (nullable) -) VALUES ( - $1, $2, $3, $4, $5, $6, $7 -) RETURNING *; +1. tenant schema/extension bootstrap with the `pentagi-tenant-bootstrap` advisory lock; +2. migrations with a schema-specific `pentagi-migrations-` advisory lock. --- Hierarchical retrieval with security joins --- name: GetFlowAgentLogs :many -SELECT al.* -FROM agentlogs al -INNER JOIN flows f ON al.flow_id = f.id -WHERE al.flow_id = $1 AND f.deleted_at IS NULL -ORDER BY al.created_at ASC; -``` +The lock is held on a dedicated `*sql.Conn` because PostgreSQL session advisory locks belong to a physical connection, not to a `*sql.DB` pool. -### Complex Query Examples +### Adding a migration -#### Message Chain Management +1. Create a uniquely ordered SQL file in `backend/migrations/sql`. +2. Add `-- +goose Up` and, where safe, a reversible `Down`. +3. Use `-- +goose StatementBegin`/`StatementEnd` for multi-statement units. +4. Preserve tenant compatibility: unqualified application objects must be created in the active tenant schema, while extensions remain in the shared extension schema. +5. Regenerate sqlc if the schema or query types changed. +6. Test both a fresh database and an upgrade from the previous migration head. -```sql --- Get conversation chains for a specific task --- name: GetTaskPrimaryMsgChains :many -SELECT mc.* -FROM msgchains mc -LEFT JOIN subtasks s ON mc.subtask_id = s.id -WHERE (mc.task_id = $1 OR s.task_id = $1) AND mc.type = 'primary_agent' -ORDER BY mc.created_at DESC; +Do not edit an already released migration. Add a new migration. --- Update conversation usage tracking with duration --- name: UpdateMsgChainUsage :one -UPDATE msgchains -SET - usage_in = usage_in + $1, - usage_out = usage_out + $2, - usage_cache_in = usage_cache_in + $3, - usage_cache_out = usage_cache_out + $4, - usage_cost_in = usage_cost_in + $5, - usage_cost_out = usage_cost_out + $6, - duration_seconds = duration_seconds + $7 -WHERE id = $8 -RETURNING *; - -// Get usage statistics for a specific flow --- name: GetFlowUsageStats :one -SELECT - COALESCE(SUM(mc.usage_in), 0) AS total_usage_in, - COALESCE(SUM(mc.usage_out), 0) AS total_usage_out, - COALESCE(SUM(mc.usage_cache_in), 0) AS total_usage_cache_in, - COALESCE(SUM(mc.usage_cache_out), 0) AS total_usage_cache_out, - COALESCE(SUM(mc.usage_cost_in), 0.0) AS total_usage_cost_in, - COALESCE(SUM(mc.usage_cost_out), 0.0) AS total_usage_cost_out -FROM msgchains mc -LEFT JOIN subtasks s ON mc.subtask_id = s.id -LEFT JOIN tasks t ON s.task_id = t.id OR mc.task_id = t.id -INNER JOIN flows f ON (mc.flow_id = f.id OR t.flow_id = f.id) -WHERE (mc.flow_id = $1 OR t.flow_id = $1) AND f.deleted_at IS NULL; -``` - -#### Container Management with Constraints - -```sql --- Upsert container with conflict resolution --- name: CreateContainer :one -INSERT INTO containers ( - type, name, image, status, flow_id, local_id, local_dir -) VALUES ( - $1, $2, $3, $4, $5, $6, $7 -) -ON CONFLICT ON CONSTRAINT containers_local_id_unique -DO UPDATE SET - type = EXCLUDED.type, - name = EXCLUDED.name, - image = EXCLUDED.image, - status = EXCLUDED.status, - flow_id = EXCLUDED.flow_id, - local_dir = EXCLUDED.local_dir -RETURNING *; -``` - -#### Role-Based Access Control - -```sql --- Complex role aggregation --- name: GetUser :one -SELECT - u.*, - r.name AS role_name, - ( - SELECT ARRAY_AGG(p.name) - FROM privileges p - WHERE p.role_id = r.id - ) AS privileges -FROM users u -INNER JOIN roles r ON u.role_id = r.id -WHERE u.id = $1; -``` - -## Code Generation with sqlc +## sqlc Query Layer ### Configuration -The package uses sqlc for code generation with the following configuration (`sqlc/sqlc.yml`): +`backend/sqlc/sqlc.yml` reads queries from `models/*.sql` and schema from `../migrations/sql/*.sql`, generating package `database` into `../pkg/database` with `emit_interface` and `emit_json_tags`. Notable overrides: -```yaml -version: "2" -sql: - - engine: "postgresql" - queries: ["models/*.sql"] - schema: ["../migrations/sql/*.sql"] - gen: - go: - package: "database" - out: "../pkg/database" - sql_package: "database/sql" - emit_interface: true - emit_json_tags: true - database: - uri: ${DATABASE_URL} -``` +- `pg_catalog.numeric` → `float64`; +- nullable `vector` / `pg_catalog.vector` → `string` (queries cast vector literals explicitly). -### Generation Command +Generated files start with `Code generated by sqlc. DO NOT EDIT.` Edit SQL or migrations and regenerate instead. -Code generation is performed using Docker to ensure consistency: +### Query files + +| Query file | Product area | Named queries | +|---|---|---| +| `flows.sql` | Flow CRUD, soft deletion, provider/model updates, flow statistics | `GetFlows`, `GetUserFlows`, `GetFlow`, `GetUserFlow`, `CreateFlow`, `UpdateFlow*`, `DeleteFlow`, `GetFlowStats`, `GetUserTotalFlowsStats`, `GetFlowsStatsByDayLast{Week,Month,3Months}` | +| `tasks.sql` | Task lifecycle and hierarchy-scoped retrieval | `GetFlowTasks`, `GetUserFlowTasks`, `GetFlowTask`, `GetUserFlowTask`, `GetTask`, `CreateTask`, `UpdateTaskStatus/Result/FinishedResult/FailedResult` | +| `subtasks.sql` | Subtask lifecycle, context, planned/completed filters | `GetFlowSubtasks`, `GetFlowTaskSubtasks`, `GetUserFlow*`, `GetTaskSubtasks`, `GetTaskPlannedSubtasks`, `GetTaskCompletedSubtasks`, `GetSubtask`, `GetFlowSubtask`, `CreateSubtask`, `UpdateSubtask*`, `DeleteSubtask(s)` | +| `containers.sql` | Flow container lookup and status | `GetContainers`, `GetUserContainers`, `GetRunningContainers`, `GetFlowContainers`, `GetFlowPrimaryContainer`, `GetUserFlowContainers`, `CreateContainer`, `UpdateContainer*` | +| `assistants.sql` | User/admin assistant access and settings | `GetFlowAssistants`, `GetUserFlowAssistants`, `GetFlowAssistant`, `GetUserFlowAssistant`, `GetAssistant`, `GetAssistantUseAgents`, `CreateAssistant`, `UpdateAssistant*`, `DeleteAssistant` | +| `msgchains.sql` | Conversation chains plus usage analytics | Chain CRUD/lookup by hierarchy/type; `UpdateMsgChainUsage`; aggregates by flow/task/subtask/provider/model/type/day/user | +| `toolcalls.sql` | Tool-call lifecycle and analytics | Hierarchy CRUD/status updates; aggregates by flow/task/subtask/function/day/user | +| `analytics.sql` | Period-based flow selection and hierarchy batches for execution analytics | `GetFlowsForPeriodLast{Week,Month,3Months}`, `GetTasksForFlow`, `GetSubtasksForTasks`, `GetMsgchainsForFlow`, `GetToolcallsForFlow`, `GetAssistantsCountForFlow` | +| `screenshots.sql` | Screenshot retrieval/creation | Flow/user/task/subtask getters + `CreateScreenshot` | +| `agentlogs.sql` | Agent delegation logs | Flow/user/task/subtask getters + `CreateAgentLog` | +| `assistantlogs.sql` | Assistant message/result/thinking logs | Create/update/delete plus flow/user getters | +| `msglogs.sql` | General message logs | Create/update plus flow/user/task/subtask getters | +| `searchlogs.sql` | Search operation logs | Flow/user/task/subtask getters + `CreateSearchLog` | +| `termlogs.sql` | Terminal logs | Container and hierarchy-scoped getters + `CreateTermLog` | +| `vecstorelogs.sql` | Vector-store audit logs | Flow/user/task/subtask getters + `CreateVectorStoreLog` | +| `users.sql` | User identity and administration | `GetUsers`, `GetUser`, `GetUserByHash`, `CreateUser`, `UpdateUser*`, `DeleteUser` | +| `roles.sql` | Roles and privileges | `GetRoles`, `GetRole`, `GetRoleByName` | +| `api_tokens.sql` | Token lifecycle | Admin and user-scoped create/update/soft-delete/list | +| `user_preferences.sql` | Preferences and favorite flows | CRUD/upsert + `AddFavoriteFlow` / `DeleteFavoriteFlow` | +| `providers.sql` | Provider configuration | Admin and user-scoped CRUD/soft-delete, lookup by type/name | +| `prompts.sql` | Prompt templates by `PROMPT_TYPE` | Admin and user-scoped CRUD, lookup/update by type | +| `flow_templates.sql` | Flow templates | User-owned CRUD | +| `resources.sql` | `user_resources` trees (**read-only** sqlc; create/update/delete go through GORM REST services) | Root/dir/recursive/all lookups for one user or all users; lookup by ID(s) | +| `knowledge.sql` | LangChain pgvector documents | Admin/user get/list/update/delete, cosine search, insert, `DeleteFlowMemoryDocuments` | + +The generated `Querier` interface in `backend/pkg/database/querier.go` currently exposes 251 methods matching the named SQL queries. Do not hand-edit it. + +### Regeneration + +From `backend/`, with the PentAGI PostgreSQL network available: ```bash -docker run --rm -v "$(pwd):/src" --network pentagi-network \ +docker run --rm \ + -v "$(pwd):/src" \ + -w /src \ + --network pentagi-network \ -e DATABASE_URL='postgres://postgres:postgres@pgvector:5432/pentagidb?sslmode=disable' \ - -w /src sqlc/sqlc:1.27.0 generate -f sqlc/sqlc.yml + sqlc/sqlc:1.27.0 generate -f sqlc/sqlc.yml ``` -This command: -1. Mounts the current directory into the container -2. Connects to the PentAGI database network -3. Uses the PostgreSQL database URL for schema introspection -4. Generates type-safe Go code from SQL queries +Use credentials matching the target database. Then review generated changes in `pkg/database`, especially `models.go`, `querier.go` and the affected `*.sql.go` file. -## Core Components - -### 1. Database Interface (`db.go`) - -Provides the foundational database transaction interface: - -```go -type DBTX interface { - ExecContext(context.Context, string, ...interface{}) (sql.Result, error) - PrepareContext(context.Context, string) (*sql.Stmt, error) - QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) - QueryRowContext(context.Context, string, ...interface{}) *sql.Row -} - -type Queries struct { - db DBTX -} -``` - -**Key Features:** -- Generic database transaction interface -- Support for both direct database connections and transactions -- Thread-safe query execution -- Context-aware operations for timeout handling - -### 2. Database Utilities (`database.go`) - -Contains utility functions and GORM integration: - -```go -// Null value converters -func StringToNullString(s string) sql.NullString -func NullStringToPtrString(s sql.NullString) *string -func Int64ToNullInt64(i *int64) sql.NullInt64 -func NullInt64ToInt64(i sql.NullInt64) *int64 -func TimeToNullTime(t time.Time) sql.NullTime - -// GORM configuration -func NewGorm(dsn, dbType string) (*gorm.DB, error) -``` - -**Key Features:** -- Null value handling for optional database fields -- GORM integration with custom logging -- Connection pooling configuration -- OpenTelemetry observability integration - -### 3. Query Interface (`querier.go`) - -Auto-generated interface containing all database operations: - -```go -type Querier interface { - // Flow operations - CreateFlow(ctx context.Context, arg CreateFlowParams) (Flow, error) - GetFlows(ctx context.Context) ([]Flow, error) - GetUserFlow(ctx context.Context, arg GetUserFlowParams) (Flow, error) - UpdateFlowStatus(ctx context.Context, arg UpdateFlowStatusParams) (Flow, error) - DeleteFlow(ctx context.Context, id int64) (Flow, error) - - // Task operations - CreateTask(ctx context.Context, arg CreateTaskParams) (Task, error) - GetFlowTasks(ctx context.Context, flowID int64) ([]Task, error) - UpdateTaskStatus(ctx context.Context, arg UpdateTaskStatusParams) (Task, error) - - // ... 150+ additional methods -} -``` - -**Features:** -- Complete CRUD operations for all entities -- User-scoped queries for multi-tenancy -- Efficient joins with foreign key relationships -- Soft delete support for critical entities - -### 4. Model Converters (`converter/converter.go`) - -Converts database models to GraphQL schema types: - -```go -func ConvertFlows(flows []database.Flow, containers []database.Container) []*model.Flow -func ConvertFlow(flow database.Flow, containers []database.Container) *model.Flow -func ConvertTasks(tasks []database.Task, subtasks []database.Subtask) []*model.Task -func ConvertAssistants(assistants []database.Assistant) []*model.Assistant -``` - -**Key Functions:** -- Transform database types to GraphQL models -- Handle relationship mapping (flows → tasks → subtasks) -- Null value processing for optional fields -- Aggregation of related entities - -## Data Models - -### Core Workflow Entities - -#### Flow -Top-level penetration testing workflow: -- `id`, `title`, `status` (active/completed/failed) -- `model`, `model_provider_name`, `model_provider_type` for AI configuration -- `language` for localization -- `tool_call_id_template` for customizing tool call ID format -- `functions` as JSON for AI behavior -- `trace_id` for observability -- `user_id` for multi-tenancy -- Soft delete with `deleted_at` - -**Note**: Prompts are no longer stored in flows. They are managed separately through the `prompts` table and loaded dynamically based on `PROMPT_TYPE`. - -#### Task -Major phases within a flow: -- `id`, `flow_id`, `title`, `status` (pending/running/done/failed) -- `input` for task parameters -- `result` JSON for task outputs -- Creation and update timestamps - -#### SubTask -Specific assignments for AI agents: -- `id`, `task_id`, `title`, `description` -- `status` (created/waiting/running/finished/failed) -- `result` and `context` JSON fields -- Agent type classification - -### Supporting Entities - -#### Container -Docker execution environments: -- `type` (primary/secondary), `name`, `image` -- `status` (starting/running/stopped) -- `local_id` for Docker integration -- `local_dir` for volume mapping - -#### Assistant -AI assistants for interactive flows: -- `title`, `status`, `model`, `model_provider_name`, `model_provider_type` -- `language` for localization -- `tool_call_id_template` for customizing tool call ID format -- `functions` configuration as JSON -- `use_agents` flag for delegation behavior -- `msgchain_id` for conversation tracking -- Flow association and soft delete - -**Note**: Prompts are managed separately through the `prompts` table, not stored in assistants. - -#### Message Chains (MsgChain) -LLM conversation management and usage tracking: -- `type` (primary_agent/assistant/generator/refiner/reporter/etc.) -- `model`, `model_provider` for tracking -- **Token usage tracking**: - - `usage_in`, `usage_out` - input/output tokens - - `usage_cache_in`, `usage_cache_out` - cached tokens (for prompt caching) - - `usage_cost_in`, `usage_cost_out` - cost tracking in currency units -- **Duration tracking**: - - `duration_seconds` - pre-calculated execution duration (DOUBLE PRECISION, NOT NULL, DEFAULT 0.0) - - Automatically incremented during updates using delta from backend - - Provides fast analytics without real-time calculations -- `chain` JSON for conversation history -- Multi-level association (flow/task/subtask) -- Creation and update timestamps for temporal analysis - -#### Provider -LLM provider configurations for multi-provider support: -- `type` - PROVIDER_TYPE enum (openai/anthropic/gemini/bedrock/deepseek/glm/kimi/qwen/minimax/ollama/custom) -- `name` - user-defined provider name -- `config` - JSON configuration for API keys and settings -- `user_id` - user ownership -- Soft delete with `deleted_at` -- Unique constraint on (name, user_id) for active providers - -#### Prompt -Centralized prompt template management: -- `type` - PROMPT_TYPE enum (primary_agent/assistant/pentester/coder/etc.) -- `prompt` - template content -- `user_id` - user ownership -- Creation and update timestamps - -### Logging Entities - -The package provides comprehensive logging for all system operations: - -- **AgentLog**: Inter-agent communication and delegation -- **AssistantLog**: Human-assistant interactions -- **MsgLog**: General message logging (thoughts/browser/terminal/file/search/advice/ask/input/done) -- **SearchLog**: External search operations (google/tavily/firecrawl/traversaal/browser/duckduckgo/perplexity/sploitus/searxng) -- **TermLog**: Terminal command execution (stdin/stdout/stderr) -- **ToolCall**: AI function calling with duration tracking - - `duration_seconds` - pre-calculated execution duration (DOUBLE PRECISION, NOT NULL, DEFAULT 0.0) - - Automatically incremented during status updates using delta from backend - - Only counts completed toolcalls (finished/failed) in analytics -- **VecStoreLog**: Vector database operations - -## LLM Usage Analytics - -The database package provides comprehensive analytics for tracking LLM usage, costs, and performance across all levels of the workflow hierarchy. This enables detailed monitoring of AI resource consumption and cost optimization. - -### Usage Tracking Fields - -The `msgchains` table tracks six key metrics for each conversation: - -| Field | Type | Description | -| ----------------- | ---------------- | ---------------------------------------- | -| `usage_in` | BIGINT | Input tokens consumed | -| `usage_out` | BIGINT | Output tokens generated | -| `usage_cache_in` | BIGINT | Cached input tokens (for prompt caching) | -| `usage_cache_out` | BIGINT | Cached output tokens | -| `usage_cost_in` | DOUBLE PRECISION | Input cost in currency units | -| `usage_cost_out` | DOUBLE PRECISION | Output cost in currency units | - -### Analytics Queries - -#### 1. Hierarchical Usage Statistics - -Get aggregated usage for specific entities: - -```go -// Get total usage for a flow -stats, err := db.GetFlowUsageStats(ctx, flowID) - -// Get total usage for a task -stats, err := db.GetTaskUsageStats(ctx, taskID) - -// Get total usage for a subtask -stats, err := db.GetSubtaskUsageStats(ctx, subtaskID) - -// Get usage for all flows (grouped by flow_id) -allStats, err := db.GetAllFlowsUsageStats(ctx) -``` - -Each query returns: -```go -type UsageStats struct { - TotalUsageIn int64 // Total input tokens - TotalUsageOut int64 // Total output tokens - TotalUsageCacheIn int64 // Total cached input tokens - TotalUsageCacheOut int64 // Total cached output tokens - TotalUsageCostIn float64 // Total input cost - TotalUsageCostOut float64 // Total output cost -} -``` - -#### 2. Provider and Model Analytics - -Track usage by LLM provider or specific model: +### Query conventions ```sql --- Get usage statistics grouped by provider --- name: GetUsageStatsByProvider :many -SELECT - mc.model_provider, - COALESCE(SUM(mc.usage_in), 0) AS total_usage_in, - COALESCE(SUM(mc.usage_out), 0) AS total_usage_out, - COALESCE(SUM(mc.usage_cache_in), 0) AS total_usage_cache_in, - COALESCE(SUM(mc.usage_cache_out), 0) AS total_usage_cache_out, - COALESCE(SUM(mc.usage_cost_in), 0.0) AS total_usage_cost_in, - COALESCE(SUM(mc.usage_cost_out), 0.0) AS total_usage_cost_out -FROM msgchains mc -LEFT JOIN subtasks s ON mc.subtask_id = s.id -LEFT JOIN tasks t ON s.task_id = t.id OR mc.task_id = t.id -INNER JOIN flows f ON (mc.flow_id = f.id OR t.flow_id = f.id) -WHERE f.deleted_at IS NULL -GROUP BY mc.model_provider -ORDER BY mc.model_provider; - --- Get usage statistics grouped by model --- name: GetUsageStatsByModel :many --- Similar structure, GROUP BY mc.model, mc.model_provider -``` - -Usage example: -```go -// Analyze costs per provider -providerStats, err := db.GetUsageStatsByProvider(ctx) -for _, stat := range providerStats { - totalCost := stat.TotalUsageCostIn + stat.TotalUsageCostOut - fmt.Printf("Provider: %s, Total Cost: $%.2f\n", - stat.ModelProvider, totalCost) -} - -// Compare model efficiency -modelStats, err := db.GetUsageStatsByModel(ctx) -``` - -#### 3. Agent Type Analytics - -Track usage by agent type (primary_agent, assistant, pentester, coder, etc.): - -```go -// Get usage by type across all flows -typeStats, err := db.GetUsageStatsByType(ctx) - -// Get usage by type for a specific flow -flowTypeStats, err := db.GetUsageStatsByTypeForFlow(ctx, flowID) -``` - -This helps identify which agent types consume the most resources. - -#### 4. Temporal Analytics - -Analyze usage trends over time: - -```go -// Last 7 days -weekStats, err := db.GetUsageStatsByDayLastWeek(ctx) - -// Last 30 days -monthStats, err := db.GetUsageStatsByDayLastMonth(ctx) - -// Last 90 days -quarterStats, err := db.GetUsageStatsByDayLast3Months(ctx) -``` - -Each query returns daily aggregates: -```go -type DailyUsageStats struct { - Date time.Time - TotalUsageIn int64 - TotalUsageOut int64 - TotalUsageCacheIn int64 - TotalUsageCacheOut int64 - TotalUsageCostIn float64 - TotalUsageCostOut float64 -} -``` - -### Usage Tracking Implementation - -When making LLM API calls, update usage metrics with duration: - -```go -// After receiving LLM response -startTime := time.Now() -// ... make LLM API call ... -durationDelta := time.Since(startTime).Seconds() - -_, err := db.UpdateMsgChainUsage(ctx, database.UpdateMsgChainUsageParams{ - UsageIn: response.Usage.PromptTokens, - UsageOut: response.Usage.CompletionTokens, - UsageCacheIn: response.Usage.PromptCacheTokens, - UsageCacheOut: response.Usage.CompletionCacheTokens, - UsageCostIn: calculateCost(response.Usage.PromptTokens, inputRate), - UsageCostOut: calculateCost(response.Usage.CompletionTokens, outputRate), - DurationSeconds: durationDelta, - ID: msgChainID, -}) -``` - -### Performance Considerations - -All analytics queries are optimized with appropriate indexes: - -- **Soft delete filtering**: `flows_deleted_at_idx` - partial index for active flows only -- **Time-based queries**: `msgchains_created_at_idx` - for temporal filtering -- **Provider analytics**: `msgchains_model_provider_idx` - for grouping by provider -- **Model analytics**: `msgchains_model_provider_composite_idx` - composite index -- **Type analytics**: `msgchains_type_flow_id_idx` - for flow-scoped type queries - -These indexes ensure fast query execution even with millions of message chain records. - -### Analytics-Specific Indexes - -Additional indexes optimized for analytics queries: - -**Assistants Analytics:** -- `assistants_deleted_at_idx` - Partial index for soft delete filtering (WHERE deleted_at IS NULL) -- `assistants_created_at_idx` - Temporal queries and sorting by creation date -- `assistants_flow_id_deleted_at_idx` - Flow-scoped queries with soft delete (GetFlowAssistants) -- `assistants_flow_id_created_at_idx` - Temporal analytics by flow (GetFlowsStatsByDay*) - -**Subtasks Analytics:** -- `subtasks_task_id_status_idx` - Task-scoped queries with status filtering -- `subtasks_status_created_at_idx` - Execution time analytics (excludes created/waiting) - -**Toolcalls Analytics:** -- `toolcalls_flow_id_status_idx` - Flow-scoped completed toolcalls counting -- `toolcalls_name_status_idx` - Function-based analytics with status filtering - -**MsgChains Analytics:** -- `msgchains_type_task_id_subtask_id_idx` - Hierarchical msgchain lookup by type -- `msgchains_type_created_at_idx` - Temporal analytics grouped by msgchain type - -**Tasks Analytics:** -- `tasks_flow_id_status_idx` - Flow-scoped task queries with status filtering - -### Cost Optimization Strategies - -Use analytics data to optimize LLM costs: - -1. **Identify expensive flows**: `GetAllFlowsUsageStats()` to find high-cost workflows -2. **Compare providers**: `GetUsageStatsByProvider()` to choose cost-effective providers -3. **Optimize agent types**: `GetUsageStatsByType()` to reduce token usage per agent -4. **Monitor trends**: Temporal queries to detect unusual spikes in usage -5. **Cache effectiveness**: Compare `usage_cache_in` vs `usage_in` to measure prompt caching benefits - -Example cost analysis: -```go -// Calculate cache savings -stats, _ := db.GetFlowUsageStats(ctx, flowID) -regularTokens := stats.TotalUsageIn + stats.TotalUsageOut -cachedTokens := stats.TotalUsageCacheIn + stats.TotalUsageCacheOut -cacheRatio := float64(cachedTokens) / float64(regularTokens+cachedTokens) -savings := stats.TotalUsageCostIn * (cacheRatio * 0.9) // Assuming 90% cache discount - -fmt.Printf("Cache effectiveness: %.1f%%\n", cacheRatio*100) -fmt.Printf("Estimated savings: $%.2f\n", savings) -``` - -## Flows and Structure Analytics - -The database package provides comprehensive analytics for tracking flow structure, execution metrics, and assistant usage across the workflow hierarchy. - -### Flow Structure Queries - -#### 1. Flow-Level Statistics - -Get structural metrics for specific flows: - -```go -// Get structure stats for a flow -stats, err := db.GetFlowStats(ctx, flowID) -// Returns: total_tasks_count, total_subtasks_count, total_assistants_count - -// Get total stats for all user's flows -allStats, err := db.GetUserTotalFlowsStats(ctx, userID) -// Returns: total_flows_count, total_tasks_count, total_subtasks_count, total_assistants_count -``` - -Each query returns: -```go -type FlowStats struct { - TotalTasksCount int64 - TotalSubtasksCount int64 - TotalAssistantsCount int64 -} - -type FlowsStats struct { - TotalFlowsCount int64 - TotalTasksCount int64 - TotalSubtasksCount int64 - TotalAssistantsCount int64 -} -``` - -#### 2. Temporal Flow Statistics - -Track flow creation and structure over time: - -```sql --- Get flows stats by day for the last week --- name: GetFlowsStatsByDayLastWeek :many -SELECT - DATE(f.created_at) AS date, - COALESCE(COUNT(DISTINCT f.id), 0)::bigint AS total_flows_count, - COALESCE(COUNT(DISTINCT t.id), 0)::bigint AS total_tasks_count, - COALESCE(COUNT(DISTINCT s.id), 0)::bigint AS total_subtasks_count, - COALESCE(COUNT(DISTINCT a.id), 0)::bigint AS total_assistants_count -FROM flows f -LEFT JOIN tasks t ON f.id = t.flow_id -LEFT JOIN subtasks s ON t.id = s.task_id -LEFT JOIN assistants a ON f.id = a.flow_id AND a.deleted_at IS NULL -WHERE f.created_at >= NOW() - INTERVAL '7 days' - AND f.deleted_at IS NULL AND f.user_id = $1 -GROUP BY DATE(f.created_at) -ORDER BY date DESC; -``` - -Usage example: -```go -// Analyze flow trends -weekStats, err := db.GetFlowsStatsByDayLastWeek(ctx, userID) -for _, stat := range weekStats { - fmt.Printf("Date: %s, Flows: %d, Tasks: %d, Subtasks: %d, Assistants: %d\n", - stat.Date, stat.TotalFlowsCount, stat.TotalTasksCount, - stat.TotalSubtasksCount, stat.TotalAssistantsCount) -} - -// Available for different periods -monthStats, err := db.GetFlowsStatsByDayLastMonth(ctx, userID) -quarterStats, err := db.GetFlowsStatsByDayLast3Months(ctx, userID) -``` - -### Flow Execution Time Analytics - -Track actual execution time and tool usage across the flow hierarchy using pre-calculated duration metrics. - -#### Analytics Queries (`analytics.sql`) - -```sql --- name: GetFlowsForPeriodLastWeek :many --- Get flow IDs created in the last week for analytics -SELECT id, title +-- name: GetUserFlow :one +SELECT * FROM flows -WHERE created_at >= NOW() - INTERVAL '7 days' - AND deleted_at IS NULL AND user_id = $1 -ORDER BY created_at DESC; - --- name: GetTasksForFlow :many --- Get all tasks for a flow -SELECT id, title, created_at, updated_at -FROM tasks -WHERE flow_id = $1 -ORDER BY id ASC; - --- name: GetSubtasksForTasks :many --- Get all subtasks for multiple tasks -SELECT id, task_id, title, status, created_at, updated_at -FROM subtasks -WHERE task_id = ANY(@task_ids::BIGINT[]) -ORDER BY id ASC; - --- name: GetMsgchainsForFlow :many --- Get all msgchains for a flow (including task and subtask level) -SELECT id, type, flow_id, task_id, subtask_id, duration_seconds, created_at, updated_at -FROM msgchains -WHERE flow_id = $1 -ORDER BY created_at ASC; - --- name: GetToolcallsForFlow :many --- Get all toolcalls for a flow -SELECT tc.id, tc.status, tc.flow_id, tc.task_id, tc.subtask_id, - tc.duration_seconds, tc.created_at, tc.updated_at -FROM toolcalls tc -LEFT JOIN tasks t ON tc.task_id = t.id -LEFT JOIN subtasks s ON tc.subtask_id = s.id -INNER JOIN flows f ON tc.flow_id = f.id -WHERE tc.flow_id = $1 AND f.deleted_at IS NULL - AND (tc.task_id IS NULL OR t.id IS NOT NULL) - AND (tc.subtask_id IS NULL OR s.id IS NOT NULL) -ORDER BY tc.created_at ASC; - --- name: GetAssistantsCountForFlow :one --- Get total count of assistants for a specific flow -SELECT COALESCE(COUNT(id), 0)::bigint AS total_assistants_count -FROM assistants -WHERE flow_id = $1 AND deleted_at IS NULL; +WHERE id = $1 + AND user_id = $2 + AND deleted_at IS NULL; ``` -Usage example: -```go -// Get execution statistics for flows in a period -flows, _ := db.GetFlowsForPeriodLastWeek(ctx, userID) +Established conventions: -for _, flow := range flows { - // Get hierarchical data - tasks, _ := db.GetTasksForFlow(ctx, flow.ID) - - // Collect task IDs - taskIDs := make([]int64, len(tasks)) - for i, task := range tasks { - taskIDs[i] = task.ID - } - - // Get all subtasks for these tasks - subtasks, _ := db.GetSubtasksForTasks(ctx, taskIDs) - - // Get msgchains and toolcalls - msgchains, _ := db.GetMsgchainsForFlow(ctx, flow.ID) - toolcalls, _ := db.GetToolcallsForFlow(ctx, flow.ID) - - // Get assistants count - assistantsCount, _ := db.GetAssistantsCountForFlow(ctx, flow.ID) - - // Build execution stats using converter functions - stats := converter.BuildFlowExecutionStats( - flow.ID, flow.Title, tasks, subtasks, msgchains, toolcalls, - int(assistantsCount), - ) - - fmt.Printf("Flow: %s, Duration: %.2fs, Toolcalls: %d, Assistants: %d\n", - stats.FlowTitle, stats.TotalDurationSeconds, - stats.TotalToolcallsCount, stats.TotalAssistantsCount) -} -``` +- `Create*`, `Get*`, `Update*`, `Delete*` for basic operations; +- `GetUser*` for user-owned paths; +- `GetFlow*`, `GetTask*`, `GetSubtask*` for hierarchy-scoped paths; +- admin/unscoped methods only where authorization is enforced by the caller; +- filter `deleted_at IS NULL` when querying soft-deletable entities; +- use foreign keys and `ON DELETE CASCADE` for owned child records; +- use `sqlc.arg(...)` for repeated or named parameters. -### Assistant Usage Tracking +Parameterized queries protect values from SQL injection. Dynamic identifiers cannot be parameterized and must be validated and quoted separately. -The database tracks assistant usage across flows: +### Transactions -```go -// Get assistant count for a flow -count, err := db.GetAssistantsCountForFlow(ctx, flowID) - -// Get all assistants for a flow -assistants, err := db.GetFlowAssistants(ctx, flowID) - -// User-scoped assistant access -userAssistants, err := db.GetUserFlowAssistants(ctx, database.GetUserFlowAssistantsParams{ - FlowID: flowID, - UserID: userID, -}) -``` - -Assistant metrics help understand: -- **Interactive flow usage**: Flows with high assistant counts indicate heavy user interaction -- **Delegation patterns**: Assistants with `use_agents` flag show delegation behavior -- **Resource allocation**: Track assistant-to-flow ratio for capacity planning - -## Usage Patterns - -### Basic Query Operations - -```go -// Initialize queries -db := database.New(sqlConnection) - -// Create a new flow -flow, err := db.CreateFlow(ctx, database.CreateFlowParams{ - Title: "Security Assessment", - Status: "active", - Model: "gpt-4", - ModelProviderName: "my-openai", - ModelProviderType: "openai", - Language: "en", - ToolCallIDTemplate: "call_{r:24:x}", - Functions: []byte(`{"tools": ["nmap", "sqlmap"]}`), - UserID: userID, -}) - -// Retrieve user's flows -flows, err := db.GetUserFlows(ctx, userID) - -// Update flow status -updatedFlow, err := db.UpdateFlowStatus(ctx, database.UpdateFlowStatusParams{ - Status: "completed", - ID: flowID, -}) -``` - -### Transaction Support +`database.Queries` accepts the `DBTX` interface, so the same generated methods work with `*sql.DB` and `*sql.Tx`: ```go tx, err := sqlDB.BeginTx(ctx, nil) @@ -911,577 +359,188 @@ if err != nil { } defer tx.Rollback() -queries := db.WithTx(tx) - -// Perform multiple operations atomically -task, err := queries.CreateTask(ctx, taskParams) -if err != nil { +qtx := queries.WithTx(tx) +if _, err := qtx.CreateTask(ctx, taskParams); err != nil { return err } - -subtask, err := queries.CreateSubtask(ctx, subtaskParams) -if err != nil { +if _, err := qtx.CreateSubtask(ctx, subtaskParams); err != nil { return err } return tx.Commit() ``` -### User-Scoped Operations +Keep transactions short and pass the caller's context through every query. -Most queries include user-scoped variants for multi-tenancy: +## Helpers (`database.go`) -```go -// Admin access - all flows -allFlows, err := db.GetFlows(ctx) +Hand-written helpers used across controllers and GraphQL resolvers: -// User access - only user's flows -userFlows, err := db.GetUserFlows(ctx, userID) +| Helper | Purpose | +|---|---| +| `NullStringToPtrString` / `PtrStringToNullString` / `StringToNullString` | Nullable string bridging | +| `Int64ToNullInt64` / `Uint64ToNullInt64` / `NullInt64ToInt64` | Nullable integer bridging | +| `TimeToNullTime` / `PtrTimeToNullTime` | Nullable timestamp bridging | +| `SanitizeUTF8` | Strip NUL bytes and replace invalid UTF-8 before storing untrusted tool output | +| `NewGorm` / `GormLogger` | Shared GORM init and optional SQL logging when `DEBUG=true` | -// User-scoped flow access with validation -flow, err := db.GetUserFlow(ctx, database.GetUserFlowParams{ - ID: flowID, - UserID: userID, -}) +## Converter Package + +`backend/pkg/database/converter` maps sqlc rows to GraphQL models and computes execution analytics: + +- entity converters for flows, containers/terminals, tasks, subtasks, assistants, screenshots, terminal/message/agent/search/vector/tool-call/assistant logs, prompts, preferences, API tokens, flow templates, user resources; +- provider/model converters between database provider configs, internal `pconfig` structures and GraphQL agent configs (including reasoning mode and call options); +- usage/toolcall/flow stats converters that adapt the typed sqlc analytics rows to GraphQL stats models; +- `BuildFlowExecutionStats` and related helpers in `analytics.go`, which combine tasks, subtasks, message-chain durations, tool-call counts and assistant activity, including overlap compensation for concurrent subtasks. + +Analytics calculation behavior belongs in `analytics.go` and its tests. Avoid copying entire generated result structs into documentation; sqlc changes them when query aliases change. + +## Knowledge Package + +`backend/pkg/database/knowledge` implements the GraphQL knowledge API on top of `knowledge.sql`, an optional LangChain `VectorStore` and an embedding provider: + +- admin reads have no `user_id` filter; user-scoped reads filter `cmetadata ->> 'user_id'`; +- writes always record the acting `userID` in metadata and publish scoped subscription events; +- create/update/search require a configured embedder/store; list/get/delete still work without embeddings; +- text sent to the embedding model is truncated to `maxEmbeddingBytes` (default 8192), while the full original text is stored in the database; +- in-memory filters refine SQL results by flow/task/subtask, doc type and related metadata. + +## GORM Integration + +`database.NewGorm(db, debug)` wraps the existing `*sql.DB`, installs `GormLogger` and enables GORM SQL logging only when `debug` is true. GORM v1 remains in use by server models and handlers; new type-safe database operations should prefer sqlc unless they need existing GORM model behavior. + +GORM `AutoMigrate` is not used. Goose migrations are the only supported schema management path. Because sqlc and GORM share one pool, do not call `gorm.Open` elsewhere with the same `DATABASE_URL` without accounting for another pool. + +Representative GORM model surfaces in `backend/pkg/server/models/` include users/roles/privileges/preferences, flows/tasks/subtasks/assistants, providers/prompts, API tokens, resources, knowledge request/response DTOs, settings, flow files and the various log entity models used by REST handlers. In particular, `user_resources` mutations are GORM-only; sqlc covers hierarchy listing and ID lookup. API token GraphQL conversion also uses the hand-written `APITokenWithSecret` helper in `backend/pkg/database/api_token_with_secret.go`. + +## Vector Operations + +PentAGI requires: + +- `vector` for embeddings and cosine-distance search; +- `pg_trgm` for GIN trigram indexes on message/log text. + +Tenant bootstrap guarantees that both extensions are reachable from every tenant. On a stock deployment they live in `public`; on Supabase they commonly live in `extensions`. + +`knowledge.sql` supports admin and user-scoped document retrieval, metadata-only and full document updates, insertion with precomputed embeddings, cosine-similarity search using `<=>`, ownership filtering through `cmetadata ->> 'user_id'`, and cleanup of flow memory documents. Embedding arguments are PostgreSQL vector literals (`[f1,f2,...]`); metadata arguments must be valid JSON text. + +The query layer currently filters by collection and metadata. No approximate HNSW/IVFFlat vector index is created by the migrations, so evaluate an appropriate pgvector index before assuming similarity searches will scale linearly to a large corpus. + +## Analytics + +Usage and execution analytics are derived from relational data rather than a separate warehouse. + +### LLM usage (`msgchains`) + +Stored fields: input/output tokens, cache input/output tokens, input/output cost, accumulated `duration_seconds`, model, provider and chain type. + +Aggregates in `msgchains.sql`: per flow/task/subtask, all flows, by provider/model/type, by type or model-agents for a flow, daily windows (week/month/3 months) and per-user totals. + +### Tool-call analytics (`toolcalls`) + +Aggregates in `toolcalls.sql`: per flow/task/subtask, all flows, by function (global and per flow), daily windows and per-user totals. Converters distinguish agent tools from ordinary tools where needed for GraphQL presentation. + +### Flow counts (`flows`) + +`GetFlowStats`, `GetUserTotalFlowsStats` and daily flow-count windows for week/month/3 months. + +### Flow execution analytics + +`analytics.sql` selects flows for a period and loads hierarchy batches. `converter.BuildFlowExecutionStats` turns that into execution duration models (task/subtask timing, generator/refiner contributions, finished tool-call counts, assistant message-chain time). + +## Indexing and Performance + +PostgreSQL automatically indexes primary keys and unique constraints. It does **not** automatically index foreign-key columns; PentAGI migrations create the needed foreign-key and query-pattern indexes explicitly. + +Current migrations include: + +- ownership/hierarchy indexes (`user_id`, `flow_id`, `task_id`, `subtask_id`); +- partial indexes for active soft-deletable rows; +- provider/model/type/time indexes for analytics; +- GIN indexes for JSON preferences; +- trigram GIN indexes for message, result and thinking text; +- path-prefix indexes for user resources. + +Large text B-tree indexes on task input/result and subtask description/result were deliberately removed by later migrations. Use full-text or trigram indexing for a concrete query pattern instead of restoring broad B-tree indexes. + +Before adding an index: capture the real query and expected cardinality, run `EXPLAIN (ANALYZE, BUFFERS)` on representative data, account for write amplification and index size, add it through a migration, and confirm the generated schema still passes sqlc. + +## Observability and Troubleshooting + +### Query logging + +Set `DEBUG=true` to enable the custom GORM logger. sqlc/libpq queries are not automatically printed by that logger; use PostgreSQL logging, tracing around the caller or a database proxy when those queries need inspection. + +Never log `DATABASE_URL`, provider configs, token values or arbitrary SQL arguments containing credentials. + +### Tenant schema mismatch + +```text +search_path resolved to schema "public", expected "acme" ``` -## Integration with PentAGI +The connection or pooler ignored the tenant search path. Do not bypass the check. Use a direct connection, configure PgBouncer as documented in [config.md](config.md#multi-tenant-postgresql-access-through-pgbouncer), or try `DATABASE_SEARCH_PATH_VIA_OPTIONS=true` for a compatible Supavisor version. -### GraphQL API Integration +### Extension schema mismatch -The database package integrates with PentAGI's GraphQL API through the converter package: - -```go -// In GraphQL resolvers -func (r *queryResolver) Flows(ctx context.Context) ([]*model.Flow, error) { - userID := auth.GetUserID(ctx) - - // Fetch from database - flows, err := r.DB.GetUserFlows(ctx, userID) - if err != nil { - return nil, err - } - - containers, err := r.DB.GetUserContainers(ctx, userID) - if err != nil { - return nil, err - } - - // Convert to GraphQL models - return converter.ConvertFlows(flows, containers), nil -} +```text +extension "vector" is installed in schema "extensions", +but multi-tenant mode requires it in "" ``` -### AI Agent Integration +Set `DATABASE_EXTENSIONS_SCHEMA` to the existing shared extension schema. Do not move provider-managed extensions unless the database operator explicitly requires it. The effective default is `public`. -The package supports AI agent operations through specialized queries: +### No migrations run in a new tenant -```go -// Log agent interactions -agentLog, err := db.CreateAgentLog(ctx, database.CreateAgentLogParams{ - Initiator: "pentester", - Executor: "researcher", - Task: "Analyze target application", - Result: resultJSON, - FlowID: flowID, - TaskID: sql.NullInt64{Int64: taskID, Valid: true}, -}) +Every tenant must have its own `.goose_db_version`. If a newly created tenant reports the public migration version but has no application tables, verify that the running binary includes schema-qualified goose table handling from `cmd/pentagi/main.go`. -// Track tool calls with duration updates -toolCall, err := db.CreateToolcall(ctx, database.CreateToolcallParams{ - CallID: callID, - Status: "received", - Name: "nmap_scan", - Args: argsJSON, - FlowID: flowID, - TaskID: sql.NullInt64{Int64: taskID, Valid: true}, - SubtaskID: sql.NullInt64{Int64: subtaskID, Valid: true}, -}) +### Constraint and scan errors -// Update status with duration delta -startTime := time.Now() -// ... execute toolcall ... -durationDelta := time.Since(startTime).Seconds() +- Foreign-key errors usually mean the parent flow/task/subtask was not created or belongs to a different scoped path. +- `sql.ErrNoRows` is expected for missing user-scoped data and should normally be translated to a not-found/access-denied result by the service layer. +- Nullable columns use `sql.Null*` values in generated models; use helpers in `pkg/database/database.go` where they make call sites clearer. +- `SanitizeUTF8` must be applied to untrusted tool output before insert into PostgreSQL text fields. -_, err = db.UpdateToolcallFinishedResult(ctx, database.UpdateToolcallFinishedResultParams{ - Result: resultJSON, - DurationSeconds: durationDelta, - ID: toolCall.ID, -}) -``` +## Development Checklist -### Vector Database Operations +### Adding or changing a query -For AI memory and semantic search: - -```go -// Log vector operations -vecLog, err := db.CreateVectorStoreLog(ctx, database.CreateVectorStoreLogParams{ - Initiator: "memorist", - Executor: "vector_db", - Filter: "vulnerability_data", - Query: "SQL injection techniques", - Action: "search", - Result: resultsJSON, - FlowID: flowID, -}) -``` - -## Best Practices - -### Error Handling - -Always handle database errors appropriately: - -```go -flow, err := db.GetUserFlow(ctx, params) -if err != nil { - if errors.Is(err, sql.ErrNoRows) { - return nil, fmt.Errorf("flow not found") - } - return nil, fmt.Errorf("database error: %w", err) -} -``` - -### Context Usage - -Use context for timeout and cancellation: - -```go -ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) -defer cancel() - -flows, err := db.GetFlows(ctx) -``` - -### Null Value Handling - -Use provided utilities for null values: - -```go -// Converting optional strings -description := database.StringToNullString(optionalDesc) - -// Converting back to pointers -descPtr := database.NullStringToPtrString(task.Description) -``` - -## Security Considerations - -### Multi-tenancy - -All user-facing operations use user-scoped queries to prevent unauthorized access: - -- `GetUserFlows()` instead of `GetFlows()` -- `GetUserFlowTasks()` instead of `GetFlowTasks()` -- User ID validation in all operations - -### Soft Deletes - -Critical entities use soft deletes to maintain audit trails: - -```sql --- Flows and assistants are soft deleted -UPDATE flows SET deleted_at = CURRENT_TIMESTAMP WHERE id = $1 - --- Most queries automatically filter soft-deleted records -WHERE f.deleted_at IS NULL -``` - -### SQL Injection Prevention - -sqlc generates parameterized queries that prevent SQL injection: - -```sql --- Safe parameterized query -SELECT * FROM flows WHERE user_id = $1 AND id = $2 -``` - -## Performance Considerations - -### Query Optimization - -The database package is designed with performance in mind: - -**Indexed Queries**: All foreign key relationships and frequently queried fields are properly indexed: -```sql --- Primary keys and foreign keys are automatically indexed --- Common query patterns use indexes for filtering and grouping - --- Flow indexes -CREATE INDEX flows_status_idx ON flows(status); -CREATE INDEX flows_title_idx ON flows(title); -CREATE INDEX flows_language_idx ON flows(language); -CREATE INDEX flows_model_provider_name_idx ON flows(model_provider_name); -CREATE INDEX flows_model_provider_type_idx ON flows(model_provider_type); -CREATE INDEX flows_user_id_idx ON flows(user_id); -CREATE INDEX flows_trace_id_idx ON flows(trace_id); -CREATE INDEX flows_deleted_at_idx ON flows(deleted_at) WHERE deleted_at IS NULL; - --- Task indexes -CREATE INDEX tasks_status_idx ON tasks(status); -CREATE INDEX tasks_title_idx ON tasks(title); -CREATE INDEX tasks_flow_id_idx ON tasks(flow_id); - --- Subtask indexes -CREATE INDEX subtasks_status_idx ON subtasks(status); -CREATE INDEX subtasks_title_idx ON subtasks(title); -CREATE INDEX subtasks_task_id_idx ON subtasks(task_id); - --- MsgChain indexes for analytics and duration tracking -CREATE INDEX msgchains_type_idx ON msgchains(type); -CREATE INDEX msgchains_flow_id_idx ON msgchains(flow_id); -CREATE INDEX msgchains_task_id_idx ON msgchains(task_id); -CREATE INDEX msgchains_subtask_id_idx ON msgchains(subtask_id); -CREATE INDEX msgchains_created_at_idx ON msgchains(created_at); -CREATE INDEX msgchains_model_provider_idx ON msgchains(model_provider); -CREATE INDEX msgchains_model_idx ON msgchains(model); -CREATE INDEX msgchains_model_provider_composite_idx ON msgchains(model, model_provider); -CREATE INDEX msgchains_created_at_flow_id_idx ON msgchains(created_at, flow_id); -CREATE INDEX msgchains_type_flow_id_idx ON msgchains(type, flow_id); - --- Toolcalls indexes for analytics and duration tracking -CREATE INDEX toolcalls_flow_id_idx ON toolcalls(flow_id); -CREATE INDEX toolcalls_task_id_idx ON toolcalls(task_id); -CREATE INDEX toolcalls_subtask_id_idx ON toolcalls(subtask_id); -CREATE INDEX toolcalls_status_idx ON toolcalls(status); -CREATE INDEX toolcalls_name_idx ON toolcalls(name); -CREATE INDEX toolcalls_created_at_idx ON toolcalls(created_at); -CREATE INDEX toolcalls_call_id_idx ON toolcalls(call_id); - --- Assistants indexes for analytics -CREATE INDEX assistants_flow_id_idx ON assistants(flow_id); -CREATE INDEX assistants_deleted_at_idx ON assistants(deleted_at) WHERE deleted_at IS NULL; -CREATE INDEX assistants_created_at_idx ON assistants(created_at); -CREATE INDEX assistants_flow_id_deleted_at_idx ON assistants(flow_id, deleted_at) WHERE deleted_at IS NULL; -CREATE INDEX assistants_flow_id_created_at_idx ON assistants(flow_id, created_at) WHERE deleted_at IS NULL; - --- Additional analytics indexes -CREATE INDEX subtasks_task_id_status_idx ON subtasks(task_id, status); -CREATE INDEX subtasks_status_created_at_idx ON subtasks(status, created_at); -CREATE INDEX toolcalls_flow_id_status_idx ON toolcalls(flow_id, status); -CREATE INDEX toolcalls_name_status_idx ON toolcalls(name, status); -CREATE INDEX msgchains_type_task_id_subtask_id_idx ON msgchains(type, task_id, subtask_id); -CREATE INDEX msgchains_type_created_at_idx ON msgchains(type, created_at); -CREATE INDEX tasks_flow_id_status_idx ON tasks(flow_id, status); - --- Provider indexes -CREATE INDEX providers_user_id_idx ON providers(user_id); -CREATE INDEX providers_type_idx ON providers(type); -CREATE INDEX providers_name_user_id_idx ON providers(name, user_id); -CREATE UNIQUE INDEX providers_name_user_id_unique ON providers(name, user_id) WHERE deleted_at IS NULL; -``` - -**Note**: Some indexes on large text fields (tasks.input, tasks.result, subtasks.description, subtasks.result) have been removed to improve write performance. These fields should use full-text search when needed. - -**Efficient Joins**: User-scoped queries use INNER JOINs to leverage PostgreSQL query planner: -```sql --- Efficient user-scoped access with proper join order -SELECT t.* -FROM tasks t -INNER JOIN flows f ON t.flow_id = f.id -- Fast foreign key join -WHERE f.user_id = $1 AND f.deleted_at IS NULL; -``` - -**Batch Operations**: Use transaction batching for bulk operations: -```go -tx, err := db.BeginTx(ctx, nil) -defer tx.Rollback() - -queries := database.New(tx) -for _, item := range items { - if _, err := queries.CreateSubtask(ctx, item); err != nil { - return err - } -} -return tx.Commit() -``` - -### Connection Pooling - -PentAGI opens two independent connection pools to the same Postgres instance: - -| Pool | Env var | Default | Used by | -|---|---|---|---| -| Shared `sql.DB` | `DATABASE_MAX_OPEN_CONNS` | `25` | sqlc `Queries` and GORM — both clients are backed by the **same** `*sql.DB` created in `main.go` | -| Shared `pgxpool` | `DATABASE_VECTOR_MAX_CONNS` | `10` | All `pgvector.Store` instances: every flow/assistant tool executor + knowledge API | - -Additional knob: `DATABASE_MAX_IDLE_CONNS` (default `5`) — idle connections kept open between requests. - -`NewGorm` accepts the already-configured `*sql.DB` so GORM never opens its own pool: - -```go -// main.go — one pool, used by both sqlc and GORM -db.SetMaxOpenConns(cfg.DBMaxOpenConns) -db.SetMaxIdleConns(cfg.DBMaxIdleConns) -db.SetConnMaxLifetime(time.Hour) - -orm, err := database.NewGorm(db) - -// pkg/database/database.go -func NewGorm(db *sql.DB) (*gorm.DB, error) { - orm, err := gorm.Open("postgres", db) - ... -} -``` - -The shared `pgxpool` is created in `main.go` and stored in `cfg.PgxPool`. All pgvector stores -(tool executors via `SetEmbedder`, knowledge API in `router.go`) use `pgvector.WithConn(cfg.PgxPool)` -instead of opening individual `pgx.Connect` calls per executor. - -**Connection budget** for the stock `vxcontrol/pgvector` image (`max_connections = 100`, -`superuser_reserved_connections = 3`): - -``` -Available for client connections = 97 - pentagi sql.DB (DATABASE_MAX_OPEN_CONNS) = 25 - pentagi pgxpool (DATABASE_VECTOR_MAX_CONNS) = 10 - pgexporter = 3 - autovacuum workers = 3 - ───────────────────────────────────────────── - Total consumed = 41 - Free buffer = 56 (≈ 58%) -``` - -Defaults are sized for 10 parallel flows with concurrent API requests. To inspect the live -budget on a running deployment: +1. Edit the appropriate file in `backend/sqlc/models`. +2. Keep user/admin scoping explicit in the query name and SQL. +3. Regenerate sqlc. +4. Review generated diffs; never hand-edit them. +5. Add tests at the service, converter or database boundary appropriate to the behavior. +6. Run: ```bash -# Postgres limits -docker exec pgvector sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -c \ - "SELECT name, setting FROM pg_settings - WHERE name IN ('"'"'max_connections'"'"', '"'"'superuser_reserved_connections'"'"');"' - -# Current usage vs. available -docker exec pgvector sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -c \ - "SELECT max_conn, used, max_conn - used AS available - FROM (SELECT current_setting('"'"'max_connections'"'"')::int AS max_conn, - count(*) AS used FROM pg_stat_activity) t;"' - -# Breakdown by client -docker exec pgvector sh -c 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -c \ - "SELECT application_name, client_addr, state, count(*) - FROM pg_stat_activity - WHERE pid <> pg_backend_pid() - GROUP BY 1, 2, 3 ORDER BY count DESC;"' +cd backend +go test ./pkg/database/... ./pkg/server/... ./pkg/graph/... +go vet ./pkg/database/... ``` -To raise the Postgres limit, add a `command` override in `docker-compose.yml`: +### Adding a table or field -```yaml -pgvector: - image: vxcontrol/pgvector:latest - command: postgres -c max_connections=200 -``` +1. Add a new goose migration. +2. Add explicit indexes for the actual foreign-key/query patterns. +3. Add or update sqlc queries. +4. Regenerate sqlc. +5. Update GraphQL/GORM converters only if the field crosses those boundaries. +6. Test fresh install, upgrade and tenant-schema startup. -### Vector Operations +### Review points -For pgvector operations, consider: -- **Batch embedding inserts** for better performance -- **Appropriate vector dimensions** (typically 512-1536) -- **Index configuration** for similarity searches +- Does every user-facing path enforce ownership? +- Is an unscoped/admin query intentionally authorized by its caller? +- Are soft-deleted rows filtered where expected? +- Does the migration work in a non-`public` tenant schema? +- Are shared extensions referenced through the configured search path? +- Does the change fit within the two-pool connection budget? +- Are generated files free of manual edits? -## Debugging and Troubleshooting +## Related Documentation -### Query Logging - -Enable query logging for debugging: -```go -// GORM logger captures all SQL operations -db.SetLogger(&GormLogger{}) -db.LogMode(true) -``` - -**Log Output Example**: -``` -INFO[0000] SELECT * FROM flows WHERE user_id = '1' AND deleted_at IS NULL component=pentagi-gorm duration=2.5ms rows_returned=3 -``` - -### Common Issues and Solutions - -#### 1. Foreign Key Constraint Violations - -**Error**: `pq: insert or update on table "tasks" violates foreign key constraint` - -**Solution**: Ensure parent entities exist before creating child entities: -```go -// Verify flow exists and user has access -flow, err := db.GetUserFlow(ctx, database.GetUserFlowParams{ - ID: flowID, - UserID: userID, -}) -if err != nil { - return fmt.Errorf("invalid flow: %w", err) -} - -// Now safe to create task -task, err := db.CreateTask(ctx, taskParams) -``` - -#### 2. Soft Delete Issues - -**Error**: Records not appearing in queries after "deletion" - -**Solution**: Check soft delete filters in custom queries: -```sql --- Always include soft delete filter -WHERE f.deleted_at IS NULL -``` - -#### 3. Null Value Handling - -**Error**: `sql: Scan error on column index 2: unsupported Scan` - -**Solution**: Use proper null value converters: -```go -// When creating -description := database.StringToNullString(optionalDesc) - -// When reading -descPtr := database.NullStringToPtrString(row.Description) -``` - -### Query Performance Analysis - -Use PostgreSQL's EXPLAIN for performance analysis: -```sql --- Analyze query performance -EXPLAIN ANALYZE SELECT f.*, COUNT(t.id) as task_count -FROM flows f -LEFT JOIN tasks t ON f.id = t.flow_id -WHERE f.user_id = $1 AND f.deleted_at IS NULL -GROUP BY f.id; -``` - -## Extending the Database Package - -### Adding New Entities - -1. **Create migration**: Add schema in `backend/migrations/sql/` -2. **Create SQL queries**: Add `.sql` file in `backend/sqlc/models/` -3. **Regenerate code**: Run sqlc generation command -4. **Add converters**: Update `converter/converter.go` for GraphQL integration - -**Example New Entity**: -```sql --- backend/sqlc/models/vulnerabilities.sql - --- name: CreateVulnerability :one -INSERT INTO vulnerabilities ( - title, severity, description, flow_id -) VALUES ( - $1, $2, $3, $4 -) RETURNING *; - --- name: GetFlowVulnerabilities :many -SELECT v.* -FROM vulnerabilities v -INNER JOIN flows f ON v.flow_id = f.id -WHERE v.flow_id = $1 AND f.deleted_at IS NULL -ORDER BY v.severity DESC, v.created_at DESC; -``` - -### Custom Query Patterns - -Follow established patterns for consistency: - -```sql --- Pattern: User-scoped access --- name: GetUser[Entity] :one/:many -SELECT [entity].* -FROM [entity] [alias] -INNER JOIN flows f ON [alias].flow_id = f.id -INNER JOIN users u ON f.user_id = u.id -WHERE [conditions] AND f.user_id = $user_id AND f.deleted_at IS NULL; - --- Pattern: Hierarchical retrieval --- name: Get[Parent][Children] :many -SELECT [child].* -FROM [child] [child_alias] -INNER JOIN [parent] [parent_alias] ON [child_alias].[parent_id] = [parent_alias].id -WHERE [parent_alias].id = $1 AND [filters]; -``` - -### Integration Testing - -Test database operations with real PostgreSQL: -```go -func TestCreateFlow(t *testing.T) { - // Setup test database - db := setupTestDB(t) - defer cleanupTestDB(t, db) - - queries := database.New(db) - - // Test operation - flow, err := queries.CreateFlow(ctx, database.CreateFlowParams{ - Title: "Test Flow", - Status: "active", - ModelProvider: "openai", - UserID: 1, - }) - - assert.NoError(t, err) - assert.Equal(t, "Test Flow", flow.Title) -} -``` - -## Security Guidelines - -### Input Validation - -Always validate inputs before database operations: -```go -func validateFlowInput(params CreateFlowParams) error { - if len(params.Title) > 255 { - return fmt.Errorf("title too long") - } - if !isValidStatus(params.Status) { - return fmt.Errorf("invalid status") - } - return nil -} -``` - -### Access Control - -Implement consistent access control patterns: -```go -// Always verify user ownership -flow, err := db.GetUserFlow(ctx, database.GetUserFlowParams{ - ID: flowID, - UserID: currentUserID, -}) -if err != nil { - return fmt.Errorf("access denied or flow not found") -} -``` - -### Audit Logging - -Use logging entities for security audit trails: -```go -// Log sensitive operations -_, err = db.CreateAgentLog(ctx, database.CreateAgentLogParams{ - Initiator: "system", - Executor: "user_action", - Task: "flow_deletion", - Result: []byte(fmt.Sprintf(`{"flow_id": %d, "user_id": %d}`, flowID, userID)), - FlowID: flowID, -}) -``` - -## Conclusion - -The database package provides a robust, secure, and performant foundation for PentAGI's data layer. By leveraging sqlc for code generation, implementing consistent security patterns, and maintaining comprehensive audit trails, it ensures reliable operation of the autonomous penetration testing system. - -Key benefits: -- **Type Safety**: Compile-time verification of SQL queries -- **Performance**: Optimized queries with proper indexing -- **Security**: Multi-tenancy and soft delete support -- **Observability**: Comprehensive logging and tracing -- **Maintainability**: Consistent patterns and generated code - -For developers working with this package, follow the established patterns for security, performance, and maintainability to ensure smooth integration with the broader PentAGI ecosystem. - -This documentation provides a comprehensive overview of the database package's architecture, functionality, and integration within the PentAGI system. +- [config.md](config.md) — database environment variables, tenancy, PgBouncer and Supavisor configuration +- [README.md](../../README.md) — deployment and development commands +- `backend/sqlc/sqlc.yml` — sqlc generation settings +- `backend/migrations/sql` — authoritative schema history +- `backend/pkg/database` — generated API and database helpers diff --git a/backend/pkg/config/config.go b/backend/pkg/config/config.go index d96f27db..4e7acb3f 100644 --- a/backend/pkg/config/config.go +++ b/backend/pkg/config/config.go @@ -18,10 +18,9 @@ import ( type Config struct { // === Core System Configuration === - DatabaseURL string `env:"DATABASE_URL" envDefault:"postgres://pentagiuser:pentagipass@pgvector:5432/pentagidb?sslmode=disable"` - Debug bool `env:"DEBUG" envDefault:"false"` - DataDir string `env:"DATA_DIR" envDefault:"./data"` - AskUser bool `env:"ASK_USER" envDefault:"false"` + Debug bool `env:"DEBUG" envDefault:"false"` + DataDir string `env:"DATA_DIR" envDefault:"./data"` + AskUser bool `env:"ASK_USER" envDefault:"false"` // TenantID namespaces every externally-visible artifact this instance creates // (PostgreSQL schema, docker object names, Graphiti group ids, telemetry identity) @@ -269,11 +268,20 @@ type Config struct { // === Agent Planning Phase Configuration === AgentPlanningStepEnabled bool `env:"AGENT_PLANNING_STEP_ENABLED" envDefault:"false"` + // === Database Configuration === + DatabaseURL string `env:"DATABASE_URL" envDefault:"postgres://pentagiuser:pentagipass@pgvector:5432/pentagidb?sslmode=disable"` + // === Database Connection Pool Sizing === DBMaxOpenConns int `env:"DATABASE_MAX_OPEN_CONNS" envDefault:"25"` DBMaxIdleConns int `env:"DATABASE_MAX_IDLE_CONNS" envDefault:"5"` DBVectorMaxConns int `env:"DATABASE_VECTOR_MAX_CONNS" envDefault:"10"` + // DatabaseExtensionsSchema/DatabaseSearchPathViaOptions only matter with + // TenantID set; see backend/docs/config.md -> "Multi-Instance Deployment + // (TENANT_ID)" for what they do and when to override them. + DatabaseExtensionsSchema string `env:"DATABASE_EXTENSIONS_SCHEMA" envDefault:""` + DatabaseSearchPathViaOptions bool `env:"DATABASE_SEARCH_PATH_VIA_OPTIONS" envDefault:"false"` + // PgxPool is the shared pgxpool.Pool for all pgvector stores. Populated by // main after pool creation; NOT sourced from environment variables. PgxPool *pgxpool.Pool `env:"-"` diff --git a/backend/pkg/config/tenant.go b/backend/pkg/config/tenant.go index 3678a717..928eccaf 100644 --- a/backend/pkg/config/tenant.go +++ b/backend/pkg/config/tenant.go @@ -149,6 +149,16 @@ func (c *Config) SchemaName() string { return c.TenantID } +// ExtensionSchema returns the schema every tenant's search_path must include for +// shared extensions to resolve. See DATABASE_EXTENSIONS_SCHEMA in +// backend/docs/config.md for details; defaults to "public". +func (c *Config) ExtensionSchema() string { + if c == nil || c.DatabaseExtensionsSchema == "" { + return "public" + } + return c.DatabaseExtensionsSchema +} + // AuthSalt returns the effective salt for cookie and JWT key derivation. Mixing // the tenant in makes one instance's session cookies and API tokens // cryptographically invalid on another even when COOKIE_SIGNING_SALT is shared diff --git a/backend/cmd/pentagi/tenant.go b/backend/pkg/database/tenant.go similarity index 68% rename from backend/cmd/pentagi/tenant.go rename to backend/pkg/database/tenant.go index d6314bd2..53134529 100644 --- a/backend/cmd/pentagi/tenant.go +++ b/backend/pkg/database/tenant.go @@ -1,4 +1,4 @@ -package main +package database import ( "context" @@ -20,11 +20,7 @@ import ( // would then fail with an opaque "type vector does not exist". var requiredExtensions = []string{"vector", "pg_trgm"} -// sharedExtensionSchema is where extensions must live so that every tenant's -// search_path can reach them. -const sharedExtensionSchema = "public" - -// ensureTenantSchema prepares this instance's PostgreSQL namespace and rewrites +// EnsureTenantSchema prepares this instance's PostgreSQL namespace and rewrites // cfg.DatabaseURL so that every downstream consumer — sqlc, gorm, goose and the // pgxpool backing the langchaingo vector store — resolves unqualified // identifiers inside it. @@ -32,12 +28,13 @@ const sharedExtensionSchema = "public" // It is a strict no-op when TENANT_ID is empty: the DSN is left untouched and // everything keeps resolving through the default "public" search path exactly // as before. -func ensureTenantSchema(ctx context.Context, cfg *config.Config) error { +func EnsureTenantSchema(ctx context.Context, cfg *config.Config) error { if !cfg.HasTenant() { return nil } schema := cfg.SchemaName() + extSchema := cfg.ExtensionSchema() // Short-lived bootstrap connection on the ORIGINAL DSN. Opening it before the // search_path rewrite means CREATE EXTENSION resolves against the default @@ -54,7 +51,7 @@ func ensureTenantSchema(ctx context.Context, cfg *config.Config) error { // Serialize concurrent first boots so two instances cannot race on schema and // extension creation in the shared catalog. - if err := withAdvisoryLock(ctx, db, "pentagi-tenant-bootstrap", func(conn *sql.Conn) error { + if err := WithAdvisoryLock(ctx, db, "pentagi-tenant-bootstrap", func(conn *sql.Conn) error { // QuoteIdentifier is belt-and-braces: ValidateTenantID already restricts // the character set, but this keeps the statement safe if that ever relaxes. if _, err := conn.ExecContext(ctx, @@ -64,7 +61,7 @@ func ensureTenantSchema(ctx context.Context, cfg *config.Config) error { } for _, ext := range requiredExtensions { - if err := ensureSharedExtension(ctx, conn, ext); err != nil { + if err := ensureSharedExtension(ctx, conn, ext, extSchema); err != nil { return err } } @@ -75,22 +72,37 @@ func ensureTenantSchema(ctx context.Context, cfg *config.Config) error { } // Rewrite the DSN once; every consumer reads cfg.DatabaseURL afterwards. - // Both lib/pq and pgx forward search_path as a libpq runtime parameter, so a - // single URL change reaches every connection in every pool. - rewritten, err := withSearchPath(cfg.DatabaseURL, schema+","+sharedExtensionSchema) + return RewriteDatabaseURLForTenant(cfg) +} + +// RewriteDatabaseURLForTenant appends the tenant search_path to cfg.DatabaseURL. +// Unlike EnsureTenantSchema it does not touch the catalog — use it for tools that +// only need to read or write tenant data (e.g. the installer's password reset). +// It is a no-op when TENANT_ID is empty. +func RewriteDatabaseURLForTenant(cfg *config.Config) error { + if !cfg.HasTenant() { + return nil + } + + rewritten, err := withSearchPath( + cfg.DatabaseURL, + cfg.SchemaName()+","+cfg.ExtensionSchema(), + cfg.DatabaseSearchPathViaOptions, + ) if err != nil { return err } cfg.DatabaseURL = rewritten - return nil } -// ensureSharedExtension guarantees that ext exists and is reachable from every -// tenant's search_path. It checks before creating so that a database whose -// extensions were pre-installed by an administrator works without the -// application needing CREATE privileges. -func ensureSharedExtension(ctx context.Context, conn *sql.Conn, ext string) error { +// ensureSharedExtension guarantees that ext exists in sharedSchema and is +// therefore reachable from every tenant's search_path. It checks before +// creating so that a database whose extensions were pre-installed by an +// administrator (or by convention — see DATABASE_EXTENSIONS_SCHEMA in +// backend/docs/config.md) works without the application needing CREATE +// privileges. +func ensureSharedExtension(ctx context.Context, conn *sql.Conn, ext, sharedSchema string) error { schema, err := extensionSchema(ctx, conn, ext) switch { case err != nil: @@ -100,24 +112,25 @@ func ensureSharedExtension(ctx context.Context, conn *sql.Conn, ext string) erro // Not installed yet — create it explicitly in the shared schema. if _, err := conn.ExecContext(ctx, fmt.Sprintf( "CREATE EXTENSION IF NOT EXISTS %s SCHEMA %s", - pq.QuoteIdentifier(ext), pq.QuoteIdentifier(sharedExtensionSchema), + pq.QuoteIdentifier(ext), pq.QuoteIdentifier(sharedSchema), )); err != nil { return fmt.Errorf( "failed to create extension %q in schema %q (a privileged user must run "+ "CREATE EXTENSION %s SCHEMA %s once): %w", - ext, sharedExtensionSchema, ext, sharedExtensionSchema, err, + ext, sharedSchema, ext, sharedSchema, err, ) } return nil - case schema != sharedExtensionSchema: + case schema != sharedSchema: // Installed, but somewhere this tenant's search_path will not reach. Fail // with an actionable message rather than letting migrations die on a // confusing "type does not exist". return fmt.Errorf( "extension %q is installed in schema %q, but multi-tenant mode requires it in %q "+ - "so every tenant can reach it; run: ALTER EXTENSION %s SET SCHEMA %s", - ext, schema, sharedExtensionSchema, ext, sharedExtensionSchema, + "so every tenant can reach it; either run ALTER EXTENSION %s SET SCHEMA %s, "+ + "or set DATABASE_EXTENSIONS_SCHEMA=%s to match where it already lives", + ext, schema, sharedSchema, ext, sharedSchema, schema, ) default: @@ -145,10 +158,10 @@ func extensionSchema(ctx context.Context, conn *sql.Conn, ext string) (string, e } } -// verifySearchPath asserts that connections really do resolve into the expected +// VerifySearchPath asserts that connections really do resolve into the expected // schema. A typo in the DSN would otherwise route a tenant silently onto public, // where every tenant would share one dataset — a quiet, catastrophic failure. -func verifySearchPath(ctx context.Context, db *sql.DB, cfg *config.Config) error { +func VerifySearchPath(ctx context.Context, db *sql.DB, cfg *config.Config) error { if !cfg.HasTenant() { return nil } @@ -168,21 +181,21 @@ func verifySearchPath(ctx context.Context, db *sql.DB, cfg *config.Config) error return nil } -// runMigrations applies pending migrations while holding an advisory lock, so +// RunMigrations applies pending migrations while holding an advisory lock, so // that two instances booting simultaneously cannot execute the same migration // set concurrently. Without a tenant the lock is still taken, which also fixes // the pre-existing race between two single-instance deployments sharing a // database. -func runMigrations(ctx context.Context, db *sql.DB, cfg *config.Config, up func(*sql.DB) error) error { - return withAdvisoryLock(ctx, db, "pentagi-migrations-"+cfg.SchemaName(), func(*sql.Conn) error { +func RunMigrations(ctx context.Context, db *sql.DB, cfg *config.Config, up func(*sql.DB) error) error { + return WithAdvisoryLock(ctx, db, "pentagi-migrations-"+cfg.SchemaName(), func(*sql.Conn) error { return up(db) }) } -// withAdvisoryLock runs fn while holding a PostgreSQL session-level advisory +// WithAdvisoryLock runs fn while holding a PostgreSQL session-level advisory // lock derived from key. The lock is taken on a dedicated connection because // advisory locks are session-scoped and *sql.DB is a pool. -func withAdvisoryLock(ctx context.Context, db *sql.DB, key string, fn func(*sql.Conn) error) error { +func WithAdvisoryLock(ctx context.Context, db *sql.DB, key string, fn func(*sql.Conn) error) error { // crc32 into the signed 32-bit space keeps the key stable and collision-free // enough for the two distinct locks this application takes. lockID := int64(int32(crc32.ChecksumIEEE([]byte(key)))) @@ -204,17 +217,25 @@ func withAdvisoryLock(ctx context.Context, db *sql.DB, key string, fn func(*sql. return fn(conn) } -// withSearchPath returns dsn with the search_path runtime parameter set. It -// supports both URL-style DSNs (the shipped default) and libpq keyword strings. -func withSearchPath(dsn, searchPath string) (string, error) { +// withSearchPath returns dsn with the tenant's search_path applied as a +// PostgreSQL startup parameter (or, with viaOptions, wrapped as +// options=--search_path= for poolers that need it — see +// DATABASE_SEARCH_PATH_VIA_OPTIONS in backend/docs/config.md). Supports both +// URL-style DSNs and libpq keyword strings. +func withSearchPath(dsn, searchPath string, viaOptions bool) (string, error) { + key, value := "search_path", searchPath + if viaOptions { + key, value = "options", "--search_path="+searchPath + } + u, err := url.Parse(dsn) if err != nil || u.Scheme == "" { // Not a URL — fall back to libpq keyword/value syntax. - return fmt.Sprintf("%s search_path=%s", dsn, searchPath), nil + return fmt.Sprintf("%s %s=%s", dsn, key, value), nil } q := u.Query() - q.Set("search_path", searchPath) + q.Set(key, value) u.RawQuery = q.Encode() return u.String(), nil diff --git a/docker-compose.yml b/docker-compose.yml index 0fbeda6a..91623065 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -140,6 +140,8 @@ services: - OAUTH_GITHUB_CLIENT_ID=${OAUTH_GITHUB_CLIENT_ID:-} - OAUTH_GITHUB_CLIENT_SECRET=${OAUTH_GITHUB_CLIENT_SECRET:-} - DATABASE_URL=postgres://${PENTAGI_POSTGRES_USER:-postgres}:${PENTAGI_POSTGRES_PASSWORD:-postgres}@pgvector:5432/${PENTAGI_POSTGRES_DB:-pentagidb}?sslmode=disable + - DATABASE_EXTENSIONS_SCHEMA=${DATABASE_EXTENSIONS_SCHEMA:-} + - DATABASE_SEARCH_PATH_VIA_OPTIONS=${DATABASE_SEARCH_PATH_VIA_OPTIONS:-} - DATABASE_MAX_OPEN_CONNS=${DATABASE_MAX_OPEN_CONNS:-} - DATABASE_MAX_IDLE_CONNS=${DATABASE_MAX_IDLE_CONNS:-} - DATABASE_VECTOR_MAX_CONNS=${DATABASE_VECTOR_MAX_CONNS:-}