import { EventEmitter } from "node:events"; 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), 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. */ 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 */ timestamp: string; /** PR badge snapshot data; null = badge cleared; omitted = no change */ prInfo?: PrInfo | null; /** Issue badge snapshot data; null = badge cleared; omitted = no change */ issueInfo?: IssueInfo | null; } /** * BadgePubSub adapter interface for cross-instance badge snapshot fan-out. * * Implementations must: * - Validate incoming messages against the BadgePubSubMessage contract * - Emit 'message' events with validated BadgePubSubMessage payloads * - Ignore malformed messages without crashing * - Support graceful shutdown via dispose() */ export interface BadgePubSub extends EventEmitter { /** Publish a badge snapshot to the shared bus */ publish(message: BadgePubSubMessage): void | Promise; /** Start receiving messages from the shared bus */ start(): void | Promise; /** Stop receiving messages and clean up resources */ dispose(): void | Promise; } export interface BadgePubSubEvents { message: [BadgePubSubMessage]; error: [Error]; } /** Factory function type for creating BadgePubSub instances */ export type BadgePubSubFactory = (options: BadgePubSubFactoryOptions) => BadgePubSub; export interface BadgePubSubFactoryOptions { /** Redis URL (when using Redis adapter) */ redisUrl?: string; /** Pub/sub channel name (when using Redis adapter) */ channel?: string; /** Server instance identifier (for deduplication) */ sourceId: string; } /** * In-memory BadgePubSub adapter for single-instance deployments and testing. * * This adapter does not actually communicate across instances; it's useful for: * - Local single-instance deployments without Redis * - Testing multi-instance scenarios with injected shared state */ export class InMemoryBadgePubSub extends EventEmitter implements BadgePubSub { private started = false; private disposed = false; publish(message: BadgePubSubMessage): void { if (this.disposed) return; // In single-instance mode, we immediately emit the message locally // This allows tests to verify the publish/subscribe flow setImmediate(() => { if (!this.disposed) { this.emit("message", message); } }); } start(): void { if (this.disposed) return; this.started = true; } dispose(): void { if (this.disposed) return; this.disposed = true; this.started = false; this.removeAllListeners(); } } /** * Redis-backed BadgePubSub adapter for multi-instance deployments. * * Requires REDIS_URL environment variable or redisUrl option. * Uses Redis pub/sub for cross-instance message delivery. * * Environment variables: * - FUSION_BADGE_PUBSUB_REDIS_URL: Redis connection URL * - FUSION_BADGE_PUBSUB_CHANNEL: Channel name (default: "fusion:badge-updates") */ export class RedisBadgePubSub extends EventEmitter implements BadgePubSub { private subscriber: import("ioredis").Redis | null = null; private publisher: import("ioredis").Redis | null = null; private started = false; private disposed = false; private readonly redisUrl: string | undefined; private readonly channel: string; private readonly sourceId: string; constructor(options: BadgePubSubFactoryOptions) { super(); this.redisUrl = options.redisUrl ?? getRedisUrlFromEnv(); this.channel = options.channel ?? getChannelFromEnv(); this.sourceId = options.sourceId; } async start(): Promise { if (this.disposed || this.started) return; const redisUrl = this.redisUrl; if (!redisUrl) { throw new Error("Redis URL is required for RedisBadgePubSub"); } try { const { Redis } = await import("ioredis"); this.subscriber = new Redis(redisUrl); this.publisher = new Redis(redisUrl); // Handle connection errors without crashing the process this.subscriber.on("error", (err: Error) => { this.emit("error", new Error(`Redis subscriber error: ${err.message}`)); }); this.publisher.on("error", (err: Error) => { this.emit("error", new Error(`Redis publisher error: ${err.message}`)); }); // Subscribe to channel and handle messages await this.subscriber.subscribe(this.channel); this.subscriber.on("message", (_channel: string, message: string) => { if (this.disposed) return; const parsed = parseBadgePubSubMessage(message, this.sourceId); if (parsed.ok) { this.emit("message", parsed.value); } // Silently ignore malformed messages }); this.started = true; } catch (err) { // Clean up on failure await this.cleanupConnections(); throw err; } } async publish(message: BadgePubSubMessage): Promise { if (this.disposed || !this.publisher || !this.started) return; try { const payload = JSON.stringify(message); await this.publisher.publish(this.channel, payload); } catch (err) { // Emit error but don't crash - caller can decide to retry this.emit("error", err instanceof Error ? err : new Error(String(err))); } } async dispose(): Promise { if (this.disposed) return; this.disposed = true; this.started = false; await this.cleanupConnections(); this.removeAllListeners(); } private async cleanupConnections(): Promise { try { if (this.subscriber) { await this.subscriber.unsubscribe(); await this.subscriber.quit(); this.subscriber = null; } } catch { // Ignore cleanup errors } try { if (this.publisher) { await this.publisher.quit(); this.publisher = null; } } catch { // Ignore cleanup errors } } } /** * Parse and validate a BadgePubSubMessage from a JSON string. * * Returns { ok: true, value } for valid messages. * Returns { ok: false } for malformed messages (should be silently ignored). * * The sourceId check prevents processing our own echoed messages. * * @internal Exported for testing purposes */ export function parseBadgePubSubMessage( json: string, localSourceId: string ): { ok: true; value: BadgePubSubMessage } | { ok: false } { try { const parsed = JSON.parse(json) as Partial; // Validate required fields if (typeof parsed.sourceId !== "string" || parsed.sourceId.length === 0) { return { ok: false }; } // Ignore our own messages (prevent echo loops) if (parsed.sourceId === localSourceId) { return { ok: false }; } if (typeof parsed.taskId !== "string" || parsed.taskId.length === 0) { return { ok: false }; } if (typeof parsed.timestamp !== "string" || parsed.timestamp.length === 0) { return { ok: false }; } // Validate prInfo if present (can be null, object, or omitted) if (parsed.prInfo !== undefined && parsed.prInfo !== null) { if (typeof parsed.prInfo !== "object") { return { ok: false }; } // Basic validation of required PrInfo fields const pr = parsed.prInfo as Partial; if (typeof pr.url !== "string" || typeof pr.number !== "number") { return { ok: false }; } } // Validate issueInfo if present (can be null, object, or omitted) if (parsed.issueInfo !== undefined && parsed.issueInfo !== null) { if (typeof parsed.issueInfo !== "object") { return { ok: false }; } // Basic validation of required IssueInfo fields const issue = parsed.issueInfo as Partial; if (typeof issue.url !== "string" || typeof issue.number !== "number") { return { ok: false }; } } return { ok: true, value: parsed as BadgePubSubMessage, }; } catch { return { ok: false }; } } /** * Create a BadgePubSub adapter based on environment configuration. * * If FUSION_BADGE_PUBSUB_REDIS_URL is set, returns a RedisBadgePubSub. * Otherwise, returns an InMemoryBadgePubSub for local-only operation. */ export function createBadgePubSub(options: { sourceId: string }): BadgePubSub { const redisUrl = getRedisUrlFromEnv(); if (redisUrl) { return new RedisBadgePubSub({ redisUrl, channel: getChannelFromEnv(), sourceId: options.sourceId, }); } return new InMemoryBadgePubSub(); } function getRedisUrlFromEnv(): string | undefined { return process.env.FUSION_BADGE_PUBSUB_REDIS_URL; } function getChannelFromEnv(): string { return process.env.FUSION_BADGE_PUBSUB_CHANNEL ?? "fusion:badge-updates"; } export { getRedisUrlFromEnv, getChannelFromEnv };