/** * @fileoverview RR V3 Port-RPC Client Composable (Shared) * @description RPC client for UI components to connect with Background Service Worker * * This composable is shared between Sidepanel, Builder, and other UI entrypoints. * * Responsibilities: * - Connect to background via chrome.runtime.Port * - Provide request/response RPC calls (with timeout and cancellation) * - Support event stream subscription * - Auto-reconnect with exponential backoff * * Design considerations: * - MV3 service worker may be terminated due to idle, causing Port disconnect * - Implement idempotent reconnection and subscription recovery */ import { computed, onUnmounted, ref, shallowRef, type ComputedRef, type Ref } from 'vue'; import type { JsonObject, JsonValue } from '@/entrypoints/background/record-replay-v3/domain/json'; import type { RunEvent } from '@/entrypoints/background/record-replay-v3/domain/events'; import type { RunId } from '@/entrypoints/background/record-replay-v3/domain/ids'; import { RR_V3_PORT_NAME, createRpcRequest, isRpcEvent, isRpcResponse, type RpcMethod, } from '@/entrypoints/background/record-replay-v3/engine/transport/rpc'; // ==================== Types ==================== /** RPC request options */ export interface RpcRequestOptions { /** Timeout in milliseconds, 0 means no timeout */ timeoutMs?: number; /** Abort signal for cancellation */ signal?: AbortSignal; } /** Composable configuration */ export interface UseRRV3RpcOptions { /** Default request timeout (ms) */ requestTimeoutMs?: number; /** Maximum reconnect attempts */ maxReconnectAttempts?: number; /** Base delay for reconnection (ms) */ baseReconnectDelayMs?: number; /** Auto-connect on initialization */ autoConnect?: boolean; /** Connection state change callback */ onConnectionChange?: (connected: boolean) => void; /** Error callback */ onError?: (error: string) => void; } /** Event listener function */ type EventListener = (event: RunEvent) => void; /** Pending request entry */ interface PendingRequest { method: RpcMethod; resolve: (value: JsonValue) => void; reject: (error: Error) => void; timeoutId: ReturnType | null; /** AbortSignal reference for cleanup */ signal?: AbortSignal; /** Abort handler for cleanup */ abortHandler?: () => void; } /** Composable return type */ export interface UseRRV3Rpc { // Connection state connected: Ref; connecting: Ref; reconnecting: Ref; reconnectAttempts: Ref; lastError: Ref; isReady: ComputedRef; // Diagnostics pendingCount: Ref; subscribedRunIds: Ref>; // Connection lifecycle connect: () => Promise; disconnect: (reason?: string) => void; ensureConnected: () => Promise; // RPC calls request: ( method: RpcMethod, params?: JsonObject, options?: RpcRequestOptions, ) => Promise; // Event subscription subscribe: (runId?: RunId | null) => Promise; unsubscribe: (runId?: RunId | null) => Promise; onEvent: (listener: EventListener) => () => void; } // ==================== Helpers ==================== function toErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } function isRunEvent(value: unknown): value is RunEvent { if (typeof value !== 'object' || value === null) return false; const obj = value as Record; return ( typeof obj.runId === 'string' && typeof obj.type === 'string' && typeof obj.seq === 'number' && typeof obj.ts === 'number' ); } // ==================== Composable ==================== /** * RR V3 Port-RPC client */ export function useRRV3Rpc(options: UseRRV3RpcOptions = {}): UseRRV3Rpc { // Configuration const DEFAULT_TIMEOUT_MS = options.requestTimeoutMs ?? 12_000; const MAX_RECONNECT_ATTEMPTS = options.maxReconnectAttempts ?? 8; const BASE_RECONNECT_DELAY_MS = options.baseReconnectDelayMs ?? 500; // Reactive state const connected = ref(false); const connecting = ref(false); const reconnecting = ref(false); const reconnectAttempts = ref(0); const lastError = ref(null); const pendingCount = ref(0); const subscribedRunIds = ref>([]); // Internal state (non-reactive) const port = shallowRef(null); const pendingRequests = new Map(); const eventListeners = new Set(); const desiredSubscriptions = new Set(); let connectPromise: Promise | null = null; let reconnectTimer: ReturnType | null = null; let manualDisconnect = false; // Computed const isReady = computed(() => connected.value && port.value !== null); // ==================== Internal Methods ==================== function setError(message: string | null): void { lastError.value = message; if (message) options.onError?.(message); } function setConnected(next: boolean): void { if (connected.value === next) return; connected.value = next; options.onConnectionChange?.(next); } function syncSubscriptionsSnapshot(): void { const arr = Array.from(desiredSubscriptions.values()); arr.sort((a, b) => { // Both null - equal if (a === null && b === null) return 0; // null comes first if (a === null) return -1; if (b === null) return 1; return String(a).localeCompare(String(b)); }); subscribedRunIds.value = arr; } /** * Clean up a pending request entry (timeout, abort listener) */ function cleanupPendingRequest(entry: PendingRequest): void { if (entry.timeoutId) { clearTimeout(entry.timeoutId); entry.timeoutId = null; } if (entry.signal && entry.abortHandler) { try { entry.signal.removeEventListener('abort', entry.abortHandler); } catch { // Ignore - signal may be invalid } } } function rejectAllPending(reason: string): void { const error = new Error(reason); for (const [requestId, entry] of pendingRequests) { cleanupPendingRequest(entry); entry.reject(error); pendingRequests.delete(requestId); } pendingCount.value = 0; } async function rehydrateSubscriptions(): Promise { if (!isReady.value || desiredSubscriptions.size === 0) return; for (const runId of desiredSubscriptions) { try { const params: JsonObject = runId === null ? {} : { runId }; await request('rr_v3.subscribe', params).catch(() => { // Best-effort, ignore errors }); } catch { // Ignore } } } function scheduleReconnect(): void { if (manualDisconnect || reconnectTimer) return; if (reconnectAttempts.value >= MAX_RECONNECT_ATTEMPTS) { reconnecting.value = false; setError('RR V3 RPC: max reconnect attempts reached'); return; } reconnecting.value = true; const delay = BASE_RECONNECT_DELAY_MS * Math.pow(2, reconnectAttempts.value); reconnectTimer = setTimeout(() => { reconnectTimer = null; reconnectAttempts.value += 1; void connect().then((ok) => { if (!ok) scheduleReconnect(); }); }, delay); } // ==================== Port Handlers ==================== function handlePortDisconnect(): void { // Capture disconnect reason for debugging const disconnectReason = chrome.runtime.lastError?.message; const reason = disconnectReason ? `RR V3 RPC disconnected: ${disconnectReason}` : 'RR V3 RPC disconnected'; port.value = null; setConnected(false); connecting.value = false; rejectAllPending(reason); // Update lastError for UI visibility (only on unexpected disconnect) if (!manualDisconnect) { setError(reason); scheduleReconnect(); } } function handlePortMessage(msg: unknown): void { // Handle RPC response if (isRpcResponse(msg)) { const entry = pendingRequests.get(msg.requestId); if (!entry) return; pendingRequests.delete(msg.requestId); pendingCount.value = pendingRequests.size; // Clean up timeout and abort listener cleanupPendingRequest(entry); if (msg.ok) { entry.resolve(msg.result as JsonValue); } else { entry.reject(new Error(msg.error || `RPC error: ${entry.method}`)); } return; } // Handle event push if (isRpcEvent(msg)) { const event = msg.event; if (!isRunEvent(event)) return; for (const listener of eventListeners) { try { listener(event); } catch (e) { console.error('[useRRV3Rpc] Event listener error:', e); } } } } // ==================== Public Methods ==================== async function connect(): Promise { if (isReady.value) return true; if (connectPromise) return connectPromise; connectPromise = (async () => { manualDisconnect = false; connecting.value = true; setError(null); try { if (typeof chrome === 'undefined' || !chrome.runtime?.connect) { setError('chrome.runtime.connect not available'); return false; } const p = chrome.runtime.connect({ name: RR_V3_PORT_NAME }); port.value = p; // Reset reconnect state reconnectAttempts.value = 0; reconnecting.value = false; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } p.onMessage.addListener(handlePortMessage); p.onDisconnect.addListener(handlePortDisconnect); setConnected(true); // Restore subscriptions void rehydrateSubscriptions(); return true; } catch (error) { setError(`Connection failed: ${toErrorMessage(error)}`); return false; } finally { connecting.value = false; connectPromise = null; } })(); return connectPromise; } function disconnect(reason?: string): void { manualDisconnect = true; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } reconnecting.value = false; const p = port.value; port.value = null; setConnected(false); connecting.value = false; rejectAllPending(reason || 'RR V3 RPC: client disconnected'); if (p) { try { p.onMessage.removeListener(handlePortMessage); p.onDisconnect.removeListener(handlePortDisconnect); p.disconnect(); } catch { // Ignore } } } async function ensureConnected(): Promise { if (isReady.value) return true; return connect(); } async function request( method: RpcMethod, params?: JsonObject, reqOptions: RpcRequestOptions = {}, ): Promise { const ready = await ensureConnected(); const p = port.value; if (!ready || !p) { throw new Error('RR V3 RPC: not connected'); } const timeoutMs = reqOptions.timeoutMs ?? DEFAULT_TIMEOUT_MS; const { signal } = reqOptions; if (signal?.aborted) { throw new Error('RPC request already aborted'); } const req = createRpcRequest(method, params); return new Promise((resolve, reject) => { const entry: PendingRequest = { method, resolve: resolve as (value: JsonValue) => void, reject, timeoutId: null, signal, }; // Helper to complete request with cleanup const complete = (fn: () => void) => { pendingRequests.delete(req.requestId); pendingCount.value = pendingRequests.size; cleanupPendingRequest(entry); fn(); }; // Timeout handling if (timeoutMs > 0) { entry.timeoutId = setTimeout(() => { complete(() => reject(new Error(`RPC timeout (${timeoutMs}ms): ${method}`))); }, timeoutMs); } // Abort handling if (signal) { const onAbort = () => { complete(() => reject(new Error('RPC request aborted'))); }; entry.abortHandler = onAbort; signal.addEventListener('abort', onAbort, { once: true }); } pendingRequests.set(req.requestId, entry); pendingCount.value = pendingRequests.size; try { p.postMessage(req); } catch (e) { complete(() => reject(new Error(`Failed to send RPC request: ${toErrorMessage(e)}`))); } }); } async function subscribe(runId: RunId | null = null): Promise { desiredSubscriptions.add(runId); syncSubscriptionsSnapshot(); try { const params: JsonObject = runId === null ? {} : { runId }; await request('rr_v3.subscribe', params); return true; } catch (error) { setError(toErrorMessage(error)); return false; } } async function unsubscribe(runId: RunId | null = null): Promise { desiredSubscriptions.delete(runId); syncSubscriptionsSnapshot(); try { const params: JsonObject = runId === null ? {} : { runId }; await request('rr_v3.unsubscribe', params); return true; } catch (error) { setError(toErrorMessage(error)); return false; } } function onEvent(listener: EventListener): () => void { eventListeners.add(listener); return () => eventListeners.delete(listener); } // ==================== Lifecycle ==================== onUnmounted(() => { disconnect('Component unmounted'); }); if (options.autoConnect) { void ensureConnected(); } return { connected, connecting, reconnecting, reconnectAttempts, lastError, isReady, pendingCount, subscribedRunIds, connect, disconnect, ensureConnected, request, subscribe, unsubscribe, onEvent, }; }