fix(11-07): enqueue broadcast jobs before GORM commits a single write

- lighthouse:after_create/update/delete also declare
  Before(gorm:commit_or_rollback_transaction); an After-only anchor put
  them past GORM's own commit, so a plain gdb.Create enqueued its
  broadcast job after the commit on the pool (deferred from 11-05)
- lighthouse gets the testcontainers Postgres harness and TestBroadcastTx
  (commit publishes once, rollback nothing, single-statement write
  enqueues on its own transaction, failed write enqueues nothing)

modules/lighthouse/README.md

@@ -39,7 +39,7 @@ The `centrifugo` sub-package is the Centrifugo driver. It has a hand-rolled `net
   - `lighthouse.BroadcastTTLer` or `Binding.TTL` replaces the ttl.
 
   The event name is `{action}.{alias}` lowercased: `lighthouse.ActionCreated`, `lighthouse.ActionUpdated` or `lighthouse.ActionDeleted`, then an alias that defaults to `<plugin>.<model>` (the Go package name, or the parent directory of a `models` package, and the type name). The payload builder receives a `lighthouse.Event` with the action, the `lighthouse.Actor`, the timestamp and the ttl. A soft delete counts as a delete. A delete's channels and payload are computed from a fresh read of the row before it is deleted, so deleting a model that holds only its id still broadcasts. An empty channel list means no broadcast.
-- Transactional delivery. GORM callbacks (`lighthouse.CallbackAfterCreate`, `lighthouse.CallbackAfterUpdate`, `lighthouse.CallbackSnapshot` and `lighthouse.CallbackAfterDelete`) are installed through `lagoon.OnDatabase`. They enqueue a `lighthouse.BroadcastArgs` job on the write's `*sql.Tx`, on the `realtime.broadcast_queue` queue with MaxAttempts 1 and the `realtime.broadcast_timeout` timeout. Channel and payload queries and the enqueue run inside a savepoint, so a failure is rolled back to it, logged at Warn with channels and event (never the payload), and the write goes on. A write with a zero primary key, such as `Model(&T{}).Where(…).Updates(…)`, is not broadcast; bulk paths suppress and emit instead. The null driver, or a driver whose `Enabled` reports false (Centrifugo without an API key), gets no jobs.
+- Transactional delivery. GORM callbacks (`lighthouse.CallbackAfterCreate`, `lighthouse.CallbackAfterUpdate`, `lighthouse.CallbackSnapshot` and `lighthouse.CallbackAfterDelete`) are installed through `lagoon.OnDatabase`. The after-write callbacks run after the model's own after hook and before GORM commits the transaction it opens for a single-statement write, so they enqueue a `lighthouse.BroadcastArgs` job on the write's `*sql.Tx` in every case (an explicit transaction or a single `Create`, `Save` or `Delete`), on the `realtime.broadcast_queue` queue with MaxAttempts 1 and the `realtime.broadcast_timeout` timeout. Channel and payload queries and the enqueue run inside a savepoint, so a failure is rolled back to it, logged at Warn with channels and event (never the payload), and the write goes on. A write with a zero primary key, such as `Model(&T{}).Where(…).Updates(…)`, is not broadcast; bulk paths suppress and emit instead. The null driver, or a driver whose `Enabled` reports false (Centrifugo without an API key), gets no jobs.
 - The broadcast job lowercases the channels and adds the `realtime.broadcast_namespace` prefix unless a channel already has it. It then publishes to one channel or broadcasts to several. A failure is logged as `realtime: broadcast failed` and is not retried. Delivery order across separate jobs is not guaranteed. The payload travels inside the job as a JSON string, so its key order survives Postgres JSONB.
 - Suppression: `lighthouse.WithoutBroadcasting` silences one model type for writes made with the context it hands to its function. Other types still broadcast, and a write through an outer context is not suppressed. `lighthouse.Service.Emit` enqueues one `lighthouse.Broadcast` on the caller's transaction and returns its error. Together they turn N row events into one summary event.
 - Centrifugo driver (`centrifugo.Driver`, driver name `centrifugo`):

modules/lighthouse/broadcast.go

@@ -229,20 +229,26 @@ type defaultPayload struct {
 	TTL       int       `json:"ttl"`
 }
 
+// commitCallback is GORM's commit of a transaction it opened itself.
+const commitCallback = "gorm:commit_or_rollback_transaction"
+
 // installCallbacks registers the broadcast callbacks on gdb, replacing
 // earlier ones so a handle shared by several apps broadcasts through the
-// most recent service.
+// most recent service. The after-write callbacks run after the model's own
+// after hook and before GORM commits a single-statement write, so the job
+// is enqueued on the write's transaction: GORM appends a callback that
+// names only an After anchor to the end of the chain, past the commit.
 func (s *Service) installCallbacks(gdb *gorm.DB) error {
 	cb := gdb.Callback()
 	if cb.Create().Get(CallbackAfterCreate) == nil {
-		if err := cb.Create().After("gorm:after_create").Register(CallbackAfterCreate, s.afterCreate); err != nil {
+		if err := cb.Create().After("gorm:after_create").Before(commitCallback).Register(CallbackAfterCreate, s.afterCreate); err != nil {
 			return err
 		}
 	} else if err := cb.Create().Replace(CallbackAfterCreate, s.afterCreate); err != nil {
 		return err
 	}
 	if cb.Update().Get(CallbackAfterUpdate) == nil {
-		if err := cb.Update().After("gorm:after_update").Register(CallbackAfterUpdate, s.afterUpdate); err != nil {
+		if err := cb.Update().After("gorm:after_update").Before(commitCallback).Register(CallbackAfterUpdate, s.afterUpdate); err != nil {
 			return err
 		}
 	} else if err := cb.Update().Replace(CallbackAfterUpdate, s.afterUpdate); err != nil {
@@ -256,7 +262,7 @@ func (s *Service) installCallbacks(gdb *gorm.DB) error {
 		return err
 	}
 	if cb.Delete().Get(CallbackAfterDelete) == nil {
-		if err := cb.Delete().After("gorm:after_delete").Register(CallbackAfterDelete, s.afterDelete); err != nil {
+		if err := cb.Delete().After("gorm:after_delete").Before(commitCallback).Register(CallbackAfterDelete, s.afterDelete); err != nil {
 			return err
 		}
 	} else if err := cb.Delete().Replace(CallbackAfterDelete, s.afterDelete); err != nil {

modules/lighthouse/broadcast_test.go

@@ -0,0 +1,362 @@
+package lighthouse
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/compass"
+	"git.golem15.com/golem15/summercms/modules/conga"
+	"git.golem15.com/golem15/summercms/modules/lagoon"
+	"gorm.io/gorm"
+)
+
+// Widget broadcasts through its Broadcastable method.
+type Widget struct {
+	ID      uint   `gorm:"column:id;primaryKey"`
+	Name    string `gorm:"column:name"`
+	OwnerID uint   `gorm:"column:owner_id"`
+}
+
+func (Widget) TableName() string { return "acme_widgets" }
+
+// BroadcastChannels publishes to the owner's widget channel; the mixed
+// case proves the job lowercases it.
+func (w *Widget) BroadcastChannels(context.Context, *gorm.DB) ([]string, error) {
+	if w.OwnerID == 0 {
+		return nil, nil
+	}
+	return []string{fmt.Sprintf("Widgets:%d", w.OwnerID)}, nil
+}
+
+// Gadget is broadcast only through a Binding installed by the test.
+type Gadget struct {
+	ID      uint   `gorm:"column:id;primaryKey"`
+	Name    string `gorm:"column:name"`
+	OwnerID uint   `gorm:"column:owner_id"`
+}
+
+func (Gadget) TableName() string { return "acme_gadgets" }
+
+const broadcastTables = `
+CREATE TABLE acme_widgets (id SERIAL PRIMARY KEY, name TEXT NOT NULL, owner_id INT NOT NULL DEFAULT 0);
+CREATE TABLE acme_gadgets (id SERIAL PRIMARY KEY, name TEXT NOT NULL UNIQUE, owner_id INT NOT NULL DEFAULT 0);
+CREATE TABLE acme_sprockets (id SERIAL PRIMARY KEY, name TEXT NOT NULL, channels TEXT NOT NULL DEFAULT '');
+`
+
+// logCapture records log messages with their attributes as strings.
+type logCapture struct {
+	mu      sync.Mutex
+	records []capturedLog
+}
+
+type capturedLog struct {
+	level slog.Level
+	msg   string
+	attrs map[string]string
+}
+
+func (h *logCapture) Enabled(context.Context, slog.Level) bool { return true }
+func (h *logCapture) WithAttrs([]slog.Attr) slog.Handler       { return h }
+func (h *logCapture) WithGroup(string) slog.Handler            { return h }
+func (h *logCapture) Handle(_ context.Context, r slog.Record) error {
+	rec := capturedLog{level: r.Level, msg: r.Message, attrs: map[string]string{}}
+	r.Attrs(func(a slog.Attr) bool {
+		rec.attrs[a.Key] = a.Value.String()
+		return true
+	})
+	h.mu.Lock()
+	h.records = append(h.records, rec)
+	h.mu.Unlock()
+	return nil
+}
+
+func (h *logCapture) count(msg string) int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	n := 0
+	for _, r := range h.records {
+		if r.msg == msg {
+			n++
+		}
+	}
+	return n
+}
+
+func (h *logCapture) find(msg string) (capturedLog, bool) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for _, r := range h.records {
+		if r.msg == msg {
+			return r, true
+		}
+	}
+	return capturedLog{}, false
+}
+
+func (h *logCapture) all() string {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	var b strings.Builder
+	for _, r := range h.records {
+		fmt.Fprintf(&b, "%s %s %v\n", r.level, r.msg, r.attrs)
+	}
+	return b.String()
+}
+
+// lhEnv is an app with the memory driver (unless kv names another), the
+// broadcast callbacks installed through the production boot order
+// (From before the database is published) and a running worker.
+type lhEnv struct {
+	app  *backpack.App
+	svc  *Service
+	db   *sql.DB
+	gdb  *gorm.DB
+	logs *logCapture
+}
+
+func newLHEnv(t *testing.T, kv map[string]any, setup func(t *testing.T, svc *Service)) lhEnv {
+	t.Helper()
+	db, dsn := migratedDB(t)
+	ctx := t.Context()
+	cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Env: "testing", Environ: []string{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	settings := map[string]any{"database.dsn": dsn, "realtime.driver": "memory"}
+	for k, v := range kv {
+		settings[k] = v
+	}
+	for k, v := range settings {
+		if err := cfg.Set(k, v); err != nil {
+			t.Fatal(err)
+		}
+	}
+	app := backpack.New(cfg)
+	logs := &logCapture{}
+	if err := app.Publish(slog.New(logs)); err != nil {
+		t.Fatal(err)
+	}
+	svc, err := From(app)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if setup != nil {
+		setup(t, svc)
+	}
+	gdb, err := lagoon.Use(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := gdb.Exec(broadcastTables).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := lagoon.Publish(app, db, gdb); err != nil {
+		t.Fatal(err)
+	}
+	w, err := conga.StartWorker(ctx, app, nil, conga.WorkerOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		if err := w.Stop(stopCtx); err != nil {
+			t.Errorf("stop worker: %v", err)
+		}
+	})
+	return lhEnv{app: app, svc: svc, db: db, gdb: gdb, logs: logs}
+}
+
+func (e lhEnv) memory(t *testing.T) *MemoryDriver {
+	t.Helper()
+	m, ok := e.svc.Driver().(*MemoryDriver)
+	if !ok {
+		t.Fatalf("driver is %T, want the memory driver", e.svc.Driver())
+	}
+	return m
+}
+
+// broadcastJobs counts summer.broadcast jobs through a separate connection.
+func (e lhEnv) broadcastJobs(t *testing.T) int {
+	t.Helper()
+	var n int
+	if err := e.db.QueryRowContext(t.Context(), `SELECT count(*) FROM river_job WHERE kind = 'summer.broadcast'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+// waitPublications waits for exactly n publications after the first from,
+// then waits a little longer to catch extra ones.
+func waitPublications(t *testing.T, m *MemoryDriver, from, n int) []Publication {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for len(m.Publications()) < from+n {
+		if time.Now().After(deadline) {
+			t.Fatalf("got %d publications, want %d", len(m.Publications())-from, n)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	time.Sleep(300 * time.Millisecond)
+	pubs := m.Publications()[from:]
+	if len(pubs) != n {
+		t.Fatalf("got %d publications, want exactly %d: %+v", len(pubs), n, pubs)
+	}
+	return pubs
+}
+
+// payloadKeys returns the top-level keys of a JSON object in order.
+func payloadKeys(t *testing.T, raw json.RawMessage) []string {
+	t.Helper()
+	dec := json.NewDecoder(strings.NewReader(string(raw)))
+	if tok, err := dec.Token(); err != nil || tok != json.Delim('{') {
+		t.Fatalf("payload %s is not an object", raw)
+	}
+	var keys []string
+	for dec.More() {
+		tok, err := dec.Token()
+		if err != nil {
+			t.Fatal(err)
+		}
+		keys = append(keys, tok.(string))
+		var skip json.RawMessage
+		if err := dec.Decode(&skip); err != nil {
+			t.Fatal(err)
+		}
+	}
+	return keys
+}
+
+// TestBroadcastTx covers RT-03 and T-11-05: model broadcasts are River jobs
+// enqueued in the write's transaction, published once after the commit and
+// never after a rollback, including single-statement writes for which GORM
+// opens its own transaction.
+func TestBroadcastTx(t *testing.T) {
+	var (
+		mu      sync.Mutex
+		inTxLog []bool
+	)
+	env := newLHEnv(t, nil, func(t *testing.T, svc *Service) {
+		err := Bind[Gadget](svc, Binding[Gadget]{
+			Alias: "acme.gadget",
+			Channels: func(_ context.Context, tx *gorm.DB, m *Gadget) ([]string, error) {
+				_, inTx := tx.Statement.ConnPool.(*sql.Tx)
+				mu.Lock()
+				inTxLog = append(inTxLog, inTx)
+				mu.Unlock()
+				return []string{fmt.Sprintf("gadgets:%d", m.OwnerID)}, nil
+			},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+	mem := env.memory(t)
+	ctx := t.Context()
+
+	t.Run("commit_publishes_once_after_commit", func(t *testing.T) {
+		from := len(mem.Publications())
+		before := env.broadcastJobs(t)
+		err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
+			if err := tx.WithContext(ctx).Create(&Widget{Name: "w1", OwnerID: 5}).Error; err != nil {
+				return err
+			}
+			var inside int
+			if err := tx.Raw(`SELECT count(*) FROM river_job WHERE kind = 'summer.broadcast'`).Scan(&inside).Error; err != nil {
+				return err
+			}
+			if inside != before+1 {
+				t.Errorf("jobs inside the write transaction = %d, want %d", inside, before+1)
+			}
+			if outside := env.broadcastJobs(t); outside != before {
+				t.Errorf("job visible outside the transaction before commit: %d, want %d", outside, before)
+			}
+			return nil
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		p := waitPublications(t, mem, from, 1)[0]
+		if p.Method != "publish" || len(p.Channels) != 1 || p.Channels[0] != "widgets:5" {
+			t.Fatalf("publication = %s %v, want publish [widgets:5]", p.Method, p.Channels)
+		}
+		if p.Event != "created.lighthouse.widget" {
+			t.Fatalf("event = %q", p.Event)
+		}
+		if got := strings.Join(payloadKeys(t, p.Payload), ","); got != "model,actor,timestamp,ttl" {
+			t.Fatalf("payload keys = %s", got)
+		}
+		var body struct {
+			Model Widget `json:"model"`
+			Actor Actor  `json:"actor"`
+			TTL   int    `json:"ttl"`
+		}
+		if err := json.Unmarshal(p.Payload, &body); err != nil {
+			t.Fatal(err)
+		}
+		if body.Model.Name != "w1" || body.TTL != DefaultTTL || body.Actor.UserID != nil || body.Actor.Name == nil || *body.Actor.Name != "System" {
+			t.Fatalf("payload = %s", p.Payload)
+		}
+	})
+
+	t.Run("rollback_publishes_nothing", func(t *testing.T) {
+		from := len(mem.Publications())
+		before := env.broadcastJobs(t)
+		rollback := errors.New("rollback")
+		err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
+			if err := tx.WithContext(ctx).Create(&Widget{Name: "w-rolled-back", OwnerID: 5}).Error; err != nil {
+				return err
+			}
+			return rollback
+		})
+		if !errors.Is(err, rollback) {
+			t.Fatalf("err = %v", err)
+		}
+		if after := env.broadcastJobs(t); after != before {
+			t.Fatalf("broadcast jobs %d -> %d after a rollback", before, after)
+		}
+		time.Sleep(300 * time.Millisecond)
+		if n := len(mem.Publications()) - from; n != 0 {
+			t.Fatalf("%d publications after a rollback", n)
+		}
+	})
+
+	t.Run("single_statement_write_enqueues_in_its_own_transaction", func(t *testing.T) {
+		from := len(mem.Publications())
+		mu.Lock()
+		inTxLog = nil
+		mu.Unlock()
+		if err := env.gdb.WithContext(ctx).Create(&Gadget{Name: "g1", OwnerID: 3}).Error; err != nil {
+			t.Fatal(err)
+		}
+		mu.Lock()
+		got := append([]bool(nil), inTxLog...)
+		mu.Unlock()
+		if len(got) != 1 || !got[0] {
+			t.Fatalf("channels ran on the write's transaction: %v, want [true] (the job must be enqueued before GORM commits)", got)
+		}
+		p := waitPublications(t, mem, from, 1)[0]
+		if p.Event != "created.acme.gadget" || p.Channels[0] != "gadgets:3" {
+			t.Fatalf("publication = %+v", p)
+		}
+	})
+
+	t.Run("failed_single_statement_write_enqueues_nothing", func(t *testing.T) {
+		before := env.broadcastJobs(t)
+		if err := env.gdb.WithContext(ctx).Create(&Gadget{Name: "g1", OwnerID: 3}).Error; err == nil {
+			t.Fatal("duplicate gadget was created")
+		}
+		if after := env.broadcastJobs(t); after != before {
+			t.Fatalf("broadcast jobs %d -> %d after a failed write", before, after)
+		}
+	})
+}

modules/lighthouse/postgres_test.go

@@ -0,0 +1,148 @@
+package lighthouse
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"net/url"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/lagoon"
+	_ "github.com/jackc/pgx/v5/stdlib"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+)
+
+var (
+	lhPG    *postgres.PostgresContainer
+	lhSQL   *sql.DB
+	lhDSN   string
+	lhPGErr error
+	dbSeq      atomic.Int64
+)
+
+func TestMain(m *testing.M) {
+	code := 1
+	if !testShort() {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		lhPGErr = startLighthousePostgres(ctx)
+		cancel()
+		if lhPGErr != nil {
+			fmt.Fprintf(os.Stderr, "lighthouse: testcontainers postgres: %v\n", lhPGErr)
+			stopLighthousePostgres()
+			os.Exit(1)
+		}
+	}
+	code = m.Run()
+	stopLighthousePostgres()
+	os.Exit(code)
+}
+
+func testShort() bool {
+	for _, a := range os.Args {
+		if a == "-test.short" {
+			return true
+		}
+	}
+	return false
+}
+
+func startLighthousePostgres(ctx context.Context) error {
+	ctr, err := postgres.Run(ctx,
+		"postgres:16-alpine",
+		postgres.WithDatabase("lighthouse"),
+		postgres.WithUsername("lighthouse"),
+		postgres.WithPassword("lighthouse"),
+		postgres.BasicWaitStrategies(),
+		testcontainers.WithEnv(map[string]string{
+			"POSTGRES_INITDB_ARGS": "--locale-provider=icu --icu-locale=pl-PL --encoding=UTF8",
+		}),
+	)
+	if err != nil {
+		return err
+	}
+	lhPG = ctr
+	dsn, err := ctr.ConnectionString(ctx, "sslmode=disable")
+	if err != nil {
+		return err
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		return err
+	}
+	if err := db.PingContext(ctx); err != nil {
+		_ = db.Close()
+		return err
+	}
+	lhSQL = db
+	lhDSN = dsn
+	return nil
+}
+
+func stopLighthousePostgres() {
+	if lhSQL != nil {
+		_ = lhSQL.Close()
+	}
+	if lhPG != nil {
+		_ = testcontainers.TerminateContainer(lhPG)
+	}
+}
+
+func adminDB(t *testing.T) *sql.DB {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("requires testcontainers postgres")
+	}
+	if lhPGErr != nil {
+		t.Fatalf("postgres unavailable: %v", lhPGErr)
+	}
+	if lhSQL == nil {
+		t.Fatal("postgres unavailable: container was not started")
+	}
+	return lhSQL
+}
+
+// migratedDB returns a dedicated ICU pl-PL database migrated with
+// lagoon.Migrate (River v7 and summer_jobs) plus its DSN.
+func migratedDB(t *testing.T) (*sql.DB, string) {
+	t.Helper()
+	admin := adminDB(t)
+	ctx := t.Context()
+	name := fmt.Sprintf("lighthouse_%d", dbSeq.Add(1))
+	if _, err := admin.ExecContext(ctx, `CREATE DATABASE `+name+` TEMPLATE template0 ENCODING 'UTF8' LOCALE_PROVIDER icu ICU_LOCALE 'pl-PL'`); err != nil && !strings.Contains(err.Error(), "already exists") {
+		t.Fatalf("create %s: %v", name, err)
+	}
+	dsn, err := dsnWithDB(lhDSN, name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = db.Close()
+		_, _ = admin.ExecContext(context.Background(), `DROP DATABASE IF EXISTS `+name+` WITH (FORCE)`)
+	})
+	gdb, err := lagoon.Use(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := lagoon.Migrate(gdb, nil); err != nil {
+		t.Fatal(err)
+	}
+	return db, dsn
+}
+
+func dsnWithDB(dsn, name string) (string, error) {
+	u, err := url.Parse(dsn)
+	if err != nil {
+		return "", err
+	}
+	u.Path = "/" + name
+	return u.String(), nil
+}