import { effect, inject, Injectable, OnDestroy, signal } from '@angular/core'; import type { MatrixClient, MatrixEvent, Room, RoomMember } from 'matrix-js-sdk'; import { SessionService } from './session.service'; /** * Matrix 客户端 —— `matrix-js-sdk` 的**唯一收敛点** * * 依据《数据访问层设计.md》§五(每域唯一 owner)与 SPEC FR-04。 * * ## NFR-02:SDK 不得进主包 * * `matrix-js-sdk` **只在首次真正需要时动态导入**(`await import()`), * 顶层仅 `import type`(编译期擦除,零运行时依赖)。 * 验证:主包 chunk 中不得出现 `matrix-js-sdk` 字样。 * * ## 身份规则(fmode-hub v4.2 P3 约定) * * - 用户 `mxid` = `@<用户 objectId 小写>:fmode.cn` * - Agent `mxid` = `@:fmode.cn` * - 凭据一律来自 provision 云函数(`RP5Ht55NeX`)**服务端签发**; * 本服务**不派生任何密码**(黄金规则 9:前端零密钥)。 * * ## 不包含的能力 * * 语音通话(`m.call.*` / WebRTC)按 SPEC FR-11 裁决**暂不实现**, * 故不引入 TURN 配置,也不依赖 `navigator.mediaDevices`。 */ /** 一条归一化后的文本消息(视图层只见此结构) */ export interface MxMessage { /** 所属房间(未读计数需要知道消息来自哪个 Agent) */ roomId: string; eventId: string; sender: string; body: string; ts: number; kind: 'text' | 'notice' | 'emote'; /** * 是否**实时到达**的消息(`data.liveEvent === true`)。 * * 🔴 `/sync` 初次同步会把**历史消息**逐条塞进 timeline,这些也会触发 * `Room.timeline` 事件。若不加区分,未读计数会把整个历史都算成未读 * (实测曾出现「刚连接就有 39 条未读」)。 * * 未读计数**只计 `live === true`** 的消息;timeline 渲染两者都要。 */ live: boolean; } /** 同步连接状态(驱动 ChatService 的 ConnectionState) */ export type MxSyncState = 'connecting' | 'connected' | 'disconnected'; function kindOf(msgtype: string | undefined): MxMessage['kind'] { if (msgtype === 'm.notice') return 'notice'; if (msgtype === 'm.emote') return 'emote'; return 'text'; } const TEXT_MSGTYPES = ['m.text', 'm.notice', 'm.emote']; @Injectable({ providedIn: 'root' }) export class MatrixClientService implements OnDestroy { /** 懒加载的 SDK 模块(首次 `load()` 时填充) */ private sdk: typeof import('matrix-js-sdk') | null = null; private client: MatrixClient | null = null; // ---------- 连接状态(供 chat 页徽标消费) ---------- private readonly _syncState = signal('disconnected'); readonly syncState = this._syncState.asReadonly(); /** * 本地 store 版本号(每次 sync 完成 +1)。 * 聊天页 `effect` 订阅它 → 重读 timeline,解决「同步晚于首读 → 历史空白」。 */ private readonly _storeVersion = signal(0); readonly storeVersion = this._storeVersion.asReadonly(); private readonly _userId = signal(''); /** 当前登录的 Matrix 用户 id(`@xxx:fmode.cn`) */ readonly userId = this._userId.asReadonly(); private token = ''; private deviceId = ''; /** 显式停止标志:避免 disconnect() 后 sync 回调再改状态 */ private stopped = false; /** * 初次同步是否已完成。 * * 🔴 未读计数的**分界线**:初次 `/sync` 会把房间历史逐条回填(且按时间 * 升序),若直接计数会把整个历史都算成未读。故此标志为 false 期间, * 未读只推进基线、不计数。 */ private initialSyncDone = false; /** 初次同步是否完成(未读计数用于区分历史与实时) */ get hasInitialSync(): boolean { return this.initialSyncDone; } /** * 全局消息流 —— **全项目唯一一处** `Room.timeline` 订阅。 * * 多路复用给:① 聊天页(当前房间)② 未读计数(所有房间)。 * 这样 sync 常驻时也只有一个监听器,不会重复触发。 */ private readonly messageHandlers = new Set<(m: MxMessage) => void>(); private globalBound = false; private readonly session = inject(SessionService); constructor() { // 🔴 登出即断连:不依赖页面记得调用 disconnect()。 // 任一页面的 doLogout() 清掉 token 后,这里自动收敛。 effect(() => { if (!this.session.token() && this.client) this.disconnect(); }); } ngOnDestroy(): void { this.disconnect(); } // ---------- 生命周期(会话级常驻) ---------- /** * 建立**会话级**连接(整个登录会话只调用一次,由 SessionService 持有者驱动)。 * * 🔴 与旧实现的关键差异:这是**常驻**的 —— 不在聊天页 `close()` 时断开, * 否则离开会话后收不到消息,未读计数永远无法累加。 * 断开只发生在**退出登录**时(`disconnect()`)。 * * @param accessToken 云函数返回的 token(**不落 localStorage**,仅存内存) * @param userId 形如 `@sr2wipsdyq:fmode.cn` * @param deviceId 云函数返回的 deviceId(可空) */ async connectSession(accessToken: string, userId: string, deviceId?: string): Promise { if (!accessToken || !userId) { throw new Error('Matrix 凭据不完整(缺 accessToken 或 userId)'); } // 同身份重复调用:幂等,不打断现有 sync if (this.client && this.token === accessToken && this._userId() === userId) { return; } this.disconnect(); const sdk = await this.load(); this.token = accessToken; this.deviceId = deviceId ?? ''; this._userId.set(userId); this.stopped = false; this.initialSyncDone = false; this._syncState.set('connecting'); this.client = sdk.createClient({ baseUrl: MATRIX_HS, accessToken, userId, ...(deviceId ? { deviceId } : {}), }); // sync 状态 → 连接状态(FR-04 AC-04.9:断线可自动重连) this.client.on(sdk.ClientEvent.Sync, (state: string) => { if (this.stopped) return; if (state === 'SYNCING' || state === 'PREPARED') this._syncState.set('connected'); else if (state === 'ERROR') this._syncState.set('disconnected'); else if (state === 'RECONNECTING') this._syncState.set('connecting'); // 🔴 首个 SYNCING = 初次同步完成 → 之后的消息才算「实时」 if (state === 'SYNCING') { this.initialSyncDone = true; /* * 🔴 通知「timeline 已就绪」。 * * `startClient()` 是异步的:`openReal()` 里紧接着读 `timelineToMessages()` * 时,房间的历史事件**往往还没落到本地 store** → 拿到空数组; * 而后续只依赖 realtime 事件补消息,历史内容永远补不上 * (表现为:刷新后会话空白、重开看不到之前的内容)。 * * 每次 SYNCING 递增版本号,聊天页据此重读一次 timeline。 */ this._storeVersion.update((n) => n + 1); } }); // 收到邀请自动接受(Agent 侧建房的场景) this.client.on(sdk.RoomMemberEvent.Membership, (_ev: MatrixEvent, member: RoomMember) => { if (this.stopped) return; if (member.membership === 'invite' && member.userId === userId) { void this.client?.joinRoom(member.roomId).catch(() => undefined); } }); // 全局消息流(未读计数 + 聊天页复用) this.bindGlobalMessageStream(); this.client.startClient({ initialSyncLimit: 50 }); } /** 断开并清理(**仅退出登录时**调用) */ disconnect(): void { this.stopped = true; this.initialSyncDone = false; this.messageHandlers.clear(); this.globalBound = false; if (this.client) { try { this.client.stopClient(); } catch { /* 忽略:销毁期异常不阻塞清理 */ } try { this.client.removeAllListeners(); } catch { /* 忽略 */ } } this.client = null; this.token = ''; this.deviceId = ''; this._userId.set(''); this._syncState.set('disconnected'); } /** 是否已建立会话级连接 */ get connected(): boolean { return !!this.client; } // ---------- 房间 ---------- /** * 确保与目标 Agent 的 DM 房间可用。 * * 优先用 provision 返回的 `roomId`(绑定表真相);未提供时按 * `m.direct` accountData 查找,仍无则创建。 */ async ensureRoom( agentMxId: string, opts?: { roomId?: string | null; agentName?: string }, ): Promise { const c = this.requireClient(); // ① 绑定表给的 roomId 优先(与容器侧同一房间) if (opts?.roomId) { const room = c.getRoom(opts.roomId); if (room && room.getMyMembership() === 'join') return opts.roomId; // 本地尚未 join → 尝试加入(可能是别的设备建的房) const joined = await c.joinRoom(opts.roomId).catch(() => null); if (joined) return opts.roomId; } // ② m.direct 里已有记录 const directEv = c.getAccountData('m.direct' as never); const direct = directEv?.getContent() as Record | undefined; for (const rid of direct?.[agentMxId] ?? []) { const room = c.getRoom(rid); if (room?.getMyMembership() === 'join') return rid; } // ③ 兜底:遍历已 join 房间找共同成员 for (const room of c.getRooms()) { if (room.getMyMembership() !== 'join') continue; if (room.getMember(agentMxId)) return room.roomId; } // ④ 都没有 → 新建 DM const sdk = await this.load(); const created = await c.createRoom({ invite: [agentMxId], is_direct: true, preset: sdk.Preset.TrustedPrivateChat, name: opts?.agentName, }); return created.room_id; } /** 该房间是否已是「已加入」状态 */ isJoined(roomId: string): boolean { return !!this.client?.getRoom(roomId)?.getMyMembership?.(); } // ---------- 消息收发 ---------- /** 发文本(AC-04.2) */ async sendText(roomId: string, text: string): Promise<{ event_id?: string }> { const c = this.requireClient(); const sdk = await this.load(); return c.sendEvent(roomId, sdk.EventType.RoomMessage, { msgtype: sdk.MsgType.Text, body: text, }); } /** 发文件卡片(`m.text` + JSON body,与容器侧解析约定一致) */ async sendFileCard( roomId: string, card: { name: string; size: number; mimeType: string; url: string }, ): Promise<{ event_id?: string }> { const c = this.requireClient(); const sdk = await this.load(); return c.sendEvent(roomId, sdk.EventType.RoomMessage, { msgtype: sdk.MsgType.Text, body: JSON.stringify({ kind: 'file', ...card }), }); } /** 当前 timeline → 归一消息(时间正序) */ timelineToMessages(roomId: string): MxMessage[] { const room = this.client?.getRoom(roomId); if (!room) return []; return room .getLiveTimeline() .getEvents() .filter((e) => e.getType() === 'm.room.message') .filter((e) => TEXT_MSGTYPES.includes(String(e.getContent()['msgtype'] ?? ''))) .map((e) => ({ roomId, eventId: e.getId() ?? '', sender: e.getSender() ?? '', body: (e.getContent()['body'] as string) ?? '', ts: e.getTs(), kind: kindOf(String(e.getContent()['msgtype'] ?? '')), // 从 timeline 读出的都视为历史(未读不依赖此值) live: false, })); } /** 向上翻历史(AC-04.4) */ async paginateBack(roomId: string, limit = 30): Promise { const c = this.requireClient(); let room = c.getRoom(roomId); if (!room) { await c.joinRoom(roomId).catch(() => null); room = c.getRoom(roomId); } if (!room) return false; return c.paginateEventTimeline(room.getLiveTimeline(), { backwards: true, limit }); } /** * 是否还有更早历史(live timeline 的 pagination token 是否耗尽)。 * 用于 UI 决定要不要显示「加载更早」入口,以及分页到顶后停止请求。 */ paginationToken(roomId: string): string | null { const room = this.client?.getRoom(roomId); if (!room || !this.sdk) return null; return room.getLiveTimeline().getPaginationToken(this.sdk.Direction.Backward) ?? null; } /** 已读回执 */ async sendReceipt(roomId: string, eventId: string): Promise { const c = this.requireClient(); const ev = c .getRoom(roomId) ?.getLiveTimeline() .getEvents() .find((e) => e.getId() === eventId); if (!ev) return; await c.sendReadReceipt(ev); } /** 输入中通知(失败非致命) */ async setTyping(roomId: string, typing: boolean, timeoutMs = 8000): Promise { try { await this.requireClient().sendTyping(roomId, typing, timeoutMs); } catch { /* 非致命:typing 丢失不影响消息链路 */ } } // ---------- 事件订阅 ---------- /** * 订阅房间新消息;返回解绑函数。 * * 不过滤 sender —— 自己发的消息也要接住(local echo 依赖 Timeline 事件)。 */ onMessage(roomId: string, cb: (m: MxMessage) => void): () => void { return this.onAnyMessage((m) => { if (m.roomId === roomId) cb(m); }); } /** * 订阅**所有房间**的新消息(未读计数用);返回解绑函数。 * * 与 `onMessage` 共用同一个底层订阅(`bindGlobalMessageStream`), * 因此无论多少订阅者,SDK 侧只有**一个** `Room.timeline` 监听器。 */ onAnyMessage(cb: (m: MxMessage) => void): () => void { this.messageHandlers.add(cb); return () => { this.messageHandlers.delete(cb); }; } /** 订阅对方 typing 状态;返回解绑函数 */ onTyping(roomId: string, cb: (typing: boolean, userId: string) => void): () => void { const c = this.requireClient(); const handler = (ev: MatrixEvent, member: RoomMember) => { if (member?.roomId !== roomId || ev.getType() !== 'm.typing') return; cb(!!member.typing, member.userId ?? ''); }; c.on('RoomMember.typing' as never, handler as never); return () => { try { c.off('RoomMember.typing' as never, handler as never); } catch { /* 忽略 */ } }; } /** 绑定全局消息流(只绑一次,幂等) */ private bindGlobalMessageStream(): void { const c = this.client; if (!c || this.globalBound) return; this.globalBound = true; c.on('Room.timeline' as never, ((ev: MatrixEvent, _room?: unknown, _toStart?: boolean, data?: { liveEvent?: boolean }) => { if (ev.getType() !== 'm.room.message') return; // 跳过本地未发送回执(echo 会再触发一次) const sending = (ev as unknown as { isSending?: () => boolean }).isSending; if (typeof sending === 'function' && sending.call(ev)) return; const msgtype = String(ev.getContent()['msgtype'] ?? ''); if (!TEXT_MSGTYPES.includes(msgtype)) return; const msg: MxMessage = { roomId: ev.getRoomId() ?? '', eventId: ev.getId() ?? '', sender: ev.getSender() ?? '', body: (ev.getContent()['body'] as string) ?? '', ts: ev.getTs(), kind: kindOf(msgtype), // 仅真正的实时事件为 true;初次 /sync 的历史回填为 false live: !!data?.liveEvent, }; for (const h of this.messageHandlers) { try { h(msg); } catch { /* 单个订阅者异常不影响其它订阅者 */ } } }) as never); } // ---------- 工具 ---------- /** `@:fmode.cn`(v4.2 P3 约定) */ static agentMxId(objectId: string): string { return `@${objectId.toLowerCase()}:${MATRIX_SERVER_NAME}`; } /** 解析文件卡片 body;非卡片返回 null */ static parseFileCard( body: string, ): { name: string; size: number; mimeType: string; url: string } | null { if (!body || body[0] !== '{') return null; try { const j = JSON.parse(body); if ((j.kind === 'file' || j.type === 'fcard') && j.url && j.name) { return { name: String(j.name), size: Number(j.size) || 0, mimeType: String(j.mimeType ?? j.mime ?? ''), url: String(j.url), }; } return null; } catch { return null; } } private requireClient(): MatrixClient { if (!this.client) throw new Error('Matrix 客户端未连接'); return this.client; } /** 动态导入 SDK(NFR-02:不进主包) */ private async load(): Promise { if (!this.sdk) { this.sdk = await import('matrix-js-sdk'); /* * 🔴 SDK 日志降噪(否则控制台卡顿)。 * * matrix-js-sdk 的 logger 默认输出 **warn 及以上**,且每条都带完整调用栈 * (push rule 缺失的告警会在每次 startClient 时批量刷屏, * 叠加 zone.js 的调度栈,一次上百行)。 * 这些告警不影响收发:push rule 是服务端推送规则,浏览器端本就不用。 * 故收敛为只保留 error;取不到 logger 时静默跳过,绝不影响连接。 */ const sdkAny = this.sdk as { logger?: { setLevel?: (l: string) => void } }; if (sdkAny.logger && typeof sdkAny.logger.setLevel === 'function') { sdkAny.logger.setLevel('error'); } } return this.sdk; } } /** homeserver 常量(与《API-契约》§1.2 一致) */ export const MATRIX_HS = 'https://matrix.fmode.cn'; export const MATRIX_SERVER_NAME = 'fmode.cn';