diff --git a/common/dbpool_crud/dbpool_crud.go b/common/dbpool_crud/dbpool_crud.go deleted file mode 100644 index ba3e6931..00000000 --- a/common/dbpool_crud/dbpool_crud.go +++ /dev/null @@ -1,162 +0,0 @@ -package dbpool_crud - -import ( - "context" - "fmt" - - "github.com/primadi/lokstra/common/logger" - "github.com/primadi/lokstra/core/service" - "github.com/primadi/lokstra/serviceapi" -) - -// DbPoolConfig represents database pool configuration -type DbPoolConfig struct { - Name string `json:"name"` - DSN string `json:"dsn"` - Schema string `json:"schema"` - RlsContext map[string]string `json:"rls_context,omitempty"` -} - -var dbpm = service.LazyLoad[serviceapi.DbPoolManager]("dbpool-manager") - -// AddDbPool adds a new database pool at runtime -// This will persist to sync_config database if using distributed sync -// Note: If pool already exists, it will be updated (upsert behavior) -func AddDbPool(config DbPoolConfig) error { - if config.Name == "" { - return fmt.Errorf("pool name is required") - } - if config.DSN == "" { - return fmt.Errorf("DSN is required") - } - if config.Schema == "" { - config.Schema = "public" // Default schema - } - - dpm := dbpm.Get() - if dpm == nil { - return fmt.Errorf("dbpool-manager service not found") - } - - // Set the pool configuration (upsert: works for both new and existing pools) - // This also auto-registers the pool as a lazy service - dpm.SetDbPoolManager(config.Name, config.DSN, config.Schema, config.RlsContext) - - // Validate configuration by attempting to get the pool - _, err := dpm.GetDbPoolManager(config.Name) - if err != nil { - // Rollback on validation failure - dpm.RemoveDbPoolManager(config.Name) - return fmt.Errorf("failed to create pool '%s': %w", config.Name, err) - } - - logger.LogInfo("✅ Added/Updated DB pool: %s (schema: %s)", config.Name, config.Schema) - return nil -} - -// UpdateDbPool updates an existing database pool configuration -// This is an alias for AddDbPool (both do upsert) -// Note: This will NOT close existing connections, new connections will use new config -func UpdateDbPool(config DbPoolConfig) error { - return AddDbPool(config) // Same implementation - upsert behavior -} - -// RemoveDbPool removes a database pool at runtime -// This will persist to sync_config database if using distributed sync -// The pool service will also be unregistered from the service registry -func RemoveDbPool(name string) error { - if name == "" { - return fmt.Errorf("pool name is required") - } - - dpm := dbpm.Get() - if dpm == nil { - return fmt.Errorf("dbpool-manager service not found") - } - - // Check if pool exists - _, _, _, err := dpm.GetDbPoolManagerInfo(name) - if err != nil { - return fmt.Errorf("pool '%s' not found: %w", name, err) - } - - // Remove from manager (also unregisters service automatically) - dpm.RemoveDbPoolManager(name) - - logger.LogInfo("✅ Removed DB pool: %s", name) - return nil -} - -// GetDbPoolInfo retrieves database pool configuration -func GetDbPoolInfo(name string) (*DbPoolConfig, error) { - if name == "" { - return nil, fmt.Errorf("pool name is required") - } - - dpm := dbpm.Get() - if dpm == nil { - return nil, fmt.Errorf("dbpool-manager service not found") - } - - dsn, schema, rlsContext, err := dpm.GetDbPoolManagerInfo(name) - if err != nil { - return nil, fmt.Errorf("pool '%s' not found: %w", name, err) - } - - return &DbPoolConfig{ - Name: name, - DSN: dsn, - Schema: schema, - RlsContext: rlsContext, - }, nil -} - -// ListDbPools returns all configured database pools -// Works with both regular and sync DbPoolManager -func ListDbPools() ([]string, error) { - dpm := dbpm.Get() - if dpm == nil { - return nil, fmt.Errorf("dbpool-manager service not found") - } - - // Use GetAllDbPoolManager from DbPoolManager interface - allPools := dpm.GetAllDbPoolManager() - if allPools == nil { - return []string{}, nil - } - - poolNames := make([]string, 0, len(allPools)) - for poolName := range allPools { - poolNames = append(poolNames, poolName) - } - - return poolNames, nil -} - -// GetDbPool returns the actual DbPool instance for direct use -func GetDbPool(name string) (serviceapi.DbPool, error) { - if name == "" { - return nil, fmt.Errorf("pool name is required") - } - - dpm := dbpm.Get() - if dpm == nil { - return nil, fmt.Errorf("dbpool-manager service not found") - } - - return dpm.GetDbPoolManager(name) -} - -// AcquireDbConn acquires a connection from a named pool -func AcquireDbConn(ctx context.Context, poolName string) (serviceapi.DbConn, error) { - if poolName == "" { - return nil, fmt.Errorf("pool name is required") - } - - dpm := dbpm.Get() - if dpm == nil { - return nil, fmt.Errorf("dbpool-manager service not found") - } - - return dpm.AcquireNamedConn(ctx, poolName) -} diff --git a/common/dbpool_crud/dbpool_crud_test.go b/common/dbpool_crud/dbpool_crud_test.go deleted file mode 100644 index c2c1a80c..00000000 --- a/common/dbpool_crud/dbpool_crud_test.go +++ /dev/null @@ -1,150 +0,0 @@ -package dbpool_crud_test - -import ( - "context" - "testing" - - "github.com/primadi/lokstra/common/dbpool_crud" - "github.com/primadi/lokstra/lokstra_init" -) - -// Example: Add a new database pool at runtime -func ExampleAddDbPool() { - // Initialize framework first - lokstra_init.UsePgxDbPoolManager(true) // Enable distributed sync - - // Add new pool - err := dbpool_crud.AddDbPool(dbpool_crud.DbPoolConfig{ - Name: "db-analytics", - DSN: "postgres://user:pass@localhost:5432/analytics", - Schema: "public", - }) - - if err != nil { - panic(err) - } - - // Pool is now available across all servers (if using distributed sync) - conn, _ := dbpool_crud.AcquireDbConn(context.Background(), "db-analytics") - defer conn.Release() - - // Use connection... -} - -// Example: Update existing pool configuration -func ExampleUpdateDbPool() { - err := dbpool_crud.UpdateDbPool(dbpool_crud.DbPoolConfig{ - Name: "db-main", - DSN: "postgres://user:pass@new-host:5432/main", - Schema: "public", - }) - - if err != nil { - panic(err) - } - - // New connections will use the updated configuration - // Existing connections remain unchanged until closed -} - -// Example: Remove a pool -func ExampleRemoveDbPool() { - err := dbpool_crud.RemoveDbPool("db-old-tenant") - if err != nil { - panic(err) - } - - // Pool removed from all servers (if using distributed sync) -} - -// Example: List all pools -func ExampleListDbPools() { - pools, err := dbpool_crud.ListDbPools() - if err != nil { - panic(err) - } - - for _, poolName := range pools { - info, _ := dbpool_crud.GetDbPoolInfo(poolName) - println("Pool:", info.Name, "Schema:", info.Schema) - } -} - -// Example: Get pool info -func ExampleGetDbPoolInfo() { - info, err := dbpool_crud.GetDbPoolInfo("db-main") - if err != nil { - panic(err) - } - - println("DSN:", info.DSN) - println("Schema:", info.Schema) -} - -// Example: Direct pool access -func ExampleGetDbPool() { - pool, err := dbpool_crud.GetDbPool("db-main") - if err != nil { - panic(err) - } - - conn, _ := pool.Acquire(context.Background()) - defer conn.Release() - - // Use connection... -} - -// Test CRUD operations -func TestDbPoolCRUD(t *testing.T) { - // Setup - lokstra_init.UsePgxDbPoolManager(false) // Use local sync for testing - - // Create - err := dbpool_crud.AddDbPool(dbpool_crud.DbPoolConfig{ - Name: "test-pool", - DSN: "postgres://localhost/test", - Schema: "test_schema", - }) - if err != nil { - t.Fatalf("Failed to add pool: %v", err) - } - - // Read - info, err := dbpool_crud.GetDbPoolInfo("test-pool") - if err != nil { - t.Fatalf("Failed to get pool info: %v", err) - } - if info.Name != "test-pool" { - t.Errorf("Expected name 'test-pool', got '%s'", info.Name) - } - if info.Schema != "test_schema" { - t.Errorf("Expected schema 'test_schema', got '%s'", info.Schema) - } - - // Update - err = dbpool_crud.UpdateDbPool(dbpool_crud.DbPoolConfig{ - Name: "test-pool", - DSN: "postgres://localhost/test2", - Schema: "test_schema2", - }) - if err != nil { - t.Fatalf("Failed to update pool: %v", err) - } - - info, _ = dbpool_crud.GetDbPoolInfo("test-pool") - if info.Schema != "test_schema2" { - t.Errorf("Expected schema 'test_schema2', got '%s'", info.Schema) - } - - // Delete - err = dbpool_crud.RemoveDbPool("test-pool") - if err != nil { - t.Fatalf("Failed to remove pool: %v", err) - } - - // Verify deletion - _, err = dbpool_crud.GetDbPoolInfo("test-pool") - if err == nil { - t.Error("Expected error when getting deleted pool") - } -} diff --git a/core/deploy/loader/builder.go b/core/deploy/loader/builder.go index e356b358..f7291e48 100644 --- a/core/deploy/loader/builder.go +++ b/core/deploy/loader/builder.go @@ -3,13 +3,11 @@ package loader import ( "fmt" "strings" - "time" "github.com/primadi/lokstra/common/logger" "github.com/primadi/lokstra/core/deploy" "github.com/primadi/lokstra/core/deploy/schema" "github.com/primadi/lokstra/core/router" - "github.com/primadi/lokstra/serviceapi" ) // Case-insensitive map lookup helpers @@ -978,115 +976,3 @@ func LoadConfig(configPaths ...string) (*schema.DeployConfig, error) { logger.LogDebug("✅ Config loaded successfully from: %v", configPaths) return config, nil } - -// LoadDbPoolDefsFromConfig auto-discovers and sets up named DB pools from config -// Requires dbpool-manager service to be already registered -func LoadDbPoolDefsFromConfig() error { - registry := deploy.Global() - config := registry.GetDeployConfig() - - // Check if dbpool-definitions section exists - if len(config.DbPoolDefinitions) == 0 { - // No dbpool-definitions section, skip - return nil - } - - dpm, ok := deploy.Global().GetServiceAny("dbpool-manager") - if !ok || dpm == nil { - return fmt.Errorf("dbpool-manager service not found in registry") - } - dbPoolManager, ok := dpm.(serviceapi.DbPoolManager) - if !ok { - return fmt.Errorf("dbpool-manager service does not implement serviceapi.DbPoolManager interface") - } - - // Setup each pool - for poolName, poolConfig := range config.DbPoolDefinitions { - // Extract DSN or build from components - dsn := poolConfig.DSN - - // Extract optional pool parameters with best practice defaults - minConns := poolConfig.MinConns - if minConns == 0 { - minConns = 2 // Best practice: 2 minimum connections - } - - maxConns := poolConfig.MaxConns - if maxConns == 0 { - maxConns = 10 // Best practice: 10 max connections - } - - maxIdleTime := 30 * time.Minute // Best practice default - if poolConfig.MaxIdleTime != "" { - if parsed, err := time.ParseDuration(poolConfig.MaxIdleTime); err == nil { - maxIdleTime = parsed - } - } - - maxLifetime := time.Hour // Best practice default - if poolConfig.MaxLifetime != "" { - if parsed, err := time.ParseDuration(poolConfig.MaxLifetime); err == nil { - maxLifetime = parsed - } - } - - // If no DSN, build from components - if dsn == "" { - host := poolConfig.Host - port := poolConfig.Port - if port == 0 { - port = 5432 // Default PostgreSQL port - } - database := poolConfig.Database - username := poolConfig.Username - password := poolConfig.Password - - if host == "" || database == "" { - return fmt.Errorf("dbpool-manager.%s: must provide either 'dsn' or 'host'+'database'", poolName) - } - - // Build DSN with best practice defaults - sslmode := poolConfig.SSLMode - if sslmode == "" { - sslmode = "disable" - } - - dsn = fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s&pool_min_conns=%d&pool_max_conns=%d&pool_max_conn_idle_time=%s&pool_max_conn_lifetime=%s", - username, password, host, port, database, sslmode, minConns, maxConns, maxIdleTime, maxLifetime) - } else { - // DSN provided - apply optional pool parameters if not already set - opts := "" - if !strings.Contains(dsn, "pool_min_conns=") { - opts += fmt.Sprintf("&pool_min_conns=%d", minConns) - } - if !strings.Contains(dsn, "pool_max_conns=") { - opts += fmt.Sprintf("&pool_max_conns=%d", maxConns) - } - if !strings.Contains(dsn, "pool_max_conn_idle_time=") { - opts += fmt.Sprintf("&pool_max_conn_idle_time=%s", maxIdleTime) - } - if !strings.Contains(dsn, "pool_max_conn_lifetime=") { - opts += fmt.Sprintf("&pool_max_conn_lifetime=%s", maxLifetime) - } - if strings.Contains(dsn, "?") { - dsn += opts - } else { - dsn += "?" + strings.TrimPrefix(opts, "&") - } - } - - // Extract schema (default: public) - schema := poolConfig.Schema - if schema == "" { - schema = "public" - } - - // Set DSN and Schema for poolName - // This also auto-registers the pool as a lazy service - dbPoolManager.SetDbPoolManager(poolName, dsn, schema, poolConfig.RlsContext) - - logger.LogDebug("✅ Registered DB pool: %s (schema: %s)", poolName, schema) - } - - return nil -} diff --git a/core/deploy/loader/loader.go b/core/deploy/loader/loader.go index 4c56f030..6f474552 100644 --- a/core/deploy/loader/loader.go +++ b/core/deploy/loader/loader.go @@ -205,7 +205,6 @@ func applyConfigOverrides(config *schema.DeployConfig) { func mergeConfigs(target, source *schema.DeployConfig) *schema.DeployConfig { result := &schema.DeployConfig{ Configs: mergeMap(target.Configs, source.Configs), - DbPoolDefinitions: mergeMaps(target.DbPoolDefinitions, source.DbPoolDefinitions), MiddlewareDefinitions: mergeMaps(target.MiddlewareDefinitions, source.MiddlewareDefinitions), ServiceDefinitions: mergeMaps(target.ServiceDefinitions, source.ServiceDefinitions), RouterDefinitions: mergeMaps(target.RouterDefinitions, source.RouterDefinitions), // Renamed from Routers diff --git a/core/deploy/loader/resolver/PROVIDER-REGISTRY.md b/core/deploy/loader/resolver/PROVIDER-REGISTRY.md index 34aa5599..c6323c81 100644 --- a/core/deploy/loader/resolver/PROVIDER-REGISTRY.md +++ b/core/deploy/loader/resolver/PROVIDER-REGISTRY.md @@ -138,24 +138,6 @@ configs: database: host: "prod-db.example.com" db:url: "postgresql://localhost:5432/mydb" # Key with colon - -dbpool-definitions: - main: - # Simple config reference (no colons in key) - host: ${@cfg:database.host} - # Interpreted as: key="database.host" ✅ - - # Config key contains colon - WITHOUT quotes (ambiguous) - url: ${@cfg:db:url} - # Interpreted as: key="db", default="url" ❌ - - # Config key contains colon - WITH quotes (clear) - url: ${@cfg:'db:url'} - # Interpreted as: key="db:url" ✅ - - # With default value - url: ${@cfg:'db:url':postgresql://localhost:5432/fallback} - # Interpreted as: key="db:url", default="postgresql://localhost:5432/fallback" ✅ ``` ### Best Practices @@ -231,11 +213,6 @@ configs: db: host: "prod-db.example.com" port: 5432 - -dbpool-definitions: - main: - host: ${@cfg:db.host} # → "prod-db.example.com" - port: ${@cfg:db.port} # → 5432 ``` **Features:** @@ -502,18 +479,19 @@ configs: aws: region: ${AWS_REGION:us-east-1} # @env provider -dbpool-definitions: +service-definitions: main: - # Mix of providers - host: ${@cfg:db.host} # @cfg provider - port: ${@cfg:db.port} # @cfg provider - database: ${DB_NAME} # @env provider (default) - username: ${@aws-secret:prod/db/username} # @aws-secret provider - password: ${@aws-secret:prod/db/password} # @aws-secret provider - min-conns: 2 - max-conns: 10 + type: dbpool_pg + config: + # Mix of providers + host: ${@cfg:db.host} # @cfg provider + port: ${@cfg:db.port} # @cfg provider + database: ${DB_NAME} # @env provider (default) + username: ${@aws-secret:prod/db/username} # @aws-secret provider + password: ${@aws-secret:prod/db/password} # @aws-secret provider + min-conns: 2 + max-conns: 10 -service-definitions: api-service: type: api-service-factory config: diff --git a/core/deploy/schema/lokstra.schema.json b/core/deploy/schema/lokstra.schema.json index 09778ac1..682d772d 100644 --- a/core/deploy/schema/lokstra.schema.json +++ b/core/deploy/schema/lokstra.schema.json @@ -294,79 +294,6 @@ } }, "additionalProperties": false - }, - "dbPoolConfig": { - "type": "object", - "description": "Database pool configuration", - "properties": { - "dsn": { - "type": "string", - "description": "PostgreSQL connection string (DSN)" - }, - "host": { - "type": "string", - "description": "Database host" - }, - "port": { - "type": "integer", - "description": "Database port", - "default": 5432, - "minimum": 1, - "maximum": 65535 - }, - "database": { - "type": "string", - "description": "Database name" - }, - "username": { - "type": "string", - "description": "Database username" - }, - "password": { - "type": "string", - "description": "Database password" - }, - "schema": { - "type": "string", - "description": "Database schema", - "default": "public" - }, - "min-conns": { - "type": "integer", - "description": "Minimum number of connections in the pool", - "default": 2, - "minimum": 0 - }, - "max-conns": { - "type": "integer", - "description": "Maximum number of connections in the pool", - "default": 10, - "minimum": 1 - }, - "max-idle-time": { - "type": "string", - "description": "Maximum idle time for connections (duration string, e.g., '30m')", - "default": "30m", - "pattern": "^[0-9]+(ns|us|µs|ms|s|m|h)$" - }, - "max-lifetime": { - "type": "string", - "description": "Maximum lifetime for connections (duration string, e.g., '1h')", - "default": "1h", - "pattern": "^[0-9]+(ns|us|µs|ms|s|m|h)$" - }, - "sslmode": { - "type": "string", - "description": "SSL mode for connection", - "enum": ["disable", "allow", "prefer", "require", "verify-ca", "verify-full"], - "default": "disable" - } - }, - "oneOf": [ - { "required": ["dsn"] }, - { "required": ["host", "database"] } - ], - "additionalProperties": false } }, "properties": { @@ -385,13 +312,6 @@ } } }, - "dbpool-definitions": { - "type": "object", - "description": "Named database pool configurations", - "patternProperties": { - "^[a-z][a-z0-9-]*$": { "$ref": "#/definitions/dbPoolConfig" } - } - }, "middleware-definitions": { "type": "object", "description": "Middleware definitions", diff --git a/core/deploy/schema/schema.go b/core/deploy/schema/schema.go index 7bdabd57..71ebc840 100644 --- a/core/deploy/schema/schema.go +++ b/core/deploy/schema/schema.go @@ -14,7 +14,6 @@ func GetSchemaBytes() []byte { // This matches the JSON schema and supports multi-file merging type DeployConfig struct { Configs map[string]any `yaml:"configs" json:"configs"` - DbPoolDefinitions map[string]*DbPoolConfig `yaml:"dbpool-definitions,omitempty" json:"dbpool-definitions,omitempty"` MiddlewareDefinitions map[string]*MiddlewareDef `yaml:"middleware-definitions,omitempty" json:"middleware-definitions,omitempty"` ServiceDefinitions map[string]*ServiceDef `yaml:"service-definitions" json:"service-definitions"` RouterDefinitions map[string]*RouterDef `yaml:"router-definitions,omitempty" json:"router-definitions,omitempty"` // Renamed from Routers @@ -25,30 +24,6 @@ type DeployConfig struct { Servers map[string]*ServerDefMap `yaml:"servers,omitempty" json:"servers,omitempty"` } -// DbPoolConfig defines configuration for a named database pool -type DbPoolConfig struct { - // Connection via DSN (highest priority) - DSN string `yaml:"dsn,omitempty" json:"dsn,omitempty"` - - // Connection via components (used if DSN not provided) - Host string `yaml:"host,omitempty" json:"host,omitempty"` - Port int `yaml:"port,omitempty" json:"port,omitempty"` - Database string `yaml:"database,omitempty" json:"database,omitempty"` - Username string `yaml:"username,omitempty" json:"username,omitempty"` - Password string `yaml:"password,omitempty" json:"password,omitempty"` - - // Schema configuration - Schema string `yaml:"schema,omitempty" json:"schema,omitempty"` // Default: "public" - RlsContext map[string]string `yaml:"rls-context,omitempty" json:"rls-context,omitempty"` // Row Level Security context variables - - // Pool configuration (optional) - MinConns int `yaml:"min-conns,omitempty" json:"min-conns,omitempty"` - MaxConns int `yaml:"max-conns,omitempty" json:"max-conns,omitempty"` - MaxIdleTime string `yaml:"max-idle-time,omitempty" json:"max-idle-time,omitempty"` // Duration string (e.g., "30m") - MaxLifetime string `yaml:"max-lifetime,omitempty" json:"max-lifetime,omitempty"` // Duration string (e.g., "1h") - SSLMode string `yaml:"sslmode,omitempty" json:"sslmode,omitempty"` // SSL mode (disable, require, etc.) -} - // RouterDef defines a router auto-generated from a service // Service name is derived from router name by removing "-router" suffix // Example: "user-service-router" → service is "user-service" diff --git a/docs/02-framework-guide/04-config/examples/07-db-pools/README.md b/docs/02-framework-guide/04-config/examples/07-db-pools/README.md deleted file mode 100644 index 7bfebd8a..00000000 --- a/docs/02-framework-guide/04-config/examples/07-db-pools/README.md +++ /dev/null @@ -1,290 +0,0 @@ -# Example 07 - Named Database Pools - -Demonstrates configuring multiple named database pools with different settings. - -## What's Demonstrated - -- ✅ Multiple database pool configurations -- ✅ DSN vs component-based configuration -- ✅ Pool sizing (min/max connections) -- ✅ Connection timeouts and lifetimes -- ✅ SSL mode configuration -- ✅ Schema-specific pools -- ✅ Service dependencies on specific pools - -## Named Database Pools - -### Configuration Options - -#### Option 1: Component-Based -```yaml -dbpool-definitions: - main-db: - host: "localhost" - port: 5432 - database: "myapp" - username: "postgres" - password: "${DB_PASSWORD}" - schema: "public" - min-conns: 2 - max-conns: 10 - max-idle-time: "30m" - max-lifetime: "1h" - sslmode: "disable" -``` - -#### Option 2: DSN-Based -```yaml -dbpool-definitions: - analytics-db: - dsn: "postgres://user:pass@host:5432/db?sslmode=require" - schema: "analytics" - min-conns: 1 - max-conns: 5 -``` - -## Pool Parameters - -### Connection Sizing - -**min-conns** (default: 2) -- Minimum connections kept alive -- Best practice: 2-4 for most apps - -**max-conns** (default: 10) -- Maximum concurrent connections -- Best practice: 10-20 for web apps - -```yaml -min-conns: 2 # Always keep 2 connections ready -max-conns: 10 # Allow up to 10 concurrent connections -``` - -### Connection Lifecycle - -**max-idle-time** (default: 30m) -- How long idle connections stay alive -- Prevents stale connections - -**max-lifetime** (default: 1h) -- Maximum lifetime of any connection -- Forces periodic connection refresh - -```yaml -max-idle-time: "30m" # Close idle connections after 30 minutes -max-lifetime: "1h" # Refresh all connections every hour -``` - -### SSL Configuration - -**sslmode** options: -- `disable` - No SSL (development only) -- `allow` - Try SSL, fallback to non-SSL -- `prefer` - Try SSL first (default) -- `require` - Require SSL -- `verify-ca` - Require SSL + verify CA -- `verify-full` - Require SSL + verify CA + hostname - -```yaml -sslmode: "require" # Production: always require SSL -``` - -## Use Cases - -### 1. Main Application Database -```yaml -main-db: - host: "localhost" - database: "myapp" - min-conns: 2 - max-conns: 10 - max-idle-time: "30m" -``` - -**Best for:** -- Primary transactional database -- High traffic, frequent queries -- CRUD operations - -### 2. Analytics Database (Read-Only) -```yaml -analytics-db: - dsn: "postgres://readonly:pass@analytics:5432/analytics" - min-conns: 1 - max-conns: 5 - max-idle-time: "10m" -``` - -**Best for:** -- Read-only queries -- Long-running analytics queries -- Lower connection count (fewer writes) - -### 3. Reporting Database -```yaml -reporting-db: - host: "reporting-server" - database: "reports" - min-conns: 1 - max-conns: 3 - max-lifetime: "2h" -``` - -**Best for:** -- Scheduled report generation -- Infrequent access -- Minimal connection pool - -## Service Dependencies - -### Single Pool Dependency -```yaml -service-definitions: - user-repository: - type: user-repository-factory - depends-on: [main-db] -``` - -**Factory signature:** -```go -func UserRepositoryFactory(deps map[string]any, config map[string]any) any { - pool := deps["main-db"].(*pgxpool.Pool) - return &UserRepository{pool: pool} -} -``` - -### Multiple Pool Dependencies -```yaml -service-definitions: - report-generator: - type: report-generator-factory - depends-on: - - mainDb:main-db - - reportDb:reporting-db -``` - -**Factory signature:** -```go -func ReportGeneratorFactory(deps map[string]any, config map[string]any) any { - mainPool := deps["mainDb"].(*pgxpool.Pool) - reportPool := deps["reportDb"].(*pgxpool.Pool) - - return &ReportGenerator{ - mainPool: mainPool, - reportPool: reportPool, - } -} -``` - -## Best Practices - -### 1. Pool Sizing Guidelines - -**Web API servers:** -```yaml -min-conns: 2 -max-conns: 10 -``` - -**Background workers:** -```yaml -min-conns: 1 -max-conns: 5 -``` - -**High-traffic services:** -```yaml -min-conns: 5 -max-conns: 20 -``` - -### 2. Connection Timeouts - -**Interactive applications:** -```yaml -max-idle-time: "30m" -max-lifetime: "1h" -``` - -**Batch processing:** -```yaml -max-idle-time: "10m" -max-lifetime: "2h" -``` - -### 3. SSL Configuration - -**Development:** -```yaml -sslmode: "disable" -``` - -**Production:** -```yaml -sslmode: "require" # or verify-full -``` - -### 4. Environment Variables - -```yaml -dbpool-definitions: - main-db: - host: "${DB_HOST}" - port: 5432 - database: "${DB_NAME}" - username: "${DB_USER}" - password: "${DB_PASSWORD}" -``` - -## Monitoring - -### Pool Health Metrics - -Track these metrics in production: -- Active connections -- Idle connections -- Wait time for connections -- Connection errors -- Query duration - -### Optimization Tips - -**Problem: Connection pool exhausted** -```yaml -# Increase max-conns -max-conns: 20 # from 10 -``` - -**Problem: Too many idle connections** -```yaml -# Decrease max-idle-time -max-idle-time: "15m" # from 30m -``` - -**Problem: Stale connections** -```yaml -# Decrease max-lifetime -max-lifetime: "30m" # from 1h -``` - -## Run - -```bash -# Set environment variables -export DB_PASSWORD="secret123" -export ANALYTICS_PASSWORD="analytics456" -export REPORTING_PASSWORD="reporting789" - -# Run application -go run main.go -``` - -## Summary - -Named database pools allow you to: -- ✅ Configure multiple databases with different settings -- ✅ Optimize connection pooling per use case -- ✅ Use DSN or component-based configuration -- ✅ Inject specific pools into services -- ✅ Monitor and tune pool performance -- ✅ Keep secrets in environment variables diff --git a/docs/02-framework-guide/04-config/examples/07-db-pools/config.yaml b/docs/02-framework-guide/04-config/examples/07-db-pools/config.yaml deleted file mode 100644 index 857f73f0..00000000 --- a/docs/02-framework-guide/04-config/examples/07-db-pools/config.yaml +++ /dev/null @@ -1,75 +0,0 @@ -# Named Database Pools Example - -# Global configs -configs: - app: - name: "DatabaseApp" - -# Named database pool configurations -dbpool-definitions: - # Main application database - main-db: - host: "localhost" - port: 5432 - database: "myapp" - username: "postgres" - password: "${DB_PASSWORD}" - schema: "public" - min-conns: 2 - max-conns: 10 - max-idle-time: "30m" - max-lifetime: "1h" - sslmode: "disable" - - # Analytics database (read-only) - analytics-db: - dsn: "postgres://readonly:${ANALYTICS_PASSWORD}@analytics-server:5432/analytics?sslmode=require" - schema: "analytics" - min-conns: 1 - max-conns: 5 - max-idle-time: "10m" - - # Reporting database - reporting-db: - host: "reporting-server" - port: 5432 - database: "reports" - username: "reporter" - password: "${REPORTING_PASSWORD}" - schema: "reports" - min-conns: 1 - max-conns: 3 - -# Service definitions -service-definitions: - user-repository: - type: user-repository-factory - depends-on: [main-db] # Uses main-db pool - config: - pool-name: "main-db" - - analytics-service: - type: analytics-service-factory - depends-on: [analytics-db] # Uses analytics-db pool - config: - pool-name: "analytics-db" - - report-generator: - type: report-generator-factory - depends-on: - - mainDb:main-db - - reportDb:reporting-db - config: - main-pool: "main-db" - report-pool: "reporting-db" - -deployments: - production: - servers: - api: - base-url: "https://api.myapp.com" - addr: ":8080" - published-services: - - user-repository - - analytics-service - - report-generator diff --git a/docs/02-framework-guide/08-database-pools.md b/docs/02-framework-guide/08-database-pools.md index 6a060a42..c4c75dfd 100644 --- a/docs/02-framework-guide/08-database-pools.md +++ b/docs/02-framework-guide/08-database-pools.md @@ -25,7 +25,7 @@ DbTx (transaction) ### 1. DbPoolManager -**Service** that manages multiple named database pools. Defined in YAML `dbpool-definitions:` section. +**Service** that manages multiple named database pools. - Manages pool configurations (DSN, schema, etc.) - Creates and caches pool instances diff --git a/docs/schema/lokstra.schema.json b/docs/schema/lokstra.schema.json index d8562e90..a4f58ca0 100644 --- a/docs/schema/lokstra.schema.json +++ b/docs/schema/lokstra.schema.json @@ -223,79 +223,6 @@ } }, "additionalProperties": false - }, - "dbPoolConfig": { - "type": "object", - "description": "Database pool configuration", - "properties": { - "dsn": { - "type": "string", - "description": "PostgreSQL connection string (DSN)" - }, - "host": { - "type": "string", - "description": "Database host" - }, - "port": { - "type": "integer", - "description": "Database port", - "default": 5432, - "minimum": 1, - "maximum": 65535 - }, - "database": { - "type": "string", - "description": "Database name" - }, - "username": { - "type": "string", - "description": "Database username" - }, - "password": { - "type": "string", - "description": "Database password" - }, - "schema": { - "type": "string", - "description": "Database schema", - "default": "public" - }, - "min-conns": { - "type": "integer", - "description": "Minimum number of connections in the pool", - "default": 2, - "minimum": 0 - }, - "max-conns": { - "type": "integer", - "description": "Maximum number of connections in the pool", - "default": 10, - "minimum": 1 - }, - "max-idle-time": { - "type": "string", - "description": "Maximum idle time for connections (duration string, e.g., '30m')", - "default": "30m", - "pattern": "^[0-9]+(ns|us|µs|ms|s|m|h)$" - }, - "max-lifetime": { - "type": "string", - "description": "Maximum lifetime for connections (duration string, e.g., '1h')", - "default": "1h", - "pattern": "^[0-9]+(ns|us|µs|ms|s|m|h)$" - }, - "sslmode": { - "type": "string", - "description": "SSL mode for connection", - "enum": ["disable", "allow", "prefer", "require", "verify-ca", "verify-full"], - "default": "disable" - } - }, - "oneOf": [ - { "required": ["dsn"] }, - { "required": ["host", "database"] } - ], - "additionalProperties": false } }, "properties": { @@ -314,13 +241,6 @@ } } }, - "dbpool-definitions": { - "type": "object", - "description": "Named database pool configurations", - "patternProperties": { - "^[a-z][a-z0-9-]*$": { "$ref": "#/definitions/dbPoolConfig" } - } - }, "middleware-definitions": { "type": "object", "description": "Middleware definitions", diff --git a/lokstra_init/initialize.go b/lokstra_init/initialize.go index 1d009e7e..b54487a5 100644 --- a/lokstra_init/initialize.go +++ b/lokstra_init/initialize.go @@ -32,19 +32,15 @@ type InitializeConfig struct { PgxSyncHeartbeatInterval time.Duration PgxSyncReconnectInterval time.Duration - // 5. EnableDbPoolManager - EnableDbPoolManager bool - IsDbPoolAutoSync bool - - // 6. EnableDbMigration + // 5. EnableDbMigration EnableDbMigration bool MigrationFolder string SkipMigrationOnProd bool - // 7. ServerInit Func + // 6. ServerInit Func ServerInitFunc func() error - // 8. Init and Run Server + // 7. Init and Run Server IsRunServer bool } @@ -62,7 +58,6 @@ func BootstrapAndRun(opts ...InitializeOption) error { LogLevel: logger.LogLevelInfo, EnableLoadConfig: true, EnableAnnotation: true, // Auto-detect @RouterService - IsDbPoolAutoSync: false, EnablePgxSyncMap: false, SkipMigrationOnProd: true, PgxSyncMapDbPoolName: "db_main", @@ -88,8 +83,6 @@ func BootstrapAndRunWithConfig(cfg *InitializeConfig) error { if len(cfg.PgxSyncMapDbPoolName) == 0 { return cfg.returnError(fmt.Errorf("PgxSyncMapDbPoolName must be set when UsePgxSyncMap is true")) } - } else if cfg.IsDbPoolAutoSync { - return cfg.returnError(fmt.Errorf("IsDbPoolAutoSync requires PgxSyncMap to be enabled")) } // 1. Set log level @@ -117,14 +110,7 @@ func BootstrapAndRunWithConfig(cfg *InitializeConfig) error { cfg.PgxSyncHeartbeatInterval, cfg.PgxSyncReconnectInterval) } - // 5. Crewate Db Pool Manager and load definitions from config - UsePgxDbPoolManager(cfg.IsDbPoolAutoSync) - - if err := loader.LoadDbPoolDefsFromConfig(); err != nil { - return cfg.returnError(err) - } - - // 6. Check DB Migrations + // 5. Check DB Migrations if cfg.EnableDbMigration { if mode := GetRuntimeMode(); mode != "prod" || !cfg.SkipMigrationOnProd { if err := CheckDbMigrationsAuto(cfg.MigrationFolder); err != nil { @@ -133,14 +119,14 @@ func BootstrapAndRunWithConfig(cfg *InitializeConfig) error { } } - // 7. Server Init Func + // 6. Server Init Func if cfg.ServerInitFunc != nil { if err := cfg.ServerInitFunc(); err != nil { return cfg.returnError(err) } } - // 8. Init and Run Server + // 7. Init and Run Server if cfg.IsRunServer { if err := lokstra_registry.RunConfiguredServer(); err != nil { return cfg.returnError(err) diff --git a/lokstra_init/migration.go b/lokstra_init/migration.go index 39957d61..b21e4929 100644 --- a/lokstra_init/migration.go +++ b/lokstra_init/migration.go @@ -6,7 +6,6 @@ import ( "os" "path/filepath" - "github.com/primadi/lokstra/common/dbpool_crud" "github.com/primadi/lokstra/common/logger" "github.com/primadi/lokstra/common/utils" "github.com/primadi/lokstra/lokstra_init/migration_runner" @@ -18,7 +17,7 @@ import ( // MigrationYamlConfig represents the migration.yaml file structure // This file is optional and located in the migrations directory type MigrationYamlConfig struct { - // DbPoolName is the database pool name from config.yaml dbpool-definitions + // DbPoolName is the database pool name from config.yaml service-definitions DbPoolName string `yaml:"dbpool-name"` // SchemaTable is the table name for tracking migrations @@ -107,11 +106,7 @@ func CheckDbMigration(cfg *MigrationConfig) error { // Apply final defaults if still empty if cfg.DbPoolName == "" { - names, err := dbpool_crud.ListDbPools() - if err != nil || len(names) == 0 { - return fmt.Errorf("no database pools defined - check your config.yaml dbpool-definitions section") - } - cfg.DbPoolName = names[0] // Use first available pool + return fmt.Errorf("no database pools defined - check your config.yaml service-definitions section") } if cfg.SchemaTable == "" { cfg.SchemaTable = "schema_migrations" @@ -120,7 +115,7 @@ func CheckDbMigration(cfg *MigrationConfig) error { // Get database pool pool, ok := lokstra_registry.GetServiceAny(cfg.DbPoolName) if !ok { - return fmt.Errorf("database pool '%s' not found - check your config.yaml dbpool-definitions section", cfg.DbPoolName) + return fmt.Errorf("database pool '%s' not found - check your config.yaml service-definitions section", cfg.DbPoolName) } dbPool, ok := pool.(serviceapi.DbPool) diff --git a/lokstra_init/migration_runner/example/config/config.yaml b/lokstra_init/migration_runner/example/config/config.yaml index 7e316803..a3b09968 100644 --- a/lokstra_init/migration_runner/example/config/config.yaml +++ b/lokstra_init/migration_runner/example/config/config.yaml @@ -3,16 +3,18 @@ configs: name: "Migration Example" version: "1.0.0" -dbpool-definitions: +service-definitions: main-db: - host: localhost - port: 5432 - database: migration_example - username: postgres - password: postgres - schema: public - min-conns: 2 - max-conns: 10 - max-idle-time: 30m - max-lifetime: 1h - sslmode: disable + type: dbpool_pg + config: + host: localhost + port: 5432 + database: migration_example + username: postgres + password: postgres + schema: public + min-conns: 2 + max-conns: 10 + max-idle-time: 30m + max-lifetime: 1h + sslmode: disable diff --git a/lokstra_init/migration_runner/example/example_multi_db.go b/lokstra_init/migration_runner/example/example_multi_db.go index 1d10a86a..f18d602a 100644 --- a/lokstra_init/migration_runner/example/example_multi_db.go +++ b/lokstra_init/migration_runner/example/example_multi_db.go @@ -21,12 +21,7 @@ func main() { log.Fatalf("Failed to load config: %v", err) } - lokstra_init.UsePgxDbPoolManager(true) - sync_config_pg.Register("db_main", 5*time.Minute, 5*time.Second) - if err := loader.LoadDbPoolDefsFromConfig(); err != nil { - log.Fatalf("Failed to load named DB pools: %v", err) - } // OPTION 1: Auto-scan all migration folders (RECOMMENDED) // Scans multi_db/ for subdirectories, runs them in alphabetical order diff --git a/lokstra_init/migration_runner/example/main.go b/lokstra_init/migration_runner/example/main.go index 5b1b93cb..f072a5dd 100644 --- a/lokstra_init/migration_runner/example/main.go +++ b/lokstra_init/migration_runner/example/main.go @@ -46,16 +46,7 @@ func MainTest() { log.Fatalf("Failed to load config: %v", err) } - // Get database pool - pool, ok := lokstra_registry.GetServiceAny(*dbName) - if !ok { - log.Fatalf("❌ Database pool '%s' not found. Check your config.yaml dbpool-definitions section", *dbName) - } - - dbPool, ok := pool.(serviceapi.DbPool) - if !ok { - log.Fatalf("❌ Service '%s' is not a DbPool", *dbName) - } + dbPool := lokstra_registry.GetService[serviceapi.DbPool](*dbName) // Create migration runner runner := migration_runner.New(dbPool, *migrationsDir) diff --git a/lokstra_init/migration_runner/example/migration.yaml b/lokstra_init/migration_runner/example/migration.yaml index e810ac49..40eaf090 100644 --- a/lokstra_init/migration_runner/example/migration.yaml +++ b/lokstra_init/migration_runner/example/migration.yaml @@ -3,8 +3,6 @@ # # This file allows you to configure migration behavior per database pool # Useful for multi-database systems (tenant-db, ledger-db, analytics-db, etc) - -# Database pool name from config.yaml dbpool-definitions section # If not specified, defaults to "main-db" dbpool-name: main-db diff --git a/lokstra_init/option.go b/lokstra_init/option.go index 6a46ccda..a1a81c46 100644 --- a/lokstra_init/option.go +++ b/lokstra_init/option.go @@ -61,14 +61,6 @@ func WithPgxSyncMapIntervals(heartBeatInterval, reconnectInterval time.Duration) } } -// enable or disable database pool auto synchronization -// default enable is false -func WithDbPoolAutoSync(enable bool) InitializeOption { - return func(c *InitializeConfig) { - c.IsDbPoolAutoSync = enable - } -} - // enable or disable database migrations with the given migration folder // default enable is false func WithDbMigrations(enable bool, folder string) InitializeOption { diff --git a/lokstra_init/pgx_dbpool_manager.go b/lokstra_init/pgx_dbpool_manager.go deleted file mode 100644 index c50823f7..00000000 --- a/lokstra_init/pgx_dbpool_manager.go +++ /dev/null @@ -1,28 +0,0 @@ -package lokstra_init - -import ( - "github.com/primadi/lokstra/common/logger" - "github.com/primadi/lokstra/lokstra_registry" - "github.com/primadi/lokstra/serviceapi" - "github.com/primadi/lokstra/services/dbpool_manager" -) - -// auto create dbpool-manager service if not exists -func UsePgxDbPoolManager(useSync bool) { - pm := lokstra_registry.GetService[serviceapi.DbPoolManager]("dbpool-manager") - if pm != nil { - return // Already registered - } - - if useSync { - // ensure lokstra_core_sql migration is applied - pm = dbpool_manager.NewPgxSyncDbPoolManager() - logger.LogDebug("[Lokstra] DbPoolManager initialized with distributed sync") - } else { - // Default: use regular pool manager (local sync.Map) - pm = dbpool_manager.NewPgxPoolManager() - logger.LogDebug("[Lokstra] DbPoolManager initialized with local sync") - } - - lokstra_registry.RegisterService("dbpool-manager", pm) -} diff --git a/project_templates/02_app_framework/01_enterprise_router_service/config/deployment.yaml b/project_templates/02_app_framework/01_enterprise_router_service/config/deployment.yaml index 0c25660c..dce462c1 100644 --- a/project_templates/02_app_framework/01_enterprise_router_service/config/deployment.yaml +++ b/project_templates/02_app_framework/01_enterprise_router_service/config/deployment.yaml @@ -5,10 +5,12 @@ configs: store: order-repository: order-repository -dbpool-definitions: +service-definitions: db_main: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} deployments: development: diff --git a/project_templates/02_app_framework/02_sync_config/config/config.yaml b/project_templates/02_app_framework/02_sync_config/config/config.yaml index 32f695e8..4612a716 100644 --- a/project_templates/02_app_framework/02_sync_config/config/config.yaml +++ b/project_templates/02_app_framework/02_sync_config/config/config.yaml @@ -1,9 +1,11 @@ # yaml-language-server: $schema=https://primadi.github.io/lokstra/schema/lokstra.schema.json -dbpool-definitions: +service-definitions: db_main: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} deployments: monolith: diff --git a/project_templates/02_app_framework/02_sync_config/main.go b/project_templates/02_app_framework/02_sync_config/main.go index a56642f5..8e5a72dd 100644 --- a/project_templates/02_app_framework/02_sync_config/main.go +++ b/project_templates/02_app_framework/02_sync_config/main.go @@ -9,7 +9,6 @@ func main() { // lokstra_init.WithAnnotations(true), // default is true // lokstra_init.WithYAMLConfigPath(true, "config"), // default is true, path is empty (config folder) lokstra_init.WithPgSyncMap(true, "db_main"), - lokstra_init.WithDbPoolAutoSync(true), lokstra_init.WithDbMigrations(true, "migrations"), lokstra_init.WithServerInitFunc(func() error { registerRouters() diff --git a/project_templates/02_app_framework/03_tenant_management/application/zz_cache.lokstra.json b/project_templates/02_app_framework/03_tenant_management/application/zz_cache.lokstra.json index 0aca8195..34d35168 100644 --- a/project_templates/02_app_framework/03_tenant_management/application/zz_cache.lokstra.json +++ b/project_templates/02_app_framework/03_tenant_management/application/zz_cache.lokstra.json @@ -3,15 +3,15 @@ "files": { "tenant_service.go": { "filename": "tenant_service.go", - "checksum": "a8d23ee05ed47fe7d1a9eb4a760199dbd8bb5477d95c59ccb846f76710f85a86", + "checksum": "969d1b3b563fa5423bd6f31929f90ad6067b4d03f80d14877f672c1c3f057d51", "annotations": 10, - "last_scan": "2025-12-18T16:07:08.4971722+07:00", + "last_scan": "2026-01-03T01:29:24.1580941+07:00", "generated": [ "zz_generated.lokstra.go" ], - "generated_mod_time": "2025-12-18T16:07:08.4971722+07:00" + "generated_mod_time": "2026-01-03T01:29:24.1580941+07:00" } }, - "updated_at": "2025-12-19T21:23:40.8371545+07:00", + "updated_at": "2026-01-03T02:13:01.786635+07:00", "generated_checksum": "df8bffc551242f55129afe8190e26cdc9c1adb32bbd9b1b36878ef257ce0e43a" } \ No newline at end of file diff --git a/project_templates/02_app_framework/03_tenant_management/config/config.yaml b/project_templates/02_app_framework/03_tenant_management/config/config.yaml index 48299b69..faf76478 100644 --- a/project_templates/02_app_framework/03_tenant_management/config/config.yaml +++ b/project_templates/02_app_framework/03_tenant_management/config/config.yaml @@ -5,14 +5,17 @@ configs: tenant-store: postgres-tenant-store user-store: postgres-user-store -dbpool-definitions: +service-definitions: db_core: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_core} + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db2} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} db_auth: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_core} - + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db2} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} servers: api-server: diff --git a/project_templates/02_app_framework/03_tenant_management/main.go b/project_templates/02_app_framework/03_tenant_management/main.go index ce1d380f..360373df 100644 --- a/project_templates/02_app_framework/03_tenant_management/main.go +++ b/project_templates/02_app_framework/03_tenant_management/main.go @@ -1,18 +1,22 @@ package main import ( + "github.com/primadi/lokstra/common/logger" "github.com/primadi/lokstra/lokstra_init" "github.com/primadi/lokstra/middleware/recovery" "github.com/primadi/lokstra/middleware/request_logger" + "github.com/primadi/lokstra/services/dbpool_pg" ) func main() { recovery.Register() request_logger.Register() + dbpool_pg.Register() lokstra_init.BootstrapAndRun( - lokstra_init.WithPgSyncMap(true, "db_core"), - lokstra_init.WithDbPoolAutoSync(true), + lokstra_init.WithLogLevel(logger.LogLevelDebug), + // lokstra_init.WithPgSyncMap(true, "db_auth"), + // lokstra_init.WithDbPoolAutoSync(true), // lokstra_init.WithDbMigrations(true, "migrations"), ) } diff --git a/project_templates/02_app_framework/03_tenant_management/repository/zz_cache.lokstra.json b/project_templates/02_app_framework/03_tenant_management/repository/zz_cache.lokstra.json index 02d2bea8..abb1ab4b 100644 --- a/project_templates/02_app_framework/03_tenant_management/repository/zz_cache.lokstra.json +++ b/project_templates/02_app_framework/03_tenant_management/repository/zz_cache.lokstra.json @@ -22,6 +22,6 @@ "generated_mod_time": "2025-12-17T17:16:58.2514677+07:00" } }, - "updated_at": "2025-12-19T21:23:40.8377344+07:00", + "updated_at": "2026-01-03T02:13:01.787159+07:00", "generated_checksum": "f3e29afd20b1546d51b0f487725dc26863d7cf2c011aba9071035b6308ae8956" } \ No newline at end of file diff --git a/serviceapi/kvstore.go b/serviceapi/kvstore.go index 126a3b0c..bcc138be 100644 --- a/serviceapi/kvstore.go +++ b/serviceapi/kvstore.go @@ -20,4 +20,10 @@ type KvStore interface { // Gets all keys matching the pattern. Keys(ctx context.Context, pattern string) ([]string, error) + + // Sets Prefix for all keys + SetPrefix(prefix string) + + // Gets Prefix for all keys + GetPrefix() string } diff --git a/services/dbpool_manager/sync_pool_manager.go b/services/dbpool_manager/sync_dbpool_manager.go similarity index 100% rename from services/dbpool_manager/sync_pool_manager.go rename to services/dbpool_manager/sync_dbpool_manager.go diff --git a/services/dbpool_pg/dbpool_postgres.go b/services/dbpool_pg/dbpool_postgres.go index dd08a485..e3b09635 100644 --- a/services/dbpool_pg/dbpool_postgres.go +++ b/services/dbpool_pg/dbpool_postgres.go @@ -59,21 +59,21 @@ func (r *rowWrapper) Scan(dest ...any) error { return r.row.Scan(dest...) } -type pgxPostgresPool struct { +type PgxPostgresPool struct { pool *pgxpool.Pool poolName string // Pool name for transaction tracking - dsn string - schema string - rlsContext map[string]string + Dsn string + Schema string + RlsContext map[string]string } // Begin implements serviceapi.DbPool. -func (p *pgxPostgresPool) Begin(ctx context.Context) (serviceapi.DbTx, error) { +func (p *PgxPostgresPool) Begin(ctx context.Context) (serviceapi.DbTx, error) { panic("Begin() should not be called directly on DbPool. Use Acquire() first to get a DbConn, then call Begin() on it.") } // Exec implements serviceapi.DbPool. -func (p *pgxPostgresPool) Exec(ctx context.Context, query string, args ...any) (serviceapi.CommandResult, error) { +func (p *PgxPostgresPool) Exec(ctx context.Context, query string, args ...any) (serviceapi.CommandResult, error) { cn, err := p.Acquire(ctx) if err != nil { return nil, fmt.Errorf("failed to acquire connection: %w", err) @@ -83,12 +83,12 @@ func (p *pgxPostgresPool) Exec(ctx context.Context, query string, args ...any) ( } // IsErrorNoRows implements serviceapi.DbPool. -func (p *pgxPostgresPool) IsErrorNoRows(err error) bool { +func (p *PgxPostgresPool) IsErrorNoRows(err error) bool { return errors.Is(err, pgx.ErrNoRows) } // IsExists implements serviceapi.DbPool. -func (p *pgxPostgresPool) IsExists(ctx context.Context, query string, args ...any) (bool, error) { +func (p *PgxPostgresPool) IsExists(ctx context.Context, query string, args ...any) (bool, error) { cn, err := p.Acquire(ctx) if err != nil { return false, fmt.Errorf("failed to acquire connection: %w", err) @@ -98,7 +98,7 @@ func (p *pgxPostgresPool) IsExists(ctx context.Context, query string, args ...an } // Ping implements serviceapi.DbPool. -func (p *pgxPostgresPool) Ping(ctx context.Context) error { +func (p *PgxPostgresPool) Ping(ctx context.Context) error { cn, err := p.Acquire(ctx) if err != nil { return fmt.Errorf("failed to acquire connection: %w", err) @@ -108,7 +108,7 @@ func (p *pgxPostgresPool) Ping(ctx context.Context) error { } // Query implements serviceapi.DbPool. -func (p *pgxPostgresPool) Query(ctx context.Context, query string, args ...any) (serviceapi.Rows, error) { +func (p *PgxPostgresPool) Query(ctx context.Context, query string, args ...any) (serviceapi.Rows, error) { cn, err := p.Acquire(ctx) if err != nil { return nil, fmt.Errorf("failed to acquire connection: %w", err) @@ -123,7 +123,7 @@ func (p *pgxPostgresPool) Query(ctx context.Context, query string, args ...any) } // QueryRow implements serviceapi.DbPool. -func (p *pgxPostgresPool) QueryRow(ctx context.Context, query string, args ...any) serviceapi.Row { +func (p *PgxPostgresPool) QueryRow(ctx context.Context, query string, args ...any) serviceapi.Row { cn, err := p.Acquire(ctx) if err != nil { return &errorRow{err: err} @@ -134,12 +134,12 @@ func (p *pgxPostgresPool) QueryRow(ctx context.Context, query string, args ...an } // Release implements serviceapi.DbPool. -func (p *pgxPostgresPool) Release() error { +func (p *PgxPostgresPool) Release() error { panic("Release() should not be called directly on DbPool. Use Acquire() first to get a DbConn, then call Release() on it.") } // SelectManyRowMap implements serviceapi.DbPool. -func (p *pgxPostgresPool) SelectManyRowMap(ctx context.Context, query string, args ...any) ([]serviceapi.RowMap, error) { +func (p *PgxPostgresPool) SelectManyRowMap(ctx context.Context, query string, args ...any) ([]serviceapi.RowMap, error) { cn, err := p.Acquire(ctx) if err != nil { return nil, fmt.Errorf("failed to acquire connection: %w", err) @@ -149,7 +149,7 @@ func (p *pgxPostgresPool) SelectManyRowMap(ctx context.Context, query string, ar } // SelectManyWithMapper implements serviceapi.DbPool. -func (p *pgxPostgresPool) SelectManyWithMapper(ctx context.Context, fnScan func(serviceapi.Row) (any, error), query string, args ...any) (any, error) { +func (p *PgxPostgresPool) SelectManyWithMapper(ctx context.Context, fnScan func(serviceapi.Row) (any, error), query string, args ...any) (any, error) { cn, err := p.Acquire(ctx) if err != nil { return nil, fmt.Errorf("failed to acquire connection: %w", err) @@ -159,7 +159,7 @@ func (p *pgxPostgresPool) SelectManyWithMapper(ctx context.Context, fnScan func( } // SelectMustOne implements serviceapi.DbPool. -func (p *pgxPostgresPool) SelectMustOne(ctx context.Context, query string, args []any, dest ...any) error { +func (p *PgxPostgresPool) SelectMustOne(ctx context.Context, query string, args []any, dest ...any) error { cn, err := p.Acquire(ctx) if err != nil { return fmt.Errorf("failed to acquire connection: %w", err) @@ -169,7 +169,7 @@ func (p *pgxPostgresPool) SelectMustOne(ctx context.Context, query string, args } // SelectOne implements serviceapi.DbPool. -func (p *pgxPostgresPool) SelectOne(ctx context.Context, query string, args []any, dest ...any) error { +func (p *PgxPostgresPool) SelectOne(ctx context.Context, query string, args []any, dest ...any) error { cn, err := p.Acquire(ctx) if err != nil { return fmt.Errorf("failed to acquire connection: %w", err) @@ -179,7 +179,7 @@ func (p *pgxPostgresPool) SelectOne(ctx context.Context, query string, args []an } // SelectOneRowMap implements serviceapi.DbPool. -func (p *pgxPostgresPool) SelectOneRowMap(ctx context.Context, query string, args ...any) (serviceapi.RowMap, error) { +func (p *PgxPostgresPool) SelectOneRowMap(ctx context.Context, query string, args ...any) (serviceapi.RowMap, error) { cn, err := p.Acquire(ctx) if err != nil { return nil, fmt.Errorf("failed to acquire connection: %w", err) @@ -189,7 +189,7 @@ func (p *pgxPostgresPool) SelectOneRowMap(ctx context.Context, query string, arg } // Transaction implements serviceapi.DbPool. -func (p *pgxPostgresPool) Transaction(ctx context.Context, fn func(tx serviceapi.DbExecutor) error) error { +func (p *PgxPostgresPool) Transaction(ctx context.Context, fn func(tx serviceapi.DbExecutor) error) error { cn, err := p.Acquire(ctx) if err != nil { return fmt.Errorf("failed to acquire connection: %w", err) @@ -199,35 +199,35 @@ func (p *pgxPostgresPool) Transaction(ctx context.Context, fn func(tx serviceapi } // SetSchemaRls implements serviceapi.DbPoolSchemaRls. -func (p *pgxPostgresPool) SetSchemaRls(schema string, rlsContext map[string]string) { - p.schema = schema - p.rlsContext = rlsContext +func (p *PgxPostgresPool) SetSchemaRls(schema string, rlsContext map[string]string) { + p.Schema = schema + p.RlsContext = rlsContext } // Shutdown implements serviceapi.DbPool. -func (p *pgxPostgresPool) Shutdown() error { +func (p *PgxPostgresPool) Shutdown() error { p.pool.Close() return nil } -func (p *pgxPostgresPool) Acquire(ctx context.Context) (serviceapi.DbConn, error) { +func (p *PgxPostgresPool) Acquire(ctx context.Context) (serviceapi.DbConn, error) { conn, err := p.pool.Acquire(ctx) if err != nil { return nil, err } - if len(p.schema) > 0 { - stmt := "SET search_path TO " + pgx.Identifier{p.schema}.Sanitize() + if len(p.Schema) > 0 { + stmt := "SET search_path TO " + pgx.Identifier{p.Schema}.Sanitize() if _, err := conn.Exec(ctx, stmt); err != nil { conn.Release() return nil, err } } - if len(p.rlsContext) > 0 { + if len(p.RlsContext) > 0 { // Build all SET LOCAL statements in one query var stmts []string - for key, value := range p.rlsContext { + for key, value := range p.RlsContext { // Key is sanitized as identifier, value is quoted as literal string stmt := fmt.Sprintf("SET LOCAL %s = '%s'", pgx.Identifier{key}.Sanitize(), value) stmts = append(stmts, stmt) @@ -245,10 +245,10 @@ func (p *pgxPostgresPool) Acquire(ctx context.Context) (serviceapi.DbConn, error }, nil } -var _ serviceapi.DbPool = (*pgxPostgresPool)(nil) -var _ serviceapi.DbPoolSchemaRls = (*pgxPostgresPool)(nil) +var _ serviceapi.DbPool = (*PgxPostgresPool)(nil) +var _ serviceapi.DbPoolSchemaRls = (*PgxPostgresPool)(nil) -func NewPgxPostgresPool(poolName string, dsn string, schema string, rlsContext map[string]string) (*pgxPostgresPool, error) { +func NewPgxPostgresPool(poolName string, dsn string, schema string, rlsContext map[string]string) (*PgxPostgresPool, error) { ctx := context.Background() pool, err := pgxpool.New(ctx, dsn) if err != nil { @@ -258,11 +258,11 @@ func NewPgxPostgresPool(poolName string, dsn string, schema string, rlsContext m return nil, err } - return &pgxPostgresPool{ + return &PgxPostgresPool{ pool: pool, poolName: poolName, - dsn: dsn, - schema: schema, - rlsContext: rlsContext, + Dsn: dsn, + Schema: schema, + RlsContext: rlsContext, }, nil } diff --git a/services/dbpool_pg/module.go b/services/dbpool_pg/module.go index 53d6a883..b9123269 100644 --- a/services/dbpool_pg/module.go +++ b/services/dbpool_pg/module.go @@ -50,31 +50,41 @@ func (cfg *Config) GetFinalDSN() string { } dsnFinal := cfg.DSN + // Determine separator: ? if no query params yet, & if already has params + separator := "&" + if !strings.Contains(dsnFinal, "?") { + separator = "?" + } + if !strings.Contains(dsnFinal, "pool_min_conns=") { - dsnFinal += fmt.Sprintf("&pool_min_conns=%d", cfg.MinConns) + dsnFinal += fmt.Sprintf("%spool_min_conns=%d", separator, cfg.MinConns) + separator = "&" } if !strings.Contains(dsnFinal, "pool_max_conns=") { - dsnFinal += fmt.Sprintf("&pool_max_conns=%d", cfg.MaxConns) + dsnFinal += fmt.Sprintf("%spool_max_conns=%d", separator, cfg.MaxConns) + separator = "&" } if !strings.Contains(dsnFinal, "pool_max_conn_idle_time=") { - dsnFinal += fmt.Sprintf("&pool_max_conn_idle_time=%s", cfg.MaxIdleTime) + dsnFinal += fmt.Sprintf("%spool_max_conn_idle_time=%s", separator, cfg.MaxIdleTime) + separator = "&" } if !strings.Contains(dsnFinal, "pool_max_conn_lifetime=") { - dsnFinal += fmt.Sprintf("&pool_max_conn_lifetime=%s", cfg.MaxLifetime) + dsnFinal += fmt.Sprintf("%spool_max_conn_lifetime=%s", separator, cfg.MaxLifetime) + separator = "&" } if !strings.Contains(dsnFinal, "sslmode=") { - dsnFinal += fmt.Sprintf("&sslmode=%s", cfg.SSLMode) + dsnFinal += fmt.Sprintf("%ssslmode=%s", separator, cfg.SSLMode) } return dsnFinal } -func Service(poolName string, cfg *Config) *pgxPostgresPool { +func Service(poolName string, cfg *Config) *PgxPostgresPool { dsn := cfg.GetFinalDSN() svc, err := NewPgxPostgresPool(poolName, dsn, cfg.Schema, cfg.RlsContext) if err != nil { - return nil + panic(fmt.Sprintf("failed to create dbpool_pg service for pool '%s': %v", poolName, err)) } return svc } diff --git a/services/kvstore/kvstore_inmemory/module.go b/services/kvstore/kvstore_inmemory/module.go new file mode 100644 index 00000000..de809d2a --- /dev/null +++ b/services/kvstore/kvstore_inmemory/module.go @@ -0,0 +1,179 @@ +package kvstore_inmemory + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/primadi/lokstra/common/utils" + "github.com/primadi/lokstra/lokstra_registry" + "github.com/primadi/lokstra/serviceapi" +) + +const SERVICE_TYPE = "kvstore_inmemory" + +var ErrKeyNotFound = errors.New("key not found") + +var ( + mu sync.RWMutex + data = make(map[string]kvEntry) + + MaxCounter int = 30 + cleanupCounter int +) + +type kvEntry struct { + value any + expiresAt *time.Time +} + +type kvStoreInMemory struct { + prefix string +} + +func checkCleanUp() { + go func() { + mu.Lock() + defer mu.Unlock() + + if cleanupCounter < MaxCounter { + cleanupCounter++ + return + } + + // Perform cleanup + now := time.Now() + for key, entry := range data { + if entry.expiresAt != nil && now.After(*entry.expiresAt) { + delete(data, key) + } + } + cleanupCounter = 0 + }() +} + +func (k *kvStoreInMemory) prefixKey(key string) string { + if k.prefix != "" { + return k.prefix + ":" + key + } + return key +} + +// Delete implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) Delete(ctx context.Context, key string) error { + mu.Lock() + delete(data, k.prefixKey(key)) + mu.Unlock() + + checkCleanUp() + return nil +} + +// DeleteKeys implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) DeleteKeys(ctx context.Context, keys ...string) error { + mu.Lock() + for _, key := range keys { + delete(data, k.prefixKey(key)) + } + mu.Unlock() + + checkCleanUp() + return nil +} + +// Get implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) Get(ctx context.Context, key string, dest any) error { + mu.RLock() + defer mu.RUnlock() + + entry, exists := data[k.prefixKey(key)] + if !exists { + return ErrKeyNotFound + } + if entry.expiresAt != nil && time.Now().After(*entry.expiresAt) { + return ErrKeyNotFound + } + dest = entry.value + return nil +} + +// GetPrefix implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) GetPrefix() string { + return k.prefix +} + +// Keys implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) Keys(ctx context.Context, pattern string) ([]string, error) { + pattern = k.prefixKey(pattern) + lenPattern := len(pattern) + wildcardPattern := lenPattern > 0 && pattern[lenPattern-1] == '*' + startItem := len(k.prefix) + if startItem > 0 { + startItem++ // to account for the colon + } + + var keys []string + + mu.RLock() + defer mu.RUnlock() + for key, value := range data { + // Simple pattern matching: only supports '*' at the end + if wildcardPattern { + prefix := pattern[:lenPattern-1] + if len(key) >= len(prefix) && key[:len(prefix)] == prefix { + if value.expiresAt == nil || time.Now().Before(*value.expiresAt) { + keys = append(keys, key[startItem:]) + } + } + } else if key == pattern { + if value.expiresAt == nil || time.Now().Before(*value.expiresAt) { + keys = append(keys, key[startItem:]) + } + } + } + return keys, nil +} + +// Set implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) Set(ctx context.Context, key string, value any, ttl time.Duration) error { + mu.Lock() + var expiresAt *time.Time + if ttl > 0 { + t := time.Now().Add(ttl) + expiresAt = &t + } + data[k.prefixKey(key)] = kvEntry{ + value: value, + expiresAt: expiresAt, + } + mu.Unlock() + + checkCleanUp() + return nil +} + +// SetPrefix implements [serviceapi.KvStore]. +func (k *kvStoreInMemory) SetPrefix(prefix string) { + k.prefix = prefix +} + +var _ serviceapi.KvStore = (*kvStoreInMemory)(nil) + +// creates a new instance of kvStoreInMemory service. +func Service(prefix string) *kvStoreInMemory { + return &kvStoreInMemory{ + prefix: prefix, + } +} + +// the factory function for kvStoreInMemory service. +func ServiceFactory(config map[string]any) any { + prefix := utils.GetValueFromMap(config, "prefix", "") + return Service(prefix) +} + +// registers the kvStoreInMemory service type. +func Register() { + lokstra_registry.RegisterServiceType(SERVICE_TYPE, ServiceFactory) +} diff --git a/services/kvstore/kvstore_postgres/module.go b/services/kvstore/kvstore_postgres/module.go new file mode 100644 index 00000000..427b9da3 --- /dev/null +++ b/services/kvstore/kvstore_postgres/module.go @@ -0,0 +1,132 @@ +package kvstore_postgres + +import ( + "context" + "errors" + "strings" + "time" + + "github.com/primadi/lokstra/common/utils" + "github.com/primadi/lokstra/lokstra_registry" + "github.com/primadi/lokstra/serviceapi" +) + +var SERVICE_TYPE = "kvstore_postgres" + +var ErrKeyNotFound = errors.New("key not found") + +type kvStorePostgres struct { + dbPool serviceapi.DbPool + prefix string +} + +func (k *kvStorePostgres) prefixKey(key string) string { + if k.prefix != "" { + return k.prefix + ":" + key + } + return key +} + +// Delete implements [serviceapi.KvStore]. +func (k *kvStorePostgres) Delete(ctx context.Context, key string) error { + _, err := k.dbPool.Exec(ctx, "DELETE FROM kvstore WHERE key = $1", k.prefixKey(key)) + return err +} + +// DeleteKeys implements [serviceapi.KvStore]. +func (k *kvStorePostgres) DeleteKeys(ctx context.Context, keys ...string) error { + _, err := k.dbPool.Exec(ctx, "DELETE FROM kvstore WHERE key = ANY($1)", func() []string { + prefixedKeys := make([]string, len(keys)) + for i, key := range keys { + prefixedKeys[i] = k.prefixKey(key) + } + return prefixedKeys + }()) + return err +} + +// Get implements [serviceapi.KvStore]. +func (k *kvStorePostgres) Get(ctx context.Context, key string, dest any) error { + err := k.dbPool.SelectMustOne(ctx, + "SELECT value FROM kvstore WHERE key = $1 AND (expiresat IS NULL OR expiresat > NOW())", + []any{k.prefixKey(key)}, &dest) + if k.dbPool.IsErrorNoRows(err) { + return ErrKeyNotFound + } + return err +} + +// GetPrefix implements [serviceapi.KvStore]. +func (k *kvStorePostgres) GetPrefix() string { + return k.prefix +} + +// Keys implements [serviceapi.KvStore]. +func (k *kvStorePostgres) Keys(ctx context.Context, pattern string) ([]string, error) { + pattern = strings.ReplaceAll(pattern, "*", "%") + rows, err := k.dbPool.Query(ctx, + "SELECT key FROM kvstore WHERE key LIKE $1 AND (expiresat IS NULL OR expiresat > NOW())", + k.prefixKey(pattern)) + if err != nil { + return nil, err + } + defer rows.Close() + + startItem := len(k.prefix) + if startItem > 0 { + startItem++ // to account for the colon + } + + var keys []string + for rows.Next() { + var key string + if err := rows.Scan(&key); err != nil { + return nil, err + } + keys = append(keys, key[startItem:]) + } + return keys, nil +} + +// Set implements [serviceapi.KvStore]. +func (k *kvStorePostgres) Set(ctx context.Context, key string, value any, ttl time.Duration) error { + var expiresAt *time.Time + if ttl > 0 { + exp := time.Now().Add(ttl) + expiresAt = &exp + } + res, err := k.dbPool.Exec(ctx, "UPDATE kvstore SET value=$1, expiresAt=$2 WHERE key=$3", + value, expiresAt, k.prefixKey(key)) + if err != nil { + return err + } + if res.RowsAffected() == 0 { + _, _ = k.dbPool.Exec(ctx, "INSERT INTO kvstore (key, value, expiresAt) VALUES ($1, $2, $3)", + k.prefixKey(key), value, expiresAt) + } + return nil +} + +// SetPrefix implements [serviceapi.KvStore]. +func (k *kvStorePostgres) SetPrefix(prefix string) { + k.prefix = prefix +} + +var _ serviceapi.KvStore = (*kvStorePostgres)(nil) + +func Service(poolName, prefix string) *kvStorePostgres { + return &kvStorePostgres{ + dbPool: lokstra_registry.GetService[serviceapi.DbPool](poolName), + prefix: prefix, + } +} + +func ServiceFactory(config map[string]any) any { + poolName := utils.GetValueFromMap(config, "pool_name", "db_main") + prefix := utils.GetValueFromMap(config, "prefix", "") + return Service(poolName, prefix) +} + +func Register() { + lokstra_registry.RegisterServiceType(SERVICE_TYPE, ServiceFactory) +} diff --git a/services/kvstore_redis/module.go b/services/kvstore/kvstore_redis/module.go similarity index 72% rename from services/kvstore_redis/module.go rename to services/kvstore/kvstore_redis/module.go index ca19e887..3b776030 100644 --- a/services/kvstore_redis/module.go +++ b/services/kvstore/kvstore_redis/module.go @@ -3,6 +3,9 @@ package kvstore_redis import ( "context" "encoding/json" + "errors" + "fmt" + "sync" "time" "github.com/primadi/lokstra/common/utils" @@ -13,6 +16,13 @@ import ( const SERVICE_TYPE = "kvstore_redis" +var ErrKeyNotFound = errors.New("key not found") + +var ( + mu sync.Mutex + poolClient = make(map[Config]*redis.Client) +) + // Config represents the configuration for Redis-based KvStore service. type Config struct { Addr string `json:"addr" yaml:"addr"` // host:port address @@ -29,6 +39,16 @@ type kvStoreRedis struct { var _ serviceapi.KvStore = (*kvStoreRedis)(nil) +// GetPrefix implements [serviceapi.KvStore]. +func (k *kvStoreRedis) GetPrefix() string { + return k.prefix +} + +// SetPrefix implements [serviceapi.KvStore]. +func (k *kvStoreRedis) SetPrefix(prefix string) { + k.prefix = prefix +} + func (k *kvStoreRedis) prefixKey(key string) string { if k.prefix != "" { return k.prefix + ":" + key @@ -47,9 +67,15 @@ func (k *kvStoreRedis) Set(ctx context.Context, key string, value any, ttl time. func (k *kvStoreRedis) Get(ctx context.Context, key string, dest any) error { data, err := k.client.Get(ctx, k.prefixKey(key)).Bytes() if err != nil { - return err + if errors.Is(err, redis.Nil) { + return ErrKeyNotFound + } + return fmt.Errorf("redis get %q: %w", key, err) } - return json.Unmarshal(data, dest) + if err := json.Unmarshal(data, dest); err != nil { + return fmt.Errorf("unmarshal key %q: %w", key, err) + } + return nil } func (k *kvStoreRedis) Delete(ctx context.Context, key string) error { @@ -74,13 +100,14 @@ func (k *kvStoreRedis) Keys(ctx context.Context, pattern string) ([]string, erro return nil, err } - // Remove prefix from returned keys - if k.prefix != "" { - prefixLen := len(k.prefix) + 1 // +1 for the colon - for i, key := range keys { - if len(key) > prefixLen { - keys[i] = key[prefixLen:] - } + startItem := len(k.prefix) + if startItem > 0 { + startItem++ // to account for the colon + } + + for i, key := range keys { + if len(key) > startItem { + keys[i] = key[startItem:] } } @@ -95,12 +122,19 @@ func (k *kvStoreRedis) Shutdown() error { } func Service(cfg *Config) *kvStoreRedis { - client := redis.NewClient(&redis.Options{ - Addr: cfg.Addr, - Password: cfg.Password, - DB: cfg.DB, - PoolSize: cfg.PoolSize, - }) + mu.Lock() + client, exists := poolClient[*cfg] + if !exists { + client = redis.NewClient(&redis.Options{ + Addr: cfg.Addr, + Password: cfg.Password, + DB: cfg.DB, + PoolSize: cfg.PoolSize, + }) + poolClient[*cfg] = client + } + mu.Unlock() + return &kvStoreRedis{ client: client, prefix: cfg.Prefix, diff --git a/services/redis/module.go b/services/redis/module.go deleted file mode 100644 index efa8bdb7..00000000 --- a/services/redis/module.go +++ /dev/null @@ -1,56 +0,0 @@ -package redis - -import ( - "github.com/primadi/lokstra/common/utils" - "github.com/primadi/lokstra/lokstra_registry" - "github.com/redis/go-redis/v9" -) - -const SERVICE_TYPE = "redis" - -// Config represents the configuration for Redis client service. -type Config struct { - Addr string `json:"addr" yaml:"addr"` // host:port address - Password string `json:"password" yaml:"password"` // password - DB int `json:"db" yaml:"db"` // database number - PoolSize int `json:"pool_size" yaml:"pool_size"` -} - -type redisService struct { - client *redis.Client -} - -func (r *redisService) Client() *redis.Client { - return r.client -} - -func (r *redisService) Shutdown() error { - if r.client != nil { - return r.client.Close() - } - return nil -} - -func Service(cfg *Config) *redisService { - client := redis.NewClient(&redis.Options{ - Addr: cfg.Addr, - Password: cfg.Password, - DB: cfg.DB, - PoolSize: cfg.PoolSize, - }) - return &redisService{client: client} -} - -func ServiceFactory(params map[string]any) any { - cfg := &Config{ - Addr: utils.GetValueFromMap(params, "addr", "localhost:6379"), - Password: utils.GetValueFromMap(params, "password", ""), - DB: utils.GetValueFromMap(params, "db", 0), - PoolSize: utils.GetValueFromMap(params, "pool_size", 10), - } - return Service(cfg) -} - -func Register() { - lokstra_registry.RegisterServiceType(SERVICE_TYPE, ServiceFactory) -} diff --git a/services/register_all.go b/services/register_all.go index 68bbe5be..f887acb7 100644 --- a/services/register_all.go +++ b/services/register_all.go @@ -8,9 +8,9 @@ import ( "github.com/primadi/lokstra/services/dbpool_pg" "github.com/primadi/lokstra/services/email_smtp" - "github.com/primadi/lokstra/services/kvstore_redis" + "github.com/primadi/lokstra/services/kvstore/kvstore_inmemory" + "github.com/primadi/lokstra/services/kvstore/kvstore_redis" "github.com/primadi/lokstra/services/metrics_prometheus" - "github.com/primadi/lokstra/services/redis" "github.com/primadi/lokstra/services/sync_config_pg" ) @@ -18,8 +18,8 @@ import ( // Note: Auth services have been moved to github.com/primadi/lokstra-auth func RegisterAllServices() { // Core services - redis.Register() kvstore_redis.Register() + kvstore_inmemory.Register() metrics_prometheus.Register() dbpool_pg.Register() email_smtp.Register() diff --git a/services/sync_config_pg/config_test.yaml b/services/sync_config_pg/config_test.yaml index c35940f1..71253d67 100644 --- a/services/sync_config_pg/config_test.yaml +++ b/services/sync_config_pg/config_test.yaml @@ -1,6 +1,8 @@ # yaml-language-server: $schema=https://primadi.github.io/lokstra/schema/lokstra.schema.json -dbpool-definitions: +service-definitions: db_main: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} \ No newline at end of file + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} \ No newline at end of file diff --git a/services/sync_config_pg/module.go b/services/sync_config_pg/module.go index f4681d92..824bc608 100644 --- a/services/sync_config_pg/module.go +++ b/services/sync_config_pg/module.go @@ -13,7 +13,6 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/primadi/lokstra/common/json" "github.com/primadi/lokstra/common/utils" - "github.com/primadi/lokstra/core/deploy" "github.com/primadi/lokstra/lokstra_registry" "github.com/primadi/lokstra/serviceapi" "github.com/primadi/lokstra/services/dbpool_pg" @@ -58,25 +57,6 @@ type syncConfigPG struct { var _ serviceapi.SyncConfig = (*syncConfigPG)(nil) -func getDsnAndSchema(cfg *Config) (string, string) { - deployConfig := deploy.Global().GetDeployConfig() - if deployConfig == nil { - panic("sync_config_pg: deploy config not found") - } - - poolConfig, ok := deployConfig.DbPoolDefinitions[cfg.DbPoolName] - if !ok { - panic(fmt.Sprintf("sync_config_pg: named pool '%s' not found in config", cfg.DbPoolName)) - } - - schema := poolConfig.Schema - if schema == "" { - schema = "public" // Default schema - } - - return poolConfig.DSN, schema -} - // NewSyncConfigPG creates a new SyncConfig instance from config // It will automatically get the database pool and create listener connection // If an instance with the same configuration already exists, it will be reused (singleton per config) @@ -91,13 +71,17 @@ func NewSyncConfigPG(cfg *Config) (serviceapi.SyncConfig, error) { } instanceMu.Unlock() - dsn, schema := getDsnAndSchema(cfg) + dbPool := lokstra_registry.GetService[serviceapi.DbPool](cfg.DbPoolName) + pgxDbPool, ok := dbPool.(*dbpool_pg.PgxPostgresPool) + if !ok { + panic("cannot create dbPool " + cfg.DbPoolName) + } // Get DSN for listener connection var listenerDB *pgxpool.Pool if cfg.EnableNotification { var err error - listenerDB, err = pgxpool.New(context.Background(), dsn) + listenerDB, err = pgxpool.New(context.Background(), pgxDbPool.Dsn) if err != nil { return nil, fmt.Errorf("failed to create listener pool: %w", err) } @@ -106,15 +90,6 @@ func NewSyncConfigPG(cfg *Config) (serviceapi.SyncConfig, error) { // Create context with cancel for goroutine management ctx, cancel := context.WithCancel(context.Background()) - dbPool, err := dbpool_pg.NewPgxPostgresPool(cfg.DbPoolName, dsn, schema, nil) - if err != nil { - if listenerDB != nil { - listenerDB.Close() - } - cancel() - return nil, fmt.Errorf("failed to create db pool: %w", err) - } - service := &syncConfigPG{ cfg: cfg, dbPool: dbPool, @@ -576,26 +551,12 @@ func ServiceFactory(mapCfg map[string]any) any { EnableNotification: utils.GetValueFromMap(mapCfg, "enable_notification", true), } - svc, err := Service(cfg) - if err != nil { - panic(fmt.Sprintf("failed to create sync_config_pg service: %v", err)) + dbPool := lokstra_registry.GetService[serviceapi.DbPool](cfg.DbPoolName) + pgxDbPool, ok := dbPool.(*dbpool_pg.PgxPostgresPool) + if !ok { + panic("cannot create dbPool " + cfg.DbPoolName) } - return svc -} - -//go:embed sync_config.sql -var sync_config_sql string - -// Register registers the SyncConfig service type -func Register(dbPoolName string, heartBeatInterval, reconnectInterval time.Duration) { - // get dsn and schema from dbPoolName, read config from deploy.Global() - dsn, schema := getDsnAndSchema(&Config{DbPoolName: dbPoolName}) - // check is sync_config table on the schema, if not create it using sync_config_sql - dbPool, err := dbpool_pg.NewPgxPostgresPool(dbPoolName, dsn, schema, nil) - if err != nil { - panic(fmt.Sprintf("failed to create connection pool for sync_config_pg registration: %v", err)) - } ctx := context.Background() // acquire a connection conn, err := dbPool.Acquire(ctx) @@ -604,7 +565,7 @@ func Register(dbPoolName string, heartBeatInterval, reconnectInterval time.Durat } defer conn.Release() // check if table exists - exists, err := conn.IsExists(ctx, "SELECT to_regclass($1)", fmt.Sprintf("%s.sync_config", schema)) + exists, err := conn.IsExists(ctx, "SELECT to_regclass($1)", fmt.Sprintf("%s.sync_config", pgxDbPool.Schema)) if err != nil { panic(fmt.Sprintf("failed to check sync_config table existence: %v", err)) } @@ -616,6 +577,19 @@ func Register(dbPoolName string, heartBeatInterval, reconnectInterval time.Durat } } + svc, err := Service(cfg) + if err != nil { + panic(fmt.Sprintf("failed to create sync_config_pg service: %v", err)) + } + + return svc +} + +//go:embed sync_config.sql +var sync_config_sql string + +// Register registers the SyncConfig service type +func Register(dbPoolName string, heartBeatInterval, reconnectInterval time.Duration) { lokstra_registry.RegisterServiceType(SERVICE_TYPE, ServiceFactory) SetDefaultSyncConfigPG(dbPoolName, heartBeatInterval, reconnectInterval) } diff --git a/services/sync_config_pg/test/config.yaml b/services/sync_config_pg/test/config.yaml index c1470b7d..b323cad4 100644 --- a/services/sync_config_pg/test/config.yaml +++ b/services/sync_config_pg/test/config.yaml @@ -7,10 +7,12 @@ configs: # heartbeat_interval: ${DBPOOL_MANAGER_HEARTBEAT_INTERVAL:5} # in minutes # reconnect_interval: ${DBPOOL_MANAGER_RECONNECT_INTERVAL:5} # in seconds -dbpool-definitions: +service-definitions: db_main: - dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} + type: dbpool_pg + config: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db2} + schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} deployments: monolith: diff --git a/services/sync_config_pg/test/main.go b/services/sync_config_pg/test/main.go index 9ab9fc66..6ec1ec95 100644 --- a/services/sync_config_pg/test/main.go +++ b/services/sync_config_pg/test/main.go @@ -8,10 +8,13 @@ import ( "github.com/primadi/lokstra/core/deploy/loader" "github.com/primadi/lokstra/lokstra_init" "github.com/primadi/lokstra/lokstra_registry" + "github.com/primadi/lokstra/services/dbpool_pg" "github.com/primadi/lokstra/services/sync_config_pg" ) func main() { + dbpool_pg.Register() + // 1. Bootstrap Lokstra framework lokstra_init.Bootstrap() @@ -20,11 +23,7 @@ func main() { panic(err) } - lokstra_init.UsePgxDbPoolManager(true) sync_config_pg.Register("db_main", 5*time.Minute, 5*time.Second) - if err := loader.LoadDbPoolDefsFromConfig(); err != nil { - panic(err) - } // 3. Register routers registerRouters() diff --git a/services/sync_config_pg/test/test.http b/services/sync_config_pg/test/test.http index a32869d2..d0f2d9f7 100644 --- a/services/sync_config_pg/test/test.http +++ b/services/sync_config_pg/test/test.http @@ -16,7 +16,7 @@ Content-Type: application/json { "key": "mykey", - "value": "myvalue2" + "value": "myvalue1" } ### Set Value (example 2)