From 88f68224738748dfb637ab83e7bd56fc02420778 Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Fri, 10 Jul 2026 15:06:22 +0800 Subject: [PATCH 1/3] docs: update api-gateway workline and contract based on new arbitration update workline to detail full P2-P6 tasks with clear milestones and dependencies revise contract to correct route prefix, JWKS method and dependency descriptions clean up outdated issue content --- .../issues/contracts/api-gateway_contract.md | 208 +++++++--- .../issues/objections/api-gateway_issue.md | 23 +- .../issues/worklines/api-gateway_workline.md | 382 ++++++++++++++++-- 3 files changed, 518 insertions(+), 95 deletions(-) diff --git a/docs/architecture/issues/contracts/api-gateway_contract.md b/docs/architecture/issues/contracts/api-gateway_contract.md index 7290487..6ce23e4 100644 --- a/docs/architecture/issues/contracts/api-gateway_contract.md +++ b/docs/architecture/issues/contracts/api-gateway_contract.md @@ -1,40 +1,90 @@ # api-gateway 对接契约 > 负责人：ai01 -> 关联：[matrix.md](./matrix.md)、[iam.proto](../../../packages/shared-proto/proto/iam.proto) +> 关联：[matrix.md](./matrix.md)、[iam.proto](../../../packages/shared-proto/proto/iam.proto)、[objections/api-gateway_issue.md](../objections/api-gateway_issue.md)、[worklines/api-gateway_workline.md](../worklines/api-gateway_workline.md)、[02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) +> 依据：[coord-final-decisions.md](../../coord-final-decisions.md) §3.8 W1-W8、[president-final-rulings.md](../../president-final-rulings.md) §2.15/§2.16/§2.19 +> 版本：v2（2026-07-10 修正：路径前缀 /api→/api/v1、JWKS 拉取方式 gRPC→HTTP、删除 GetEffectiveAccess、错误码加 GW_ 前缀） --- ## §1 我提供什么（对外接口） -### 1.1 gRPC 接口（如有） +### 1.1 gRPC 接口 -无。api-gateway 是 HTTP 入口，不对外提供 gRPC。 +无。api-gateway 是 HTTP 入口，不对外提供 gRPC（依据 [president-final-rulings.md](../../president-final-rulings.md) §2.16 裁决：gateway 保持 HTTP 透传，不改为 gRPC 客户端）。 -### 1.2 HTTP 端点（如有） +### 1.2 HTTP 端点 -| Method | Path | 用途 | 认证 | -| ------ | -------------- | ------------------------------------------- | ---------------------- | -| ANY | /api/auth/* | 代理到 iam 认证相关（登录/注册/刷新 token） | 公开（登录注册免认证） | -| ANY | /api/teacher/* | 代理到 teacher-bff GraphQL（:3003） | JWT 必需 | -| ANY | /api/student/* | 代理到 student-bff GraphQL（:3009） | JWT 必需 | -| ANY | /api/parent/* | 代理到 parent-bff GraphQL（:3010） | JWT 必需 | -| ANY | /api/admin/* | 代理到 teacher-bff GraphQL admin namespace | JWT 必需 + admin 角色 | -| GET | /healthz | 网关健康检查（liveness） | 公开 | -| GET | /readyz | 网关就绪检查（readiness，含 iam 连通性） | 公开 | -| GET | /metrics | Prometheus 指标端点 | 公开（内网） | +#### 1.2.1 健康检查与指标端点（公开，无鉴权） -### 1.3 GraphQL schema（如 BFF） +| Method | Path | 用途 | 认证 | 阶段 | +| ------ | ---------- | ------------------------------------------- | ---- | ---- | +| GET | /healthz | 网关存活探针（liveness） | 公开 | P1 | +| GET | /readyz | 网关就绪探针（readiness，并行 ping 下游 /healthz）| 公开 | P2 | +| GET | /metrics | Prometheus 指标端点（7 个业务指标 + Go runtime）| 公开（内网）| P2 | -不适用。api-gateway 仅做 HTTP 反向代理 + JWT 验签，不解析 GraphQL。 +#### 1.2.2 反向代理路由（JWT 必需，除白名单外） -### 1.4 Kafka 事件发布（如有） +> **路径前缀**：所有业务路由统一 `/api/v1/*` 前缀（与 [main.go](../../../services/api-gateway/main.go) L59 `r.Group("/api/v1")` 一致）。 +> +> **API 版本化**：按 [president-final-rulings.md](../../president-final-rulings.md) §2.15 裁决，各服务 Controller 内加 `/v1` 前缀。Gateway 透传时不改路径，前端调用 `/api/v1//v1/*`（ISSUE-003 待 coord 仲裁最终方案，本表暂列方案 A）。 -无。api-gateway 不发布事件。 +| Method | Path（Gateway 外部路径） | 目标服务 | 端口 | 鉴权 | 公开子路径 | 阶段 | +| ------ | --------------------------------------------------- | ------------------------- | ----- | ---- | --------------------------------------- | ---- | +| ANY | /api/v1/iam/v1/*path + /api/v1/iam/v1 | iam | 3002 | JWT | `/iam/v1/register` `/login` `/refresh` | P2 | +| ANY | /api/v1/teacher/v1/*path + /api/v1/teacher/v1 | teacher-bff（GraphQL） | 3003 | JWT | — | P2 | +| ANY | /api/v1/student/v1/*path + /api/v1/student/v1 | student-bff（GraphQL） | 3009 | JWT | — | P3 | +| ANY | /api/v1/parent/v1/*path + /api/v1/parent/v1 | parent-bff（GraphQL） | 3010 | JWT | — | P4 | +| ANY | /api/v1/classes/v1/*path + /api/v1/classes/v1 | core-edu（classes 合并） | 3004 | JWT | — | P3 | +| ANY | /api/v1/exams/v1/*path + /api/v1/homework/v1/*path + /api/v1/grades/v1/*path | core-edu | 3004 | JWT | — | P3 | +| ANY | /api/v1/textbooks/v1/*path + /api/v1/chapters/v1/*path + /api/v1/knowledge-points/v1/*path + /api/v1/questions/v1/*path | content | 3005 | JWT | — | P4 | +| ANY | /api/v1/analytics/v1/*path + /api/v1/dashboard/v1/*path | data-ana | 3006 | JWT | — | P4 | +| ANY | /api/v1/notifications/v1/*path + /api/v1/messages/v1/*path | msg | 3007 | JWT | — | P5 | +| ANY | /api/v1/ai/v1/*path + /api/v1/ai/v1 | ai | 3008 | JWT | — | P5 | +| ANY | /api/v1/admin/v1/*path + /api/v1/admin/v1 | teacher-bff admin namespace | 3003 | JWT + admin 角色 | — | P6 | + +> **注 1**：每个前缀同时注册无尾斜杠与通配符两条路由（`/iam/v1` + `/iam/v1/*path`），因为 `r.RedirectTrailingSlash = false`（[main.go](../../../services/api-gateway/main.go) L34）。 +> +> **注 2**：admin-portal P6 阶段复用 teacher-bff + admin schema 命名空间（依据 [coord.md](../coord.md) §1.3 ARB-001 裁决）。 +> +> **注 3**：当前 P1 代码暂未加 `/v1` 前缀（如 `/api/v1/iam/*path`），待 ISSUE-003 仲裁后 P2.7 任务统一迁移。 + +#### 1.2.3 透传请求头（Gateway 注入，下游读取） + +| 头名 | 来源 | 用途 | 下游消费方 | +| -------------- | ----------------------------- | ----------------------------- | ---------- | +| `x-user-id` | JWT `sub` claim | 用户身份传递 | 所有下游 | +| `x-user-roles` | JWT `roles` claim（逗号分隔）| 角色传递 | 所有下游 | +| `x-data-scope` | JWT `data_scope` claim | 数据范围传递（P2 新增） | 所有下游 | +| `X-Request-Id` | 客户端透传或 Gateway 生成 UUID | 请求 ID（全链路追踪） | 所有下游 | +| `traceparent` | OTel SDK 自动处理 | W3C Trace Context（链路追踪）| 所有下游 | +| `tracestate` | OTel SDK 自动处理 | W3C Trace Context 扩展 | 所有下游 | + +### 1.3 GraphQL schema + +不适用。api-gateway 仅做 HTTP 反向代理 + JWT 验签，不解析 GraphQL（GraphQL 由 BFF 层处理）。 + +### 1.4 Kafka 事件发布 + +无。api-gateway 不发布 Kafka 事件（纯同步 HTTP 反向代理）。 ### 1.5 错误码前缀 -`GW_`（如 GW_UNAUTHORIZED、GW_RATE_LIMITED、GW_CIRCUIT_OPEN、GW_BACKEND_UNAVAILABLE） +`GW_`（依据 [coord-final-decisions.md](../../coord-final-decisions.md) W1 / G14 裁决） + +### 1.6 错误码清单 + +| 错误码 | HTTP | 触发条件 | 响应体（ActionState 信封） | +| ------------------------ | ---- | --------------------- | ----------------------------------------------------------------------- | +| `GW_UNAUTHORIZED` | 401 | 缺失 Authorization 头 | `{success:false,error:{code:"GW_UNAUTHORIZED",message:"..."}}` | +| `GW_INVALID_TOKEN` | 401 | JWT 签名/格式错误 | 同上 | +| `GW_INVALID_CLAIMS` | 401 | JWT claims 解析失败 | 同上 | +| `GW_RATE_LIMITED` | 429 | 超出令牌桶限流 | `{success:false,error:{code:"GW_RATE_LIMITED",message:"...",retry_after:60}}` | +| `GW_CIRCUIT_OPEN` | 503 | 下游熔断打开 | `{success:false,error:{code:"GW_CIRCUIT_OPEN",message:"...",retry_after:30}}` | +| `GW_REQUEST_TOO_LARGE` | 413 | 请求体超 10MB | `{success:false,error:{code:"GW_REQUEST_TOO_LARGE",message:"..."}}` | +| `GW_INTERNAL_ERROR` | 500 | panic 兜底 | `{success:false,error:{code:"GW_INTERNAL_ERROR",message:"...",request_id:"..."}}` | + +> 依据 W1 / W2 裁决：错误码统一 `GW_` 前缀，响应体统一 ActionState 信封 `{success,error:{code,message}}`。 --- @@ -42,22 +92,48 @@ ### 2.1 gRPC 调用（同步） -| 被调用方 | Service.RPC | 用途 | mock 策略 | -| ---------- | ----------------------------- | ---------------------------------------- | ----------------------------------------------------------------- | -| iam (ai06) | IamService.GetPublicKey | 启动时拉取 RS256 公钥，用于 JWT 验签 | iam 就绪前使用本地固定 mock 公钥（与 mock 私钥配对签发 mock JWT） | -| iam (ai06) | IamService.GetEffectiveAccess | 权限校验（可选，部分路由需要细粒度权限） | iam 就绪前放行所有请求（仅校验 JWT 签名） | +**无**。依据 [president-final-rulings.md](../../president-final-rulings.md) §2.16 裁决"gateway 保持 HTTP 透传，不改为 gRPC 客户端"，api-gateway 不消费任何 gRPC 接口。 -### 2.2 Kafka 事件订阅（异步） +> **勘误**（v1 版本曾错误声明消费 `IamService.GetPublicKey` 和 `IamService.GetEffectiveAccess`，v2 已删除）： +> - JWT 公钥拉取走 HTTP JWKS 端点，非 gRPC（见 §2.3） +> - Gateway 不做权限点校验，不消费 `GetEffectiveAccess`（权限由下游服务 Controller `@RequirePermission` 自校验，见 [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §2 + B3 裁决） +> - [matrix.md](./matrix.md) §2 gRPC 接口提供方矩阵中"iam 消费方"应移除 api-gateway（已提请 ISSUE-001） + +### 2.2 Kafka 事件订阅 无。api-gateway 不订阅 Kafka 事件。 -### 2.3 HTTP 调用（如有） +### 2.3 HTTP 调用（同步） -| 被调用方 | Method.Path | 用途 | mock 策略 | -| ------------------ | ------------- | --------------------------- | -------------------------------------------------- | -| teacher-bff (ai03) | POST /graphql | 反向代理教师端 GraphQL 请求 | teacher-bff 就绪前返回 502，前端使用本地 mock 数据 | -| student-bff (ai04) | POST /graphql | 反向代理学生端 GraphQL 请求 | student-bff 就绪前返回 502 | -| parent-bff (ai05) | POST /graphql | 反向代理家长端 GraphQL 请求 | parent-bff 就绪前返回 502 | +| 被调用方 | Method.Path | 用途 | mock 策略 | 阶段 | +| ------------------ | ----------------------------------- | ------------------------------------------ | --------------------------------------------------------------- | ---- | +| iam (ai06) | `GET /.well-known/jwks.json` | 拉 RS256 公钥集（JWKS），TTL 5min 缓存 | iam 就绪前使用本地固定 mock RS256 公钥（与 mock 私钥配对签发 mock JWT）| P2 | +| iam (ai06) | `GET /healthz` | /readyz 下游健康检查 | iam 就绪前 /readyz 软失败（返回 200 + degraded） | P2 | +| teacher-bff (ai03) | `POST /graphql`（反向代理） | 代理教师端 GraphQL 请求 | teacher-bff 就绪前返回 503 + Retry-After，前端降级到本地 mock | P2 | +| student-bff (ai04) | `POST /graphql`（反向代理） | 代理学生端 GraphQL 请求 | student-bff 就绪前返回 503 | P3 | +| parent-bff (ai05) | `POST /graphql`（反向代理） | 代理家长端 GraphQL 请求 | parent-bff 就绪前返回 503 | P4 | +| core-edu (ai08) | `GET /healthz` + 业务 REST（反向代理）| /readyz 检查 + 业务请求代理 | core-edu 就绪前 /readyz 软失败 | P3 | +| content (ai09) | `GET /healthz` + 业务 REST | /readyz 检查 + 业务请求代理 | content 就绪前 /readyz 软失败 | P4 | +| data-ana (ai11) | `GET /healthz` + 业务 REST | /readyz 检查 + 业务请求代理 | data-ana 就绪前 /readyz 软失败 | P4 | +| msg (ai10) | `GET /healthz` + 业务 REST | /readyz 检查 + 业务请求代理 | msg 就绪前 /readyz 软失败 | P5 | +| ai (ai12) | `GET /healthz` + 业务 REST | /readyz 检查 + 业务请求代理 | ai 就绪前 /readyz 软失败 | P5 | + +> **JWKS 缓存策略**（[packages/shared-go/jwks/jwks.go](../../../packages/shared-go/jwks/jwks.go)）： +> - TTL 5min，到期后台异步刷新（不阻塞请求） +> - kid 未命中时强制同步刷新一次 +> - 刷新失败保留旧公钥集继续服务（fail-open 1 次后 fail-close） +> - 启动时同步拉取一次，失败则 panic 拒绝启动 + +### 2.4 shared-go 包依赖 + +依据 [president-final-rulings.md](../../president-final-rulings.md) §2.19 裁决，ai01 直接 import `packages/shared-go`： + +| 模块 | 用途 | 接入阶段 | +| ----------------- | ------------------------------------------ | -------- | +| `shared-go/jwks` | JWKS Fetcher（HTTP 拉 RS256 公钥 + 缓存） | P2.2 | +| `shared-go/logger`| 结构化日志（zap 或 slog，待 ISSUE-004 仲裁）| P2.1 | +| `shared-go/tracer`| OTel tracer 初始化（评估接入，若接口兼容） | P2.1 | +| `shared-go/env` | 环境变量加载（评估接入，若接口兼容） | P2.1 | --- @@ -65,38 +141,70 @@ ### 3.1 我依赖的上游就绪标志 -- [ ] iam gRPC 50052 启用（ai06）—— GetPublicKey 拉取验签公钥 -- [ ] teacher-bff GraphQL :3003 启用（ai03） -- [ ] student-bff GraphQL :3009 启用（ai04） -- [ ] parent-bff GraphQL :3010 启用（ai05） +| 上游 | 就绪标志 | 阶段 | 状态 | +| ---- | -------- | ---- | ---- | +| coord | shared-go 包骨架（tracer/logger/jwks/env 4 模块）| 批次 0 | ✅ 已完成 | +| coord | ISSUE-001 仲裁（JWKS HTTP 确认）| P2 启动前 | ⏳ 待仲裁 | +| coord | ISSUE-002 仲裁（shared-go 接入）| P2 启动前 | ⏳ 待仲裁 | +| coord | ISSUE-003 仲裁（API 版本化路由方案）| P2.7 前 | ⏳ 待仲裁 | +| coord | ISSUE-004 仲裁（zap vs slog）| P2.1 前 | ⏳ 待仲裁 | +| iam (ai06) | `GET /.well-known/jwks.json` HTTP 端点 + RS256 JWT 签发 | P2 | ⏳ | +| iam (ai06) | `GET /healthz` 端点 | P2 | ⏳ | +| teacher-bff (ai03) | `POST /graphql` :3003 启用 + `GET /healthz` | P2 | ⏳ | +| student-bff (ai04) | `POST /graphql` :3009 启用 + `GET /healthz` | P3 | ⏳ | +| core-edu (ai08) | `GET /healthz` + classes 合并 | P3 | ⏳ | +| parent-bff (ai05) | `POST /graphql` :3010 启用 + `GET /healthz` | P4 | ⏳ | +| content (ai09) | `GET /healthz` + REST 端点 | P4 | ⏳ | +| data-ana (ai11) | `GET /healthz` + REST 端点 | P4 | ⏳ | +| msg (ai10) | `GET /healthz` + REST 端点 | P5 | ⏳ | +| ai (ai12) | `GET /healthz` + REST 端点 | P5 | ⏳ | +| push-gateway (ai02) | :8081 启用 + /internal/push（WebSocket 协作评估）| P5 | ⏳ | ### 3.2 我的就绪标志（供下游消费） -- [ ] api-gateway HTTP :8080 启用（/healthz 返回 200） -- [ ] /readyz 返回 200（含 iam 连通性检查通过） -- [ ] JWT 验签链路打通（使用 iam 公钥校验 access_token） -- [ ] /api/auth/* 代理到 iam 认证链路可用 -- [ ] /api/teacher/* /api/student/* /api/parent/* 反向代理到各 BFF 可用 -- [ ] 限流（IP 级令牌桶）+ 熔断（各后端独立熔断器）生效 +| 阶段 | 就绪标志 | 状态 | +| ---- | -------- | ---- | +| P2 | api-gateway HTTP :8080 启用（/healthz 返回 200）| ⏳ | +| P2 | /readyz 返回 200（含 iam + teacher-bff 连通性检查通过）| ⏳ | +| P2 | JWT RS256 验签链路打通（使用 iam JWKS 公钥校验 access_token）| ⏳ | +| P2 | /api/v1/iam/v1/* 代理到 iam 认证链路可用 | ⏳ | +| P2 | /api/v1/teacher/v1/* 反向代理到 teacher-bff GraphQL 可用 | ⏳ | +| P2 | 限流（IP 级令牌桶）+ 熔断（共享 downstream）+ CORS 白名单生效 | ⏳ | +| P2 | 7 个业务指标暴露在 /metrics | ⏳ | +| P2 | 错误响应统一 ActionState 信封 + GW_ 前缀 | ⏳ | +| P3 | /api/v1/student/v1/* + /api/v1/exams/v1/* 等路由可用 | ⏳ | +| P4 | /api/v1/parent/v1/* + /api/v1/textbooks/v1/* + /api/v1/analytics/v1/* 路由可用 | ⏳ | +| P5 | /api/v1/notifications/v1/* + /api/v1/ai/v1/* 路由可用 | ⏳ | +| P6 | 限流迁 Redis + 测试覆盖率 ≥ 80% | ⏳ | --- ## §4 Mock 策略 -### 4.1 我提供的 mock +### 4.1 我提供的 mock（供下游各前端 portal 消费） -在 api-gateway 真实就绪前，为下游（各前端 portal）提供以下 mock： +在 api-gateway 真实就绪前，为下游（teacher-portal / student-portal / parent-portal / admin-portal）提供以下 mock： -- **HTTP mock**：使用 MSW（Mock Service Worker）或本地 nginx 拦截 - - /api/auth/login 返回固定 JWT（mock 签发）+ UserInfo - - /api/teacher/* /api/student/* /api/parent/* 直接返回各 BFF 的 mock GraphQL 响应 - - /healthz /readyz 返回 200 +- **HTTP mock**（由各前端 portal 自行用 MSW 拦截，api-gateway 不提供 mock 服务）： + - `/api/v1/iam/v1/login` 返回固定 JWT（mock 签发）+ UserInfo + - `/api/v1/teacher/v1/*` `/api/v1/student/v1/*` `/api/v1/parent/v1/*` 直接返回各 BFF 的 mock GraphQL 响应 + - `/healthz` `/readyz` 返回 200 - **JWT mock**：前端开发期使用固定 mock JWT（api-gateway 就绪前不走真实验签） -### 4.2 我消费的 mock +### 4.2 我消费的 mock（在真实上游就绪前） 在真实上游就绪前，api-gateway 使用以下 mock： -- **iam 公钥**：使用本地固定 mock RS256 公钥（与 mock 私钥配对），验签 mock JWT -- **iam 权限校验**：GetEffectiveAccess 返回 allowed=true，放行所有请求 -- **各 BFF 代理**：BFF 就绪前返回 503 + Retry-After，前端降级到本地 mock 数据 +- **iam JWKS 公钥**：使用本地固定 mock RS256 公钥（与 mock 私钥配对），验签 mock JWT（DevMode 下生效） +- **iam /readyz 检查**：iam 就绪前软失败，返回 200 + `degraded: true` +- **各 BFF 代理**：BFF 就绪前返回 503 + `Retry-After`，前端降级到本地 mock 数据 +- **下游 /healthz 检查**：未就绪服务软失败（返回 200 + degraded），已就绪服务硬失败（返回 503） + +--- + +## §5 变更记录 + +| 版本 | 日期 | 变更内容 | 变更者 | +| ---- | ---------- | ------------------------------------------------------------------------ | ------ | +| v1 | 2026-07-09 | 初始创建（含错误：/api/* 路径、gRPC GetPublicKey、GetEffectiveAccess） | ai01 | +| v2 | 2026-07-10 | 修正：路径 /api→/api/v1、JWKS 拉取 gRPC→HTTP、删除 GetEffectiveAccess、错误码加 GW_ 前缀、对齐 ActionState 信封、补充 shared-go 依赖、补充 admin-portal P6 路由 | ai01 | diff --git a/docs/architecture/issues/objections/api-gateway_issue.md b/docs/architecture/issues/objections/api-gateway_issue.md index 4488d96..258f76e 100644 --- a/docs/architecture/issues/objections/api-gateway_issue.md +++ b/docs/architecture/issues/objections/api-gateway_issue.md @@ -1,24 +1,3 @@ # api-gateway 问题记录 -> 负责人：ai01 -> 关联：[coord.md](../coord.md)、[contracts/api-gateway_contract.md](../contracts/api-gateway_contract.md) -> 规则：AI 遇到问题时在此追加条目，coord 仲裁后更新状态 - ---- - -## 问题列表 - - - -（暂无问题） +> 负 \ No newline at end of file diff --git a/docs/architecture/issues/worklines/api-gateway_workline.md b/docs/architecture/issues/worklines/api-gateway_workline.md index 43fd5ed..c61eab0 100644 --- a/docs/architecture/issues/worklines/api-gateway_workline.md +++ b/docs/architecture/issues/worklines/api-gateway_workline.md @@ -1,57 +1,393 @@ # api-gateway 工作排期 > 负责人：ai01 -> 关联：[workline.md](../workline.md)、[coord.md](../coord.md)、[contracts/api-gateway_contract.md](../contracts/api-gateway_contract.md) +> 关联：[workline.md](../workline.md)、[coord.md](../coord.md)、[contracts/api-gateway_contract.md](../contracts/api-gateway_contract.md)、[objections/api-gateway_issue.md](../objections/api-gateway_issue.md) > 模式：全并行（各 AI 一口气完成 P2-P6 全部代码，最后统一集成测试） +> 依据：[coord-final-decisions.md](../../coord-final-decisions.md) §3.8 W1-W8、[president-final-rulings.md](../../president-final-rulings.md) §2.15/§2.16/§2.19、[02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §9 --- ## §1 总览 -api-gateway 是 Edu 系统统一入口，负责路由、JWT 验签、限流、熔断、CORS。全阶段目标：P2 路由+JWT → P3 限流加固 → P4-P6 持续优化。 +api-gateway 是 Edu 系统统一入口（L3 网关层），负责路由转发、JWT RS256 验签、限流、熔断、CORS、可观测性。无业务状态，纯 HTTP 反向代理。 + +**全阶段目标**： +- **P2**：路由表 + JWT RS256（HTTP JWKS） + shared-go 接入 + 错误码 GW_ 前缀 + ActionState 信封 + slog + /readyz 真实检查 + 业务 metrics + tracer 资源属性 + DevMode 防护（遵循 W1-W8 / G1-G17 裁决） +- **P3**：路由扩展（student-bff :3009）+ core-edu 路由 +- **P4**：路由扩展（parent-bff :3010）+ content / data-ana 路由 +- **P5**：路由扩展（msg / ai 路由）+ 接入 push-gateway 协作（WebSocket 升级透传评估） +- **P6**：限流迁 Redis + per-服务实例熔断评估 + 测试覆盖率 ≥ 80% + 安全加固 + +**当前状态（2026-07-10）**：P1 已交付（classes 域 CRUD 端到端跑通），P2 升级未启动。已有仲裁核查发现 6 项未遵循裁决（见 [objections/api-gateway_issue.md](../objections/api-gateway_issue.md) §0.4）。 --- -## §2 全阶段甘特图（P2-P6，各 AI 自行细化） +## §2 全阶段甘特图（P2-P6） ```mermaid gantt - title ai01 api-gateway 全阶段排期 + title ai01 api-gateway 全阶段排期（P2-P6） dateFormat YYYY-MM-DD axisFormat %m-%d - section P2 基础 - 路由表+双入口+shared-go :a1a, 2026-07-10, 3d - JWT校验+JWKS fetcher :a1b, after a1a, 3d - 限流+熔断+CORS :a1c, after a1b, 2d + section P2 基础升级（批次1） + P2.0 修复P0遗留 :crit, p2a, 2026-07-10, 1d + P2.1 shared-go接入 :crit, p2b, after p2a, 2d + P2.2 JWT RS256+JWKS :crit, p2c, after p2b, 3d + P2.3 错误码GW_+ActionState :crit, p2d, after p2c, 1d + P2.4 slog+metrics+tracer :crit, p2e, after p2d, 2d + P2.5 /readyz真实检查 :crit, p2f, after p2e, 1d + P2.6 DevMode防护 :p2g, after p2f, 1d + P2.7 路由表扩展iam/teacher :p2h, after p2g, 1d - section P3-P6 持续优化 - 路由扩展(student/parent/admin) :a1d, after a1c, 2d - 指标+链路加固 :a1e, after a1d, 2d + section P3 路由扩展（批次2） + P3.1 student-bff路由 :p3a, after p2h, 1d + P3.2 core-edu路由 :p3b, after p3a, 1d + P3.3 API版本化/v1迁移 :p3c, after p3b, 2d + + section P4 路由扩展（批次3） + P4.1 parent-bff路由 :p4a, after p3c, 1d + P4.2 content路由 :p4b, after p4a, 1d + P4.3 data-ana路由 :p4c, after p4b, 1d + + section P5 路由扩展（批次4） + P5.1 msg路由 :p5a, after p4c, 1d + P5.2 ai路由 :p5b, after p5a, 1d + P5.3 push-gateway协作评估 :p5c, after p5b, 2d + + section P6 硬化（批次5） + P6.1 限流迁Redis :p6a, after p5c, 3d + P6.2 per-服务熔断评估 :p6b, after p6a, 2d + P6.3 测试覆盖率80% :p6c, after p6b, 3d + P6.4 安全加固 :p6d, after p6c, 2d ``` -> **注意**：以上为 coord 初始规划，ai01 接管后必须自行细化为完整 P2-P6 排期。 +**关键路径**（红色 crit）：P2.0 → P2.1 → P2.2 → P2.3 → P2.4 → P2.5 → P2.6 → P2.7 +**总时间线**：P2 约 11 天 + P3-P5 约 9 天 + P6 约 10 天 = 约 30 天 --- ## §3 详细任务 -### P2：路由表 + JWT + 限流 +### P2.0：修复 P0 遗留问题 - **负责人**：ai01 +- **依赖**：无 - **交付物**： - - `services/api-gateway/internal/routing/router.go` — 路由表 - - `/api/v1/teacher/*` → teacher-bff:3003 代理 - - `/api/v1/iam/*` → iam:3002 代理 - - JWT RS256 验签（shared-go/jwks） - - 限流 + 熔断 + CORS -- **依赖**：shared-go 骨架（批次 0 已完成）+ iam GetPublicKey（ai06） -- **验收标准**：路由双入口 + JWT 验签 + 限流 + CORS 白名单 -- **完整 P3-P6 任务**：⚠️ 由 ai01 自行补充 + - `services/api-gateway/go.mod` L3 改为 `go 1.22`（修复 ISSUE-005） + - `go.work` L1 改为 `go 1.22` + - 删除 [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §7.1 issue #3 死代码引用（ISSUE-008） + - 修正 [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §6 审计表 metrics 行（ISSUE-007） + - 修正 [README.md](../../../services/api-gateway/README.md) L38 删除不存在的 `/health` 兼容端点描述 + - 修正 [proxy.go](../../../services/api-gateway/internal/proxy/proxy.go) L24 删除冗余 `TrimPrefix("/api")` +- **验收标准**：`go build ./...` + `go vet ./...` 通过；文档与代码一致 +- **对应 ISSUE**：ISSUE-005 / ISSUE-007 / ISSUE-008 + +### P2.1：shared-go 包接入 + +- **负责人**：ai01 +- **依赖**：[packages/shared-go](../../../packages/shared-go/) 已建立（批次 0 已完成）；coord 仲裁 ISSUE-002 / ISSUE-004 +- **交付物**： + - `go.work` 增加 `./packages/shared-go` + - `services/api-gateway/go.mod` 增加 `github.com/edu-cloud/shared-go` 依赖 + - 按 ISSUE-004 仲裁结果接入 shared-go/logger（zap 或 slog，取决于 coord 裁决） + - [tracer.go](../../../services/api-gateway/internal/observability/tracer.go) 评估接入 shared-go/tracer（若接口兼容） + - [config.go](../../../services/api-gateway/internal/config/config.go) 评估接入 shared-go/env +- **验收标准**：`go build ./...` 通过；import shared-go 成功；logger 输出结构化 JSON +- **对应 ISSUE**：ISSUE-002 / ISSUE-004 + +### P2.2：JWT RS256 升级 + JWKS 缓存 + +- **负责人**：ai01 +- **依赖**：iam (ai06) 暴露 `GET /.well-known/jwks.json` HTTP 端点；coord 仲裁 ISSUE-001（确认 HTTP JWKS） +- **交付物**： + - [auth.go](../../../services/api-gateway/internal/middleware/auth.go) 改用 shared-go/jwks.Fetcher（`jwks.NewFetcher(cfg.JWKSURL)`） + - HS256 逻辑废弃，`cfg.JWTSecret` 仅 DevMode 下用作 mock 密钥 + - JWKS 缓存策略：TTL 5min（shared-go/jwks 默认），kid 未命中时强制刷新，刷新失败保留旧公钥 + - 启动时同步拉取一次 JWKS，失败则 panic 拒绝启动 + - claims 增加 `data_scope` 字段提取，注入 `x-data-scope` 头 + - `internal/config/config.go` 增加 `JWKSURL` 字段（环境变量 `IAM_JWKS_URL`） +- **验收标准**：JWT RS256 验签通过；JWKS 缓存命中率达 99%+；kid 未命中自动刷新 +- **对应裁决**：W1（错误码加 GW_ 前缀）、§2.16（HTTP JWKS，非 gRPC） + +### P2.3：错误码 GW_ 前缀 + ActionState 信封 + +- **负责人**：ai01 +- **依赖**：无 +- **交付物**： + - [auth.go](../../../services/api-gateway/internal/middleware/auth.go) 错误码改为 `GW_UNAUTHORIZED` / `GW_INVALID_TOKEN` / `GW_INVALID_CLAIMS` + - [ratelimit.go](../../../services/api-gateway/internal/middleware/ratelimit.go) 响应体改为 `{success:false,error:{code:"GW_RATE_LIMITED",message:"...",retry_after:60}}` + - [circuit-breaker.go](../../../services/api-gateway/internal/middleware/circuit-breaker.go) 响应体改为 `{success:false,error:{code:"GW_CIRCUIT_OPEN",message:"...",retry_after:30}}` + - [recovery.go](../../../services/api-gateway/internal/middleware/recovery.go) 响应体改为 `{success:false,error:{code:"GW_INTERNAL_ERROR",message:"...",request_id:"..."}}` + - `RequestBodyLimit` 超限响应改为 `{success:false,error:{code:"GW_REQUEST_TOO_LARGE",message:"..."}}` +- **验收标准**：所有错误响应符合 ActionState 信封；错误码统一 `GW_` 前缀 +- **对应裁决**：W1 / W2 / G14 +- **对应 ISSUE**：ISSUE-009 + +### P2.4：slog + 业务 metrics + tracer 资源属性 + +- **负责人**：ai01 +- **依赖**：P2.1 shared-go 接入完成 +- **交付物**： + - 按 ISSUE-004 仲裁结果统一 logger（zap 或 slog） + - 所有 `log.Printf` / `log.Println` / `log.Fatal` 改为结构化日志（带 `request_id` / `trace_id` / `user_id` / `method` / `path` / `status` / `latency_ms` 字段） + - 新增 `internal/observability/metrics.go`，注册 7 个业务指标： + - `api_gateway_http_requests_total`（Counter，method/endpoint/status） + - `api_gateway_http_request_duration_seconds`（Histogram，method/endpoint） + - `api_gateway_circuit_breaker_state`（Gauge，service/state） + - `api_gateway_rate_limited_total`（Counter，ip） + - `api_gateway_proxy_upstream_duration_seconds`（Histogram，upstream） + - `api_gateway_jwks_refresh_total`（Counter，result） + - `api_gateway_auth_failures_total`（Counter，reason） + - [tracer.go](../../../services/api-gateway/internal/observability/tracer.go) 资源属性补全：`service.name` + `service.version`（编译时注入）+ `deployment.environment`（ENV 变量）+ `host.name` + - Metrics 中间件：在 Auth 之后、CircuitBreaker 之前注册（统计通过鉴权的请求） +- **验收标准**：`/metrics` 端点返回 7 个业务指标；日志为 JSON 结构化；tracer 资源属性完整 +- **对应裁决**：W3 / W5 / W6 / G4 / G5 / G6 + +### P2.5：/readyz 真实健康检查 + +- **负责人**：ai01 +- **依赖**：所有下游服务实现 `/healthz`（P2 阶段 iam / teacher-bff 已就绪） +- **交付物**： + - [health.go](../../../services/api-gateway/internal/health/health.go) `Readyz` 重构为并行 ping 下游 `/healthz` + - 下游清单从 `cfg.ServicesURL` 动态读取（iam / classes / teacher-bff / core-edu / content / msg / ai / data-ana） + - 超时 2s，任一不可达返回 503 + `{"status":"error","unhealthy":["iam","core-edu"]}` + - 全部可达返回 200 + `{"status":"ok"}` + - 可选依赖软失败规则：未启用 gRPC 的下游（P3-P5 阶段未就绪的服务）失败仅告警，返回 200 + `degraded: true`（依据 president-final-rulings.md §3.3） +- **验收标准**：/readyz 真实检查下游；某服务下线时返回 503 +- **对应裁决**：W4 / G2 + +### P2.6：DevMode 生产防护 + +- **负责人**：ai01 +- **依赖**：无 +- **交付物**： + - [config.go](../../../services/api-gateway/internal/config/config.go) `Load()` 增加 `ENV` 环境变量读取 + - 若 `DevMode=true && ENV=production` 则 `panic` 拒绝启动 + - 启动日志打印 `ENV` / `DevMode` 状态 +- **验收标准**：`DEV_MODE=true ENV=production` 启动失败；`DEV_MODE=true ENV=development` 启动成功 +- **对应裁决**：W7 + +### P2.7：路由表扩展（iam / teacher） + +- **负责人**：ai01 +- **依赖**：coord 仲裁 ISSUE-003（API 版本化路由规则）；iam (ai06) / teacher-bff (ai03) P2 就绪 +- **交付物**： + - 按 ISSUE-003 仲裁结果更新路由（方案 A/B/C 之一） + - 若方案 A：iam 路由改为 `/api/v1/iam/v1/*path`，proxy 透传 `/iam/v1/*path` + - teacher-bff 路由保持 `/api/v1/teacher/*path` + - [02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §4.1 路由表同步更新 +- **验收标准**：路由表与代码一致；前端调用 `/api/v1/iam/v1/auth/login` 透传到 iam 服务 +- **对应裁决**：§2.15 + +--- + +### P3.1：student-bff 路由 + +- **负责人**：ai01 +- **依赖**：student-bff (ai04) P3 就绪 +- **交付物**： + - `main.go` 增加 student-bff 路由：`/api/v1/student` + `/api/v1/student/*path` → `cfg.StudentBffURL`（:3009） + - `config.go` 增加 `StudentBffURL` 字段（环境变量 `STUDENT_BFF_URL`） +- **验收标准**：`/api/v1/student/*` 代理到 student-bff:3009 + +### P3.2：core-edu 路由 + +- **负责人**：ai01 +- **依赖**：core-edu (ai08) P3 就绪；classes 服务已合并入 core-edu（C1 裁决） +- **交付物**： + - `main.go` classes 路由目标改为 core-edu（`cfg.ClassesServiceURL` → `cfg.CoreEduServiceURL`） + - 或保留 classes 路由别名，proxy 到 core-edu + - exams / homework / grades 路由已在 P1 实现，无需修改 +- **验收标准**：`/api/v1/classes/*` 代理到 core-edu:3004 + +### P3.3：API 版本化 /v1 迁移 + +- **负责人**：ai01 +- **依赖**：P2.7 路由规则仲裁结果；core-edu (ai08) P3 就绪 +- **交付物**： + - 按 ISSUE-003 仲裁结果，全服务路由统一加 `/v1` 前缀 + - core-edu 路由：`/api/v1/exams/v1/*path` 等（若方案 A） + - 更新 [02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §4.1 路由表 +- **验收标准**：所有业务路由含 `/v1` 版本前缀 + +--- + +### P4.1：parent-bff 路由 + +- **负责人**：ai01 +- **依赖**：parent-bff (ai05) P4 就绪 +- **交付物**： + - `main.go` 增加 parent-bff 路由：`/api/v1/parent` + `/api/v1/parent/*path` → `cfg.ParentBffURL`（:3010） + - `config.go` 增加 `ParentBffURL` 字段 +- **验收标准**：`/api/v1/parent/*` 代理到 parent-bff:3010 + +### P4.2：content 路由 + +- **负责人**：ai01 +- **依赖**：content (ai09) P4 就绪 +- **交付物**： + - `main.go` 已有 content 路由（P1 实现：textbooks / chapters / knowledge-points / questions） + - 验证路由目标 `cfg.ContentServiceURL`（:3005）正确 + - 按 P3.3 版本化规则加 `/v1` 前缀 +- **验收标准**：content 路由可用 + 版本化 + +### P4.3：data-ana 路由 + +- **负责人**：ai01 +- **依赖**：data-ana (ai11) P4 就绪 +- **交付物**： + - `main.go` 已有 data-ana 路由（P1 实现：analytics） + - 增加 `dashboard` 路由别名：`/api/v1/dashboard` + `/*path` → data-ana + - 按版本化规则加 `/v1` 前缀 +- **验收标准**：data-ana 路由可用 + dashboard 别名可用 + +--- + +### P5.1：msg 路由 + +- **负责人**：ai01 +- **依赖**：msg (ai10) P5 就绪 +- **交付物**： + - `main.go` 已有 msg 路由（P1 实现：notifications） + - 增加 `messages` 路由别名：`/api/v1/messages` + `/*path` → msg + - 按版本化规则加 `/v1` 前缀 +- **验收标准**：msg 路由可用 + messages 别名可用 + +### P5.2：ai 路由 + +- **负责人**：ai01 +- **依赖**：ai (ai12) P5 就绪 +- **交付物**： + - `main.go` 已有 ai 路由（P1 实现） + - 按版本化规则加 `/v1` 前缀 +- **验收标准**：ai 路由可用 + +### P5.3：push-gateway 协作评估 + +- **负责人**：ai01 +- **依赖**：push-gateway (ai02) P5 就绪 +- **交付物**： + - 评估 WebSocket 升级请求是否需要 Gateway 透传到 push-gateway + - 若需要：增加 `/ws` + `/sse` 路由透传到 push-gateway（:8081） + - 若不需要：文档说明 WebSocket 直连 push-gateway，不经过 Gateway + - 更新 [02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §7 交互点清单 +- **验收标准**：WebSocket 推送链路可用（透传或直连） + +--- + +### P6.1：限流迁 Redis + +- **负责人**：ai01 +- **依赖**：Redis 基础设施就绪 +- **交付物**： + - [ratelimit.go](../../../services/api-gateway/internal/middleware/ratelimit.go) 改用 Redis 令牌桶（`github.com/go-redis/redis_rate/v10`） + - 支持多副本一致限流 + - 增加用户级限流（基于 `x-user-id` 头） + - 登录接口额外加用户级限流（防爆破） + - 保留 DevMode 下内存令牌桶回退 +- **验收标准**：多副本部署时限流一致；用户级限流生效 + +### P6.2：per-服务实例熔断评估 + +- **负责人**：ai01 +- **依赖**：P6.1 完成 +- **交付物**： + - 评估是否将共享 `downstream` 熔断器拆分为 per-服务实例（iam / core-edu / teacher-bff 等） + - 若拆分：每个下游服务独立熔断状态，互不影响 + - 若不拆分：保持 W8 裁决现状，文档说明理由 + - 按 W8 裁决，此项 P6 单独评估，不强制拆分 +- **验收标准**：评估报告 + 决策记录 + +### P6.3：测试覆盖率 80% + +- **负责人**：ai01 +- **依赖**：P2-P5 全部完成 +- **交付物**： + - 补全 [auth_test.go](../../../services/api-gateway/internal/middleware/auth_test.go)：JWKS 验签 / claims 解析 / DevMode 旁路 / 公开路径白名单 + - 补全 [cors_test.go](../../../services/api-gateway/internal/middleware/cors_test.go)：白名单匹配 / 预检请求 / Vary 头 + - 补全 [security_test.go](../../../services/api-gateway/internal/middleware/security_test.go)：安全头设置 / Server 头移除 + - 补全 [recovery_test.go](../../../services/api-gateway/internal/middleware/recovery_test.go)：panic 捕获 / request_id 生成 / ActionState 信封 + - 补全 [requestid_test.go](../../../services/api-gateway/internal/middleware/requestid_test.go)：透传 / 生成 / 响应头 + - 补全 [proxy_test.go](../../../services/api-gateway/internal/proxy/proxy_test.go)：路径前缀去除 / Host 改写 + - 补全 [health_test.go](../../../services/api-gateway/internal/health/health_test.go)：/readyz 下游检查 / 软失败规则 +- **验收标准**：`go test ./... -cover` 覆盖率 ≥ 80% + +### P6.4：安全加固 + +- **负责人**：ai01 +- **依赖**：P6.1-P6.3 完成 +- **交付物**： + - 评估 IP 黑名单 / WAF 规则是否在 Gateway 层实现（建议在 Istio 层做，本服务不介入） + - 限流策略表 per-路由细化（02 §4.2） + - 熔断阈值 per-服务配置（02 §4.3） + - CORS 白名单生产环境强制配置（禁止 `*`） + - 审计日志：记录所有 401 / 403 / 429 / 503 响应 +- **验收标准**：安全扫描通过；审计日志可追溯 --- ## §4 依赖与就绪信号 -- **我依赖**：iam GetPublicKey RPC（ai06） -- **我的就绪信号**：api-gateway :8080 可访问 + JWT 验签可用 +### 4.1 我依赖的上游就绪标志 + +| 上游 | 就绪标志 | 阶段 | 状态 | +| ---- | -------- | ---- | ---- | +| coord | shared-go 包骨架（tracer/logger/jwks/env）| 批次 0 | ✅ 已完成 | +| coord | ISSUE-001 仲裁（JWKS vs gRPC）| P2 启动前 | ⏳ 待仲裁 | +| coord | ISSUE-002 仲裁（shared-go 接入）| P2 启动前 | ⏳ 待仲裁 | +| coord | ISSUE-003 仲裁（API 版本化路由）| P2.7 前 | ⏳ 待仲裁 | +| coord | ISSUE-004 仲裁（zap vs slog）| P2.1 前 | ⏳ 待仲裁 | +| iam (ai06) | `GET /.well-known/jwks.json` HTTP 端点 + RS256 签发 | P2 | ⏳ | +| iam (ai06) | gRPC 50052 启用（不影响 Gateway，Gateway 走 HTTP）| P2 | ⏳ | +| teacher-bff (ai03) | `POST /graphql` :3003 启用 | P2 | ⏳ | +| student-bff (ai04) | `POST /graphql` :3009 启用 | P3 | ⏳ | +| core-edu (ai08) | gRPC 50053 启用 + classes 合并 | P3 | ⏳ | +| parent-bff (ai05) | `POST /graphql` :3010 启用 | P4 | ⏳ | +| content (ai09) | gRPC 50054 启用 | P4 | ⏳ | +| data-ana (ai11) | gRPC 50055 启用 | P4 | ⏳ | +| msg (ai10) | gRPC 50056 启用 | P5 | ⏳ | +| ai (ai12) | gRPC 50058 启用 | P5 | ⏳ | +| push-gateway (ai02) | :8081 启用 + /internal/push | P5 | ⏳ | + +### 4.2 我的就绪标志（供下游消费） + +| 阶段 | 就绪标志 | 状态 | +| ---- | -------- | ---- | +| P2 | api-gateway :8080 可访问 + JWT RS256 验签可用 + 7 个业务指标 + /readyz 真实检查 | ⏳ | +| P2 | /api/v1/iam/* + /api/v1/teacher/* 路由可用 | ⏳ | +| P3 | /api/v1/student/* + /api/v1/exams/* 等路由可用 | ⏳ | +| P4 | /api/v1/parent/* + /api/v1/textbooks/* + /api/v1/analytics/* 路由可用 | ⏳ | +| P5 | /api/v1/notifications/* + /api/v1/ai/* 路由可用 | ⏳ | +| P6 | 限流迁 Redis + 测试覆盖率 ≥ 80% | ⏳ | + +--- + +## §5 风险与应对 + +| 风险 | 概率 | 影响 | 应对 | +| ---- | ---- | ---- | ---- | +| coord 仲裁延期（ISSUE-001/002/003/004）| 中 | P2 阻塞 | ai01 先按 HTTP JWKS + shared-go + 方案 A 推进，仲裁后调整 | +| iam JWKS 端点延期 | 中 | P2.2 阻塞 | DevMode 下用本地 mock RS256 公钥（与 mock 私钥配对） | +| shared-go 接口不兼容 | 低 | P2.1 阻塞 | ai01 自行适配，或反馈 coord 修改 shared-go | +| 下游服务未实现 /healthz | 中 | /readyz 误报 | 软失败规则：未就绪服务失败仅告警，返回 200 + degraded | +| JWKS 缓存过期时 iam 不可达 | 低 | 全量 401 | fail-open 1 次后 fail-close；监控 `jwks_refresh_total` 指标 | +| DevMode 旁路误开到生产 | 低 | 鉴权绕过 | P2.6 生产防护（W7 裁决） | + +--- + +## §6 与其他模块的协作 + +| 模块 | 协作内容 | 时机 | +| ---- | -------- | ---- | +| iam (ai06) | JWKS HTTP 端点 + JWT RS256 签发 | P2 | +| teacher-bff (ai03) | 反向代理 :3003 GraphQL | P2 | +| student-bff (ai04) | 反向代理 :3009 GraphQL | P3 | +| parent-bff (ai05) | 反向代理 :3010 GraphQL | P4 | +| core-edu (ai08) | 反向代理 :3004 + classes 合并 | P3 | +| content (ai09) | 反向代理 :3005 | P4 | +| data-ana (ai11) | 反向代理 :3006 | P4 | +| msg (ai10) | 反向代理 :3007 | P5 | +| ai (ai12) | 反向代理 :3008 | P5 | +| push-gateway (ai02) | WebSocket 升级透传评估 | P5 | +| coord | shared-go 包维护 + ISSUE 仲裁 | 持续 | From 8e9103989611da0bc3dc89e01df778e97abea314 Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Fri, 10 Jul 2026 16:41:57 +0800 Subject: [PATCH 2/3] =?UTF-8?q?docs(api-gateway):=20=E8=A1=A5=E5=85=85?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E6=9E=B6=E6=9E=84=E9=97=AE=E9=A2=98=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E4=B8=8E=E8=A3=81=E5=86=B3=E6=A0=B8=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 完成api-gateway模块边界确认，添加W1-W8裁决遵循情况核查与架构不一致问题梳理 汇总8项待coord仲裁的issue，包含模块契约、依赖、版本、文档等问题 --- .../issues/objections/api-gateway_issue.md | 167 +++++++++++++++++- 1 file changed, 166 insertions(+), 1 deletion(-) diff --git a/docs/architecture/issues/objections/api-gateway_issue.md b/docs/architecture/issues/objections/api-gateway_issue.md index 258f76e..be0ee29 100644 --- a/docs/architecture/issues/objections/api-gateway_issue.md +++ b/docs/architecture/issues/objections/api-gateway_issue.md @@ -1,3 +1,168 @@ # api-gateway 问题记录 -> 负 \ No newline at end of file +> 负责人：ai01 +> 关联：[coord.md](../coord.md)、[contracts/api-gateway_contract.md](../contracts/api-gateway_contract.md)、[worklines/api-gateway_workline.md](../worklines/api-gateway_workline.md) +> 依据：[coord-final-decisions.md](../../coord-final-decisions.md) §3.8 W1-W8、[president-final-rulings.md](../../president-final-rulings.md) §2.15/§2.16/§2.19 +> 规则：AI 遇到问题时在此追加条目，coord 仲裁后更新状态 + +--- + +## §0 模块边界确认（审查结论） + +### 0.1 边界定义（来自 [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §2） + +- **层级**：L3 网关层（004 §3.1 六层架构） +- **职责**：所有外部请求的统一入口、JWT 校验、用户身份注入（`x-user-id`/`x-user-roles`/`x-data-scope`）、限流、熔断、CORS、安全头、请求 ID 注入、反向代理 +- **不负责**： + - WebSocket 长连接管理（push-gateway 负责） + - 用户身份认证逻辑（iam 负责，本服务只校验 JWT 签名） + - 权限点解析（下游服务 Controller `@RequirePermission` 自校验，本服务只透传身份头） + - 业务数据持久化（无 DB） +- **通信方式**：入口 HTTP/REST，出口 HTTP 反向代理（`httputil.ReverseProxy`），不消费 gRPC，不接入 Kafka + +> **边界审查结论**：边界定义清晰，与 [02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §7 交互点清单、[contracts/api-gateway_contract.md](../contracts/api-gateway_contract.md) v2 一致。本服务是纯同步 HTTP 反向代理，不持有业务状态。 + +### 0.2 W1-W8 裁决遵循情况核查（2026-07-10） + +对照 [coord-final-decisions.md](../../coord-final-decisions.md) §3.8 W1-W8 裁决，逐项核查当前代码（P1 交付态）： + +| # | 裁决项 | 强制最终方案 | 当前代码状态 | 遵循 | 对应 ISSUE | +| --- | -------------------- | --------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---- | ---------- | +| W1 | **错误码前缀** | `GW_*`（GW_UNAUTHORIZED 等） | [auth.go](../../../services/api-gateway/internal/middleware/auth.go) 用 `UNAUTHORIZED`/`INVALID_TOKEN`/`INVALID_CLAIMS`（无 GW_ 前缀） | ❌ | ISSUE-009 | +| W2 | **错误信封** | 首次实现即对齐 ActionState | [ratelimit.go](../../../services/api-gateway/internal/middleware/ratelimit.go) L51-54 响应 `{"error":"rate_limited","retry_after":60}`（无 success/error.code 结构）；[circuit-breaker.go](../../../services/api-gateway/internal/middleware/circuit-breaker.go) L55-58 同样无信封；[recovery.go](../../../services/api-gateway/internal/middleware/recovery.go) L24-27 无信封；auth.go 部分对齐 | ❌ | ISSUE-009 | +| W3 | **slog** | 首次实现即用 slog | 全部用 `log.Printf`：[main.go](../../../services/api-gateway/main.go) L70/160/169/176、[config.go](../../../services/api-gateway/internal/config/config.go) L38/41、[recovery.go](../../../services/api-gateway/internal/middleware/recovery.go) L22、[circuit-breaker.go](../../../services/api-gateway/internal/middleware/circuit-breaker.go) L38、[tracer.go](../../../services/api-gateway/internal/observability/tracer.go) L25/31/43/51/65/70 | ❌ | ISSUE-004 | +| W4 | **/readyz** | 首次实现即真实健康检查（轮询下游 /healthz） | [health.go](../../../services/api-gateway/internal/health/health.go) L19-23 直接返回 200，有 TODO 注释"P7 接入服务发现后" | ❌ | — | +| W5 | **业务 metrics** | 首次实现即基础业务指标 | [main.go](../../../services/api-gateway/main.go) L56 已注册 `/metrics` 端点（promhttp.Handler），但未注册 7 个业务指标（无 `internal/observability/metrics.go` 文件） | ⚠️ | ISSUE-007 | +| W6 | **tracer 资源属性** | 首次实现即完整（version/env/host） | [tracer.go](../../../services/api-gateway/internal/observability/tracer.go) L47-49 只有 `service.name`，缺 `service.version`/`deployment.environment`/`host.name` | ❌ | — | +| W7 | **DevMode 防护** | 首次实现即生产防护（DEV_MODE 仅非生产环境生效）| [config.go](../../../services/api-gateway/internal/config/config.go) `Load()` 无 `ENV` 环境变量读取，DevMode=true 时无生产防护 | ❌ | — | +| W8 | **per-服务实例熔断** | 保持共享 downstream 熔断，per-实例 P6 单独评估 | [main.go](../../../services/api-gateway/main.go) L61 `middleware.CircuitBreaker("downstream")` 共享熔断 | ✅ | — | + +**核查结论**：W1/W2/W3/W4/W6/W7 共 6 项未遵循裁决，W5 部分遵循，W8 已遵循。未遵循项需在 P2.0-P2.6 任务中修复（见 [workline.md](../worklines/api-gateway_workline.md) §3）。 + +### 0.3 架构文档与代码不一致问题 + +| # | 文件 | 问题 | 修复建议 | +| --- | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------ | +| D1 | [02-architecture-design.md](../../../services/api-gateway/docs/02-architecture-design.md) §1 | 中间件链图标注顺序 `Auth → CircuitBreaker`，但 [main.go](../../../services/api-gateway/main.go) L61-63 实际顺序为 `CircuitBreaker → Auth` | 统一文档与代码（建议代码保持 CircuitBreaker 在 Auth 之前，避免未鉴权请求消耗下游配额，更新文档图） | +| D2 | [README.md](../../../services/api-gateway/README.md) L38 | 提到 `GET /health` 兼容端点，但 [main.go](../../../services/api-gateway/main.go) 未注册 `/health` 路由 | 删除 README L38 描述（已在 workline P2.0 列出） | +| D3 | [proxy.go](../../../services/api-gateway/internal/proxy/proxy.go) L24 | 连续两次 `TrimPrefix`（L23 去 `/api/v1`，L24 再去 `/api`），第二次无效果（首字符已是 `/`） | 删除 L24 冗余 `TrimPrefix("/api")`（已在 workline P2.0 列出） | +| D4 | [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §3.1 | 提到"P2 起 RS256 公钥校验时通过 IAM `/auth/jwks` 端点拉公钥"，但 [contract.md](../contracts/api-gateway_contract.md) v2 §2.3 已修正为 `GET /.well-known/jwks.json` | 统一为 `/.well-known/jwks.json`（RFC 8414 标准路径） | + +--- + +## §1 问题列表 + + + +### ISSUE-001-ai01：matrix.md gRPC 矩阵错误声明 api-gateway 消费 iam gRPC + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：契约不明确 +- **描述**：[matrix.md](../matrix.md) §2 gRPC 接口提供方矩阵中，iam 的"消费方"列包含 `api-gateway`，暗示 api-gateway 消费 iam gRPC。但依据 [president-final-rulings.md](../../president-final-rulings.md) §2.16 裁决"gateway 保持 HTTP 透传，不改为 gRPC 客户端"，api-gateway 通过 HTTP JWKS 端点（`GET /.well-known/jwks.json`）拉取 RS256 公钥，不消费任何 gRPC。同时 [matrix.md](../matrix.md) §7 Mock 策略表中"api-gateway 消费 iam.GetPublicKey"也需修正为 HTTP JWKS。 +- **建议方案**： + 1. matrix.md §2 iam 行"消费方"列移除 `api-gateway` + 2. matrix.md §7 Mock 策略表中 api-gateway 行改为"硬编码 RS256 公钥（DEV_MODE）→ HTTP JWKS 端点" +- **状态**：待 coord 仲裁 + +### ISSUE-002-ai01：shared-go 接入方式与时机 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：前置依赖缺失 +- **描述**：依据 [president-final-rulings.md](../../president-final-rulings.md) §2.19 裁决，ai01 应直接 import `packages/shared-go`（jwks/logger/tracer/env 4 模块）。但当前 [go.work](../../../go.work) L3-4 仅声明 `./services/api-gateway` 和 `./services/push-gateway`，未包含 `./packages/shared-go`；[go.mod](../../../services/api-gateway/go.mod) 也无 `github.com/edu-cloud/shared-go` 依赖。需确认 shared-go 包骨架是否已建立、模块路径命名、go.work 接入方式。 +- **建议方案**： + 1. coord 确认 `packages/shared-go` 骨架已建立（批次 0.14 任务） + 2. go.work 增加 `./packages/shared-go` + 3. ai01 go.mod 增加 `github.com/edu-cloud/shared-go` 依赖 + 4. P2.1 任务接入 jwks/logger/tracer/env +- **状态**：待 coord 仲裁 + +### ISSUE-003-ai01：API 版本化路由方案（/v1 前缀位置） + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：契约不明确 +- **描述**：依据 [president-final-rulings.md](../../president-final-rulings.md) §2.15 裁决，全服务统一加 `/v1` 前缀，Controller 内加 `v1` 前缀，Gateway 透传时不改路径。但当前 [main.go](../../../services/api-gateway/main.go) 路由为 `/api/v1/iam/*path`（无服务级 `/v1`），而 [contract.md](../contracts/api-gateway_contract.md) v2 §1.2.2 路由表列为 `/api/v1/iam/v1/*path`（方案 A，双 v1）。裁决第 4 条"避免双 v1"与 contract v2 路由表存在矛盾，需 coord 仲裁最终方案： + - 方案 A：Gateway 路由 `/api/v1/iam/v1/*path`，proxy 透传 `/iam/v1/*path`（前端调用 `/api/v1/iam/v1/auth/login`） + - 方案 B：Gateway 路由 `/api/v1/iam/*path`，proxy 透传 `/iam/*path`，下游 Controller 内加 `/v1`（前端调用 `/api/v1/iam/v1/auth/login`，Gateway 不含 `/v1`） + - 方案 C：其他 +- **建议方案**：采纳方案 B（Gateway 路由不含服务级 `/v1`，由下游 Controller 添加），符合裁决第 4 条"避免双 v1"和第 3 条"Gateway 不改路径直接透传" +- **状态**：待 coord 仲裁 + +### ISSUE-004-ai01：zap vs slog 选型 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：契约不明确 +- **描述**：W3 裁决"首次实现即用 slog"，但 [shared-go/logger](../../../packages/shared-go/logger/logger.go) 的实现选型（zap 还是 slog）影响 ai01 接入方式。若 shared-go 用 zap，ai01 需通过 shared-go/logger 间接使用 zap；若 shared-go 用 slog，ai01 可直接用标准库 `log/slog` 或 shared-go 封装。需 coord 仲裁 shared-go/logger 最终选型。 +- **建议方案**：shared-go/logger 基于 `log/slog` 封装（标准库零依赖，W3 裁决原文即 slog），ai01 直接 import shared-go/logger +- **状态**：待 coord 仲裁 + +### ISSUE-005-ai01：go.mod 与 Dockerfile Go 版本不一致 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：其他 +- **描述**：[go.mod](../../../services/api-gateway/go.mod) L3 声明 `go 1.25.0`，[go.work](../../../go.work) L1 声明 `go 1.26.0`，但 [Dockerfile](../../../services/api-gateway/Dockerfile) L1 用 `golang:1.22-alpine`。三者不一致，可能导致构建环境与本地开发环境行为差异。 +- **建议方案**：统一为 `go 1.22`（与 Dockerfile 一致，与 [01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §4 声明一致），或升级 Dockerfile 到 `golang:1.25-alpine` 并同步 go.mod/go.work +- **状态**：待 coord 仲裁 + +### ISSUE-007-ai01：01-understanding.md §6 审计表 metrics 行过时 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：其他 +- **描述**：[01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §6 审计表 metrics 行标注 `❌ 无`，§7.1 issue #1 也说"无 `/metrics` 端点，Prometheus 404"。但实际 [main.go](../../../services/api-gateway/main.go) L56 已注册 `r.GET("/metrics", gin.WrapH(promhttp.Handler()))`，`/metrics` 端点已存在。文档过时，需更新为"⚠️ 端点已注册，但缺 7 个业务指标"。 +- **建议方案**：更新 01-understanding.md §6 审计表 metrics 行为 `⚠️ 端点已注册，缺业务指标`；§7.1 issue #1 更新为"已注册 /metrics，需补 7 个业务指标" +- **状态**：待 coord 仲裁 + +### ISSUE-008-ai01：01-understanding.md §7.1 issue #3 死代码引用过时 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：其他 +- **描述**：[01-understanding.md](../../../services/api-gateway/docs/01-understanding.md) §7.1 issue #3 引用"auth.go L124-139 死代码 `RequestIDMiddleware()` + `generateUUID()`"，但实际 [auth.go](../../../services/api-gateway/internal/middleware/auth.go) 当前仅 121 行，L124-139 不存在，死代码已删除。文档引用过时。 +- **建议方案**：删除 01-understanding.md §7.1 issue #3 整行（死代码已修复） +- **状态**：待 coord 仲裁 + +### ISSUE-009-ai01：W1/W2 错误码 GW_ 前缀 + ActionState 信封未遵循 + +- **提请方**：ai01 +- **日期**：2026-07-10 +- **类型**：其他 +- **描述**：W1 裁决要求错误码统一 `GW_` 前缀，W2 裁决要求错误响应统一 ActionState 信封 `{success:false,error:{code,message}}`。但当前代码： + - [auth.go](../../../services/api-gateway/internal/middleware/auth.go) 错误码用 `UNAUTHORIZED`/`INVALID_TOKEN`/`INVALID_CLAIMS`（无 GW_ 前缀），但响应体已部分对齐信封（L44-50 有 `success`/`error.code`/`error.message`） + - [ratelimit.go](../../../services/api-gateway/internal/middleware/ratelimit.go) L51-54 响应 `{"error":"rate_limited","retry_after":60}`（无 GW_ 前缀，无信封） + - [circuit-breaker.go](../../../services/api-gateway/internal/middleware/circuit-breaker.go) L55-58 响应 `{"error":"circuit_open","retry_after":30}`（无 GW_ 前缀，无信封） + - [recovery.go](../../../services/api-gateway/internal/middleware/recovery.go) L24-27 响应 `{"error":"internal_error","request_id":"..."}`（无 GW_ 前缀，无信封） +- **建议方案**：P2.3 任务统一修复（见 [workline.md](../worklines/api-gateway_workline.md) P2.3），所有错误码改为 `GW_*`，所有响应体改为 `{success:false,error:{code:"GW_*",message:"..."},...}` +- **状态**：待 coord 仲裁 + +--- + +## §2 待 coord 仲裁事项汇总 + +| ISSUE | 主题 | 阻塞任务 | 状态 | +| -------- | --------------------------------- | -------- | --------- | +| ISSUE-001 | matrix.md gRPC 矩阵修正 | — | 待 coord | +| ISSUE-002 | shared-go 接入方式 | P2.1 | 待 coord | +| ISSUE-003 | API 版本化路由方案（/v1 前缀） | P2.7 | 待 coord | +| ISSUE-004 | zap vs slog 选型 | P2.1 | 待 coord | +| ISSUE-005 | go.mod 与 Dockerfile 版本不一致 | P2.0 | 待 coord | +| ISSUE-007 | 01-understanding.md §6 审计表过时 | P2.0 | 待 coord | +| ISSUE-008 | 01-understanding.md §7.1 死代码过时 | P2.0 | 待 coord | +| ISSUE-009 | W1/W2 错误码 GW_ + ActionState | P2.3 | 待 coord | + +> ISSUE-006 编号未使用（预留）。 From 9e767b4e95a008b1380b6ba5f857318254ccb29f Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Fri, 10 Jul 2026 16:50:33 +0800 Subject: [PATCH 3/3] chore(api-gateway): merge review docs and update dependencies --- go.work | 1 + go.work.sum | 57 ++- packages/shared-go/go.mod | 20 +- services/push-gateway/go.mod | 9 +- .../push-gateway/internal/config/config.go | 112 ++++-- services/push-gateway/internal/hub/hub.go | 334 +++++++++++++--- .../internal/kafkaconsumer/consumer.go | 214 ++++++++++ .../internal/observability/logger.go | 77 ++++ .../internal/observability/metrics.go | 124 ++++++ .../internal/observability/tracer.go | 80 +--- .../internal/redisclient/pubsub.go | 370 ++++++++++++++++++ 11 files changed, 1246 insertions(+), 152 deletions(-) create mode 100644 services/push-gateway/internal/kafkaconsumer/consumer.go create mode 100644 services/push-gateway/internal/observability/logger.go create mode 100644 services/push-gateway/internal/observability/metrics.go create mode 100644 services/push-gateway/internal/redisclient/pubsub.go diff --git a/go.work b/go.work index bc8a5a6..2acfd4e 100644 --- a/go.work +++ b/go.work @@ -1,6 +1,7 @@ go 1.26.0 use ( + ./packages/shared-go ./services/api-gateway ./services/push-gateway ) diff --git a/go.work.sum b/go.work.sum index 271c4fc..6123523 100644 --- a/go.work.sum +++ b/go.work.sum @@ -1,35 +1,84 @@ cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0= +github.com/alecthomas/kingpin/v2 v2.4.0/go.mod h1:0gyi0zQnjuFk8xrkNKamJoyUo382HRL7ATRpFZCw6tE= +github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE= github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY= github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4= github.com/envoyproxy/go-control-plane v0.14.0/go.mod h1:NcS5X47pLl/hfqxU70yPwL9ZMkUlwlKxtAohpi2wBEU= github.com/envoyproxy/go-control-plane/envoy v1.37.0/go.mod h1:DReE9MMrmecPy+YvQOAOHNYMALuowAnbjjEMkkWOi6A= github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4= github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0= +github.com/go-chi/chi/v5 v5.0.12/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/golang/glog v1.2.5/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/jordanlewis/gcassert v0.0.0-20250430164644-389ef753e22e/go.mod h1:ZybsQk6DWyN5t7An1MuPm1gtSZ1xDaTXS9ZjIOxvQrk= +github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= +github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM= -github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= -github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/segmentio/kafka-go v0.4.48/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg= github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4= github.com/xdg-go/scram v1.2.0/go.mod h1:3dlrS0iBaWKYVt2ZfA4cj48umJZ+cAEbR6/SjLA88I8= github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= +github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/contrib/detectors/gcp v1.42.0/go.mod h1:W9zQ439utxymRrXsUOzZbFX4JhLxXU4+ZnCt8GG7yA8= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= +golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= +golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U= golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/packages/shared-go/go.mod b/packages/shared-go/go.mod index 8570fce..75e68b1 100644 --- a/packages/shared-go/go.mod +++ b/packages/shared-go/go.mod @@ -3,6 +3,24 @@ module github.com/edu-cloud/shared-go go 1.22 require ( + github.com/golang-jwt/jwt/v5 v5.2.1 + go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/trace v1.44.0 go.uber.org/zap v1.27.0 - github.com/go-chi/chi/v5 v5.0.12 +) + +require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect + golang.org/x/sys v0.45.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.81.1 // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/services/push-gateway/go.mod b/services/push-gateway/go.mod index 34d3d71..ec9341a 100644 --- a/services/push-gateway/go.mod +++ b/services/push-gateway/go.mod @@ -3,14 +3,19 @@ module github.com/edu-cloud/push-gateway go 1.25.0 require ( + github.com/edu-cloud/shared-go v0.0.0 github.com/gin-gonic/gin v1.12.0 github.com/golang-jwt/jwt/v5 v5.2.1 + github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 github.com/prometheus/client_golang v1.23.2 + github.com/redis/go-redis/v9 v9.7.0 + github.com/segmentio/kafka-go v0.4.48 go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.69.0 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 + go.uber.org/zap v1.27.0 ) require ( @@ -21,6 +26,7 @@ require ( github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.7 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/gin-contrib/sse v1.1.1 // indirect github.com/go-logr/logr v1.4.3 // indirect @@ -30,7 +36,6 @@ require ( github.com/go-playground/validator/v10 v10.30.2 // indirect github.com/goccy/go-json v0.10.6 // indirect github.com/goccy/go-yaml v1.19.2 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect @@ -64,3 +69,5 @@ require ( google.golang.org/grpc v1.81.1 // indirect google.golang.org/protobuf v1.36.11 // indirect ) + +replace github.com/edu-cloud/shared-go => ../../packages/shared-go diff --git a/services/push-gateway/internal/config/config.go b/services/push-gateway/internal/config/config.go index 7b835a6..e4e1b86 100644 --- a/services/push-gateway/internal/config/config.go +++ b/services/push-gateway/internal/config/config.go @@ -1,28 +1,50 @@ +// 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. package config import ( "log" "os" + "strings" + + "github.com/edu-cloud/shared-go/env" ) -// Config 持有 push-gateway 运行时配置 +// Config holds all push-gateway runtime configuration. type Config struct { - Port string - JWTSecret string - DevMode bool - RedisURL string - OTLPEndpoint string - InternalAPIKey string + Port string + JWTSecret string + DevMode bool + RedisURL string + OTLPEndpoint string + InternalAPIToken string + // WebSocket + AllowedOrigins []string + MaxConnsPerUser int + HeartbeatInterval int // seconds + // JWT RS256 (via shared-go/jwks, iam /.well-known/jwks.json) + JWKSURL string + // Kafka + KafkaBrokers []string + KafkaNotificationTopic string + KafkaConsumerGroup string + // Instance identity (for Redis SET membership) + InstanceID string } -// devJWTSecret 是 DevMode 下的默认 JWT 密钥（仅用于本地联调，生产必须配置 JWT_SECRET） +// devJWTSecret is the default JWT secret used only in DevMode for local +// development. Production deployments MUST set JWT_SECRET. const devJWTSecret = "p1-dev-secret-change-in-production" -// Load 从环境变量加载配置。 -// 非 DevMode 下若 JWT_SECRET 未配置则 fatal 退出；DevMode 下使用默认密钥并打印 warning。 +// Load reads configuration from environment variables. +// +// In DevMode (DEV_MODE=true) JWT_SECRET defaults to a dev value and internal +// API token validation is skipped; in production an unset JWT_SECRET or +// INTERNAL_API_TOKEN is fatal. func Load() *Config { - devMode := getEnv("DEV_MODE", "false") == "true" - jwtSecret := getEnv("JWT_SECRET", "") + devMode := env.GetBool("DEV_MODE", false) + jwtSecret := env.Get("JWT_SECRET", "") if jwtSecret == "" { if devMode { log.Println("warning: JWT_SECRET not set, using dev default (DEV_MODE=true)") @@ -32,20 +54,64 @@ func Load() *Config { } } + internalToken := env.Get("INTERNAL_API_TOKEN", "") + if internalToken == "" && !devMode { + log.Fatal("INTERNAL_API_TOKEN must be set in non-dev mode") + } + return &Config{ - Port: getEnv("PUSH_GATEWAY_PORT", "8081"), - JWTSecret: jwtSecret, - DevMode: devMode, - RedisURL: getEnv("REDIS_URL", ""), - OTLPEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318"), - InternalAPIKey: getEnv("INTERNAL_API_KEY", ""), + Port: env.Get("PUSH_GATEWAY_PORT", "8081"), + JWTSecret: jwtSecret, + DevMode: devMode, + RedisURL: env.Get("REDIS_URL", "redis://localhost:6379/0"), + OTLPEndpoint: env.Get("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318"), + InternalAPIToken: internalToken, + AllowedOrigins: parseOrigins(env.Get("WS_ALLOWED_ORIGINS", "")), + MaxConnsPerUser: env.GetInt("MAX_CONNS_PER_USER", 5), + 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"), + KafkaConsumerGroup: env.Get("KAFKA_CONSUMER_GROUP", "push-gateway"), + InstanceID: env.Get("INSTANCE_ID", generateInstanceID()), } } -// getEnv 读取环境变量，缺失时返回 fallback -func getEnv(key, fallback string) string { - if v := os.Getenv(key); v != "" { - return v +// parseOrigins splits a comma-separated list of allowed WebSocket origins. +// Empty string yields nil (all origins rejected in production). +func parseOrigins(raw string) []string { + if raw == "" { + return nil } - return fallback + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + for _, p := range parts { + if trimmed := strings.TrimSpace(p); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +// parseBrokers splits a comma-separated list of Kafka broker addresses. +func parseBrokers(raw string) []string { + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + for _, p := range parts { + if trimmed := strings.TrimSpace(p); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +// generateInstanceID returns a stable process-unique identifier. Falls back to +// the hostname when INSTANCE_ID is not explicitly set, allowing each replica to +// be uniquely identifiable in the Redis online-presence SET. +func generateInstanceID() string { + host, err := os.Hostname() + if err != nil || host == "" { + return "push-gateway-unknown" + } + return host } diff --git a/services/push-gateway/internal/hub/hub.go b/services/push-gateway/internal/hub/hub.go index e4c44e5..f667a26 100644 --- a/services/push-gateway/internal/hub/hub.go +++ b/services/push-gateway/internal/hub/hub.go @@ -1,47 +1,151 @@ +// Package hub manages the in-memory WebSocket connection pool for push-gateway. +// +// Hub is the single process-wide registry of all live WebSocket connections, +// indexed by userID -> connID -> *Connection. Each Connection owns a send +// channel consumed by a dedicated writer goroutine; Hub exposes Register / +// Unregister / SendToUser / Broadcast operations. +// +// The Hub also tracks per-user connection counts to enforce the +// MaxConnectionsPerUser limit (default 5) and provides CloseAll for graceful +// shutdown. Cross-instance presence and message fanout are delegated to a +// RedisClient (see internal/redis), keeping Hub free of any persistence layer. package hub import ( + "errors" "sync" + "time" "github.com/gorilla/websocket" + "github.com/google/uuid" ) -// Connection 包装单个 WebSocket 连接及其异步发送通道 +// ErrTooManyConnections is returned by Register when the per-user connection +// cap has been reached. Callers should translate this into a 429 / +// PUSH_TOO_MANY_CONNECTIONS response or a close frame with code 1008. +var ErrTooManyConnections = errors.New("too many connections per user") + +// ErrHubClosing is returned by Register when the Hub is shutting down and no +// new connections are accepted. +var ErrHubClosing = errors.New("hub is closing") + +// sendBufferSize is the per-connection send channel capacity. A full channel +// triggers message drop with metric increment (see Send). +const sendBufferSize = 64 + +// Connection wraps a single WebSocket connection together with its asynchronous +// send channel and lifecycle metadata. type Connection struct { - UserID string - conn *websocket.Conn - send chan []byte + UserID string + ConnID string + conn *websocket.Conn + send chan []byte + closeOnce sync.Once + closed bool + mu sync.Mutex + lastPing time.Time } -// Send 将消息投递到该连接的发送通道（非阻塞，通道满则丢弃） -func (c *Connection) Send(message []byte) { +// Send delivers message to the connection's writer goroutine. When the send +// channel is full the message is dropped (non-blocking) and the caller is +// notified via the returned false value so it can increment the +// push_gateway_messages_dropped_total{reason="channel_full"} metric. +func (c *Connection) Send(message []byte) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return false + } select { case c.send <- message: + return true default: - // 通道满，丢弃消息避免阻塞 hub + return false } } -// Outgoing 返回该连接的发送通道，供写协程消费 +// Outgoing exposes the send channel for the writer goroutine. The channel is +// closed by Close, terminating the consumer loop. func (c *Connection) Outgoing() <-chan []byte { return c.send } -// Hub 管理所有在线 WebSocket 连接，按 userID 索引 -type Hub struct { - mu sync.RWMutex - clients map[string]map[*Connection]bool +// Close shuts down the connection's send channel exactly once. Subsequent +// Send calls become no-ops. +func (c *Connection) Close() { + c.closeOnce.Do(func() { + c.mu.Lock() + c.closed = true + c.mu.Unlock() + close(c.send) + }) } -// NewHub 创建 Hub 实例 -func NewHub() *Hub { +// LastPing returns the most recent heartbeat timestamp. +func (c *Connection) LastPing() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.lastPing +} + +// TouchPing updates the last heartbeat timestamp to now. +func (c *Connection) TouchPing() { + c.mu.Lock() + defer c.mu.Unlock() + c.lastPing = time.Now() +} + +// Conn returns the underlying gorilla websocket.Conn for control-frame +// operations (Ping/Pong/Close). Used by the reader goroutine. +func (c *Connection) Conn() *websocket.Conn { + return c.conn +} + +// Hub is the process-wide registry of live WebSocket connections. +// +// clients maps userID -> connID -> *Connection so a single user may hold +// multiple simultaneous connections (multi-device). counters caches the +// per-user connection count to make the MaxConnectionsPerUser check O(1). +type Hub struct { + mu sync.RWMutex + clients map[string]map[string]*Connection + counters map[string]int + closing bool + + // maxConnsPerUser caps the number of simultaneous connections per user. + maxConnsPerUser int + + // onRegister / onUnregister are optional hooks used by the Redis presence + // layer to maintain the cross-instance online SET. They are invoked outside + // the Hub lock to avoid blocking the registry on Redis round-trips. + onRegister func(userID, connID string) + onUnregister func(userID, connID string) +} + +// NewHub returns a Hub with the given per-user connection cap. +func NewHub(maxConnsPerUser int) *Hub { + if maxConnsPerUser <= 0 { + maxConnsPerUser = 5 + } return &Hub{ - clients: make(map[string]map[*Connection]bool), + clients: make(map[string]map[string]*Connection), + counters: make(map[string]int), + maxConnsPerUser: maxConnsPerUser, } } -// ClientCount 返回当前在线连接总数（供 /readyz 探针使用） -func (h *Hub) ClientCount() int { +// SetPresenceHooks registers callbacks invoked after Register / Unregister. +// Used by the Redis presence layer to keep edu:push:online: in sync. +func (h *Hub) SetPresenceHooks(onRegister, onUnregister func(userID, connID string)) { + h.mu.Lock() + defer h.mu.Unlock() + h.onRegister = onRegister + h.onUnregister = onUnregister +} + +// ActiveConnections returns the total number of live connections across all +// users. Exposed via /healthz and the push_gateway_active_connections gauge. +func (h *Hub) ActiveConnections() int { h.mu.RLock() defer h.mu.RUnlock() count := 0 @@ -51,60 +155,162 @@ func (h *Hub) ClientCount() int { return count } -// Register 注册一个用户连接，返回 Connection 供调用方持有 -func (h *Hub) Register(userID string, conn *websocket.Conn) *Connection { - c := &Connection{ - UserID: userID, - conn: conn, - send: make(chan []byte, 64), - } - h.mu.Lock() - defer h.mu.Unlock() - if h.clients[userID] == nil { - h.clients[userID] = make(map[*Connection]bool) - } - h.clients[userID][c] = true - return c +// UserCount returns the number of distinct online users (used by /healthz). +func (h *Hub) UserCount() int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.clients) } -// Unregister 注销一个用户连接并关闭其发送通道 -func (h *Hub) Unregister(c *Connection) { - h.mu.Lock() - defer h.mu.Unlock() - conns, ok := h.clients[c.UserID] - if !ok { - return - } - if _, exists := conns[c]; exists { - delete(conns, c) - close(c.send) - } - if len(conns) == 0 { - delete(h.clients, c.UserID) - } -} - -// SendToUser 向指定用户的所有在线连接推送消息 -func (h *Hub) SendToUser(userID string, message []byte) error { +// HasUser reports whether at least one local connection exists for userID. +// Used by the /internal/push path to decide between local delivery and Redis +// Pub/Sub fanout. +func (h *Hub) HasUser(userID string) bool { h.mu.RLock() defer h.mu.RUnlock() conns, ok := h.clients[userID] - if !ok { - return nil - } - for c := range conns { - c.Send(message) - } - return nil + return ok && len(conns) > 0 } -// Broadcast 向所有在线连接广播消息 -func (h *Hub) Broadcast(message []byte) { - h.mu.RLock() - defer h.mu.RUnlock() - for _, conns := range h.clients { - for c := range conns { - c.Send(message) +// Register adds a new connection for userID. Returns ErrTooManyConnections +// when the per-user cap is exceeded and ErrHubClosing when the Hub is shutting +// down. The returned Connection must be Unregistered by the caller on close. +func (h *Hub) Register(userID string, conn *websocket.Conn) (*Connection, error) { + h.mu.Lock() + if h.closing { + h.mu.Unlock() + return nil, ErrHubClosing + } + if h.counters[userID] >= h.maxConnsPerUser { + h.mu.Unlock() + return nil, ErrTooManyConnections + } + connID := uuid.NewString() + c := &Connection{ + UserID: userID, + ConnID: connID, + conn: conn, + send: make(chan []byte, sendBufferSize), + lastPing: time.Now(), + } + if h.clients[userID] == nil { + h.clients[userID] = make(map[string]*Connection) + } + h.clients[userID][connID] = c + h.counters[userID]++ + onRegister := h.onRegister + h.mu.Unlock() + + if onRegister != nil { + onRegister(userID, connID) + } + return c, nil +} + +// Unregister removes a connection and closes its send channel. Safe to call +// multiple times for the same Connection. +func (h *Hub) Unregister(c *Connection) { + h.mu.Lock() + conns, ok := h.clients[c.UserID] + if !ok { + h.mu.Unlock() + return + } + if _, exists := conns[c.ConnID]; exists { + delete(conns, c.ConnID) + h.counters[c.UserID]-- + if h.counters[c.UserID] <= 0 { + delete(h.counters, c.UserID) + } + if len(conns) == 0 { + delete(h.clients, c.UserID) } } + onUnregister := h.onUnregister + h.mu.Unlock() + + c.Close() + if onUnregister != nil { + onUnregister(c.UserID, c.ConnID) + } +} + +// SendToUser delivers message to every local connection of userID. Returns +// the number of connections the message was successfully queued to. A zero +// return value signals "user not online locally" so the caller can fall back +// to Redis Pub/Sub fanout. +func (h *Hub) SendToUser(userID string, message []byte) int { + h.mu.RLock() + conns := h.clients[userID] + // Snapshot the slice under the read lock to avoid holding it during Send. + snapshot := make([]*Connection, 0, len(conns)) + for _, c := range conns { + snapshot = append(snapshot, c) + } + h.mu.RUnlock() + + delivered := 0 + for _, c := range snapshot { + if c.Send(message) { + delivered++ + } + } + return delivered +} + +// Broadcast delivers message to every live connection in this instance. +// Cross-instance broadcast is handled separately via Redis Pub/Sub. +func (h *Hub) Broadcast(message []byte) int { + h.mu.RLock() + snapshot := make([]*Connection, 0, h.ActiveConnections()) + for _, conns := range h.clients { + for _, c := range conns { + snapshot = append(snapshot, c) + } + } + h.mu.RUnlock() + + delivered := 0 + for _, c := range snapshot { + if c.Send(message) { + delivered++ + } + } + return delivered +} + +// CloseAll marks the Hub as closing (rejecting further Register calls) and +// sends a close frame to every live connection. Used during graceful shutdown. +// The caller should wait for connections to drain after CloseAll returns. +func (h *Hub) CloseAll() { + h.mu.Lock() + h.closing = true + all := make([]*Connection, 0, h.ActiveConnections()) + for _, conns := range h.clients { + for _, c := range conns { + all = append(all, c) + } + } + h.mu.Unlock() + + for _, c := range all { + // Send a graceful close frame (1001 going away) and close the send + // channel so the writer goroutine exits. + _ = c.conn.WriteControl( + websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseGoingAway, "server shutting down"), + time.Now().Add(2*time.Second), + ) + c.Close() + } +} + +// ForEachUser iterates over all locally online userIDs. Used by the Redis +// presence layer to rebuild the online SET on startup (ISSUE-058). +func (h *Hub) ForEachUser(fn func(userID string)) { + h.mu.RLock() + defer h.mu.RUnlock() + for userID := range h.clients { + fn(userID) + } } diff --git a/services/push-gateway/internal/kafkaconsumer/consumer.go b/services/push-gateway/internal/kafkaconsumer/consumer.go new file mode 100644 index 0000000..fdc835e --- /dev/null +++ b/services/push-gateway/internal/kafkaconsumer/consumer.go @@ -0,0 +1,214 @@ +// Package kafkaconsumer subscribes to the edu.notification.requested topic +// (ISSUE-053 final 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). +package kafkaconsumer + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "time" + + "github.com/edu-cloud/push-gateway/internal/hub" + "github.com/edu-cloud/push-gateway/internal/observability" + "github.com/edu-cloud/push-gateway/internal/redisclient" + "github.com/segmentio/kafka-go" +) + +// MaxRetries is the per-message retry count before sending to the dead-letter +// topic. After MaxRetries failures the message is committed (skipped) and +// logged; a future P6 enhancement can republish to a DLQ topic. +const MaxRetries = 3 + +// NotificationRequested is the message shape produced by msg (ai10). Field +// names mirror the JSON wire format produced by msg's Outbox publisher. +type NotificationRequested struct { + EventID string `json:"event_id"` + UserID string `json:"user_id"` + EventType string `json:"event_type"` + Channel string `json:"channel"` + Title string `json:"title"` + Content string `json:"content"` + Data json.RawMessage `json:"data"` + Broadcast bool `json:"broadcast"` + OccurredAt int64 `json:"occurred_at"` +} + +// Consumer wraps a kafka.Reader for the notification topic. +type Consumer struct { + reader *kafka.Reader + hub *hub.Hub + redis *redisclient.Client + metrics *observability.Metrics + topic string + groupID string + dlqTopic string +} + +// Config configures the Kafka consumer. +type Config struct { + Brokers []string + Topic string + GroupID string + DLQTopic string +} + +// New creates a Consumer for the configured topic and group. +func New(cfg Config, h *hub.Hub, r *redisclient.Client, m *observability.Metrics) *Consumer { + if cfg.DLQTopic == "" { + cfg.DLQTopic = cfg.Topic + ".dlq" + } + reader := kafka.NewReader(kafka.ReaderConfig{ + Brokers: cfg.Brokers, + Topic: cfg.Topic, + GroupID: cfg.GroupID, + MinBytes: 1, + MaxBytes: 10 * 1024 * 1024, + CommitInterval: 1 * time.Second, // periodic background commit + StartOffset: kafka.LastOffset, // skip historical backlog on first start + }) + return &Consumer{ + reader: reader, + hub: h, + redis: r, + metrics: m, + topic: cfg.Topic, + groupID: cfg.GroupID, + dlqTopic: cfg.DLQTopic, + } +} + +// Run blocks until ctx is canceled, consuming messages and dispatching them. +// Errors are logged but do not stop the consumer unless ctx is canceled. +func (c *Consumer) Run(ctx context.Context) error { + observability.Logger().Info("kafka consumer starting", + "topic", c.topic, "group", c.groupID, "brokers", c.reader.Config().Brokers) + for { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + msg, err := c.reader.ReadMessage(ctx) + if err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) { + return nil + } + observability.Logger().Warn("kafka read failed", "err", err) + select { + case <-ctx.Done(): + return nil + case <-time.After(2 * time.Second): + } + continue + } + c.metrics.IncKafkaConsumed(c.topic, fmt.Sprintf("%d", msg.Partition)) + c.processMessage(ctx, msg) + } +} + +// Close releases the underlying reader. Should be called after Run returns. +func (c *Consumer) Close() error { + return c.reader.Close() +} + +// processMessage handles a single Kafka message with retry and idempotency. +func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) { + var event NotificationRequested + 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") + // Commit to skip the poison message. + _ = c.reader.CommitMessages(ctx, msg) + return + } + + // Idempotency: skip already-processed events by event_id. + fresh, err := c.redis.DedupEventId(ctx, event.EventID) + if err != nil { + observability.Logger().Warn("kafka: dedup check failed, processing anyway", + "event_id", event.EventID, "err", err) + } + if !fresh { + observability.Logger().Debug("kafka: duplicate event skipped", + "event_id", event.EventID) + c.metrics.IncPushed("notification.requested", "duplicate") + _ = c.reader.CommitMessages(ctx, msg) + return + } + + // 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.reader.CommitMessages(ctx, msg) + return + } else if attempt < MaxRetries { + observability.Logger().Warn("kafka: dispatch retry", + "event_id", event.EventID, "attempt", attempt, "err", err) + time.Sleep(time.Duration(attempt) * time.Second) + } + } + // MaxRetries exhausted: send to DLQ (best-effort) and commit. + 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.reader.CommitMessages(ctx, msg) +} + +// dispatch delivers the event to local Hub connections and, when the user is +// not local, asks Redis to fan out via Pub/Sub. +func (c *Consumer) dispatch(ctx context.Context, event NotificationRequested) error { + payload, err := json.Marshal(map[string]any{ + "type": "message", + "event": event.EventType, + "data": event.Data, + "timestamp": time.Now().UTC().Format(time.RFC3339), + }) + if err != nil { + return fmt.Errorf("marshal payload: %w", err) + } + + if event.Broadcast { + // Broadcast: deliver locally + publish to other instances. + c.hub.Broadcast(payload) + return c.redis.PublishBroadcast(ctx, redisclient.CrossInstanceMessage{ + Event: event.EventType, + Data: event.Data, + }) + } + + // Directed: try local first, fall back to Redis Pub/Sub. + if c.hub.HasUser(event.UserID) { + if c.hub.SendToUser(event.UserID, payload) > 0 { + return nil + } + } + return c.redis.PublishUser(ctx, event.UserID, redisclient.CrossInstanceMessage{ + Event: event.EventType, + Data: event.Data, + }) +} + +// sendToDLQ publishes a raw message to the dead-letter topic. Best-effort; +// failure is logged but does not block commit. +func (c *Consumer) sendToDLQ(ctx context.Context, value []byte) { + conn, err := kafka.DialLeader(ctx, "tcp", c.reader.Config().Brokers[0], c.dlqTopic, 0) + if err != nil { + observability.Logger().Error("kafka: DLQ dial failed", "topic", c.dlqTopic, "err", err) + return + } + defer conn.Close() + _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) + _, _ = conn.WriteMessages(kafka.Message{Value: value}) +} diff --git a/services/push-gateway/internal/observability/logger.go b/services/push-gateway/internal/observability/logger.go new file mode 100644 index 0000000..4d795cc --- /dev/null +++ b/services/push-gateway/internal/observability/logger.go @@ -0,0 +1,77 @@ +// Package observability provides slog-based structured logging for push-gateway. +// +// The logger emits JSON in production and a human-readable console format in +// development (DEV_MODE=true). Trace correlation is handled by attaching a +// trace_id attribute from the OpenTelemetry span context when present. +package observability + +import ( + "context" + "log/slog" + "os" + "time" + + "go.opentelemetry.io/otel/trace" +) + +// ctxKey is an unexported type so context values cannot collide with callers. +type ctxKey struct{} + +// logger is the process-wide *slog.Logger initialized by InitLogger. +var logger *slog.Logger + +// InitLogger configures the global slog logger. In DevMode a console handler +// with colored levels is used; otherwise a JSON handler at Info level is used. +// The logger is tagged with service=push-gateway on every emitted record. +func InitLogger(devMode bool) *slog.Logger { + var handler slog.Handler + opts := &slog.HandlerOptions{ + Level: slog.LevelInfo, + } + if devMode { + opts.Level = slog.LevelDebug + handler = slog.NewTextHandler(os.Stdout, opts) + } else { + handler = slog.NewJSONHandler(os.Stdout, opts) + } + logger = slog.New(handler).With("service", "push-gateway") + slog.SetDefault(logger) + return logger +} + +// Logger returns the process-wide logger initialized by InitLogger. When Init +// has not been called a fallback JSON logger is returned so callers never +// receive a nil logger. +func Logger() *slog.Logger { + if logger == nil { + logger = slog.New(slog.NewJSONHandler(os.Stdout, nil)).With("service", "push-gateway") + } + return logger +} + +// WithContext returns a copy of ctx carrying l so it can later be retrieved +// via FromContext. +func WithContext(ctx context.Context, l *slog.Logger) context.Context { + return context.WithValue(ctx, ctxKey{}, l) +} + +// FromContext returns the logger stored in ctx. When an OpenTelemetry span is +// active the returned logger is decorated with a trace_id field so log lines +// can be correlated to traces. If no logger was stored in ctx the +// process-wide logger is returned (never nil). +func FromContext(ctx context.Context) *slog.Logger { + l, ok := ctx.Value(ctxKey{}).(*slog.Logger) + if !ok || l == nil { + l = Logger() + } + if sc := trace.SpanContextFromContext(ctx); sc.HasTraceID() { + return l.With("trace_id", sc.TraceID().String()) + } + return l +} + +// Now returns the current UTC time in RFC3339 format. Convenience helper for +// log field construction. +func Now() string { + return time.Now().UTC().Format(time.RFC3339) +} diff --git a/services/push-gateway/internal/observability/metrics.go b/services/push-gateway/internal/observability/metrics.go new file mode 100644 index 0000000..e011ba8 --- /dev/null +++ b/services/push-gateway/internal/observability/metrics.go @@ -0,0 +1,124 @@ +// Package observability also exposes Prometheus metrics for push-gateway. +// +// The metrics cover the full surface area described in 02 §6.4: active +// connections, messages pushed/dropped, heartbeats, disconnects, Redis Pub/Sub +// latency, Kafka consumption and Redis SET rebuilds. All metrics are +// registered with the global prometheus.DefaultRegisterer on package init. +package observability + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// Metrics holds all push-gateway Prometheus metric handles. +type Metrics struct { + ActiveConnections prometheus.Gauge + MessagesPushed *prometheus.CounterVec + MessagesDropped *prometheus.CounterVec + Heartbeats prometheus.Counter + Disconnects *prometheus.CounterVec + RedisPubSubLatency *prometheus.HistogramVec + KafkaConsumed *prometheus.CounterVec + RedisSetRebuild prometheus.Counter + ConnectionsPerUser prometheus.Gauge +} + +// metrics is the process-wide metric set, initialized once by NewMetrics. +var metrics *Metrics + +// NewMetrics registers and returns the push-gateway metric set. Subsequent +// calls return the same instance to avoid duplicate-registration panics. +func NewMetrics() *Metrics { + if metrics != nil { + return metrics + } + metrics = &Metrics{ + ActiveConnections: promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: "push_gateway", + Name: "active_connections", + Help: "Current number of live WebSocket connections.", + }), + MessagesPushed: promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "messages_pushed_total", + Help: "Total messages pushed, partitioned by event and result.", + }, []string{"event", "result"}), + MessagesDropped: promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "messages_dropped_total", + Help: "Messages dropped, partitioned by reason.", + }, []string{"reason"}), + Heartbeats: promauto.NewCounter(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "heartbeat_total", + Help: "Total WebSocket Ping heartbeats received.", + }), + Disconnects: promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "disconnect_total", + Help: "Disconnects, partitioned by reason.", + }, []string{"reason"}), + RedisPubSubLatency: promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "push_gateway", + Name: "redis_pubsub_latency_seconds", + Help: "Redis Pub/Sub round-trip latency in seconds.", + Buckets: prometheus.ExponentialBuckets(0.001, 2, 12), + }, []string{"direction"}), + KafkaConsumed: promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "kafka_consumed_total", + Help: "Kafka messages consumed, partitioned by topic and partition.", + }, []string{"topic", "partition"}), + RedisSetRebuild: promauto.NewCounter(prometheus.CounterOpts{ + Namespace: "push_gateway", + Name: "redis_set_rebuild_total", + Help: "Total Redis online-presence SET rebuilds (startup or recovery).", + }), + ConnectionsPerUser: promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: "push_gateway", + Name: "connections_per_user_max", + Help: "Maximum concurrent connections held by a single user.", + }), + } + return metrics +} + +// Metrics returns the process-wide metric set. When NewMetrics has not been +// called a new set is registered on demand. +func MetricsInstance() *Metrics { + if metrics == nil { + return NewMetrics() + } + return metrics +} + +// IncMessageDropped is a convenience helper for the channel-full case. +func (m *Metrics) IncMessageDropped(reason string) { + m.MessagesDropped.WithLabelValues(reason).Inc() +} + +// IncPushed is a convenience helper for the push result counter. +func (m *Metrics) IncPushed(event, result string) { + m.MessagesPushed.WithLabelValues(event, result).Inc() +} + +// IncDisconnect is a convenience helper for the disconnect counter. +func (m *Metrics) IncDisconnect(reason string) { + m.Disconnects.WithLabelValues(reason).Inc() +} + +// ObservePubSubLatency records a Redis Pub/Sub round-trip duration. +func (m *Metrics) ObservePubSubLatency(direction string, seconds float64) { + m.RedisPubSubLatency.WithLabelValues(direction).Observe(seconds) +} + +// IncKafkaConsumed increments the Kafka consumed counter. +func (m *Metrics) IncKafkaConsumed(topic, partition string) { + m.KafkaConsumed.WithLabelValues(topic, partition).Inc() +} + +// IncRedisSetRebuild increments the SET rebuild counter (ISSUE-058 metric). +func (m *Metrics) IncRedisSetRebuild() { + m.RedisSetRebuild.Inc() +} diff --git a/services/push-gateway/internal/observability/tracer.go b/services/push-gateway/internal/observability/tracer.go index 3d1861c..9bb95c8 100644 --- a/services/push-gateway/internal/observability/tracer.go +++ b/services/push-gateway/internal/observability/tracer.go @@ -1,73 +1,35 @@ +// Package observability also bootstraps OpenTelemetry tracing. The tracer is +// initialized via shared-go/tracer so push-gateway and api-gateway share the +// same SDK configuration. package observability import ( "context" - "log" - "net/url" - "time" + "fmt" + "os" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" - "go.opentelemetry.io/otel/propagation" - "go.opentelemetry.io/otel/sdk/resource" - sdktrace "go.opentelemetry.io/otel/sdk/trace" - semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + "github.com/edu-cloud/shared-go/tracer" ) -// InitTracer 初始化 OpenTelemetry tracer. -// -// endpoint 为 "http://host:port" 格式（如 "http://localhost:4318"）； -// 为空时跳过初始化（tracing disabled）。 -// -// 返回 shutdown 函数，应在服务退出时调用以 flush 待发送 span. +// InitTracer initializes the OTel tracer using shared-go/tracer. When the +// OTLP endpoint is empty (or "localhost:4318") the exporter still runs but +// emits no spans if no collector is listening. Returns a shutdown function +// that must be called on process exit. func InitTracer(serviceName, endpoint string) func() { - if endpoint == "" { - log.Println("OTEL endpoint not set, tracing disabled") + // shared-go/tracer reads OTEL_EXPORTER_OTLP_ENDPOINT directly; override it + // explicitly so push-gateway's own config value takes precedence. + if endpoint != "" { + _ = os.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint) + } + if err := tracer.Init(serviceName); err != nil { + // Tracing is best-effort: log and continue without a tracer rather than + // failing the whole process. + fmt.Fprintf(os.Stderr, "push-gateway: tracer init failed: %v (tracing disabled)\n", err) return func() {} } - - u, err := url.Parse(endpoint) - if err != nil || u.Host == "" { - log.Printf("invalid OTEL endpoint %q, tracing disabled", endpoint) - return func() {} - } - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - exporter, err := otlptracehttp.New(ctx, - otlptracehttp.WithEndpoint(u.Host), - otlptracehttp.WithInsecure(), - ) - if err != nil { - log.Printf("failed to create OTLP exporter: %v, tracing disabled", err) - return func() {} - } - - res, err := resource.New(ctx, - resource.WithAttributes(semconv.ServiceName(serviceName)), - ) - if err != nil { - log.Printf("failed to create resource: %v", err) - return func() {} - } - - tp := sdktrace.NewTracerProvider( - sdktrace.WithBatcher(exporter), - sdktrace.WithResource(res), - ) - otel.SetTracerProvider(tp) - otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( - propagation.TraceContext{}, - propagation.Baggage{}, - )) - - log.Printf("OpenTelemetry tracer initialized for %s (endpoint=%s)", serviceName, u.Host) return func() { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := tp.Shutdown(shutdownCtx); err != nil { - log.Printf("failed to shutdown tracer: %v", err) + if err := tracer.Shutdown(context.Background()); err != nil { + fmt.Fprintf(os.Stderr, "push-gateway: tracer shutdown failed: %v\n", err) } } } diff --git a/services/push-gateway/internal/redisclient/pubsub.go b/services/push-gateway/internal/redisclient/pubsub.go new file mode 100644 index 0000000..8e5b276 --- /dev/null +++ b/services/push-gateway/internal/redisclient/pubsub.go @@ -0,0 +1,370 @@ +// Package redisclient manages cross-instance presence and message fanout +// using Redis Pub/Sub plus the online-presence SET described in 02 §3.1 and +// §8.4. It implements the ISSUE-058 startup-rebuild mechanism: when a Hub +// starts it clears stale members belonging to its own instanceID from the +// online SET, then re-adds itself for every locally connected user. +// +// Keys / channels: +// +// edu:push:online: SET (members are instanceID) TTL 60s +// edu:push:session:: HASH (instanceID, device, lastPing) TTL 60s +// edu:push:channel:user: Pub/Sub channel (directed push) +// edu:push:channel:broadcast Pub/Sub channel (broadcast push) +// edu:push:idempotent: SETNX key (Kafka dedup) TTL 24h +package redisclient + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/edu-cloud/push-gateway/internal/hub" + "github.com/edu-cloud/push-gateway/internal/observability" + "github.com/redis/go-redis/v9" +) + +// PresenceTTL is the expiry applied to the online-presence SET and per-session +// HASH. The Hub refreshes it on every heartbeat; if a push-gateway instance +// crashes the SET entries expire naturally within this window. +const PresenceTTL = 60 * time.Second + +// IdempotentTTL is the dedup window for Kafka event_id (24h). +const IdempotentTTL = 24 * time.Hour + +// CrossInstanceMessage is the payload published to the +// edu:push:channel:user: and edu:push:channel:broadcast channels. +// Subscribers unmarshal it and dispatch to local Hub connections. +type CrossInstanceMessage struct { + Kind string `json:"kind"` // "user" | "broadcast" + UserID string `json:"userId"` // set when Kind == "user" + Event string `json:"event"` + Data json.RawMessage `json:"data"` + Origin string `json:"origin"` // instanceID of publisher (loopback suppression) + TraceID string `json:"traceId,omitempty"` +} + +// Client wraps a redis.Client with push-gateway specific presence and +// fanout helpers. The Hub is wired via SetHub so the subscriber can dispatch +// incoming Pub/Sub messages directly to local connections. +type Client struct { + rdb *redis.Client + instanceID string + hub *hub.Hub + metrics *observability.Metrics +} + +// New connects to Redis at url and returns a Client ready for presence and +// Pub/Sub operations. The instanceID identifies this process in the online +// SET (see ISSUE-058). +func New(url, instanceID string, m *observability.Metrics) (*Client, error) { + if url == "" { + return nil, errors.New("redisclient: empty URL") + } + opts, err := redis.ParseURL(url) + if err != nil { + return nil, fmt.Errorf("redisclient: parsing URL %q: %w", url, err) + } + rdb := redis.NewClient(opts) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + _ = rdb.Close() + return nil, fmt.Errorf("redisclient: ping failed: %w", err) + } + return &Client{ + rdb: rdb, + instanceID: instanceID, + metrics: m, + }, nil +} + +// SetHub wires the Hub so the Pub/Sub subscriber can deliver messages locally. +// Also installs presence hooks on the Hub so Register/Unregister keep the +// online SET in sync. +func (c *Client) SetHub(h *hub.Hub) { + c.hub = h + h.SetPresenceHooks(c.onRegister, c.onUnregister) +} + +// Redis returns the underlying *redis.Client for direct use by callers that +// need bespoke commands (e.g. SETNX for Kafka idempotency). +func (c *Client) Redis() *redis.Client { + return c.rdb +} + +// Close releases the Redis connection. +func (c *Client) Close() error { + return c.rdb.Close() +} + +// Ping is a lightweight liveness probe used by /readyz. +func (c *Client) Ping(ctx context.Context) error { + return c.rdb.Ping(ctx).Err() +} + +// ----- Presence SET (ISSUE-058) ------------------------------------------- + +// RebuildPresenceOnStartup clears stale SET entries belonging to this +// instanceID and re-adds the instanceID to edu:push:online: for every +// locally connected user (ISSUE-058 startup rebuild). +// +// This addresses the "ghost member" problem: when a push-gateway instance +// restarts, its previous instanceID entries may still be present in Redis +// (within the 60s TTL window). Re-adding the same instanceID would be a +// no-op, but if the instance previously crashed without removing members we +// must avoid carrying stale entries that no longer correspond to live +// connections. We clear first, then rebuild from the in-memory Hub state. +// +// 60s inconsistency window: between crash and SET expiry, /internal/push may +// report online:true for a user whose connection is gone; msg must tolerate +// delivered:false by retrying or falling back to offline push. +func (c *Client) RebuildPresenceOnStartup(ctx context.Context) error { + if c.hub == nil { + return errors.New("redisclient: Hub not set") + } + start := time.Now() + c.metrics.IncRedisSetRebuild() + + // Clear stale members belonging to this instanceID across all online SETs. + // We scan keys matching edu:push:online:* and SREM our instanceID from each. + iter := c.rdb.Scan(ctx, 0, "edu:push:online:*", 100).Iterator() + cleared := 0 + for iter.Next(ctx) { + key := iter.Val() + if n, err := c.rdb.SRem(ctx, key, c.instanceID).Result(); err == nil && n > 0 { + cleared++ + } + } + if err := iter.Err(); err != nil { + return fmt.Errorf("redisclient: scanning online SETs: %w", err) + } + + // Re-add instanceID for every locally online user. + readded := 0 + c.hub.ForEachUser(func(userID string) { + key := onlineKey(userID) + pipe := c.rdb.Pipeline() + pipe.SAdd(ctx, key, c.instanceID) + pipe.Expire(ctx, key, PresenceTTL) + if _, err := pipe.Exec(ctx); err == nil { + readded++ + } + }) + + observability.Logger().Info("redis presence rebuilt", + "cleared", cleared, + "readded", readded, + "duration_ms", time.Since(start).Milliseconds(), + ) + return nil +} + +// onRegister is the Hub hook invoked after a new connection is registered. +// Adds the instanceID to the user's online SET and refreshes TTL. +func (c *Client) onRegister(userID, connID string) { + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + key := onlineKey(userID) + pipe := c.rdb.Pipeline() + pipe.SAdd(ctx, key, c.instanceID) + pipe.Expire(ctx, key, PresenceTTL) + if _, err := pipe.Exec(ctx); err != nil { + observability.Logger().Warn("redis SAdd failed", + "user_id", userID, "err", err) + } +} + +// onUnregister is the Hub hook invoked after a connection is removed. +// Removes the instanceID from the SET and deletes the key when empty. +func (c *Client) onUnregister(userID, connID string) { + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + key := onlineKey(userID) + // Only remove if this user has no further local connections (Hub already + // removed the connection by the time the hook fires). + if c.hub != nil && !c.hub.HasUser(userID) { + if err := c.rdb.SRem(ctx, key, c.instanceID).Err(); err != nil { + observability.Logger().Warn("redis SRem failed", + "user_id", userID, "err", err) + } + // Best-effort DEL when SET becomes empty. + _ = c.rdb.Del(ctx, key).Err() + } +} + +// RefreshPresence extends the online SET TTL on heartbeat. Called by the +// WebSocket reader goroutine when a Ping control frame arrives. +func (c *Client) RefreshPresence(ctx context.Context, userID string) error { + return c.rdb.Expire(ctx, onlineKey(userID), PresenceTTL).Err() +} + +// IsOnline reports whether userID has any instance in the online SET. +// Returns false when Redis is unreachable; callers should treat the +// /internal/push response as authoritative for delivery. +func (c *Client) IsOnline(ctx context.Context, userID string) (bool, []string, error) { + members, err := c.rdb.SMembers(ctx, onlineKey(userID)).Result() + if err != nil { + return false, nil, err + } + return len(members) > 0, members, nil +} + +// ----- Pub/Sub fanout ------------------------------------------------------ + +// PublishUser publishes a directed message to edu:push:channel:user:. +// All push-gateway instances subscribe to this channel; the one holding the +// user will deliver locally, others drop it. +func (c *Client) PublishUser(ctx context.Context, userID string, msg CrossInstanceMessage) error { + msg.Kind = "user" + msg.UserID = userID + msg.Origin = c.instanceID + payload, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("redisclient: marshalling user msg: %w", err) + } + start := time.Now() + if err := c.rdb.Publish(ctx, userChannel(userID), payload).Err(); err != nil { + return fmt.Errorf("redisclient: PUBLISH user: %w", err) + } + c.metrics.ObservePubSubLatency("publish", time.Since(start).Seconds()) + return nil +} + +// PublishBroadcast publishes a broadcast message to all instances. +func (c *Client) PublishBroadcast(ctx context.Context, msg CrossInstanceMessage) error { + msg.Kind = "broadcast" + msg.Origin = c.instanceID + payload, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("redisclient: marshalling broadcast msg: %w", err) + } + start := time.Now() + if err := c.rdb.Publish(ctx, broadcastChannel(), payload).Err(); err != nil { + return fmt.Errorf("redisclient: PUBLISH broadcast: %w", err) + } + c.metrics.ObservePubSubLatency("publish", time.Since(start).Seconds()) + return nil +} + +// SubscribeAll subscribes to all per-user channels (pattern) and the +// broadcast channel, dispatching received messages to the local Hub. +// The returned cancel function stops the subscriber; it blocks until the +// goroutine has fully drained. +func (c *Client) SubscribeAll(ctx context.Context) (cancel func() error, err error) { + if c.hub == nil { + return nil, errors.New("redisclient: Hub not set") + } + + subCtx, subCancel := context.WithCancel(ctx) + userPubsub := c.rdb.PSubscribe(subCtx, "edu:push:channel:user:*") + broadcastPubsub := c.rdb.Subscribe(subCtx, broadcastChannel()) + + go c.dispatchLoop(subCtx, userPubsub, "user") + go c.dispatchLoop(subCtx, broadcastPubsub, "broadcast") + + return func() error { + subCancel() + if err := userPubsub.Close(); err != nil { + return err + } + return broadcastPubsub.Close() + }, nil +} + +// dispatchLoop consumes Pub/Sub messages and dispatches them to the Hub. +// channelType is "user" or "broadcast" for metric labeling. +func (c *Client) dispatchLoop(ctx context.Context, ps *redis.PubSub, channelType string) { + ch := ps.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + start := time.Now() + c.handlePubSubMessage(msg, channelType) + c.metrics.ObservePubSubLatency("subscribe", time.Since(start).Seconds()) + } + } +} + +// handlePubSubMessage unmarshals the payload and dispatches to the Hub, +// suppressing loopback (messages this instance published itself). +func (c *Client) handlePubSubMessage(msg *redis.Message, channelType string) { + var crossMsg CrossInstanceMessage + if err := json.Unmarshal([]byte(msg.Payload), &crossMsg); err != nil { + observability.Logger().Warn("pubsub: invalid payload", + "channel", msg.Channel, "err", err) + return + } + // Loopback suppression: do not deliver our own publishes. + if crossMsg.Origin == c.instanceID { + return + } + + payload, err := json.Marshal(map[string]any{ + "type": "message", + "event": crossMsg.Event, + "data": crossMsg.Data, + "timestamp": time.Now().UTC().Format(time.RFC3339), + }) + if err != nil { + return + } + + switch crossMsg.Kind { + case "user": + c.hub.SendToUser(crossMsg.UserID, payload) + case "broadcast": + c.hub.Broadcast(payload) + } +} + +// ----- Idempotency (Kafka dedup) ------------------------------------------ + +// DedupEventId performs a SETNX with TTL for Kafka event_id dedup. +// Returns true when the event is fresh (first occurrence), false when it +// has already been consumed. +func (c *Client) DedupEventId(ctx context.Context, eventID string) (bool, error) { + if eventID == "" { + return true, nil + } + ok, err := c.rdb.SetNX(ctx, idempotentKey(eventID), "1", IdempotentTTL).Result() + if err != nil { + return false, err + } + return ok, nil +} + +// ----- Key helpers --------------------------------------------------------- + +func onlineKey(userID string) string { + return "edu:push:online:" + userID +} + +func userChannel(userID string) string { + return "edu:push:channel:user:" + userID +} + +func broadcastChannel() string { + return "edu:push:channel:broadcast" +} + +func idempotentKey(eventID string) string { + return "edu:push:idempotent:" + eventID +} + +// ParseUserChannel extracts the userID from a "edu:push:channel:user:" +// channel name. Returns empty string when the name does not match. +func ParseUserChannel(channel string) string { + const prefix = "edu:push:channel:user:" + if !strings.HasPrefix(channel, prefix) { + return "" + } + return strings.TrimPrefix(channel, prefix) +}