summercms/modules/lighthouse/broadcast_test.go

package lighthouse

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/bouncer"
	"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)
		}
	})
}

// Sprocket broadcasts through every optional method contract.
type Sprocket struct {
	ID       uint   `gorm:"column:id;primaryKey"`
	Name     string `gorm:"column:name"`
	Channels string `gorm:"column:channels"`
}

func (Sprocket) TableName() string { return "acme_sprockets" }

func (s *Sprocket) BroadcastChannels(_ context.Context, tx *gorm.DB) ([]string, error) {
	if s.Channels == "" {
		return nil, nil
	}
	if s.Channels == "fail" {
		return nil, errors.New("channels failed")
	}
	if s.Channels == "swallow" {
		// A channel function that treats a failed read as "no channels".
		_ = tx.Exec(`SELECT * FROM acme_missing_table`).Error
		return nil, nil
	}
	if s.Channels == "abort-tx" {
		// A failed statement aborts a Postgres transaction unless it runs
		// inside a savepoint.
		return nil, tx.Exec(`SELECT * FROM acme_missing_table`).Error
	}
	return strings.Split(s.Channels, ","), nil
}

func (Sprocket) BroadcastAlias() string { return "acme.sprocket" }
func (Sprocket) BroadcastTTL() int      { return 15 }

func (s *Sprocket) BroadcastPayload(_ context.Context, _ *gorm.DB, ev Event) (any, error) {
	if s.Name == "bad-payload" {
		return nil, errors.New("payload failed")
	}
	return struct {
		ID     uint   `json:"id"`
		Action Action `json:"action"`
		TTL    int    `json:"ttl"`
		Actor  Actor  `json:"actor"`
	}{s.ID, ev.Action, ev.TTL, ev.Actor}, nil
}

func (s *Sprocket) ShouldBroadcast(a Action) bool { return a != ActionUpdated }

// TestSuppression covers D-08: WithoutBroadcasting[T] silences one model
// type for the ctx it hands out and nothing else. Another type still
// broadcasts, nesting adds types, a pointer type argument names the
// struct, and a write through a stale outer ctx is not suppressed.
func TestSuppression(t *testing.T) {
	env := newLHEnv(t, nil, func(t *testing.T, svc *Service) {
		if err := Bind[Gadget](svc, Binding[Gadget]{Channels: func(_ context.Context, _ *gorm.DB, m *Gadget) ([]string, error) {
			return []string{fmt.Sprintf("gadgets:%d", m.OwnerID)}, nil
		}}); err != nil {
			t.Fatal(err)
		}
	})
	mem := env.memory(t)
	ctx := t.Context()
	events := func(pubs []Publication) []string {
		var out []string
		for _, p := range pubs {
			out = append(out, p.Event+"@"+strings.Join(p.Channels, ","))
		}
		return out
	}

	t.Run("silences_widget_not_gadget", func(t *testing.T) {
		from := len(mem.Publications())
		err := WithoutBroadcasting[Widget](ctx, func(ctx context.Context) error {
			return lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
				if err := tx.WithContext(ctx).Create(&Widget{Name: "quiet", OwnerID: 1}).Error; err != nil {
					return err
				}
				return tx.WithContext(ctx).Create(&Gadget{Name: "loud", OwnerID: 1}).Error
			})
		})
		if err != nil {
			t.Fatal(err)
		}
		got := events(waitPublications(t, mem, from, 1))
		if len(got) != 1 || got[0] != "created.lighthouse.gadget@gadgets:1" {
			t.Fatalf("publications = %v, want only the gadget", got)
		}
	})

	t.Run("nesting_and_pointer_type_argument", func(t *testing.T) {
		before := env.broadcastJobs(t)
		err := WithoutBroadcasting[*Widget](ctx, func(ctx context.Context) error {
			return WithoutBroadcasting[Gadget](ctx, func(ctx context.Context) error {
				if err := env.gdb.WithContext(ctx).Create(&Widget{Name: "nested", OwnerID: 2}).Error; err != nil {
					return err
				}
				return env.gdb.WithContext(ctx).Create(&Gadget{Name: "nested", OwnerID: 2}).Error
			})
		})
		if err != nil {
			t.Fatal(err)
		}
		if after := env.broadcastJobs(t); after != before {
			t.Fatalf("broadcast jobs %d -> %d with both types suppressed", before, after)
		}
	})

	t.Run("stale_outer_ctx_is_not_suppressed", func(t *testing.T) {
		from := len(mem.Publications())
		outer := ctx
		err := WithoutBroadcasting[Widget](ctx, func(context.Context) error {
			return env.gdb.WithContext(outer).Create(&Widget{Name: "outer", OwnerID: 3}).Error
		})
		if err != nil {
			t.Fatal(err)
		}
		got := events(waitPublications(t, mem, from, 1))
		if got[0] != "created.lighthouse.widget@widgets:3" {
			t.Fatalf("publications = %v", got)
		}
	})

	t.Run("fn_error_is_returned", func(t *testing.T) {
		boom := errors.New("boom")
		if err := WithoutBroadcasting[Widget](nil, func(ctx context.Context) error {
			if ctx == nil {
				t.Error("nil ctx was not replaced")
			}
			return boom
		}); !errors.Is(err, boom) {
			t.Fatalf("err = %v", err)
		}
		if suppressed(nil, reflect.TypeFor[Widget]()) {
			t.Fatal("a nil ctx suppresses")
		}
	})
}

type bulkPayload struct {
	Reason string `json:"reason"`
	Count  int    `json:"count"`
}

// TestBulkEmitsOnce covers D-08 and RT-03: N suppressed creates plus one
// Emit in the same transaction publish exactly one summary event, with the
// payload bytes in declaration order, and nothing when the transaction
// rolls back.
func TestBulkEmitsOnce(t *testing.T) {
	env := newLHEnv(t, map[string]any{"realtime.broadcast_namespace": "Acme"}, nil)
	mem := env.memory(t)
	ctx := t.Context()

	from := len(mem.Publications())
	err := WithoutBroadcasting[Widget](ctx, func(ctx context.Context) error {
		return lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
			for i := range 3 {
				if err := tx.WithContext(ctx).Create(&Widget{Name: fmt.Sprintf("bulk-%d", i), OwnerID: 9}).Error; err != nil {
					return err
				}
			}
			return env.svc.Emit(ctx, tx, Broadcast{Channels: []string{"Widgets:9"}, Event: "collection.bulk_updated", Payload: bulkPayload{Reason: "bulk_create", Count: 3}})
		})
	})
	if err != nil {
		t.Fatal(err)
	}
	p := waitPublications(t, mem, from, 1)[0]
	if p.Event != "collection.bulk_updated" || p.Channels[0] != "acme:widgets:9" || p.Method != "publish" {
		t.Fatalf("publication = %+v", p)
	}
	if string(p.Payload) != `{"reason":"bulk_create","count":3}` {
		t.Fatalf("payload = %s, want the ordered bytes", p.Payload)
	}

	before := env.broadcastJobs(t)
	rollback := errors.New("rollback")
	err = lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
		if err := env.svc.Emit(ctx, tx, Broadcast{Channels: []string{"widgets:9"}, Event: "collection.bulk_updated", Payload: bulkPayload{Reason: "bulk_create", Count: 1}}); 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("an Emit in a rolled-back transaction left %d job(s)", after-before)
	}

	// Emit without a transaction enqueues on its own; several channels are
	// one broadcast.
	from = len(mem.Publications())
	if err := env.svc.Emit(nil, nil, Broadcast{Channels: []string{"a:1", "b:2"}, Event: "acme.pinged", Payload: nil}); err != nil {
		t.Fatal(err)
	}
	p = waitPublications(t, mem, from, 1)[0]
	if p.Method != "broadcast" || strings.Join(p.Channels, ",") != "acme:a:1,acme:b:2" || string(p.Payload) != "null" {
		t.Fatalf("publication = %+v payload %s", p, p.Payload)
	}

	if err := env.svc.Emit(ctx, env.gdb, Broadcast{Event: "x"}); err != nil {
		t.Fatalf("Emit without channels = %v, want nil", err)
	}
	if err := env.svc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}}); err == nil {
		t.Fatal("Emit without an event name succeeded")
	}
	if err := env.svc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}, Event: "x", Payload: make(chan int)}); err == nil {
		t.Fatal("Emit with an unencodable payload succeeded")
	}
	var nilSvc *Service
	if err := nilSvc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}, Event: "x"}); err == nil {
		t.Fatal("Emit on a nil service succeeded")
	}
}

// failingDriver fails every publication.
type failingDriver struct{ calls *int32 }

func (failingDriver) Name() string    { return "acme-failing" }
func (failingDriver) Routes() []Route { return nil }
func (d failingDriver) Publish(context.Context, string, string, json.RawMessage) error {
	return errors.New("realtime server is down")
}
func (d failingDriver) Broadcast(context.Context, []string, string, json.RawMessage) error {
	return errors.New("realtime server is down")
}

func init() {
	RegisterDriver("acme-failing", func(*backpack.App, *Service) (Driver, error) { return failingDriver{}, nil })
}

// TestBroadcastEdges covers the remaining broadcast rules: a zero-key batch
// update is skipped, updates and deletes broadcast (a delete through an
// id-only model carries the reloaded row), the method-based contract
// (alias, ttl, payload, filter), multi-channel broadcasts, the actor of a
// frontend and a backend principal, failures inside the savepoint that
// never abort the write, and a failed publish that is logged and not
// retried.
func TestBroadcastEdges(t *testing.T) {
	env := newLHEnv(t, nil, nil)
	name := "Ann"
	env.svc.SetUserLookup(func(_ context.Context, id uint) (User, bool, error) {
		switch id {
		case 7:
			return User{ID: 7, Name: &name}, true, nil
		case 8:
			return User{}, false, errors.New("lookup failed")
		}
		return User{}, false, nil
	})
	mem := env.memory(t)
	ctx := t.Context()

	w := Widget{Name: "edge", OwnerID: 4}
	if err := env.gdb.WithContext(ctx).Create(&w).Error; err != nil {
		t.Fatal(err)
	}
	waitPublications(t, mem, 0, 1)

	t.Run("zero_key_batch_update_is_skipped", func(t *testing.T) {
		before := env.broadcastJobs(t)
		if err := env.gdb.WithContext(ctx).Model(&Widget{}).Where("owner_id = ?", 4).Update("name", "batch").Error; err != nil {
			t.Fatal(err)
		}
		if after := env.broadcastJobs(t); after != before {
			t.Fatalf("a zero-key batch update enqueued %d job(s)", after-before)
		}
	})

	t.Run("update_carries_the_frontend_actor", func(t *testing.T) {
		from := len(mem.Publications())
		uctx := bouncer.WithUser(ctx, &bouncer.Principal{ID: 7})
		w.Name = "renamed"
		if err := env.gdb.WithContext(uctx).Save(&w).Error; err != nil {
			t.Fatal(err)
		}
		p := waitPublications(t, mem, from, 1)[0]
		var body struct {
			Actor Actor `json:"actor"`
		}
		if err := json.Unmarshal(p.Payload, &body); err != nil {
			t.Fatal(err)
		}
		if p.Event != "updated.lighthouse.widget" || body.Actor.UserID == nil || *body.Actor.UserID != 7 || body.Actor.Name == nil || *body.Actor.Name != "Ann" {
			t.Fatalf("publication %s payload %s", p.Event, p.Payload)
		}
	})

	t.Run("actor_rules", func(t *testing.T) {
		if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 3, Backend: true})); a.UserID != nil || *a.Name != "System" {
			t.Fatalf("backend actor = %+v, want System", a)
		}
		if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 8})); a.UserID == nil || *a.UserID != 8 || a.Name != nil {
			t.Fatalf("lookup-error actor = %+v, want the id and a null name", a)
		}
		if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 99})); a.UserID == nil || a.Name != nil {
			t.Fatalf("unknown-user actor = %+v", a)
		}
		raw, _ := json.Marshal(SystemActor())
		if string(raw) != `{"user_id":null,"name":"System"}` {
			t.Fatalf("SystemActor = %s", raw)
		}
	})

	t.Run("delete_of_an_id_only_model_carries_the_row", func(t *testing.T) {
		from := len(mem.Publications())
		if err := env.gdb.WithContext(ctx).Delete(&Widget{ID: w.ID}).Error; err != nil {
			t.Fatal(err)
		}
		p := waitPublications(t, mem, from, 1)[0]
		var body struct {
			Model Widget `json:"model"`
		}
		if err := json.Unmarshal(p.Payload, &body); err != nil {
			t.Fatal(err)
		}
		if p.Event != "deleted.lighthouse.widget" || p.Channels[0] != "widgets:4" || body.Model.Name != "renamed" {
			t.Fatalf("publication %s %v payload %s", p.Event, p.Channels, p.Payload)
		}
	})

	t.Run("method_contract_and_multi_channel", func(t *testing.T) {
		from := len(mem.Publications())
		s := Sprocket{Name: "multi", Channels: "Room:1,room:2"}
		if err := env.gdb.WithContext(ctx).Create(&s).Error; err != nil {
			t.Fatal(err)
		}
		p := waitPublications(t, mem, from, 1)[0]
		if p.Method != "broadcast" || strings.Join(p.Channels, ",") != "room:1,room:2" || p.Event != "created.acme.sprocket" {
			t.Fatalf("publication = %+v", p)
		}
		if !strings.HasPrefix(string(p.Payload), fmt.Sprintf(`{"id":%d,"action":"created","ttl":15,`, s.ID)) {
			t.Fatalf("payload = %s", p.Payload)
		}
		before := env.broadcastJobs(t)
		s.Name = "filtered"
		if err := env.gdb.WithContext(ctx).Save(&s).Error; err != nil {
			t.Fatal(err)
		}
		if after := env.broadcastJobs(t); after != before {
			t.Fatal("ShouldBroadcast(updated) = false did not veto the update")
		}
		if err := env.gdb.WithContext(ctx).Create(&Sprocket{Name: "silent"}).Error; err != nil {
			t.Fatal(err)
		}
		if after := env.broadcastJobs(t); after != before {
			t.Fatal("a model without channels enqueued a job")
		}
	})

	t.Run("failures_are_logged_and_never_abort_the_write", func(t *testing.T) {
		before := env.broadcastJobs(t)
		err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := tx.WithContext(ctx).Create(&Sprocket{Name: "chan-fail", Channels: "fail"}).Error; err != nil {
				return err
			}
			if err := tx.WithContext(ctx).Create(&Sprocket{Name: "bad-payload", Channels: "room:3"}).Error; err != nil {
				return err
			}
			// The transaction is still usable after both failures.
			return tx.WithContext(ctx).Create(&Widget{Name: "after-failures", OwnerID: 0}).Error
		})
		if err != nil {
			t.Fatalf("a broadcast failure aborted the write: %v", err)
		}
		var n int64
		if err := env.gdb.Model(&Sprocket{}).Where("name IN ?", []string{"chan-fail", "bad-payload"}).Count(&n).Error; err != nil || n != 2 {
			t.Fatalf("committed sprockets = %d (err %v), want 2", n, err)
		}
		if after := env.broadcastJobs(t); after != before {
			t.Fatalf("failed broadcasts enqueued %d job(s)", after-before)
		}
		if env.logs.count("realtime: broadcast skipped") < 2 {
			t.Fatalf("no skip warnings:\n%s", env.logs.all())
		}
	})

	t.Run("a_failed_query_inside_the_savepoint_keeps_the_transaction", func(t *testing.T) {
		err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := tx.WithContext(ctx).Create(&Sprocket{Name: "abort", Channels: "abort-tx"}).Error; err != nil {
				return err
			}
			return tx.WithContext(ctx).Create(&Widget{Name: "after-abort", OwnerID: 0}).Error
		})
		if err != nil {
			t.Fatalf("a failed query in the broadcast savepoint aborted the write: %v", err)
		}
		var n int64
		if err := env.gdb.Model(&Widget{}).Where("name = ?", "after-abort").Count(&n).Error; err != nil || n != 1 {
			t.Fatalf("write after the failed broadcast query: %d rows (err %v)", n, err)
		}
	})
}

// TestBroadcastSwallowedReadFailure covers a channel function that
// swallows a failed read: the savepoint must still be rolled back, or the
// failed statement leaves the caller's transaction aborted.
func TestBroadcastSwallowedReadFailure(t *testing.T) {
	env := newLHEnv(t, nil, nil)
	ctx := t.Context()
	err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
		if err := tx.WithContext(ctx).Create(&Sprocket{Name: "swallowed", Channels: "swallow"}).Error; err != nil {
			return err
		}
		return tx.WithContext(ctx).Create(&Widget{Name: "after-swallow", OwnerID: 0}).Error
	})
	if err != nil {
		t.Fatalf("a swallowed read failure in the broadcast savepoint aborted the write: %v", err)
	}
	// A single-statement write runs its callbacks inside GORM's own
	// transaction; the commit must still succeed.
	if err := env.gdb.WithContext(ctx).Create(&Sprocket{Name: "swallowed-implicit", Channels: "swallow"}).Error; err != nil {
		t.Fatalf("single-statement write with a swallowed read failure: %v", err)
	}
	var n int64
	if err := env.gdb.Model(&Sprocket{}).Where("name LIKE ?", "swallowed%").Count(&n).Error; err != nil || n != 2 {
		t.Fatalf("committed sprockets = %d (err %v), want 2", n, err)
	}
}

// TestBroadcastPublishFailure covers D-09: a failed publish is logged at
// Warn without the payload and the one-attempt job is not retried.
func TestBroadcastPublishFailure(t *testing.T) {
	env := newLHEnv(t, map[string]any{"realtime.driver": "acme-failing"}, nil)
	ctx := t.Context()
	if err := env.gdb.WithContext(ctx).Create(&Widget{Name: "secret-payload-value", OwnerID: 6}).Error; err != nil {
		t.Fatal(err)
	}
	deadline := time.Now().Add(10 * time.Second)
	for env.logs.count("realtime: broadcast failed") == 0 {
		if time.Now().After(deadline) {
			t.Fatalf("no failure log:\n%s", env.logs.all())
		}
		time.Sleep(20 * time.Millisecond)
	}
	rec, _ := env.logs.find("realtime: broadcast failed")
	if rec.level != slog.LevelWarn || !strings.Contains(rec.attrs["channels"], "widgets:6") || strings.Contains(env.logs.all(), "secret-payload-value") {
		t.Fatalf("failure log = %+v", rec)
	}
	var state string
	var attempt int
	for time.Now().Before(deadline) {
		if err := env.db.QueryRowContext(ctx, `SELECT state::text, attempt FROM river_job WHERE kind = 'summer.broadcast' ORDER BY id DESC LIMIT 1`).Scan(&state, &attempt); err != nil {
			t.Fatal(err)
		}
		if state == "completed" {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if state != "completed" || attempt != 1 {
		t.Fatalf("job state %q attempt %d, want completed after one attempt (never retried)", state, attempt)
	}
}