# 微通信架构与端到端流程设计（MQTT/Kafka/Supabase/ClickHouse + 个推）

本方案围绕当前既有逻辑“设备/客户端通过 MQTT 上报 →（Kafka 或）直接写入 Supabase → 前端 chat/index 接收 → 前端发送消息至 MQTT 并写入 Supabase”进行系统化设计，覆盖后端组件、数据流、推送、可靠性与前端实现要点。

## 1. 总览

- 目标
  - 低延迟、多终端的聊天与事件通知，既支持 App/Web，也支持物联设备/边缘端。
  - 统一消息落库（PostgreSQL on Supabase）+ 实时分发（Supabase Realtime / MQTT）。
  - 推送采用“个推”在 App 后台或离线时补充通知到达率。
- 核心组件
  - MQTT Broker（EMQX/Mosquitto）：收消息/发消息的设备级/轻端入口，QoS≥1。
  - 网关服务 gateway-mqtt（可选：内置 Kafka Bridge）：
    - 认证/ACL、协议转换、消息校验与标准化。
    - 转发到 Kafka（chat.inbound）或直接调用 Supabase（REST/RPC/pg）落库。
  - Kafka（可选/增强）：
    - 主题：chat.inbound（入口）、chat.persisted（落库成功后）、chat.push（通知任务）。
    - 解耦高峰、可回放、做扩展处理（统计、风控、审计）。
  - Supabase（Postgres + Realtime + Edge Functions）：
    - 表：chat_conversations / chat_participants / chat_messages / chat_notifications（已建表，见 `doc_chat/create_chat_tables.sql`）。
    - 触发器：chat_on_message_insert（更新 last_message_at + 写通知）。
    - RLS：按参与者隔离访问（已配置）。
    - Realtime：前端订阅消息/通知的实时流。
  - ClickHouse（列存/OLAP/时序聚合）：
    - 承载健康/传感类高频时序数据（大跨度查询、长保留、低成本）。
    - 与 Kafka/Telegraf 对接，建立明细表 + 物化视图做分钟/小时聚合。
  - 网关运行报表（Supabase 内）：
    - 表：`gateway_nodes`、`gateway_heartbeats`（见 `doc_chat/create_gateway_reporting.sql`）。
    - Node 网关周期写入心跳与计数，供看板与告警使用。
  - push-notify-worker（个推）：
    - 通过 Supabase Webhook/Edge Function 或 Kafka chat.push 触发，调用个推 API 下发推送。

## 2. 事件/消息模型

采用统一 envelope（参考 CloudEvents 思想）：

```json
{
  "id": "<uuid>",              // 客户端生成/网关补齐（用于幂等）
  "ts": "2025-09-23T10:00:00Z", // ISO 时间
  "type": "chat.message",       // 事件类型
  "source": "mqtt|web|server",  // 来源
  "conversation_id": "...",
  "sender_id": "...",
  "content": "...",            // 文本或 URL（音频/文件）
  "content_type": "text|audio|image|file|markdown|json",
  "metadata": {"duration_ms": 1234, "mime": "audio/mpeg", "size": 102400}
}
```

- 客户端发送到 MQTT 主题时携带此结构；
- 网关校验/补齐后落库为 chat_messages 的一行（content/metadata 等字段对应）。

另：健康/传感时序（Telemetry）建议采用“轻 envelope + 列存模型”的二段式：

```json
{
  "id": "<uuid>",
  "ts": "2025-09-23T10:00:00Z",
  "type": "ts.metric",
  "user_id": "...",
  "device_id": "...",
  "metric": "hr|spo2|temp|bp_systolic|bp_diastolic|steps",
  "value": 72.0,
  "meta": { "unit": "bpm", "src": "ble" }
}
```
由网关投递至 Kafka `ts.health`，并落地 ClickHouse `health_raw`（示例见 `doc_chat/kafka_mqtt_clickhouse.md`）。

## 3. 主题与分区规划（MQTT/Kafka）

- MQTT 主题
  - chat/send/{conversationId}：客户端→服务端，发送消息入口。
  - chat/recv/{conversationId}：服务端→客户端，转发已确认写库的消息（可选）。
  - presence/{userId}：在线状态（使用 Last Will 设置离线告警）。
  - ack/{messageId}/{userId}（可选）：消息已达/已读 ACK 路由。
  - 建议 QoS1（至少一次），禁用 retain（聊天消息不保留），presence 可保留。
  - ACL：
    - 仅参与者可发布 chat/send/{conversationId}；
    - 仅参与者可订阅 chat/recv/{conversationId}；
    - 用户仅能发布/订阅自身 presence/{userId} 与 ack 自己路由。
- Kafka 主题（可选）
  - chat.inbound（key=conversationId）：保证同会话有序；
  - chat.persisted：写库成功后的广播，用于下游计算或转发；
  - chat.push：推送任务。
  - ts.health（key=user_id 或 device_id）：健康/传感数据入口，供 ClickHouse 消费。

## 4. 后端流程设计

```text
[Client/Device]
   |  MQTT publish chat/send/{conversationId}
   v
[gateway-mqtt]
   |  校验/标准化/幂等
   |--(A) 直写 Supabase PostgREST/Edge Function --> [Postgres]
   |--(B) 投递 Kafka chat.inbound ---------------> [persist-worker -> Postgres]
                                      |(trigger)  -> chat_notifications
                                      |(optional) -> MQTT chat/recv/{conversationId}
                                      |            -> Kafka chat.push
                                      v
                                  [push-notify-worker -> 个推]

---- Telemetry 时序流 ----
  |--(T1) 网关将时序上报写入 Kafka ts.health ------> [ClickHouse 明细表 health_raw]
  |--(T2) 物化/降采样视图做分钟/小时聚合 ----------> [Grafana 可视化]
  |--(T3) 最近快照/告警回写 Supabase（可选） ------> [user_metrics_last / alerts]
```

### 4.1 MQTT → Supabase（入口）

1) 客户端/设备 发布到 chat/send/{conversationId}（QoS1）。
2) gateway-mqtt 收到消息：
   - 校验 JWT/设备证书，鉴权是否会话参与者；
   - 解析 envelope，若无 id 则补 uuid；
   - 幂等检查（可选：Redis setNx(messageId) 窗口 24h）。
3) 落库：
   - 方案 A：直写 Supabase PostgREST（chat_messages.insert）或 RPC。
   - 方案 B：写入 Kafka chat.inbound → persist-worker 落库（高峰更稳）。
4) Postgres 触发器 chat_on_message_insert：
   - 更新 chat_conversations.last_message_at；
   - 插入 chat_notifications（除发送者外所有参与者）。
5) 可选：服务端再发布到 MQTT chat/recv/{conversationId}（供仅连 MQTT 的轻端即时收到），或依赖 Supabase Realtime 给连 H5/App 的用户推送。

### 4.2 Supabase → 个推（通知）

- 触发方式：
  - Edge Function 订阅 Postgres Changes（或 Supabase Webhook）监听 chat_notifications insert；
  - 或 Kafka chat.push 消费者。
- push-notify-worker：
  - 查询 userId 的个推 cid；
  - 生成通知文案（可带会话标题、消息摘要、角标）；
  - 调用个推 API 下发；记录投递结果（可回写到一个 delivery_logs 表）。

### 4.3 前端发送（chat/index）

1) 音频/附件：先 S3 预签名上传，获得 URL 与 metadata。
2) 生成消息 id（uuid），组装 envelope。
3) 并行/串行两路：
   - 发布到 MQTT chat/send/{conversationId}（低延迟通知网关/其他 MQTT 客户端）。
   - 调用 Supabase 插入 chat_messages（确保落库）。
4) UI 乐观更新；若落库失败则回滚/重试。

### 4.4 前端接收

- H5/App（连 Supabase Realtime）：
  - 订阅 chat_messages（INSERT）与 chat_notifications（INSERT）。
  - 房间内做去重（根据 id，已在 room.uvue 实现）。
- 仅 MQTT 客户端：
  - 订阅 chat/recv/{conversationId} 主题获取已确认消息（可由网关回发）。

### 4.5 Telemetry → ClickHouse（健康/传感数据）

1) 设备/网关 MQTT 上报（可以复用同一 broker，不同主题前缀如 `ts/send/{deviceId}`）。
2) gateway-mqtt 将清洗后的数据写入 Kafka `ts.health`（key 使用 user_id 或 device_id 保证局部有序）。
3) ClickHouse 落地：明细表 `health_raw`（MergeTree，按月份分区，TTL 保留），并建立 1m/1h 物化视图。
4) 前端时序图：Grafana 直连 ClickHouse；应用内查询可通过后端代理（统一鉴权，参见 `kafka_mqtt_clickhouse.md`）。
5) 最近快照/告警回写 Supabase（`user_metrics_last`/`alerts`），与业务数据/权限体系对齐。

## 5. 推送（个推）策略

- 触发：插入 chat_notifications 后对目标用户触发推送。
- 去重：个推透传携带 message_id/conversation_id，客户端去重。
- 点击动作：深链到 pages/sport/chat/room?conversation_id=xxx。
- 频率限制：同会话短时多条合并为“你有 X 条新消息”。
- 前台/后台判定：前台不推或用本地通知；后台/离线才用个推。

## 6. 安全与权限

- MQTT 认证：JWT（携带 user_id、过期时间）或 TLS 证书；ACL 映射会话参与者。
- Supabase RLS：已基于 chat_participants；确保 Edge Function 走 service role 仅内部调用。
- S3 附件：仅用预签名上传；公有读或 CDN 鉴权按需配置。
- Idempotency：message_id 唯一约束（Postgres 可加唯一索引），网关 Redis 幂等窗。
 - Telemetry：ClickHouse 作为原始明细库不直接暴露到前端；前端通过后端代理或预聚合接口访问，鉴权仍以 Supabase 用户为中心。

## 7. 可靠性与一致性

- 顺序：以 conversationId 为 key 的 Kafka 分区（或直接依赖单库顺序 + created_at 排序）。
- 重试：
  - 前端 AkReq 已支持网络重试；
  - 网关→Supabase 写库失败使用指数退避重试；
  - push 失败重试并退避；
- 断线：
  - MQTT QoS1 + Clean Session；
  - 前端 Realtime 自动重连（建议在 AkSupaRealtime 增加重连与订阅恢复）。
- 未读计数：
  - 客户端进入会话更新 chat_participants.last_read_at；
  - 会话列表通过 last_message_at 与 last_read_at 计算角标（服务端或客户端）。
 - 网关运行健康：
   - Node 网关定期上报 `gateway_heartbeats`（CPU/内存/连接状态/消息计数/错误等）。
   - 提供 `gateway_status_latest` 视图与 `gateway_daily_stats` 物化视图做看板/告警。

## 8. 与现有表结构的映射

- chat_messages.content 存文本或 S3 URL；content_type 包含 "audio"（已支持）。
- chat_on_message_insert 已生成 chat_notifications，供 Realtime 与个推使用。
- 参与者列表通过 chat_participants 查询；权限与 RLS 已覆盖。
 - 网关运行报表：
   - `gateway_nodes`：注册网关（以 `mqtt_client_id` 作为唯一标识）。
   - `gateway_heartbeats`：周期心跳与计数；`gateway_status_latest` 展示最近一次状态。
   - SQL 见 `doc_chat/create_gateway_reporting.sql`。

## 9. 前端实现要点（结合现有代码）

- 单 WS 连接多订阅：`ChatDataService.ensureRealtime()` 已实现。
- 房间消息去重：`room.uvue` 已按 id 去重。
- 录音与上传：`AudioUploadService` 预签名 + multipart，返回 URL 后调用 `sendAudioMessage()`。
- 发送路径：保持 MQTT 发布 + Supabase 插入的“双写”，若担心两者不一致，可切换为“仅插库 → 后端转发到 MQTT”。
- 订阅释放：路由切换时调用 `dispose()` 与 `closeRealtime()` 清理。
 - Telemetry 展示：趋势/历史直接走 ClickHouse（Grafana 或后端代理 API）；业务侧角标/卡片展示用 Supabase 快照表（低延迟）。

## 10. 运维与监控

- 网关与推送服务加 Prometheus 指标：QPS、失败率、延时、重试数。
- Kafka Lag 监控（若使用 Kafka）。
- Postgres 指标与 Realtime 通道数监控。
- ClickHouse：插入延迟、分区大小、查询耗时、物化视图滞后；Grafana 有现成数据源。
- 个推调用成功率，退避与熔断策略。
- 统一 compose 部署（见 `doc_chat/docker-compose.yml`）：`kafka`、`redis`、`clickhouse`、`grafana`、`mqtt` 一体化联调。

## 11. 最小可行落地顺序

1) 网关直写 Supabase（跳过 Kafka），完成 MQTT → DB → Realtime 闭环；
2) 接入个推：从 chat_notifications 插入触发；
3) 加上 MQTT 回发（chat/recv）供仅 MQTT 客户端；
4) 引入 Kafka 做解耦与扩展（如需要）；
5) 增强 Realtime 重连与订阅恢复；完善未读角标与已读回执。
6) Telemetry 引入 ClickHouse：接入 Kafka `ts.health` → 明细表 + 物化视图；前端改走 ClickHouse/Grafana。

## 12. 配置与文件索引（本仓库）

- Docker 一体化：`doc_chat/docker-compose.yml`
- MQTT Broker 配置：`doc_chat/mosquitto/config/mosquitto.conf`
- 网关服务：
  - 代码：`server/gateway-mqtt-node/src/index.js`（MQTT→Kafka/Supabase、心跳上报）
  - Worker：`server/gateway-mqtt-node/src/worker.js`（Kafka chat.inbound → Supabase 持久化）
  - 环境模板：`server/gateway-mqtt-node/.env.example`
- 聊天表结构与触发器：`doc_chat/create_chat_tables.sql`
- 网关运行报表：`doc_chat/create_gateway_reporting.sql`
- 时序选型与 ClickHouse 方案：`doc_chat/kafka_mqtt_clickhouse.md`

---
如需我生成网关示例（Node.js/Go，鉴权+插库）或 Supabase Edge Function 模板（监听 notifications → 个推），告诉我技术栈偏好即可。

---

## 附录：按会话划分 Supabase Realtime 通道方案

### 背景

- 现状：前端通过单一 `chat_messages` 订阅获取所有对话的 INSERT 事件，再由客户端根据 `conversation_id` 做过滤。
- 问题：
  - 参加多个会话的用户会收到不相关会话的通知，虽然 UI 再过滤，但仍产生额外的网络和序列化开销。
  - 如果后续需要对同一个 Supabase Realtime 连接做权限隔离（例如服务端使用 service key 代理订阅），需要显式限定查询范围。

### 设计目标

1. 每个会话使用独立的 Realtime channel，通过 Postgres Changes 的 `filter` 或 `event` 参数仅推送该会话的数据。
2. 在保持单连接多订阅的基础上，动态创建/关闭频道，避免过多 WebSocket 连接。
3. 兼容移动端（uni-app X）与 Web，同步维护订阅生命周期。

### 频道命名与过滤

| 场景 | Supabase channel 名称 | Postgres Changes 配置 |
|------|------------------------|------------------------|
| 会话消息 | `chat:msg:<conversationId>` | `{ event: "INSERT", schema: "public", table: "chat_messages", filter: "conversation_id=eq.<conversationId>" }` |
| 会话通知 | `chat:notify:<conversationId>`（可选） | `{ event: "INSERT", schema: "public", table: "chat_notifications", filter: "conversation_id=eq.<conversationId>" }` |

> 命名规则保留 `chat:` 前缀，方便统一管理；`conversationId` 建议做 URL encode 避免特殊字符。

### 前端改造步骤

1. **订阅管理器**：在 `ChatDataService.ensureRealtime()` 基础上，增加 `ensureConversationChannel(conversationId)`：
   - 若已有 `convChannels[conversationId]` 直接返回。
   - 否则调用 `realtime.channel(name)`，并使用 `channel.on('postgres_changes', {...})` 设置监听。
   - 调用 `channel.subscribe()`，成功后写入 Map；失败时重试或回退为 table 级订阅。
2. **进入会话**：`room.uvue` 调用 `ChatDataService.subscribeMessages(conversationId)` 时，改为：
   - `ensureConversationChannel(conversationId)`。
   - 在 channel 回调里只派发该会话的消息（无需再过滤）。
3. **退出会话**：在 `onUnload/onHide` 调用 `ChatDataService.releaseConversationChannel(conversationId)`：
   - 取消监听，调用 `channel.unsubscribe()` 并从 Map 移除。
   - 可设置闲置超时，避免频繁退出/进入导致频繁订阅。
4. **多会话同时打开**（例如侧边栏预览）：允许同时保持多个频道订阅，根据 UI 激活状态决定是否真正渲染进入消息队列。

### 权限与安全

- 使用匿名/终端用户 token 连接 Realtime 时，RLS 会自动限制仅返回参与者可访问的 `conversation_id`。即便如此，按会话过滤可以进一步降低服务端负载。
- 若使用 service key 统一代理订阅（例如后端服务转发消息），务必在 channel 过滤项中带上目标会话 ID，避免泄露。

### 服务器端同步策略

- 若网关或中间层也监听 Realtime，可采用同样的 channel 命名规范，或通过 `joinTopicIfNeeded('chat:msg:<id>', payload)` 方式统一管理。
- Channel 生命周期可和应用级订阅保持一致，并在服务器端维护引用计数，最后一个消费者离开时再真正 `unsubscribe()`。

### 回退策略

- 如果遇到 Supabase 版本限制无法使用 `filter`，可以退回到表级订阅，但保留 channel 命名（仍使用 `chat:msg:<id>`），在客户端回调内做过滤。这样可以平滑升级。
- 同理，在网络不佳时可侦测订阅失败并自动切换到现有的“全量监听 + 客户端过滤”。

---
该方案不会改变已有聊天流程，只是优化 Realtime 的订阅粒度，实现“每个会话一个 channel”。配合现有的缓存/去重逻辑可直接落地，下一步可在 `ChatDataService` 中实现上述方法并编写单元测试覆盖多订阅场景。