/** * Telegram bus local transport boundary * Zones: multi-instance bus, IPC transport, Windows named pipes, Unix sockets * Owns endpoint derivation, transport-kind detection, retry/error classification, and small timing policy. */ import { createHash } from "node:crypto"; import { createConnection } from "node:net"; import { join, resolve } from "node:path"; export type TelegramBusTransportKind = "pipe" | "socket"; export type TelegramBusTransportEventRecorder = ( phase: string, details: Record, ) => void; export type TelegramBusTransportEndpointDiagnostics = Record< string, unknown > & { endpoint: string; transport: TelegramBusTransportKind; }; export interface TelegramBusTransportRetryPolicy { attempts: number; delayMs: number; } export interface TelegramBusTransportRetryPolicyOverrides { attempts?: number; delayMs?: number; } export type TelegramBusTransportOperation = "registration" | "operation"; export const TELEGRAM_BUS_REGISTRATION_RETRY: TelegramBusTransportRetryPolicy = { attempts: 10, delayMs: 150, }; export const TELEGRAM_BUS_OPERATION_RETRY: TelegramBusTransportRetryPolicy = { attempts: 3, delayMs: 100, }; export function getTelegramBusPipePath(input: { agentDir: string; scope: string; }): string { const digest = createHash("sha256") .update(resolve(input.agentDir)) .digest("base64url") .slice(0, 16); const scope = input.scope.replace(/[^a-zA-Z0-9_.-]/g, "_").slice(0, 80); return `\\\\.\\pipe\\pi-telegram-${digest}-${scope}`; } export function isTelegramBusPipePath(endpoint: string): boolean { return /^\\\\[.?]\\pipe\\/i.test(endpoint); } export function getTelegramBusTransportKind( endpoint: string, ): TelegramBusTransportKind { return isTelegramBusPipePath(endpoint) ? "pipe" : "socket"; } export function getTelegramBusEndpointDiagnostics( endpoint: string, ): TelegramBusTransportEndpointDiagnostics { return { endpoint, transport: getTelegramBusTransportKind(endpoint), }; } export function getTelegramBusTransportRetryPolicy(input: { endpoint: string; operation: TelegramBusTransportOperation; overrides?: TelegramBusTransportRetryPolicyOverrides; }): TelegramBusTransportRetryPolicy | undefined { const base = input.operation === "registration" ? TELEGRAM_BUS_REGISTRATION_RETRY : isTelegramBusPipePath(input.endpoint) ? TELEGRAM_BUS_OPERATION_RETRY : undefined; if (!base && !input.overrides) return undefined; return { attempts: Math.max(1, input.overrides?.attempts ?? base?.attempts ?? 1), delayMs: Math.max(0, input.overrides?.delayMs ?? base?.delayMs ?? 0), }; } function normalizeTelegramBusEndpointScope(value: string): string { return value.replace(/[^a-zA-Z0-9_.-]/g, "_").slice(0, 80); } export function getTelegramBusLeaderEndpoint(input: { agentDir: string; platform: NodeJS.Platform | string; profileName?: string; }): string { const profileScope = input.profileName ? normalizeTelegramBusEndpointScope(input.profileName) : undefined; return input.platform === "win32" ? getTelegramBusPipePath({ agentDir: input.agentDir, scope: profileScope ? `bus-${profileScope}` : "bus", }) : join( input.agentDir, "tmp", "telegram", profileScope ? `bus.${profileScope}.sock` : "bus.sock", ); } export function getTelegramBusFollowerEndpoint(input: { agentDir: string; platform: NodeJS.Platform | string; instanceId: string; profileName?: string; }): string { const instanceScope = normalizeTelegramBusEndpointScope(input.instanceId); const profileScope = input.profileName ? normalizeTelegramBusEndpointScope(input.profileName) : undefined; return input.platform === "win32" ? getTelegramBusPipePath({ agentDir: input.agentDir, scope: profileScope ? `follower-${profileScope}-${instanceScope}` : `follower-${instanceScope}`, }) : join( input.agentDir, "tmp", "telegram", "followers", ...(profileScope ? [profileScope] : []), `${instanceScope}.sock`, ); } export interface TelegramBusTransportErrorInfo { message: string; code?: string; syscall?: string; kind: "connect" | "timeout" | "auth" | "protocol" | "unknown"; retryable: boolean; } export function classifyTelegramBusTransportError( error: unknown, ): TelegramBusTransportErrorInfo { const message = error instanceof Error ? error.message : String(error); const maybeNodeError = error as NodeJS.ErrnoException; const code = typeof maybeNodeError?.code === "string" ? maybeNodeError.code : undefined; const syscall = typeof maybeNodeError?.syscall === "string" ? maybeNodeError.syscall : undefined; const timeout = code === "ETIMEDOUT" || /timed out/i.test(message); const retryable = code === "ENOENT" || code === "ECONNREFUSED" || code === "EPIPE" || code === "ECONNRESET" || code === "EBUSY" || timeout; const kind = timeout ? "timeout" : code === "ENOENT" || code === "ECONNREFUSED" ? "connect" : "unknown"; return { message, code, syscall, kind, retryable }; } export function isRetryableTelegramBusTransportError(error: unknown): boolean { return classifyTelegramBusTransportError(error).retryable; } export function createTelegramBusTransportTimeoutError( message: string, ): NodeJS.ErrnoException { const error = new Error(message) as NodeJS.ErrnoException; error.code = "ETIMEDOUT"; return error; } export function delayTelegramBusTransportRetry(ms: number): Promise { return new Promise((resolve) => { setTimeout(resolve, ms); }); } export interface TelegramBusTransportProbeResult extends TelegramBusTransportEndpointDiagnostics { reachable: boolean; error?: TelegramBusTransportErrorInfo; } export function probeTelegramBusEndpoint(input: { endpoint: string; timeoutMs?: number; }): Promise { const timeoutMs = input.timeoutMs ?? 250; const diagnostics = getTelegramBusEndpointDiagnostics(input.endpoint); return new Promise((resolve) => { const socket = createConnection(input.endpoint); let settled = false; const settle = (result: TelegramBusTransportProbeResult): void => { if (settled) return; settled = true; clearTimeout(timeout); socket.destroy(); resolve(result); }; const timeout = setTimeout(() => { settle({ ...diagnostics, reachable: false, error: classifyTelegramBusTransportError( createTelegramBusTransportTimeoutError( "Timed out probing Telegram bus endpoint", ), ), }); }, timeoutMs); socket.once("connect", () => settle({ ...diagnostics, reachable: true })); socket.once("error", (error) => settle({ ...diagnostics, reachable: false, error: classifyTelegramBusTransportError(error), }), ); }); }