《互联网产品的客户端架构·内容、社交与实时互动》10.2 多端已读：一致性从哪里来

// file: src/notify/read-state.ts
// tier: L1
// verify: node --test test/notify/read-state.test.ts
export interface ReadCursor {
  conversationId: string;
  position: number;
}

export interface ServerReadState {
  conversationId: string;
  total: number;
  readPosition: number;
}

export interface Receipt {
  conversationId: string;
  position: number;
  attempts: number;
}

export interface FlushReport {
  sent: number;
  failed: number;
  remaining: number;
}

export type ReceiptSender = (cursor: ReadCursor) => Promise<void>;

// 已读的事实是「读到了哪个位置」，不是「是否已读」。
export function mergeCursor(a: ReadCursor, b: ReadCursor): ReadCursor {
  assertCursor(a);
  assertCursor(b);
  if (a.conversationId !== b.conversationId) throw new Error("不能合并不同会话的已读位置");
  return a.position >= b.position ? { ...a } : { ...b };
}

// 服务端总数是权威值，但本地已经乐观读掉的部分要先扣掉，最后夹到 0。
export function effectiveUnread(state: ServerReadState, localPosition: number): number {
  assertServerState(state);
  if (!Number.isInteger(localPosition) || localPosition < 0) {
    throw new Error("localPosition 必须是非负整数");
  }
  const ahead = Math.max(0, localPosition - state.readPosition);
  return Math.max(0, state.total - ahead);
}

export class ReadStateStore {
  #positions = new Map<string, number>();
  #pending = new Map<string, { position: number; attempts: number }>();
  #flushing = false;

  positionOf(conversationId: string): number {
    return this.#positions.get(conversationId) ?? 0;
  }

  // 本地已读只前进不后退；返回 false 表示这次操作没有改变任何东西。
  markRead(cursor: ReadCursor): boolean {
    assertCursor(cursor);
    const current = this.positionOf(cursor.conversationId);
    if (cursor.position <= current) return false;
    this.#positions.set(cursor.conversationId, cursor.position);
    const pending = this.#pending.get(cursor.conversationId);
    if (!pending || cursor.position > pending.position) {
      this.#pending.set(cursor.conversationId, {
        position: cursor.position,
        attempts: pending?.attempts ?? 0,
      });
    }
    return true;
  }

  // 另一台设备的已读位置只用来前进本地位置，不需要再作为回执发回服务端。
  mergeRemote(cursor: ReadCursor): boolean {
    assertCursor(cursor);
    const current = this.positionOf(cursor.conversationId);
    if (cursor.position <= current) return false;
    this.#positions.set(cursor.conversationId, cursor.position);
    return true;
  }

  pendingReceipts(): Receipt[] {
    return [...this.#pending].map(([conversationId, entry]) => ({
      conversationId,
      position: entry.position,
      attempts: entry.attempts,
    }));
  }

  // 每个会话只保留最新位置。发送失败累加 attempts，成功则清掉已被覆盖的条目。
  async flush(send: ReceiptSender): Promise<FlushReport> {
    if (this.#flushing) throw new Error("已有回执发送在进行中");
    this.#flushing = true;
    let sent = 0;
    let failed = 0;
    try {
      for (const [conversationId, entry] of [...this.#pending]) {
        const target = entry.position;
        try {
          await send({ conversationId, position: target });
          sent += 1;
          const current = this.#pending.get(conversationId);
          // 发送期间用户可能又往前读了；只有没被超越时才清掉。
          if (current && current.position <= target) this.#pending.delete(conversationId);
        } catch {
          failed += 1;
          const current = this.#pending.get(conversationId);
          if (current) {
            this.#pending.set(conversationId, {
              position: current.position,
              attempts: current.attempts + 1,
            });
          }
        }
      }
      return { sent, failed, remaining: this.#pending.size };
    } finally {
      this.#flushing = false;
    }
  }

  // 服务端同步回来的已读位置与总数：位置取最大，回执被确认后从队列移除。
  reconcile(state: ServerReadState): { unread: number; readPosition: number } {
    assertServerState(state);
    const current = this.positionOf(state.conversationId);
    const position = Math.max(current, state.readPosition);
    if (position > current) this.#positions.set(state.conversationId, position);
    const pending = this.#pending.get(state.conversationId);
    if (pending && pending.position <= state.readPosition) {
      this.#pending.delete(state.conversationId);
    }
    return { unread: effectiveUnread(state, position), readPosition: position };
  }
}

function assertCursor(cursor: ReadCursor): void {
  if (typeof cursor.conversationId !== "string" || cursor.conversationId.length === 0) {
    throw new Error("conversationId 不能为空");
  }
  if (!Number.isInteger(cursor.position) || cursor.position < 0) {
    throw new Error("position 必须是非负整数");
  }
}

function assertServerState(state: ServerReadState): void {
  if (typeof state.conversationId !== "string" || state.conversationId.length === 0) {
    throw new Error("conversationId 不能为空");
  }
  if (!Number.isInteger(state.total) || state.total < 0) throw new Error("total 必须是非负整数");
  if (!Number.isInteger(state.readPosition) || state.readPosition < 0) {
    throw new Error("readPosition 必须是非负整数");
  }
}

$ node --test test/notify/read-state.test.ts
✔ 本地已读位置只前进，旧位置不产生新回执 (0.933416ms)
✔ 两端同时已读取最大位置 (0.472833ms)
✔ 同一会话只保留最新回执，尝试次数不丢 (0.237083ms)
✔ 回执失败后保留，重试成功后清理 (0.135417ms)
✔ 发送期间前进的位置不会被旧回执清掉 (0.136834ms)
✔ 服务端总数扣掉本地乐观已读并夹到 0 (0.127416ms)
✔ 服务端已确认的回执从队列移除，同步位置不回退 (0.075834ms)
ℹ tests 7
ℹ suites 0
ℹ pass 7
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 118.070458