在线客服源码中的会话分配与消息存储怎么实现

在线客服系统的核心不只是“把访客发来的内容保存下来”，还包括把会话交给合适的坐席、避免并发分配造成超载、保证消息在重试时不重复，以及在网络中断后能够按顺序恢复历史记录。处理不好这些边界，系统在本地演示时可能正常，一旦多个访客同时进入、坐席状态频繁变化或客户端重复提交，就容易出现一条会话分给多人、消息重复、序号错乱等问题。

本文用 PostgreSQL 和 Node.js 的 pg 驱动说明一套可落地的最小实现。示例重点是数据模型、数据库事务和并发控制，不依赖特定在线客服源码框架。代码适合用作原型或现有系统的实现参考；上线时还需要接入真实的用户身份认证、授权、连接管理、限流、日志和监控。

一、先明确适用场景和分配规则

本文讨论的是以下常见流程：

1. 访客创建一个待接入会话。
2. 系统从在线坐席中选择一位没有超过承载上限的坐席。
3. 坐席接入后，访客与坐席持续发送消息。
4. 消息写入数据库，并可按会话和序号分页读取。
5. 会话结束后，坐席的活动会话数不再计入容量。

“在线”“容量”“合适”都需要产品先给出明确定义。在线状态可能来自坐席主动心跳，也可能来自 WebSocket 连接；容量可能是每位坐席最多处理的活动会话数，也可能按技能组、租户或优先级分别计算。本文采用一个简单规则：只给 online = true 的坐席分配，按当前活动会话数从少到多尝试，且不超过坐席设置的 max_conversations。

这种规则适用于单个客服工作组、容量相对简单的系统。它不是复杂调度器的替代品。若需要技能匹配、VIP 优先、区域路由、排班和超时回收，应先把这些条件建模为可查询的字段或独立调度规则，再在事务内执行分配，而不是只在应用内取一个“当前最空闲”的坐席就直接写入。

二、数据模型：会话、坐席与消息

以下 DDL 面向 PostgreSQL。每个会话有一个当前坐席（未分配时为空），每条消息有会话内递增的序号。client_message_id 用于识别客户端重试的同一条消息；数据库唯一约束是幂等能力的最终保障，不能只靠应用内缓存。

CREATE TABLE agents (
  id                  BIGSERIAL PRIMARY KEY,
  display_name        TEXT NOT NULL,
  online              BOOLEAN NOT NULL DEFAULT FALSE,
  max_conversations   INTEGER NOT NULL DEFAULT 5
                      CHECK (max_conversations >= 0)
);

CREATE TABLE conversations (
  id                  BIGSERIAL PRIMARY KEY,
  visitor_id          TEXT NOT NULL,
  status              TEXT NOT NULL DEFAULT 'waiting'
                      CHECK (status IN ('waiting', 'assigned', 'closed')),
  assigned_agent_id   BIGINT REFERENCES agents(id),
  last_message_seq    BIGINT NOT NULL DEFAULT 0
                      CHECK (last_message_seq >= 0),
  created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX conversations_waiting_idx
  ON conversations (created_at, id)
  WHERE status = 'waiting';

CREATE INDEX conversations_agent_active_idx
  ON conversations (assigned_agent_id)
  WHERE status = 'assigned';

CREATE TABLE messages (
  id                  BIGSERIAL PRIMARY KEY,
  conversation_id     BIGINT NOT NULL REFERENCES conversations(id),
  seq                 BIGINT NOT NULL,
  sender_type         TEXT NOT NULL
                      CHECK (sender_type IN ('visitor', 'agent', 'system')),
  sender_id           TEXT NOT NULL,
  client_message_id   TEXT NOT NULL,
  body                TEXT NOT NULL,
  created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (conversation_id, seq),
  UNIQUE (conversation_id, client_message_id)
);

CREATE INDEX messages_history_idx
  ON messages (conversation_id, seq);

sender_id 在这个最小例子中使用文本，便于表示访客 ID 或坐席 ID。正式系统通常会将访客、坐席和租户建成独立表，并使用外键或统一的用户表。消息正文可以使用 TEXT 保存纯文本；若要支持附件，通常另建附件表保存对象存储的 key、文件类型、大小和校验信息，不建议把大文件二进制直接塞进消息行。

索引要服务于实际查询。消息历史以会话和序号范围读取，因此使用 (conversation_id, seq)；待分配会话通常按创建时间取最早的一条，因此为等待状态建部分索引。不要在没有数据规模和查询计划依据时堆叠大量索引：每增加一个索引，写入和维护成本也会增加。

三、并发分配：用数据库事务保护容量

常见错误是先在应用里查询坐席负载，再单独插入分配记录。两个请求可能同时查到“这位坐席还剩一个名额”，然后都分给他。解决思路是将“锁定坐席、统计其活动会话、写入分配结果”放在同一个数据库事务中，并且所有会影响容量的分配和关闭操作都遵循同一套锁规则。

下面的示例用 FOR UPDATE SKIP LOCKED 锁定候选坐席。一个事务正在检查某位坐席时，另一个事务会跳过被锁定的坐席，尝试其他人。锁定后再统计活动会话数，因此只要所有分配入口都先锁坐席行，检查和写入之间就不会被另一条遵守同一规则的分配操作插队。

安装依赖并设置连接字符串：

npm install express pg

DATABASE_URL='postgres://user:password@localhost:5432/support' node app.js

以下代码使用 Express 和 pg。为便于说明，创建会话和分配写在一个接口中。生产系统可以把“访客创建会话”和“后台调度器分配”拆成不同任务，但必须保留事务边界和并发约束。

const express = require('express');
const { Pool } = require('pg');

const app = express();
app.use(express.json({ limit: '32kb' }));

const pool = new Pool({
  connectionString: process.env.DATABASE_URL
});

async function createAndAssign(visitorId) {
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    const conversationResult = await client.query(
      `INSERT INTO conversations (visitor_id, status)
       VALUES ($1, 'waiting')
       RETURNING id, visitor_id, status`,
      [visitorId]
    );

    const conversation = conversationResult.rows[0];

    // 锁住候选坐席行。并发事务会跳过已被锁定的坐席。
    const candidates = await client.query(
      `SELECT id, max_conversations
       FROM agents
       WHERE online = TRUE
       ORDER BY id
       FOR UPDATE SKIP LOCKED`
    );

    let assignedAgentId = null;

    for (const agent of candidates.rows) {
      const countResult = await client.query(
        `SELECT count(*)::int AS active_count
         FROM conversations
         WHERE assigned_agent_id = $1
           AND status = 'assigned'`,
        [agent.id]
      );

      const activeCount = countResult.rows[0].active_count;

      if (activeCount < agent.max_conversations) {
        assignedAgentId = agent.id;
        break;
      }
    }

    if (assignedAgentId !== null) {
      await client.query(
        `UPDATE conversations
         SET status = 'assigned',
             assigned_agent_id = $2,
             updated_at = now()
         WHERE id = $1`,
        [conversation.id, assignedAgentId]
      );

      conversation.status = 'assigned';
      conversation.assigned_agent_id = assignedAgentId;
    }

    await client.query('COMMIT');
    return conversation;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

app.post('/conversations', async (req, res, next) => {
  try {
    // 示例假定 visitorId 已由认证中间件验证后放在请求上下文中。
    // 不要直接信任客户端提交的 visitorId。
    const visitorId = req.auth && req.auth.visitorId;

    if (!visitorId) {
      return res.status(401).json({ error: 'unauthenticated' });
    }

    const conversation = await createAndAssign(visitorId);
    res.status(201).json(conversation);
  } catch (err) {
    next(err);
  }
});

app.use((err, req, res, next) => {
  console.error(err);
  res.status(500).json({ error: 'internal_error' });
});

app.listen(3000, () => {
  console.log('Listening on port 3000');
});

代码中的 req.auth 是身份认证中间件写入的上下文，不是 Express 自动提供的属性。单独运行示例前，需要接入认证中间件，或者在本地测试中明确提供可信的开发身份；不要为了让接口“能用”就把请求体里的任意 visitorId 当作可信身份。示例里的错误处理也只是最小形式，生产环境应区分可预期的业务错误和服务端故障。

这里选择候选坐席时按 ID 排序，是为了让例子保持简单，并不代表负载均衡。可将查询改为按技能、队列优先级或其他确定性规则排序；但不能把并发安全寄托在排序上，关键仍是锁、事务以及所有写入路径一致遵守锁规则。

容量统计为什么不会超卖

假设坐席 A 还有一个名额。第一个事务锁住 A 的行并统计出活动数为 4；另一个事务选择候选时会跳过 A。如果第一个事务分配成功并提交，后续事务再获得 A 的行锁时，重新统计就会看到新的会话数。要使这一点成立，关闭会话、重新分配会话等改变活动数的操作，也必须先锁定相关坐席行，或者使用更强的数据库约束和专门的容量计数方案。

当等待会话数量很大、坐席数量很多时，上面的逐个统计会增加查询开销。可以维护坐席的活动计数，但要在同一事务中增减，并处理重复事件、会话转移和异常恢复。无论选实时计数还是计数器，都要明确一个不变量：同一条活动会话只能贡献一次负载，结束或转移后必须准确扣减。不要同时把两个方案当作互不相关的“近似值”。

若没有任何符合条件的在线坐席，接口返回的会话会保持 waiting，这是明确的业务状态，不应伪装成“已分配”。系统可以用后台任务稍后重试，也可以在坐席上线时触发分配。调度任务同样要通过事务领取等待会话，并防止多个调度实例同时处理同一条会话。

四、消息存储：幂等、顺序和权限

消息写入需要处理三类问题：同一请求被重试时不能重复保存；同一会话中的消息需要有可恢复的顺序；发送者必须确实有权限在该会话发言。HTTP 超时、WebSocket 重连和移动端离线重发都可能造成请求重复，因此客户端应为每次逻辑发送生成一个稳定的 clientMessageId，重试时沿用它。不能在每次重试时重新生成 ID，否则数据库无法识别同一条操作。

以下函数将“验证会话、检查幂等键、更新序号、插入消息”放入一个事务。会话行锁用于串行化同一会话的序号分配。不同会话之间仍可以并行写入。重复请求会读取并返回已存在的消息，不会再增加序号。

async function storeMessage({
  conversationId,
  senderType,
  senderId,
  clientMessageId,
  body
}) {
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    const conversationResult = await client.query(
      `SELECT id, status, assigned_agent_id, last_message_seq
       FROM conversations
       WHERE id = $1
       FOR UPDATE`,
      [conversationId]
    );

    if (conversationResult.rowCount === 0) {
      const err = new Error('conversation_not_found');
      err.statusCode = 404;
      throw err;
    }

    const conversation = conversationResult.rows[0];

    // 权限判断必须使用认证后的身份，而不是信任请求体中的 senderId。
    // 这里仅展示坐席与访客的基本归属检查；实际项目应接入完整鉴权逻辑。
    if (senderType === 'agent' &&
        String(conversation.assigned_agent_id) !== String(senderId)) {
      const err = new Error('agent_not_assigned');
      err.statusCode = 403;
      throw err;
    }

    if (senderType === 'visitor' &&
        conversation.status === 'closed') {
      const err = new Error('conversation_closed');
      err.statusCode = 409;
      throw err;
    }

    const existing = await client.query(
      `SELECT id, conversation_id, seq, sender_type, sender_id,
              client_message_id, body, created_at
       FROM messages
       WHERE conversation_id = $1
         AND client_message_id = $2`,
      [conversationId, clientMessageId]
    );

    if (existing.rowCount > 0) {
      await client.query('COMMIT');
      return existing.rows[0];
    }

    const nextSeq = Number(conversation.last_message_seq) + 1;

    const inserted = await client.query(
      `INSERT INTO messages (
         conversation_id, seq, sender_type, sender_id,
         client_message_id, body
       )
       VALUES ($1, $2, $3, $4, $5, $6)
       RETURNING id, conversation_id, seq, sender_type, sender_id,
                 client_message_id, body, created_at`,
      [
        conversationId,
        nextSeq,
        senderType,
        senderId,
        clientMessageId,
        body
      ]
    );

    await client.query(
      `UPDATE conversations
       SET last_message_seq = $2,
           updated_at = now()
       WHERE id = $1`,
      [conversationId, nextSeq]
    );

    await client.query('COMMIT');
    return inserted.rows[0];
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

序号使用会话行上的 last_message_seq，而不是对消息表执行 MAX(seq) + 1。后者在并发写入时可能由多个请求算出相同的下一序号；唯一约束虽然能阻止脏数据，但会让其中一个请求失败。锁定会话行后更新序号，则为同一会话提供了明确的串行化点。消息表的唯一约束仍然要保留，作为最后一道数据完整性保护。

消息 API 的调用方式可以是：

app.post('/conversations/:id/messages', async (req, res, next) => {
  try {
    const conversationId = Number(req.params.id);
    const auth = req.auth;

    if (!auth) {
      return res.status(401).json({ error: 'unauthenticated' });
    }

    const { clientMessageId, body } = req.body || {};

    if (!Number.isSafeInteger(conversationId) || conversationId <= 0) {
      return res.status(400).json({ error: 'invalid_conversation_id' });
    }

    if (typeof clientMessageId !== 'string' ||
        clientMessageId.length < 1 ||
        clientMessageId.length > 128) {
      return res.status(400).json({ error: 'invalid_client_message_id' });
    }

    if (typeof body !== 'string' || body.trim().length === 0) {
      return res.status(400).json({ error: 'invalid_body' });
    }

    // 根据认证主体确定身份，不允许客户端自行指定 senderType 或 senderId。
    const message = await storeMessage({
      conversationId,
      senderType: auth.type,
      senderId: String(auth.id),
      clientMessageId,
      body
    });

    res.status(201).json(message);
  } catch (err) {
    next(err);
  }
});

示例对会话 ID 和幂等键设置了接口层格式检查，正文只检查非空，没有设定业务字数上限。express.json({ limit: '32kb' }) 限制的是 JSON 请求体大小，不等同于正文字符数限制；一个请求的 JSON 编码、字段名称和其他字段也会占用字节。正式系统应根据产品需求明确正文字符数或字节数上限，并在服务端和数据库存储策略中保持一致。传给数据库的值使用参数化查询，避免将用户输入拼进 SQL 字符串。

示例的坐席权限检查只能说明基本思路：发言坐席必须是该会话当前分配的坐席。访客校验也不能仅检查 senderType，必须核对认证身份是否属于此会话。多租户系统还应校验会话和身份属于同一租户。权限判断失败应拒绝请求，而不能因前端隐藏了按钮就认为安全。

五、历史读取与断线恢复

历史记录不宜一次性返回整段会话。可以使用会话内序号做游标，获取某个序号之后的消息，既便于断线恢复，也避免使用不断变化的偏移量。示例查询如下：

SELECT id, seq, sender_type, sender_id, client_message_id, body, created_at
FROM messages
WHERE conversation_id = $1
  AND seq > $2
ORDER BY seq ASC
LIMIT $3;

LIMIT 应由服务端设置合理上限，不能把未经校验的任意数值直接传给数据库。连接恢复时，客户端可提交最后确认的序号，服务端返回其后的消息；如果需要让客户端确认服务端已接收消息，也应区分“已接收”“已送达设备”和“已读”等状态，它们不是同一件事。可以建立单独的送达回执或已读位置表，避免在消息正文行上不断覆盖与消息内容不同的状态。

序号是数据库中的会话内顺序，不代表消息一定按网络到达时间排列，也不代表多个设备上展示顺序可以只靠客户端时间戳保证。客户端本地时间可能不准；服务端写入事务形成的序号更适合作为历史顺序依据。若业务需要表达编辑、撤回或删除，通常应使用额外事件或状态字段，并保留审计要求，而不是让客户端直接改写历史记录。

六、关闭会话和重新分配也要维护不变量

前面的分配示例通过统计 status = 'assigned' 的会话数计算负载。因此关闭操作至少要在同一事务中更新会话状态。若实现了“关闭会话必须锁定坐席”的规则，关闭时也应先锁坐席行，再锁会话行；所有代码路径尽量采用一致的锁定顺序，避免不同流程分别先锁坐席、先锁会话而产生死锁。

一种简单的应用约定是：分配、关闭和转移都先锁坐席，再锁会话；对于尚未分配的会话，先锁会话并在获得坐席时按统一顺序处理。复杂场景下更稳妥的做法是设计明确的锁顺序，发现 PostgreSQL 死锁错误时安全重试整个事务。不能只捕获错误后从失败的中间步骤继续运行。

会话从坐席 A 转给坐席 B 时，状态更新必须原子完成。若使用实时统计，分配结果提交前的状态变更会自然影响统计结果；若使用计数器，则必须在同一个事务中给 A 减一、给 B 加一，并确保重复转移不会重复加减。重复关闭也应具有幂等效果，不能让计数器被多扣一次。

七、消息数据库与实时推送的职责不同

WebSocket、SSE 或消息代理负责实时传输，关系数据库负责可靠保存和历史查询。不能把“已经向 WebSocket 发送成功”当作“消息已经可靠入库”，也不能因为数据库写入成功就假定客户端一定收到。一个常用顺序是先提交数据库，再尝试推送；客户端断线时通过历史接口补齐。

如果业务要求数据库提交后必须可靠地向多个消费者发布事件，可以在同一事务中写入消息和 outbox 事件，再由后台任务把 outbox 发布到消息代理并标记处理状态。这样可以避免“数据库提交成功，但进程在发布事件前崩溃”造成的通知丢失。即使采用 outbox，消费者仍应按事件 ID 做幂等处理，因为消息代理和后台任务可能重复投递。

附件上传通常先经过鉴权和文件校验，再写入对象存储，数据库保存附件元数据和存储标识。上传过程和消息事务很难成为一个跨服务的原子事务，应明确失败后的清理或补偿流程。不要把用户提供的文件名直接当作对象存储路径，也不要把可公开访问的长期凭据放进消息正文。

八、安全边界和参数说明

• 身份来源：访客 ID、坐席 ID 和租户 ID 必须来自服务端验证后的身份上下文。请求体中的同名字段不是授权凭证。
• 输入校验：校验 ID、正文类型、幂等键长度和请求体大小；还应按产品要求限制消息长度、频率和附件类型。具体上限应由需求和存储策略确定，不能把示例中的 32kb 当成通用标准。
• SQL 安全：示例使用 PostgreSQL 参数占位符传递数据。参数化查询可以避免输入值被当作 SQL 语法解析，但不能替代业务权限校验。
• 内容展示：保存纯文本不等于输出时安全。将正文渲染到网页时应按 HTML 上下文进行转义；若产品允许富文本，应使用经过审查的白名单净化方案。不要把用户正文直接拼接进 HTML。
• 敏感信息：按业务要求减少不必要的个人信息，设置访问控制、备份保护和保留期限。数据库中有历史消息，并不意味着所有运维人员都应该能查看。
• 错误响应：对外避免回传 SQL、连接串和堆栈信息；内部记录可关联请求 ID 的错误日志，但日志中也应避免无必要地记录完整敏感正文。
• 滥用防护：针对访客和坐席接口设置适当的速率限制与连接限制。速率限制不是身份认证，也不能替代会话级授权。

九、常见错误及其原因

1. 先查负载，再单独写分配

这是典型的并发竞态。两个事务可能同时看到同一名额。应把锁定、统计和写入放在同一个事务内，并要求所有改变负载的操作遵守同一规则。

2. 只在应用缓存中保存幂等键

进程重启、多个服务实例和缓存过期都会让缓存失效。对需要持久幂等的消息，应在数据库设置唯一约束，并定义客户端重试时复用同一个键。

3. 用客户端时间戳排序消息

设备时钟可能偏差，网络也可能延迟或乱序。历史顺序应由服务端定义，例如本文的会话序号；客户端时间可作为显示或诊断信息，不能取代服务端顺序。

4. 把“在线”理解为绝对准确

断网后服务端通常要经过心跳超时才知道坐席离线，在线状态天然可能滞后。要明确心跳周期、过期判定和恢复策略；分配时可再检查数据库里的在线状态，但不能假定它与现实连接状态完全同步。

5. 消息写入成功就直接返回“已送达”

入库、推送到连接、客户端接收和用户阅读是不同阶段。接口应使用准确的状态语义，断线恢复应基于历史查询或事件游标。

6. 用“查不到会话”代替完整授权

即使会话 ID 很难猜，也不能据此授权。读取历史、发送消息、关闭会话和下载附件都应逐项检查调用者是否属于该会话及租户。

7. 在生产中无限制读取历史

一条会话可能积累大量消息。使用序号游标和服务端分页上限，避免把整段历史一次载入内存或通过单次响应传输。

十、实施前的检查清单

• 分配规则是否明确在线判定、容量口径、排队顺序和无可用坐席时的行为？
• 所有分配、关闭和转移入口是否使用一致的事务与锁定约定？
• 消息是否有数据库级幂等约束，以及会话内稳定顺序？
• 认证身份是否由服务端生成，所有消息和历史接口是否执行对象级授权？
• API 是否设置符合业务需要的请求体、正文和分页限制？
• 断线补取、重复投递、服务重启和数据库事务失败是否经过测试？
• 数据库迁移、备份恢复、索引维护和消息保留策略是否纳入运维流程？

会话分配与消息存储的可靠性，主要来自清晰的不变量：一条活动会话只归属一个当前坐席；容量判断和分配写入处于同一并发控制范围；同一逻辑消息有稳定幂等键；同一会话的消息顺序由服务端统一产生；所有读写都经过身份和会话权限校验。把这些规则落实到数据库约束和事务里，再用实时通道负责通知，才更容易应对重试、并发和断线等真实场景。