import { FileReadTimeoutError, readFileUtf8WithTimeout } from '@main/utils/fsRead'; import { getTeamsBasePath } from '@main/utils/pathDecoder'; import * as fs from 'fs'; import * as path from 'path'; import type { InboxMessage } from '@shared/types'; const MAX_INBOX_FILE_BYTES = 10 * 1024 * 1024; // 10MB — skip corrupt/oversized inbox files const INBOX_READ_CONCURRENCY = process.platform === 'win32' ? 4 : 12; async function mapLimit( items: readonly T[], limit: number, fn: (item: T) => Promise ): Promise { const results = new Array(items.length); let index = 0; const workerCount = Math.max(1, Math.min(limit, items.length)); const workers = new Array(workerCount).fill(0).map(async () => { while (true) { const i = index++; if (i >= items.length) return; results[i] = await fn(items[i]); } }); await Promise.all(workers); return results; } export class TeamInboxReader { async listInboxNames(teamName: string): Promise { const inboxDir = path.join(getTeamsBasePath(), teamName, 'inboxes'); let entries: string[]; try { entries = await fs.promises.readdir(inboxDir); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { return []; } throw error; } return entries .filter((name) => name.endsWith('.json') && !name.startsWith('.')) .map((name) => name.replace(/\.json$/, '')); } async getMessagesFor(teamName: string, member: string): Promise { const inboxPath = path.join(getTeamsBasePath(), teamName, 'inboxes', `${member}.json`); let raw: string; try { const stat = await fs.promises.stat(inboxPath); // Avoid hangs on non-regular files (FIFO, sockets) and unbounded memory usage on huge files. if (!stat.isFile() || stat.size > MAX_INBOX_FILE_BYTES) { return []; } raw = await readFileUtf8WithTimeout(inboxPath, 5_000); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { return []; } if (error instanceof FileReadTimeoutError) { return []; } throw error; } let parsed: unknown; try { parsed = JSON.parse(raw) as unknown; } catch { return []; } if (!Array.isArray(parsed)) { return []; } const messages: InboxMessage[] = []; for (const item of parsed) { if (!item || typeof item !== 'object') { continue; } const row = item as Partial; if ( typeof row.from !== 'string' || typeof row.text !== 'string' || typeof row.timestamp !== 'string' ) { continue; } messages.push({ from: row.from, to: typeof row.to === 'string' ? row.to : undefined, text: row.text, timestamp: row.timestamp, read: typeof row.read === 'boolean' ? row.read : false, summary: typeof row.summary === 'string' ? row.summary : undefined, color: typeof row.color === 'string' ? row.color : undefined, messageId: typeof row.messageId === 'string' ? row.messageId : undefined, source: typeof row.source === 'string' ? (row.source as InboxMessage['source']) : undefined, leadSessionId: typeof row.leadSessionId === 'string' ? row.leadSessionId : undefined, attachments: Array.isArray(row.attachments) ? row.attachments : undefined, toolSummary: typeof row.toolSummary === 'string' ? row.toolSummary : undefined, toolCalls: Array.isArray(row.toolCalls) ? (row.toolCalls as unknown[]) .filter( (tc): tc is { name: string; preview?: string } => tc != null && typeof tc === 'object' && typeof (tc as Record).name === 'string' ) .map((tc) => ({ name: tc.name, preview: typeof tc.preview === 'string' ? tc.preview : undefined, })) : undefined, }); } messages.sort((a, b) => { const bt = Date.parse(b.timestamp); const at = Date.parse(a.timestamp); if (Number.isNaN(bt) || Number.isNaN(at)) { return 0; } return bt - at; }); return messages; } async getMessages(teamName: string): Promise { const members = await this.listInboxNames(teamName); const chunks = await mapLimit(members, INBOX_READ_CONCURRENCY, async (member) => { try { const msgs = await this.getMessagesFor(teamName, member); for (const msg of msgs) { if (!msg.to) { msg.to = member; } } return msgs; } catch { return [] as InboxMessage[]; } }); const merged = chunks.flat(); merged.sort((a, b) => { const bt = Date.parse(b.timestamp); const at = Date.parse(a.timestamp); if (Number.isNaN(bt) || Number.isNaN(at)) { return 0; } return bt - at; }); return merged; } }