[HUB-4300] Introduce Versioned Migration Flow

README.md

@@ -11,7 +11,7 @@ Prometheus instrumentation, HTTP client/server utilities, and more.
 - `pkg/log`: Structured logging, audit logs, HTTP request/response logging
 - `pkg/logging`: Global logger facade for convenience
 - `pkg/middlewares`: Auth, tenant, tracing, URL filter middlewares
-- `pkg/migrate`: Migration runner for PostgreSQL (SQL files)
+- `pkg/migrate`: Migration runner for PostgreSQL (SQL files). See [Documentation](docs/migration/README.md)
 - `pkg/prom`: Prometheus metrics utilities for HTTP client/server
 - `pkg/standard`: Opinionated server/gateway wiring
 - `pkg/ticket`: Lightweight JWT ticket verification/claims

docs/migration/README.md

@@ -0,0 +1,90 @@
+# Migration Approaches
+
+This repository supports **two migration flows** for services using `pkg/db`.
+Choose the versioned flow if you need per‑version interleaving, or the legacy
+flow for a single AutoMigrate pass.
+
+## 1) Versioned Migration Flow
+
+Use when you need **interleaving** per migration version.
+
+**API**
+- `WithVersionedMigrationFunc(func(*gorm.DB, uint) error)`
+- Optional: `WithMigrationVersion(version)`
+
+**Behavior**
+- Runs `setup.sql` once (idempotent).
+- Runs `fdw.up.sql` / `fdw.down.sql` once (optional).
+- For each version from current+1 to target:
+  1. Versioned before script (optional)
+  2. `AutoMigrate(version)` (service implementation)
+  3. Versioned after script (optional)
+  4. Record version **after** the full sequence completes
+
+**Notes**
+- Missing before/after scripts are skipped.
+- The version bump represents completion of before + AutoMigrate + after.
+- **Recording a version** means updating the `migrations` table used by
+  `golang-migrate` with the new version number.
+- **Supported naming (versioned flow only):**
+  - Before: `{version}_{name}.before.sql` **or** `{version}_{name}.before.up.sql`
+  - After: `{version}_{name}.after.sql` **or** `{version}_{name}.after.up.sql`
+
+**Diagram**
+```mermaid
+flowchart TB
+  %% Versioned Migration Logic
+
+  B0["Start migrations"] --> B1["setup.sql (optional, idempotent)"]
+  B1 --> B2["fdw.up.sql (optional)"]
+  B2 --> V1["For each version v = current+1 .. target"]
+  V1 --> V2["{version}_{name}.before.sql or .before.up.sql (optional)"]
+  V2 --> V3["AutoMigrate(version)"]
+  V3 --> V4["{version}_{name}.after.sql or .after.up.sql (optional)"]
+  V4 --> V5["Record version v (migrations table)"]
+  V5 -.-> V1
+  V5 --> B3["fdw.down.sql (optional)"]
+```
+
+## 2) Legacy Migration Flow
+
+Use when you want a single AutoMigrate call and minimal changes to existing
+services.
+
+**API**
+- `WithMigrationFunc(func(*gorm.DB) error)`
+- Optional: `WithMigrationVersion(version)`
+
+**Behavior**
+- Runs **one** AutoMigrate (latest models).
+- Runs `setup.sql` (idempotent), then `fdw.up.sql` (optional).
+- Runs numbered SQL migrations via `golang-migrate`:
+  - `{version}_{name}.up.sql` / `{version}_{name}.down.sql`
+- Runs `fdw.down.sql` (optional).
+
+**Diagram**
+```mermaid
+flowchart TB
+  %% Legacy Migration Logic
+
+  A0["Start migrations"] --> A1["AutoMigrate (once, latest models)"]
+  A1 --> A2["setup.sql (optional, idempotent)"]
+  A2 --> A3["fdw.up.sql (optional)"]
+  A3 --> L1["For each version v = current+1 .. target"]
+  L1 --> L2["{version}_{name}.up.sql"]
+  L2 -.-> L1
+  L2 --> A4["fdw.down.sql (optional)"]
+```
+
+## File Naming Summary
+
+Versioned flow:
+- Before: `{version}_{name}.before.sql` or `{version}_{name}.before.up.sql`
+- After: `{version}_{name}.after.sql` or `{version}_{name}.after.up.sql`
+
+Legacy flow:
+- SQL migrations: `{version}_{name}.up.sql`
+
+Shared:
+- Setup: `setup.sql`
+- FDW: `fdw.up.sql`, `fdw.down.sql`

docs/migration/migration-legacy-example-v5-to-v7.mmd

@@ -0,0 +1,10 @@
+flowchart TB
+  %% Legacy Example: Current v5 -> Target v7
+
+  A0["Current DB version: 5"] --> A1["Target version: 7"]
+  A1 --> A2["AutoMigrate (once, latest models)"]
+  A2 --> A3["setup.sql (optional, idempotent)"]
+  A3 --> A4["fdw.up.sql (optional)"]
+  A4 --> A5["006_fix_constraint.up.sql"]
+  A5 --> A6["007_add_feature.up.sql"]
+  A6 --> A7["fdw.down.sql (optional)"]

docs/migration/migration-legacy.mmd

@@ -0,0 +1,10 @@
+flowchart TB
+  %% Legacy Migration Logic (as on main)
+
+  A0["Start migrations"] --> A1["AutoMigrate (once, latest models)"]
+  A1 --> A2["setup.sql (optional, idempotent)"]
+  A2 --> A3["fdw.up.sql (optional)"]
+  A3 --> L1["For each version v = current+1 .. target"]
+  L1 --> L2["{version}_{name}.up.sql"]
+  L2 -.-> L1
+  L2 --> A4["fdw.down.sql (optional)"]

docs/migration/migration-versioned-example-v5-to-v7.mmd

@@ -0,0 +1,18 @@
+flowchart TB
+  %% Versioned Example: Current v5 -> Target v7
+
+  B0["Current DB version: 5"] --> B1["Target version: 7"]
+  B1 --> B2["setup.sql (optional, idempotent)"]
+  B2 --> B3["fdw.up.sql (optional)"]
+
+  B3 --> V1["006_fix_constraint.before.sql (optional)"]
+  V1 --> V2["AutoMigrate(6)"]
+  V2 --> V3["006_fix_constraint.after.sql (optional)"]
+  V3 --> V3b["Record version 6"]
+
+  V3b --> V4["007_add_feature.before.sql (optional)"]
+  V4 --> V5["AutoMigrate(7)"]
+  V5 --> V6["007_add_feature.after.sql (optional)"]
+  V6 --> V6b["Record version 7"]
+
+  V6b --> B4["fdw.down.sql (optional)"]

docs/migration/migration-versioned.mmd

@@ -0,0 +1,12 @@
+flowchart TB
+  %% Versioned Migration Logic (WithVersionedMigrationFunc)
+
+  B0["Start migrations"] --> B1["setup.sql (optional, idempotent)"]
+  B1 --> B2["fdw.up.sql (optional)"]
+  B2 --> V1["For each version v = current+1 .. target"]
+  V1 --> V2["{version}_{name}.before.sql or .before.up.sql (optional)"]
+  V2 --> V3["AutoMigrate(version)"]
+  V3 --> V4["{version}_{name}.after.sql or .after.up.sql (optional)"]
+  V4 --> V5["Record version v"]
+  V5 -.-> V1
+  V5 --> B3["fdw.down.sql (optional)"]

pkg/db/gorm.go

@@ -2,6 +2,8 @@ package db
 
 import (
 	"context"
+	"database/sql"
+	stderrors "errors"
 	"time"
 
 	"github.com/pkg/errors"
@@ -59,7 +61,7 @@ func Initialize(runCtx context.Context, opts *ConnectionOptions) <-chan struct{}
 			defer logging.LogInfof("database connection closed")
 		}()
 
-		err = runMigration(conn, opts.MigrationFunc, opts.MigrationVersion, opts.MigrationStartFromZero)
+		err = runMigration(conn, opts.MigrationFunc, opts.VersionedMigrationFunc, opts.MigrationVersion, opts.MigrationStartFromZero)
 		if err != nil {
 			if opts.MigrationHaltOnError {
 				logging.LogErrorf(err, "database migration failed - aborting")
@@ -149,28 +151,180 @@ func retryExponential(runCtx context.Context, attempts uint, waitPeriod time.Dur
 }
 
 // runMigration Executes Migrations on the database
-func runMigration(conn *gorm.DB, migFn MigrationFunc, migrationVersion uint, startFromZero bool) error {
+func runMigration(
+	conn *gorm.DB,
+	legacyFn MigrationFunc,
+	versionedFn VersionedMigrationFunc,
+	migrationVersion uint,
+	startFromZero bool,
+) error {
 	if conn == nil {
 		logging.LogErrorf(ErrDBConnection, "MigrateDB() - db handle is nil")
 		return ErrDBConnection
 	}
-	// Run GORM automigrations as supplied by service
-	err := migFn(conn)
+	if legacyFn != nil && versionedFn != nil {
+		return errors.New("both MigrationFunc (legacy) and VersionedMigrationFunc are set; please configure only one migration flow")
+	}
+	if migrationVersion == 0 {
+		// No SQL migrations; run whichever automigration function is provided.
+		if versionedFn != nil {
+			return versionedFn(conn, 0)
+		}
+		if legacyFn != nil {
+			return legacyFn(conn)
+		}
+		return nil
+	}
+
+	// Prefer explicit versioned flow when configured.
+	if versionedFn != nil {
+		return runMigrationVersioned(conn, versionedFn, migrationVersion, startFromZero)
+	}
+
+	sqlDB, err := conn.DB()
 	if err != nil {
+		logging.LogErrorf(err, "error getting sql DB")
 		return err
 	}
 
+	return runMigrationLegacy(sqlDB, conn, legacyFn, migrationVersion, startFromZero)
+}
+
+func runMigrationLegacy(sqlDB *sql.DB, conn *gorm.DB, legacyFn MigrationFunc, migrationVersion uint, startFromZero bool) error {
+	// Preserve legacy behavior: AutoMigrate once, then SQL migrations via golang-migrate.
+	if legacyFn != nil {
+		if err := legacyFn(conn); err != nil {
+			return err
+		}
+	}
+
+	if migrationVersion == 0 {
+		return nil
+	}
+	migration := migrate.NewMigration(sqlDB, migrationsSource, migrationsTable, logging.Logger())
+	return migration.MigrateDB(context.Background(), migrationVersion, startFromZero)
+}
+
+func runMigrationVersioned(conn *gorm.DB, migFn VersionedMigrationFunc, migrationVersion uint, startFromZero bool) error {
 	sqlDB, err := conn.DB()
 	if err != nil {
 		logging.LogErrorf(err, "error getting sql DB")
 		return err
 	}
 
-	// Run manual migrations defined in sql scripts if needed
-	if migrationVersion > 0 {
-		migration := migrate.NewMigration(sqlDB, migrationsSource, migrationsTable, logging.Logger())
-		err = migration.MigrateDB(context.Background(), migrationVersion, startFromZero)
+	ctx := context.Background()
+	migration := migrate.NewMigration(sqlDB, migrationsSource, migrationsTable, logging.Logger())
+
+	if err := migration.ExecuteSetup(ctx); err != nil {
+		return err
+	}
+	if err := migration.ExecuteFdwUp(ctx); err != nil {
+		return err
+	}
+	defer func() {
+		if err := migration.ExecuteFdwDown(ctx); err != nil {
+			logging.LogErrorf(err, "error executing fdw down script")
+		}
+	}()
+
+	return runMigrationVersions(ctx, conn, migration, migFn, migrationVersion, startFromZero)
+}
+
+func runMigrationVersions(
+	ctx context.Context,
+	conn *gorm.DB,
+	migration *migrate.Migration,
+	migFn VersionedMigrationFunc,
+	migrationVersion uint,
+	startFromZero bool,
+) error {
+	mpg, cleanup, err := migration.MigrateInstanceForVersionTracking()
+	if err != nil {
+		return err
+	}
+	if cleanup != nil {
+		defer cleanup()
+	}
+
+	currentVersion, dirty, needsRecordTarget, err := currentMigrationVersion(mpg, migrationVersion, startFromZero)
+	if err != nil {
+		return err
+	}
+	if dirty {
+		return errors.Errorf("database migration is dirty at version %d", currentVersion)
+	}
+
+	// Legacy behavior: when the database has no version info and startFromZero is false,
+	// run AutoMigrate once, then record the target version without running per-version hooks.
+	if needsRecordTarget {
+		if migFn != nil {
+			if err := migFn(conn, migrationVersion); err != nil {
+				logging.LogErrorf(err, "error running auto migration for version %d", migrationVersion)
+				return err
+			}
+		}
+		if err := migrate.SetVersion(mpg, migrationVersion); err != nil {
+			logging.LogErrorf(err, "error setting migration version to %d", migrationVersion)
+			return err
+		}
+		return nil
+	}
+
+	for version := currentVersion + 1; version <= migrationVersion; version++ {
+		if err := applyMigrationVersion(ctx, conn, migration, migFn, mpg, version); err != nil {
+			return err
+		}
+		logging.LogInfof("migration for version %d executed successfully", version)
+	}
+
+	return nil
+}
+
+func currentMigrationVersion(mpg migrate.VersionSetter, migrationVersion uint, startFromZero bool) (uint, bool, bool, error) {
+	currentVersion, dirty, err := mpg.Version()
+	if err == nil {
+		return currentVersion, dirty, false, nil
+	}
+	if !stderrors.Is(err, migrate.ErrNilVersion) {
+		return 0, false, false, err
+	}
+	if startFromZero {
+		return 0, false, false, nil
+	}
+	// Caller should run a single AutoMigrate and then record the target version.
+	return migrationVersion, false, true, nil
+}
+
+func applyMigrationVersion(
+	ctx context.Context,
+	conn *gorm.DB,
+	migration *migrate.Migration,
+	migFn VersionedMigrationFunc,
+	mpg migrate.VersionSetter,
+	version uint,
+) error {
+	if _, err := migration.ExecuteBeforeUp(ctx, version); err != nil {
+		logging.LogErrorf(err, "error running before migration for version %d", version)
+		return err
+	}
+
+	if migFn != nil {
+		if err := migFn(conn, version); err != nil {
+			logging.LogErrorf(err, "error running auto migration for version %d", version)
+			return err
+		}
+	}
+
+	if _, err := migration.ExecuteAfterUp(ctx, version); err != nil {
+		logging.LogErrorf(err, "error running after migration for version %d", version)
+		return err
+	}
+
+	// Record version after full before/auto/after sequence.
+	if err := migrate.SetVersion(mpg, version); err != nil {
+		logging.LogErrorf(err, "error setting migration version to %d", version)
+		return err
 	}
 
-	return err
+	return nil
 }

pkg/db/gorm_migration_guard_test.go

@@ -0,0 +1,17 @@
+package db
+
+import (
+	"testing"
+
+	"gorm.io/gorm"
+)
+
+func TestRunMigrationRejectsBothLegacyAndVersionedFuncs(t *testing.T) {
+	legacyFn := func(_ *gorm.DB) error { return nil }
+	versionedFn := func(_ *gorm.DB, _ uint) error { return nil }
+
+	err := runMigration(&gorm.DB{}, legacyFn, versionedFn, 1, true)
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+}