《互联网产品的客户端架构·内容、社交与实时互动》10.1 未读数的来源与端上合并

// 说明性代码：不对应实验文件
let total = 0;

function onRealtimeEvent(): void {
  total += 1;
  renderBadge(total);
}

function onSnapshot(snapshot: { total: number }): void {
  total = snapshot.total;
  renderBadge(total);
}

// file: src/notify/unread.ts
// tier: L1
// verify: node --test test/notify/unread.test.ts
export type NotifyKind = "like" | "comment" | "follow" | "system";

export const NOTIFY_KINDS: readonly NotifyKind[] = ["like", "comment", "follow", "system"];

export interface UnreadSnapshot {
  counts: Partial<Record<NotifyKind, number>>;
  ackSeq: number;
}

export interface UnreadEvent {
  id: string;
  kind: NotifyKind;
  seq: number;
}

export interface UnreadBadge {
  total: number;
  byKind: Record<NotifyKind, number>;
  overflow: boolean;
  text: string;
}

export type EventStatus = "applied" | "duplicate" | "covered";

export interface ApplyEventResult {
  status: EventStatus;
}

export interface SnapshotResult {
  applied: boolean;
  stale: boolean;
  dropped: number;
}

// 未读数 = 服务端快照 + 快照水位之后端上收到、尚未被服务端确认的事件。
// ackSeq 是「服务端计数已经包含到这里」的水位，不是最后收到的 seq。
export class UnreadMerger {
  #base: Record<NotifyKind, number>;
  #ackSeq = 0;
  #pending = new Map<string, UnreadEvent>();
  #seen = new Set<string>();
  #seenOrder: string[] = [];
  #seenLimit: number;
  #badgeLimit: number;

  constructor(options: { seenLimit?: number; badgeLimit?: number } = {}) {
    const seenLimit = options.seenLimit ?? 4096;
    const badgeLimit = options.badgeLimit ?? 99;
    if (!Number.isInteger(seenLimit) || seenLimit < 1) throw new Error("seenLimit 必须是正整数");
    if (!Number.isInteger(badgeLimit) || badgeLimit < 1) throw new Error("badgeLimit 必须是正整数");
    this.#seenLimit = seenLimit;
    this.#badgeLimit = badgeLimit;
    this.#base = emptyCounts();
  }

  get ackSeq(): number {
    return this.#ackSeq;
  }

  get pendingCount(): number {
    return this.#pending.size;
  }

  // 只接受水位不低于当前快照的响应；迟到的旧快照不能把数字改小。
  applySnapshot(snapshot: UnreadSnapshot): SnapshotResult {
    const counts = normalizeCounts(snapshot.counts);
    if (!Number.isInteger(snapshot.ackSeq) || snapshot.ackSeq < 0) {
      throw new Error("ackSeq 必须是非负整数");
    }
    if (snapshot.ackSeq < this.#ackSeq) return { applied: false, stale: true, dropped: 0 };
    this.#base = counts;
    this.#ackSeq = snapshot.ackSeq;
    let dropped = 0;
    for (const [id, event] of this.#pending) {
      if (event.seq <= snapshot.ackSeq) {
        this.#pending.delete(id);
        dropped += 1;
      }
    }
    return { applied: true, stale: false, dropped };
  }

  applyEvent(event: UnreadEvent): ApplyEventResult {
    assertEvent(event);
    if (this.#seen.has(event.id)) return { status: "duplicate" };
    this.#remember(event.id);
    // 水位之内的事件已经被快照计数包含，再加一次就是双计。
    if (event.seq <= this.#ackSeq) return { status: "covered" };
    this.#pending.set(event.id, { id: event.id, kind: event.kind, seq: event.seq });
    return { status: "applied" };
  }

  applyEvents(events: UnreadEvent[]): { applied: number; duplicate: number; covered: number } {
    const result = { applied: 0, duplicate: 0, covered: 0 };
    for (const event of events) {
      const status = this.applyEvent(event).status;
      result[status] += 1;
    }
    return result;
  }

  badge(): UnreadBadge {
    const byKind = emptyCounts();
    for (const kind of NOTIFY_KINDS) byKind[kind] = this.#base[kind];
    for (const event of this.#pending.values()) byKind[event.kind] += 1;
    const total = NOTIFY_KINDS.reduce((sum, kind) => sum + byKind[kind], 0);
    const overflow = total > this.#badgeLimit;
    return {
      total,
      byKind,
      overflow,
      text: overflow ? `${this.#badgeLimit}+` : String(total),
    };
  }

  #remember(id: string): void {
    this.#seen.add(id);
    this.#seenOrder.push(id);
    // 仍在待确认集合里的事件不能丢掉去重标记，否则重发会被当成新事件。
    const cap = this.#seenOrder.length;
    let scanned = 0;
    while (this.#seenOrder.length > this.#seenLimit && scanned < cap) {
      scanned += 1;
      const oldest = this.#seenOrder.shift() as string;
      if (this.#pending.has(oldest)) {
        this.#seenOrder.push(oldest);
        continue;
      }
      this.#seen.delete(oldest);
    }
  }
}

function emptyCounts(): Record<NotifyKind, number> {
  return { like: 0, comment: 0, follow: 0, system: 0 };
}

function normalizeCounts(counts: Partial<Record<NotifyKind, number>>): Record<NotifyKind, number> {
  const result = emptyCounts();
  for (const kind of NOTIFY_KINDS) {
    const value = counts[kind] ?? 0;
    if (!Number.isInteger(value) || value < 0) throw new Error(`${kind} 计数必须是非负整数`);
    result[kind] = value;
  }
  return result;
}

function assertEvent(event: UnreadEvent): void {
  if (typeof event.id !== "string" || event.id.length === 0) throw new Error("事件 id 不能为空");
  if (!NOTIFY_KINDS.includes(event.kind)) throw new Error(`未知通知类型: ${event.kind}`);
  if (!Number.isInteger(event.seq) || event.seq < 1) throw new Error("seq 必须是正整数");
}

$ node --test test/notify/unread.test.ts
✔ 快照按类型分组，水位之后的事件逐个累加 (0.644125ms)
✔ 同一事件重复到达只计一次 (0.100917ms)
✔ 水位之内的事件已被快照包含，不再累加 (0.096959ms)
✔ 迟到的旧快照被标记 stale，不把数字改小 (0.963584ms)
✔ 快照确认后清理待确认事件，避免回滚时双计 (0.126833ms)
✔ 超过角标上限显示 99+，但底层计数不截断 (0.077542ms)
✔ 批量应用按 applied/duplicate/covered 分类计数 (0.14525ms)
ℹ tests 7
ℹ suites 0
ℹ pass 7
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 115.68625