backend-ts/src/app-utils.ts
Rama main · réplica de lectura del repositorio original.
import { existsSync, readdirSync, rmSync } from 'node:fs';
import { isAbsolute, resolve } from 'node:path';
import { randomUUID } from 'node:crypto';
import fastifyStatic from '@fastify/static';
import type { FastifyInstance } from 'fastify';
import type { SpawnPtyOptions, PtyLike } from './harness/terminal/remote-terminal.js';
import type { ContainerRegistry } from './harness/container/container-registry.js';
import type { RuntimeChannelServer } from './harness/container/runtime-channel-server.js';
import type { DockerToolRouter } from './harness/container/docker-tool-router.js';
import type { SessionService } from './session/session.service.js';
import type { TaskTerminalService } from './session/task-terminal.service.js';
import type { FeishuInboundQueueRepository } from './feishu/inbound-queue.repository.js';
import type { MysqlFeishuProgressCardRepository } from './feishu/progress-card.repository.js';
import type { AuditLogService } from './audit/audit.service.js';
import { TERMINAL_AUDIT_META, type TerminalAuditRecorder } from './harness/terminal/terminal-manager.js';
import { truncateAuditError } from './audit/audit.interceptor.js';
import { PtyChannelHub } from './harness/container/pty-bridge.js';
import { harnessLog } from './harness/log.js';
// ========== 通用工具 ==========
export function expandHome(v: string): string {
if (v.startsWith('$HOME')) return v.replace(/^\$HOME/, process.env.HOME ?? '');
if (v.startsWith('~/')) return resolve(process.env.HOME ?? '', v.slice(2));
return v;
}
/**
* 共享目录路径解析（共享目录方案 §4.1 误配防护）：空值或相对路径若被 resolve 静默
* 按 cwd 解析，容器模式下 assertSafe 的 owningRoot 校验对 share 条目恒真通过
* （shareDir 自身即登记根），唯一兜底的父目录防护又依赖 cwd 与 distDir 的相对
* 布局——误配时整个后端工作树（含 .env 密钥）会被 ro 挂进所有用户容器。
* 启动即拒：必须在 resolve 之前校验原始值，否则相对路径已被静默转绝对。
*/
export function resolveShareDir(raw: string): string {
const expanded = expandHome(raw);
if (!expanded || !isAbsolute(expanded)) {
throw new Error(`MAO_SHARE_DIR must be a non-empty absolute path, got '${raw}'`);
}
return resolve(expanded);
}
export function parseSqlTimeMs(value: string | null | undefined): number | null {
if (value == null || value === '') return null;
const ms = Date.parse(value.includes('T') ? value : value.replace(' ', 'T'));
return Number.isFinite(ms) ? ms : null;
}
export function runtimeSessionCleanup(runtimeRoot: string): (userId: number, sessionId: number) => void {
return (userId: number, sessionId: number): void => {
// 归属者目录无条件回收（即使根目录枚举失败也不影响）。
const cleaned = new Set<string>([resolve(runtimeRoot, String(userId), String(sessionId))]);
try {
// 群聊会话的运行期产物按「运行身份」分别落盘（技术方案 2026-11-04 §5.2/§5.9）：
// dump_user_messages / open_web_page 缓存 / 技能软链等都在 {runtimeRoot}/{发言者}/{sid} 下，
// 删除会话时不能只回收归属者目录，否则每个发言者都会留下永不回收的残留。
// sessionId 全局唯一，按 {runtimeRoot}/*/{sid} 扫描删除（删除是低频操作，可接受一次枚举）。
// 严格只认纯数字一级目录（用户 id 形态）：runtimeRoot 下还有 channels 等非用户目录，
// channel 目录名恰为 userId——若全量扫描，sessionId == 某 userId 时会误删该用户
// 的容器通道目录（socket/.token），打挂其 docker 运行面（评审 P0，2026-11-04）。
for (const entry of readdirSync(runtimeRoot, { withFileTypes: true })) {
if (!entry.isDirectory() || !/^\d+$/.test(entry.name)) continue;
cleaned.add(resolve(runtimeRoot, entry.name, String(sessionId)));
}
} catch (e) {
console.error(`Failed to enumerate runtime dirs for session ${sessionId}`, e);
}
try {
for (const sessionDir of cleaned) {
if (!existsSync(sessionDir)) continue;
try {
rmSync(sessionDir, { recursive: true, force: true });
} catch (e) {
// 单个目录 EBUSY/EACCES 不应跳过其余发言者目录（残留可下次再清，但不能漏清）。
console.error(`Failed to clean runtime dir ${sessionDir} for session ${sessionId}`, e);
}
}
} catch (e) {
console.error(`Failed to clean runtime dir for session ${sessionId}`, e);
}
};
}
// ========== 上传静态托管 ==========
// 分隔符双兼容：fastifyStatic setHeaders 拿到的是平台本地路径（Windows 为反斜杠），
// 正则只认 / 会漏掉 electron 更新清单，导致 Windows 部署错误长缓存。
// 字符类内写作 [\\/]（反斜杠 + 正斜杠都接受）
const UPLOAD_CACHE_EXCLUDED = /(^|[\\/])(releases[\\/][^\\/]*\.(json|ya?ml)|latest[^\\/]*\.ya?ml)$/i;
function uploadCacheHeaders(path: string): Record<string, string> {
if (UPLOAD_CACHE_EXCLUDED.test(path)) return { 'Cache-Control': 'no-cache' };
return { 'Cache-Control': 'public, max-age=604800' };
}
export async function registerUploadStatic(app: FastifyInstance, uploadDir: string, apiPrefix: string): Promise<void> {
await app.register(fastifyStatic, {
root: uploadDir,
prefix: '/uploads/',
decorateReply: false,
setHeaders: (reply, path) => {
for (const [k, v] of Object.entries(uploadCacheHeaders(path))) reply.header(k, v);
},
});
await app.register(fastifyStatic, {
root: uploadDir,
prefix: `${apiPrefix.replace(/\/$/, '')}/uploads/`,
decorateReply: false,
setHeaders: (reply, path) => {
for (const [k, v] of Object.entries(uploadCacheHeaders(path))) reply.header(k, v);
},
});
}
// ========== 终端审计 ==========
export function createTerminalAudit(auditService: AuditLogService): TerminalAuditRecorder {
return (event, terminal, ctx) => {
const meta = TERMINAL_AUDIT_META[event];
void auditService.record({
action: meta.action,
objectType: 'terminal',
objectId: terminal.terminalId,
method: meta.method,
path: meta.path(terminal.sessionId, terminal.terminalId),
userId: terminal.userId,
username: ctx?.username ?? null,
ip: ctx?.ip ?? null,
status: ctx?.errorMessage == null ? 200 : 500,
success: ctx?.errorMessage == null ? 1 : 0,
errorMessage: truncateAuditError(ctx?.errorMessage),
}).catch((e) => console.error('Failed to record terminal audit log', e));
};
}
// ========== 容器 PTY 工厂 ==========
export interface ContainerPtyFactoryDeps {
dockerRouter: DockerToolRouter;
containerRegistry: ContainerRegistry;
channelServer: RuntimeChannelServer;
/** 键为 containerKey(userId, agentId)（2026-11-07 复合容器方案 §5.7）。 */
ptyChannelHubs: Map<string, PtyChannelHub>;
ptyHubGeneration: Map<string, number>;
nodePtyFactory: (options: SpawnPtyOptions) => Promise<PtyLike> | PtyLike;
}
export function createContainerPtyFactory(deps: ContainerPtyFactoryDeps): (options: SpawnPtyOptions) => Promise<PtyLike> {
const { dockerRouter, containerRegistry, channelServer, ptyChannelHubs, ptyHubGeneration, nodePtyFactory } = deps;
return async (options: SpawnPtyOptions): Promise<PtyLike> => {
const userId = options.userId;
if (userId != null && await dockerRouter.enabledFor(userId)) {
// agentId 由 terminal.routes 从会话填充；缺省（旧调用方/测试）视为无法路由，
// 明确报错而不是落到宿主机 node-pty（docker 用户的工作区文件宿主机不可见）
if (options.agentId == null) {
throw new Error(`容器终端缺少 Agent 维度（agentId），无法路由: userId=${userId}`);
}
const agentId = options.agentId;
const key = `${userId}:${agentId}`;
const entry = await containerRegistry.ensureRunning(userId, agentId, options.cwd);
const connection = channelServer.getConnection(userId, agentId);
if (!connection) throw new Error(`容器终端通道未就绪: userId=${userId}, agentId=${agentId}`);
let hub = ptyChannelHubs.get(key);
if (!hub || ptyHubGeneration.get(key) !== entry.startedAtMs) {
hub?.dispose();
hub = new PtyChannelHub(connection);
ptyChannelHubs.set(key, hub);
ptyHubGeneration.set(key, entry.startedAtMs);
} else {
hub.bindConnection(connection);
}
const bridge = await hub.open(userId, options);
harnessLog('info', `[container] terminal pty opened in container: user=${userId}, agent=${agentId}, pid=${bridge.pid}`);
return bridge;
}
return nodePtyFactory(options);
};
}
// ========== 取消持久化 ==========
export interface PersistCancelledDeps {
sessionService: SessionService;
taskTerminal: TaskTerminalService;
feishuProgressCardRepo: MysqlFeishuProgressCardRepository;
}
export function createPersistCancelledHelper(deps: PersistCancelledDeps) {
const { sessionService, taskTerminal, feishuProgressCardRepo } = deps;
return async (sessionId: number): Promise<boolean> => {
const session = await sessionService.getSession(sessionId).catch(() => null);
if (session == null) return false;
if (session.phase !== 'RUNNING' && session.phase !== 'RESUMING') return false;
await sessionService.cleanupIncompleteTail(sessionId);
await taskTerminal.finishExecution(sessionId, session.userId, 'CANCELLED', randomUUID());
try { await feishuProgressCardRepo.deleteBySessionId(sessionId); } catch (error) {
console.warn(`清理飞书进度卡片映射失败, sessionId=${sessionId}: ${error instanceof Error ? error.message : String(error)}`);
}
return true;
};
}
// ========== 平台队列接力 ==========
export interface PlatformQueueRelayDeps {
sessionService: { getSession(sessionId: number): Promise<{ userId?: number | null } | null> };
/** WS 侧排队消息消费入口（streaming-ws-handler 的 autoConsumeQueue）。 */
wsHandler: { autoConsumeQueue(sessionId: number, userId: number, options?: { relayToIm?: boolean }): Promise<void> };
}
/**
* IM （飞书/钉钉/微信）执行收尾、入站队列排空且会话空闲后的接力消费：
* 按会话归属者消费 message_queue（定时任务/Webhook/插队）——否则「定时任务在会话忙时入队、
* 之后该会话只有 IM 通道执行」的消息会永远停在 QUEUED（管理后台显示「排队中」）。
* relayToIm=false 防止「IM 链 → 平台链 → IM 链」互相无限传递。失败静默留日志，不外抛。
*/
export function createPlatformQueueRelay(deps: PlatformQueueRelayDeps): (sessionId: number) => void {
const { sessionService, wsHandler } = deps;
return (sessionId: number): void => {
void (async () => {
const session = await sessionService.getSession(sessionId).catch(() => null);
if (session?.userId == null) return;
await wsHandler.autoConsumeQueue(sessionId, session.userId, { relayToIm: false });
})().catch((error: unknown) => {
console.warn(`IM 队列排空后接力消费平台队列失败, sessionId=${sessionId}: ${error instanceof Error ? error.message : String(error)}`);
});
};
}
// ========== 排队位置 ==========
export interface QueuePositionDeps {
feishuInboundQueueRepo: FeishuInboundQueueRepository;
}
export function createQueuePositionHelper(deps: QueuePositionDeps) {
const { feishuInboundQueueRepo } = deps;
return async (sessionId: number): Promise<number> => (await feishuInboundQueueRepo.countPending(sessionId));
}