From 6c1f94e57ccc5c144488dabddd3c3427e5573e53 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Tue, 29 Sep 2026 15:25:51 +0200 Subject: [PATCH] feat(11-01): add lagoon.OnDatabase and after-commit transactions - OnDatabase runs a callback once the database is published (now, or when lagoon.Publish runs), so GORM callbacks registered at Boot also install under serve, where Boot runs before the database is opened - Transaction runs AfterCommit callbacks in order after a successful commit; nested calls are savepoints whose callbacks drop with them - the lagoon:after_commit GORM callback flushes single-statement AfterCommit work after GORM's own commit; outside a transaction it runs immediately --- modules/lagoon/README.md | 31 ++++- modules/lagoon/connection.go | 16 ++- modules/lagoon/ondatabase.go | 95 +++++++++++++ modules/lagoon/ondatabase_test.go | 85 ++++++++++++ modules/lagoon/transaction.go | 182 +++++++++++++++++++++++++ modules/lagoon/transaction_test.go | 208 +++++++++++++++++++++++++++++ 6 files changed, 612 insertions(+), 5 deletions(-) create mode 100644 modules/lagoon/ondatabase.go create mode 100644 modules/lagoon/ondatabase_test.go create mode 100644 modules/lagoon/transaction.go create mode 100644 modules/lagoon/transaction_test.go diff --git a/modules/lagoon/README.md b/modules/lagoon/README.md index 896dbef..f625d42 100644 --- a/modules/lagoon/README.md +++ b/modules/lagoon/README.md @@ -13,6 +13,8 @@ Postgres data layer: the shared GORM connection, per-plugin migrations, model he ## Features - One shared pool: `lagoon.Open`, `lagoon.Use` and `lagoon.OpenFromApp` return a `*sql.DB` and a `*gorm.DB` built on that same pool; `lagoon.Publish` makes both available on the `backpack.App`. +- Database-ready hooks: `lagoon.OnDatabase` runs a callback with the pool and GORM handle as soon as the database is published, immediately when it already is, otherwise when `lagoon.Publish` runs. Plugins register GORM callbacks through it from Boot, which runs before the `serve` command publishes the database. +- After-commit work: `lagoon.Transaction` runs a function in a transaction and then the callbacks registered with `lagoon.AfterCommit`, in order, only after the commit succeeds; a nested `lagoon.Transaction` is a savepoint whose callbacks are dropped with it when it fails. A single-statement write for which GORM opens its own transaction runs its `lagoon.AfterCommit` callbacks from the `lagoon:after_commit` GORM callback (`lagoon.AfterCommitCallback`) once GORM commits, and never when the write fails. Outside both, including inside a plain GORM `Transaction`, `lagoon.AfterCommit` runs the callback immediately. A panicking callback is logged and never turns a committed write into an error. - Database check at connect time: `lagoon.CheckLocale` refuses a database whose default collation is not the ICU `pl-PL` locale, so ordering matches the database default without per-query `COLLATE`. - Per-plugin migrations: `lagoon.Migrate` runs the framework's `system_files` set (`attach.Migrations`), backend admin identity set (`lagoon.BackendAdminMigrations`) and job-queue set (`lagoon.QueueMigrations`: River's schema pinned at `lagoon.RiverSchemaVersion`, then the `lagoon.JobsTable` record table, under the `lagoon.QueueHistoryID` history), then every `pact.HasMigrations` set in plugin activation order, each in its own `summer_migrations_` history table (`lagoon.HistoryTableName`). `lagoon.RollbackLast` and `lagoon.Status` cover rollback and history. - Mass assignment: `lagoon.Fill` copies only allow-listed keys onto a model by GORM column name and silently drops the rest, logging each dropped key once outside production. A `json.Number` (from a decoder using `UseNumber`) fills integer, unsigned and float fields. A value that does not fit its column (a fraction, an exponent or an overflow for an integer field, or a value of the wrong type) is a `lagoon.FillTypeError` naming the key, so a caller can answer it as a validation failure on that field. `lagoon.HasFillable` and `lagoon.HasHidden` are the Go forms of `$fillable` and `$hidden`. @@ -64,6 +66,29 @@ func createPost(ctx context.Context, gdb *gorm.DB, tr *phrasebook.Translator, in } ``` +A plugin registers its GORM callbacks from Boot through `lagoon.OnDatabase`, so they are installed whenever the database is published, and defers side effects until the write commits: + +```go +func (p *Plugin) Boot(app *backpack.App) error { + return lagoon.OnDatabase(app, func(_ *sql.DB, gdb *gorm.DB) error { + return gdb.Callback().Create().After("gorm:after_create").Register("acme:post_created", func(db *gorm.DB) { + if db.Error != nil { + return + } + lagoon.AfterCommit(db.Statement.Context, db, func(ctx context.Context, db *gorm.DB) { + // runs only once the insert is committed + }) + }) + }) +} + +func publish(ctx context.Context, gdb *gorm.DB, post *Post) error { + return lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error { + return tx.Create(post).Error // acme:post_created work waits for this commit + }) +} +``` + A plugin ships its schema as an ordered gormigrate set; `migrate` runs it after the framework sets: ```go @@ -87,7 +112,11 @@ func (p *Plugin) Migrations() []*gormigrate.Migration { | `lagoon.OpenFromApp` | Opens the shared pool from `database.dsn` and publishes the `app.key` encryption keys. | | `lagoon.Open` | Opens and pings a DSN, checks the locale and returns the pool plus a GORM handle on it. | | `lagoon.Use` | Returns a GORM handle on an existing pool after the same checks. | -| `lagoon.Publish` | Stores the pool and GORM handle on the `backpack.App`. | +| `lagoon.Publish` | Stores the pool and GORM handle on the `backpack.App`, then runs the callbacks queued by `lagoon.OnDatabase`. | +| `lagoon.OnDatabase` | Runs a callback with the pool and GORM handle once the database is published. | +| `lagoon.Transaction` | Runs a function in a transaction (a savepoint when nested) and its `lagoon.AfterCommit` callbacks after the commit. | +| `lagoon.AfterCommit` | Registers work to run after the surrounding transaction commits. | +| `lagoon.AfterCommitCallback` | Name of the GORM callback, `lagoon:after_commit`, that runs single-statement after-commit work. | | `lagoon.DSN` | Reads `database.dsn` from config. | | `lagoon.CheckLocale` | Fails unless the database default locale is ICU `pl-PL`. | | `lagoon.Migrate` | Runs framework and plugin migrations in order. | diff --git a/modules/lagoon/connection.go b/modules/lagoon/connection.go index faf393e..32f06d9 100644 --- a/modules/lagoon/connection.go +++ b/modules/lagoon/connection.go @@ -21,8 +21,9 @@ const ( // Open pings dsn through pgx stdlib, requires Postgres 16 ICU pl-PL, and // returns that exact *sql.DB plus a GORM handle opened on it. // -// Phase 11 owns a separate pgxpool.Pool for River LISTEN/NOTIFY. Do not -// create that listener pool here; application queries share this *sql.DB. +// The job worker in the conga module opens its own single-connection +// pgxpool.Pool for River LISTEN/NOTIFY. Do not create that listener pool +// here; application queries, including River's, share this *sql.DB. func Open(ctx context.Context, dsn string) (*sql.DB, *gorm.DB, error) { dsn = strings.TrimSpace(dsn) if dsn == "" { @@ -76,6 +77,9 @@ func gormFromSQL(sqlDB *sql.DB) (*gorm.DB, error) { if got != sqlDB { return nil, fmt.Errorf("lagoon: GORM is not using the shared *sql.DB") } + if err := registerAfterCommit(gdb); err != nil { + return nil, fmt.Errorf("lagoon: register %s: %w", AfterCommitCallback, err) + } return gdb, nil } @@ -106,7 +110,8 @@ func DSN(cfg *compass.Config) string { } // Publish stores the shared SQL pool and GORM handle on the app. Both -// handles refer to the same *sql.DB. +// handles refer to the same *sql.DB. It then runs the callbacks queued by +// OnDatabase and returns the first error. func Publish(app *backpack.App, sqlDB *sql.DB, gdb *gorm.DB) error { if app == nil { return fmt.Errorf("lagoon: app is nil") @@ -117,7 +122,10 @@ func Publish(app *backpack.App, sqlDB *sql.DB, gdb *gorm.DB) error { if err := app.Publish(sqlDB); err != nil { return err } - return app.Publish(gdb) + if err := app.Publish(gdb); err != nil { + return err + } + return runDatabaseHooks(app, sqlDB, gdb) } // CheckLocale fails unless the connected database uses ICU locale pl-PL. diff --git a/modules/lagoon/ondatabase.go b/modules/lagoon/ondatabase.go new file mode 100644 index 0000000..e2fca66 --- /dev/null +++ b/modules/lagoon/ondatabase.go @@ -0,0 +1,95 @@ +package lagoon + +import ( + "database/sql" + "fmt" + "sync" + + "git.golem15.com/golem15/summercms/modules/backpack" + "gorm.io/gorm" +) + +// databaseHooks is the per-app queue of OnDatabase callbacks waiting for +// Publish. It lives on the app, never in a package variable. +type databaseHooks struct { + mu sync.Mutex + pending []func(*sql.DB, *gorm.DB) error +} + +// OnDatabase runs fn with the shared pool and GORM handle as soon as the +// database is available: immediately when the app has already published a +// *gorm.DB, otherwise when Publish runs. Plugins register GORM callbacks +// through it from Boot, which runs before the serve command publishes the +// database. The error of an immediate call is returned; the error of a +// queued call is returned by Publish. +func OnDatabase(app *backpack.App, fn func(sqlDB *sql.DB, gdb *gorm.DB) error) error { + if app == nil { + return fmt.Errorf("lagoon: app is nil") + } + if fn == nil { + return fmt.Errorf("lagoon: database hook is nil") + } + h, err := hooksFor(app) + if err != nil { + return err + } + h.mu.Lock() + sqlDB, gdb, ok := publishedDB(app) + if !ok { + h.pending = append(h.pending, fn) + h.mu.Unlock() + return nil + } + h.mu.Unlock() + return fn(sqlDB, gdb) +} + +// runDatabaseHooks drains the OnDatabase queue once the handles are +// published and returns the first error. +func runDatabaseHooks(app *backpack.App, sqlDB *sql.DB, gdb *gorm.DB) error { + h, ok := app.Lookup[*databaseHooks]() + if !ok || h == nil { + return nil + } + h.mu.Lock() + pending := h.pending + h.pending = nil + h.mu.Unlock() + for _, fn := range pending { + if err := fn(sqlDB, gdb); err != nil { + return fmt.Errorf("lagoon: database hook: %w", err) + } + } + return nil +} + +func hooksFor(app *backpack.App) (*databaseHooks, error) { + if h, ok := app.Lookup[*databaseHooks](); ok && h != nil { + return h, nil + } + h := &databaseHooks{} + if err := app.Publish(h); err != nil { + if existing, ok := app.Lookup[*databaseHooks](); ok && existing != nil { + return existing, nil + } + return nil, fmt.Errorf("lagoon: %w", err) + } + return h, nil +} + +// publishedDB returns the published handles. A published *gorm.DB is +// enough: its own pool stands in when no *sql.DB was published. +func publishedDB(app *backpack.App) (*sql.DB, *gorm.DB, bool) { + gdb, ok := app.Lookup[*gorm.DB]() + if !ok || gdb == nil { + return nil, nil, false + } + if sqlDB, ok := app.Lookup[*sql.DB](); ok && sqlDB != nil { + return sqlDB, gdb, true + } + sqlDB, err := gdb.DB() + if err != nil { + return nil, nil, false + } + return sqlDB, gdb, true +} diff --git a/modules/lagoon/ondatabase_test.go b/modules/lagoon/ondatabase_test.go new file mode 100644 index 0000000..5af4c5f --- /dev/null +++ b/modules/lagoon/ondatabase_test.go @@ -0,0 +1,85 @@ +package lagoon + +import ( + "database/sql" + "errors" + "testing" + + "git.golem15.com/golem15/summercms/modules/backpack" + "gorm.io/gorm" +) + +// TestOnDatabaseAfterActivate covers the production boot order: plugins +// register database hooks at Boot, before serve publishes the database. +func TestOnDatabaseAfterActivate(t *testing.T) { + db, _ := dedicatedDB(t, "lagoon_ondatabase") + gdb, err := Use(t.Context(), db) + if err != nil { + t.Fatal(err) + } + + t.Run("before_publish_runs_on_publish", func(t *testing.T) { + app := backpack.New(nil) + calls := 0 + var gotSQL *sql.DB + var gotGDB *gorm.DB + if err := OnDatabase(app, func(s *sql.DB, g *gorm.DB) error { + calls++ + gotSQL, gotGDB = s, g + return nil + }); err != nil { + t.Fatal(err) + } + if calls != 0 { + t.Fatalf("hook ran %d times before Publish", calls) + } + if err := Publish(app, db, gdb); err != nil { + t.Fatal(err) + } + if calls != 1 || gotSQL != db || gotGDB != gdb { + t.Fatalf("after Publish: calls=%d sql=%v gdb=%v", calls, gotSQL == db, gotGDB == gdb) + } + }) + + t.Run("after_publish_runs_now", func(t *testing.T) { + app := backpack.New(nil) + if err := Publish(app, db, gdb); err != nil { + t.Fatal(err) + } + calls := 0 + if err := OnDatabase(app, func(*sql.DB, *gorm.DB) error { calls++; return nil }); err != nil { + t.Fatal(err) + } + if calls != 1 { + t.Fatalf("hook ran %d times, want 1", calls) + } + }) + + t.Run("gorm_only_published_runs_now", func(t *testing.T) { + app := backpack.New(nil) + if err := app.Publish(gdb); err != nil { + t.Fatal(err) + } + var gotSQL *sql.DB + if err := OnDatabase(app, func(s *sql.DB, _ *gorm.DB) error { gotSQL = s; return nil }); err != nil { + t.Fatal(err) + } + if gotSQL != db { + t.Fatal("hook did not receive the pool behind the published *gorm.DB") + } + }) + + t.Run("errors_surface", func(t *testing.T) { + boom := errors.New("boom") + app := backpack.New(nil) + if err := OnDatabase(app, func(*sql.DB, *gorm.DB) error { return boom }); err != nil { + t.Fatal(err) + } + if err := Publish(app, db, gdb); !errors.Is(err, boom) { + t.Fatalf("Publish err = %v, want the hook error", err) + } + if err := OnDatabase(app, func(*sql.DB, *gorm.DB) error { return boom }); !errors.Is(err, boom) { + t.Fatalf("OnDatabase err = %v, want the hook error", err) + } + }) +} diff --git a/modules/lagoon/transaction.go b/modules/lagoon/transaction.go new file mode 100644 index 0000000..929f8d7 --- /dev/null +++ b/modules/lagoon/transaction.go @@ -0,0 +1,182 @@ +package lagoon + +import ( + "context" + "fmt" + "log/slog" + "sync" + + "gorm.io/gorm" +) + +// AfterCommitCallback is the name of the GORM callback that runs the +// AfterCommit work of a single-statement write once GORM commits it. +const AfterCommitCallback = "lagoon:after_commit" + +const statementBufferKey = "lagoon:after_commit" + +type afterCommitKey struct{} + +// afterCommitBuffer holds the callbacks registered inside one transaction. +type afterCommitBuffer struct { + mu sync.Mutex + fns []func(context.Context, *gorm.DB) +} + +func (b *afterCommitBuffer) add(fns ...func(context.Context, *gorm.DB)) { + b.mu.Lock() + defer b.mu.Unlock() + b.fns = append(b.fns, fns...) +} + +func (b *afterCommitBuffer) take() []func(context.Context, *gorm.DB) { + b.mu.Lock() + defer b.mu.Unlock() + fns := b.fns + b.fns = nil + return fns +} + +// Transaction runs fn in a transaction whose AfterCommit callbacks run, in +// registration order, only after the commit succeeds. Writes inside fn must +// use the ctx and tx it receives. A Transaction nested in another becomes a +// savepoint: its callbacks join the outer transaction's only when fn +// succeeds, so work dropped with the savepoint never runs its callbacks. +// A panicking callback is logged and never turns a committed write into an +// error. +func Transaction(ctx context.Context, gdb *gorm.DB, fn func(ctx context.Context, tx *gorm.DB) error) error { + if gdb == nil { + return fmt.Errorf("lagoon: gorm db is nil") + } + if ctx == nil { + ctx = context.Background() + } + if parent, ok := ctx.Value(afterCommitKey{}).(*afterCommitBuffer); ok && parent != nil { + child := &afterCommitBuffer{} + childCtx := context.WithValue(ctx, afterCommitKey{}, child) + err := gdb.WithContext(childCtx).Transaction(func(tx *gorm.DB) error { + return fn(childCtx, tx) + }) + if err != nil { + return err + } + parent.add(child.take()...) + return nil + } + buf := &afterCommitBuffer{} + txCtx := context.WithValue(ctx, afterCommitKey{}, buf) + if err := gdb.WithContext(txCtx).Transaction(func(tx *gorm.DB) error { + return fn(txCtx, tx) + }); err != nil { + return err + } + runAfterCommit(ctx, gdb.Session(&gorm.Session{NewDB: true, Context: ctx}), buf.take()) + return nil +} + +// AfterCommit registers fn to run after the surrounding transaction commits. +// Inside Transaction it is buffered until that transaction commits. Inside a +// single-statement write for which GORM opened its own transaction (for +// example from a GORM create callback) it runs after that commit through the +// AfterCommitCallback callback, and not at all when the write fails. +// Anywhere else, including inside a plain gorm Transaction, fn runs +// immediately with db. +func AfterCommit(ctx context.Context, db *gorm.DB, fn func(ctx context.Context, db *gorm.DB)) { + if fn == nil { + return + } + if ctx == nil && db != nil && db.Statement != nil { + ctx = db.Statement.Context + } + if ctx == nil { + ctx = context.Background() + } + if buf := bufferFrom(ctx, db); buf != nil { + buf.add(fn) + return + } + if db != nil && db.Statement != nil { + if _, started := db.InstanceGet("gorm:started_transaction"); started { + buf := &afterCommitBuffer{} + if existing, ok := db.InstanceGet(statementBufferKey); ok { + if b, ok := existing.(*afterCommitBuffer); ok && b != nil { + buf = b + } + } + buf.add(fn) + db.InstanceSet(statementBufferKey, buf) + return + } + } + runAfterCommit(ctx, db, []func(context.Context, *gorm.DB){fn}) +} + +func bufferFrom(ctx context.Context, db *gorm.DB) *afterCommitBuffer { + if buf, ok := ctx.Value(afterCommitKey{}).(*afterCommitBuffer); ok && buf != nil { + return buf + } + if db != nil && db.Statement != nil && db.Statement.Context != nil { + if buf, ok := db.Statement.Context.Value(afterCommitKey{}).(*afterCommitBuffer); ok && buf != nil { + return buf + } + } + return nil +} + +// flushStatementAfterCommit is the AfterCommitCallback GORM callback: it +// runs the statement's buffered callbacks once GORM's own transaction has +// committed, on a fresh session over the pool. +func flushStatementAfterCommit(db *gorm.DB) { + raw, ok := db.InstanceGet(statementBufferKey) + if !ok { + return + } + buf, ok := raw.(*afterCommitBuffer) + if !ok || buf == nil { + return + } + fns := buf.take() + if db.Error != nil || len(fns) == 0 { + return + } + ctx := db.Statement.Context + if ctx == nil { + ctx = context.Background() + } + runAfterCommit(ctx, db.Session(&gorm.Session{NewDB: true, Context: ctx}), fns) +} + +func runAfterCommit(ctx context.Context, db *gorm.DB, fns []func(context.Context, *gorm.DB)) { + for _, fn := range fns { + func() { + defer func() { + if r := recover(); r != nil { + slog.Default().Warn("lagoon: after-commit callback panicked", "panic", fmt.Sprint(r)) + } + }() + fn(ctx, db) + }() + } +} + +// registerAfterCommit installs AfterCommitCallback on the create, update and +// delete processors once per GORM handle. +func registerAfterCommit(gdb *gorm.DB) error { + cb := gdb.Callback() + if cb.Create().Get(AfterCommitCallback) == nil { + if err := cb.Create().After("gorm:commit_or_rollback_transaction").Register(AfterCommitCallback, flushStatementAfterCommit); err != nil { + return err + } + } + if cb.Update().Get(AfterCommitCallback) == nil { + if err := cb.Update().After("gorm:commit_or_rollback_transaction").Register(AfterCommitCallback, flushStatementAfterCommit); err != nil { + return err + } + } + if cb.Delete().Get(AfterCommitCallback) == nil { + if err := cb.Delete().After("gorm:commit_or_rollback_transaction").Register(AfterCommitCallback, flushStatementAfterCommit); err != nil { + return err + } + } + return nil +} diff --git a/modules/lagoon/transaction_test.go b/modules/lagoon/transaction_test.go new file mode 100644 index 0000000..709e65d --- /dev/null +++ b/modules/lagoon/transaction_test.go @@ -0,0 +1,208 @@ +package lagoon + +import ( + "context" + "database/sql" + "errors" + "reflect" + "sync" + "testing" + + "gorm.io/gorm" +) + +type acItem struct { + ID uint `gorm:"column:id;primaryKey"` + Name string `gorm:"column:name"` +} + +func (acItem) TableName() string { return "lagoon_ac_items" } + +// recorder collects after-commit callback names and whether the row they +// were registered for was visible to another connection when they ran. +type recorder struct { + mu sync.Mutex + got []string +} + +func (r *recorder) add(name string) { + r.mu.Lock() + defer r.mu.Unlock() + r.got = append(r.got, name) +} + +func (r *recorder) names() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.got...) +} + +func committedCount(t *testing.T, db *sql.DB, name string) int { + t.Helper() + var n int + if err := db.QueryRowContext(context.Background(), `SELECT count(*) FROM lagoon_ac_items WHERE name = $1`, name).Scan(&n); err != nil { + t.Error(err) + } + return n +} + +// TestTransactionAfterCommit covers the after-commit seam: callbacks run +// only after a successful commit, in order, and never for rolled-back work. +func TestTransactionAfterCommit(t *testing.T) { + db, _ := dedicatedDB(t, "lagoon_after_commit") + gdb, err := Use(t.Context(), db) + if err != nil { + t.Fatal(err) + } + if err := gdb.Exec(`CREATE TABLE lagoon_ac_items (id SERIAL PRIMARY KEY, name TEXT NOT NULL UNIQUE)`).Error; err != nil { + t.Fatal(err) + } + ctx := t.Context() + + t.Run("commit_runs_in_order", func(t *testing.T) { + var r recorder + err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { + if committedCount(t, db, "ordered") != 1 { + t.Error("callback ran before the row was committed") + } + r.add("a") + }) + if err := tx.Create(&acItem{Name: "ordered"}).Error; err != nil { + return err + } + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("b") }) + if len(r.names()) != 0 { + t.Error("callback ran inside the transaction") + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if got := r.names(); !reflect.DeepEqual(got, []string{"a", "b"}) { + t.Fatalf("callbacks = %v, want [a b]", got) + } + }) + + t.Run("rollback_runs_none", func(t *testing.T) { + var r recorder + rollback := errors.New("rollback") + err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("x") }) + if err := tx.Create(&acItem{Name: "rolled-back"}).Error; err != nil { + return err + } + return rollback + }) + if !errors.Is(err, rollback) { + t.Fatalf("err = %v", err) + } + if got := r.names(); len(got) != 0 { + t.Fatalf("callbacks after rollback = %v", got) + } + if committedCount(t, db, "rolled-back") != 0 { + t.Fatal("row survived the rollback") + } + }) + + t.Run("nested_failure_drops_inner", func(t *testing.T) { + var r recorder + err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("outer") }) + inner := Transaction(ctx, tx, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("inner-failed") }) + if err := tx.Create(&acItem{Name: "inner-failed"}).Error; err != nil { + return err + } + return errors.New("inner fails") + }) + if inner == nil { + t.Error("inner transaction did not fail") + } + return Transaction(ctx, tx, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("inner-ok") }) + return tx.Create(&acItem{Name: "inner-ok"}).Error + }) + }) + if err != nil { + t.Fatal(err) + } + if got := r.names(); !reflect.DeepEqual(got, []string{"outer", "inner-ok"}) { + t.Fatalf("callbacks = %v, want [outer inner-ok]", got) + } + if committedCount(t, db, "inner-failed") != 0 || committedCount(t, db, "inner-ok") != 1 { + t.Fatal("savepoint work was not rolled back or kept as expected") + } + }) + + t.Run("panicking_callback_does_not_fail_commit", func(t *testing.T) { + var r recorder + err := Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error { + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { panic("boom") }) + AfterCommit(ctx, tx, func(context.Context, *gorm.DB) { r.add("after-panic") }) + return nil + }) + if err != nil { + t.Fatalf("committed transaction reported %v", err) + } + if got := r.names(); !reflect.DeepEqual(got, []string{"after-panic"}) { + t.Fatalf("callbacks = %v", got) + } + }) + + t.Run("implicit_single_statement", func(t *testing.T) { + // A separate handle so the test callback stays off the shared one. + g2, err := Use(ctx, db) + if err != nil { + t.Fatal(err) + } + var r recorder + if err := g2.Callback().Create().Before("gorm:create").Register("lagoon_test:after_commit", func(stmt *gorm.DB) { + item, ok := stmt.Statement.Dest.(*acItem) + if !ok { + return + } + name := item.Name + AfterCommit(stmt.Statement.Context, stmt, func(context.Context, *gorm.DB) { + if committedCount(t, db, name) != 1 { + t.Errorf("callback for %s ran before commit", name) + } + r.add(name) + }) + }); err != nil { + t.Fatal(err) + } + if err := g2.WithContext(ctx).Create(&acItem{Name: "implicit"}).Error; err != nil { + t.Fatal(err) + } + if err := g2.WithContext(ctx).Create(&acItem{Name: "implicit"}).Error; err == nil { + t.Fatal("duplicate insert succeeded") + } + if got := r.names(); !reflect.DeepEqual(got, []string{"implicit"}) { + t.Fatalf("callbacks = %v, want only the committed insert", got) + } + }) + + t.Run("plain_gorm_transaction_runs_now", func(t *testing.T) { + var got *gorm.DB + err := gdb.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + AfterCommit(ctx, tx, func(_ context.Context, d *gorm.DB) { got = d }) + if got != tx { + t.Error("callback did not run immediately with the tx handle") + } + return nil + }) + if err != nil { + t.Fatal(err) + } + }) + + t.Run("outside_transaction_runs_now", func(t *testing.T) { + ran := false + AfterCommit(ctx, gdb, func(context.Context, *gorm.DB) { ran = true }) + if !ran { + t.Fatal("callback outside a transaction did not run immediately") + } + }) +}