Files
multica/server/cmd/server/scope_authorizer_test.go
LinYushen 91424752ac feat(realtime): phase 0 — extract Broadcaster interface + add metrics (MUL-1138) (#1429)
* feat(realtime): phase 0 — extract Broadcaster interface + add metrics

Phase 0 of the WebSocket horizontal-scaling plan tracked in MUL-1138.
This change is intentionally behavior-preserving: it sets up the seams
needed for later phases (subscribe/unsubscribe protocol, scope-level
fanout, Redis Streams relay) without altering any wire protocol or
producer call sites.

What changed
- New realtime.Broadcaster interface covering the three fanout methods
  producers already use on *Hub (BroadcastToWorkspace, SendToUser,
  Broadcast). *Hub continues to satisfy it; a future Redis-backed
  implementation can be dropped in without touching listeners.
- registerListeners now depends on realtime.Broadcaster instead of
  *realtime.Hub, isolating the bus → realtime fanout layer behind an
  interface.
- New realtime.Metrics singleton with atomic counters: connects,
  disconnects, active connections, slow-client evictions, total
  messages sent/dropped, and per-event-type send counters. Wired into
  Hub register/unregister/broadcast paths and into every listener.
- New GET /health/realtime endpoint returning a JSON snapshot of the
  metrics so we can observe baseline fanout pressure before phase 1.

Why phase 0 first
GPT-Boy's only-Redis plan and CC-Girl's review both call out the same
prerequisite: get a Broadcaster seam and visibility in place before
introducing scope-level subscriptions or a Redis relay. Doing this as
a standalone step keeps each later PR focused and trivially revertable.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(realtime): only-Redis fanout — scopes, subscribe protocol, Redis Streams relay (MUL-1138)

Implements the final-version plan agreed in MUL-1138 on top of phase 0:

* Hub: 4 scope types (workspace/user/task/chat), per-client subscription
  set, subscribe/unsubscribe WS frames, ScopeAuthorizer hook for
  task/chat scope auth, first/last-subscriber callbacks for the relay,
  workspace+user auto-subscribe on connect.
* RedisRelay: Broadcaster impl that XADDs every event into
  ws:scope:{type}:{id}:stream and XREADGROUPs only the scopes for which
  this node has live subscribers. Per-node consumer group, heartbeat,
  stale-consumer sweeper, MAXLEN cap, lag/disconnect metrics.
* Listeners: route task:* events to ScopeTask, chat:* events to
  ScopeChat; workspace remains the default for everything else.
* events.Event: optional TaskID / ChatSessionID hints so the listener
  layer can pick the right scope without re-parsing payloads.
* Handler: publishTask / publishChat helpers; chat + task message
  publishers updated to use them.
* main.go: when REDIS_URL is set, wrap the hub with NewRedisRelay and
  pass the relay (instead of the hub) to registerListeners. A
  db-backed ScopeAuthorizer enforces that task/chat subscribes belong
  to the caller's workspace.
* Metrics: per-scope subscribe/deny counters, redis connect state, node
  id, lag/dropped counters surfaced via /health/realtime.

Behavior in single-node mode (REDIS_URL unset) is unchanged.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(realtime): address PR #1429 review must-fix items (MUL-1138)

- listeners: keep task/chat events on workspace fanout until the WS
  client supports scope-subscribe + reconnect-replay. Routing them
  through BroadcastToScope today (without any client subscriber) would
  silently drop every chat / task message and break the live timeline,
  chat unread badges, and pending-task UI. The server-side scope infra
  (Hub subscribe/unsubscribe, ScopeAuthorizer, Redis Streams relay)
  stays in place so flipping the switch in the client follow-up PR is
  a one-line change.

- scope_authorizer: ScopeChat now enforces CreatorID == userID, mirroring
  the HTTP layer (handler/chat.go: GetChatSession / SendChatMessage /
  MarkChatSessionRead). Without this, any workspace member who learned a
  session_id could subscribe to chat:message / chat:done /
  chat:session_read for a peer's private chat. The same creator-only
  check is applied to ScopeTask when the task is a chat task
  (task.ChatSessionID set). Issue tasks remain workspace-scoped.

- Refactor scope authorizer to depend on a narrow scopeAuthQuerier
  interface so its decisions can be unit-tested without a live DB.

- Add tests:
  * listeners_scope_test.go pins the workspace-fanout fallback for
    task:message / task:progress / chat:message / chat:done /
    chat:session_read.
  * scope_authorizer_test.go covers chat creator-only access, chat-task
    creator-only access, and issue-task workspace-only access (creator
    allowed, peer denied, cross-workspace denied, missing session
    denied, empty userID denied).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: CC-Girl <cc-girl@multica.ai>
2026-04-23 13:36:55 +08:00

187 lines
5.5 KiB
Go

package main
import (
"context"
"errors"
"testing"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/realtime"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// fakeScopeQuerier implements scopeAuthQuerier with in-memory maps.
type fakeScopeQuerier struct {
tasks map[[16]byte]db.AgentTaskQueue
issues map[[16]byte]db.Issue
sessions map[[16]byte]db.ChatSession
}
func (f *fakeScopeQuerier) GetAgentTask(_ context.Context, id pgtype.UUID) (db.AgentTaskQueue, error) {
if t, ok := f.tasks[id.Bytes]; ok {
return t, nil
}
return db.AgentTaskQueue{}, errors.New("not found")
}
func (f *fakeScopeQuerier) GetIssue(_ context.Context, id pgtype.UUID) (db.Issue, error) {
if i, ok := f.issues[id.Bytes]; ok {
return i, nil
}
return db.Issue{}, errors.New("not found")
}
func (f *fakeScopeQuerier) GetChatSession(_ context.Context, id pgtype.UUID) (db.ChatSession, error) {
if s, ok := f.sessions[id.Bytes]; ok {
return s, nil
}
return db.ChatSession{}, errors.New("not found")
}
func mustUUID(t *testing.T) (string, pgtype.UUID) {
t.Helper()
u, err := uuid.NewRandom()
if err != nil {
t.Fatal(err)
}
return u.String(), pgtype.UUID{Bytes: u, Valid: true}
}
// TestScopeAuthorizer_ChatRequiresCreator pins must-fix #2 from PR #1429:
// ScopeChat MUST verify CreatorID == userID. A workspace peer that knows the
// session_id must NOT be able to subscribe to chat:message / chat:done /
// chat:session_read for that private session.
func TestScopeAuthorizer_ChatRequiresCreator(t *testing.T) {
wsStr, wsUUID := mustUUID(t)
creatorStr, creatorUUID := mustUUID(t)
otherStr, _ := mustUUID(t)
sessStr, sessUUID := mustUUID(t)
otherWsStr, _ := mustUUID(t)
otherWsStrOnly, otherWsUUID := mustUUID(t)
_ = otherWsStrOnly
q := &fakeScopeQuerier{
sessions: map[[16]byte]db.ChatSession{
sessUUID.Bytes: {
ID: sessUUID,
WorkspaceID: wsUUID,
CreatorID: creatorUUID,
},
},
}
a := newScopeAuthorizer(q)
ctx := context.Background()
// Creator in matching workspace → allowed.
ok, err := a.AuthorizeScope(ctx, creatorStr, wsStr, realtime.ScopeChat, sessStr)
if err != nil || !ok {
t.Fatalf("creator should be allowed: ok=%v err=%v", ok, err)
}
// Same workspace, different (peer) member → must be denied.
ok, err = a.AuthorizeScope(ctx, otherStr, wsStr, realtime.ScopeChat, sessStr)
if err != nil || ok {
t.Fatalf("peer must be denied: ok=%v err=%v", ok, err)
}
// Cross-workspace creator (e.g. session in workspace A, request in
// workspace B) → must be denied even though creator matches.
ok, err = a.AuthorizeScope(ctx, creatorStr, otherWsStr, realtime.ScopeChat, sessStr)
if err != nil || ok {
t.Fatalf("cross-workspace must be denied: ok=%v err=%v", ok, err)
}
_ = otherWsUUID
// Empty userID → must be denied (defensive).
ok, err = a.AuthorizeScope(ctx, "", wsStr, realtime.ScopeChat, sessStr)
if err != nil || ok {
t.Fatalf("empty userID must be denied: ok=%v err=%v", ok, err)
}
// Unknown session → denied.
_, missingStr := mustUUID(t)
_ = missingStr
missingUUID, _ := uuid.NewRandom()
ok, err = a.AuthorizeScope(ctx, creatorStr, wsStr, realtime.ScopeChat, missingUUID.String())
if err != nil || ok {
t.Fatalf("unknown session must be denied: ok=%v err=%v", ok, err)
}
}
// TestScopeAuthorizer_ChatTaskRequiresCreator pins must-fix #2 for the
// task-scope path of chat tasks (task.ChatSessionID set, no IssueID): only
// the chat session creator may subscribe to that task's stream, since
// task:message for chat tasks contains assistant chat content.
func TestScopeAuthorizer_ChatTaskRequiresCreator(t *testing.T) {
wsStr, wsUUID := mustUUID(t)
creatorStr, creatorUUID := mustUUID(t)
otherStr, _ := mustUUID(t)
sessStr, sessUUID := mustUUID(t)
taskStr, taskUUID := mustUUID(t)
_ = sessStr
q := &fakeScopeQuerier{
tasks: map[[16]byte]db.AgentTaskQueue{
taskUUID.Bytes: {
ID: taskUUID,
ChatSessionID: sessUUID,
},
},
sessions: map[[16]byte]db.ChatSession{
sessUUID.Bytes: {
ID: sessUUID,
WorkspaceID: wsUUID,
CreatorID: creatorUUID,
},
},
}
a := newScopeAuthorizer(q)
ctx := context.Background()
ok, err := a.AuthorizeScope(ctx, creatorStr, wsStr, realtime.ScopeTask, taskStr)
if err != nil || !ok {
t.Fatalf("creator should be allowed for chat task: ok=%v err=%v", ok, err)
}
ok, err = a.AuthorizeScope(ctx, otherStr, wsStr, realtime.ScopeTask, taskStr)
if err != nil || ok {
t.Fatalf("peer must be denied for chat task: ok=%v err=%v", ok, err)
}
}
// TestScopeAuthorizer_IssueTaskWorkspaceOnly verifies issue tasks remain
// workspace-scoped (any member who can see the issue may subscribe).
func TestScopeAuthorizer_IssueTaskWorkspaceOnly(t *testing.T) {
wsStr, wsUUID := mustUUID(t)
memberStr, _ := mustUUID(t)
otherWsStr, _ := mustUUID(t)
taskStr, taskUUID := mustUUID(t)
_, issueUUID := mustUUID(t)
q := &fakeScopeQuerier{
tasks: map[[16]byte]db.AgentTaskQueue{
taskUUID.Bytes: {
ID: taskUUID,
IssueID: issueUUID,
},
},
issues: map[[16]byte]db.Issue{
issueUUID.Bytes: {
ID: issueUUID,
WorkspaceID: wsUUID,
},
},
}
a := newScopeAuthorizer(q)
ctx := context.Background()
ok, err := a.AuthorizeScope(ctx, memberStr, wsStr, realtime.ScopeTask, taskStr)
if err != nil || !ok {
t.Fatalf("member in workspace should be allowed: ok=%v err=%v", ok, err)
}
ok, err = a.AuthorizeScope(ctx, memberStr, otherWsStr, realtime.ScopeTask, taskStr)
if err != nil || ok {
t.Fatalf("cross-workspace must be denied: ok=%v err=%v", ok, err)
}
}