用户关掉 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 创建了多条重复消息，补拉又把它们全部拉回本地，页面上的气泡数量翻倍。
复盘发现两个问题叠加：
- 重试没有稳定的
clientId，服务端无法判断「这是同一条消息」。 - 补拉没有按
lastSeq过滤，把本地已经发出去的消息又当成新消息插入。
修复做了三件事：发送前生成 会话id#本地序号 作为 clientId 并持久化；重试只复用该 id；补拉带 afterSeq，收到自己消息的回显时按 clientId 合并而不是新增。另一个被顺手修掉的问题是：会话关闭时直接清空缓存，用户再打开看不到历史；修复后关闭只改状态并保留下载，未发送消息标记失败但保留原文。
这个复盘留下的规则是：离线链路的核心不是「重试」，而是「同一条消息在多次重试后仍然能被认出来」。
小结
工单会话要同时管理本地快照、离线队列与服务端序号。入队时生成稳定的 clientId，补发时复用，补拉时按 lastSeq 取增量，合并时同时按 id 与 clientId 去重。关闭与超时不能清空历史，只能把状态改为只读，并把未发送消息标记失败。
send 负责入队，flush 负责补发，receive 负责合并，sync 负责一次重连的完整动作。四者合起来，用户关掉 App 再回来时，看到的才是一条连续的会话，而不是一段被截断又重新开始的对话。
阅读导航：上一节：11.2 售后申请的状态与证据 · 下一节：12.1 埋点模型与事件契约 。
常见错误与边界
- 离线队列只放内存，进程被杀后未发送消息全部丢失。
- 每次重试生成新的消息 id，服务端创建多条重复消息。
- 补拉按时间而不是按序号，时钟漂移后漏消息或重复拉取。
- 只按消息 id 去重，忽略自己消息回显的
clientId，页面出现两个气泡。 - 补拉结果不完整时直接当成全部，用户看不到更早的历史。
- 会话关闭时清空缓存，用户重新进入后历史消失。
- 会话关闭后仍允许发送，消息进入一个永远不会被处理的队列。
- 空闲超时后自动补发旧队列，把已过期的诉求重新发给客服。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。