fmode-app/src/app/core/group-store.service.ts

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<Record<string, GroupEntry[]>>({});

  /** 自增序号：保证同一毫秒内的用户消息也有稳定 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<string, string[]>,
  ): 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<eventId>）。
   *
   * 实时流与补读会并发收到同一条事件；`appendReply` 已按 eventId 去重，
   * 但补读还要据此跳过已处理的条目，避免每轮重复扫描。
   */
  private readonly _consumed = new Map<string, Set<string>>();

  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<Record<string, string[]>>({});

  /** 某次提问派给了哪些成员 */
  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<string, StoredEntry[]> = {};
      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<string, StoredEntry[]>;
        targets?: Record<string, string[]>;
      };
      const bySession: Record<string, GroupEntry[]> = {};
      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<string, GroupEntry[]>();
  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;
}