PolyWeather/docs/superpowers/specs/2026-05-27-redis-stream-realtime-event-architecture-design.md

Redis Stream Realtime Event Architecture Design

日期: 2026-05-27 范围: PolyWeather 网站图表实时观测事件层 目标服务器: 2 vCPU / 8 GB RAM / 50 GB 系统盘 实施状态: 2026-05-28 已落地为 v1.8.1

背景

当前实时层已经具备生产化雏形：

• 后端已有 city_observation_patch.v1 schema。
• web/realtime_event_store.py 使用 SQLite 保存 observation_patch_events，支持 since_revision replay。
• /api/events?cities=...&since_revision=...&replay_limit=... 已经是前端 SSE 入口。
• 前端 frontend/hooks/use-sse-patches.ts 使用 EventSource，并按当前可见城市订阅。

这解决了单进程、单实例下的实时推送和短窗口 replay，但还不够“一步到位”：

• 多 worker / 多实例时，进程内广播不能共享。
• SQLite replay log 可以补数据，但不适合作为跨实例 live fanout。
• 服务重启、浏览器后台恢复、SSE 断线后，需要更稳定的事件源。
• 后续如果把采集器拆成独立进程，需要一个明确的后端事件总线。

目标

升级为“Redis Stream 主事件日志 + SSE 浏览器传输 + SQLite fallback”的架构：

• Redis Stream 是生产环境实时事件主通道。
• SSE 仍然是浏览器唯一实时入口。
• 前端协议保持 city_observation_patch.v1，不让浏览器感知 Redis。
• 继续使用 numeric revision，兼容现有前端 since_revision 逻辑。
• 支持按城市订阅、断线 replay、多 worker fanout。
• 事件只保留短窗口；当前默认按 Redis Stream MAXLEN 控制，建议以约 24 小时 replay 为目标，不做 72 小时保留。

非目标

这次不做以下事情：

• 不引入 Kafka、RabbitMQ 或复杂 consumer group。
• 不把天气历史数据全部 event sourcing。
• 不让浏览器直接连接 Redis。
• 不把 DEB 预测、模型曲线、概率分布都改成事件流。
• 不做永久行情归档；超过 replay 窗口时继续走 HTTP full detail resync。

推荐架构

flowchart LR
    Collector["WeatherDataCollector / city refresh"] --> Ingest["POST /api/internal/collector-patch"]
    Ingest --> Normalize["city_observation_patch.v1 normalizer"]
    Normalize --> Store["RealtimeEventStore interface"]
    Store --> Redis["Redis Stream: stream:city_observation"]
    Store --> SQLite["SQLite fallback: observation_patch_events"]
    Redis --> WorkerSub["Redis XREAD subscriber per API worker"]
    WorkerSub --> LocalFanout["Local SSE fanout"]
    SQLite --> ReplayFallback["SQLite replay fallback"]
    SSE["GET /api/events?cities=...&since_revision=..."] --> Browser["Frontend EventSource"]
    LocalFanout --> SSE
    Redis --> SSE
    ReplayFallback --> SSE
    Browser --> Chart["Chart applies patch"]
    Browser -->|resync_required| Detail["HTTP /api/city/{city}/detail"]

关键原则：

• Redis Stream 负责后端事件保存、短窗口 replay、多实例共享。
• SSE 负责服务端到浏览器的单向推送。
• SQLite 只做 fallback 和本地开发，不再作为生产主事件总线。
• 前端只认 patch event，不关心事件来自 Redis 还是 SQLite。

事件模型

继续使用现有 schema：

{
  "type": "city_observation_patch.v1",
  "revision": 12345,
  "city": "taipei",
  "source": "cwa",
  "obs_time": "2026-05-27T10:00:00+08:00",
  "observed_at_utc": "2026-05-27T02:00:00Z",
  "observed_at_local": "2026-05-27T10:00:00+08:00",
  "city_local_date": "2026-05-27",
  "city_timezone": "Asia/Taipei",
  "source_cadence_sec": 600,
  "ts": 1780000000000,
  "payload": {
    "temp": 34.2,
    "max_so_far": 34.2,
    "station_code": "466920",
    "station_label": "中央气象署台北站",
    "series_key": "settlement",
    "unit": "celsius"
  }
}

revision 仍然是前端公开排序键，必须是全局单调递增整数。

Redis Stream 自身的 stream ID，例如 1780000000000-0，只作为后端内部定位信息，不暴露给前端作为主 revision。

Redis Stream 设计

Stream key:

stream:city_observation

Revision key:

counter:city_observation_revision

每条 stream entry 字段：

• revision: 全局递增整数，公开给 SSE/前端。
• type: city_observation_patch.v1
• city: normalized city key。
• source: source code，例如 cwa, metar, amos, amsc_awos。
• obs_time: 原始观测时间。
• payload_json: compact JSON。
• created_at_ms: 写入 Redis 的 UTC epoch ms。
• producer_id: API worker / collector instance id，便于排查。

写入必须通过 Lua 脚本原子完成：

1. INCR counter:city_observation_revision
2. XADD stream:city_observation MAXLEN ~ {maxlen} * revision ... payload_json ...
3. 返回 {revision, stream_id}

原因：

• 避免 INCR 成功但 XADD 失败后造成非必要 gap。
• 让 revision 分配和事件入流成为一个原子操作。
• 保持现有前端 numeric since_revision 不变。

Replay 策略

GET /api/events 收到 since_revision 后：

1. 读取 Redis Stream 最旧 entry，拿到 oldest revision。
2. 如果 since_revision > 0 且 since_revision < oldest_revision - 1，发送 resync_required。
3. 从 Redis Stream 顺序扫描事件。
4. 过滤 revision > since_revision。
5. 过滤 city IN subscribed_cities。
6. 返回最多 replay_limit + 1 条。
7. 如果超过 replay_limit，发送 resync_required，让前端 HTTP resync。

当前事件量很小，单个 stream 顺序扫描 24 小时事件是可以接受的：

• 30 城市 * 1 条/分钟 * 24 小时 = 43,200 条。
• replay 通常只发生在断线重连，不是高频查询。

如果后续城市量扩大，或者确实需要超过 24 小时的 replay，再加 per-city stream 或 Redis sorted index。

Live Fanout

每个 API worker 启动一个 Redis subscriber loop：

XREAD BLOCK 5000 STREAMS stream:city_observation {last_seen_stream_id}

读取到事件后：

1. 解析 city_observation_patch.v1。
2. 按本进程 SSE 连接的 cities subscription 过滤。
3. 推送给本进程匹配的浏览器连接。
4. 更新 worker 内存中的 last_seen_stream_id。

ingest endpoint 写入 Redis 后不直接广播本地连接，由 subscriber loop 统一 fanout。这样避免同一个 worker 上“写入后本地广播一次、subscriber 又广播一次”的重复事件。

前端仍然用 revision 去重。即使极端情况下收到重复 revision，也应忽略旧 revision。

SQLite Fallback

保留当前 SQLite store，但角色调整为：

• 本地开发无需 Redis 时使用。
• Redis 不可用且允许降级时使用。
• 单实例临时运行时可用。

新增环境变量：

POLYWEATHER_EVENT_STORE=redis
POLYWEATHER_EVENT_STORE_FALLBACK=sqlite
POLYWEATHER_REDIS_URL=redis://127.0.0.1:6379/0
POLYWEATHER_REDIS_STREAM_KEY=stream:city_observation
POLYWEATHER_REDIS_STREAM_MAXLEN=50000
POLYWEATHER_PATCH_EVENT_RETENTION_HOURS=24
POLYWEATHER_REDIS_REQUIRED=true

推荐生产策略：

• POLYWEATHER_EVENT_STORE=redis
• POLYWEATHER_REDIS_REQUIRED=true
• Redis 写入失败时 ingest 返回 503，不广播不可 replay 的事件。

推荐本地策略：

• 不配置 Redis，自动使用 SQLite。
• 或 POLYWEATHER_EVENT_STORE=sqlite。

Redis 容量与配置

当前 VPS 可以跑本机 Redis。

建议：

bind 127.0.0.1
protected-mode yes
appendonly yes
appendfsync everysec
maxmemory 512mb
maxmemory-policy noeviction

默认 MAXLEN ~ 50000 足够约 24 小时事件：

• 30 城市 * 1/min * 24h = 43,200 条。
• payload 约 1-3 KB，实际内存取决于 Redis Stream overhead。
• 512 MB Redis maxmemory 对当前事件量足够。

注意：这个事件流不是永久数据仓库。超过 24 小时窗口后，前端应 HTTP resync 当前城市 detail。

模块边界

新增/调整后端接口：

web/realtime_event_store.py

定义统一接口：

class RealtimeEventStore:
    def append_event(self, event: dict) -> dict: ...
    def replay_events(self, *, cities: set[str] | None, since_revision: int, limit: int) -> list[dict]: ...
    def replay_requires_resync(self, *, cities: set[str] | None, since_revision: int, replay_count: int, limit: int) -> bool: ...
    def latest_revision(self) -> int: ...

保留现有 SQLite 实现，新增 Redis 实现。

web/redis_realtime_event_store.py

职责：

• Redis 连接与健康检查。
• Lua append script。
• Redis Stream replay。
• maxlen trim。
• entry 到 SSE event 的反序列化。

web/sse_manager.py

职责：

• 管理本进程 SSE 连接。
• 记录每个连接订阅城市集合。
• 将 Redis subscriber 读到的 event fanout 给匹配连接。

web/routers/sse_router.py

职责：

• /api/internal/collector-patch 规范化并写入 event store。
• /api/events 处理 connected、replay、heartbeat、live stream。
• 当 replay 不完整时发送 resync_required。

前端

前端原则上不需要大改：

• use-sse-patches.ts 继续使用 EventSource。
• lastRevision 仍然是 number。
• resync_required 继续触发当前可见城市 HTTP detail refresh。
• 继续按可见城市列表构建 cities 参数。

运行流程

首屏

1. 前端 HTTP 加载 terminal rows。
2. 可见图表 HTTP 加载 full detail。
3. 前端连接 /api/events?cities=...&since_revision=...&replay_limit=500。
4. 后端先 replay Redis Stream 中缺失事件，再进入 live stream。

新观测

1. 采集器产生 city_patch 或 v1 patch。
2. ingest endpoint 规范化成 city_observation_patch.v1。
3. Redis Lua script 写入 stream 并生成 revision。
4. 每个 API worker 的 subscriber loop 读到事件。
5. worker fanout 给订阅对应城市的 SSE 连接。
6. 前端 apply patch，图表无痛追加点。

浏览器后台恢复

1. EventSource 如果断线，前端按指数退避重连。
2. URL 带 since_revision=lastRevision。
3. Redis replay 补齐后台期间错过的事件。
4. 如果 revision 太旧或 replay 超限，前端收到 resync_required，HTTP 重拉 full detail。

错误处理

• Redis append 失败且 POLYWEATHER_REDIS_REQUIRED=true：ingest 返回 503，不广播。
• Redis append 失败且允许 fallback：写入 SQLite，并在 health 状态标记 degraded。
• Redis replay 失败：SSE 发送 resync_required，然后继续尝试 live stream。
• Redis subscriber loop 断开：指数退避重连；期间 SSE heartbeat 仍可维持连接，但不会有 live patch。
• 前端收到 revision 倒退或重复：忽略。
• 前端收到未知 event type：忽略并记录 debug。

监控与诊断

新增健康指标：

• 当前 event store mode: redis / sqlite / degraded_sqlite
• Redis ping latency
• latest revision
• stream length
• oldest revision
• subscriber connected
• SSE active connection count
• dropped/resync_required count

建议暴露到现有 health endpoint 或日志：

{
  "realtime": {
    "store": "redis",
    "redis_connected": true,
    "stream_len": 43210,
    "latest_revision": 123456,
    "oldest_revision": 80200,
    "subscriber_connected": true,
    "sse_connections": 9
  }
}

测试策略

后端：

• Redis store append 会生成全局递增 numeric revision。
• Redis replay 支持 city 过滤。
• Redis replay 支持 since_revision。
• Redis replay 超出 retention/limit 返回 resync required。
• Redis 不可用时，根据 env 严格失败或 fallback SQLite。
• SQLite store 现有测试保持通过。
• SSE router 在 Redis store 下仍发送 connected、replay、heartbeat。

前端：

• use-sse-patches.ts 保持 revision numeric。
• 重连 URL 保留 since_revision。
• 收到 resync_required 会触发 visible city refresh。
• 重复 revision 不重复追加图表点。

验收命令：

python -m pytest tests/test_realtime_patch_schema.py tests/test_realtime_event_store.py tests/test_sse_replay.py
python -m pytest tests/test_redis_realtime_event_store.py
cd frontend
npm run test:business
npm run typecheck
npm run build

迁移步骤

1. 保留现有 SQLite implementation，抽出 event store factory。
2. 新增 Redis implementation 和单元测试。
3. 新增 Redis subscriber loop，但先在本地开发环境跑。
4. /api/events 接入 event store factory。
5. 线上安装 Redis，只监听 127.0.0.1。
6. 先用 POLYWEATHER_EVENT_STORE=redis、POLYWEATHER_REDIS_REQUIRED=false 灰度。
7. 观察 stream length、latest revision、SSE error、resync count。
8. 稳定后切换 POLYWEATHER_REDIS_REQUIRED=true。
9. 保留 SQLite fallback 代码，但生产不主动降级，避免多实例状态分裂。

验收标准

• 断开浏览器网络 1-5 分钟后恢复，图表自动补齐期间 patch。
• 后台挂页面再回来，图表不需要等待 full loading 才更新当前点。
• 多 worker 下，任意 worker 写入的 patch 都能推到其他 worker 的 SSE 连接。
• /api/events?cities=taipei,shanghai&since_revision=... 只 replay 指定城市。
• Redis 重启后，前端收到 resync_required 并 HTTP 重建画面。
• 前端无需知道 Redis 存在。
• 现有 city_observation_patch.v1 schema 不破坏。

结论

这是一种“一步到位但不过度设计”的方案：

• 一步到位的是事件契约、replay、跨实例 fanout 和生产部署边界。
• 不过度设计的是不用 Kafka、不做永久行情库、不重写前端协议。

对当前产品最关键的是：图表像股票行情一样无痛刷新，断线可补，后台恢复可追上，多实例以后不用再推倒实时层。