《互联网产品的客户端架构·内容、社交与实时互动》8.2 输入中状态的节流与超时

// file: src/presence/typing.ts
// tier: L2
// verify: node --test --test-concurrency=1 test/presence/typing.test.ts

export interface TypingPolicy {
  throttleMs: number;
  timeoutMs: number;
}

export const DEFAULT_TYPING_POLICY: TypingPolicy = {
  throttleMs: 3_000,
  timeoutMs: 6_000,
};

export type Clock = () => number;

interface RemoteEntry {
  expiresAt: number;
  lastSignalAt: number;
}

export class TypingTracker {
  #throttleMs: number;
  #timeoutMs: number;
  #now: Clock;
  #lastSentAt = new Map<string, number>();
  #startedAt = new Map<string, number>();
  #remote = new Map<string, Map<string, RemoteEntry>>();
  #closedAt = new Map<string, Map<string, number>>();

  constructor(policy: TypingPolicy = DEFAULT_TYPING_POLICY, now: Clock = () => Date.now()) {
    if (!Number.isInteger(policy.throttleMs) || policy.throttleMs < 1) {
      throw new Error("throttleMs 必须是正整数");
    }
    if (!Number.isInteger(policy.timeoutMs) || policy.timeoutMs <= policy.throttleMs) {
      throw new Error("timeoutMs 必须大于 throttleMs，否则心跳之间会闪烁");
    }
    this.#throttleMs = policy.throttleMs;
    this.#timeoutMs = policy.timeoutMs;
    this.#now = now;
  }

  // 本地按键：返回是否应立即上行发送 start。首次与每个节流窗口的第一下都发送。
  noteKeystroke(conversationId: string, at: number = this.#now()): boolean {
    assertId(conversationId, "conversationId");
    assertTime(at);
    const last = this.#lastSentAt.get(conversationId);
    if (last === undefined || at - last >= this.#throttleMs) {
      this.#lastSentAt.set(conversationId, at);
      this.#startedAt.set(conversationId, at);
      return true;
    }
    return false;
  }

  // 发送、清空或失焦：只有远端可能仍显示「正在输入」时才需要发 stop。
  stopLocal(conversationId: string, at: number = this.#now()): boolean {
    assertId(conversationId, "conversationId");
    assertTime(at);
    const started = this.#startedAt.get(conversationId);
    this.#lastSentAt.delete(conversationId);
    this.#startedAt.delete(conversationId);
    if (started === undefined) return false;
    // start 发出后若已超过远端超时，对方早已自动隐藏，不必再补一条 stop。
    return at - started < this.#timeoutMs;
  }

  ingest(conversationId: string, userId: string, kind: "start" | "stop", at: number): void {
    assertId(conversationId, "conversationId");
    assertId(userId, "userId");
    assertTime(at);
    const entries = this.#remote.get(conversationId) ?? new Map<string, RemoteEntry>();
    if (kind === "stop") {
      entries.delete(userId);
      this.#rememberClose(conversationId, userId, at);
      this.#remote.set(conversationId, entries);
      return;
    }
    const closed = this.#closedAt.get(conversationId)?.get(userId);
    // stop 之后才到达的旧 start 不能复活输入状态。
    if (closed !== undefined && at <= closed) return;
    const previous = entries.get(userId);
    // 乱序的旧 start 不能缩短或覆盖已经更新的 start。
    if (previous && at < previous.lastSignalAt) return;
    entries.set(userId, { expiresAt: at + this.#timeoutMs, lastSignalAt: at });
    this.#remote.set(conversationId, entries);
  }

  isTyping(conversationId: string, userId: string, at: number = this.#now()): boolean {
    assertTime(at);
    this.prune(conversationId, at);
    const entry = this.#remote.get(conversationId)?.get(userId);
    return entry !== undefined && entry.expiresAt > at;
  }

  activeTypers(conversationId: string, viewerId?: string, at: number = this.#now()): string[] {
    assertTime(at);
    this.prune(conversationId, at);
    const entries = this.#remote.get(conversationId);
    if (!entries) return [];
    return [...entries.keys()].filter((userId) => userId !== viewerId).sort();
  }

  prune(conversationId: string, at: number = this.#now()): number {
    const entries = this.#remote.get(conversationId);
    if (!entries) return 0;
    let removed = 0;
    for (const [userId, entry] of entries) {
      if (entry.expiresAt <= at) {
        entries.delete(userId);
        removed += 1;
      }
    }
    return removed;
  }

  #rememberClose(conversationId: string, userId: string, at: number): void {
    const closed = this.#closedAt.get(conversationId) ?? new Map<string, number>();
    closed.set(userId, Math.max(closed.get(userId) ?? Number.NEGATIVE_INFINITY, at));
    this.#closedAt.set(conversationId, closed);
  }
}

function assertId(value: string, field: string): void {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(`${field} 必须是非空字符串`);
  }
}

function assertTime(value: number): void {
  if (!Number.isFinite(value)) throw new Error("typing 时间必须是有限数字");
}

$ node --test --test-concurrency=1 test/presence/typing.test.ts
✔ 首次按键立即发送，节流窗口内保持静默 (0.639333ms)
✔ 持续输入时按节流窗口重发，作为远端租约的心跳 (0.079542ms)
✔ 没发过 start 时不发 stop，停止后再次输入立即重发 (0.090459ms)
✔ start 已超过远端超时后才停止，不必再发 stop (0.461583ms)
✔ 远端 start 在超时前活跃，到达超时边界后自动消失 (0.519708ms)
✔ 远端 stop 立即清除状态，不等超时 (0.152542ms)
✔ 会话之间完全隔离，stop 不跨会话生效 (0.090583ms)
✔ start 丢包时不凭本地按键推断，stop 丢包时由超时兜底 (0.063292ms)
✔ stop 之后到达的旧 start 不会复活，活跃列表按 id 排序并排除自己 (0.31125ms)
ℹ tests 9
ℹ suites 0
ℹ pass 9
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 146.103083