feat(FN-1745): scope WebSocket channels to project context for multi-project support

- Scope /api/tasks/:id/logs/stream SSE endpoint to resolved project context
- Make badge WebSocket/pubsub channels project-aware using scopeKey
- Fix terminal WebSocket project/service resolution with auth middleware
- Add cross-instance badge broadcast via BadgePubSub with projectId metadata
- Update AGENTS.md documentation for badge WebSocket project-scoping
- Add tests for project-scoped WebSocket and SSE endpoints
- Add tests for project detection in WebSocket upgrade handling
This commit is contained in:
gsxdsm
2026-04-14 08:13:25 -07:00
parent 38d859e843
commit 271f3f7c16
8 changed files with 395 additions and 115 deletions

View File

@@ -5,8 +5,8 @@ import type { IssueInfo, PrInfo } from "@fusion/core";
* Badge snapshot message envelope for shared pub/sub.
*
* This contract is used for cross-instance badge updates. Each message includes
* a sourceId (server instance identifier), taskId, timestamp, and optional
* prInfo/issueInfo snapshot data.
* a sourceId (server instance identifier), projectId (for cross-project isolation),
* taskId, timestamp, and optional prInfo/issueInfo snapshot data.
*
* Explicit null values indicate a badge was removed; omitted fields mean no change
* to that badge type's data.
@@ -14,6 +14,8 @@ import type { IssueInfo, PrInfo } from "@fusion/core";
export interface BadgePubSubMessage {
/** Unique identifier for the originating server instance (for deduplication) */
sourceId: string;
/** Project scope key for cross-project isolation (e.g., project ID or "default") */
projectId?: string;
/** Task identifier */
taskId: string;
/** ISO timestamp when the snapshot was captured */

View File

@@ -8450,7 +8450,7 @@ describe("Terminal WebSocket close handler", () => {
const getSessionMock = vi.fn().mockReturnValue({
id: "term-ws-test",
shell: "/bin/zsh",
cwd: "/test/project",
cwd: "/fake/root",
scrollbackBuffer: "hello",
lastActivityAt: new Date(),
});
@@ -8474,8 +8474,9 @@ describe("Terminal WebSocket close handler", () => {
const app = express();
const server = http.createServer(app);
const store = createMockStore();
setupTerminalWebSocket(app, server);
setupTerminalWebSocket(app, server, store);
class FakeWebSocket extends EventEmitter {
send = vi.fn();
close = vi.fn(() => this.emit("close"));
@@ -8504,7 +8505,7 @@ describe("Terminal WebSocket close handler", () => {
const getSessionMock = vi.fn().mockReturnValue({
id: "term-ws-err",
shell: "/bin/zsh",
cwd: "/test/project",
cwd: "/fake/root",
lastActivityAt: new Date(),
});
const getScrollbackAndClearPendingMock = vi.fn().mockReturnValue(null);
@@ -8527,8 +8528,9 @@ describe("Terminal WebSocket close handler", () => {
const app = express();
const server = http.createServer(app);
const store = createMockStore();
setupTerminalWebSocket(app, server);
setupTerminalWebSocket(app, server, store);
class FakeWebSocket extends EventEmitter {
send = vi.fn();
close = vi.fn(() => this.emit("close"));
@@ -8561,7 +8563,7 @@ describe("Terminal WebSocket close handler", () => {
const getSessionMock = vi.fn().mockReturnValue({
id: "term-ws-unsub",
shell: "/bin/zsh",
cwd: "/test/project",
cwd: "/fake/root",
lastActivityAt: new Date(),
});
const getScrollbackAndClearPendingMock = vi.fn().mockReturnValue(null);
@@ -8584,8 +8586,9 @@ describe("Terminal WebSocket close handler", () => {
const app = express();
const server = http.createServer(app);
const store = createMockStore();
setupTerminalWebSocket(app, server);
setupTerminalWebSocket(app, server, store);
class FakeWebSocket extends EventEmitter {
send = vi.fn();
close = vi.fn(() => this.emit("close"));

View File

@@ -304,10 +304,12 @@ describe("API Error Handling Middleware", () => {
describe("Terminal WebSocket heartbeat", () => {
let app: ReturnType<typeof express>;
let server: http.Server;
let store: TaskStore;
beforeEach(() => {
app = express();
server = http.createServer(app);
store = createMockStore();
vi.useFakeTimers();
vi.spyOn(console, "warn").mockImplementation(() => {});
vi.spyOn(console, "info").mockImplementation(() => {});
@@ -349,7 +351,7 @@ describe("Terminal WebSocket heartbeat", () => {
/** Setup terminal WebSocket and trigger a connection */
function setupAndConnect(ws: any, req: any): void {
const wss = setupTerminalWebSocket(app, server);
const wss = setupTerminalWebSocket(app, server, store);
// The function sets up wss on the server's upgrade event.
// We need to access the WebSocketServer directly to emit a connection.
@@ -368,7 +370,7 @@ describe("Terminal WebSocket heartbeat", () => {
mockTerminalService.getSession.mockReturnValue({
id: "session-1",
shell: "/bin/bash",
cwd: "/project",
cwd: "/fake/root",
lastActivityAt: new Date(),
});
@@ -394,7 +396,7 @@ describe("Terminal WebSocket heartbeat", () => {
mockTerminalService.getSession.mockReturnValue({
id: "session-2",
shell: "/bin/bash",
cwd: "/project",
cwd: "/fake/root",
lastActivityAt: new Date(),
});
@@ -420,7 +422,7 @@ describe("Terminal WebSocket heartbeat", () => {
mockTerminalService.getSession.mockReturnValue({
id: "session-3",
shell: "/bin/bash",
cwd: "/project",
cwd: "/fake/root",
lastActivityAt: new Date(),
});
@@ -457,7 +459,7 @@ describe("Terminal WebSocket heartbeat", () => {
mockTerminalService.getSession.mockReturnValue({
id: "stale-session",
shell: "/bin/bash",
cwd: "/project",
cwd: "/fake/root",
lastActivityAt: tenMinutesAgo,
});
@@ -479,7 +481,7 @@ describe("Terminal WebSocket heartbeat", () => {
mockTerminalService.getSession.mockReturnValue({
id: "fresh-session",
shell: "/bin/bash",
cwd: "/project",
cwd: "/fake/root",
lastActivityAt: new Date(Date.now() - 60_000),
});

View File

@@ -66,6 +66,35 @@ process.on("beforeExit", () => {
clearAiSessionCleanupInterval();
});
/**
* Module-level helper for resolving a scoped TaskStore.
* Mirrors /api/events semantics: prefer engine's store, fallback to resolver, fallback to default store.
* Used by realtime endpoints that need project-aware store resolution.
*
* @param projectId - The project ID to resolve, or undefined for the default store
* @param store - The default TaskStore to use when projectId is undefined
* @param engineManager - Optional engine manager for per-project engine store access
* @returns The resolved TaskStore
*/
export async function resolveScopedStore(
projectId: string | undefined,
store: TaskStore,
engineManager?: import("@fusion/engine").ProjectEngineManager,
): Promise<TaskStore> {
if (!projectId) {
return store;
}
if (engineManager) {
const engine = engineManager.getEngine(projectId);
if (engine) {
return engine.getTaskStore();
}
}
return await getOrCreateProjectStore(projectId);
}
export interface ServerOptions {
/** Optional ProjectEngine — when provided, subsystems (onMerge, automationStore,
* missionAutopilot, missionExecutionLoop, heartbeatMonitor) are derived from it.
@@ -320,9 +349,18 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
}
});
/**
* Shared project-resolution helper for realtime endpoints.
* Uses module-level resolveScopedStore with current closure context.
*/
async function resolveProjectScopedStore(projectId: string | undefined): Promise<TaskStore> {
return resolveScopedStore(projectId, store, options?.engineManager);
}
// Per-task SSE endpoint for live agent log streaming
app.get("/api/tasks/:id/logs/stream", (req, res) => {
app.get("/api/tasks/:id/logs/stream", async (req, res) => {
const taskId = req.params.id;
const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
@@ -332,23 +370,28 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
res.write(": connected\n\n");
// agent:log events are emitted by the in-process TaskExecutor via
// store.appendAgentLog(). The executor is always bound to the default
// `store` passed to createServer — never to a project-scoped store
// created by getOrCreateProjectStore — so we must listen on `store`
// directly. Using getOrCreateProjectStore here would attach the listener
// to a different EventEmitter instance that the executor never writes to,
// breaking real-time log streaming.
// Resolve the store for this request:
// - With projectId: use scoped store from engine or resolver (ensures multi-project isolation)
// - Without projectId: use default store (preserves existing single-project behavior)
//
// Per-entry text and detail fields are serialized in full — there is no
// SSE-level truncation. The 500-entry cap is applied client-side in the
// React hooks (useAgentLogs / useMultiAgentLogs).
let scopedStore: TaskStore;
try {
scopedStore = await resolveProjectScopedStore(projectId);
} catch (err) {
res.write(`event: error\ndata: ${JSON.stringify({ message: "Failed to resolve project store" })}\n\n`);
res.end();
return;
}
const onAgentLog = (entry: { taskId: string; text: string; type: string; timestamp: string }) => {
if (entry.taskId !== taskId) return;
res.write(`event: agent:log\ndata: ${JSON.stringify(entry)}\n\n`);
};
store.on("agent:log", onAgentLog);
scopedStore.on("agent:log", onAgentLog);
const heartbeat = setInterval(() => {
res.write(": heartbeat\n\n");
@@ -356,7 +399,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
req.on("close", () => {
clearInterval(heartbeat);
store.off("agent:log", onAgentLog);
scopedStore.off("agent:log", onAgentLog);
});
});
@@ -633,7 +676,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
if (!dashboardApp.__fnWebSocketsAttached) {
dashboardApp.__fnWebSocketsAttached = true;
setupTerminalWebSocket(dashboardApp, server);
setupTerminalWebSocket(dashboardApp, server, store, options);
setupBadgeWebSocket(dashboardApp, server, store, options);
}
@@ -650,11 +693,14 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
export function setupTerminalWebSocket(
app: ReturnType<typeof express>,
server: import("http").Server,
store: TaskStore,
options?: ServerOptions,
): void {
const terminalService = getTerminalService();
const wss = new WebSocketServer({ noServer: true });
// Default terminal service for stale eviction (uses default store's root dir)
const defaultTerminalService = getTerminalService(store.getRootDir());
server.on("upgrade", (req, socket, head) => {
const pathname = new URL(req.url || "", `http://${req.headers.host}`).pathname;
if (pathname !== "/api/terminal/ws") {
@@ -669,22 +715,52 @@ export function setupTerminalWebSocket(
// Store reference on app for access
(app as DashboardExpressApp).terminalWsServer = wss;
wss.on("connection", (ws: WebSocket, req) => {
wss.on("connection", async (ws: WebSocket, req) => {
// Parse query params from URL
const url = new URL(req.url || "", `http://${req.headers.host}`);
const sessionId = url.searchParams.get("sessionId");
const projectId = url.searchParams.get("projectId") ?? undefined;
if (!sessionId) {
ws.close(4000, "Missing sessionId");
return;
}
// Resolve the scoped terminal service
let terminalService: ReturnType<typeof getTerminalService>;
let scopedRootDir: string;
try {
if (projectId) {
// When projectId is provided, resolve the scoped store and get its root dir
const scopedStore = await resolveScopedStore(projectId, store, options?.engineManager);
scopedRootDir = scopedStore.getRootDir();
terminalService = getTerminalService(scopedRootDir);
} else {
// Without projectId, use the default store's root dir
scopedRootDir = store.getRootDir();
terminalService = getTerminalService(scopedRootDir);
}
} catch (err) {
console.error("[terminal] Failed to resolve project scope:", err);
ws.close(4510, "Failed to resolve project scope");
return;
}
const session = terminalService.getSession(sessionId);
if (!session) {
ws.close(4004, "Session not found");
return;
}
// Security check: reject sessions that don't belong to this project's root
// Session cwd must be within the resolved project root
if (!session.cwd.startsWith(scopedRootDir)) {
console.warn(`[terminal] Session ${sessionId} cwd ${session.cwd} does not belong to project root ${scopedRootDir}`);
ws.close(4503, "Session does not belong to this project");
return;
}
const MAX_MISSED_PONGS = 2; // Allow 2 missed pongs (~90s) before terminating
// Track if connection is alive
@@ -818,7 +894,7 @@ export function setupTerminalWebSocket(
// TerminalService (default 5 minutes of inactivity).
const staleEvictionInterval = setInterval(() => {
try {
terminalService.evictStaleSessions();
defaultTerminalService.evictStaleSessions();
} catch {
// Ignore errors during periodic eviction
}
@@ -842,7 +918,8 @@ export function setupBadgeWebSocket(
const wsManager = new WebSocketManager();
// Structured badge snapshot cache for local subscriptions and pub/sub sync
// Maps taskId -> BadgeSnapshot with timestamp
// Maps "{projectId}:{taskId}" -> BadgeSnapshot with timestamp
// Uses "default" for unscoped/default project
const badgeSnapshots = new Map<string, BadgeSnapshot>();
// Server instance ID for pub/sub deduplication
@@ -852,10 +929,31 @@ export function setupBadgeWebSocket(
const badgePubSub = options?.badgePubSub ?? createBadgePubSub({ sourceId: serverId });
void badgePubSub.start();
// Prime cache with existing tasks
// Track scoped stores for multi-project support
const scopedStores = new Map<string, TaskStore>();
// Helper to get or create a scoped store
const getScopedStore = async (projectId: string): Promise<TaskStore> => {
// Always use the default store for the "default" scope
if (projectId === "default") {
return store;
}
let scopedStore = scopedStores.get(projectId);
if (scopedStore) {
return scopedStore;
}
// Create scoped store
scopedStore = await resolveScopedStore(projectId, store, options?.engineManager);
scopedStores.set(projectId, scopedStore);
return scopedStore;
};
// Prime cache with existing tasks from default store
void store.listTasks({ slim: true, includeArchived: false }).then((tasks) => {
for (const task of tasks) {
badgeSnapshots.set(task.id, {
badgeSnapshots.set(`default:${task.id}`, {
prInfo: task.prInfo ?? null,
issueInfo: task.issueInfo ?? null,
timestamp: new Date().toISOString(),
@@ -881,95 +979,165 @@ export function setupBadgeWebSocket(
dashboardApp.badgeWsServer = wss;
dashboardApp.badgeWsManager = wsManager;
const broadcastBadgeSnapshot = (taskId: string, snapshot: BadgeSnapshot): void => {
wsManager.broadcastBadgeUpdate(taskId, snapshot);
/**
* Broadcast a badge snapshot to subscribed clients within a project scope.
*/
const broadcastBadgeSnapshot = (taskId: string, snapshot: BadgeSnapshot, projectId: string = "default"): void => {
wsManager.broadcastBadgeUpdate(taskId, snapshot, projectId);
};
const onTaskUpdated = (task: Task) => {
const previousSnapshot = badgeSnapshots.get(task.id);
const nextSnapshot: BadgeSnapshot = {
prInfo: task.prInfo ?? null,
issueInfo: task.issueInfo ?? null,
timestamp: new Date().toISOString(),
};
// Update local cache immediately
badgeSnapshots.set(task.id, nextSnapshot);
/**
* Get or create scoped store and attach badge listeners.
* Returns cleanup function.
*/
const attachScopedListeners = async (
projectId: string,
scopedStore: TaskStore
): Promise<() => void> => {
const scopeKey = projectId === "default" ? "default" : projectId;
// Check if badge data actually changed
if (snapshotsEqual(previousSnapshot, nextSnapshot)) {
const onTaskUpdated = (task: Task) => {
const cacheKey = `${scopeKey}:${task.id}`;
const previousSnapshot = badgeSnapshots.get(cacheKey);
const nextSnapshot: BadgeSnapshot = {
prInfo: task.prInfo ?? null,
issueInfo: task.issueInfo ?? null,
timestamp: new Date().toISOString(),
};
// Update local cache immediately
badgeSnapshots.set(cacheKey, nextSnapshot);
// Check if badge data actually changed
if (snapshotsEqual(previousSnapshot, nextSnapshot)) {
return;
}
// Always publish to shared bus (even if no local subscribers)
// This ensures other instances receive the update
const pubSubMessage: BadgePubSubMessage = {
sourceId: serverId,
projectId,
taskId: task.id,
timestamp: nextSnapshot.timestamp,
prInfo: nextSnapshot.prInfo,
issueInfo: nextSnapshot.issueInfo,
};
void badgePubSub.publish(pubSubMessage);
// Broadcast to local websocket subscribers if any
if (wsManager.getSubscriptionCount(task.id, projectId) > 0) {
broadcastBadgeSnapshot(task.id, nextSnapshot, projectId);
}
};
const onTaskCreated = (task: Task) => {
const cacheKey = `${scopeKey}:${task.id}`;
badgeSnapshots.set(cacheKey, {
prInfo: task.prInfo ?? null,
issueInfo: task.issueInfo ?? null,
timestamp: new Date().toISOString(),
});
};
const onTaskDeleted = (task: Task) => {
const cacheKey = `${scopeKey}:${task.id}`;
badgeSnapshots.delete(cacheKey);
};
scopedStore.on("task:updated", onTaskUpdated);
scopedStore.on("task:created", onTaskCreated);
scopedStore.on("task:deleted", onTaskDeleted);
return () => {
scopedStore.off("task:updated", onTaskUpdated);
scopedStore.off("task:created", onTaskCreated);
scopedStore.off("task:deleted", onTaskDeleted);
};
};
// Store cleanup functions for scoped listeners
const scopedCleanups = new Map<string, () => void>();
// Attach listeners to default store
void (async () => {
const cleanup = await attachScopedListeners("default", store);
scopedCleanups.set("default", cleanup);
})();
/**
* Ensure scoped listeners are attached for a project.
*/
const ensureScopedListeners = async (projectId: string): Promise<void> => {
if (scopedCleanups.has(projectId)) {
return;
}
// Always publish to shared bus (even if no local subscribers)
// This ensures other instances receive the update
const pubSubMessage: BadgePubSubMessage = {
sourceId: serverId,
taskId: task.id,
timestamp: nextSnapshot.timestamp,
prInfo: nextSnapshot.prInfo,
issueInfo: nextSnapshot.issueInfo,
};
void badgePubSub.publish(pubSubMessage);
// Broadcast to local websocket subscribers if any
if (wsManager.getSubscriptionCount(task.id) > 0) {
broadcastBadgeSnapshot(task.id, nextSnapshot);
}
const scopedStore = await getScopedStore(projectId);
const cleanup = await attachScopedListeners(projectId, scopedStore);
scopedCleanups.set(projectId, cleanup);
};
const onTaskCreated = (task: Task) => {
badgeSnapshots.set(task.id, {
prInfo: task.prInfo ?? null,
issueInfo: task.issueInfo ?? null,
timestamp: new Date().toISOString(),
});
};
const onTaskDeleted = (task: Task) => {
badgeSnapshots.delete(task.id);
};
store.on("task:updated", onTaskUpdated);
store.on("task:created", onTaskCreated);
store.on("task:deleted", onTaskDeleted);
// Handle remote badge updates from other instances via pub/sub
badgePubSub.on("message", (message: BadgePubSubMessage) => {
// Use provided projectId or default scope
const projectId = message.projectId ?? "default";
const cacheKey = `${projectId}:${message.taskId}`;
// Update local cache with remote snapshot
const remoteSnapshot: BadgeSnapshot = {
prInfo: message.prInfo,
issueInfo: message.issueInfo,
timestamp: message.timestamp,
};
badgeSnapshots.set(message.taskId, remoteSnapshot);
badgeSnapshots.set(cacheKey, remoteSnapshot);
// Rebroadcast to local websocket subscribers
// (No need to check for echo - pub/sub adapter already filtered our own messages)
if (wsManager.getSubscriptionCount(message.taskId) > 0) {
broadcastBadgeSnapshot(message.taskId, remoteSnapshot);
if (wsManager.getSubscriptionCount(message.taskId, projectId) > 0) {
broadcastBadgeSnapshot(message.taskId, remoteSnapshot, projectId);
}
});
wsManager.on("subscription:changed", (taskId, subscriberCount) => {
wsManager.on("subscription:changed", (taskId, subscriberCount, projectId) => {
// Send cached snapshot to late subscriber if available
// This ensures a client subscribing after a remote update still sees the latest state
if (subscriberCount > 0) {
const cachedSnapshot = badgeSnapshots.get(taskId);
const cacheKey = `${projectId}:${taskId}`;
const cachedSnapshot = badgeSnapshots.get(cacheKey);
if (cachedSnapshot) {
broadcastBadgeSnapshot(taskId, cachedSnapshot);
broadcastBadgeSnapshot(taskId, cachedSnapshot, projectId);
}
}
});
wss.on("connection", (ws: WebSocket) => {
wsManager.addClient(ws, randomUUID());
wss.on("connection", (ws: WebSocket, req) => {
// Parse projectId from URL query params
const url = new URL(req.url || "", `http://${req.headers.host}`);
const projectId = url.searchParams.get("projectId") ?? "default";
// Ensure scoped listeners are attached for this project
void ensureScopedListeners(projectId);
// Add client bound to this project scope
wsManager.addClient(ws, randomUUID(), projectId);
});
server.once("close", () => {
store.off("task:updated", onTaskUpdated);
store.off("task:created", onTaskCreated);
store.off("task:deleted", onTaskDeleted);
// Clean up all scoped listeners
for (const cleanup of scopedCleanups.values()) {
cleanup();
}
scopedCleanups.clear();
for (const scopedStore of scopedStores.values()) {
// Don't close the default store - it's managed externally
if (scopedStore !== store) {
scopedStore.stopWatching?.();
scopedStore.close?.();
}
}
scopedStores.clear();
for (const client of wss.clients) {
client.terminate();

View File

@@ -33,18 +33,23 @@ export type BadgeServerMessage = BadgeUpdatedMessage | WebSocketErrorMessage;
export interface SubscribeMessage {
type: "subscribe";
taskId: string;
projectId?: string;
}
export interface UnsubscribeMessage {
type: "unsubscribe";
taskId: string;
projectId?: string;
}
export type BadgeClientMessage = SubscribeMessage | UnsubscribeMessage;
interface ClientState {
ws: WebSocket;
/** Subscribed channels (e.g., "badge:project-123:FN-001") */
subscriptions: Set<string>;
/** The project scope this client is bound to */
projectId: string;
isAlive: boolean;
handlers: {
pong: () => void;
@@ -57,7 +62,7 @@ interface ClientState {
export interface WebSocketManagerEvents {
"client:connected": [clientId: string, totalClients: number];
"client:disconnected": [clientId: string, totalClients: number];
"subscription:changed": [taskId: string, subscriberCount: number];
"subscription:changed": [taskId: string, subscriberCount: number, projectId: string];
}
export interface WebSocketManagerOptions {
@@ -75,13 +80,20 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
this.heartbeatIntervalMs = options.heartbeatIntervalMs ?? 30_000;
}
addClient(ws: WebSocket, clientId: string): void {
/**
* Add a new client to the manager.
* @param ws - The WebSocket connection
* @param clientId - Unique identifier for this client
* @param projectId - The project scope this client is bound to (defaults to "default")
*/
addClient(ws: WebSocket, clientId: string, projectId: string = "default"): void {
this.removeClient(clientId);
const handlers = this.createClientHandlers(clientId);
const state: ClientState = {
ws,
subscriptions: new Set<string>(),
projectId,
isAlive: true,
handlers,
};
@@ -118,11 +130,19 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
this.emit("client:disconnected", clientId, this.clients.size);
}
subscribe(clientId: string, taskId: string): void {
/**
* Subscribe a client to badge updates for a task within their project scope.
* @param clientId - The client ID to subscribe
* @param taskId - The task ID to subscribe to
* @param projectIdOverride - Optional project scope override (uses client's bound scope if not provided)
*/
subscribe(clientId: string, taskId: string, projectIdOverride?: string): void {
const state = this.clients.get(clientId);
if (!state) return;
const channel = toBadgeChannel(taskId);
// Use client's bound scope by default, or override if explicitly provided
const scopeKey = projectIdOverride ?? state.projectId;
const channel = toBadgeChannel(scopeKey, taskId);
if (state.subscriptions.has(channel)) return;
state.subscriptions.add(channel);
@@ -134,16 +154,33 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
}
subscribers.add(clientId);
this.emit("subscription:changed", taskId, subscribers.size);
this.emit("subscription:changed", taskId, subscribers.size, scopeKey);
}
unsubscribe(clientId: string, taskId: string): void {
const channel = toBadgeChannel(taskId);
/**
* Unsubscribe a client from badge updates for a task.
* @param clientId - The client ID to unsubscribe
* @param taskId - The task ID to unsubscribe from
* @param projectIdOverride - Optional project scope override (uses client's bound scope if not provided)
*/
unsubscribe(clientId: string, taskId: string, projectIdOverride?: string): void {
const state = this.clients.get(clientId);
if (!state) return;
const scopeKey = projectIdOverride ?? state.projectId;
const channel = toBadgeChannel(scopeKey, taskId);
this.removeChannelSubscription(clientId, channel);
}
broadcastBadgeUpdate(taskId: string, badgeData: BadgeUpdate): void {
const subscribers = this.channelSubscribers.get(toBadgeChannel(taskId));
/**
* Broadcast a badge update to all clients subscribed to the task within the scope.
* @param taskId - The task ID
* @param badgeData - The badge data to broadcast
* @param projectId - Optional project scope (defaults to "default")
*/
broadcastBadgeUpdate(taskId: string, badgeData: BadgeUpdate, projectId?: string): void {
const scopeKey = projectId ?? "default";
const subscribers = this.channelSubscribers.get(toBadgeChannel(scopeKey, taskId));
if (!subscribers || subscribers.size === 0) return;
const message: BadgeUpdatedMessage = {
@@ -171,14 +208,26 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
return this.clients.size > 0;
}
getSubscriptionCount(taskId: string): number {
return this.channelSubscribers.get(toBadgeChannel(taskId))?.size ?? 0;
/**
* Get subscription count for a task within a project scope.
* @param taskId - The task ID
* @param projectId - Optional project scope (defaults to "default")
*/
getSubscriptionCount(taskId: string, projectId?: string): number {
const scopeKey = projectId ?? "default";
return this.channelSubscribers.get(toBadgeChannel(scopeKey, taskId))?.size ?? 0;
}
getSubscribedTaskIds(): string[] {
/**
* Get all subscribed task IDs, optionally filtered by project scope.
* @param projectId - Optional project scope filter (returns all if not specified)
*/
getSubscribedTaskIds(projectId?: string): string[] {
const scopeKey = projectId ?? "default";
const prefix = `badge:${scopeKey}:`;
return [...this.channelSubscribers.entries()]
.filter(([, subscribers]) => subscribers.size > 0)
.map(([channel]) => fromBadgeChannel(channel));
.filter(([channel, subscribers]) => subscribers.size > 0 && channel.startsWith(prefix))
.map(([channel]) => fromBadgeChannel(scopeKey, channel));
}
dispose(): void {
@@ -223,11 +272,11 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
}
if (parsed.value.type === "subscribe") {
this.subscribe(clientId, parsed.value.taskId);
this.subscribe(clientId, parsed.value.taskId, parsed.value.projectId);
return;
}
this.unsubscribe(clientId, parsed.value.taskId);
this.unsubscribe(clientId, parsed.value.taskId, parsed.value.projectId);
}
private removeChannelSubscription(clientId: string, channel: string): void {
@@ -241,14 +290,15 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
subscribers.delete(clientId);
const taskId = fromBadgeChannel(channel);
// Extract taskId and projectId from channel for the event
const { taskId, projectId } = extractPartsFromChannel(channel);
if (subscribers.size === 0) {
this.channelSubscribers.delete(channel);
this.emit("subscription:changed", taskId, 0);
this.emit("subscription:changed", taskId ?? "", 0, projectId ?? "default");
return;
}
this.emit("subscription:changed", taskId, subscribers.size);
this.emit("subscription:changed", taskId ?? "", subscribers.size, projectId ?? "default");
}
private safeSend(ws: WebSocket, message: BadgeServerMessage): boolean {
@@ -294,12 +344,43 @@ export class WebSocketManager extends EventEmitter<WebSocketManagerEvents> {
}
}
function toBadgeChannel(taskId: string): string {
return `badge:${taskId}`;
/**
* Create a badge channel key with project scope.
* Format: badge:{projectId}:{taskId}
*/
function toBadgeChannel(projectId: string, taskId: string): string {
return `badge:${projectId}:${taskId}`;
}
function fromBadgeChannel(channel: string): string {
return channel.replace(/^badge:/, "");
/**
* Extract taskId from a badge channel key.
*/
function fromBadgeChannel(projectId: string, channel: string): string {
const prefix = `badge:${projectId}:`;
return channel.startsWith(prefix) ? channel.slice(prefix.length) : channel;
}
/**
* Extract taskId and projectId from any badge channel key.
*/
function extractPartsFromChannel(channel: string): { taskId: string | null; projectId: string | null } {
// Channel format: badge:{projectId}:{taskId}
const match = channel.match(/^badge:([^:]+):(.+)$/);
if (match) {
return { projectId: match[1], taskId: match[2] };
}
return { projectId: null, taskId: null };
}
/**
* Extract taskId from any badge channel key (without knowing the projectId).
* @deprecated Use extractPartsFromChannel instead
*/
function extractTaskIdFromChannel(channel: string): string | null {
// Channel format: badge:{projectId}:{taskId}
// We need to find the taskId after the second colon
const match = channel.match(/^badge:[^:]+:(.+)$/);
return match ? match[1] : null;
}
function parseClientMessage(raw: WebSocket.RawData):
@@ -324,11 +405,17 @@ function parseClientMessage(raw: WebSocket.RawData):
return { ok: false, error: "taskId is required" };
}
// projectId is optional - validates that it's a non-empty string if provided
if (value.projectId !== undefined && (typeof value.projectId !== "string" || value.projectId.trim().length === 0)) {
return { ok: false, error: "projectId must be a non-empty string" };
}
return {
ok: true,
value: {
type: value.type,
taskId: value.taskId.trim(),
projectId: value.projectId?.trim(),
},
};
} catch {