From dc1b51c7e6927023cc7977de57b5bcbd7bbcc3d0 Mon Sep 17 00:00:00 2001 From: Primadi Setiawan Date: Mon, 22 Dec 2025 17:47:06 +0700 Subject: [PATCH] Fix ValidateYAMLConfig bugs, ctx.BeginTransaction --- .vscode/settings.json | 3 +- .../dbpool_crud.go | 2 +- .../dbpool_crud_test.go | 32 +- core/deploy/loader/loader.go | 32 +- core/deploy/loader/test/loader_test.go | 138 ++++++++ core/deploy/schema/schema.go | 16 +- core/request/context.go | 126 +++++++- docs/02-framework-guide/08-database-pools.md | 88 ++++- .../06-services/dbpool-manager.md | 17 +- docs/QUICK-REFERENCE.md | 117 +++++++ docs/examples/manual-transaction-control.md | 306 ++++++++++++++++++ lokstra_init/initialize.go | 1 + lokstra_init/migration.go | 7 +- lokstra_init/migration_runner/IMPROVEMENTS.md | 208 ------------ .../migration_runner/lokstra_core.sql | 62 ++++ lokstra_init/migration_runner/migration.go | 40 +-- lokstra_init/pgx_dbpool_manager.go | 1 + .../application/tenant_service.go | 6 +- .../application/zz_cache.lokstra.json | 2 +- .../03_tenant_management/config/config.yaml | 6 +- .../03_tenant_management/main.go | 6 +- .../{001_init.sql => 001_init.up.sql} | 0 .../repository/zz_cache.lokstra.json | 2 +- serviceapi/transaction.go | 6 + services/sync_config_pg/module.go | 31 ++ services/sync_config_pg/sync_config.sql | 56 ++++ 26 files changed, 1002 insertions(+), 309 deletions(-) rename common/{dbpool_manager => dbpool_crud}/dbpool_crud.go (99%) rename common/{dbpool_manager => dbpool_crud}/dbpool_crud_test.go (74%) create mode 100644 docs/examples/manual-transaction-control.md delete mode 100644 lokstra_init/migration_runner/IMPROVEMENTS.md create mode 100644 lokstra_init/migration_runner/lokstra_core.sql rename project_templates/02_app_framework/03_tenant_management/migrations/{001_init.sql => 001_init.up.sql} (100%) create mode 100644 services/sync_config_pg/sync_config.sql diff --git a/.vscode/settings.json b/.vscode/settings.json index f3ebed86..2aff1692 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -6,7 +6,8 @@ }, "[markdown]": { "files.exclude": {} - } + }, + "go.diagnostic.vulncheck": "Off" // "markdown.validate.ignoredLinks": ["**/.github/copilot-instructions.md"] // "yaml.schemas": { // "./schema/lokstra.json": [ diff --git a/common/dbpool_manager/dbpool_crud.go b/common/dbpool_crud/dbpool_crud.go similarity index 99% rename from common/dbpool_manager/dbpool_crud.go rename to common/dbpool_crud/dbpool_crud.go index e2f2453b..ba3e6931 100644 --- a/common/dbpool_manager/dbpool_crud.go +++ b/common/dbpool_crud/dbpool_crud.go @@ -1,4 +1,4 @@ -package dbpool_manager +package dbpool_crud import ( "context" diff --git a/common/dbpool_manager/dbpool_crud_test.go b/common/dbpool_crud/dbpool_crud_test.go similarity index 74% rename from common/dbpool_manager/dbpool_crud_test.go rename to common/dbpool_crud/dbpool_crud_test.go index 4a109175..c2c1a80c 100644 --- a/common/dbpool_manager/dbpool_crud_test.go +++ b/common/dbpool_crud/dbpool_crud_test.go @@ -1,10 +1,10 @@ -package dbpool_manager_test +package dbpool_crud_test import ( "context" "testing" - "github.com/primadi/lokstra/common/dbpool_manager" + "github.com/primadi/lokstra/common/dbpool_crud" "github.com/primadi/lokstra/lokstra_init" ) @@ -14,7 +14,7 @@ func ExampleAddDbPool() { lokstra_init.UsePgxDbPoolManager(true) // Enable distributed sync // Add new pool - err := dbpool_manager.AddDbPool(dbpool_manager.DbPoolConfig{ + err := dbpool_crud.AddDbPool(dbpool_crud.DbPoolConfig{ Name: "db-analytics", DSN: "postgres://user:pass@localhost:5432/analytics", Schema: "public", @@ -25,7 +25,7 @@ func ExampleAddDbPool() { } // Pool is now available across all servers (if using distributed sync) - conn, _ := dbpool_manager.AcquireDbConn(context.Background(), "db-analytics") + conn, _ := dbpool_crud.AcquireDbConn(context.Background(), "db-analytics") defer conn.Release() // Use connection... @@ -33,7 +33,7 @@ func ExampleAddDbPool() { // Example: Update existing pool configuration func ExampleUpdateDbPool() { - err := dbpool_manager.UpdateDbPool(dbpool_manager.DbPoolConfig{ + err := dbpool_crud.UpdateDbPool(dbpool_crud.DbPoolConfig{ Name: "db-main", DSN: "postgres://user:pass@new-host:5432/main", Schema: "public", @@ -49,7 +49,7 @@ func ExampleUpdateDbPool() { // Example: Remove a pool func ExampleRemoveDbPool() { - err := dbpool_manager.RemoveDbPool("db-old-tenant") + err := dbpool_crud.RemoveDbPool("db-old-tenant") if err != nil { panic(err) } @@ -59,20 +59,20 @@ func ExampleRemoveDbPool() { // Example: List all pools func ExampleListDbPools() { - pools, err := dbpool_manager.ListDbPools() + pools, err := dbpool_crud.ListDbPools() if err != nil { panic(err) } for _, poolName := range pools { - info, _ := dbpool_manager.GetDbPoolInfo(poolName) + info, _ := dbpool_crud.GetDbPoolInfo(poolName) println("Pool:", info.Name, "Schema:", info.Schema) } } // Example: Get pool info func ExampleGetDbPoolInfo() { - info, err := dbpool_manager.GetDbPoolInfo("db-main") + info, err := dbpool_crud.GetDbPoolInfo("db-main") if err != nil { panic(err) } @@ -83,7 +83,7 @@ func ExampleGetDbPoolInfo() { // Example: Direct pool access func ExampleGetDbPool() { - pool, err := dbpool_manager.GetDbPool("db-main") + pool, err := dbpool_crud.GetDbPool("db-main") if err != nil { panic(err) } @@ -100,7 +100,7 @@ func TestDbPoolCRUD(t *testing.T) { lokstra_init.UsePgxDbPoolManager(false) // Use local sync for testing // Create - err := dbpool_manager.AddDbPool(dbpool_manager.DbPoolConfig{ + err := dbpool_crud.AddDbPool(dbpool_crud.DbPoolConfig{ Name: "test-pool", DSN: "postgres://localhost/test", Schema: "test_schema", @@ -110,7 +110,7 @@ func TestDbPoolCRUD(t *testing.T) { } // Read - info, err := dbpool_manager.GetDbPoolInfo("test-pool") + info, err := dbpool_crud.GetDbPoolInfo("test-pool") if err != nil { t.Fatalf("Failed to get pool info: %v", err) } @@ -122,7 +122,7 @@ func TestDbPoolCRUD(t *testing.T) { } // Update - err = dbpool_manager.UpdateDbPool(dbpool_manager.DbPoolConfig{ + err = dbpool_crud.UpdateDbPool(dbpool_crud.DbPoolConfig{ Name: "test-pool", DSN: "postgres://localhost/test2", Schema: "test_schema2", @@ -131,19 +131,19 @@ func TestDbPoolCRUD(t *testing.T) { t.Fatalf("Failed to update pool: %v", err) } - info, _ = dbpool_manager.GetDbPoolInfo("test-pool") + info, _ = dbpool_crud.GetDbPoolInfo("test-pool") if info.Schema != "test_schema2" { t.Errorf("Expected schema 'test_schema2', got '%s'", info.Schema) } // Delete - err = dbpool_manager.RemoveDbPool("test-pool") + err = dbpool_crud.RemoveDbPool("test-pool") if err != nil { t.Fatalf("Failed to remove pool: %v", err) } // Verify deletion - _, err = dbpool_manager.GetDbPoolInfo("test-pool") + _, err = dbpool_crud.GetDbPoolInfo("test-pool") if err == nil { t.Error("Expected error when getting deleted pool") } diff --git a/core/deploy/loader/loader.go b/core/deploy/loader/loader.go index 300f4f33..4c56f030 100644 --- a/core/deploy/loader/loader.go +++ b/core/deploy/loader/loader.go @@ -8,7 +8,6 @@ import ( "strings" "github.com/primadi/lokstra/common/utils" - "github.com/primadi/lokstra/core/deploy/loader/internal" "github.com/primadi/lokstra/core/deploy/loader/resolver" "github.com/primadi/lokstra/core/deploy/schema" "github.com/xeipuuv/gojsonschema" @@ -73,6 +72,16 @@ func loadConfig(paths ...string) (*schema.DeployConfig, error) { } } + // STEP 1.5: Validate merged config BEFORE any resolution + // This catches schema errors early before decode filters them out + mergedBytes, err := yaml.Marshal(merged) + if err != nil { + return nil, fmt.Errorf("failed to marshal merged config for validation: %w", err) + } + if err := ValidateConfigYAML(mergedBytes); err != nil { + return nil, fmt.Errorf("validation failed: %w", err) + } + // STEP 2: Normalize shorthand servers (must be before getting server key) normalizeShorthandServers(merged) @@ -122,11 +131,6 @@ func loadConfig(paths ...string) (*schema.DeployConfig, error) { // STEP 9: Normalize server definitions (convert helper fields to apps) normalizeServerDefinitions(&finalConfig) - // STEP 10: Validate final config - if err := ValidateConfig(&finalConfig); err != nil { - return nil, fmt.Errorf("validation failed: %w", err) - } - return &finalConfig, nil } @@ -248,15 +252,21 @@ func mergeMaps[T any](target, source map[string]*T) map[string]*T { return result } -// ValidateConfig validates a deployment configuration against JSON schema -func ValidateConfig(config *schema.DeployConfig) error { +// ValidateConfigYAML validates raw YAML bytes against JSON schema +// This must be called BEFORE YAML decode to catch invalid fields +func ValidateConfigYAML(yamlData []byte) error { // Load embedded schema from schema package schemaData := schema.GetSchemaBytes() schemaLoader := gojsonschema.NewBytesLoader(schemaData) - // Convert config to map for validation - configMap := internal.ConfigToMap(config) - documentLoader := gojsonschema.NewGoLoader(configMap) // Validate + // Parse YAML to generic map (don't use strict struct decode) + var configMap map[string]any + if err := yaml.Unmarshal(yamlData, &configMap); err != nil { + return fmt.Errorf("failed to parse YAML: %w", err) + } + + // Validate against schema + documentLoader := gojsonschema.NewGoLoader(configMap) result, err := gojsonschema.Validate(schemaLoader, documentLoader) if err != nil { return fmt.Errorf("validation error: %w", err) diff --git a/core/deploy/loader/test/loader_test.go b/core/deploy/loader/test/loader_test.go index 967a7cb6..480297cc 100644 --- a/core/deploy/loader/test/loader_test.go +++ b/core/deploy/loader/test/loader_test.go @@ -343,3 +343,141 @@ func TestSmartMerging(t *testing.T) { t.Errorf("expected helper published-services to be cleared, got %d items", len(apiServer.HelperPublishedServices)) } } + +func TestValidateConfigYAML_Valid(t *testing.T) { + validYAML := ` +configs: + app-name: test-app + db-host: localhost + +service-definitions: + user-service: + type: user-service-factory + depends-on: + - db-pool + +deployments: + development: + servers: + api: + base-url: http://localhost:8080 + apps: + - addr: ":8080" + published-services: + - user-service +` + + err := loader.ValidateConfigYAML([]byte(validYAML)) + if err != nil { + t.Errorf("valid YAML should pass validation: %v", err) + } +} + +func TestValidateConfigYAML_InvalidField(t *testing.T) { + invalidYAML := ` +configs: + app-name: test-app + +servise-definitions: + user-service: + type: user-service-factory +` + + err := loader.ValidateConfigYAML([]byte(invalidYAML)) + if err == nil { + t.Error("expected validation error for invalid field 'servise-definitions'") + } +} + +func TestValidateConfigYAML_MissingRequiredField(t *testing.T) { + // Service definition without 'type' field + invalidYAML := ` +service-definitions: + user-service: + depends-on: + - db-pool +` + + err := loader.ValidateConfigYAML([]byte(invalidYAML)) + if err == nil { + t.Error("expected validation error for missing 'type' field") + } +} + +func TestValidateConfigYAML_InvalidType(t *testing.T) { + // addr should be string, not number + invalidYAML := ` +deployments: + development: + servers: + api: + base-url: http://localhost:8080 + apps: + - addr: 8080 + published-services: + - user-service +` + + err := loader.ValidateConfigYAML([]byte(invalidYAML)) + if err == nil { + t.Error("expected validation error for wrong type (number instead of string)") + } +} + +func TestValidateConfigYAML_InvalidDependsOn(t *testing.T) { + // depends-on with invalid service name pattern + invalidYAML := ` +service-definitions: + user-service: + type: user-service-factory + depends-on: + - DB_POOL +` + + err := loader.ValidateConfigYAML([]byte(invalidYAML)) + if err == nil { + t.Error("expected validation error for invalid depends-on pattern (uppercase with underscore)") + } +} + +func TestValidateConfigYAML_MalformedYAML(t *testing.T) { + malformedYAML := ` +configs: + app-name: test-app + invalid yaml: + - missing proper indentation + - broken structure +` + + err := loader.ValidateConfigYAML([]byte(malformedYAML)) + if err == nil { + t.Error("expected error for malformed YAML") + } +} + +func TestValidateConfigYAML_AdditionalProperties(t *testing.T) { + // Unknown top-level field + invalidYAML := ` +configs: + app-name: test-app + +unknown-field: + something: value +` + + err := loader.ValidateConfigYAML([]byte(invalidYAML)) + if err == nil { + t.Error("expected validation error for unknown top-level field") + } +} + +func TestValidateConfigYAML_EmptyConfig(t *testing.T) { + emptyYAML := `{}` + + err := loader.ValidateConfigYAML([]byte(emptyYAML)) + // Empty config might be valid depending on schema requirements + // Adjust assertion based on actual schema rules + if err != nil { + t.Logf("Empty config validation: %v", err) + } +} diff --git a/core/deploy/schema/schema.go b/core/deploy/schema/schema.go index bf970ea8..7bdabd57 100644 --- a/core/deploy/schema/schema.go +++ b/core/deploy/schema/schema.go @@ -130,18 +130,18 @@ type ConfigDef struct { // MiddlewareDef defines a middleware instance type MiddlewareDef struct { - Name string `yaml:"name"` - Type string `yaml:"type"` // Factory type - Config map[string]any `yaml:"config"` // Optional config + Name string `yaml:"name,omitempty" json:"name,omitempty"` // Optional: defaults to map key + Type string `yaml:"type" json:"type"` // Factory type + Config map[string]any `yaml:"config,omitempty" json:"config,omitempty"` // Optional config } // ServiceDef defines a service instance type ServiceDef struct { - Name string `yaml:"name"` - Type string `yaml:"type"` // Factory type - DependsOn []string `yaml:"depends-on"` // Dependencies (can be "paramName:serviceName") - Router *RouterDef `yaml:"router,omitempty"` // Embedded router definition (auto-generated router for this service) - Config map[string]any `yaml:"config"` // Optional config + Name string `yaml:"name,omitempty" json:"name,omitempty"` // Optional: defaults to map key + Type string `yaml:"type" json:"type"` // Factory type + DependsOn []string `yaml:"depends-on,omitempty" json:"depends-on,omitempty"` // Dependencies (can be "paramName:serviceName") + Router *RouterDef `yaml:"router,omitempty" json:"router,omitempty"` // Embedded router definition (auto-generated router for this service) + Config map[string]any `yaml:"config,omitempty" json:"config,omitempty"` // Optional config } // ReverseProxyDef defines a reverse proxy configuration diff --git a/core/request/context.go b/core/request/context.go index 596394af..71061256 100644 --- a/core/request/context.go +++ b/core/request/context.go @@ -2,12 +2,23 @@ package request import ( "context" + "fmt" "net/http" "github.com/primadi/lokstra/core/response" "github.com/primadi/lokstra/serviceapi" ) +// StatusCodeError represents an error state based on HTTP status code +// Used to trigger transaction rollback for non-2xx status codes +type StatusCodeError struct { + StatusCode int +} + +func (e *StatusCodeError) Error() string { + return fmt.Sprintf("HTTP status code %d indicates error", e.StatusCode) +} + type Context struct { // Embedding standard context for easy access context.Context @@ -28,6 +39,12 @@ type Context struct { handlers []HandlerFunc value map[string]any + + // Transaction finalizers to be called automatically in FinalizeResponse + // Map of poolName -> finalizer function + txFinalizers map[string]func(*error) + // Track order of transaction creation for proper LIFO finalization + txPoolOrder []string } func NewContext(w http.ResponseWriter, r *http.Request, handlers []HandlerFunc) *Context { @@ -59,19 +76,116 @@ func (c *Context) Next() error { } // Begins a transaction for the specified pool name -// Returns a finalize function to be deferred -// that accepts a pointer to error to determine commit/rollback - -func (c *Context) BeginTransaction(poolName string) func(*error) { +// The transaction will be automatically finalized (commit/rollback) when FinalizeResponse is called +// No need to defer the returned function anymore - it's handled automatically +func (c *Context) BeginTransaction(poolName string) { newCtx, finalizeCtx := serviceapi.BeginTransaction(c, poolName) c.Context = newCtx // Update embedded context with transaction context - return finalizeCtx + + // Initialize map if needed + if c.txFinalizers == nil { + c.txFinalizers = make(map[string]func(*error)) + } + + // Store finalizer by pool name + c.txFinalizers[poolName] = finalizeCtx + c.txPoolOrder = append(c.txPoolOrder, poolName) +} + +// RollbackTransaction manually rolls back a specific transaction +// Use this for edge cases like dry-run or testing where you want to return 200 OK but rollback changes +// +// ⚠️ WARNING: Only call this in the SAME handler that called BeginTransaction. +// Do NOT use manual commit/rollback if your handler calls other handlers (nested calls), +// as it can cause transaction state inconsistency. +// +// Safe usage: +// - Dry-run operations (single handler, no nested calls) +// - Testing/validation (single handler, no nested calls) +// +// Unsafe usage: +// - Handler A commits manually, then calls Handler B (nested) +// - Service layer that might be called by multiple handlers +func (c *Context) RollbackTransaction(poolName string) { + if finalizer, exists := c.txFinalizers[poolName]; exists { + // Force rollback by passing a non-nil error + rollbackErr := error(&StatusCodeError{StatusCode: http.StatusInternalServerError}) + finalizer(&rollbackErr) + // Remove from finalizers to prevent double-finalization + delete(c.txFinalizers, poolName) + // Also remove from order tracking + c.removeTxFromOrder(poolName) + } +} + +// CommitTransaction manually commits a specific transaction +// Use this for edge cases where you need explicit control over transaction lifecycle +// +// ⚠️ WARNING: Only call this in the SAME handler that called BeginTransaction. +// Do NOT use manual commit/rollback if your handler calls other handlers (nested calls), +// as it can cause transaction state inconsistency. +// +// Safe usage: +// - Conditional commit based on business logic (single handler) +// - Partial success handling (single handler, no nested calls) +// +// Unsafe usage: +// - Handler A commits manually, then calls Handler B (nested) +// - Service layer that might be called by multiple handlers +func (c *Context) CommitTransaction(poolName string) { + if finalizer, exists := c.txFinalizers[poolName]; exists { + // Force commit by passing nil error + var commitErr error + finalizer(&commitErr) + // Remove from finalizers to prevent double-finalization + delete(c.txFinalizers, poolName) + // Also remove from order tracking + c.removeTxFromOrder(poolName) + } +} + +// removeTxFromOrder removes a pool name from the order tracking +func (c *Context) removeTxFromOrder(poolName string) { + for i, name := range c.txPoolOrder { + if name == poolName { + c.txPoolOrder = append(c.txPoolOrder[:i], c.txPoolOrder[i+1:]...) + break + } + } } // Finalizes the response, writing status code and body if not already written +// Also automatically finalizes all transactions (commit on success, rollback on error) func (c *Context) FinalizeResponse(err error) { + // IMPORTANT: Always finalize transactions, even if response was manually written + // Use defer to ensure transactions are finalized in all code paths + defer func() { + // Finalize all remaining transactions (in reverse order - LIFO) + // Skip transactions that were manually committed/rolled back + // Determine if transaction should commit or rollback based on: + // 1. Explicit error from handler + // 2. Status code >= 400 (client/server errors) + statusCode := c.StatusCode() + var txErr error + if err != nil { + txErr = err + } else if statusCode >= http.StatusBadRequest { + // Create error from status code to trigger rollback + txErr = &StatusCodeError{StatusCode: statusCode} + } + + // Call finalizers in reverse order (LIFO) for remaining transactions + for i := len(c.txPoolOrder) - 1; i >= 0; i-- { + poolName := c.txPoolOrder[i] + if finalizer, exists := c.txFinalizers[poolName]; exists { + finalizer(&txErr) + } + } + }() + if c.W.ManualWritten() { - // User already wrote directly to ResponseWriter, do nothing + // User already wrote directly to ResponseWriter, skip response writing + // but still finalize transactions (via defer above) return } diff --git a/docs/02-framework-guide/08-database-pools.md b/docs/02-framework-guide/08-database-pools.md index 24b55b2c..6a060a42 100644 --- a/docs/02-framework-guide/08-database-pools.md +++ b/docs/02-framework-guide/08-database-pools.md @@ -82,7 +82,7 @@ type DbTx interface { ## Transaction via Context (Recommended) -The recommended way to handle transactions is using `ctx.BeginTransaction()` from `request.Context`: +The recommended way to handle transactions is using `ctx.BeginTransaction(poolName)` from `request.Context`. Transactions are automatically finalized (commit/rollback) when the response is written. ## Setup Database Pools @@ -442,10 +442,11 @@ lokstra_registry.InitAndRunServer() ### Basic Transaction Usage ```go -func (s *UserService) CreateUser(ctx *request.Context, user *User) (err error) { +func (s *UserService) CreateUser(ctx *request.Context, user *User) error { // Begin transaction for pool named "main-db" // Transaction will be created automatically on first DB operation - defer ctx.BeginTransaction("main-db")(&err) + // Will auto-commit on success or rollback on error in FinalizeResponse + ctx.BeginTransaction("main-db") // All database operations using this ctx will join the same transaction if err := s.userRepo.Create(ctx, user); err != nil { @@ -465,12 +466,11 @@ func (s *UserService) CreateUser(ctx *request.Context, user *User) (err error) { ```go // @Route "POST /" func (s *TenantService) CreateTenant(ctx *request.Context, - req *domain.CreateTenantRequest) (result *domain.Tenant, err error) { + req *domain.CreateTenantRequest) (*domain.Tenant, error) { // Begin transaction for pool "db_auth" // All subsequent DB operations using ctx will join this transaction - finishTx := ctx.BeginTransaction("db_auth") - defer finishTx(&err) + ctx.BeginTransaction("db_auth") // These operations will automatically join the transaction existing, err := s.TenantStore.GetByName(ctx, req.Name) @@ -491,14 +491,74 @@ func (s *TenantService) CreateTenant(ctx *request.Context, 2. **Lazy Creation**: Transaction is created only when first DB operation occurs (not immediately) 3. **Pool Name Based**: Transaction is tracked by pool name, not pool instance 4. **Auto-Join**: All DB operations using the same context automatically join the transaction -5. **Auto-Commit/Rollback**: - - Returns `nil` → Transaction commits automatically - - Returns error → Transaction rolls back automatically +5. **Auto-Finalization**: Happens automatically in `FinalizeResponse()`: + - Returns `nil` + status < 400 → **Commit** + - Returns error OR status >= 400 → **Rollback** **Key Points:** - No need to manually inject DbPool - just use the pool name - Transaction is created lazily (zero overhead if no DB operations) - All operations in the same context automatically share the transaction +- Rollback happens on **any** error status (400+), even if handler returns nil error + +**Example - Status-based rollback:** +```go +func (s *Service) Create(ctx *request.Context, req *Request) error { + ctx.BeginTransaction("db") + + s.repo.Create(ctx, data) + + // Even though error is nil, transaction will rollback because status = 400 + return ctx.Api.BadRequest("Validation failed") // ← Triggers rollback! +} +``` + +### Manual Transaction Control + +For advanced scenarios (dry-run, testing, conditional commit): + +```go +// Dry-run: Execute operations but don't persist +func (s *Service) DryRun(ctx *request.Context, req *Request) error { + ctx.BeginTransaction("main-db") + + // Execute all operations + result, err := s.repo.Create(ctx, req) + if err != nil { + return err + } + + // Manual rollback - changes discarded + ctx.RollbackTransaction("main-db") + + // Return 200 OK with results + return ctx.Api.Ok(map[string]any{ + "message": "Dry run successful", + "result": result, + }) +} + +// Conditional commit +func (s *Service) BatchProcess(ctx *request.Context, items []Item) error { + ctx.BeginTransaction("main-db") + + successCount := s.processItems(ctx, items) + + if successCount < len(items) * 0.8 { + ctx.RollbackTransaction("main-db") // Below threshold + return ctx.Api.Ok(map[string]any{"status": "rolled_back"}) + } + + ctx.CommitTransaction("main-db") // Above threshold + return ctx.Api.Ok(map[string]any{"status": "committed"}) +} +``` + +**Available Methods:** +- `ctx.RollbackTransaction(poolName)` - Force rollback +- `ctx.CommitTransaction(poolName)` - Force commit + +**See:** [Manual Transaction Control Examples](../examples/manual-transaction-control.md) ### Using Without Request Context (Service Layer) @@ -527,12 +587,12 @@ func (s *UserService) DoWork(ctx context.Context) (err error) { Each pool name has its own transaction context: ```go -func (s *Service) Transfer(ctx *request.Context, amount float64) (err error) { +func (s *Service) Transfer(ctx *request.Context, amount float64) error { // Transaction for main database - defer ctx.BeginTransaction("main-db")(&err) + ctx.BeginTransaction("main-db") // Transaction for analytics database (separate) - defer ctx.BeginTransaction("analytics-db")(&err) + ctx.BeginTransaction("analytics-db") // Operations on main-db join main-db transaction s.mainRepo.Deduct(ctx, amount) @@ -551,8 +611,8 @@ Sometimes you need to execute operations outside of a transaction (e.g., audit l ```go import "github.com/primadi/lokstra/serviceapi" -func (s *Service) CreateWithAudit(ctx *request.Context, data *Data) (err error) { - defer ctx.BeginTransaction("main-db")(&err) +func (s *Service) CreateWithAudit(ctx *request.Context, data *Data) error { + ctx.BeginTransaction("main-db") // This joins the transaction if err := s.repo.Create(ctx, data); err != nil { diff --git a/docs/03-api-reference/06-services/dbpool-manager.md b/docs/03-api-reference/06-services/dbpool-manager.md index ee6f2253..ffe26b7d 100644 --- a/docs/03-api-reference/06-services/dbpool-manager.md +++ b/docs/03-api-reference/06-services/dbpool-manager.md @@ -484,22 +484,23 @@ func (s *Service) DoWork(ctx context.Context) (err error) { ### Request Context Integration -In HTTP handlers, use `request.Context.BeginTransaction()`: +In HTTP handlers, use `request.Context.BeginTransaction()`. Transactions are automatically finalized when the response is written: ```go -func (s *UserService) CreateUser(ctx *request.Context, user *User) (err error) { - defer ctx.BeginTransaction("main-db")(&err) +func (s *UserService) CreateUser(ctx *request.Context, user *User) error { + // Mark context as needing transaction - no defer needed! + ctx.BeginTransaction("main-db") // All operations join the transaction if err := s.userRepo.Create(ctx, user); err != nil { - return err // Auto rollback + return err // Auto rollback in FinalizeResponse } if err := s.auditRepo.Log(ctx, "user_created", user.ID); err != nil { - return err // Auto rollback + return err // Auto rollback in FinalizeResponse } - return nil // Auto commit + return nil // Auto commit in FinalizeResponse } ``` @@ -519,10 +520,10 @@ type TxContext struct { **How It Works:** -1. **Marking Phase**: `BeginTransaction` adds `TxContext` marker to context +1. **Marking Phase**: `ctx.BeginTransaction()` adds `TxContext` marker to context 2. **Lazy Creation**: First DB operation checks context, finds marker, creates transaction 3. **Auto-Join**: Subsequent DB operations detect existing transaction and reuse it -4. **Finalization**: Deferred function commits (on success) or rolls back (on error) +4. **Auto-Finalization**: When `FinalizeResponse` is called, transactions are automatically committed (on success) or rolled back (on error) **Transaction Flow:** ``` diff --git a/docs/QUICK-REFERENCE.md b/docs/QUICK-REFERENCE.md index 85fdbd5b..0bf2150a 100644 --- a/docs/QUICK-REFERENCE.md +++ b/docs/QUICK-REFERENCE.md @@ -116,6 +116,123 @@ func(params *SearchParams) ([]Result, error) { return results, nil } --- +## Database & Transactions + +### Inject DB Pool + +```go +// @Service "user-repository" +type UserRepository struct { + // @Inject "main-db" + DB serviceapi.DbPool +} + +func (r *UserRepository) GetUser(ctx context.Context, id string) (*User, error) { + conn, err := r.DB.Acquire(ctx) + if err != nil { + return nil, err + } + defer conn.Release() + + var user User + err = conn.QueryRow(ctx, + "SELECT id, name, email FROM users WHERE id = $1", id, + ).Scan(&user.ID, &user.Name, &user.Email) + + return &user, err +} +``` + +### Transactions (Auto-Finalized) + +```go +// ✅ RECOMMENDED: Automatic management +func (s *Service) Create(ctx *request.Context, req *Request) error { + ctx.BeginTransaction("main-db") // No defer needed! + + s.repo1.Create(ctx, data1) + s.repo2.Create(ctx, data2) + + return ctx.Api.Ok(data) // Auto-commit (status 200) +} + +// ❌ Error or 400+ status → Auto-rollback +func (s *Service) Update(ctx *request.Context, req *Request) error { + ctx.BeginTransaction("main-db") + + if err := s.repo.Update(ctx, data); err != nil { + return err // ← Rollback + } + + return ctx.Api.BadRequest("Invalid") // ← Also rollback (status 400) +} +``` + +**Auto-finalization rules:** +- **Commit:** `nil` error + status < 400 +- **Rollback:** Error returned OR status >= 400 + +### Manual Transaction Control + +```go +// Dry-run: Execute but rollback +func (s *Service) DryRun(ctx *request.Context) error { + ctx.BeginTransaction("main-db") + + result, _ := s.repo.Create(ctx, data) + + ctx.RollbackTransaction("main-db") // ← Force rollback + return ctx.Api.Ok(result) // 200 OK, no data saved +} + +// Conditional commit +func (s *Service) Batch(ctx *request.Context, items []Item) error { + ctx.BeginTransaction("main-db") + + successCount := s.process(ctx, items) + + if successCount < len(items)*0.8 { + ctx.RollbackTransaction("main-db") // < 80% success + } else { + ctx.CommitTransaction("main-db") // >= 80% success + } + + return ctx.Api.Ok(map[string]any{"success": successCount}) +} +``` + +### Multiple Pools + +```go +func (s *Service) CrossDatabase(ctx *request.Context) error { + ctx.BeginTransaction("db-auth") + ctx.BeginTransaction("db-tenant") // Independent + + s.authRepo.Create(ctx, authData) // Uses db-auth tx + s.tenantRepo.Create(ctx, tenantData) // Uses db-tenant tx + + return ctx.Api.Ok("done") // Both auto-commit +} +``` + +### Service Layer (Without Request Context) + +```go +import "github.com/primadi/lokstra/serviceapi" + +func (s *Service) DoWork(ctx context.Context) (err error) { + ctx, finish := serviceapi.BeginTransaction(ctx, "main-db") + defer finish(&err) // ← Manual defer needed + + s.repo1.Create(ctx, ...) + s.repo2.Update(ctx, ...) + + return nil // Auto-commit +} +``` + +--- + ## Service Patterns (Recommended: Use Annotations) ### Annotation-Based Service (Recommended) diff --git a/docs/examples/manual-transaction-control.md b/docs/examples/manual-transaction-control.md new file mode 100644 index 00000000..0c22f46d --- /dev/null +++ b/docs/examples/manual-transaction-control.md @@ -0,0 +1,306 @@ +# Manual Transaction Control + +This guide covers advanced scenarios where you need explicit control over transaction commit/rollback behavior. + +## Automatic Transaction Management (Default) + +By default, transactions are automatically managed based on response status: + +```go +// @Route "POST /" +func (s *UserService) CreateUser(ctx *request.Context, req *CreateUserRequest) error { + ctx.BeginTransaction("main-db") + + // These operations join the transaction + user, err := s.repo.Create(ctx, req) + if err != nil { + return err // ✅ Auto rollback + } + + return ctx.Api.Ok(user) // ✅ Auto commit +} +``` + +**Auto-rollback triggers:** +- Handler returns `error` +- Response status code >= 400 (e.g., `ctx.Api.BadRequest()`) + +**Auto-commit triggers:** +- Handler returns `nil` error +- Response status code < 400 (e.g., `ctx.Api.Ok()`) + +## Manual Control + +### 1. Dry-Run Mode (Rollback with 200 OK) + +```go +// @Route "POST /dry-run" +func (s *UserService) DryRunCreate(ctx *request.Context, req *CreateUserRequest) error { + ctx.BeginTransaction("main-db") + + // Execute all operations normally + user, err := s.repo.Create(ctx, req) + if err != nil { + return err + } + + s.auditRepo.Log(ctx, "user_created", user.ID) + + // ✅ Manual rollback - data not persisted + ctx.RollbackTransaction("main-db") + + // Return 200 OK with simulation results + return ctx.Api.Ok(map[string]any{ + "message": "Dry run successful", + "user": user, + "note": "No data was actually saved", + }) +} +``` + +### 2. Testing & Validation + +```go +// @Route "POST /validate" +func (s *UserService) ValidateOperation(ctx *request.Context, req *ComplexRequest) error { + ctx.BeginTransaction("main-db") + + // Test complex multi-step operation + result1, err := s.repo.Step1(ctx, req) + if err != nil { + return err + } + + result2, err := s.repo.Step2(ctx, result1) + if err != nil { + return err + } + + // Validate business rules + if !s.validateBusinessRules(result2) { + // ✅ Manual rollback with custom response + ctx.RollbackTransaction("main-db") + return ctx.Api.Ok(map[string]any{ + "valid": false, + "message": "Business rules not satisfied", + "details": result2, + }) + } + + // ✅ Manual commit + ctx.CommitTransaction("main-db") + + return ctx.Api.Ok(map[string]any{ + "valid": true, + "message": "All validations passed", + }) +} +``` + +### 3. Partial Success Handling + +```go +// @Route "POST /batch" +func (s *UserService) BatchCreate(ctx *request.Context, req *BatchRequest) error { + ctx.BeginTransaction("main-db") + + var succeeded []string + var failed []string + + for _, item := range req.Items { + if err := s.repo.Create(ctx, item); err != nil { + failed = append(failed, item.ID) + } else { + succeeded = append(succeeded, item.ID) + } + } + + // Business logic: commit only if at least 80% succeeded + successRate := float64(len(succeeded)) / float64(len(req.Items)) + + if successRate < 0.8 { + // ✅ Rollback all + ctx.RollbackTransaction("main-db") + return ctx.Api.Ok(map[string]any{ + "status": "rolled_back", + "reason": "Success rate below threshold", + "succeeded": succeeded, + "failed": failed, + }) + } + + // ✅ Commit partial success, auto commit because return 200 + // no need to do this: + // ctx.CommitTransaction("main-db") + + return ctx.Api.Ok(map[string]any{ + "status": "committed", + "succeeded": succeeded, + "failed": failed, + }) +} +``` + +### 4. Multiple Pools with Selective Control + +```go +// @Route "POST /cross-db" +func (s *UserService) CrossDatabaseOperation(ctx *request.Context, req *Request) error { + // Start transactions on both pools + ctx.BeginTransaction("db_auth") + ctx.BeginTransaction("db_tenant") + + // Auth DB operations + user, err := s.authRepo.Create(ctx, req.User) + if err != nil { + return err // Both auto-rollback + } + + // Tenant DB operations + tenant, err := s.tenantRepo.Create(ctx, req.Tenant) + if err != nil { + // ✅ Manually rollback auth DB first + ctx.RollbackTransaction("db_auth") + return err // Tenant DB also auto-rollback + } + + // Business rule: Commit auth but rollback tenant + if req.DryRunTenant { + ctx.RollbackTransaction("db_tenant") // ✅ Tenant rolled back + // db_auth will auto-commit (200 OK) + } + + return ctx.Api.Ok(map[string]any{ + "user": user, + "tenant": tenant, + }) +} +``` + +## Best Practices + +### ⚠️ CRITICAL: Avoid Manual Control with Nested Calls + +**DO NOT** use manual commit/rollback if your handler calls other handlers: + +```go +// ❌ DANGEROUS: Manual commit with nested handler call +func (s *ServiceA) Create(ctx *request.Context) error { + ctx.BeginTransaction("db") + + s.repo.Create(ctx, data) + + // ❌ BAD: Manual commit here + ctx.CommitTransaction("db") + + // ❌ DANGER: Calling another handler creates nested transaction + return s.serviceB.Process(ctx) // Transaction state corrupted! +} +``` + +**Why dangerous?** +- ServiceA commits transaction +- ServiceB calls `BeginTransaction()` on same context +- Transaction context still exists but already committed +- Can cause data corruption or inconsistent state + +**Solution: Let auto-finalization handle it** + +```go +// ✅ SAFE: Auto-finalization +func (s *ServiceA) Create(ctx *request.Context) error { + ctx.BeginTransaction("db") + + s.repo.Create(ctx, data) + + // ✅ GOOD: No manual commit, let it auto-finalize + return s.serviceB.Process(ctx) // Safe! +} +``` + +### ✅ DO + +```go +// Use manual control for explicit edge cases +func (s *Service) DryRun(ctx *request.Context) error { + ctx.BeginTransaction("db") + // ... operations + ctx.RollbackTransaction("db") // Clear intent + return ctx.Api.Ok(result) +} +``` + +### ❌ DON'T + +```go +// Don't use manual control for normal error handling +func (s *Service) Create(ctx *request.Context) error { + ctx.BeginTransaction("db") + + user, err := s.repo.Create(ctx, data) + if err != nil { + ctx.RollbackTransaction("db") // ❌ Unnecessary + return err // Auto-rollback already happens + } + + ctx.CommitTransaction("db") // ❌ Unnecessary + return ctx.Api.Ok(user) // Auto-commit already happens +} +``` + +**Rule of thumb:** Only use manual control when automatic behavior doesn't match your use case **AND** you're in a single, isolated handler with no nested calls. + +### Safe vs Unsafe Manual Control + +| Scenario | Safe? | Reason | +|----------|-------|--------| +| **Single handler, dry-run** | ✅ Safe | No nested calls, transaction fully controlled | +| **Single handler, validation** | ✅ Safe | No nested calls, clear lifecycle | +| **Handler → calls another handler** | ❌ **UNSAFE** | Nested context, transaction state corrupted | +| **Service layer shared by handlers** | ❌ **UNSAFE** | Called from multiple contexts | +| **Middleware with manual control** | ❌ **UNSAFE** | Applied to many handlers | +| **Batch processing (no nested calls)** | ✅ Safe | Single handler, isolated logic | + +### When in Doubt: Use Auto-Finalization + +If you're unsure whether manual control is safe: +- **DON'T** use manual commit/rollback +- Let auto-finalization handle it +- Manual control is for **rare edge cases only** + +## Transaction Lifecycle + +``` +BeginTransaction(poolName) + ↓ +Store finalizer in map[poolName] + ↓ +Handler executes + ↓ +┌─────────────────────────────────┐ +│ Manual Control (Optional) │ +│ - RollbackTransaction(poolName) │ +│ - CommitTransaction(poolName) │ +└─────────────────────────────────┘ + ↓ +FinalizeResponse() + ↓ +Auto-finalize remaining transactions +(skip already manually finalized) +``` + +## Common Patterns + +| Use Case | Pattern | Example | +|----------|---------|---------| +| **Normal CRUD** | Auto (default) | Return error → rollback, return success → commit | +| **Dry-Run** | Manual rollback + 200 OK | `ctx.RollbackTransaction()` then `ctx.Api.Ok()` | +| **Validation Test** | Manual based on validation | Rollback if invalid, commit if valid | +| **Batch Processing** | Conditional manual control | Commit if threshold met, rollback otherwise | +| **Audit Logs** | Split transaction | Use `serviceapi.WithoutTransaction()` for audit | + +## See Also + +- [Transaction Guide](../02-framework-guide/08-database-pools.md#transaction-management) +- [Database Pools](../03-api-reference/06-services/dbpool-manager.md) +- [Request Context API](../03-api-reference/02-core/request-context.md) diff --git a/lokstra_init/initialize.go b/lokstra_init/initialize.go index 2bbfbd66..1d009e7e 100644 --- a/lokstra_init/initialize.go +++ b/lokstra_init/initialize.go @@ -83,6 +83,7 @@ func BootstrapAndRun(opts ...InitializeOption) error { // Initialize lokstra framework with given config func BootstrapAndRunWithConfig(cfg *InitializeConfig) error { + // Validate config if cfg.EnablePgxSyncMap { if len(cfg.PgxSyncMapDbPoolName) == 0 { return cfg.returnError(fmt.Errorf("PgxSyncMapDbPoolName must be set when UsePgxSyncMap is true")) diff --git a/lokstra_init/migration.go b/lokstra_init/migration.go index 00c23342..39957d61 100644 --- a/lokstra_init/migration.go +++ b/lokstra_init/migration.go @@ -6,6 +6,7 @@ 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" @@ -106,7 +107,11 @@ func CheckDbMigration(cfg *MigrationConfig) error { // Apply final defaults if still empty if cfg.DbPoolName == "" { - cfg.DbPoolName = "main-db" + 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 } if cfg.SchemaTable == "" { cfg.SchemaTable = "schema_migrations" diff --git a/lokstra_init/migration_runner/IMPROVEMENTS.md b/lokstra_init/migration_runner/IMPROVEMENTS.md deleted file mode 100644 index 06efb426..00000000 --- a/lokstra_init/migration_runner/IMPROVEMENTS.md +++ /dev/null @@ -1,208 +0,0 @@ -# Migration Runner Improvements - -## ✅ Implemented Improvements - -### 1. **Version Conflict Detection** - -**Problem:** Multiple developers could create migrations with the same version number but different descriptions, causing silent data loss. - -**Solution:** Added strict validation during `Load()`: - -```go -// Example conflict scenario: -// Developer A: 001_create_users.up.sql -// Developer B: 001_create_orders.up.sql - -// Error output: -❌ Migration conflict detected for version 001: - Found: 001_create_users.*.sql - Found: 001_create_orders.*.sql - → Same version cannot have different descriptions! - → Please rename one of the migrations to use a different version number. -``` - -**Benefits:** -- ✅ Detects conflicts immediately -- ✅ Clear error message with resolution steps -- ✅ Prevents silent migration loss -- ✅ Works for both UP and DOWN migrations - ---- - -### 2. **Better Error Messages** - -**Before:** -``` -Error: no DOWN SQL for migration 2 -``` - -**After:** -``` -❌ Cannot rollback migration 002_add_indexes - → Missing file: 002_add_indexes.down.sql - → Create the DOWN migration file or manually remove this version from schema_migrations table -``` - -**Improvements:** -- ✅ Shows version with leading zeros (002 vs 2) -- ✅ Includes migration description -- ✅ Shows exact filename needed -- ✅ Provides actionable solutions -- ✅ Uses emoji for better readability - ---- - -### 3. **Create Command (Auto-versioning)** - -**Feature:** Auto-generate migration file pairs with proper versioning. - -```bash -# Command -go run main.go --cmd=create --name="create_users_table" - -# Output -✅ Created migration files: - → 001_create_users_table.up.sql - → 001_create_users_table.down.sql - -📝 Next steps: - 1. Edit the migration files with your SQL - 2. Run: go run main.go --cmd=up -``` - -**How it works:** -1. Scans existing migrations to find highest version -2. Auto-increments to next version (001 → 002 → 003...) -3. Creates both UP and DOWN files with templates -4. Validates migration name format (snake_case) -5. Prevents accidental overwrites - -**Benefits:** -- ✅ No manual version numbering -- ✅ No version conflicts (uses next available) -- ✅ Helpful SQL templates included -- ✅ Creates both files atomically -- ✅ Validates naming conventions - ---- - -## Additional Validations - -### Duplicate File Detection - -```go -// Prevents duplicate UP or DOWN files for same version -❌ Duplicate UP migration file for version 001_create_users - → Only one .up.sql file allowed per version -``` - -### Name Format Validation - -```bash -# Invalid names are rejected -go run main.go --cmd=create --name="Create Users" - -❌ Invalid migration name: 'Create Users' - → Use snake_case with lowercase letters, numbers, and underscores only - → Example: create_users_table, add_email_index -``` - ---- - -## Migration File Templates - -**UP Migration Template:** -```sql --- Migration: create_users_table --- Created: auto-generated --- Version: 001 - --- Add your UP migration SQL here --- Example: --- CREATE TABLE users ( --- id SERIAL PRIMARY KEY, --- name VARCHAR(255) NOT NULL, --- email VARCHAR(255) UNIQUE NOT NULL, --- created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP --- ); -``` - -**DOWN Migration Template:** -```sql --- Migration: create_users_table (ROLLBACK) --- Created: auto-generated --- Version: 001 - --- Add your DOWN migration SQL here (reverse of UP migration) --- Example: --- DROP TABLE IF EXISTS users; -``` - ---- - -## Usage Examples - -### Typical Workflow - -```bash -# 1. Create new migration -go run main.go --cmd=create --name="create_users_table" - -# 2. Edit generated files (add your SQL) -# Edit: migrations/001_create_users_table.up.sql -# Edit: migrations/001_create_users_table.down.sql - -# 3. Run migration -go run main.go --cmd=up - -# 4. Check status -go run main.go --cmd=status - -# 5. Rollback if needed -go run main.go --cmd=down -``` - -### Team Collaboration Scenario - -**Developer A (morning):** -```bash -go run main.go --cmd=create --name="create_users_table" -# Creates: 001_create_users_table.*.sql -git add migrations/ -git commit -m "Add user table migration" -``` - -**Developer B (afternoon, before pulling):** -```bash -go run main.go --cmd=create --name="create_orders_table" -# Creates: 001_create_orders_table.*.sql (same version!) -git pull # Gets Developer A's changes -``` - -**Developer B (after pull):** -```bash -go run main.go --cmd=up -# Error: Version conflict detected! -# Fix: Rename to version 002 -mv migrations/001_create_orders_table.up.sql migrations/002_create_orders_table.up.sql -mv migrations/001_create_orders_table.down.sql migrations/002_create_orders_table.down.sql - -go run main.go --cmd=up -# Success! -``` - ---- - -## Summary - -| Feature | Before | After | -|---------|--------|-------| -| **Conflict Detection** | ❌ Silent overwrite | ✅ Error with clear message | -| **Error Messages** | ❌ Cryptic | ✅ Actionable with solutions | -| **File Creation** | ⚠️ Manual | ✅ Auto-generated with `create` | -| **Version Numbering** | ⚠️ Manual | ✅ Auto-incremented | -| **Duplicate Detection** | ❌ None | ✅ Enforced | -| **Name Validation** | ❌ None | ✅ Snake_case required | -| **Templates** | ❌ None | ✅ Helpful SQL templates | - -All improvements maintain backward compatibility with existing migration files! 🎉 diff --git a/lokstra_init/migration_runner/lokstra_core.sql b/lokstra_init/migration_runner/lokstra_core.sql new file mode 100644 index 00000000..d1912dfb --- /dev/null +++ b/lokstra_init/migration_runner/lokstra_core.sql @@ -0,0 +1,62 @@ +CREATE SCHEMA IF NOT EXISTS lokstra_core; + +SET SEARCH_PATH TO lokstra_core; + +CREATE TABLE IF NOT EXISTS %s ( + version INTEGER PRIMARY KEY, + applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + description TEXT +); + +-- Create sync_config table +CREATE TABLE IF NOT EXISTS sync_config ( + key VARCHAR(255) PRIMARY KEY, + value JSONB NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +-- Create index on updated_at for faster queries +CREATE INDEX IF NOT EXISTS idx_sync_config_updated_at ON sync_config (updated_at); + +-- Create trigger function for automatic NOTIFY on changes +CREATE OR REPLACE FUNCTION sync_config_notify() +RETURNS TRIGGER AS $$ +DECLARE + notification JSON; +BEGIN + -- Build notification payload + IF (TG_OP = 'DELETE') THEN + notification = json_build_object( + 'action', 'delete', + 'key', OLD.key, + 'value', null + ); + ELSE + notification = json_build_object( + 'action', lower(TG_OP), + 'key', NEW.key, + 'value', NEW.value + ); + END IF; + + -- Send notification to default channel + -- Channel name can be customized per deployment + PERFORM pg_notify('config_changes', notification::text); + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Create trigger for INSERT, UPDATE, DELETE +CREATE OR REPLACE TRIGGER sync_config_notify_trigger + AFTER INSERT OR UPDATE OR DELETE ON sync_config + FOR EACH ROW + EXECUTE FUNCTION sync_config_notify(); + +-- Add comments +COMMENT ON TABLE sync_config IS 'Configuration key-value store with real-time sync support'; +COMMENT ON COLUMN sync_config.key IS 'Configuration key (unique identifier)'; +COMMENT ON COLUMN sync_config.value IS 'Configuration value stored as JSONB'; +COMMENT ON COLUMN sync_config.updated_at IS 'Timestamp of last update'; +COMMENT ON FUNCTION sync_config_notify() IS 'Trigger function to send NOTIFY on config changes'; +COMMENT ON TRIGGER sync_config_notify_trigger ON sync_config IS 'Automatically sends pg_notify when config changes'; diff --git a/lokstra_init/migration_runner/migration.go b/lokstra_init/migration_runner/migration.go index 26e16424..720cbb63 100644 --- a/lokstra_init/migration_runner/migration.go +++ b/lokstra_init/migration_runner/migration.go @@ -2,6 +2,7 @@ package migration_runner import ( "context" + _ "embed" "fmt" "os" "path/filepath" @@ -13,6 +14,9 @@ import ( "github.com/primadi/lokstra/serviceapi" ) +//go:embed lokstra_core.sql +var lokstra_core_sql string + // Migration represents a single database migration type Migration struct { Version int @@ -292,23 +296,9 @@ func (r *Runner) loadFromSubfolders(entries []os.DirEntry, minVersion int, migra return nil } -// ensureSchemaTable creates the schema_migrations table if it doesn't exist -func (r *Runner) ensureSchemaTable(ctx context.Context) error { - conn, err := r.dbPool.Acquire(ctx) - if err != nil { - return fmt.Errorf("failed to acquire connection: %w", err) - } - defer conn.Release() - - createTableSQL := fmt.Sprintf(` - CREATE TABLE IF NOT EXISTS %s ( - version INTEGER PRIMARY KEY, - applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - description TEXT - ) - `, r.schemaTable) - - _, err = conn.Exec(ctx, createTableSQL) +// EnsureSchemaTable creates the schema_migrations table if it doesn't exist +func (r *Runner) EnsureSchemaTable(ctx context.Context) error { + _, err := r.dbPool.Exec(ctx, fmt.Sprintf(lokstra_core_sql, r.schemaTable)) if err != nil { return fmt.Errorf("failed to create schema table: %w", err) } @@ -345,7 +335,7 @@ func (r *Runner) getAppliedVersions(ctx context.Context) (map[int]bool, error) { // Up runs all pending UP migrations func (r *Runner) Up(ctx context.Context) error { - if err := r.ensureSchemaTable(ctx); err != nil { + if err := r.EnsureSchemaTable(ctx); err != nil { return err } @@ -382,16 +372,10 @@ func (r *Runner) Up(ctx context.Context) error { // getCurrentVersion returns the highest applied migration version // Returns 0 if no migrations have been applied func (r *Runner) getCurrentVersion(ctx context.Context) (int, error) { - conn, err := r.dbPool.Acquire(ctx) - if err != nil { - return 0, fmt.Errorf("failed to acquire connection: %w", err) - } - defer conn.Release() - query := fmt.Sprintf("SELECT COALESCE(MAX(version), 0) FROM %s", r.schemaTable) var maxVersion int - err = conn.QueryRow(ctx, query).Scan(&maxVersion) + err := r.dbPool.QueryRow(ctx, query).Scan(&maxVersion) if err != nil { return 0, fmt.Errorf("failed to get current version: %w", err) } @@ -444,7 +428,7 @@ func (r *Runner) DownN(ctx context.Context, n int) error { return err } - if err := r.ensureSchemaTable(ctx); err != nil { + if err := r.EnsureSchemaTable(ctx); err != nil { return err } @@ -524,7 +508,7 @@ func (r *Runner) runDownMigration(ctx context.Context, m *Migration) error { // Version returns the current migration version func (r *Runner) Version(ctx context.Context) (int, error) { - if err := r.ensureSchemaTable(ctx); err != nil { + if err := r.EnsureSchemaTable(ctx); err != nil { return 0, err } @@ -550,7 +534,7 @@ func (r *Runner) Status(ctx context.Context) (string, error) { return "", err } - if err := r.ensureSchemaTable(ctx); err != nil { + if err := r.EnsureSchemaTable(ctx); err != nil { return "", err } diff --git a/lokstra_init/pgx_dbpool_manager.go b/lokstra_init/pgx_dbpool_manager.go index 3a8bd7ee..c50823f7 100644 --- a/lokstra_init/pgx_dbpool_manager.go +++ b/lokstra_init/pgx_dbpool_manager.go @@ -15,6 +15,7 @@ func UsePgxDbPoolManager(useSync bool) { } if useSync { + // ensure lokstra_core_sql migration is applied pm = dbpool_manager.NewPgxSyncDbPoolManager() logger.LogDebug("[Lokstra] DbPoolManager initialized with distributed sync") } else { diff --git a/project_templates/02_app_framework/03_tenant_management/application/tenant_service.go b/project_templates/02_app_framework/03_tenant_management/application/tenant_service.go index 7b61613b..29eee641 100644 --- a/project_templates/02_app_framework/03_tenant_management/application/tenant_service.go +++ b/project_templates/02_app_framework/03_tenant_management/application/tenant_service.go @@ -20,11 +20,11 @@ type TenantService struct { // @Route "POST /" func (s *TenantService) CreateTenant(ctx *request.Context, - req *domain.CreateTenantRequest) (result *domain.Tenant, err error) { + req *domain.CreateTenantRequest) (*domain.Tenant, error) { // Begin transaction - lazy created on first database operation - finishTx := ctx.BeginTransaction("db_auth") - defer finishTx(&err) + // Auto-finalized (commit/rollback) when response is written + ctx.BeginTransaction("db_auth") // Check if tenant name already exists existing, err := s.TenantStore.GetByName(ctx, req.Name) 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 8941cd45..0aca8195 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 @@ -12,6 +12,6 @@ "generated_mod_time": "2025-12-18T16:07:08.4971722+07:00" } }, - "updated_at": "2025-12-18T16:07:14.4438215+07:00", + "updated_at": "2025-12-19T21:23:40.8371545+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 744ccddf..48299b69 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 @@ -6,9 +6,13 @@ configs: user-store: postgres-user-store dbpool-definitions: + db_core: + dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} + schema: ${GLOBAL_DB_SCHEMA:lokstra_core} db_auth: dsn: ${GLOBAL_DB_DSN:postgres://postgres:adm1n@localhost:5432/lokstra_db} - schema: ${GLOBAL_DB_SCHEMA:lokstra_auth} + schema: ${GLOBAL_DB_SCHEMA:lokstra_core} + 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 6b144562..ce1d380f 100644 --- a/project_templates/02_app_framework/03_tenant_management/main.go +++ b/project_templates/02_app_framework/03_tenant_management/main.go @@ -10,5 +10,9 @@ func main() { recovery.Register() request_logger.Register() - lokstra_init.BootstrapAndRun() + lokstra_init.BootstrapAndRun( + lokstra_init.WithPgSyncMap(true, "db_core"), + lokstra_init.WithDbPoolAutoSync(true), + // lokstra_init.WithDbMigrations(true, "migrations"), + ) } diff --git a/project_templates/02_app_framework/03_tenant_management/migrations/001_init.sql b/project_templates/02_app_framework/03_tenant_management/migrations/001_init.up.sql similarity index 100% rename from project_templates/02_app_framework/03_tenant_management/migrations/001_init.sql rename to project_templates/02_app_framework/03_tenant_management/migrations/001_init.up.sql 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 fa02e17d..02d2bea8 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-18T16:07:14.4443577+07:00", + "updated_at": "2025-12-19T21:23:40.8377344+07:00", "generated_checksum": "f3e29afd20b1546d51b0f487725dc26863d7cf2c011aba9071035b6308ae8956" } \ No newline at end of file diff --git a/serviceapi/transaction.go b/serviceapi/transaction.go index 32cc2181..f7c675aa 100644 --- a/serviceapi/transaction.go +++ b/serviceapi/transaction.go @@ -126,6 +126,12 @@ func finalizeTx(ctx context.Context, txCtx *TxContext, err *error) { if txCtx.Conn != nil { _ = txCtx.Conn.Release() } + + // Reset state to prevent reuse issues + txCtx.Tx = nil + txCtx.Conn = nil + txCtx.committed = false + txCtx.rolledBack = false } // WithoutTransaction creates a child context that explicitly ignores any parent transactions. diff --git a/services/sync_config_pg/module.go b/services/sync_config_pg/module.go index 9e724451..f4681d92 100644 --- a/services/sync_config_pg/module.go +++ b/services/sync_config_pg/module.go @@ -2,6 +2,7 @@ package sync_config_pg import ( "context" + _ "embed" "fmt" "hash/crc32" "sort" @@ -583,8 +584,38 @@ func ServiceFactory(mapCfg map[string]any) any { 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) + if err != nil { + panic(fmt.Sprintf("failed to acquire connection for sync_config_pg registration: %v", err)) + } + defer conn.Release() + // check if table exists + exists, err := conn.IsExists(ctx, "SELECT to_regclass($1)", fmt.Sprintf("%s.sync_config", schema)) + if err != nil { + panic(fmt.Sprintf("failed to check sync_config table existence: %v", err)) + } + if !exists { + // create table + _, err = conn.Exec(ctx, sync_config_sql) + if err != nil { + panic(fmt.Sprintf("failed to create sync_config table: %v", err)) + } + } + lokstra_registry.RegisterServiceType(SERVICE_TYPE, ServiceFactory) SetDefaultSyncConfigPG(dbPoolName, heartBeatInterval, reconnectInterval) } diff --git a/services/sync_config_pg/sync_config.sql b/services/sync_config_pg/sync_config.sql new file mode 100644 index 00000000..7cf470e8 --- /dev/null +++ b/services/sync_config_pg/sync_config.sql @@ -0,0 +1,56 @@ +CREATE SCHEMA IF NOT EXISTS lokstra_core; + +SET SEARCH_PATH TO lokstra_core; + +-- Create sync_config table +CREATE TABLE IF NOT EXISTS sync_config ( + key VARCHAR(255) PRIMARY KEY, + value JSONB NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +-- Create index on updated_at for faster queries +CREATE INDEX IF NOT EXISTS idx_sync_config_updated_at ON sync_config (updated_at); + +-- Create trigger function for automatic NOTIFY on changes +CREATE OR REPLACE FUNCTION sync_config_notify() +RETURNS TRIGGER AS $$ +DECLARE + notification JSON; +BEGIN + -- Build notification payload + IF (TG_OP = 'DELETE') THEN + notification = json_build_object( + 'action', 'delete', + 'key', OLD.key, + 'value', null + ); + ELSE + notification = json_build_object( + 'action', lower(TG_OP), + 'key', NEW.key, + 'value', NEW.value + ); + END IF; + + -- Send notification to default channel + -- Channel name can be customized per deployment + PERFORM pg_notify('config_changes', notification::text); + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Create trigger for INSERT, UPDATE, DELETE +CREATE OR REPLACE TRIGGER sync_config_notify_trigger + AFTER INSERT OR UPDATE OR DELETE ON sync_config + FOR EACH ROW + EXECUTE FUNCTION sync_config_notify(); + +-- Add comments +COMMENT ON TABLE sync_config IS 'Configuration key-value store with real-time sync support'; +COMMENT ON COLUMN sync_config.key IS 'Configuration key (unique identifier)'; +COMMENT ON COLUMN sync_config.value IS 'Configuration value stored as JSONB'; +COMMENT ON COLUMN sync_config.updated_at IS 'Timestamp of last update'; +COMMENT ON FUNCTION sync_config_notify() IS 'Trigger function to send NOTIFY on config changes'; +COMMENT ON TRIGGER sync_config_notify_trigger ON sync_config IS 'Automatically sends pg_notify when config changes';