fix(cabana): backend subject events reach pier as *pact.MCPSubjectChanged, fired after the transaction commits

modules/cabana/subject_events.go

@@ -4,7 +4,10 @@ import (
 	"context"
 
 	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/lagoon"
 	"git.golem15.com/golem15/summercms/modules/pact"
+
+	"gorm.io/gorm"
 )
 
 // backendSubjectGuard is the guard name backend subject events carry: the
@@ -13,9 +16,12 @@ const backendSubjectGuard = "backend"
 
 // fireSubjectChanged fires one pact.MCPSubjectChanged of the given kind for
 // every administrator id through the application's event bus, so the MCP
-// endpoint re-lists a linked admin's tools or ends their sessions. A missing
-// application or bus is a no-op and a listener's error is logged: the save
-// that caused the event never fails because of it.
+// endpoint re-lists a linked admin's tools or ends their sessions. The event
+// is a *pact.MCPSubjectChanged, the type pier listens for. Inside a
+// transaction it fires only after the commit, so a rolled-back save fires
+// nothing; outside one it fires at once. A missing application or bus is a
+// no-op and a listener's error is logged: the save that caused the event
+// never fails because of it.
 func fireSubjectChanged(ctx context.Context, app *backpack.App, kind string, adminIDs ...uint) {
 	if app == nil || app.Events == nil {
 		return
@@ -23,13 +29,16 @@ func fireSubjectChanged(ctx context.Context, app *backpack.App, kind string, adm
 	if ctx == nil {
 		ctx = context.Background()
 	}
-	for _, id := range adminIDs {
-		if id == 0 {
-			continue
+	tx, _ := TxFromContext(ctx)
+	lagoon.AfterCommit(ctx, tx, func(ctx context.Context, _ *gorm.DB) {
+		for _, id := range adminIDs {
+			if id == 0 {
+				continue
+			}
+			event := &pact.MCPSubjectChanged{Guard: backendSubjectGuard, UserID: id, Kind: kind}
+			if err := app.Events.Fire(ctx, event); err != nil {
+				moduleLog.WarnContext(ctx, "cabana: subject changed listener failed", "kind", kind, "admin_id", id, "error", err)
+			}
 		}
-		event := pact.MCPSubjectChanged{Guard: backendSubjectGuard, UserID: id, Kind: kind}
-		if err := app.Events.Fire(ctx, event); err != nil {
-			moduleLog.WarnContext(ctx, "cabana: subject changed listener failed", "kind", kind, "admin_id", id, "error", err)
-		}
-	}
+	})
 }

modules/cabana/subject_events_test.go

@@ -0,0 +1,74 @@
+package cabana
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/lagoon"
+	"git.golem15.com/golem15/summercms/modules/pact"
+
+	"gorm.io/gorm"
+)
+
+// subjectListener records every *pact.MCPSubjectChanged the bus delivers,
+// the type pier listens for.
+func subjectListener(app *backpack.App) *[]pact.MCPSubjectChanged {
+	got := &[]pact.MCPSubjectChanged{}
+	app.Events.Listen("test", func(_ context.Context, ev *pact.MCPSubjectChanged) error {
+		*got = append(*got, *ev)
+		return nil
+	})
+	return got
+}
+
+func TestFireSubjectChangedReachesPointerListener(t *testing.T) {
+	app := backpack.New(nil)
+	got := subjectListener(app)
+
+	fireSubjectChanged(context.Background(), app, pact.MCPSubjectPermissions, 7, 0, 9)
+
+	if len(*got) != 2 {
+		t.Fatalf("delivered %d events, want 2: %+v", len(*got), *got)
+	}
+	for i, id := range []uint{7, 9} {
+		ev := (*got)[i]
+		if ev.Guard != backendSubjectGuard || ev.UserID != id || ev.Kind != pact.MCPSubjectPermissions {
+			t.Fatalf("event %d = %+v", i, ev)
+		}
+	}
+}
+
+func TestFireSubjectChangedWaitsForCommit(t *testing.T) {
+	_, db := newListService(t)
+	app := backpack.New(nil)
+	got := subjectListener(app)
+
+	err := lagoon.Transaction(context.Background(), db, func(ctx context.Context, tx *gorm.DB) error {
+		fireSubjectChanged(withTx(ctx, tx), app, pact.MCPSubjectDeleted, 3)
+		if len(*got) != 0 {
+			t.Fatalf("fired inside the transaction: %+v", *got)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(*got) != 1 || (*got)[0].UserID != 3 || (*got)[0].Kind != pact.MCPSubjectDeleted {
+		t.Fatalf("after commit got %+v, want one deleted event for 3", *got)
+	}
+
+	*got = nil
+	rollback := errors.New("rollback")
+	err = lagoon.Transaction(context.Background(), db, func(ctx context.Context, tx *gorm.DB) error {
+		fireSubjectChanged(withTx(ctx, tx), app, pact.MCPSubjectDeleted, 4)
+		return rollback
+	})
+	if !errors.Is(err, rollback) {
+		t.Fatalf("err = %v, want rollback", err)
+	}
+	if len(*got) != 0 {
+		t.Fatalf("a rolled-back transaction fired %+v", *got)
+	}
+}