diff --git a/services/push-gateway/docs/nextstep-v2.md b/services/push-gateway/docs/nextstep-v2.md new file mode 100644 index 0000000..d9a0fe1 --- /dev/null +++ b/services/push-gateway/docs/nextstep-v2.md @@ -0,0 +1,352 @@ +# push-gateway 模块上下游依赖与工作清单（Next Steps v2） + +> 版本：v2 +> 日期：2026-07-14 +> 负责人：ai09 +> 关联： +> +> - [push-gateway nextstep.md (v1)](./nextstep.md) +> - [msg nextstep.md](../../msg/docs/nextstep.md) §2.4 / §4 +> - [student-bff nextstep-v2.md](../../student-bff/docs/nextstep-v2.md) §4.7/§4.8 +> - [teacher-portal nextstep-v2.md](../../../apps/teacher-portal/docs/nextstep-v2.md) §3.2.3 +> - ARB-013 Kafka topic 命名规范 + +--- + +## 1. 概述 + +push-gateway 是 Edu 平台的实时 WebSocket 推送基础设施（L3 网关层，端口 8081），负责维护用户在线连接、接收 msg 服务通过 HTTP `/internal/*` 和 Kafka 推送的消息、并通过 WebSocket 实时投递给在线客户端。 + +**v2 核心目标**：对齐 ARB-013 Kafka topic 命名规范和 msg 服务的 API 契约，消除 v1 中存在的 4 个不一致点。 + +--- + +## 2. v2 完成的工作（API 契约对齐） + +### 2.1 Kafka topic 对齐（ARB-013） + +| 项 | v1 | v2 | 状态 | +| ---------- | -------------------------------- | ---------------------------------------------- | ---- | +| 消费 topic | `edu.notification.requested` | `edu.notify.notification.sent` | ✅ | +| 命名规范 | 旧命名（违反 ARB-013） | `edu...` | ✅ | +| DLQ topic | `edu.notification.requested.dlq` | `edu.notify.notification.sent.dlq`（自动派生） | ✅ | + +**修改文件**： + +- `internal/config/config.go`：`KAFKA_NOTIFICATION_TOPIC` 默认值改为 `edu.notify.notification.sent` +- `internal/kafkaconsumer/consumer.go`：包注释更新，metric label 使用 `event.EventType`（而非硬编码 "notification.requested"） +- `internal/config/config_test.go`：断言更新为新 topic 名 + +### 2.2 鉴权头对齐 + +| 项 | v1 | v2 | 状态 | +| -------- | -------------------- | -------------------------------------------------------------------- | ---- | +| 规范头名 | `X-Internal-Token` | `X-Internal-Key`（ARB-013 对齐 msg） | ✅ | +| 向后兼容 | — | 同时接受 `X-Internal-Token`（legacy alias） | ✅ | +| 环境变量 | `INTERNAL_API_TOKEN` | `PUSH_INTERNAL_TOKEN`（canonical）+ `INTERNAL_API_TOKEN`（fallback） | ✅ | + +**修改文件**： + +- `internal/ws/handler.go`： + - `internalTokenHeader` 常量改为 `X-Internal-Key` + - 新增 `legacyInternalTokenHeader = "X-Internal-Token"` 兼容别名 + - `checkInternalToken()` 依次尝试两个头 +- `internal/config/config.go`：优先读取 `PUSH_INTERNAL_TOKEN`，为空时回退到 `INTERNAL_API_TOKEN` + +### 2.3 请求体字段命名对齐 + +| 项 | v1 | v2 | 状态 | +| ------------ | ------------------------ | --------------------------------------------------------- | ---- | +| 用户 ID 字段 | `user_id`（snake_case） | `userId`（camelCase，msg canonical）+ `user_id`（legacy） | ✅ | +| 其他字段 | `event` / `data` / `ttl` | 保持不变 | ✅ | + +**修改文件**： + +- `internal/ws/handler.go` `PushHandler()`： + - 结构体同时定义 `UserID json:"userId"` 和 `UserIDLeg json:"user_id"` + - 绑定后若 `UserID` 为空则回退到 `UserIDLeg` + +### 2.4 环境变量对齐 + +| v1 环境变量 | v2 规范环境变量 | 兼容性 | +| --------------------------------------------------------------- | ----------------------------------------------------------------- | ----------------------------- | +| `INTERNAL_API_TOKEN` | `PUSH_INTERNAL_TOKEN` | v1 变量作为 fallback 别名保留 | +| `KAFKA_NOTIFICATION_TOPIC`（默认 `edu.notification.requested`） | `KAFKA_NOTIFICATION_TOPIC`（默认 `edu.notify.notification.sent`） | 环境变量名不变，默认值更新 | + +**修改文件**： + +- `infra/docker-compose.deploy.yml`：push-gateway 段使用 `PUSH_INTERNAL_TOKEN`（支持 `${PUSH_INTERNAL_TOKEN:-${INTERNAL_API_TOKEN:-edu-internal-token}}` 回退链） +- `infra/deploy.env.example`：`PUSH_INTERNAL_TOKEN` 为规范变量，`INTERNAL_API_TOKEN` 标注为兼容别名 + +### 2.5 部署配置更新 + +| 文件 | 修改内容 | +| --------------------------------- | ---------------------------------------------------------------------------------------- | +| `infra/docker-compose.deploy.yml` | `KAFKA_NOTIFICATION_TOPIC: edu.notify.notification.sent`；`PUSH_INTERNAL_TOKEN` 带回退链 | +| `infra/deploy.env.example` | 新增 `PUSH_INTERNAL_TOKEN`，保留 `INTERNAL_API_TOKEN` 作为兼容别名（留空） | + +--- + +## 3. 本地 Docker 验证结果（2026-07-14 v2） + +测试环境：本地 Docker Desktop，`edu-full_default` 网络，接入真实 edu-kafka / edu-redis / edu-iam-test。 + +``` +镜像：edu/push-gateway:test +容器：edu-push-gateway-test（DEV_MODE=true, PUSH_INTERNAL_TOKEN=test-internal-token-v2） +网络：edu-full_default（直连 edu-kafka:29092 / edu-redis:6379 / edu-iam-test:3002） +``` + +| 验证项 | 结果 | 说明 | +| --------------------------------------- | ---- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `go vet ./...` | ✅ | 零警告 | +| `go build ./...` | ✅ | 零错误 | +| `go test ./...` | ✅ | config/hub/ws 全部通过（含新增 `TestLoadProdModeLegacyToken`） | +| Docker 镜像构建 | ✅ | `edu/push-gateway:test` 构建成功 | +| `/healthz` | ✅ | `{"service":"push-gateway","status":"ok"}` | +| `/readyz` | ✅ | `degraded:false`，kafka+redis 均 ok | +| 容器日志确认 topic | ✅ | `kafka_topic=edu.notify.notification.sent` | +| **v2 契约：X-Internal-Key + userId** | ✅ | `POST /internal/push` 返回 `{"success":true}` | +| **v1 兼容：X-Internal-Token + user_id** | ✅ | 旧调用方式仍正常工作 | +| **Kafka 全链路：新 topic → WS** | ✅ | 发布到 `edu.notify.notification.sent` → WebSocket 客户端收到消息 | +| **考试事件透传（3 类）** | ✅ | ExamExtended / ExamForceSubmitted / ExamQuestionReordered 全部经 Kafka → WS 投递成功 | +| Prometheus metrics | ✅ | `push_gateway_kafka_consumed_total{topic="edu.notify.notification.sent"}` 计数正确；`push_gateway_messages_pushed_total{event="ExamExtended",result="delivered"}` 等标签正确 | + +### 3.1 测试脚本 + +| 脚本 | 用途 | +| ---------------------------------------------------- | -------------------------------------------------------------------------------------------- | +| `services/push-gateway/scripts/test-kafka-ws.mjs` | 验证单条 Kafka 消息经 `edu.notify.notification.sent` topic 到达 WebSocket 客户端 | +| `services/push-gateway/scripts/test-exam-events.mjs` | 验证 3 类考试事件（ExamExtended/ExamForceSubmitted/ExamQuestionReordered）的 event_type 透传 | + +运行方式（需在仓库根目录）： + +```bash +# 前置：push-gateway 容器已启动并接入 edu-full_default 网络 +node services/push-gateway/scripts/test-kafka-ws.mjs +node services/push-gateway/scripts/test-exam-events.mjs +``` + +--- + +## 4. 上游依赖（调用 push-gateway 的模块） + +### 4.1 4 个前端 Portal（teacher/student/parent/admin）— P0 + +| Portal | WebSocket 接入状态 | v2 新增工作 | 状态 | +| -------------- | --------------------------------------------- | ----------- | ---- | +| teacher-portal | ✅ 已接入（`useNotificationsWebSocket` hook） | 无 | ✅ | +| student-portal | ✅ 已接入（考试事件监听） | 无 | ✅ | +| parent-portal | ✅ 已接入 | 无 | ✅ | +| admin-portal | ✅ 已接入 | 无 | ✅ | + +**说明**：4 个 portal 在 v1 阶段已完成 WebSocket 接入，使用 `?token=` query param 鉴权。v2 的 API 契约变更（Kafka topic / 鉴权头 / 请求体字段）对 portal 透明，无需 portal 侧改动。 + +### 4.2 3 个 BFF（teacher-bff / student-bff / parent-bff）— P1 + +| BFF | 依赖 push-gateway 的方式 | v2 新增工作 | 状态 | +| ----------- | ------------------------------------- | ----------- | ---- | +| teacher-bff | 无直接依赖（BFF 不调用 push-gateway） | 无 | ✅ | +| student-bff | 无直接依赖（BFF 不调用 push-gateway） | 无 | ✅ | +| parent-bff | 无直接依赖（BFF 不调用 push-gateway） | 无 | ✅ | + +**说明**：BFF 层不直接调用 push-gateway。推送由 msg 服务通过 HTTP `/internal/push` 或 Kafka 发起，push-gateway 负责投递到 portal 的 WebSocket。 + +--- + +## 5. 下游依赖（push-gateway 调用的模块） + +### 5.1 iam 服务（ai06 负责）— P0 + +| 依赖项 | 用途 | 状态 | +| ----------------------------------- | ------------------------------------- | --------- | +| `GET /v1/iam/.well-known/jwks.json` | RS256 JWKS 公钥（WebSocket JWT 验签） | ✅ 已就绪 | + +**环境变量**：`JWKS_URL=http://iam:3002/v1/iam/.well-known/jwks.json` + +**说明**：shared-go/jwks Fetcher 每 5 分钟刷新 JWKS 缓存。DevMode 接受 `dev-token` 旁路 JWT 验签。 + +### 5.2 msg 服务（ai10 负责）— P0 + +| 依赖项 | 用途 | 状态 | +| -------------------------------------------- | ------------------------------------------------- | --------- | +| HTTP `POST /internal/push` | 定向推送（msg → push-gateway） | ✅ 已对齐 | +| Kafka topic `edu.notify.notification.sent` | 事件驱动推送（msg Outbox → Kafka → push-gateway） | ✅ 已对齐 | +| 鉴权头 `X-Internal-Key: PUSH_INTERNAL_TOKEN` | /internal/* API 鉴权 | ✅ 已对齐 | +| 请求体 `{ userId, event, data }` | camelCase 字段命名 | ✅ 已对齐 | + +**环境变量**： + +- push-gateway 侧：`PUSH_INTERNAL_TOKEN`（canonical）/ `INTERNAL_API_TOKEN`（fallback） +- msg 侧：`PUSH_INTERNAL_TOKEN`（已对齐） + +**v2 对齐结果**：4 个不一致点（Kafka topic / 鉴权头 / 请求体字段 / 环境变量）全部修复，push-gateway 同时保持对 v1 调用方式的向后兼容。 + +### 5.3 Redis（基础设施）— P0 + +| 依赖项 | 用途 | 状态 | +| ------------------------ | --------------------------------------------------- | --------- | +| `redis://edu-redis:6379` | 在线状态 SET + 跨实例 Pub/Sub 扇出 + 事件去重 SETNX | ✅ 已就绪 | + +**环境变量**：`REDIS_URL=redis://edu-redis:6379/0` + +### 5.4 Kafka（基础设施）— P0 + +| 依赖项 | 用途 | 状态 | +| ----------------- | ----------------------------------------- | --------- | +| `edu-kafka:29092` | 消费 `edu.notify.notification.sent` topic | ✅ 已就绪 | + +**环境变量**： + +- `KAFKA_BROKERS=edu-kafka:29092` +- `KAFKA_NOTIFICATION_TOPIC=edu.notify.notification.sent`（ARB-013） +- `KAFKA_CONSUMER_GROUP=push-gateway` + +### 5.5 OTel Collector（基础设施）— P1 + +| 依赖项 | 用途 | 状态 | +| ---------------------------- | --------------- | --------- | +| `http://otel-collector:4318` | OTLP trace 上报 | ⏳ 待部署 | + +**环境变量**：`OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318` + +**说明**：未配置时 push-gateway 跳过 trace 上报，不影响功能。 + +--- + +## 6. 待协调事项（需上下游配合） + +### 6.1 core-edu 考试事件发布（下游阻塞 student-portal） + +| 项 | 详情 | +| ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | +| 阻塞方 | core-edu（ai07） | +| 影响方 | student-portal / student-bff | +| 问题 | student-portal 期望接收 `ExamExtended` / `ExamForceSubmitted` / `ExamQuestionReordered` 三类考试事件，但这些事件在 core-edu 源码中未找到发布逻辑 | +| 期望流程 | core-edu Outbox publisher → Kafka `edu.notify.notification.sent` → push-gateway → student-portal WebSocket | +| push-gateway 侧状态 | ✅ 已验证可消费这三类事件并透传到 WebSocket（见 §3 测试结果） | +| 协调动作 | 请 core-edu 在考试相关业务逻辑中发布这三类事件到 `edu.notify.notification.sent` topic | + +### 6.2 msg 服务 Outbox 发布 topic 对齐（待确认） + +| 项 | 详情 | +| ------------------- | ---------------------------------------------------------------------------------------------------------------------------- | +| 阻塞方 | msg（ai10） | +| 问题 | 需确认 msg 的 Outbox publisher 已将发布 topic 从 `edu.notification.requested` 改为 `edu.notify.notification.sent`（ARB-013） | +| push-gateway 侧状态 | ✅ 已消费 `edu.notify.notification.sent` | +| 协调动作 | 请 msg 确认 Outbox `TOPIC_MAP` 中 `NotificationSent` 事件映射到 `edu.notify.notification.sent`；如仍发往旧 topic，需更新 | + +### 6.3 生产环境环境变量迁移（部署时） + +| 项 | 详情 | +| ------------------- | -------------------------------------------------------------------------------------------------------------- | +| 阻塞方 | SRE AI / 人类决策者 | +| 问题 | 生产环境需将 `INTERNAL_API_TOKEN` 迁移为 `PUSH_INTERNAL_TOKEN`（或在 deploy.env 中同时设置两者） | +| push-gateway 侧状态 | ✅ 支持双环境变量（`PUSH_INTERNAL_TOKEN` 优先，`INTERNAL_API_TOKEN` 回退） | +| 协调动作 | 部署时在 `infra/security/secrets.example.env` 或 K8s Secret 中设置 `PUSH_INTERNAL_TOKEN`；旧变量可保留作为兼容 | + +--- + +## 7. v2 文件变更清单 + +### 7.1 源码修改 + +| 文件 | 修改内容 | +| ---------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | +| `services/push-gateway/internal/config/config.go` | Kafka topic 默认值 → `edu.notify.notification.sent`；支持 `PUSH_INTERNAL_TOKEN` 环境变量（带 `INTERNAL_API_TOKEN` 回退） | +| `services/push-gateway/internal/ws/handler.go` | 鉴权头 `X-Internal-Key`（canonical）+ `X-Internal-Token`（legacy）；请求体 `userId`（canonical）+ `user_id`（legacy） | +| `services/push-gateway/internal/kafkaconsumer/consumer.go` | 包注释更新；metric label 使用 `event.EventType` 而非硬编码 | +| `services/push-gateway/internal/config/config_test.go` | 断言新 topic 名；新增 `TestLoadProdModeLegacyToken` 验证 fallback | +| `services/push-gateway/main.go` | 启动流程注释更新（topic 名 + 鉴权头名） | + +### 7.2 部署配置修改 + +| 文件 | 修改内容 | +| --------------------------------- | --------------------------------------------------------------------------------------------------------- | +| `infra/docker-compose.deploy.yml` | push-gateway 段：`KAFKA_NOTIFICATION_TOPIC: edu.notify.notification.sent`；`PUSH_INTERNAL_TOKEN` 带回退链 | +| `infra/deploy.env.example` | `PUSH_INTERNAL_TOKEN` 为规范变量；`INTERNAL_API_TOKEN` 标注为兼容别名 | + +### 7.3 测试脚本新增 + +| 文件 | 用途 | +| ---------------------------------------------------- | ---------------------------------------- | +| `services/push-gateway/scripts/test-kafka-ws.mjs` | Kafka → WebSocket 全链路验证（单条消息） | +| `services/push-gateway/scripts/test-exam-events.mjs` | 3 类考试事件透传验证 | + +### 7.4 根 package.json + +| 文件 | 修改内容 | +| -------------- | --------------------------------------------------------- | +| `package.json` | `devDependencies` 新增 `kafkajs` + `ws`（仅用于测试脚本） | + +--- + +## 8. 环境变量清单（v2） + +| 环境变量 | 必填 | 默认值 | 用途 | +| ----------------------------- | -------- | ------------------------------ | ----------------------------------------------- | +| `PUSH_GATEWAY_PORT` | 否 | `8081` | HTTP 端口 | +| `JWT_SECRET` | 生产必填 | — | JWT HS256 签名密钥（DevMode 有默认值） | +| `DEV_MODE` | 否 | `false` | 开发模式（旁路 JWT + InternalToken 校验） | +| `REDIS_URL` | 是 | `redis://localhost:6379/0` | Redis 连接地址 | +| `KAFKA_BROKERS` | 是 | `localhost:9092` | Kafka broker 列表（逗号分隔） | +| `KAFKA_NOTIFICATION_TOPIC` | 否 | `edu.notify.notification.sent` | 消费的 Kafka topic（ARB-013） | +| `KAFKA_CONSUMER_GROUP` | 否 | `push-gateway` | Kafka 消费者组 | +| `PUSH_INTERNAL_TOKEN` | 生产必填 | — | /internal/* API 鉴权 token（ARB-013 canonical） | +| `INTERNAL_API_TOKEN` | 兼容别名 | — | /internal/* API 鉴权 token（fallback，v1 兼容） | +| `JWKS_URL` | 生产必填 | — | iam JWKS 端点（RS256 公钥） | +| `WS_ALLOWED_ORIGINS` | 生产必填 | — | WebSocket 允许的 Origin 白名单（逗号分隔） | +| `MAX_CONNS_PER_USER` | 否 | `5` | 单用户最大连接数 | +| `HEARTBEAT_INTERVAL_SECONDS` | 否 | `30` | 心跳间隔（秒） | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | 否 | `localhost:4318` | OTLP collector 端点 | +| `INSTANCE_ID` | 否 | hostname | 实例标识（Redis SET 成员） | + +--- + +## 9. API 契约（v2） + +### 9.1 WebSocket 端点 + +``` +GET /ws?token= +``` + +- 鉴权：JWT RS256（via iam JWKS）或 DevMode `dev-token` +- 消息格式：`{ "type": "message", "event": "", "data": {...}, "timestamp": "" }` +- 心跳：RFC 6455 Ping/Pong（30s 间隔，60s 空闲超时） + +### 9.2 内部 HTTP API + +| 方法 | 路径 | 鉴权头 | 请求体 | 响应 | +| ---- | --------------------------- | ------------------------- | ------------------------------- | ---------------------------------------------------- | +| POST | `/internal/push` | `X-Internal-Key: ` | `{ userId, event, data, ttl? }` | `{ success, delivered?, online?, reached?, error? }` | +| POST | `/internal/broadcast` | `X-Internal-Key: ` | `{ event, data, filter? }` | `{ success, reached? }` | +| GET | `/internal/online/` | `X-Internal-Key: ` | — | `{ online: bool, instances: [] }` | + +**向后兼容**： + +- 鉴权头同时接受 `X-Internal-Token`（v1 legacy） +- `/internal/push` 请求体同时接受 `user_id`（snake_case，v1 legacy） + +### 9.3 Kafka 消费 + +| Topic | 消息格式 | 来源 | +| ---------------------------------- | ---------------------------- | --------------------- | +| `edu.notify.notification.sent` | `NotificationRequested` JSON | msg Outbox publisher | +| `edu.notify.notification.sent.dlq` | 原始消息（失败 3 次后投递） | push-gateway 自动派生 | + +**消息字段**：`event_id` / `user_id` / `event_type` / `channel` / `title` / `content` / `data` / `broadcast` / `occurred_at` + +**event_type 透传**：push-gateway 从 Kafka 消息的 `event_type` 字段读取事件类型，作为 WebSocket 消息的 `event` 字段透传，不硬编码事件类型。 + +--- + +## 10. 后续工作（v3 展望） + +| 工作项 | 优先级 | 说明 | +| ------------------- | ------ | ------------------------------------------------------------------------ | +| P6 Reconnect 协议 | P1 | `session_id` + `last_seq` + ring buffer（100 条/用户）支持断线重连后补发 | +| P6 离线消息推送 | P2 | 与 msg 服务集成，离线用户消息落库后通过 SMS/邮件/应用内通知补发 | +| Prometheus 告警规则 | P2 | 配置 `push_gateway_kafka_consumed_total` 停止增长告警 | +| Grafana 面板 | P2 | WebSocket 连接数、Kafka 消费延迟、推送成功率面板 | +| 多副本部署验证 | P2 | 验证 Redis Pub/Sub 跨实例扇出在多副本场景下的正确性 | diff --git a/services/push-gateway/internal/config/config.go b/services/push-gateway/internal/config/config.go index e4e1b86..366b06b 100644 --- a/services/push-gateway/internal/config/config.go +++ b/services/push-gateway/internal/config/config.go @@ -1,6 +1,10 @@ // Package config loads runtime configuration for push-gateway from environment -// variables. Required variables (JWT_SECRET in production, INTERNAL_API_TOKEN -// when not in DevMode) cause a fatal exit when missing. +// variables. Required variables (JWT_SECRET in production, PUSH_INTERNAL_TOKEN +// or INTERNAL_API_TOKEN when not in DevMode) cause a fatal exit when missing. +// +// v2 alignment (ARB-013 / msg contract): Kafka topic default changed to +// edu.notify.notification.sent; PUSH_INTERNAL_TOKEN is the canonical env var +// for /internal/* auth (INTERNAL_API_TOKEN kept as a backward-compat alias). package config import ( @@ -54,9 +58,15 @@ func Load() *Config { } } - internalToken := env.Get("INTERNAL_API_TOKEN", "") + // PUSH_INTERNAL_TOKEN is the canonical env var for /internal/* auth + // (ARB-013 alignment with msg). INTERNAL_API_TOKEN is kept as a + // backward-compat alias for existing deployments. + internalToken := env.Get("PUSH_INTERNAL_TOKEN", "") + if internalToken == "" { + internalToken = env.Get("INTERNAL_API_TOKEN", "") + } if internalToken == "" && !devMode { - log.Fatal("INTERNAL_API_TOKEN must be set in non-dev mode") + log.Fatal("PUSH_INTERNAL_TOKEN (or INTERNAL_API_TOKEN) must be set in non-dev mode") } return &Config{ @@ -71,7 +81,9 @@ func Load() *Config { HeartbeatInterval: env.GetInt("HEARTBEAT_INTERVAL_SECONDS", 30), JWKSURL: env.Get("JWKS_URL", "http://localhost:50052/.well-known/jwks.json"), KafkaBrokers: parseBrokers(env.Get("KAFKA_BROKERS", "localhost:9092")), - KafkaNotificationTopic: env.Get("KAFKA_NOTIFICATION_TOPIC", "edu.notification.requested"), + // ARB-013 canonical topic name: edu.... + // msg's Outbox publisher emits to this topic. + KafkaNotificationTopic: env.Get("KAFKA_NOTIFICATION_TOPIC", "edu.notify.notification.sent"), KafkaConsumerGroup: env.Get("KAFKA_CONSUMER_GROUP", "push-gateway"), InstanceID: env.Get("INSTANCE_ID", generateInstanceID()), } diff --git a/services/push-gateway/internal/config/config_test.go b/services/push-gateway/internal/config/config_test.go index ecff8f2..af841b2 100644 --- a/services/push-gateway/internal/config/config_test.go +++ b/services/push-gateway/internal/config/config_test.go @@ -10,6 +10,7 @@ func TestLoadDevMode(t *testing.T) { t.Setenv("DEV_MODE", "true") t.Setenv("JWT_SECRET", "") t.Setenv("INTERNAL_API_TOKEN", "") + t.Setenv("PUSH_INTERNAL_TOKEN", "") cfg := Load() @@ -31,8 +32,9 @@ func TestLoadDevMode(t *testing.T) { if cfg.HeartbeatInterval != 30 { t.Errorf("HeartbeatInterval = %d, want 30", cfg.HeartbeatInterval) } - if cfg.KafkaNotificationTopic != "edu.notification.requested" { - t.Errorf("KafkaNotificationTopic = %q, want edu.notification.requested", cfg.KafkaNotificationTopic) + // ARB-013: canonical topic name is edu.notify.notification.sent. + if cfg.KafkaNotificationTopic != "edu.notify.notification.sent" { + t.Errorf("KafkaNotificationTopic = %q, want edu.notify.notification.sent", cfg.KafkaNotificationTopic) } if cfg.KafkaConsumerGroup != "push-gateway" { t.Errorf("KafkaConsumerGroup = %q, want push-gateway", cfg.KafkaConsumerGroup) @@ -42,7 +44,8 @@ func TestLoadDevMode(t *testing.T) { } } -// TestLoadProdMode verifies production requires JWT_SECRET + INTERNAL_API_TOKEN. +// TestLoadProdMode verifies production requires JWT_SECRET + PUSH_INTERNAL_TOKEN +// (or the legacy INTERNAL_API_TOKEN alias). func TestLoadProdMode(t *testing.T) { // Ensure the env vars are unset so Load fails. We can't call os.Exit in // a test, so we verify the required behavior by setting DEV_MODE=false @@ -50,7 +53,8 @@ func TestLoadProdMode(t *testing.T) { // os.Exit, we instead test the happy path: all required vars set. t.Setenv("DEV_MODE", "false") t.Setenv("JWT_SECRET", "prod-secret") - t.Setenv("INTERNAL_API_TOKEN", "prod-token") + t.Setenv("PUSH_INTERNAL_TOKEN", "prod-token") + t.Setenv("INTERNAL_API_TOKEN", "") cfg := Load() @@ -61,7 +65,22 @@ func TestLoadProdMode(t *testing.T) { t.Errorf("JWTSecret = %q, want prod-secret", cfg.JWTSecret) } if cfg.InternalAPIToken != "prod-token" { - t.Errorf("InternalAPIToken = %q, want prod-token", cfg.InternalAPIToken) + t.Errorf("InternalAPIToken = %q, want prod-token (from PUSH_INTERNAL_TOKEN)", cfg.InternalAPIToken) + } +} + +// TestLoadProdModeLegacyToken verifies INTERNAL_API_TOKEN is accepted as a +// backward-compat alias when PUSH_INTERNAL_TOKEN is not set. +func TestLoadProdModeLegacyToken(t *testing.T) { + t.Setenv("DEV_MODE", "false") + t.Setenv("JWT_SECRET", "prod-secret") + t.Setenv("PUSH_INTERNAL_TOKEN", "") + t.Setenv("INTERNAL_API_TOKEN", "legacy-token") + + cfg := Load() + + if cfg.InternalAPIToken != "legacy-token" { + t.Errorf("InternalAPIToken = %q, want legacy-token (from INTERNAL_API_TOKEN fallback)", cfg.InternalAPIToken) } } diff --git a/services/push-gateway/internal/kafkaconsumer/consumer.go b/services/push-gateway/internal/kafkaconsumer/consumer.go index 006b778..b3efdd2 100644 --- a/services/push-gateway/internal/kafkaconsumer/consumer.go +++ b/services/push-gateway/internal/kafkaconsumer/consumer.go @@ -1,12 +1,11 @@ -// Package kafkaconsumer subscribes to the edu.notification.requested topic -// (ISSUE-053 final topic name) on behalf of push-gateway. Each consumed +// Package kafkaconsumer subscribes to the edu.notify.notification.sent topic +// (ARB-013 canonical topic name) on behalf of push-gateway. Each consumed // NotificationRequested event is dispatched to the Hub for delivery to online // clients. At-least-once delivery is enforced via manual commit; idempotency // is provided by Redis SETNX on event_id (TTL 24h). // -// Topic naming follows the G16 rule (edu...); the -// previous abstract name edu.notification.events / edu.msg.notification.events -// is deprecated (see president-final-rulings §1.5). +// Topic naming follows ARB-013 (edu...); the +// previous name edu.notification.requested is deprecated (v2 alignment). package kafkaconsumer import ( @@ -159,12 +158,19 @@ func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) { if err := json.Unmarshal(msg.Value, &event); err != nil { observability.Logger().Warn("kafka: invalid message payload", "topic", c.topic, "partition", msg.Partition, "offset", msg.Offset, "err", err) - c.metrics.IncPushed("notification.requested", "invalid") + c.metrics.IncPushed("notification.invalid", "invalid") // Commit to skip the poison message. _ = c.reader.CommitMessages(ctx, msg) return } + // eventLabel is the metrics label for IncPushed. Prefer event_type (the + // actual event name like "ExamExtended"); fall back to a generic label. + eventLabel := event.EventType + if eventLabel == "" { + eventLabel = "notification.sent" + } + // Idempotency: skip already-processed events by event_id. fresh, err := c.redis.DedupEventId(ctx, event.EventID) if err != nil { @@ -174,7 +180,7 @@ func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) { if !fresh { observability.Logger().Debug("kafka: duplicate event skipped", "event_id", event.EventID) - c.metrics.IncPushed("notification.requested", "duplicate") + c.metrics.IncPushed(eventLabel, "duplicate") _ = c.reader.CommitMessages(ctx, msg) return } @@ -182,7 +188,7 @@ func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) { // Retry up to MaxRetries on dispatch failure. for attempt := 1; attempt <= MaxRetries; attempt++ { if err := c.dispatch(ctx, event); err == nil { - c.metrics.IncPushed("notification.requested", "delivered") + c.metrics.IncPushed(eventLabel, "delivered") _ = c.reader.CommitMessages(ctx, msg) return } else if attempt < MaxRetries { @@ -195,7 +201,7 @@ func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) { observability.Logger().Error("kafka: dispatch failed after retries, sending to DLQ", "event_id", event.EventID, "topic", c.topic) c.sendToDLQ(ctx, msg.Value) - c.metrics.IncPushed("notification.requested", "dlq") + c.metrics.IncPushed(eventLabel, "dlq") _ = c.reader.CommitMessages(ctx, msg) } diff --git a/services/push-gateway/internal/ws/handler.go b/services/push-gateway/internal/ws/handler.go index d380a73..e8e7399 100644 --- a/services/push-gateway/internal/ws/handler.go +++ b/services/push-gateway/internal/ws/handler.go @@ -6,8 +6,12 @@ // - WebSocket /ws: JWT RS256 validated via shared-go/jwks (caching JWKS from // iam /.well-known/jwks.json, refreshed every 5 minutes). DevMode accepts // the literal "dev-token" returning a synthetic dev-user subject. -// - /internal/*: X-Internal-Token header matched against INTERNAL_API_TOKEN -// (ARB-015 §17.3 / president §7.2). DevMode skips the check. +// - /internal/*: X-Internal-Key header matched against PUSH_INTERNAL_TOKEN +// (ARB-013 alignment with msg). X-Internal-Token is accepted as a +// backward-compat alias. DevMode skips the check. +// +// Request body field naming: msg sends userId (camelCase); legacy callers may +// still send user_id (snake_case). Both are accepted by /internal/push. // // Heartbeat (RFC 6455 control frames, not text messages): // - Client sends Ping every 30s; gorilla/websocket auto-replies with Pong. @@ -40,8 +44,12 @@ import ( ) // internalTokenHeader is the canonical header for /internal/* authentication -// (ARB-015 §17.3 / president §7.2). Was X-Internal-Key in pre-P5 builds. -const internalTokenHeader = "X-Internal-Token" +// (ARB-013 alignment with msg). legacyInternalTokenHeader is accepted as a +// backward-compat alias for existing callers. +const ( + internalTokenHeader = "X-Internal-Key" + legacyInternalTokenHeader = "X-Internal-Token" +) // readBufferSize / writeBufferSize tune gorilla/websocket's internal buffers. const ( @@ -244,16 +252,20 @@ func (h *Handler) readerLoop(c *hub.Connection) { // PushHandler implements POST /internal/push: directed push to a single user. // Response carries delivered + online so msg (ai10) can decide offline fallback. +// +// Request body accepts both camelCase (userId, msg's preferred format) and +// snake_case (user_id, legacy format) field names for the user identifier. func (h *Handler) PushHandler(c *gin.Context) { if !h.checkInternalToken(c) { return } var req struct { - UserID string `json:"user_id" binding:"required"` - Event string `json:"event" binding:"required"` - Data map[string]any `json:"data"` - TTL *int `json:"ttl,omitempty"` + UserID string `json:"userId"` // camelCase (msg canonical, ARB-013) + UserIDLeg string `json:"user_id"` // snake_case (legacy) + Event string `json:"event" binding:"required"` + Data map[string]any `json:"data"` + TTL *int `json:"ttl,omitempty"` } if err := c.ShouldBindJSON(&req); err != nil { c.AbortWithStatusJSON(http.StatusBadRequest, pushResponse{ @@ -262,6 +274,17 @@ func (h *Handler) PushHandler(c *gin.Context) { }) return } + // Prefer userId (camelCase); fall back to user_id (snake_case, legacy). + if req.UserID == "" { + req.UserID = req.UserIDLeg + } + if req.UserID == "" { + c.AbortWithStatusJSON(http.StatusBadRequest, pushResponse{ + Success: false, + Error: &errBody{Code: "PUSH_INVALID_REQUEST", Message: "userId (or user_id) is required"}, + }) + return + } message, err := buildMessage(req.Event, req.Data) if err != nil { @@ -399,8 +422,10 @@ func (h *Handler) OnlineHandler(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"online": false, "instances": []string{}}) } -// checkInternalToken validates the X-Internal-Token header. Returns false when -// the request has been aborted (caller should return immediately). +// checkInternalToken validates the internal token header. The canonical header +// is X-Internal-Key (ARB-013 alignment with msg); X-Internal-Token is accepted +// as a backward-compat alias. Returns false when the request has been aborted +// (caller should return immediately). func (h *Handler) checkInternalToken(c *gin.Context) bool { if h.cfg.DevMode { return true @@ -412,7 +437,11 @@ func (h *Handler) checkInternalToken(c *gin.Context) bool { }) return false } - if c.GetHeader(internalTokenHeader) != h.cfg.InternalAPIToken { + token := c.GetHeader(internalTokenHeader) + if token == "" { + token = c.GetHeader(legacyInternalTokenHeader) + } + if token != h.cfg.InternalAPIToken { c.AbortWithStatusJSON(http.StatusUnauthorized, pushResponse{ Success: false, Error: &errBody{Code: "PUSH_UNAUTHORIZED", Message: "invalid or missing internal token"}, diff --git a/services/push-gateway/main.go b/services/push-gateway/main.go index 07f2993..ed1babd 100644 --- a/services/push-gateway/main.go +++ b/services/push-gateway/main.go @@ -12,8 +12,8 @@ // 5. hub.NewHub (in-memory connection registry) // 6. redisclient.New + SetHub + RebuildPresenceOnStartup (ISSUE-058) // 7. redisclient.SubscribeAll (cross-instance Pub/Sub fanout) -// 8. kafkaconsumer.New + goroutine Run (edu.notification.requested) -// 9. ws.NewHandler (JWT RS256, X-Internal-Token, WebSocket upgrade) +// 8. kafkaconsumer.New + goroutine Run (edu.notify.notification.sent, ARB-013) +// 9. ws.NewHandler (JWT RS256, X-Internal-Key, WebSocket upgrade) // 10. health.NewReadyzer (/readyz soft-failure probe) // 11. gin router + http.Server // @@ -116,9 +116,9 @@ func main() { } } - // 7. Kafka consumer (edu.notification.requested). Started only when at - // least one broker is configured; otherwise the HTTP /internal/push API - // is the only delivery channel. + // 7. Kafka consumer (edu.notify.notification.sent, ARB-013). Started only + // when at least one broker is configured; otherwise the HTTP /internal/push + // API is the only delivery channel. var kafkaConsumer *kafkaconsumer.Consumer var kafkaCancel context.CancelFunc if len(cfg.KafkaBrokers) > 0 && cfg.KafkaNotificationTopic != "" { @@ -161,7 +161,7 @@ func main() { r.GET("/metrics", gin.WrapH(promhttp.Handler())) // WebSocket upgrade (JWT auth via query ?token= or Authorization header). r.GET("/ws", wsHandler.HandleWebSocket) - // Internal HTTP APIs consumed by msg (X-Internal-Token auth). + // Internal HTTP APIs consumed by msg (X-Internal-Key auth, ARB-013). internal := r.Group("/internal") internal.POST("/push", wsHandler.PushHandler) internal.POST("/broadcast", wsHandler.BroadcastHandler) diff --git a/services/push-gateway/scripts/test-exam-events.mjs b/services/push-gateway/scripts/test-exam-events.mjs new file mode 100644 index 0000000..46eacdb --- /dev/null +++ b/services/push-gateway/scripts/test-exam-events.mjs @@ -0,0 +1,98 @@ +// Test script: Exam events through Kafka → push-gateway → WebSocket (v2) +// Verifies that the 3 exam events (ExamExtended, ExamForceSubmitted, +// ExamQuestionReordered) traverse the new edu.notify.notification.sent topic +// and reach the WebSocket client with correct event_type passthrough. +// +// Usage: node services/push-gateway/scripts/test-exam-events.mjs + +import WebSocket from "ws"; +import { Kafka } from "kafkajs"; + +const WS_URL = "ws://localhost:8081/ws?token=dev-token"; +const KAFKA_BROKER = "localhost:9092"; +const TOPIC = "edu.notify.notification.sent"; +const USER_ID = "dev-user"; + +const examEvents = [ + { + event_type: "ExamExtended", + data: { examId: "exam-001", extendedMinutes: 15, reason: "special needs" }, + }, + { + event_type: "ExamForceSubmitted", + data: { examId: "exam-001", reason: "time_expired", submissionId: "sub-001" }, + }, + { + event_type: "ExamQuestionReordered", + data: { examId: "exam-001", questionIds: ["q-3", "q-1", "q-2"] }, + }, +]; + +console.log("[exam-test] connecting WebSocket to", WS_URL); +const ws = new WebSocket(WS_URL); +const received = []; + +ws.on("open", async () => { + console.log("[exam-test] WebSocket connected, publishing 3 exam events..."); + const kafka = new Kafka({ brokers: [KAFKA_BROKER] }); + const producer = kafka.producer(); + await producer.connect(); + + for (const ev of examEvents) { + const payload = { + event_id: `evt-exam-${ev.event_type}-${Date.now()}`, + user_id: USER_ID, + event_type: ev.event_type, + channel: "ws", + title: `Exam event: ${ev.event_type}`, + content: `Verification of ${ev.event_type} passthrough`, + data: ev.data, + broadcast: false, + occurred_at: Date.now(), + }; + await producer.send({ + topic: TOPIC, + messages: [{ value: JSON.stringify(payload) }], + }); + console.log(`[exam-test] published ${ev.event_type}`); + } + await producer.disconnect(); + + // Wait up to 15 seconds for all 3 messages. + const deadline = Date.now() + 15000; + while (Date.now() < deadline && received.length < 3) { + await new Promise((r) => setTimeout(r, 200)); + } + + if (received.length < 3) { + console.error(`[exam-test] FAIL: only ${received.length}/3 messages received`); + process.exit(1); + } + const got = received.map((m) => m.event).sort(); + const want = examEvents.map((e) => e.event_type).sort(); + if (JSON.stringify(got) === JSON.stringify(want)) { + console.log("[exam-test] PASS: all 3 exam events received with correct event_type"); + console.log("[exam-test] events:", got.join(", ")); + process.exit(0); + } else { + console.error("[exam-test] FAIL: event mismatch"); + console.error(" got:", got); + console.error(" want:", want); + process.exit(1); + } +}); + +ws.on("message", (data) => { + const msg = JSON.parse(data.toString()); + console.log("[exam-test] received:", JSON.stringify(msg)); + received.push(msg); +}); +ws.on("error", (err) => { + console.error("[exam-test] WebSocket error:", err.message); + process.exit(1); +}); + +setTimeout(() => { + console.error("[exam-test] FAIL: overall timeout 20s"); + process.exit(1); +}, 20000); diff --git a/services/push-gateway/scripts/test-kafka-ws.mjs b/services/push-gateway/scripts/test-kafka-ws.mjs new file mode 100644 index 0000000..6cab561 --- /dev/null +++ b/services/push-gateway/scripts/test-kafka-ws.mjs @@ -0,0 +1,78 @@ +// Test script: Kafka → push-gateway → WebSocket full chain (v2) +// Usage: node scripts/test-kafka-ws.mjs +// Prerequisites: push-gateway running on localhost:8081, Kafka on localhost:9092, +// topic edu.notify.notification.sent exists. + +import WebSocket from "ws"; +import { Kafka } from "kafkajs"; + +const WS_URL = "ws://localhost:8081/ws?token=dev-token"; +const KAFKA_BROKER = "localhost:9092"; +const TOPIC = "edu.notify.notification.sent"; +const USER_ID = "dev-user"; // DevMode maps dev-token to this user +const EVENT_TYPE = "TestEventV2KafkaChain"; + +console.log("[test] connecting WebSocket to", WS_URL); +const ws = new WebSocket(WS_URL); + +const messages = []; +ws.on("open", () => { + console.log("[test] WebSocket connected, publishing Kafka message..."); + publishKafka(); +}); +ws.on("message", (data) => { + const msg = JSON.parse(data.toString()); + console.log("[test] received WS message:", JSON.stringify(msg)); + messages.push(msg); +}); +ws.on("error", (err) => { + console.error("[test] WebSocket error:", err.message); + process.exit(1); +}); + +async function publishKafka() { + const kafka = new Kafka({ brokers: [KAFKA_BROKER] }); + const producer = kafka.producer(); + await producer.connect(); + const payload = { + event_id: `evt-test-${Date.now()}`, + user_id: USER_ID, + event_type: EVENT_TYPE, + channel: "ws", + title: "v2 Kafka chain test", + content: "Verifying edu.notify.notification.sent consumption", + data: { message: "hello from v2 kafka chain test" }, + broadcast: false, + occurred_at: Date.now(), + }; + await producer.send({ + topic: TOPIC, + messages: [{ value: JSON.stringify(payload) }], + }); + console.log("[test] Kafka message published to", TOPIC); + await producer.disconnect(); + + // Wait up to 10 seconds for the WS message. + const deadline = Date.now() + 10000; + while (Date.now() < deadline && messages.length === 0) { + await new Promise((r) => setTimeout(r, 200)); + } + if (messages.length === 0) { + console.error("[test] FAIL: no WS message received within 10s"); + process.exit(1); + } + const got = messages[0]; + if (got.event === EVENT_TYPE && got.type === "message") { + console.log("[test] PASS: WS message matches expected event_type"); + process.exit(0); + } else { + console.error("[test] FAIL: WS message mismatch:", JSON.stringify(got)); + process.exit(1); + } +} + +// Hard timeout. +setTimeout(() => { + console.error("[test] FAIL: overall timeout 15s"); + process.exit(1); +}, 15000);