《互联网产品的客户端架构·交易、增长与平台化》11.3 工单会话的离线与恢复

客服工单最容易被低估的是「用户关掉 App 之后」。消息发没发出去、重连后会不会重复、会话关闭后历史还在不在,决定了用户是否信任这个入口。本节用注入存储与假时钟实现本地缓存、离线队列、按序号补拉与去重,并复盘一次重连后消息翻倍的完整排查过程。

用户关掉 App 之后,消息去哪了

客服会话与普通聊天最大的差别,是它承载售后与投诉的事实。用户在地铁里发了一句「我要退货」,网络断了,App 被系统杀掉;二十分钟后重新打开,这句话可能已经消失,也可能发了两遍。前者让用户重复描述,后者让客服看到重复工单。

这条链路有三个必须回答的问题:消息在本地存在哪里、离线时排在哪里、重连后如何补拉与去重。三者缺一,用户就会觉得「这个客服入口不可靠」。

三个存储位置,三套责任

位置存什么负责什么不负责什么
本地快照会话状态、消息、lastSeq、本地序号冷启动恢复、离线展示服务端事实
离线队列未确认的发送消息与 clientId断网重试、顺序补发决定消息是否真的送达
服务端消息 id、seq、会话状态排序、去重、最终事实端上草稿与未发送内容

三个决定:

  • 本地快照要持久化,不只放内存。 进程被杀后,内存里的队列与历史一起消失。
  • 离线消息要有稳定的 clientId。 重试复用同一个 clientId,服务端才能把它识别为同一条消息;每次重试换新 id,就会产生重复。
  • 补拉按 lastSeq 而不是按时间。 时间会漂移,序号是服务端给的单调事实。

一个可运行的工单会话模块

下面的模块属于 L2:时间、存储与传输全部由调用方注入,模块本身不建连接、不发真实请求。测试用假时钟推进空闲超时,用内存存储模拟冷启动。

// file: src/support/ticket-session.ts
// tier: L2
// verify: node --test test/support/ticket-session.test.ts

export type SessionState = "open" | "closed" | "expired";
export type MessageStatus = "queued" | "sent" | "failed";
export type Direction = "client" | "agent" | "system";

export interface TicketMessage {
  id: string | null; clientId: string | null; direction: Direction; body: string;
  at: number; seq: number | null; status: MessageStatus; failure: string | null;
}
export interface TicketSnapshot {
  sessionId: string; state: SessionState; messages: TicketMessage[]; lastSeq: number;
  nextLocalSeq: number; updatedAt: number; closeReason: string | null;
}
export interface TicketStore {
  load(sessionId: string): TicketSnapshot | null;
  save(snapshot: TicketSnapshot): void;
}
export interface PushAck { clientId: string; id: string; seq: number; at: number }
export interface RemoteMessage {
  id: string; clientId: string | null; direction: Direction; body: string;
  at: number; seq: number;
}
export interface PushBatch {
  sessionId: string;
  messages: Array<{ clientId: string; body: string; at: number }>;
}
export type PushResult = { ok: true; acked: PushAck[] } | { ok: false; retryable: boolean };
export type PullResult = { ok: true; messages: RemoteMessage[]; complete: boolean } | { ok: false };

// 传输层由调用方注入;本模块不创建连接、不发真实请求。
export interface Transport {
  push(batch: PushBatch): PushResult;
  pull(sessionId: string, afterSeq: number): PullResult;
}
export type SessionOptions = { ttlMs?: number; maxQueue?: number; maxBody?: number };

function localSeq(message: TicketMessage): number {
  return message.clientId === null ? 0 : Number(message.clientId.split("#").pop() ?? "0");
}

export class TicketSession {
  sessionId: string;
  state: SessionState;
  messages: TicketMessage[];
  lastSeq: number;
  nextLocalSeq: number;
  updatedAt: number;
  closeReason: string | null;
  now: () => number;
  store: TicketStore;
  ttlMs: number;
  maxQueue: number;
  maxBody: number;

  constructor(snapshot: TicketSnapshot, now: () => number, store: TicketStore,
    options: SessionOptions = {}) {
    Object.assign(this, snapshot);
    this.messages = snapshot.messages.map((message) => ({ ...message }));
    this.now = now;
    this.store = store;
    this.ttlMs = options.ttlMs ?? 30 * 60_000;
    this.maxQueue = options.maxQueue ?? 50;
    this.maxBody = options.maxBody ?? 2_000;
  }

  snapshot(): TicketSnapshot {
    return {
      sessionId: this.sessionId, state: this.state,
      messages: this.messages.map((message) => ({ ...message })),
      lastSeq: this.lastSeq, nextLocalSeq: this.nextLocalSeq,
      updatedAt: this.updatedAt, closeReason: this.closeReason,
    };
  }
  history(): TicketMessage[] {
    return this.messages.map((message) => ({ ...message }));
  }
  canSend(): boolean { return this.state === "open"; }
  queued(): TicketMessage[] {
    return this.messages.filter((message) => message.status === "queued")
      .sort((left, right) => localSeq(left) - localSeq(right));
  }
  save(): void { this.store.save(this.snapshot()); }

  send(body: string): TicketMessage {
    if (this.state !== "open") throw new Error(`会话状态为 ${this.state},不能发送消息`);
    const text = body.trim();
    if (text === "") throw new Error("消息内容不能为空");
    if (text.length > this.maxBody) throw new Error(`消息超过 ${this.maxBody} 字`);
    if (this.queued().length >= this.maxQueue) throw new Error(`离线队列已满(${this.maxQueue} 条)`);
    const sequence = this.nextLocalSeq + 1;
    this.nextLocalSeq = sequence;
    const message: TicketMessage = {
      id: null, clientId: `${this.sessionId}#${sequence}`, direction: "client",
      body: text, at: this.now(), seq: null, status: "queued", failure: null,
    };
    this.messages.push(message);
    this.updatedAt = this.now();
    this.save();
    return { ...message };
  }

  // 一次刷新把队列整体交给传输层;重试复用同一个 clientId,服务端据此去重。
  flush(transport: Transport) {
    const queued = this.queued();
    if (queued.length === 0) return { ok: true, sent: 0, remaining: 0, retryable: false };
    const result = transport.push({
      sessionId: this.sessionId,
      messages: queued.map((message) => ({
        clientId: message.clientId ?? "", body: message.body, at: message.at,
      })),
    });
    if (!result.ok) {
      if (!result.retryable) this.failQueued("rejected");
      this.updatedAt = this.now();
      this.save();
      return { ok: false, sent: 0, remaining: this.queued().length, retryable: result.retryable };
    }
    const acked = new Map(result.acked.map((ack) => [ack.clientId, ack]));
    let sent = 0;
    for (const message of queued) {
      const ack = message.clientId === null ? undefined : acked.get(message.clientId);
      if (ack === undefined) continue;
      message.id = ack.id; message.seq = ack.seq; message.at = ack.at;
      message.status = "sent"; message.failure = null; sent += 1;
      if (ack.seq > this.lastSeq) this.lastSeq = ack.seq;
    }
    this.updatedAt = this.now();
    this.save();
    return { ok: true, sent, remaining: this.queued().length, retryable: false };
  }

  failQueued(failure: string): number {
    let count = 0;
    for (const message of this.messages) {
      if (message.status !== "queued") continue;
      message.status = "failed"; message.failure = failure; count += 1;
    }
    return count;
  }

  // 只有服务端明确拒绝的消息可以手动重试;会话关闭导致的失败不允许复活。
  retryFailed(): number {
    if (this.state !== "open") return 0;
    let count = 0;
    for (const message of this.messages) {
      if (message.status === "failed" && message.failure === "rejected") {
        message.status = "queued"; message.failure = null; count += 1;
      }
    }
    if (count > 0) { this.updatedAt = this.now(); this.save(); }
    return count;
  }

  // 服务端消息按 seq 合并;自己消息的回显按 clientId 合并,不新增气泡。
  receive(remote: RemoteMessage[]) {
    let added = 0;
    let duplicates = 0;
    const sorted = [...remote].sort((left, right) =>
      left.seq !== right.seq ? left.seq - right.seq : left.at - right.at
    );
    for (const item of sorted) {
      if (!Number.isSafeInteger(item.seq) || item.seq <= 0) { duplicates += 1; continue; }
      if (item.clientId !== null) {
        const own = this.messages.find((message) =>
          message.clientId === item.clientId && message.direction === "client"
        );
        if (own !== undefined) {
          own.id = item.id; own.seq = item.seq; own.at = item.at;
          if (own.status !== "failed") own.status = "sent";
          own.failure = null;
          if (item.seq > this.lastSeq) this.lastSeq = item.seq;
          duplicates += 1;
          continue;
        }
      }
      if (this.messages.some((message) => message.id === item.id)) {
        duplicates += 1;
        if (item.seq > this.lastSeq) this.lastSeq = item.seq;
        continue;
      }
      this.messages.push({
        id: item.id, clientId: item.clientId, direction: item.direction,
        body: item.body, at: item.at, seq: item.seq, status: "sent", failure: null,
      });
      added += 1;
      if (item.seq > this.lastSeq) this.lastSeq = item.seq;
    }
    this.messages.sort((left, right) => {
      if (left.seq !== null && right.seq !== null) return left.seq - right.seq;
      if (left.seq === null && right.seq === null) return localSeq(left) - localSeq(right);
      return left.seq === null ? 1 : -1;
    });
    this.updatedAt = this.now();
    if (added > 0 || duplicates > 0) this.save();
    return { added, duplicates };
  }

  // 重连一次做两件事:先补发队列,再按 lastSeq 拉取缺失消息。
  sync(transport: Transport) {
    const flushed = this.flush(transport);
    if (!flushed.ok) {
      return { ok: false, sent: flushed.sent, added: 0, duplicates: 0, gap: false, hasMore: false };
    }
    const afterSeq = this.lastSeq;
    const pulled = transport.pull(this.sessionId, afterSeq);
    if (!pulled.ok) {
      return { ok: false, sent: flushed.sent, added: 0, duplicates: 0, gap: false, hasMore: false };
    }
    const received = this.receive(pulled.messages);
    const seqs = pulled.messages.map((message) => message.seq)
      .filter((seq) => Number.isSafeInteger(seq) && seq > 0);
    const minSeq = seqs.length > 0 ? Math.min(...seqs) : null;
    const gap = (minSeq !== null && minSeq > afterSeq + 1) || pulled.complete === false;
    return {
      ok: true, sent: flushed.sent, added: received.added,
      duplicates: received.duplicates, gap, hasMore: pulled.complete === false,
    };
  }

  // 关闭时未发出的消息不能静默丢弃,标记失败并保留原文供用户复制。
  close(reason: string, at?: number): number {
    if (this.state !== "open") throw new Error(`会话状态为 ${this.state},不能重复关闭`);
    this.state = "closed";
    this.closeReason = reason.trim() === "" ? "closed" : reason.trim();
    const failed = this.failQueued("session_closed");
    this.updatedAt = at ?? this.now();
    this.save();
    return failed;
  }

  // 恢复时先判断空闲超时;超时的会话转为只读,不自动发送旧队列。
  applyIdle(): boolean {
    if (this.state !== "open") return false;
    if (this.now() - this.updatedAt <= this.ttlMs) return false;
    this.state = "expired";
    this.closeReason = "idle_timeout";
    this.failQueued("session_expired");
    this.updatedAt = this.now();
    this.save();
    return true;
  }
}

export function openTicket(sessionId: string, store: TicketStore, now: () => number,
  options: SessionOptions = {}): TicketSession {
  const id = sessionId.trim();
  if (id === "") throw new Error("会话标识不能为空");
  if (store.load(id) !== null) throw new Error(`会话 ${id} 已存在,请使用恢复入口`);
  const snapshot: TicketSnapshot = {
    sessionId: id, state: "open", messages: [], lastSeq: 0, nextLocalSeq: 0,
    updatedAt: now(), closeReason: null,
  };
  const session = new TicketSession(snapshot, now, store, options);
  session.save();
  return session;
}

export function resumeTicket(sessionId: string, store: TicketStore, now: () => number,
  options: SessionOptions = {}): TicketSession | null {
  const snapshot = store.load(sessionId);
  if (snapshot === null) return null;
  if (snapshot.sessionId !== sessionId) throw new Error("会话快照标识不一致");
  const session = new TicketSession(snapshot, now, store, options);
  session.applyIdle();
  return session;
}

四个设计点。

send 只入队,不直接发送。 发送成功与否由 flush 决定。这样离线、弱网与页面切走都走同一条路径,不需要为每种情况写一套逻辑。

flush 把队列整体交给传输层,并按 ack 更新状态。 可重试失败保留队列,不可重试失败标记 failed;只有服务端明确拒绝的消息允许手动重试,会话关闭导致的失败不能复活。

receive 同时按 id 与 clientId 去重。 服务端可能回显用户自己发的消息,这条回显的 id 是新的,但 clientId 与本地一致。只按 id 去重会多出一个气泡,只按 clientId 去重会漏掉客服消息。

sync 先补发再补拉,并用序号跳跃判断缺口。 补拉结果不完整或最小序号大于 lastSeq + 1 时,调用方应继续翻页或提示「正在同步历史」,而不是假装消息已经齐全。

测试与真实输出

本机运行结果如下:

$ node --test --test-concurrency=1 test/support/ticket-session.test.ts
✔ 新会话发送后进入本地队列并持久化 (0.76775ms)
✔ 空白消息被拒绝 (0.24725ms)
✔ 超长消息被拒绝 (0.090125ms)
✔ 离线队列达到上限后拒绝继续发送 (0.134417ms)
✔ 发送成功后记录服务端序号并清空队列 (0.589208ms)
✔ 可重试失败保留队列并在下次复用同一个 clientId (0.126ms)
✔ 不可重试失败标记为拒绝,用户手动重试后可以再发 (0.141416ms)
✔ 补拉的服务端消息按序号排序 (0.188583ms)
✔ 重复的服务端消息按 id 去重 (0.092667ms)
✔ 自己的消息回显按 clientId 合并而不是新增气泡 (0.164958ms)
✔ 重连补拉使用已确认的最大序号 (0.139584ms)
✔ 补拉序号出现跳跃时报告缺口 (0.101333ms)
✔ 服务端还有下一页时报告未完成 (0.061125ms)
✔ 快照恢复后保留历史、队列与本地序号 (0.165333ms)
✔ 关闭会话把未发送消息标记失败并阻止继续发送 (0.122167ms)
✔ 关闭后的会话恢复为只读并保留历史 (0.088209ms)
✔ 空闲超时后恢复为过期并标记旧队列 (0.06225ms)
✔ 一次同步完成补发与补拉 (0.105375ms)
✔ 快照是副本,外部修改不会影响会话 (0.075125ms)
ℹ tests 19
ℹ suites 0
ℹ pass 19
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 127.78525

十九条用例覆盖入队、补发、补拉、去重、恢复与关闭。它们不验证真实网络的重连时序,那需要 L3 的本地 HTTP 回调。

一次「重连后消息翻倍」的复盘

某交易客户端的客服页在每次重试时生成新的消息 id,本地队列与服务端各存一份。一次地铁断网后,用户发了三条消息,端上重试了四次;恢复网络后,服务端按新 id 创建了多条重复消息,补拉又把它们全部拉回本地,页面上的气泡数量翻倍。

复盘发现两个问题叠加:

  1. 重试没有稳定的 clientId,服务端无法判断「这是同一条消息」。
  2. 补拉没有按 lastSeq 过滤,把本地已经发出去的消息又当成新消息插入。

修复做了三件事:发送前生成 会话id#本地序号 作为 clientId 并持久化;重试只复用该 id;补拉带 afterSeq,收到自己消息的回显时按 clientId 合并而不是新增。另一个被顺手修掉的问题是:会话关闭时直接清空缓存,用户再打开看不到历史;修复后关闭只改状态并保留下载,未发送消息标记失败但保留原文。

这个复盘留下的规则是:离线链路的核心不是「重试」,而是「同一条消息在多次重试后仍然能被认出来」。

小结

工单会话要同时管理本地快照、离线队列与服务端序号。入队时生成稳定的 clientId,补发时复用,补拉时按 lastSeq 取增量,合并时同时按 id 与 clientId 去重。关闭与超时不能清空历史,只能把状态改为只读,并把未发送消息标记失败。

send 负责入队,flush 负责补发,receive 负责合并,sync 负责一次重连的完整动作。四者合起来,用户关掉 App 再回来时,看到的才是一条连续的会话,而不是一段被截断又重新开始的对话。

阅读导航:上一节:11.2 售后申请的状态与证据 · 下一节:12.1 埋点模型与事件契约 。

常见错误与边界

  • 离线队列只放内存,进程被杀后未发送消息全部丢失。
  • 每次重试生成新的消息 id,服务端创建多条重复消息。
  • 补拉按时间而不是按序号,时钟漂移后漏消息或重复拉取。
  • 只按消息 id 去重,忽略自己消息回显的 clientId,页面出现两个气泡。
  • 补拉结果不完整时直接当成全部,用户看不到更早的历史。
  • 会话关闭时清空缓存,用户重新进入后历史消失。
  • 会话关闭后仍允许发送,消息进入一个永远不会被处理的队列。
  • 空闲超时后自动补发旧队列,把已过期的诉求重新发给客服。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查