11-02: this plan registers the broadcast job through the conga.Manager API from 11-01 while 11-02 adds periodic jobs inside conga; no shared files, order-independent
tokens
raw_tokens
tasks
confidence
180000
180000
3
low
truths
artifacts
key_links
prohibitions
Per D-11, `modules/lighthouse` is transport-neutral: it owns the Publisher/Driver interfaces, the channel-namespace authorizer Registry, channel rules, Route/Surface/Mount, the Broadcastable contract with WithoutBroadcasting and Emit, and the broadcast River job; drivers are chosen by `realtime.driver` (centrifugo, memory, log, null; default null) and models and authorizers never import a driver package.
Per D-12 and RT-01, `GET /api/realtime/token` behind `jwt.auth` returns 200 `{"token":...}` signed HS256 with `realtime.centrifugo.token_secret` whose decoded claims are exactly `sub` (the user id as a string), `exp` (now + token_ttl, default 3600) and `info` = {"name": the user's name or null}; with an empty token_secret it returns 503 `{"error":"WebSocket not configured"}`, and without a resolvable user 401 `{"error":"Unauthorized"}`.
Per D-12, the token issuer ports all five PHP JwtTokenGenerator generators (ForUser, Subscription, Anonymous with exp now+300 and sub "", ForIdentifier with info encoded as `[]` when empty, SubscriptionForIdentifier) and refuses to sign with an empty secret.
Per D-12 and RT-02, `POST /api/realtime/subscribe` always answers HTTP 200: an empty or mismatched `X-Centrifugo-Secret` (constant-time compare; an empty configured proxy_secret denies), an empty or "0" user, a double `presence:` prefix, more than three segments, an unknown namespace or an authorizer denial all return exactly `{"error":{"code":403,"message":"Access denied"}}` with the reason logged only; an allow returns `{"result":{"info":[]}}` (object info when non-empty) and, for a `presence:` channel, also `allow` (default ["prs"]) and `override` merging presence/join_leave true and force_push_join_leave false with the authorizer's overrides.
Per D-13, routes are declared by the driver as `lighthouse.Route{Name, Method, Path, Surface, Handler}` and mounted by the app with one `lighthouse.Mount(r, driver, lighthouse.Surfaces{UserAuth, ServerToServer, Public, Middleware})` call; fonoteka mounts the token route with `jwt.auth` before `throttle:ws-api` (user decision 5) and the subscribe route in a raw group with `throttle:ws-api`; Mount refuses a UserAuth route when the UserAuth surface is empty.
Per D-14, fonoteka registers the `ws-api` bucket: 120 requests per minute keyed by the authenticated user id, else by the trusted-proxy client IP.
Per D-16, fonoteka registers `collection` and `wishlist` authorizers ported from CollectionChannelAuthorizer (owner or editor of a kind=collection collection) and WishlistChannelAuthorizer (a wishlist_subscriptions row, or a kind=wishlist collection owned by a household peer), both parsing the id segment with PHP (int)-cast semantics; there is no separate websockets plugin and PROJECT.md records that deviation.
Per D-06, D-07 and RT-03, a create, update or delete of a broadcastable model enqueues one River job on the `realtime.broadcast_queue` queue (default `broadcasts`) inside the write transaction, with MaxAttempts 1 and timeout `realtime.broadcast_timeout` (default 5s); nothing is published when the write rolls back; a failed publish or a failed enqueue is logged as a warning and never fails the write.
Per D-08, `lighthouse.WithoutBroadcasting[T](ctx, fn)` silences only T's broadcasts for writes made with the ctx handed to fn (other types still broadcast, a stale outer ctx is not suppressed), and `(*lighthouse.Service).Emit` enqueues one explicit summary event in the same transaction, so N album creates under suppression plus one Emit publish exactly one `collection.bulk_updated`.
Per D-09, the event name is `{action}.{alias}` lowercased (alias defaults to `<plugin>.<model>` from the Go package path and can be overridden), the default payload is `{model, actor, timestamp, ttl}` with ttl 60, a delete snapshot is taken before the row is deleted, `ShouldBroadcast(action)` can veto, one channel uses Centrifugo publish and several use broadcast, and channels are lowercased and prefixed with `realtime.broadcast_namespace` unless already prefixed.
Per D-12, the Centrifugo HTTP client POSTs `{api_url}/publish` with `{"channel":c,"data":{"event":e,"payload":p,"timestamp":"...+00:00"}}`, `/broadcast` with `channels`, `/presence` and `/unsubscribe` with header `Authorization: apikey <api_key>` and a 5s timeout; success is any 2xx; with an empty api_key it sends nothing.
Album broadcasts match PHP Album overrides: channels `["collection:<id>"]` only when the album's collection has kind `collection` (else none), payload `{id, collection_id, action, actor, timestamp}` plus `album` (the current classes.SerializeAlbum of the reloaded row) except for deleted, actor `{user_id, name}` from the frontend principal or `{user_id: null, name: "System"}` for no principal or a backend admin.
Edge (RT-01 concurrency): concurrent token requests for one user each get an independently signed token with no shared mutable state (race detector clean).
statement
verification
Edge (RT-01 concurrency): delivery order across separate broadcast jobs is not guaranteed, as with PHP's queued BroadcastEventJob.
backstop
Edge (RT-02 adjacency): a two-segment `collection:5` and a three-segment `ns:entity:id` channel reach their authorizer, four segments deny, one `presence:` prefix is stripped before the namespace lookup while the authorizer receives the full original channel, and `presence:presence:` denies.
Edge (RT-02 empty): a missing or empty user, user "0", a missing or empty channel, and an empty namespace all return the generic HTTP 200 deny (a missing channel, which Centrifugo never sends, is denied instead of PHP's TypeError 500 and is recorded as deliberate hardening).
Edge (RT-02 encoding): namespace lookup is byte-exact and case-sensitive with no lowercasing or Unicode normalization, as in PHP; the id segment is parsed with PHP (int)-cast semantics confirmed with `php -r` (for example `5abc` is 5 and `abc` is 0, which denies).
Edge (RT-02 ordering): registering a namespace twice is a boot error (PHP's last-wins registry is not reproduced; fail-loud like duplicate middleware names) and `Registry.Namespaces()` returns a sorted list.
Edge (RT-02 idempotency): every subscribe re-runs the authorizer against current database state with no caching, so a user removed as editor is denied on the next subscribe.
Edge (RT-02 concurrency): concurrent subscribe requests are handled independently and the registry is safe for concurrent reads (race detector clean).
The connection token info claim MUST NOT carry anything but the display name (no email, no ids beyond sub), keeping the PHP WS-004 rule
resolved
test
requirement_id
category
statement
status
verification
RT-02
privacy
A denied subscribe MUST NOT reveal the internal reason, the namespace's existence or the user's membership in its response; the body is always the generic Access denied
resolved
test
requirement_id
category
statement
status
verification
RT-03
safety
A model broadcast MUST NOT be published for a write that rolled back
resolved
test
requirement_id
category
statement
status
verification
RT-03
privacy
An album change MUST NOT be broadcast to any channel other than its own kind=collection collection channel; wishlist albums publish nothing
resolved
test
Phase Goal
ROADMAP Phase 11 goal (verbatim, not in user-story form): River jobs run on the correct dual-driver split, Centrifugo publishing and channel authorization match the existing server, and Typesense sync stays a re-gated pre-filter — all brought up before the API phases that depend on them.
This plan's slice: the Nuxt app can fetch a Centrifugo connection token from GET /api/realtime/token, Centrifugo's subscribe proxy is re-authorized on every subscribe, and album changes are published only after their write commits (RT-01, RT-02, RT-03; ROADMAP SC-3 and SC-4).
Create the transport-neutral realtime package `lighthouse` with its Centrifugo driver sub-package, and bind Płytarium to it in fonoteka.go: config, authorizers, the Mount call, the ws-api bucket and the Album broadcast binding.
Purpose: notifications (Phase 13) and the Albums API (Phase 12) publish through this; the Nuxt client and Centrifugo stay unchanged. Decisions implemented: D-06, D-07, D-08, D-09, D-11, D-12, D-13, D-14, D-16; user decision 5 (jwt.auth before throttle:ws-api); RESEARCH Patterns 5, 7, 8, 9 and Pitfalls 7-11, 15.
Output: modules/lighthouse (+ centrifugo) with README, root README row, fonoteka.go realtime wiring and smoke tests, PROJECT.md D-16 note and the RT-01 requirement note.
Repos: summercms.go (framework, planning docs) and fonoteka.go (application). Framework text and tests use neutral names (acme, blog); the application names stay in fonoteka.go. Planning docs and code in separate commits. Never add co-author tags.
RT-03 came back unclassified from the spec-less edge probe and stays unresolved. Planner reading for manual review: batch updates through Model(&T{}).Where(...) carry a zero primary key and are skipped (Pitfall 8), so bulk paths must use WithoutBroadcasting plus Emit; a soft delete is a delete for broadcasting; a restore is not broadcast (PHP has no afterRestore hook in getBroadcastHooks).
The Album created/updated payload is built inside the write transaction by the GORM callback; SaveAlbum syncs artist pivots after tx.Save, so the automatic created payload can carry stale artists. PHP's store/update suppress automatic broadcasts and publish one explicit event after the write; Phase 12's controllers must follow that pattern (WithoutBroadcasting + Emit), which this plan makes expressible.
Task 1: The Nuxt token request reaches a Centrifugo token through the neutral realtime package and the app's Mount
D-11 and D-13: every broadcastable model, authorizer and driver is written against the lighthouse interfaces and the Route/Surface mount contract; the user locked both in CONTEXT.md, so the weight is flagged without a checkpoint.
Plan 11-01 is executed: `go doc ./modules/conga Manager.Enqueue` and `go doc ./modules/lagoon OnDatabase` exit 0.
modules/lighthouse/lighthouse.go, modules/lighthouse/drivers.go, modules/lighthouse/route.go, modules/lighthouse/users.go, modules/lighthouse/README.md, modules/lighthouse/centrifugo/config.go, modules/lighthouse/centrifugo/client.go, modules/lighthouse/centrifugo/token.go, modules/lighthouse/centrifugo/handlers.go, modules/lighthouse/centrifugo/driver.go, README.md, ../fonoteka.go/config/realtime.yaml, ../fonoteka.go/README.md, ../fonoteka.go/plugins/golem15/fonoteka/plugin.go, ../fonoteka.go/plugins/golem15/fonoteka/routes.go, ../fonoteka.go/plugins/golem15/fonoteka/realtime.go, ../fonoteka.go/plugins/golem15/fonoteka/realtime_smoke_test.go, ../fonoteka.go/plugins/golem15/fonoteka/go.mod, ../fonoteka.go/plugins/golem15/fonoteka/go.sum
modules/postcard/mailer.go, modules/postcard/drivers.go, modules/wristband/server.go (lines 170-200), modules/wristband/token.go (golang-jwt usage), modules/wire/response.go, modules/bouncer/context.go, modules/pact/capabilities.go (Router), modules/surf/router.go (Group, GroupRaw, joinPath), modules/surf/limiter.go (Bucket, BucketProvider), ../fonoteka.go/plugins/golem15/fonoteka/plugin.go (Boot, Buckets), ../fonoteka.go/plugins/golem15/fonoteka/routes.go, ../fonoteka.go/plugins/golem15/fonoteka/plugin_boot_test.go (bootConfig, bootDB), ../fonoteka.go/plugins/golem15/user/models/user.go, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/routes.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/config/config.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/classes/JwtTokenGenerator.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/classes/CentrifugoClient.php, /media/nvme/dev/golem15/fonoteka/vue-fonoteka-app/app/composables/useCentrifugo.ts, .planning/phases/11-jobs-realtime-and-search-infrastructure/11-RESEARCH.md (Patterns 7-8, Pitfalls 9-11)
(1) lighthouse core (D-11): lighthouse.go with `type Service struct` (app, driver, registry, user lookup, namespace, queue, timeout, logger, bindings), `func From(app *backpack.App) (*Service, error)` doing lookup-or-publish of `*Service` on the app; the first call reads `realtime.driver` (default `null`), `realtime.broadcast_namespace` (default ""), `realtime.broadcast_queue` (default `broadcasts`) and `realtime.broadcast_timeout` (default 5s; int seconds or duration string, the postcard idiom), builds the driver through the registered factory (unknown name is an error listing the registered names and saying the driver package must be imported), and publishes. drivers.go: `type Publisher interface { Publish(ctx, channel, event string, payload json.RawMessage) error; Broadcast(ctx, channels []string, event string, payload json.RawMessage) error }`, `type Driver interface { Publisher; Name() string; Routes() []Route }`, `type DriverFactory func(app *backpack.App, svc *Service) (Driver, error)`, `func RegisterDriver(name string, f DriverFactory)` (init-time registry like database/sql; a duplicate name panics), and built-ins registered in init: `null` (discards), `log` (logs channel names and event at Info, never the payload), `memory` (`MemoryDriver` with a mutex-guarded `[]Publication{Method, Channels, Event, Payload, Timestamp}` and a copying `Publications()`; postcard MemoryDriver precedent). None of the built-ins declares routes. users.go: `type User struct { ID uint; Name *string }`, `type UserLookup func(ctx context.Context, id uint) (User, bool, error)`, `(*Service).SetUserLookup`, `(*Service).User(ctx, id)` (no lookup registered yields User{ID} with a nil Name), `type Actor struct` with `UserID *uint` (JSON key user_id) and `Name *string` (JSON key name), `SystemActor()` ({nil, "System"}) and `(*Service).Actor(ctx)` (no principal or `Principal.Backend` gives SystemActor, because PHP's frontend `auth()->user()` is null in backend requests; otherwise the user id and looked-up name).
(2) route.go (D-13): type Surface int with UserAuth, ServerToServer, Public; type Route struct { Name, Method, Path string; Surface Surface; Handler http.HandlerFunc }; type Surfaces struct { UserAuth, ServerToServer, Public, Middleware []string }; func Mount(r pact.Router, d Driver, s Surfaces) error registers each route under prefix "/" with the surface middleware followed by Middleware: UserAuth and Public through r.Group, ServerToServer through r.GroupRaw; a UserAuth route with an empty s.UserAuth is an error (never mount the token route without a guard, T-11-19); a nil driver or a driver without routes mounts nothing.
(3) centrifugo sub-package (D-12), import path .../modules/lighthouse/centrifugo: config.go type Config struct read from realtime.centrifugo.* with the PHP defaults (api_url http://127.0.0.1:8001/api, token_ttl 3600, ws_url /ws, token_path /api/realtime/token, subscribe_path /api/realtime/subscribe, empty api_key, token_secret and proxy_secret). client.go (RESEARCH Pattern 8): NewClient(cfg, *http.Client) with a 5s timeout; Publish POSTs {api_url}/publish with an ordered struct body {channel, data: {event, payload, timestamp}} where timestamp is wire.Time of now, headers Authorization: apikey <key> and Content-Type: application/json, encoder with SetEscapeHTML(false); Broadcast the same with channels to /broadcast; Presence(ctx, channel) (map[string]any, error) returns result.presence (empty map when disabled or on failure, logged); Unsubscribe(ctx, userID uint, channel) POSTs {"user":"<id>","channel":c} to /unsubscribe; success is any 2xx (Centrifugo's 200-with-error bodies count as success as in PHP; debug-log them); empty api_key returns ErrNotConfigured without a request; the key never appears in logs or errors; Enabled() and DebugInfo() ({APIURL, Enabled, APIKeySet}). token.go (RESEARCH Pattern 7): TokenIssuer with injectable clock; ForUser(u lighthouse.User) claims sub = decimal id string, exp = now + ttl, info = {"name": u.Name}; Subscription(u, channel) claims sub, channel, exp; Anonymous() sub "" and exp now+300; ForIdentifier(identifier string, info map[string]any) with info encoded as the JSON [] when empty (Pitfall 9); SubscriptionForIdentifier(identifier, channel); all HS256 via golang-jwt/v5 and ErrNotConfigured for an empty secret; Configured(). handlers.go: TokenHandler(svc, issuer) http.HandlerFunc: no principal or no user from svc.User gives 401 {"error":"Unauthorized"}; empty token_secret gives 503 {"error":"WebSocket not configured"} (checked after the user, PHP order); else 200 {"token":"..."}; bodies through wire.WriteJSON with Cache-Control: no-cache, private (Laravel default, confirmed later against PHP by plan 11-06). driver.go: Driver implementing lighthouse.Driver (name centrifugo, Publish/Broadcast via Client, Routes returning token GET token_path UserAuth and, in Task 2, subscribe); func init() { lighthouse.RegisterDriver("centrifugo", ...) }.
(4) fonoteka.go wiring: config/realtime.yaml with driver: centrifugo, empty secrets and the PHP defaults (comment the SUMMER_REALTIME__CENTRIFUGO__* env names). New plugins/golem15/fonoteka/realtime.go: blank-imports the centrifugo driver package and defines func (p *Plugin) wireRealtime(app *backpack.App) error that calls lighthouse.From(app), stores the service on the Plugin (new field realtime *lighthouse.Service) and sets a UserLookup that loads usermodels.User by id through a lazily resolved *gorm.DB (app.Lookup[*gorm.DB](), lazyInvTokenGuard precedent). plugin.go Boot calls p.wireRealtime(app); Buckets gains "ws-api": {Max: 120, Decay: time.Minute, Key: ...} keyed wsapi:u:<id> from bouncer.User else wsapi:ip: plus surf.ClientIP (D-14). routes.go calls lighthouse.Mount(r, p.realtime.Driver(), lighthouse.Surfaces{UserAuth: surf.Use("jwt.auth"), ServerToServer: surf.Use(), Middleware: surf.Use("throttle:ws-api")}) once, so jwt.auth runs before the throttle (user decision 5). Run go mod tidy in plugins/golem15/fonoteka.
(5) Smoke test ../fonoteka.go/plugins/golem15/fonoteka/realtime_smoke_test.go TestRealtimeTokenRoute through the assembled router (surf.BuildRouter of both plugins with a test config): a valid frontend JWT returns 200 and the token decodes with the configured secret to exactly the claims sub/exp/info{name}; an empty token_secret returns 503 with the exact body; a request without Authorization gets the jwt.auth 401; the response has Content-Type application/json and no trailing newline. Also a small httptest-backed check that Client.Publish sends the exact path, Authorization: apikey header and body keys.
(6) Docs: new modules/lighthouse/README.md in the standard structure (H1, summary "Transport-neutral realtime: a publisher interface with pluggable drivers, subscribe-time channel authorization, and model broadcasts enqueued in the write transaction.", both import lines, Overview, Features, Usage with an acme example, API reference for lighthouse and centrifugo, Configuration table, Dependencies, Testing), root README.md row with the same sentence, and a ## Configuration section in ../fonoteka.go/README.md mapping CENTRIFUGO_API_URL, CENTRIFUGO_API_KEY, CENTRIFUGO_SECRET, CENTRIFUGO_TOKEN_TTL, CENTRIFUGO_WS_URL, CENTRIFUGO_PROXY_SECRET, BROADCAST_NAMESPACE, BROADCAST_QUEUE and BROADCAST_TIMEOUT to their SUMMER_REALTIME__* names. Check identifiers with go doc ./modules/lighthouse <Identifier> and go doc ./modules/lighthouse/centrifugo <Identifier>.
go vet ./... && go test ./modules/lighthouse/... -count=1 && (cd ../fonoteka.go && go vet ./plugins/golem15/fonoteka/... && go test ./plugins/golem15/fonoteka -run '^(TestRealtimeTokenRoute|TestAllRouteGroupsBoot|TestFullRouteTableAuthGroupMutualExclusivity)$' -count=1 -race -v)
<fails_when>Any command exits non-zero; the verbose run lacks "--- PASS" for TestRealtimeTokenRoute, TestAllRouteGroupsBoot or TestFullRouteTableAuthGroupMutualExclusivity, prints "no tests to run", "--- SKIP" or "DATA RACE".</fails_when>
<acceptance_criteria>
- go doc ./modules/lighthouse Mount, go doc ./modules/lighthouse Surfaces, go doc ./modules/lighthouse RegisterDriver and go doc ./modules/lighthouse/centrifugo TokenIssuer.ForUser exit 0.
- grep -c 'apikey ' modules/lighthouse/centrifugo/client.go prints at least 1.
- grep -c 'WebSocket not configured' modules/lighthouse/centrifugo/handlers.go prints 1.
- grep -n 'lighthouse.Mount' ../fonoteka.go/plugins/golem15/fonoteka/routes.go shows surf.Use("jwt.auth") as UserAuth and surf.Use("throttle:ws-api") as Middleware.
- grep -c '"ws-api"' ../fonoteka.go/plugins/golem15/fonoteka/plugin.go prints 1.
- grep -c '\[lighthouse\](modules/lighthouse/README.md)' README.md prints 1.
- grep -c 'SUMMER_REALTIME__CENTRIFUGO__TOKEN_SECRET' ../fonoteka.go/README.md prints at least 1.
</acceptance_criteria>
A logged-in client gets a PHP-shaped Centrifugo token from the mounted route, the unconfigured and unauthenticated cases keep PHP's bodies, and the driver can publish byte-shaped requests to Centrifugo.
Task 2: Centrifugo's subscribe proxy re-authorizes every subscribe through the namespace registry and Płytarium's collection and wishlist authorizers
modules/lighthouse/channel.go, modules/lighthouse/registry.go, modules/lighthouse/lighthouse.go, modules/lighthouse/README.md, modules/lighthouse/centrifugo/handlers.go, modules/lighthouse/centrifugo/driver.go, ../fonoteka.go/plugins/golem15/fonoteka/realtime.go, ../fonoteka.go/plugins/golem15/fonoteka/classes/ws/collection_authorizer.go, ../fonoteka.go/plugins/golem15/fonoteka/classes/ws/wishlist_authorizer.go, ../fonoteka.go/plugins/golem15/fonoteka/realtime_smoke_test.go
/media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/http/controllers/ProxyController.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/classes/AuthorizerRegistry.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/classes/AuthorizationResult.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/fonoteka/classes/ws/CollectionChannelAuthorizer.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/fonoteka/classes/ws/WishlistChannelAuthorizer.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/fonoteka/models/Collection.php (scopeWishlistsVisibleTo), ../fonoteka.go/plugins/golem15/fonoteka/classes/active_collection.go (AccessibleByMembership), ../fonoteka.go/plugins/golem15/fonoteka/models/collection.go, ../fonoteka.go/plugins/golem15/fonoteka/models/wishlist_subscription.go, modules/lighthouse/centrifugo/handlers.go (Task 1), .planning/phases/11-jobs-realtime-and-search-infrastructure/11-RESEARCH.md (Pattern 9, Pitfall 9, Known Threat Patterns T-11-01..T-11-03)
- Missing, wrong or empty-configured proxy secret: 200 with the exact generic deny body; the log line holds the reason and the client IP but never either secret.
- user "" or "0": deny; user "5" as a JSON string or number: the authorizer receives 5.
- `collection:{id}` for an owner or editor of a kind=collection row: 200 `{"result":{"info":[]}}`; the same id of a kind=wishlist row: deny.
- `presence:collection:{id}` for a member: deny (segment 1 is "collection", PHP (int) gives 0, malformed), matching PHP as is.
- `presence:acme:room:1` with an allowing test authorizer: result has info, allow ["prs"] and the three override keys; an authorizer override of join_leave wins over the default.
- `presence:presence:x`, `a:b:c:d`, `unknown:1`: deny.
- `wishlist:{id}` for a wishlist_subscriptions row or for a household peer of the wishlist owner: allow; for a stranger: deny.
- Removing an editor between two subscribes flips allow to deny.
(1) channel.go (RESEARCH Pattern 9): `func ParseChannel(channel string) (namespace string, presence bool)` porting parseChannel exactly: a `presence:presence:` prefix yields "" (deny), one `presence:` prefix is stripped for the lookup (presence reported true), splitting on ":" with more than three segments yields "", otherwise segment 0; `func ChannelID(channel string) int64` returns segment 1 of the original channel (0 when missing) converted with PHP (int)-cast semantics — before coding, run `php -r 'foreach (["5","05"," 5","5abc","abc","-3","","1e3","0x1A","9999999999999999999"] as $s) var_dump((int)$s);'` and implement exactly what it prints, keeping those inputs as a table test; `func FormatChannels(namespace string, channels []string) []string` lowercases each channel and prefixes `lower(namespace)+":"` unless empty namespace or already prefixed (BroadcastEventJob.formatChannels); `ClientID(ctx) string` returns the proxy's client id placed in ctx.
(2) registry.go (D-11): type Result struct { Allowed bool; Info map[string]any; Capabilities []string; Overrides map[string]any; reason string } with Reason() string, constructors Allowed(info map[string]any) Result and Denied(reason string) Result; type Authorizer interface { Authorize(ctx context.Context, userID uint, channel string) Result } and AuthorizerFunc; type Registry struct (RWMutex) with Register(namespace string, a Authorizer) error (empty namespace, one containing ":", nil authorizer or a duplicate are errors), Get(namespace) (Authorizer, bool), Namespaces() []string (sorted); (*Service).Registry() returns the service's registry.
(3) ProxyHandler in centrifugo/handlers.go (D-12, T-11-01..T-11-03): body capped at 64 KiB with http.MaxBytesReader; secret check first with crypto/subtle.ConstantTimeCompare against proxy_secret (empty configured secret denies); decode {user, channel, client} accepting user as a JSON string or number (anything else counts as empty); empty or "0" user denies; namespace from ParseChannel; unknown namespace denies; call the authorizer with ctx carrying the client id, the user id (PHP (int) of the user, negative mapped to 0) and the full original channel. Allow: {"result":{"info":...}} with info [] when empty; presence adds allow (Capabilities or ["prs"]) and override (defaults presence/join_leave {"value":true}, force_push_join_leave {"value":false}, then the authorizer's Overrides on top). Deny: slog Warn Subscription denied with reason and the PHP log context (ip for secret failures; user, channel, client and internal_reason otherwise) and the exact body {"error":{"code":403,"message":"Access denied"}} with HTTP 200 (Centrifugo reads non-200 as internal error 100). Always Content-Type: application/json and Cache-Control: no-cache, private, no trailing newline. A missing channel denies (Centrifugo always sends one; PHP would 500 with a TypeError), recorded as deliberate hardening in the SUMMARY. Driver.Routes adds subscribe POST subscribe_path ServerToServer.
(4) fonoteka.go authorizers (D-16), new package classes/ws: CollectionAuthorizer{App} resolves *gorm.DB lazily per call (deny with reason "database unavailable" when unpublished); id := lighthouse.ChannelID(channel), not positive denies "malformed channel"; the user must exist in users (else "user not found"); allowed when a golem15_fonoteka_collections row with that id, kind collection, not soft-deleted, passes classes.AccessibleByMembership(userID); else "not owner/editor of collection". WishlistAuthorizer{App}: same id parsing; allowed when a golem15_fonoteka_wishlist_subscriptions row has (user_id, collection_id), or when a kind wishlist collection with that id has owner_id in the peer set (owner ids plus editor user ids of every collection the user reaches through AccessibleByMembership, mirroring scopeWishlistsVisibleTo); else "not a subscriber of this wishlist". realtime.go registers collection and wishlist on the service registry during wireRealtime.
(5) Smoke test TestRealtimeSubscribeProxy in realtime_smoke_test.go covering the behavior list against Postgres (use a fresh lagoon.Use(ctx, bootSQL) handle and seeded users/collections), run with -race; plus TestChannelIDMatchesPHP for the php -r table (the test may live in the lighthouse package as a unit test instead). Update modules/lighthouse/README.md (proxy contract, registry, channel rules).
go vet ./... && go test ./modules/lighthouse/... -count=1 -race && (cd ../fonoteka.go && go test ./plugins/golem15/fonoteka -run '^(TestRealtimeSubscribeProxy|TestRealtimeTokenRoute)$' -count=1 -race -v)
<fails_when>Any command exits non-zero; the verbose run lacks "--- PASS" for TestRealtimeSubscribeProxy or TestRealtimeTokenRoute, prints "no tests to run", "--- SKIP" or "DATA RACE".</fails_when>
<acceptance_criteria>
- grep -c 'ConstantTimeCompare' modules/lighthouse/centrifugo/handlers.go prints 1.
- grep -c '"Access denied"' modules/lighthouse/centrifugo/handlers.go prints 1.
- grep -c 'force_push_join_leave' modules/lighthouse/centrifugo/handlers.go prints at least 1.
- grep -c 'Register("collection"' ../fonoteka.go/plugins/golem15/fonoteka/realtime.go and grep -c 'Register("wishlist"' ../fonoteka.go/plugins/golem15/fonoteka/realtime.go each print 1.
- go doc ./modules/lighthouse Registry.Register, go doc ./modules/lighthouse ChannelID and go doc ./modules/lighthouse/centrifugo ProxyHandler exit 0.
- TestRealtimeSubscribeProxy asserts the allow body bytes {"result":{"info":[]}} exactly.
</acceptance_criteria>
Every Centrifugo subscribe is re-authorized against the database through the namespace registry, denials are generic HTTP 200 bodies with reasons only in logs, and Płytarium's collection and wishlist rules are ported.
Task 3: Album writes publish to Centrifugo only after commit, and a suppressed bulk write emits exactly one summary event
modules/lighthouse/broadcast.go, modules/lighthouse/suppress.go, modules/lighthouse/job.go, modules/lighthouse/lighthouse.go, modules/lighthouse/README.md, ../fonoteka.go/plugins/golem15/fonoteka/realtime.go, ../fonoteka.go/plugins/golem15/fonoteka/realtime_smoke_test.go, .planning/PROJECT.md, .planning/REQUIREMENTS.md
/media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/traits/BroadcastableModel.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/websockets/jobs/BroadcastEventJob.php, /media/nvme/dev/golem15/fonoteka/plugins/golem15/fonoteka/models/Album.php (lines 415-495), /media/nvme/dev/golem15/fonoteka/plugins/golem15/fonoteka/controllers/api/AlbumApiController.php (store, bulk, publishAlbumBroadcast), ../fonoteka.go/plugins/golem15/fonoteka/classes/serialize.go, ../fonoteka.go/plugins/golem15/fonoteka/models/album.go, ../fonoteka.go/plugins/golem15/fonoteka/classes/artist_resolver.go (callback registration precedent), modules/conga/conga.go (Enqueue, Register), modules/lagoon/ondatabase.go, $(go env GOMODCACHE)/gorm.io/gorm@v1.31.2/callbacks/transaction.go, .planning/PROJECT.md (Constraints and Key Decisions), .planning/REQUIREMENTS.md (RT-01), .planning/phases/11-jobs-realtime-and-search-infrastructure/11-RESEARCH.md (Pattern 5, Pitfalls 7, 8, 15)
- An album created inside lagoon.Transaction under a frontend principal produces one POST /publish on `collection:` with data.event `created.fonoteka.album` and payload keys id, collection_id, action, actor, timestamp, album; actor is {user_id, name}.
- The same write rolled back produces no publish within 2s.
- A delete of an album loaded with only its id still publishes `deleted.fonoteka.album` with id, collection_id, action, actor, timestamp and no album key.
- An album in a kind=wishlist collection publishes nothing.
- Three album creates inside WithoutBroadcasting[models.Album] plus one svc.Emit of `collection.bulk_updated` {reason: bulk_create, count: 3} produce exactly one publish; suppressing Album never suppresses another bound type (asserted with an acme type in plan 11-07); a write through a stale outer ctx inside fn is not suppressed.
- Centrifugo answering 500 leaves the album committed and logs a warning; the job is not retried.
- With broadcast_namespace `acme`, the published channel is `acme:collection:`.
(1) broadcast.go (D-06, D-09, Pattern 5): `type Action string` with `ActionCreated`, `ActionUpdated`, `ActionDeleted`; `type Event struct { Action Action; Actor Actor; Timestamp wire.Time; TTL int }`; model-method contract `Broadcastable` (`BroadcastChannels(ctx context.Context, tx *gorm.DB) ([]string, error)`) with optional `BroadcastPayloader` (`BroadcastPayload(ctx, tx, Event) (any, error)`), `BroadcastAliaser` (`BroadcastAlias() string`), `BroadcastFilter` (`ShouldBroadcast(Action) bool`), `BroadcastTTLer` (`BroadcastTTL() int`); plus `type Binding[T any] struct { Alias string; Channels func(ctx, tx *gorm.DB, m *T) ([]string, error); Payload func(ctx, tx *gorm.DB, m *T, ev Event) (any, error); ShouldBroadcast func(Action) bool; TTL int }` and `func Bind[T any](svc *Service, b Binding[T]) error` (for payloads that need packages the models-leaf rule keeps out of models; nil Channels or a duplicate binding is an error). Default alias: last Go package path segment, or the one before it when that segment is `models`, plus "." plus the lowercased type name; event name `strings.ToLower(action + "." + alias)`; default payload `{"model": , "actor", "timestamp", "ttl"}` with ttl 60. Callbacks installed through `lagoon.OnDatabase` from From (idempotent per *gorm.DB: Register when `Get(name)` is nil, else Replace): `lighthouse:after_create` After `gorm:after_create`, `lighthouse:after_update` After `gorm:after_update`, `lighthouse:snapshot` Before `gorm:before_delete` (reload the full row by primary key with an unscoped fresh session when needed, compute channels and the deleted payload now and keep them with InstanceSet) and `lighthouse:after_delete` After `gorm:after_delete` (enqueue the snapshot only when db.Error is nil). Each callback: skip when the statement type has no binding or contract, when its primary key is zero (Pitfall 8; iterate slice elements for batch creates), when ctx suppresses the type, or when ShouldBroadcast vetoes; compute channels (empty means no broadcast), actor via svc.Actor(db.Statement.Context), payload, then `conga.From(app).Enqueue(ctx, db, BroadcastArgs{...}, conga.EnqueueOpts{Queue: svc.queue})`, all inside a GORM savepoint (`SavePoint`/`RollbackTo`) so a failed query or enqueue can never abort the write transaction; failures are logged Warn with channels and event, never payloads.
(2) suppress.go (D-08, Pitfall 7): func WithoutBroadcasting[T any](ctx context.Context, fn func(ctx context.Context) error) error adds reflect.TypeFor[T]() (pointer types normalized to their element) to an immutable set in a new ctx value and calls fn with that ctx only; doc comment: writes must use gdb.WithContext(ctx) with the ctx handed to fn. type Broadcast struct { Channels []string; Event string; Payload any } and (*Service).Emit(ctx, db *gorm.DB, b Broadcast) error enqueue exactly one job on the same transaction (returns the error, since the caller asked explicitly).
(3) job.go (D-06, D-07): type BroadcastArgs struct { Channels []string; Event string; Payload json.RawMessage } with Kind summer.broadcast; From registers it once with conga.Job(..., conga.OnQueue(queue), conga.MaxAttempts(1), conga.Timeout(timeout)); the worker applies FormatChannels with the namespace, returns nil for zero channels, calls driver.Publish for one channel and driver.Broadcast for several, and on error logs Warn realtime: broadcast failed (channels, event) and returns nil so River never retries (PHP tries = 1).
(4) fonoteka Album binding in realtime.go: lighthouse.Bind[models.Album](svc, lighthouse.Binding[models.Album]{Alias: "fonoteka.album", Channels: albumChannels, Payload: albumPayload}); albumChannels loads the collection by CollectionID through tx with the default soft-delete scope and returns []string{"collection:" + id} only when Kind is collection; albumPayload returns an ordered struct {id, collection_id, action, actor, timestamp, album?} where album (omitted for deleted) is classes.SerializeAlbum of the album reloaded through tx with Genre, Styles and Artists preloaded.
(5) Smoke TestAlbumBroadcastSmoke in realtime_smoke_test.go for the behavior list: realtime config pointing api_url at an httptest server recording requests with api_key test-only-api-key, a fresh lagoon.Use handle, migrations, conga.StartWorker for the app, writes through lagoon.Transaction with bouncer.WithUser ctx; wait up to 5s for publishes.
(6) Docs: modules/lighthouse/README.md (broadcasting, suppression, Emit, Bind, queue and timeout config) in the code commit. Separate planning-docs commit in summercms.go: .planning/PROJECT.md Constraints line on the two repositories notes that websockets is not a separate app plugin (D-16: the framework realtime package plus fonoteka's config, authorizers and Mount call replace it) and a Key Decisions row for it; .planning/REQUIREMENTS.md RT-01 gets an appended note that the Centrifugo client is a hand-rolled net/http client per D-12 rather than the originally named library, so the verifier does not flag the wording.
go vet ./... && go test ./... && (cd ../fonoteka.go && go vet ./... ./plugins/golem15/fonoteka/... ./plugins/golem15/user/... && go test ./plugins/golem15/fonoteka -run '^(TestAlbumBroadcastSmoke|TestRealtimeSubscribeProxy|TestRealtimeTokenRoute)$' -count=1 -race -v && go test ./... ./plugins/golem15/fonoteka/... ./plugins/golem15/user/...)
<fails_when>Any command exits non-zero; the verbose run lacks "--- PASS" for TestAlbumBroadcastSmoke, TestRealtimeSubscribeProxy or TestRealtimeTokenRoute, prints "no tests to run", "--- SKIP" or "DATA RACE"; any package reports FAIL.</fails_when>
<acceptance_criteria>
- go doc ./modules/lighthouse WithoutBroadcasting, go doc ./modules/lighthouse Bind, go doc ./modules/lighthouse Service.Emit and go doc ./modules/lighthouse BroadcastArgs exit 0.
- grep -c 'SavePoint' modules/lighthouse/broadcast.go prints at least 1.
- grep -l 'MaxAttempts(1)' modules/lighthouse/*.go lists at least one file.
- grep -c 'Bind\[models.Album\]' ../fonoteka.go/plugins/golem15/fonoteka/realtime.go prints 1.
- TestAlbumBroadcastSmoke asserts exactly one publish for the suppressed bulk case and zero for the rolled-back write.
- grep -c 'D-16' .planning/PROJECT.md prints at least 1 and grep -c 'D-12' .planning/REQUIREMENTS.md prints at least 1.
</acceptance_criteria>
Album creates, updates and deletes publish Płytarium-shaped events to collection channels only after commit, bulk writes can suppress per-row events and emit one summary, publish failures never touch the write, and both repositories pass their full suites.
<threat_model>
Trust Boundaries
Boundary
Description
Browser (frontend JWT) → GET /api/realtime/token
Authenticated users obtain a Centrifugo connection token
Centrifugo server → POST /api/realtime/subscribe
Server-to-server call authenticated by a shared secret; body names the user and channel
Write transaction → River broadcasts queue → Centrifugo HTTP API
Model data leaves the database for subscribers of a channel
Operator config → driver selection and secrets
api_key, token_secret, proxy_secret
STRIDE Threat Register
Threat ID
Category
Component
Severity
Disposition
Mitigation Plan
T-11-01
Spoofing
subscribe proxy
high
mitigate
Constant-time X-Centrifugo-Secret check before any parsing; an empty configured proxy_secret denies (Task 2).
T-11-02
Elevation of Privilege
channel parsing and authorizers
high
mitigate
parseChannel ported exactly (presence:presence:, over three segments, namespace lookup); authorizers check kind (collection vs wishlist) and membership against the database on every subscribe; PHP (int) id semantics pinned by a php -r table test (Task 2).
T-11-03
Information Disclosure
deny responses
medium
mitigate
One generic 200 deny body; reasons and context only in logs (Task 2).
T-11-04
Information Disclosure
connection token info claim
medium
mitigate
ForUser puts only name in info; test asserts the exact claim set sub/exp/info (Task 1).
T-11-05
Information Disclosure
broadcast audience and rolled-back writes
high
mitigate
Enqueue happens inside the write transaction (rollback removes the job); Album channels only for kind=collection; payload built from SerializeAlbum, which carries no hidden or encrypted fields (Task 3).
T-11-06
Information Disclosure
logs (api key, tokens, secrets, payloads)
medium
mitigate
Client never logs the key or header; handlers never log secrets; broadcast failures log channels and event only; the log driver omits payloads (Tasks 1-3).
T-11-10
Denial of Service
token and subscribe flooding
medium
mitigate
ws-api bucket 120/min per user or IP on both routes (jwt.auth runs first on token); subscribe body capped at 64 KiB (Tasks 1-2).
T-11-19
Elevation of Privilege
Mount of a user route without a guard
high
mitigate
Mount refuses a UserAuth route when the UserAuth surface is empty (Task 1).
T-11-20
Denial of Service
broadcast failure aborting the write transaction
medium
mitigate
Broadcast queries and enqueue run inside a savepoint; errors are logged and rolled back to the savepoint only (Task 3).
T-11-SC
Tampering
Go module installs
high
mitigate
No new module: golang-jwt v5.3.1 and River v0.47.0 are already pinned; the official Centrifugo Go client is deliberately not added (D-12).
</threat_model>
After Task 3: `go vet ./... && go test ./...` in summercms.go and the fonoteka.go full vet/test command pass; TestRealtimeTokenRoute, TestRealtimeSubscribeProxy and TestAlbumBroadcastSmoke pass under -race.
Manual smoke (not blocking, collected at /gsd-verify-work): start Centrifugo v6 with the production secret layout, run `fonoteka serve`, log in through the Nuxt app, change an album and see the event arrive in the browser.
<success_criteria>
lighthouse and lighthouse/centrifugo exist with README and root row; drivers selected by realtime.driver.
The token route and subscribe proxy match PHP bodies and statuses; the registry re-validates every subscribe.
Album broadcasts are transactional, suppressible per type, and bulk-summarizable with one Emit.
fonoteka registers the ws-api bucket, both authorizers, the Mount call and the Album binding; PROJECT.md records D-16.
</success_criteria>
Create `.planning/phases/11-jobs-realtime-and-search-infrastructure/11-03-SUMMARY.md` when done.
