feat: announce material revisions on materials:{id}

- every revision bump (API writes and /bmadmin saves) enqueues
  updated.jz.bm.material with {material_id, revision} in a savepoint
- the materials namespace authorizer allows a subscribe only to a
  visible material the user may manage
- api.ManagesMaterials and api.CanWatchMaterial

classes/material_revision.go

@@ -0,0 +1,92 @@
+package classes
+
+import (
+	"context"
+	"log/slog"
+	"regexp"
+	"strconv"
+
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+	"gorm.io/gorm"
+)
+
+// The live heads-up of a material's revision (D-20): an open editor learns
+// that a newer revision was saved before its own save, which still meets
+// the 409. Only ids and the revision number travel on the channel, never
+// content.
+const (
+	// MaterialsNamespace is the realtime namespace jz.bm owns.
+	MaterialsNamespace = "materials"
+	// MaterialRevisedEvent is the event published after a content write.
+	MaterialRevisedEvent = "updated.jz.bm.material"
+)
+
+// materialChannelRE is the only accepted form of a material channel:
+// lighthouse.ChannelID reads segments with a PHP-style cast, which would
+// take "materials:5abc" for material 5.
+var materialChannelRE = regexp.MustCompile(`^materials:[1-9][0-9]{0,18}$`)
+
+// MaterialChannel is the channel of material id.
+func MaterialChannel(id uint) string {
+	return MaterialsNamespace + ":" + strconv.FormatUint(uint64(id), 10)
+}
+
+// MaterialChannelID returns the material id of a channel of the form
+// materials:{id}, and false for any other channel (a presence: prefix, a
+// leading zero, trailing characters, a third segment).
+func MaterialChannelID(channel string) (uint, bool) {
+	if !materialChannelRE.MatchString(channel) {
+		return 0, false
+	}
+	n, err := strconv.ParseUint(channel[len(MaterialsNamespace)+1:], 10, strconv.IntSize)
+	if err != nil || n == 0 {
+		return 0, false
+	}
+	return uint(n), true
+}
+
+// MaterialRevised is the payload of MaterialRevisedEvent. The field order
+// is fixed so the payload's key order is stable.
+type MaterialRevised struct {
+	MaterialID uint `json:"material_id"`
+	Revision   uint `json:"revision"`
+}
+
+// MaterialRevisionNotifier publishes MaterialRevisedEvent after a write
+// that bumped a material's revision. A nil notifier, or one without a
+// realtime service, publishes nothing.
+type MaterialRevisionNotifier struct {
+	// Realtime is the app's lighthouse service; nil turns the event off.
+	Realtime *lighthouse.Service
+	// Log is where enqueue failures are logged; nil means slog.Default.
+	Log *slog.Logger
+}
+
+// Revised enqueues the event of material id at revision on the write's
+// transaction tx, so it is published only when that write commits. The
+// enqueue runs in a savepoint: a failure rolls back the savepoint only, is
+// logged and never fails or loses the write. Without a transaction it does
+// nothing.
+func (n *MaterialRevisionNotifier) Revised(ctx context.Context, tx *gorm.DB, id, revision uint) {
+	if n == nil || n.Realtime == nil || tx == nil {
+		return
+	}
+	b := lighthouse.Broadcast{
+		Channels: []string{MaterialChannel(id)},
+		Event:    MaterialRevisedEvent,
+		Payload:  MaterialRevised{MaterialID: id, Revision: revision},
+	}
+	err := tx.Transaction(func(sp *gorm.DB) error {
+		return n.Realtime.Emit(ctx, sp, b)
+	})
+	if err != nil {
+		n.log().Warn("jz.bm: material revision event not queued", "material_id", id, "revision", revision, "error", err)
+	}
+}
+
+func (n *MaterialRevisionNotifier) log() *slog.Logger {
+	if n.Log != nil {
+		return n.Log
+	}
+	return slog.Default()
+}

classes/material_revision_test.go

@@ -0,0 +1,48 @@
+package classes
+
+import (
+	"context"
+	"testing"
+
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+	"gorm.io/gorm"
+)
+
+func TestMaterialChannel(t *testing.T) {
+	for _, id := range []uint{1, 5, 42, 1234567890} {
+		ch := MaterialChannel(id)
+		got, ok := MaterialChannelID(ch)
+		if !ok || got != id {
+			t.Errorf("MaterialChannelID(%q) = %d, %v, want %d, true", ch, got, ok, id)
+		}
+	}
+	if got := MaterialChannel(5); got != "materials:5" {
+		t.Errorf("MaterialChannel(5) = %q", got)
+	}
+	for _, ch := range []string{
+		"materials:5abc",
+		"materials:05",
+		"materials:0",
+		"materials:5:x",
+		"presence:materials:5",
+		"materials:",
+		"materials",
+		"materials:-5",
+		"materials: 5",
+		"materials:99999999999999999999",
+		"user:5",
+		"",
+	} {
+		if id, ok := MaterialChannelID(ch); ok {
+			t.Errorf("MaterialChannelID(%q) = %d, true, want a refusal", ch, id)
+		}
+	}
+}
+
+func TestMaterialRevisionNotifierNil(t *testing.T) {
+	ctx := context.Background()
+	var nilNotifier *MaterialRevisionNotifier
+	nilNotifier.Revised(ctx, &gorm.DB{}, 5, 2)
+	(&MaterialRevisionNotifier{}).Revised(ctx, &gorm.DB{}, 5, 2)
+	(&MaterialRevisionNotifier{Realtime: &lighthouse.Service{}}).Revised(ctx, nil, 5, 2)
+}

controllers/api/api.go

@@ -78,6 +78,9 @@ type Handlers struct {
 	// means a search (q) answers 503 and the syncs after a write are
 	// skipped.
 	MaterialIndex *classes.MaterialIndexer
+	// Revisions publishes the live heads-up after a write that bumped a
+	// material's revision (D-20). Nil means no live event.
+	Revisions *classes.MaterialRevisionNotifier
 }
 
 func (h *Handlers) now() time.Time {

controllers/api/material_access.go

@@ -170,6 +170,12 @@ func replaceMaterialShares(tx *gorm.DB, materialID uint, userIDs []uint) error {
 
 // managesMaterials tells whether the user holds jz.bm.access_materials.
 func (h *Handlers) managesMaterials(ctx context.Context, db *gorm.DB, userID uint) (bool, error) {
+	return ManagesMaterials(ctx, db, userID)
+}
+
+// ManagesMaterials tells whether the user holds jz.bm.access_materials. The
+// realtime connection-token guard admits such users (D-20).
+func ManagesMaterials(ctx context.Context, db *gorm.DB, userID uint) (bool, error) {
 	principal, err := userclasses.PrincipalByID(ctx, db, userID)
 	if err != nil {
 		return false, err
@@ -177,6 +183,26 @@ func (h *Handlers) managesMaterials(ctx context.Context, db *gorm.DB, userID uin
 	return principal != nil && cabana.Allows(principal, []string{controllers.PermMaterials}), nil
 }
 
+// CanWatchMaterial tells whether the user may follow material id's live
+// revision channel (D-20): the material is visible to them at now and they
+// may manage it (canManageMaterial). It reads current state on every call;
+// a material they cannot see is false without an error.
+func CanWatchMaterial(ctx context.Context, db *gorm.DB, now time.Time, userID, id uint) (bool, error) {
+	var m models.Material
+	res := visibleMaterials(db.WithContext(ctx), now, userID).Where("jz_bm_materials.id = ?", id).Limit(1).Find(&m)
+	if res.Error != nil {
+		return false, res.Error
+	}
+	if res.RowsAffected == 0 {
+		return false, nil
+	}
+	manager, err := ManagesMaterials(ctx, db, userID)
+	if err != nil {
+		return false, err
+	}
+	return canManageMaterial(m, userID, manager), nil
+}
+
 // canManageMaterial tells whether a material manager may change the
 // material's access: it is theirs, or it has no author yet (the first change
 // makes them its author).

controllers/api/material_revision.go

@@ -27,10 +27,18 @@ func staleRevision(sent *uint, stored uint) bool {
 
 // revise bumps the revision of material id inside tx when its content
 // differs from before (models.ReviseMaterial). Every content writer of the
-// API calls it after its write.
+// API calls it after its write. A bump also enqueues the live heads-up on
+// tx (D-20), published only when the write commits; its failure never fails
+// the write.
 func (h *Handlers) revise(ctx context.Context, tx *gorm.DB, id uint, before string) error {
-	_, _, err := models.ReviseMaterial(tx.WithContext(ctx), id, before)
-	return err
+	rev, bumped, err := models.ReviseMaterial(tx.WithContext(ctx), id, before)
+	if err != nil {
+		return err
+	}
+	if bumped {
+		h.Revisions.Revised(ctx, tx, id, rev)
+	}
+	return nil
 }
 
 // writeConflict answers 409 to a write that sent a stale revision, with the

controllers/materials_hooks.go

@@ -50,6 +50,29 @@ func installedMaterialIndex() MaterialIndex {
 	return h.ix
 }
 
+// MaterialRevisionNotifier publishes the live heads-up after an admin save
+// bumped a material's revision (classes.MaterialRevisionNotifier). It
+// enqueues on tx and never fails the save.
+type MaterialRevisionNotifier interface {
+	Revised(ctx context.Context, tx *gorm.DB, id, revision uint)
+}
+
+type materialRevisionNotifierHolder struct{ n MaterialRevisionNotifier }
+
+var materialRevisionNotifier atomic.Value
+
+// UseMaterialRevisionNotifier installs the notifier the materials admin
+// calls after a save bumped the revision; nil turns it off. Boot installs
+// it, as UseMaterialIndex.
+func UseMaterialRevisionNotifier(n MaterialRevisionNotifier) {
+	materialRevisionNotifier.Store(materialRevisionNotifierHolder{n: n})
+}
+
+func installedMaterialRevisionNotifier() MaterialRevisionNotifier {
+	h, _ := materialRevisionNotifier.Load().(materialRevisionNotifierHolder)
+	return h.n
+}
+
 // FormAfterCreate queues the file text job in the save transaction (the
 // form's deferred files commit in the same one) and syncs the new material
 // once it commits.
@@ -88,7 +111,8 @@ func (c materialsAdmin) FormBeforeUpdate(ctx context.Context, model any) error {
 }
 
 // FormAfterUpdate bumps the revision when the save changed the material's
-// content, then queues the file text job in the save transaction and syncs
+// content, and then enqueues the live heads-up on the save transaction
+// (D-20), then queues the file text job in the save transaction and syncs
 // the material once it commits. The compare runs here, not in
 // FormBeforeUpdate, because the course, the zjazd and the tags are final
 // only after the row write and the pivot sync. A revise error fails the save
@@ -102,9 +126,13 @@ func (materialsAdmin) FormAfterUpdate(ctx context.Context, model any) error {
 	}
 	if pending, found := pendingRevisions.LoadAndDelete(m.ID); found {
 		if tx, ok := cabana.TxFromContext(ctx); ok {
-			if _, _, err := models.ReviseMaterial(tx, m.ID, pending.(string)); err != nil {
+			rev, bumped, err := models.ReviseMaterial(tx, m.ID, pending.(string))
+			if err != nil {
 				return err
 			}
+			if n := installedMaterialRevisionNotifier(); bumped && n != nil {
+				n.Revised(ctx, tx, m.ID, rev)
+			}
 		}
 	}
 	return materialAfterCommit(ctx, model, false)

controllers/materials_hooks_test.go

@@ -36,6 +36,14 @@ func (r *recordingIndex) QueueExtract(_ context.Context, tx *gorm.DB, id uint) {
 
 func itoa(id uint) string { return strconv.FormatUint(uint64(id), 10) }
 
+// recordingNotifier records the live revision events of the materials
+// admin.
+type recordingNotifier struct{ calls []string }
+
+func (r *recordingNotifier) Revised(_ context.Context, _ *gorm.DB, id, revision uint) {
+	r.calls = append(r.calls, itoa(id)+"@"+itoa(revision))
+}
+
 func TestMaterialsAdminHooks(t *testing.T) {
 	t.Cleanup(func() { UseMaterialIndex(nil) })
 	var c materialsAdmin
@@ -95,8 +103,13 @@ func TestMaterialsAdminHooks(t *testing.T) {
 // store no snapshot, a model that is not a material is an error, and the
 // embedded date check still runs first.
 func TestMaterialsAdminRevision(t *testing.T) {
-	t.Cleanup(func() { UseMaterialIndex(nil) })
+	t.Cleanup(func() {
+		UseMaterialIndex(nil)
+		UseMaterialRevisionNotifier(nil)
+	})
 	UseMaterialIndex(nil)
+	notifier := &recordingNotifier{}
+	UseMaterialRevisionNotifier(notifier)
 	var c materialsAdmin
 	ctx := context.Background()
 	m := &models.Material{ID: 41, Title: "Notatka", Type: "notatka"}
@@ -110,6 +123,18 @@ func TestMaterialsAdminRevision(t *testing.T) {
 	if err := c.FormAfterUpdate(ctx, m); err != nil {
 		t.Errorf("after update without a transaction: %v", err)
 	}
+	// Outside a write nothing is revised, so nothing is announced.
+	if len(notifier.calls) != 0 {
+		t.Errorf("notifier called outside a transaction: %v", notifier.calls)
+	}
+	// Turning the notifier off is safe.
+	UseMaterialRevisionNotifier(nil)
+	if n := installedMaterialRevisionNotifier(); n != nil {
+		t.Errorf("notifier after UseMaterialRevisionNotifier(nil) = %v, want nil", n)
+	}
+	if err := c.FormAfterUpdate(ctx, m); err != nil {
+		t.Errorf("after update without a notifier: %v", err)
+	}
 
 	if err := c.FormBeforeUpdate(ctx, &models.Course{ID: 41}); err == nil {
 		t.Error("before update with a course: no error")

material_watch_postgres_test.go

@@ -0,0 +1,185 @@
+package bm_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strconv"
+	"testing"
+	"time"
+
+	userclasses "git.golem15.com/golem15/sm-user-plugin/classes"
+	"git.golem15.com/golem15/summercms/modules/surf"
+	"git.golem15.com/jakub/sm-bm-plugin/classes"
+	"git.golem15.com/jakub/sm-bm-plugin/controllers/api"
+	"git.golem15.com/jakub/sm-bm-plugin/models"
+	"gorm.io/gorm"
+)
+
+// The realtime config of the material watch test: the Centrifugo driver, so
+// Emit enqueues broadcast jobs.
+const (
+	watchProxySecret = "material-watch-proxy-secret-0123456789"
+	watchTokenSecret = "bm-test-only-material-watch-token-secret-32b"
+	watchAppKey      = "bm-test-only-material-watch-app-key"
+)
+
+// TestMaterialWatchAgainstDatabase checks the live revision heads-up (D-20)
+// against Postgres: who may follow materials:{id} (api.CanWatchMaterial),
+// and that a management PUT which changes the content queues exactly one
+// broadcast on the material's channel while a visibility-only PUT queues
+// none.
+func TestMaterialWatchAgainstDatabase(t *testing.T) {
+	app, plugins, db := migratedWith(t,
+		"SUMMER_REALTIME__DRIVER=centrifugo",
+		"SUMMER_REALTIME__CENTRIFUGO__PROXY_SECRET="+watchProxySecret,
+		"SUMMER_REALTIME__CENTRIFUGO__TOKEN_SECRET="+watchTokenSecret,
+		"SUMMER_APP__KEY="+watchAppKey,
+	)
+	seedFixture(t, plugins, "testdata/seed.example.json")
+	var course models.Course
+	if err := db.Order("id").First(&course).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	author := createUser(t, db, "bm-watch-author@example.test")
+	other := createUser(t, db, "bm-watch-other@example.test")
+	student := createUser(t, db, "bm-watch-student@example.test")
+	var groupID uint
+	if err := db.Raw(`
+INSERT INTO user_groups (name, code, permissions, created_at, updated_at)
+VALUES ('BM materials', 'bm-materials', '{"jz.bm.access_materials":1}', NOW(), NOW())
+RETURNING id`).Scan(&groupID).Error; err != nil {
+		t.Fatal(err)
+	}
+	for _, id := range []uint{author, other} {
+		if err := db.Exec(`INSERT INTO users_groups (user_id, user_group_id) VALUES (?, ?)`, id, groupID).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	now := time.Now()
+	past := now.Add(-time.Hour)
+	insert := func(title string, published *time.Time, private bool, authorID *uint) uint {
+		t.Helper()
+		var id uint
+		err := db.Raw(`
+INSERT INTO jz_bm_materials (course_id, title, type, body, published_at, is_private, author_user_id, created_at, updated_at)
+VALUES (?, ?, 'notatka', 'tekst', ?, ?, ?, NOW(), NOW())
+RETURNING id`, course.ID, title, published, private, authorID).Scan(&id).Error
+		if err != nil {
+			t.Fatal(err)
+		}
+		return id
+	}
+	authored := insert("Autorska", &past, false, &author)
+	orphan := insert("Bez autora", &past, false, nil)
+	hidden := insert("Prywatna", &past, true, nil)
+	draft := insert("Szkic", nil, false, nil)
+
+	for _, c := range []struct {
+		name   string
+		userID uint
+		id     uint
+		want   bool
+	}{
+		{"the author manager", author, authored, true},
+		{"another manager on an authored material", other, authored, false},
+		{"any manager on a material with no author", other, orphan, true},
+		{"a non-manager", student, orphan, false},
+		{"a private material the user cannot see", other, hidden, false},
+		{"a draft", other, draft, false},
+		{"an unknown material", author, draft + 1000, false},
+	} {
+		got, err := api.CanWatchMaterial(t.Context(), db, now, c.userID, c.id)
+		if err != nil {
+			t.Errorf("%s: %v", c.name, err)
+			continue
+		}
+		if got != c.want {
+			t.Errorf("%s: CanWatchMaterial = %v, want %v", c.name, got, c.want)
+		}
+	}
+
+	secret, token, err := userclasses.MintAPIToken("usr_", author, "BM watch", []string{"read", "write"}, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(token).Error; err != nil {
+		t.Fatal(err)
+	}
+	handler, err := surf.Assemble(app, plugins)
+	if err != nil {
+		t.Fatal(err)
+	}
+	send := func(method, path string, payload any) (int, map[string]any) {
+		t.Helper()
+		raw, err := json.Marshal(payload)
+		if err != nil {
+			t.Fatal(err)
+		}
+		req := httptest.NewRequest(method, path, bytes.NewReader(raw))
+		req.Header.Set("Authorization", "Bearer "+secret)
+		req.Header.Set("Content-Type", "application/json")
+		rec := httptest.NewRecorder()
+		handler.ServeHTTP(rec, req)
+		var out map[string]any
+		_ = json.Unmarshal(rec.Body.Bytes(), &out)
+		return rec.Code, out
+	}
+
+	note := map[string]any{"title": "Na żywo", "type": "notatka", "course_id": course.ID, "body": "pierwsza\n"}
+	code, out := send(http.MethodPost, "/api/v1/bm/materials", note)
+	if code != http.StatusCreated {
+		t.Fatalf("create = %d %v", code, out)
+	}
+	d, _ := out["data"].(map[string]any)
+	id := uint(d["id"].(float64))
+	path := "/api/v1/bm/materials/" + strconv.FormatUint(uint64(id), 10)
+	channel := classes.MaterialChannel(id)
+	if n := revisionJobs(t, db, channel); n != 0 {
+		t.Fatalf("create queued %d revision broadcasts, want 0", n)
+	}
+
+	note["body"] = "druga\n"
+	if code, out := send(http.MethodPut, path, note); code != http.StatusOK {
+		t.Fatalf("content PUT = %d %v", code, out)
+	}
+	if n := revisionJobs(t, db, channel); n != 1 {
+		t.Fatalf("content PUT queued %d revision broadcasts on %s, want 1", n, channel)
+	}
+
+	note["is_private"] = true
+	if code, out := send(http.MethodPut, path, note); code != http.StatusOK {
+		t.Fatalf("visibility PUT = %d %v", code, out)
+	}
+	if n := revisionJobs(t, db, channel); n != 1 {
+		t.Errorf("visibility PUT changed the revision broadcasts to %d, want still 1", n)
+	}
+}
+
+// revisionJobs counts the queued broadcast jobs that announce a material
+// revision on channel.
+func revisionJobs(t *testing.T, db *gorm.DB, channel string) int {
+	t.Helper()
+	var rows []string
+	if err := db.Raw(`SELECT args::text FROM river_job WHERE kind = 'summer.broadcast'`).Scan(&rows).Error; err != nil {
+		t.Fatal(err)
+	}
+	n := 0
+	for _, raw := range rows {
+		var args struct {
+			Channels []string `json:"channels"`
+			Event    string   `json:"event"`
+		}
+		if err := json.Unmarshal([]byte(raw), &args); err != nil {
+			t.Fatal(err)
+		}
+		if args.Event == classes.MaterialRevisedEvent && slices.Contains(args.Channels, channel) {
+			n++
+		}
+	}
+	return n
+}

plugin.go

@@ -66,6 +66,13 @@ type Plugin struct {
 	// chatMember is the chat-group check of the realtime access guard; nil
 	// means classes.IsChatMember. Tests inject their own.
 	chatMember classes.MemberFunc
+	// watchMaterial is the materials channel check of the realtime
+	// authorizer; nil means api.CanWatchMaterial at time.Now. Tests inject
+	// their own.
+	watchMaterial func(ctx context.Context, db *gorm.DB, userID, id uint) (bool, error)
+	// materialManager is the material-manager check of the realtime access
+	// guard; nil means api.ManagesMaterials. Tests inject their own.
+	materialManager func(ctx context.Context, db *gorm.DB, userID uint) (bool, error)
 	// materialIndex searches the materials index and keeps it current. Nil
 	// when Boot did not run: then search answers 503 and syncs are skipped.
 	materialIndex *classes.MaterialIndexer

realtime.go

@@ -7,6 +7,8 @@ import (
 	usermodels "git.golem15.com/golem15/sm-user-plugin/models"
 	"git.golem15.com/golem15/summercms/modules/backpack"
 	"git.golem15.com/golem15/summercms/modules/lighthouse"
+	"git.golem15.com/jakub/sm-bm-plugin/classes"
+	"git.golem15.com/jakub/sm-bm-plugin/controllers"
 	"gorm.io/gorm"
 
 	// The app always ships the Centrifugo driver; config/realtime.yaml
@@ -19,15 +21,19 @@ import (
 var errDatabaseUnavailable = errors.New("jz.bm: database unavailable")
 
 // wireRealtime binds the app to the framework realtime service: it builds
-// the service selected by realtime.driver and installs the user lookup the
-// connection token reads its display name from. Channel authorizers belong
-// to the plugins that own the channels (golem15.presentations).
+// the service selected by realtime.driver, installs the user lookup the
+// connection token reads its display name from and the notifier the
+// materials admin announces revisions with, and registers the authorizer of
+// the materials namespace. Channel authorizers belong to the plugins that
+// own the channels: jz.bm owns materials (classes.MaterialsNamespace),
+// golem15.presentations its own.
 func (p *Plugin) wireRealtime(app *backpack.App) error {
 	svc, err := lighthouse.From(app)
 	if err != nil {
 		return err
 	}
 	p.realtime = svc
+	controllers.UseMaterialRevisionNotifier(&classes.MaterialRevisionNotifier{Realtime: svc})
 	svc.SetUserLookup(func(ctx context.Context, id uint) (lighthouse.User, bool, error) {
 		// Boot runs before serve publishes the database, so the handle is
 		// resolved per call.
@@ -45,5 +51,5 @@ func (p *Plugin) wireRealtime(app *backpack.App) error {
 		}
 		return lighthouse.User{ID: u.ID, Name: u.Name}, true, nil
 	})
-	return nil
+	return p.registerMaterialsRealtime(svc)
 }

realtime_materials.go

@@ -0,0 +1,50 @@
+package bm
+
+import (
+	"context"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+	"git.golem15.com/jakub/sm-bm-plugin/classes"
+	"git.golem15.com/jakub/sm-bm-plugin/controllers/api"
+	"gorm.io/gorm"
+)
+
+// registerMaterialsRealtime registers the authorizer of the materials
+// namespace: channel materials:{id} carries the live revision heads-up of
+// one material (D-20) and is open only to whoever may manage it.
+func (p *Plugin) registerMaterialsRealtime(svc *lighthouse.Service) error {
+	return svc.Registry().Register(classes.MaterialsNamespace, lighthouse.AuthorizerFunc(p.authorizeMaterial))
+}
+
+// authorizeMaterial allows a subscribe to materials:{id} only when the
+// material is visible to the user and they may manage it (jz.bm.access_materials
+// and its author, or any such user while it has no author). It checks the
+// current state on every subscribe. The deny reasons are for the logs.
+func (p *Plugin) authorizeMaterial(ctx context.Context, userID uint, channel string) lighthouse.Result {
+	id, ok := classes.MaterialChannelID(channel)
+	if !ok {
+		return lighthouse.Denied("malformed material channel")
+	}
+	db := p.database()
+	if db == nil {
+		return lighthouse.Denied("database unavailable")
+	}
+	allowed, err := p.canWatchMaterial(ctx, db, userID, id)
+	if err != nil {
+		return lighthouse.Denied("material lookup failed")
+	}
+	if !allowed {
+		return lighthouse.Denied("not the material's manager")
+	}
+	return lighthouse.Allowed(nil)
+}
+
+// canWatchMaterial runs the watchMaterial seam, or api.CanWatchMaterial at
+// the current time by default.
+func (p *Plugin) canWatchMaterial(ctx context.Context, db *gorm.DB, userID, id uint) (bool, error) {
+	if p.watchMaterial != nil {
+		return p.watchMaterial(ctx, db, userID, id)
+	}
+	return api.CanWatchMaterial(ctx, db, time.Now(), userID, id)
+}

realtime_materials_test.go

@@ -0,0 +1,82 @@
+package bm
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/compass"
+	"gorm.io/gorm"
+)
+
+// watchPlugin builds jz.bm on an app with a published (fake) database
+// unless noDB, and the given materials channel check.
+func watchPlugin(t *testing.T, noDB bool, watch func(context.Context, *gorm.DB, uint, uint) (bool, error)) *Plugin {
+	t.Helper()
+	cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Environ: []string{"SUMMER_ENV=testing"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	app := backpack.New(cfg)
+	if !noDB {
+		if err := app.Publish[*gorm.DB](&gorm.DB{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	return &Plugin{app: app, watchMaterial: watch}
+}
+
+func TestAuthorizeMaterial(t *testing.T) {
+	ctx := context.Background()
+	type call struct{ userID, id uint }
+	var calls []call
+	watch := func(_ context.Context, _ *gorm.DB, userID, id uint) (bool, error) {
+		calls = append(calls, call{userID, id})
+		switch userID {
+		case 7:
+			return true, nil
+		case 13:
+			return false, errors.New("database down")
+		}
+		return false, nil
+	}
+
+	p := watchPlugin(t, false, watch)
+	if res := p.authorizeMaterial(ctx, 7, "materials:5"); !res.Allowed {
+		t.Errorf("manager on materials:5 denied: %s", res.Reason())
+	}
+	if len(calls) != 1 || calls[0] != (call{7, 5}) {
+		t.Errorf("seam calls %v, want [{7 5}]", calls)
+	}
+
+	for _, ch := range []string{"materials:5abc", "materials:05", "materials:0", "materials:5:x", "presence:materials:5", "materials:", "materials"} {
+		calls = nil
+		res := p.authorizeMaterial(ctx, 7, ch)
+		if res.Allowed {
+			t.Errorf("%q allowed", ch)
+		}
+		if res.Reason() != "malformed material channel" {
+			t.Errorf("%q: reason %q", ch, res.Reason())
+		}
+		if len(calls) != 0 {
+			t.Errorf("%q called the seam: %v", ch, calls)
+		}
+	}
+
+	if res := p.authorizeMaterial(ctx, 13, "materials:5"); res.Allowed || res.Reason() != "material lookup failed" {
+		t.Errorf("seam error: allowed %v, reason %q", res.Allowed, res.Reason())
+	}
+	if res := p.authorizeMaterial(ctx, 8, "materials:5"); res.Allowed || res.Reason() != "not the material's manager" {
+		t.Errorf("non-manager: allowed %v, reason %q", res.Allowed, res.Reason())
+	}
+
+	calls = nil
+	noDB := watchPlugin(t, true, watch)
+	if res := noDB.authorizeMaterial(ctx, 7, "materials:5"); res.Allowed || res.Reason() != "database unavailable" {
+		t.Errorf("no database: allowed %v, reason %q", res.Allowed, res.Reason())
+	}
+	if len(calls) != 0 {
+		t.Errorf("no database called the seam: %v", calls)
+	}
+}

routes.go

@@ -342,6 +342,7 @@ func (p *Plugin) mountRealtime(r pact.Router) error {
 func (p *Plugin) handlers() *api.Handlers {
 	h := &api.Handlers{
 		MaterialIndex: p.materialIndex,
+		Revisions:     &classes.MaterialRevisionNotifier{Realtime: p.realtime},
 		DB: func() *gorm.DB {
 			if p.app == nil {
 				return nil