import { decryptText, encryptText } from './crypto.js'; import { getPool, withTransaction } from './db.js'; import type { AppConfig } from './config.js'; import { AgentBusListener, type AgentBusListenerOptions, type AgentBusLogger, type AgentBusSocketFactory } from './agentbus.js'; import { TaskError, type TaskContext } from './task-service.js'; import { diagnosticError, diagnosticMetadataKeys } from './diagnostics.js'; export interface PublicAgentBusChannel { id: string; organization_id: string; channel_type: 'agentbus'; display_name: string; external_user_ref: string | null; agentbus_bot_address: string | null; enabled: boolean; status: 'disabled' | 'connecting' | 'connected' | 'error'; key_configured: true; deletable: boolean; last_connected_at: string | null; last_error: string | null; created_at: string; updated_at: string; } export interface AgentBusChannelSecret extends PublicAgentBusChannel { ws_url: string; ws_token: string; bot_address: string; } export interface AgentBusChannelMutation { displayName: string; externalUserRef?: string; agentbusKey: string; botAddress?: string; enabled?: boolean; } export interface AgentBusChannelUpdate { displayName?: string; externalUserRef?: string; botAddress?: string; enabled?: boolean; } function text(value: unknown): string { return value == null ? '' : String(value).trim(); } const LEGACY_CHANNEL_REF = 'legacy-env'; const DUPLICATE_LEGACY_CHANNEL_ERROR = '检测到重复的兼容渠道,已停用。'; function iso(value: unknown): string | null { return value ? new Date(String(value)).toISOString() : null; } function publicChannel( row: Record, legacyEnvironmentManaged = false ): PublicAgentBusChannel { const status = text(row.status); const environmentManaged = text(row.external_user_ref) === LEGACY_CHANNEL_REF && legacyEnvironmentManaged; return { id: text(row.id), organization_id: text(row.organization_id), channel_type: 'agentbus', display_name: text(row.display_name), external_user_ref: text(row.external_user_ref) || null, agentbus_bot_address: text(row.agentbus_bot_address) || null, enabled: row.enabled === true || text(row.enabled) === 'true', status: ['disabled', 'connecting', 'connected', 'error'].includes(status) ? status as PublicAgentBusChannel['status'] : 'error', key_configured: true, deletable: !environmentManaged, last_connected_at: iso(row.last_connected_at), last_error: text(row.last_error) || null, created_at: new Date(String(row.created_at)).toISOString(), updated_at: new Date(String(row.updated_at)).toISOString() }; } function booleanValue(value: unknown): boolean { return value === true || text(value) === 'true'; } /** * The channel directory is persisted in PostgreSQL, while listener state is * owned by the current process. Prefer the live listener state whenever a * channel has an active listener so an older process cannot leave the panel * showing a stale `disabled` status after a restart. */ export function mergeRuntimeChannelStatuses( channels: PublicAgentBusChannel[], runtimeChannels: ReadonlyArray> ): PublicAgentBusChannel[] { const runtimeById = new Map( runtimeChannels .map((runtime) => [text(runtime.channel_id), runtime] as const) .filter(([channelId]) => Boolean(channelId)) ); return channels.map((channel) => { const runtime = runtimeById.get(channel.id); if (!runtime || !channel.enabled || !booleanValue(runtime.enabled)) return channel; if (booleanValue(runtime.connected) && booleanValue(runtime.session_ready)) { return { ...channel, status: 'connected', last_error: null }; } if (channel.status === 'disabled' || channel.status === 'connecting') { return { ...channel, status: 'connecting', last_error: null }; } return channel; }); } export class AgentBusChannelService { private readonly runtimeStatusWrites = new Map>(); constructor( private readonly config: AppConfig, private readonly logger?: AgentBusLogger ) {} private log( level: 'info' | 'warn' | 'error', metadata: Record, message: string ): void { try { this.logger?.[level](metadata, message); } catch { // Channel persistence and encryption never depend on operational logs. } } private legacyEnvironmentManaged(): boolean { return this.config.agentBusEnabled && Boolean(text(this.config.AGENTBUS_WS_URL)) && Boolean(text(this.config.AGENTBUS_WS_TOKEN)) && Boolean(text(this.config.AGENTBUS_BOT_ADDRESS)); } async list(organizationId: string): Promise { const result = await getPool(this.config).query( `WITH canonical_legacy AS ( SELECT DISTINCT ON (organization_id) id, organization_id FROM user_channels WHERE external_user_ref = '${LEGACY_CHANNEL_REF}' ORDER BY organization_id, created_at ASC, id ASC ) SELECT channel.id, channel.organization_id, channel.display_name, channel.external_user_ref, channel.agentbus_bot_address, channel.enabled, channel.status, channel.last_connected_at, channel.last_error, channel.created_at, channel.updated_at FROM user_channels channel LEFT JOIN canonical_legacy legacy ON legacy.id = channel.id WHERE channel.organization_id = $1 AND (channel.external_user_ref IS DISTINCT FROM '${LEGACY_CHANNEL_REF}' OR legacy.id IS NOT NULL) ORDER BY channel.created_at ASC, channel.id ASC`, [organizationId] ); const legacyEnvironmentManaged = this.legacyEnvironmentManaged(); return (result.rows as Record[]) .map((row) => publicChannel(row, legacyEnvironmentManaged)); } async listEnabledSecrets(organizationId: string): Promise { const result = await getPool(this.config).query( `WITH canonical_legacy AS ( SELECT DISTINCT ON (organization_id) id, organization_id FROM user_channels WHERE external_user_ref = '${LEGACY_CHANNEL_REF}' ORDER BY organization_id, created_at ASC, id ASC ) SELECT channel.id, channel.organization_id, channel.display_name, channel.external_user_ref, channel.agentbus_bot_address, channel.enabled, channel.status, channel.last_connected_at, channel.last_error, channel.created_at, channel.updated_at, channel.agentbus_ws_token_ciphertext FROM user_channels channel LEFT JOIN canonical_legacy legacy ON legacy.id = channel.id WHERE channel.organization_id = $1 AND channel.enabled = true AND (channel.external_user_ref IS DISTINCT FROM '${LEGACY_CHANNEL_REF}' OR legacy.id IS NOT NULL) ORDER BY channel.created_at ASC, channel.id ASC`, [organizationId] ); const channels: AgentBusChannelSecret[] = []; for (const row of result.rows as Record[]) { const botAddress = text(row.agentbus_bot_address) || text(this.config.AGENTBUS_BOT_ADDRESS); const wsUrl = text(this.config.AGENTBUS_WS_URL); try { const wsToken = decryptText(this.config, text(row.agentbus_ws_token_ciphertext)); if (!wsUrl || !wsToken || !botAddress) { this.log('warn', { agentbus_event: 'channel_configuration_incomplete', channel_id: text(row.id), ws_url_present: Boolean(wsUrl), ws_token_present: Boolean(wsToken), bot_address_present: Boolean(botAddress) }, 'AgentBus channel configuration is incomplete'); await this.setRuntimeStatus( organizationId, text(row.id), 'error', '渠道连接缺少全局 WebSocket 地址或 bot address。' ); continue; } channels.push({ ...publicChannel(row, this.legacyEnvironmentManaged()), ws_url: wsUrl, ws_token: wsToken, bot_address: botAddress }); } catch (error) { this.log('error', { agentbus_event: 'channel_key_decryption_failed', channel_id: text(row.id), ...diagnosticError(error, 'agentbus_channel_key_decryption_failed') }, 'AgentBus channel key decryption failed'); await this.setRuntimeStatus(organizationId, text(row.id), 'error', '渠道 key 无法解密。'); } } return channels; } async create(context: TaskContext, input: AgentBusChannelMutation): Promise { const displayName = text(input.displayName).slice(0, 120); const agentbusKey = text(input.agentbusKey); if (!displayName) throw new TaskError('channel_name_required', '渠道名称不能为空。', 400); if (!agentbusKey) throw new TaskError('channel_key_required', 'AgentBus key 不能为空。', 400); try { const result = await withTransaction(this.config, async (client) => { const inserted = await client.query( `INSERT INTO user_channels (organization_id, display_name, external_user_ref, agentbus_ws_token_ciphertext, agentbus_bot_address, enabled, status, created_by) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id, organization_id, display_name, external_user_ref, agentbus_bot_address, enabled, status, last_connected_at, last_error, created_at, updated_at`, [ context.organizationId, displayName, text(input.externalUserRef).slice(0, 200) || null, encryptText(this.config, agentbusKey), text(input.botAddress).slice(0, 200) || null, input.enabled !== false, input.enabled === false ? 'disabled' : 'connecting', context.userId || null ] ); await this.audit(client, context, 'agentbus_channel.created', text(inserted.rows[0].id), { display_name: displayName, external_user_ref_present: Boolean(text(input.externalUserRef)), enabled: input.enabled !== false }); return inserted.rows[0] as Record; }); return publicChannel(result, this.legacyEnvironmentManaged()); } catch (error) { if (error && typeof error === 'object' && String((error as { code?: unknown }).code || '') === '23505') { throw new TaskError('channel_name_conflict', '同一组织下的渠道名称已存在。', 409); } throw error; } } async update( context: TaskContext, channelId: string, input: AgentBusChannelUpdate ): Promise { const current = await this.getRow(context.organizationId, channelId); const displayName = input.displayName === undefined ? text(current.display_name) : text(input.displayName).slice(0, 120); if (!displayName) throw new TaskError('channel_name_required', '渠道名称不能为空。', 400); const currentExternalUserRef = text(current.external_user_ref) || null; const externalUserRef = currentExternalUserRef === LEGACY_CHANNEL_REF ? LEGACY_CHANNEL_REF : input.externalUserRef === undefined ? currentExternalUserRef : (text(input.externalUserRef).slice(0, 200) || null); const enabled = input.enabled === undefined ? current.enabled === true || text(current.enabled) === 'true' : input.enabled; try { const result = await withTransaction(this.config, async (client) => { const updated = await client.query( `UPDATE user_channels SET display_name = $1, external_user_ref = $2, agentbus_bot_address = $3, enabled = $4, status = CASE WHEN $4 THEN 'connecting' ELSE 'disabled' END, last_error = CASE WHEN $4 THEN NULL ELSE last_error END, updated_at = now() WHERE organization_id = $5 AND id = $6 RETURNING id, organization_id, display_name, external_user_ref, agentbus_bot_address, enabled, status, last_connected_at, last_error, created_at, updated_at`, [ displayName, externalUserRef, input.botAddress === undefined ? (text(current.agentbus_bot_address) || null) : (text(input.botAddress).slice(0, 200) || null), enabled, context.organizationId, channelId ] ); if (!updated.rowCount) throw new TaskError('channel_not_found', '用户渠道不存在。', 404); await this.audit(client, context, 'agentbus_channel.updated', channelId, { enabled, display_name: displayName }); return updated.rows[0] as Record; }); return publicChannel(result, this.legacyEnvironmentManaged()); } catch (error) { if (error && typeof error === 'object' && String((error as { code?: unknown }).code || '') === '23505') { throw new TaskError('channel_name_conflict', '同一组织下的渠道名称已存在。', 409); } throw error; } } async rotateKey(context: TaskContext, channelId: string, agentbusKey: string): Promise { const key = text(agentbusKey); if (!key) throw new TaskError('channel_key_required', 'AgentBus key 不能为空。', 400); const result = await withTransaction(this.config, async (client) => { const updated = await client.query( `UPDATE user_channels SET agentbus_ws_token_ciphertext = $1, status = CASE WHEN enabled THEN 'connecting' ELSE 'disabled' END, last_error = NULL, updated_at = now() WHERE organization_id = $2 AND id = $3 RETURNING id, organization_id, display_name, external_user_ref, agentbus_bot_address, enabled, status, last_connected_at, last_error, created_at, updated_at`, [encryptText(this.config, key), context.organizationId, channelId] ); if (!updated.rowCount) throw new TaskError('channel_not_found', '用户渠道不存在。', 404); await this.audit(client, context, 'agentbus_channel.key_rotated', channelId, {}); return updated.rows[0] as Record; }); return publicChannel(result, this.legacyEnvironmentManaged()); } async delete( context: TaskContext, channelId: string ): Promise<{ channel_id: string; deleted: true }> { return withTransaction(this.config, async (client) => { const lookup = await client.query( `SELECT id, display_name, external_user_ref, enabled FROM user_channels WHERE organization_id = $1 AND id = $2 FOR UPDATE`, [context.organizationId, channelId] ); if (!lookup.rowCount) throw new TaskError('channel_not_found', '用户渠道不存在。', 404); const row = lookup.rows[0] as Record; if (text(row.external_user_ref) === LEGACY_CHANNEL_REF && this.legacyEnvironmentManaged()) { throw new TaskError( 'channel_managed_by_environment', '该兼容渠道仍由旧 AgentBus 环境变量托管;请先移除环境配置并重启服务,再删除数据库渠道。', 409 ); } const related = await client.query( `SELECT (SELECT count(*)::int FROM tasks WHERE channel_id = $1) AS linked_task_count, (SELECT count(*)::int FROM agentbus_deliveries WHERE channel_id = $1) AS delivery_count`, [channelId] ); const relatedRow = related.rows[0] as Record; await this.audit(client, context, 'agentbus_channel.deleted', channelId, { display_name: text(row.display_name), enabled: booleanValue(row.enabled), linked_task_count: Number(relatedRow.linked_task_count || 0), removed_delivery_count: Number(relatedRow.delivery_count || 0) }); const deleted = await client.query( 'DELETE FROM user_channels WHERE organization_id = $1 AND id = $2 RETURNING id', [context.organizationId, channelId] ); if (!deleted.rowCount) throw new TaskError('channel_not_found', '用户渠道不存在。', 404); return { channel_id: text(deleted.rows[0].id), deleted: true }; }); } async ensureLegacyChannel(organizationId: string): Promise { const wsToken = text(this.config.AGENTBUS_WS_TOKEN); const botAddress = text(this.config.AGENTBUS_BOT_ADDRESS); if (!text(this.config.AGENTBUS_WS_URL) || !wsToken || !botAddress) return; await withTransaction(this.config, async (client) => { const existing = await client.query( `SELECT id FROM user_channels WHERE organization_id = $1 AND external_user_ref = '${LEGACY_CHANNEL_REF}' ORDER BY created_at ASC, id ASC FOR UPDATE`, [organizationId] ); const rows = existing.rows as Array>; const duplicateIds = rows.slice(1).map((row) => text(row.id)).filter(Boolean); if (duplicateIds.length) { await client.query( `UPDATE user_channels SET enabled = false, status = 'disabled', last_error = $1, updated_at = now() WHERE organization_id = $2 AND id = ANY($3::uuid[])`, [DUPLICATE_LEGACY_CHANNEL_ERROR, organizationId, duplicateIds] ); } if (rows.length) return; await client.query( `INSERT INTO user_channels (organization_id, display_name, external_user_ref, agentbus_ws_token_ciphertext, agentbus_bot_address, enabled, status) SELECT $1, '默认 AgentBus 渠道', '${LEGACY_CHANNEL_REF}', $2, $3, true, 'connecting' WHERE NOT EXISTS ( SELECT 1 FROM user_channels WHERE organization_id = $1 AND external_user_ref = '${LEGACY_CHANNEL_REF}' ) ON CONFLICT (organization_id, display_name) DO NOTHING`, [organizationId, encryptText(this.config, wsToken), botAddress] ); }); } async setRuntimeStatus( organizationId: string, channelId: string, status: PublicAgentBusChannel['status'], errorMessage: string | null = null, epoch: number | null = null ): Promise { const previous = this.runtimeStatusWrites.get(channelId) || Promise.resolve(); const next = previous .catch(() => undefined) .then(async () => { await getPool(this.config).query( `UPDATE user_channels SET status = $1, last_error = $2, last_session_epoch = COALESCE($3::bigint, last_session_epoch), last_connected_at = CASE WHEN $1 = 'connected' THEN now() ELSE last_connected_at END, updated_at = now() WHERE organization_id = $4 AND id = $5 AND ( $1 = 'error' OR ($3::bigint IS NOT NULL AND (last_session_epoch IS NULL OR last_session_epoch <= $3::bigint)) OR ($3::bigint IS NULL AND last_session_epoch IS NULL) )`, [status, errorMessage, epoch, organizationId, channelId] ); }); this.runtimeStatusWrites.set(channelId, next); try { await next; } finally { if (this.runtimeStatusWrites.get(channelId) === next) this.runtimeStatusWrites.delete(channelId); } } private async getRow(organizationId: string, channelId: string): Promise> { const result = await getPool(this.config).query( `SELECT id, organization_id, display_name, external_user_ref, agentbus_bot_address, enabled, status, last_connected_at, last_error, created_at, updated_at FROM user_channels WHERE organization_id = $1 AND id = $2`, [organizationId, channelId] ); if (!result.rowCount) throw new TaskError('channel_not_found', '用户渠道不存在。', 404); return result.rows[0] as Record; } private async audit( client: import('pg').PoolClient, context: TaskContext, eventType: string, entityId: string, metadata: Record ): Promise { await client.query( `INSERT INTO audit_events (organization_id, actor_user_id, event_type, entity_type, entity_id, request_id, metadata) VALUES ($1, $2, $3, 'user_channel', $4, $5, $6)`, [context.organizationId, context.userId || null, eventType, entityId, context.requestId, metadata] ); this.log('info', { agentbus_event: 'channel_audit_staged', request_id: context.requestId, domain_event: eventType, entity_type: 'user_channel', entity_id: entityId, actor_present: Boolean(context.userId), metadata_keys: diagnosticMetadataKeys(metadata) }, 'AgentBus channel audit event staged'); } } export interface AgentBusManagerOptions { config: AppConfig; tasks: AgentBusListenerOptions['tasks']; organizationId: string; scheduleParseQueue: () => Promise; logger?: AgentBusLogger; socketFactory?: AgentBusSocketFactory; } export class AgentBusManager { private readonly channels: AgentBusChannelService; private readonly listeners = new Map(); private readonly options: AgentBusManagerOptions; private started = false; private reloadInFlight: Promise | null = null; private reloadRequested = false; constructor(options: AgentBusManagerOptions) { this.options = options; this.channels = new AgentBusChannelService(options.config, options.logger); } private log( level: 'info' | 'warn' | 'error', metadata: Record, message: string ): void { try { this.options.logger?.[level](metadata, message); } catch { // Runtime channel management must not depend on log delivery. } } get channelService(): AgentBusChannelService { return this.channels; } async start(): Promise { if (!this.options.config.agentBusEnabled) return; if (this.started) return; this.started = true; this.log('info', { agentbus_event: 'manager_starting' }, 'AgentBus manager starting'); await this.reload(); } async reload(): Promise { if (!this.started) return; this.reloadRequested = true; this.log('info', { agentbus_event: 'manager_reload_requested', reload_in_flight: Boolean(this.reloadInFlight), current_listeners: this.listeners.size }, 'AgentBus manager reload requested'); if (this.reloadInFlight) return this.reloadInFlight; this.reloadInFlight = (async () => { while (this.started && this.reloadRequested) { this.reloadRequested = false; await this.channels.ensureLegacyChannel(this.options.organizationId); for (const listener of this.listeners.values()) listener.stop(false); this.listeners.clear(); const secrets = await this.channels.listEnabledSecrets(this.options.organizationId); this.log('info', { agentbus_event: 'manager_channels_loaded', enabled_channels: secrets.length }, 'AgentBus enabled channels loaded'); for (const channel of secrets) { const listener = new AgentBusListener({ config: this.options.config, tasks: this.options.tasks, organizationId: this.options.organizationId, scheduleParseQueue: this.options.scheduleParseQueue, socketFactory: this.options.socketFactory, logger: this.options.logger, channel: { id: channel.id, displayName: channel.display_name, wsUrl: channel.ws_url, wsToken: channel.ws_token, botAddress: channel.bot_address }, onStatusChange: (status, error, epoch) => this.channels.setRuntimeStatus( this.options.organizationId, channel.id, status, error, epoch ).catch((statusError) => { this.log('warn', { agentbus_event: 'channel_status_persist_failed', channel_id: channel.id, runtime_status: status, session_epoch: epoch, ...diagnosticError(statusError, 'agentbus_channel_status_persist_failed') }, 'AgentBus channel status persistence failed'); }) }); this.listeners.set(channel.id, listener); listener.start(); } this.log('info', { agentbus_event: 'manager_reload_completed', active_listeners: this.listeners.size, reload_requested_again: this.reloadRequested }, 'AgentBus manager reload completed'); } })() .catch((error) => { this.log('error', { agentbus_event: 'manager_reload_failed', ...diagnosticError(error, 'agentbus_manager_reload_failed') }, 'AgentBus manager reload failed'); throw error; }) .finally(() => { this.reloadInFlight = null; }); return this.reloadInFlight; } async stop(): Promise { this.started = false; this.reloadRequested = false; if (this.reloadInFlight) await this.reloadInFlight.catch(() => undefined); // Process shutdown is not an administrative channel disable. Do not let // an older process write `disabled` after a newer process has connected. for (const listener of this.listeners.values()) listener.stop(false); this.listeners.clear(); this.log('info', { agentbus_event: 'manager_stopped' }, 'AgentBus manager stopped'); } status(): { enabled: boolean; connected: boolean; session_ready: boolean; address: string | null; channels: Array>; } { const channels = [...this.listeners.entries()].map(([, listener]) => listener.status()); return { enabled: this.options.config.agentBusEnabled, connected: channels.some((channel) => channel.connected === true), session_ready: channels.some((channel) => channel.session_ready === true), address: (channels.find((channel) => channel.address)?.address as string | undefined) || null, channels }; } }