import { computed, inject, Injectable, signal } from '@angular/core'; import { Message } from './api/models'; import { GroupEntry } from './group-turns'; import { turnIdAt } from './group-matrix-transcript'; import { SessionService } from './session.service'; import { ChatSessionService } from './chat-session.service'; import { isUserUploadUrl } from './s3-link'; /** * 群聊消息存储(2026-11-05 新增) * * ## 为什么需要独立存储 * * 伪群聊的回复物理上落在**各容器的 DM 房间**里 —— 而那个房间里还装着 * 用户与该容器的**全部单聊历史**。直接回读房间会让私聊内容串进群聊页。 * * 实测(2026-11-05):刚进群聊页就能看到容器几天前的单聊长回复, * 还被合并成一个「已完成 · 189848s」的回合。 * * 所以:群聊有自己的消息表,**只收群内发生的消息**,与 DM 房间彻底隔离。 * * ## 什么能进群聊 * * | 来源 | 能否进群 | * |---|---| * | 用户在群里发的消息 | ✅ | * | **订阅建立之后**实时到达的容器回复 | ✅ | * | `/sync` 初次回填的历史 | ❌ 那是单聊历史 | * | 向上翻页拿回的旧事件 | ❌ 同上报废(`live` 标志区分不了,见下) | * * 🔴 `live === true` 并不等于「刚到的」:matrix-js-sdk 的 * `liveEvent: !toStartOfTimeline && ...`,向上翻页同样是 true。 * 所以还要叠加**时间水位**:`ts > watermark` 才算新回复。 * * ## 持久化 * * 本地 localStorage,按用户隔离(与 `ChatSessionService` 同模式)。 * 不上云:容器房间里有原件,云端再存一份会产生两份真相(见 ADR-007)。 */ /** 单会话最多保留的条目数,超出按时间裁剪 */ export const MAX_ENTRIES_PER_GROUP = 500; /** localStorage 键前缀(与 chat-session 的 `fmode_chat_sessions_v2_` 同风格) */ const STORAGE_PREFIX = 'fmode_group_entries_v1_'; /** 一条用户群内消息的落库形态 */ interface StoredEntry extends GroupEntry { /** 写入时的会话 id(冗余一份,便于按 key 存) */ sessionId: string; } @Injectable({ providedIn: 'root' }) export class GroupStoreService { private readonly session = inject(SessionService); /** 群会话的成果链接写入(2026-11-10,与单聊同口径) */ private readonly sessions = inject(ChatSessionService); /** sessionId → 全部条目(按 at 升序) */ private readonly _bySession = signal>({}); /** 自增序号:保证同一毫秒内的用户消息也有稳定 id */ private seq = 0; /** * 是否已成功从磁盘载入过。 * * 🔴 只有载入过才允许写盘:否则「未登录 → 写空表」会抹掉已有数据。 */ private loaded = false; /** ⚠️ 测试用:重置内存态(同时复位「已加载」标记) */ resetForTest(): void { this._bySession.set({}); this._targetsOf.set({}); this.seq = 0; this.loaded = false; } /** 某会话的全部条目 */ entriesOf(sessionId: string): GroupEntry[] { return this._bySession()[sessionId] ?? []; } /** * 该会话是否已有记录。 * * 页面用它区分「真的没聊过」与「还没加载」—— * 两者在 UI 上都显示空态,但语义不同。 */ hasAny = computed(() => Object.keys(this._bySession()).length > 0); // ---------- 写入 ---------- /** * 记录一条用户群内消息。 * * @returns 这条消息的回合 id(= 条目 id),后续容器回复都挂在它下面 */ appendMine( sessionId: string, text: string, targets: string[], at = Date.now(), ): string { /* * 🔴 id 必须**可复现**(2026-11-07 群聊落 Matrix):`me-<毫秒>`。 * * 原实现 `me-<毫秒>-<内存自增序号>`:序号每次启动从 0 开始, * 换设备后同一条消息算出的 id 不同 —— 从 Matrix 重建出来的条目 * 与本地条目对不上,合并会变成两条一模一样的气泡。 */ this.seq += 1; const id = turnIdAt(at, new Set(this.entriesOf(sessionId).map((e) => e.id))); const entry: GroupEntry = { id, turnId: id, fromAgentId: null, at, message: { id, conversationId: 'group', sender: 'me', kind: 'text', text, ts: new Date(at).toISOString(), state: 'sent', }, }; // 目标成员记在这里,供回合聚合渲染「已派给谁」 this._targetsOf.update((m) => ({ ...m, [id]: targets })); this.write(sessionId, [entry]); return id; } /** * 记录一条容器回复。 * * @param turnId 属于哪次提问。未知时归入该会话**最近一次**提问。 * @returns 实际写入的 turnId;找不到任何回合时返回 null(丢弃) */ appendReply( sessionId: string, agentId: string, msg: Message, at: number, turnId?: string, ): string | null { const list = this.entriesOf(sessionId); // 按 eventId 去重 —— 同一条 Matrix 事件只入群一次 if (msg.id && list.some((e) => e.id === msg.id)) return null; const resolved = turnId ?? [...list].reverse().find((e) => e.fromAgentId === null)?.turnId ?? null; if (!resolved) return null; const entry: GroupEntry = { id: msg.id, turnId: resolved, fromAgentId: agentId, message: msg, at }; this.write(sessionId, [entry]); /* * 记录成果链接到**群会话**(2026-11-10)。 * * 原来群会话的 `deliverableUrls` 从来没人写,右栏「成果」只能退化成 * 「按成员容器拉全量」—— 成员在别的单聊里产出的文件也会混进来。 * 现在与单聊同口径:只认 agent 产出、且不是用户上传的文件。 */ const url = msg.file?.cloudUrl || (msg.file?.url?.startsWith('https://s3.') ? msg.file.url : undefined); if (url && !isUserUploadUrl(url)) this.sessions.addDeliverableUrl(sessionId, url); return resolved; } /** * 记录一次「用户在本回合发的附件」(2026-11-07)。 * * 群聊发文件时,每个目标房间各有一条文件事件;这里只记**首条成功的那条** * (按 name+size 去重),用户气泡上显示一份即可。 * * @param eventId Matrix 文件事件 id;没有时回落到 turnId */ appendUserFile( sessionId: string, turnId: string, file: Message['file'], at: number, eventId?: string, ): void { if (!file) return; const list = this.entriesOf(sessionId); const dup = list.some( (e) => e.message.file?.name === file.name && e.message.file?.size === file.size, ); if (dup) return; const entry: GroupEntry = { id: eventId || `${turnId}#file-${at}`, turnId, fromAgentId: null, at, message: { id: eventId || turnId, conversationId: 'group', sender: 'me', kind: 'text', file, ts: new Date(at).toISOString(), state: 'sent', }, }; this.write(sessionId, [entry]); } /** * 用 Matrix 重建结果合并进某群(2026-11-07,2026-11-08 改为**保守合并**)。 * * 调用方已完成「本地优先」合并;这里只负责落表 + 并集 targets + 落盘。 * * 🔴 原来是**整表替换**:`[sessionId]: [...entries]`。 * Matrix 重建时如果少推导出条目(分页没翻到、`isNewReply` 过滤、 * 房间一时读不到),本地已有的内容就被直接抹掉 —— * 用户看到的是「后续回复把之前的输出覆盖了」。 * 现在改为:Matrix 结果为准,但**Matrix 没有的本地条目保留**。 */ replaceFromMatrix( sessionId: string, entries: GroupEntry[], targets: Record, ): void { this._bySession.update((all) => { const local = all[sessionId] ?? []; const seen = new Set(entries.map((e) => e.id)); // 本地有、Matrix 没有的条目:保留(不因为重建不全就丢内容) const kept = local.filter((e) => e.id && !seen.has(e.id)); return { ...all, [sessionId]: [...entries, ...kept].sort((a, b) => a.at - b.at), }; }); this._targetsOf.update((m) => { const next = { ...m }; for (const [turnId, list] of Object.entries(targets)) { const cur = next[turnId] ?? []; next[turnId] = [...cur, ...list.filter((x) => !cur.includes(x))]; } return next; }); this.persist(); } /** 清空某群的消息(用户手动清除 / 群解散) */ clear(sessionId: string): void { this._bySession.update((all) => { const next = { ...all }; delete next[sessionId]; return next; }); this._targetsOf.update((m) => { const next = { ...m }; for (const k of Object.keys(next)) if (next[k]?.length === 0) delete next[k]; return next; }); this.persist(); } // ---------- 目标成员(回合聚合需要) ---------- /** * 已消费的事件 id(`sessionId|agentId` → Set)。 * * 实时流与补读会并发收到同一条事件;`appendReply` 已按 eventId 去重, * 但补读还要据此跳过已处理的条目,避免每轮重复扫描。 */ private readonly _consumed = new Map>(); private consumedKey(sessionId: string, agentId: string): string { return `${sessionId}|${agentId}`; } /** 该事件是否已被本轮消费 */ isConsumed(sessionId: string, agentId: string, eventId: string): boolean { return this._consumed.get(this.consumedKey(sessionId, agentId))?.has(eventId) ?? false; } /** 标记已消费(内存态即可:补读最长 135 秒,刷新后从头重建索引不丢内容) */ markConsumed(sessionId: string, agentId: string, eventId: string): void { const key = this.consumedKey(sessionId, agentId); let set = this._consumed.get(key); if (!set) { set = new Set(); this._consumed.set(key, set); } set.add(eventId); } private readonly _targetsOf = signal>({}); /** 某次提问派给了哪些成员 */ targetsOf(turnId: string): string[] { return this._targetsOf()[turnId] ?? []; } // ---------- 内部 ---------- private write(sessionId: string, entries: GroupEntry[]): void { this._bySession.update((all) => { const cur = all[sessionId] ?? []; /* * 按 id 去重(2026-11-07):Matrix 重建与实时流可能写同一条事件, * 原来的纯追加会产生重复气泡。 */ const seen = new Set(cur.map((e) => e.id)); const merged = [...cur, ...entries.filter((e) => !seen.has(e.id))].sort( (a, b) => a.at - b.at, ); /* * 🔴 不再按回合裁剪内存表:Matrix 才是历史的真源,本地裁掉等于 * 「刷新后重建出来又被裁回去」。只在写盘时裁(见 persist)。 */ return { ...all, [sessionId]: merged }; }); this.persist(); } private key(): string | null { const uid = this.session.me()?.id; return uid ? `${STORAGE_PREFIX}${uid}` : null; } private persist(): void { /* * 🔴 未加载完成时**绝不写盘**(2026-11-05 评审修复)。 * * 时序陷阱:深链接进入群聊页时,`load()` 可能在登录态就绪前就被调用, * 此时 `key()` 为 null 直接返回;若之后 `persist()` 照常写, * 会把「空表」盖到已有数据上 —— 表现为刷新后群内历史全丢。 * 所以:没成功加载过,就不允许覆盖。 */ if (!this.loaded) return; const key = this.key(); // 未登录不落盘 —— 否则下一个账号会读到上一个账号的群聊 if (!key) return; try { const payload: Record = {}; for (const [sid, list] of Object.entries(this._bySession())) { // 只裁**落盘副本**:内存表保持完整,会话内翻看不受影响 payload[sid] = trimByTurn(list, MAX_ENTRIES_PER_GROUP).map((e) => ({ ...e, sessionId: sid, /* * 🔴 本地预览地址(`blob:`)不能落盘(评审 2026-11-07): * 刷新后 blob 已失效,气泡会显示一张裂图 / 空卡。 * 落盘时去掉,刷新后由 Matrix 重建补上真实地址。 */ message: e.message.file?.url?.startsWith('blob:') ? { ...e.message, file: { ...e.message.file, url: '' } } : e.message, })); } localStorage.setItem( key, JSON.stringify({ entries: payload, targets: this._targetsOf() }), ); } catch { /* 隐私模式 / 超限:只保留内存态 */ } } /** 从 localStorage 载入(幂等;未登录时不置 `loaded`,等登录后再调) */ load(): void { const key = this.key(); if (!key) return; try { const raw = localStorage.getItem(key); // 没有历史数据也是「已载入」—— 允许后续正常写入新数据 if (!raw) { this.loaded = true; return; } const parsed = JSON.parse(raw) as { entries?: Record; targets?: Record; }; const bySession: Record = {}; for (const [sid, list] of Object.entries(parsed.entries ?? {})) { if (!Array.isArray(list)) continue; bySession[sid] = list .filter((e) => e && typeof e.at === 'number' && e.message && e.turnId) .map(({ sessionId: _sid, ...rest }) => rest) .sort((a, b) => a.at - b.at); } this._bySession.set(bySession); this._targetsOf.set(parsed.targets ?? {}); this.loaded = true; } catch { // 坏数据当没有 —— 不能让一条脏 JSON 把群聊页整个卡死 this._bySession.set({}); this._targetsOf.set({}); this.loaded = true; } } } /** * 按**回合**整组裁剪,保留最近的若干个回合。 * * 🔴 不能按条数切(2026-11-05 评审修复): * 按时间切片会把某个回合的用户消息先裁掉、只留下它的回复, * 变成「孤儿回复」——`buildGroupTurns` 会丢弃它,表现为对话凭空断掉。 * 整组裁掉才是干净的状态。 */ export function trimByTurn(list: GroupEntry[], max: number): GroupEntry[] { if (list.length <= max) return list; // 按 turnId 分组(保持首次出现顺序) const groups = new Map(); for (const e of list) { const g = groups.get(e.turnId); if (g) g.push(e); else groups.set(e.turnId, [e]); } const keys = [...groups.keys()]; const kept: GroupEntry[] = []; let count = 0; // 从最新回合往前留,直到超过上限为止 for (let i = keys.length - 1; i >= 0; i--) { const g = groups.get(keys[i])!; if (count + g.length > max && kept.length > 0) break; kept.unshift(...g); count += g.length; } return kept; } /** * 判断一条实时事件是否算「群内新回复」的纯函数部分。 * * `live` 单独不可靠(翻页拿到的事件 live 也是 true), * 必须叠加时间水位:只有不早于订阅建立时该房间最后一条事件的时间才算新。 * * 🔴 边界用 `>=` 而不是 `>`(2026-11-05 评审修复): * 水位 = 订阅时房间最后一条的 ts。若容器恰好在水位那一毫秒回了消息, * 用 `>` 会把它判成历史丢掉 → 状态卡在「思考中」直到 90 秒超时。 * 放宽到 `>=` 后可能重复收到同一条,但 `appendReply` 按 eventId 去重兜底。 * * @param live Matrix 事件的 `live` 标志 * @param ts 事件时间(ms) * @param watermark 订阅建立时该房间的时间水位(ms);0 表示没有水位约束 */ export function isFreshReply(live: boolean, ts: number, watermark: number): boolean { if (!live) return false; if (watermark <= 0) return true; return ts >= watermark; }