feat(p5): messaging, push gateway and AI assistant services

P5 阶段交付物：
- services/msg: 消息通知服务（NestJS）
  - notifications: 发送通知 + ES 全文检索 + search
  - config/elasticsearch.ts: ES Client 单例
  - package.json: 补充 @opentelemetry/sdk-node + exporter-trace-otlp-http
- services/push-gateway: WebSocket 推送网关（Go Gin）
  - internal/hub/hub.go: WebSocket 连接池管理（Register/Unregister/SendToUser）
  - internal/ws/handler.go: JWT 鉴权 + WebSocket 升级 + 内部推送 API
- services/ai: AI 辅助服务（Python FastAPI）
  - /chat + /chat/stream（SSE 流式）
  - /generate/question + /optimize/expression
  - config.py: OpenAI 兼容 API 配置
- packages/shared-proto/proto/msg.proto: NotificationService 契约（send/search）
- packages/shared-proto/proto/ai.proto: AiService 契约（含 stream 方法）

packages/shared-proto/proto/ai.proto

@@ -0,0 +1,60 @@
+syntax = "proto3";
+
+package next_edu_cloud.ai.v1;
+
+service AiService {
+  rpc Chat(ChatRequest) returns (ChatResponse);
+  rpc StreamChat(ChatRequest) returns (stream ChatChunk);
+  rpc GenerateQuestion(GenerateQuestionRequest) returns (GeneratedQuestion);
+  rpc OptimizeExpression(OptimizeExpressionRequest) returns (OptimizedExpression);
+}
+
+message ChatRequest {
+  repeated ChatMessage messages = 1;
+  string model = 2;
+  double temperature = 3;
+}
+
+message ChatMessage {
+  string role = 1;
+  string content = 2;
+}
+
+message ChatResponse {
+  string content = 1;
+  string model = 2;
+  Usage usage = 3;
+}
+
+message Usage {
+  int32 prompt_tokens = 1;
+  int32 completion_tokens = 2;
+  int32 total_tokens = 3;
+}
+
+message ChatChunk {
+  string content = 1;
+  bool done = 2;
+}
+
+message GenerateQuestionRequest {
+  string prompt = 1;
+  string subject = 2;
+  string difficulty = 3;
+}
+
+message GeneratedQuestion {
+  string question = 1;
+  string answer = 2;
+  string explanation = 3;
+}
+
+message OptimizeExpressionRequest {
+  string text = 1;
+  string context = 2;
+}
+
+message OptimizedExpression {
+  string optimized = 1;
+  repeated string suggestions = 2;
+}

packages/shared-proto/proto/msg.proto

@@ -0,0 +1,48 @@
+syntax = "proto3";
+
+package next_edu_cloud.msg.v1;
+
+service NotificationService {
+  rpc SendNotification(SendNotificationRequest) returns (Notification);
+  rpc ListNotifications(ListNotificationsRequest) returns (ListNotificationsResponse);
+  rpc MarkAsRead(MarkAsReadRequest) returns (Empty);
+  rpc SearchNotifications(SearchNotificationsRequest) returns (SearchNotificationsResponse);
+}
+
+message Notification {
+  string id = 1;
+  string user_id = 2;
+  string type = 3;
+  string title = 4;
+  string content = 5;
+  string channel = 6;
+  bool is_read = 7;
+  int64 created_at = 8;
+}
+
+message SendNotificationRequest {
+  string user_id = 1;
+  string type = 2;
+  string title = 3;
+  string content = 4;
+  string channel = 5;
+}
+
+message ListNotificationsRequest {
+  string user_id = 1;
+  bool only_unread = 2;
+}
+
+message ListNotificationsResponse {
+  repeated Notification notifications = 1;
+}
+
+message MarkAsReadRequest { string id = 1; }
+message SearchNotificationsRequest {
+  string user_id = 1;
+  string query = 2;
+}
+message SearchNotificationsResponse {
+  repeated Notification notifications = 1;
+}
+message Empty {}

services/ai/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.12-slim
+WORKDIR /app
+RUN pip install uv
+COPY pyproject.toml .
+RUN uv sync --no-dev
+COPY src ./src
+EXPOSE 3008
+CMD ["uv", "run", "uvicorn", "src.ai.main:app", "--host", "0.0.0.0", "--port", "3008"]

services/ai/README.md

@@ -0,0 +1,45 @@
+# AI 网关服务
+
+> 版本：0.1（P5 骨架）
+> 端口：3008
+
+## 职责
+
+AI 网关限界上下文（Python 实现），统一封装 LLM 调用（多模型路由、重试、限流、成本控制）。
+提供辅助出题、表达优化、分层提问等能力。通过 gRPC 查询 content 题库与 data-ana 学情数据。
+
+## 技术栈
+
+- Python 3.12 + FastAPI 0.115
+- Pydantic 2 + pydantic-settings
+- OpenTelemetry（LLM 调用链追踪）
+- prometheus-client + structlog
+- SSE 流式响应
+
+## 开发
+
+```bash
+uv sync
+uv run uvicorn src.ai.main:app --reload --port 3008
+```
+
+## API
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | /healthz | 健康检查 |
+| POST | /chat | LLM 聊天接口 |
+| POST | /chat/stream | 流式聊天（SSE） |
+| POST | /generate/question | 生成题目 |
+| POST | /optimize/expression | 优化表达 |
+| GET | /metrics | Prometheus 指标 |
+
+## 环境变量
+
+| 变量 | 默认值 | 说明 |
+|------|--------|------|
+| port | 3008 | 服务端口 |
+| openai_api_key | - | OpenAI API 密钥 |
+| anthropic_api_key | - | Anthropic API 密钥 |
+| otel_endpoint | http://localhost:4318 | OpenTelemetry OTLP 端点 |
+| log_level | info | 日志级别 |

services/ai/pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "ai-service"
+version = "0.1.0"
+description = "AI 网关服务 - LLM 集成 + RAG"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi>=0.115.0",
+    "uvicorn[standard]>=0.30.0",
+    "pydantic>=2.9.0",
+    "pydantic-settings>=2.5.0",
+    "httpx>=0.27.0",
+    "opentelemetry-api>=1.27.0",
+    "opentelemetry-sdk>=1.27.0",
+    "prometheus-client>=0.20.0",
+    "structlog>=24.4.0",
+]
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "W", "UP", "B", "SIM"]

services/ai/src/ai/config.py

@@ -0,0 +1,18 @@
+"""配置管理."""
+
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """应用配置."""
+
+    port: int = 3008
+    openai_api_key: str = ""
+    anthropic_api_key: str = ""
+    otel_endpoint: str = "http://localhost:4318"
+    log_level: str = "info"
+
+    model_config = {"env_file": ".env", "env_prefix": ""}
+
+
+settings = Settings()

services/ai/src/ai/main.py

@@ -0,0 +1,111 @@
+"""AI 网关服务入口."""
+
+from contextlib import asynccontextmanager
+
+import structlog
+from fastapi import FastAPI
+from fastapi.responses import StreamingResponse
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from prometheus_client import make_asgi_app
+from pydantic import BaseModel
+
+logger = structlog.get_logger()
+tracer = trace.get_tracer(__name__)
+
+
+def init_tracer() -> None:
+    """初始化 OpenTelemetry."""
+    provider = TracerProvider()
+    exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
+    provider.add_span_processor(BatchSpanProcessor(exporter))
+    trace.set_tracer_provider(provider)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """应用生命周期."""
+    init_tracer()
+    logger.info("ai service starting")
+    yield
+    logger.info("ai service stopping")
+
+
+app = FastAPI(
+    title="AI Gateway Service",
+    version="0.1.0",
+    lifespan=lifespan,
+)
+
+app.mount("/metrics", make_asgi_app())
+
+
+class ChatRequest(BaseModel):
+    """聊天请求."""
+
+    messages: list[dict]
+    model: str = "gpt-4o-mini"
+    temperature: float = 0.7
+    stream: bool = False
+
+
+class ChatResponse(BaseModel):
+    """聊天响应."""
+
+    content: str
+    model: str
+    usage: dict
+
+
+@app.get("/healthz")
+async def healthz():
+    """健康检查."""
+    return {"status": "ok", "service": "ai"}
+
+
+@app.post("/chat", response_model=ChatResponse)
+async def chat(req: ChatRequest):
+    """LLM 聊天接口."""
+    with tracer.start_as_current_span("ai_chat"):
+        # P5 骨架：实际调用 OpenAI/Anthropic API
+        # 需要从环境变量获取 API key
+        return {
+            "content": "P5 skeleton - LLM integration pending",
+            "model": req.model,
+            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
+        }
+
+
+@app.post("/chat/stream")
+async def chat_stream(req: ChatRequest):
+    """流式聊天（SSE）."""
+
+    async def generate():
+        with tracer.start_as_current_span("ai_chat_stream"):
+            # P5 骨架：流式调用 LLM
+            yield "data: P5 skeleton\n\n"
+            yield "data: [DONE]\n\n"
+
+    return StreamingResponse(generate(), media_type="text/event-stream")
+
+
+@app.post("/generate/question")
+async def generate_question(prompt: str):
+    """生成题目."""
+    with tracer.start_as_current_span("generate_question"):
+        return {
+            "success": True,
+            "data": {"question": "P5 skeleton - question generation pending"},
+        }
+
+
+@app.post("/optimize/expression")
+async def optimize_expression(text: str):
+    """优化表达."""
+    with tracer.start_as_current_span("optimize_expression"):
+        return {
+            "success": True,
+            "data": {"optimized": "P5 skeleton - expression optimization pending"},
+        }

services/msg/Dockerfile

@@ -0,0 +1,19 @@
+# Build stage
+FROM node:20-alpine AS builder
+WORKDIR /app
+RUN npm install -g pnpm
+COPY package.json pnpm-lock.yaml* ./
+RUN pnpm install --frozen-lockfile
+COPY tsconfig.json nest-cli.json ./
+COPY src ./src
+RUN pnpm build
+
+# Runtime stage
+FROM node:20-alpine
+WORKDIR /app
+RUN npm install -g pnpm
+COPY package.json pnpm-lock.yaml* ./
+RUN pnpm install --prod --frozen-lockfile
+COPY --from=builder /app/dist ./dist
+EXPOSE 3007
+CMD ["node", "dist/main.js"]

services/msg/README.md

@@ -0,0 +1,43 @@
+# Msg 消息通知服务
+
+> 版本：0.1（P5 骨架）
+> 端口：3007
+
+## 职责
+
+消息通知限界上下文，管理通知的多渠道分发（站内信、邮件、短信、推送）。
+消费 Kafka 事件，写入 MySQL + Elasticsearch，调用 Push Gateway 实时推送。
+
+## 技术栈
+
+- NestJS 10 + TypeScript 5
+- Drizzle ORM + MySQL 8
+- Elasticsearch 8（全文检索）
+- Kafka（事件消费，骨架）
+- pino + prom-client + OpenTelemetry
+
+## 开发
+
+```bash
+pnpm install
+pnpm dev   # http://localhost:3007
+```
+
+## API
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| POST | /notifications | 发送通知 |
+| GET | /notifications | 查询用户通知列表 |
+| POST | /notifications/:id/read | 标记已读 |
+| GET | /notifications/search?q= | 全文检索通知 |
+
+## 环境变量
+
+| 变量 | 默认值 | 说明 |
+|------|--------|------|
+| PORT | 3007 | 服务端口 |
+| DATABASE_URL | - | MySQL 连接串 |
+| KAFKA_BROKERS | localhost:9092 | Kafka broker 列表 |
+| ES_URL | http://localhost:9200 | Elasticsearch 地址 |
+| JWT_SECRET | - | JWT 密钥 |

services/msg/nest-cli.json

@@ -0,0 +1,8 @@
+{
+  "$schema": "https://json.schemastore.org/nest-cli",
+  "collection": "@nestjs/schematics",
+  "sourceRoot": "src",
+  "compilerOptions": {
+    "deleteOutDir": true
+  }
+}

services/msg/package.json

@@ -0,0 +1,38 @@
+{
+  "name": "@edu/msg-service",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "nest start --watch",
+    "build": "nest build",
+    "start": "node dist/main.js",
+    "test": "vitest run",
+    "lint": "eslint src --ext .ts",
+    "typecheck": "tsc --noEmit"
+  },
+  "dependencies": {
+    "@nestjs/common": "^10.4.0",
+    "@nestjs/core": "^10.4.0",
+    "@nestjs/platform-express": "^10.4.0",
+    "drizzle-orm": "^0.31.0",
+    "mysql2": "^3.11.0",
+    "kafkajs": "^2.2.0",
+    "@elastic/elasticsearch": "^8.15.0",
+    "pino": "^9.4.0",
+    "prom-client": "^15.1.0",
+    "@opentelemetry/api": "^1.9.0",
+    "zod": "^3.23.0",
+    "uuid": "^10.0.0",
+    "reflect-metadata": "^0.2.2",
+    "rxjs": "^7.8.0",
+    "@opentelemetry/sdk-node": "^0.55.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.55.0"
+  },
+  "devDependencies": {
+    "@nestjs/cli": "^10.4.0",
+    "@types/node": "^22.0.0",
+    "typescript": "^5.6.0",
+    "vitest": "^2.1.0"
+  }
+}

services/msg/src/app.module.ts

@@ -0,0 +1,7 @@
+import { Module } from '@nestjs/common';
+import { NotificationsModule } from './notifications/notifications.module.js';
+
+@Module({
+  imports: [NotificationsModule],
+})
+export class AppModule {}

services/msg/src/config/database.ts

@@ -0,0 +1,24 @@
+import { drizzle } from 'drizzle-orm/mysql2';
+import mysql from 'mysql2/promise';
+import { env } from './env.js';
+
+let pool: mysql.Pool | null = null;
+
+export function getDb() {
+  if (!pool) {
+    pool = mysql.createPool({
+      uri: env.DATABASE_URL,
+      waitForConnections: true,
+      connectionLimit: 10,
+      queueLimit: 0,
+    });
+  }
+  return drizzle(pool);
+}
+
+export async function closeDb(): Promise<void> {
+  if (pool) {
+    await pool.end();
+    pool = null;
+  }
+}

services/msg/src/config/elasticsearch.ts

@@ -0,0 +1,13 @@
+import { Client } from '@elastic/elasticsearch';
+import { env } from './env.js';
+
+export const esClient = new Client({ node: env.ES_URL });
+
+export async function checkEsConnection(): Promise<void> {
+  try {
+    await esClient.ping();
+    console.log('Elasticsearch connected');
+  } catch (err) {
+    console.error('Elasticsearch connection failed:', err);
+  }
+}

services/msg/src/config/env.ts

@@ -0,0 +1,27 @@
+import { z } from 'zod';
+
+const envSchema = z.object({
+  PORT: z.string().default('3007'),
+  DATABASE_URL: z.string().url(),
+  REDIS_URL: z.string().url().optional(),
+  JWT_SECRET: z.string(),
+  JWT_ISSUER: z.string().default('next-edu-cloud'),
+  KAFKA_BROKERS: z.string().default('localhost:9092'),
+  ES_URL: z.string().url().default('http://localhost:9200'),
+  OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url().optional(),
+  LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'),
+  NODE_ENV: z.enum(['development', 'production', 'test']).default('development'),
+});
+
+export type Env = z.infer<typeof envSchema>;
+
+export function loadEnv(): Env {
+  const result = envSchema.safeParse(process.env);
+  if (!result.success) {
+    console.error('❌ Invalid environment variables:', result.error.flatten().fieldErrors);
+    throw new Error('Invalid environment configuration');
+  }
+  return result.data;
+}
+
+export const env = loadEnv();

services/msg/src/main.ts

@@ -0,0 +1,31 @@
+import 'reflect-metadata';
+import { NestFactory } from '@nestjs/core';
+import { AppModule } from './app.module.js';
+import { GlobalErrorFilter } from './shared/errors/global-error.filter.js';
+import { initTracer, shutdownTracer } from './shared/observability/tracer.js';
+import { env } from './config/env.js';
+import { logger } from './shared/observability/logger.js';
+
+async function bootstrap(): Promise<void> {
+  initTracer();
+
+  const app = await NestFactory.create(AppModule, {
+    logger: ['log', 'error', 'warn'],
+  });
+
+  app.useGlobalFilters(new GlobalErrorFilter());
+  app.enableShutdownHooks();
+
+  await app.listen(env.PORT);
+  logger.info({ port: env.PORT }, 'Msg service started');
+
+  process.on('SIGTERM', async () => {
+    await app.close();
+    await shutdownTracer();
+  });
+}
+
+bootstrap().catch((err: unknown) => {
+  logger.error({ err }, 'Failed to start msg service');
+  process.exit(1);
+});

services/msg/src/notifications/notifications.controller.ts

@@ -0,0 +1,32 @@
+import { Body, Controller, Get, Param, Post, Query, Req } from '@nestjs/common';
+import { NotificationsService } from './notifications.service.js';
+import type { SendNotificationDto } from './notifications.service.js';
+
+@Controller('notifications')
+export class NotificationsController {
+  constructor(private readonly service: NotificationsService) {}
+
+  @Post()
+  async send(@Body() body: unknown) {
+    const result = await this.service.send(body as SendNotificationDto);
+    return { success: true, data: result };
+  }
+
+  @Get()
+  async list(@Req() req: { userId: string }, @Query('unread') unread: string) {
+    const result = await this.service.listByUser(req.userId, unread === 'true');
+    return { success: true, data: result };
+  }
+
+  @Post(':id/read')
+  async markAsRead(@Param('id') id: string) {
+    await this.service.markAsRead(id);
+    return { success: true };
+  }
+
+  @Get('search')
+  async search(@Req() req: { userId: string }, @Query('q') q: string) {
+    const result = await this.service.search(req.userId, q);
+    return { success: true, data: result };
+  }
+}

services/msg/src/notifications/notifications.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { NotificationsController } from './notifications.controller.js';
+import { NotificationsService } from './notifications.service.js';
+
+@Module({
+  controllers: [NotificationsController],
+  providers: [NotificationsService],
+})
+export class NotificationsModule {}

services/msg/src/notifications/notifications.schema.ts

@@ -0,0 +1,24 @@
+import { mysqlTable, varchar, char, timestamp, text, boolean, json } from 'drizzle-orm/mysql-core';
+
+export const notifications = mysqlTable('msg_notifications', {
+  id: char('id', { length: 36 }).notNull().primaryKey(),
+  userId: char('user_id', { length: 36 }).notNull(),
+  type: varchar('type', { length: 50 }).notNull(),
+  title: varchar('title', { length: 200 }).notNull(),
+  content: text('content').notNull(),
+  channel: varchar('channel', { length: 20 }).notNull().default('in_app'),
+  isRead: boolean('is_read').notNull().default(false),
+  metadata: json('metadata'),
+  createdAt: timestamp('created_at').notNull().defaultNow(),
+});
+
+export const notificationPreferences = mysqlTable('msg_notification_preferences', {
+  userId: char('user_id', { length: 36 }).notNull().primaryKey(),
+  emailEnabled: boolean('email_enabled').notNull().default(true),
+  smsEnabled: boolean('sms_enabled').notNull().default(false),
+  pushEnabled: boolean('push_enabled').notNull().default(true),
+  inAppEnabled: boolean('in_app_enabled').notNull().default(true),
+});
+
+export type Notification = typeof notifications.$inferSelect;
+export type NotificationPreference = typeof notificationPreferences.$inferSelect;

services/msg/src/notifications/notifications.service.ts

@@ -0,0 +1,110 @@
+import { Injectable } from '@nestjs/common';
+import { getDb } from '../config/database.js';
+import { notifications, notificationPreferences } from './notifications.schema.js';
+import type { NotificationPreference } from './notifications.schema.js';
+import { v4 as uuidv4 } from 'uuid';
+import { eq, and } from 'drizzle-orm';
+import { esClient } from '../config/elasticsearch.js';
+import { logger } from '../shared/observability/logger.js';
+
+export interface SendNotificationDto {
+  userId: string;
+  type: string;
+  title: string;
+  content: string;
+  channel?: string;
+  metadata?: Record<string, unknown>;
+}
+
+@Injectable()
+export class NotificationsService {
+  async send(dto: SendNotificationDto) {
+    const id = uuidv4();
+    const db = getDb();
+
+    // 检查用户偏好
+    const [pref] = await db.select().from(notificationPreferences).where(eq(notificationPreferences.userId, dto.userId));
+
+    const channel = dto.channel || 'in_app';
+    if (pref && !this.isChannelEnabled(pref, channel)) {
+      logger.info({ userId: dto.userId, channel }, 'Notification skipped by preference');
+      return { skipped: true };
+    }
+
+    // 写入 DB
+    await db.insert(notifications).values({
+      id,
+      userId: dto.userId,
+      type: dto.type,
+      title: dto.title,
+      content: dto.content,
+      channel,
+      metadata: dto.metadata,
+    });
+
+    // 索引到 ES（全文检索）
+    try {
+      await esClient.index({
+        index: 'notifications',
+        id,
+        document: {
+          userId: dto.userId,
+          type: dto.type,
+          title: dto.title,
+          content: dto.content,
+          channel,
+          createdAt: new Date().toISOString(),
+        },
+      });
+    } catch (err) {
+      logger.error({ err }, 'Failed to index notification in ES');
+    }
+
+    // TODO: 触发 Push Gateway 推送（P5 后期）
+
+    return { id, skipped: false };
+  }
+
+  async listByUser(userId: string, onlyUnread: boolean = false) {
+    const db = getDb();
+    const conditions = onlyUnread
+      ? and(eq(notifications.userId, userId), eq(notifications.isRead, false))
+      : eq(notifications.userId, userId);
+    return db.select().from(notifications).where(conditions);
+  }
+
+  async markAsRead(id: string) {
+    const db = getDb();
+    await db.update(notifications).set({ isRead: true }).where(eq(notifications.id, id));
+  }
+
+  async search(userId: string, query: string) {
+    const result = await esClient.search({
+      index: 'notifications',
+      query: {
+        bool: {
+          must: [
+            { term: { userId } },
+            {
+              multi_match: {
+                query,
+                fields: ['title', 'content'],
+              },
+            },
+          ],
+        },
+      },
+    });
+    return result.hits.hits;
+  }
+
+  private isChannelEnabled(pref: NotificationPreference, channel: string): boolean {
+    switch (channel) {
+      case 'email': return pref.emailEnabled;
+      case 'sms': return pref.smsEnabled;
+      case 'push': return pref.pushEnabled;
+      case 'in_app': return pref.inAppEnabled;
+      default: return true;
+    }
+  }
+}

services/msg/src/shared/errors/application-error.ts

@@ -0,0 +1,96 @@
+export type ErrorType =
+  | 'validation'
+  | 'not_found'
+  | 'permission_denied'
+  | 'conflict'
+  | 'business'
+  | 'database'
+  | 'internal';
+
+export interface ErrorDetails {
+  [key: string]: unknown;
+}
+
+export abstract class ApplicationError extends Error {
+  abstract readonly type: ErrorType;
+  abstract readonly statusCode: number;
+  readonly code: string;
+  readonly details?: ErrorDetails;
+  // FIX #1: traceId 改为可写，以便 GlobalErrorFilter 注入请求级 traceId
+  traceId?: string;
+
+  constructor(message: string, code: string, details?: ErrorDetails) {
+    super(message);
+    this.name = this.constructor.name;
+    this.code = code;
+    this.details = details;
+  }
+
+  toJSON(): Record<string, unknown> {
+    return {
+      success: false,
+      error: {
+        code: this.code,
+        message: this.message,
+        details: this.details,
+        traceId: this.traceId,
+      },
+    };
+  }
+}
+
+export class ValidationError extends ApplicationError {
+  readonly type = 'validation' as const;
+  readonly statusCode = 400;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, 'MSG_VALIDATION_ERROR', details);
+  }
+}
+
+export class NotFoundError extends ApplicationError {
+  readonly type = 'not_found' as const;
+  readonly statusCode = 404;
+  constructor(resource: string, id: string) {
+    super(`${resource} not found: ${id}`, 'MSG_NOT_FOUND', { resource, id });
+  }
+}
+
+export class PermissionDeniedError extends ApplicationError {
+  readonly type = 'permission_denied' as const;
+  readonly statusCode = 403;
+  constructor(permission: string) {
+    super(`Permission denied: ${permission}`, 'MSG_PERMISSION_DENIED', { permission });
+  }
+}
+
+export class ConflictError extends ApplicationError {
+  readonly type = 'conflict' as const;
+  readonly statusCode = 409;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, 'MSG_CONFLICT', details);
+  }
+}
+
+export class BusinessError extends ApplicationError {
+  readonly type = 'business' as const;
+  readonly statusCode = 422;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, 'MSG_BUSINESS_ERROR', details);
+  }
+}
+
+export class DatabaseError extends ApplicationError {
+  readonly type = 'database' as const;
+  readonly statusCode = 500;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, 'MSG_DATABASE_ERROR', details);
+  }
+}
+
+export class InternalError extends ApplicationError {
+  readonly type = 'internal' as const;
+  readonly statusCode = 500;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, 'MSG_INTERNAL_ERROR', details);
+  }
+}

services/msg/src/shared/errors/global-error.filter.ts

@@ -0,0 +1,77 @@
+import { Catch, ExceptionFilter, ArgumentsHost, HttpException, Logger } from '@nestjs/common';
+import { Request, Response } from 'express';
+import { ZodError } from 'zod';
+import { ApplicationError } from './application-error.js';
+
+@Catch()
+export class GlobalErrorFilter implements ExceptionFilter {
+  private readonly logger = new Logger(GlobalErrorFilter.name);
+
+  catch(exception: unknown, host: ArgumentsHost): void {
+    const ctx = host.switchToHttp();
+    const response = ctx.getResponse<Response>();
+    const request = ctx.getRequest<Request>();
+
+    const traceId = (request.headers['x-request-id'] as string | undefined) ?? 'unknown';
+
+    let statusCode = 500;
+    let body: Record<string, unknown>;
+
+    if (exception instanceof ApplicationError) {
+      exception.traceId = traceId;
+      statusCode = exception.statusCode;
+      body = exception.toJSON();
+    } else if (exception instanceof ZodError) {
+      // FIX #3: 捕获 Zod 解析错误，转换为结构化 ValidationError 响应
+      statusCode = 400;
+      body = {
+        success: false,
+        error: {
+          code: 'MSG_VALIDATION_ERROR',
+          message: 'Validation failed',
+          details: exception.flatten(),
+          traceId,
+        },
+      };
+    } else if (exception instanceof HttpException) {
+      statusCode = exception.getStatus();
+      const res = exception.getResponse();
+      const message = this.extractHttpMessage(res, exception);
+      body = {
+        success: false,
+        error: {
+          code: 'HTTP_ERROR',
+          message,
+          traceId,
+        },
+      };
+    } else {
+      this.logger.error(
+        `Unhandled exception: ${exception}`,
+        exception instanceof Error ? exception.stack : undefined,
+      );
+      body = {
+        success: false,
+        error: {
+          code: 'INTERNAL_ERROR',
+          message: 'An unexpected error occurred',
+          traceId,
+        },
+      };
+    }
+
+    response.status(statusCode).json(body);
+  }
+
+  private extractHttpMessage(res: string | object, exception: HttpException): string {
+    if (typeof res === 'string') {
+      return res;
+    }
+    if (res && typeof res === 'object' && 'message' in res) {
+      // 从 HttpException 响应体收窄类型（NestJS 约定包含 message 字段）
+      const msg = (res as { message: unknown }).message;
+      return typeof msg === 'string' ? msg : exception.message;
+    }
+    return exception.message;
+  }
+}

services/msg/src/shared/observability/logger.ts

@@ -0,0 +1,20 @@
+import pino from 'pino';
+import { env } from '../../config/env.js';
+
+export const logger = pino({
+  level: env.LOG_LEVEL,
+  // 修复：pino 默认字段选项为 `base`，而非 `defaultFields`
+  base: {
+    service: 'msg',
+    version: '0.1.0',
+  },
+  transport:
+    env.NODE_ENV === 'development'
+      ? {
+          target: 'pino-pretty',
+          options: { colorize: true },
+        }
+      : undefined,
+});
+
+export type Logger = typeof logger;

services/msg/src/shared/observability/metrics.ts

@@ -0,0 +1,24 @@
+import promClient from 'prom-client';
+
+const registry = new promClient.Registry();
+// 修复：在本地 registry 上设置默认标签（原代码误用全局 register）
+registry.setDefaultLabels({ service: 'msg' });
+
+registry.registerMetric(
+  new promClient.Counter({
+    name: 'msg_requests_total',
+    help: 'Total number of msg requests',
+    labelNames: ['method', 'endpoint', 'status'],
+  }),
+);
+
+registry.registerMetric(
+  new promClient.Histogram({
+    name: 'msg_request_duration_seconds',
+    help: 'Msg request duration in seconds',
+    labelNames: ['method', 'endpoint'],
+    buckets: [0.01, 0.05, 0.1, 0.3, 0.5, 1, 3, 5],
+  }),
+);
+
+export { registry as metricsRegistry };

services/msg/src/shared/observability/tracer.ts

@@ -0,0 +1,25 @@
+import { NodeSDK } from '@opentelemetry/sdk-node';
+import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
+import { env } from '../../config/env.js';
+
+let sdk: NodeSDK | null = null;
+
+export function initTracer(): void {
+  if (!env.OTEL_EXPORTER_OTLP_ENDPOINT) return;
+
+  sdk = new NodeSDK({
+    serviceName: 'msg',
+    traceExporter: new OTLPTraceExporter({
+      url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
+    }),
+  });
+
+  sdk.start();
+  console.log('Tracer initialized');
+}
+
+export async function shutdownTracer(): Promise<void> {
+  if (sdk) {
+    await sdk.shutdown();
+  }
+}

services/msg/tsconfig.json

@@ -0,0 +1,15 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": {
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "target": "ES2022",
+    "experimentalDecorators": true,
+    "emitDecoratorMetadata": true,
+    "outDir": "./dist",
+    "rootDir": "./src",
+    "types": ["node"]
+  },
+  "include": ["src/**/*"],
+  "exclude": ["node_modules", "dist", "test"]
+}

services/push-gateway/Dockerfile

@@ -0,0 +1,13 @@
+FROM golang:1.22-alpine AS builder
+WORKDIR /app
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -o push-gateway .
+
+FROM alpine:3.20
+RUN apk --no-cache add ca-certificates
+WORKDIR /app
+COPY --from=builder /app/push-gateway .
+EXPOSE 8081
+CMD ["./push-gateway"]

services/push-gateway/README.md

@@ -0,0 +1,42 @@
+# Push Gateway 推送网关服务
+
+> 版本：0.1（P5 骨架）
+> 端口：8081
+
+## 职责
+
+实时推送基础设施服务（Go 实现），管理 WebSocket 长连接。
+接收 Msg 服务的推送请求，维护用户在线连接池，将消息实时投递到浏览器/移动端。
+
+## 技术栈
+
+- Go 1.22 + Gin 1.10
+- gorilla/websocket 1.5
+- golang-jwt/jwt/v5（JWT 鉴权）
+- zap（结构化日志，骨架）
+
+## 开发
+
+```bash
+go mod tidy
+go run main.go   # :8081
+```
+
+## API
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| GET | /healthz | 健康检查 |
+| GET | /ws?token=JWT | WebSocket 升级端点（JWT 鉴权） |
+| POST | /internal/push | 内部推送 API（Msg 服务调用） |
+
+## 环境变量
+
+| 变量 | 默认值 | 说明 |
+|------|--------|------|
+| PUSH_GATEWAY_PORT | 8081 | 服务端口 |
+| JWT_SECRET | p1-dev-secret-change-in-production | JWT 密钥 |
+
+## WebSocket 心跳
+
+客户端发送 `ping` 文本帧，服务端回复 `pong`。

services/push-gateway/go.mod

@@ -0,0 +1,11 @@
+module github.com/edu-cloud/push-gateway
+
+go 1.22
+
+require (
+	github.com/gin-gonic/gin v1.10.0
+	github.com/golang-jwt/jwt/v5 v5.2.1
+	github.com/google/uuid v1.6.0
+	github.com/gorilla/websocket v1.5.3
+	go.uber.org/zap v1.27.0
+)

services/push-gateway/internal/config/config.go

@@ -0,0 +1,22 @@
+package config
+
+import "os"
+
+type Config struct {
+	Port      string
+	JWTSecret string
+}
+
+func Load() *Config {
+	return &Config{
+		Port:      getEnv("PUSH_GATEWAY_PORT", "8081"),
+		JWTSecret: getEnv("JWT_SECRET", "p1-dev-secret-change-in-production"),
+	}
+}
+
+func getEnv(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}

services/push-gateway/internal/hub/hub.go

@@ -0,0 +1,59 @@
+package hub
+
+import (
+	"sync"
+
+	"github.com/gorilla/websocket"
+)
+
+// Hub 管理 WebSocket 客户端连接
+type Hub struct {
+	mu      sync.RWMutex
+	clients map[string]map[*websocket.Conn]bool // userID -> connections
+}
+
+func NewHub() *Hub {
+	return &Hub{
+		clients: make(map[string]map[*websocket.Conn]bool),
+	}
+}
+
+func (h *Hub) Register(userID string, conn *websocket.Conn) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.clients[userID] == nil {
+		h.clients[userID] = make(map[*websocket.Conn]bool)
+	}
+	h.clients[userID][conn] = true
+}
+
+func (h *Hub) Unregister(userID string, conn *websocket.Conn) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if conns, ok := h.clients[userID]; ok {
+		delete(conns, conn)
+		if len(conns) == 0 {
+			delete(h.clients, userID)
+		}
+	}
+}
+
+// SendToUser 向指定用户的所有连接推送消息
+func (h *Hub) SendToUser(userID string, message []byte) error {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	conns, ok := h.clients[userID]
+	if !ok {
+		return nil
+	}
+
+	for conn := range conns {
+		if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
+			return err
+		}
+	}
+	return nil
+}

services/push-gateway/internal/ws/handler.go

@@ -0,0 +1,99 @@
+package ws
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/edu-cloud/push-gateway/internal/hub"
+	"github.com/gin-gonic/gin"
+	"github.com/golang-jwt/jwt/v5"
+	"github.com/gorilla/websocket"
+)
+
+var upgrader = websocket.Upgrader{
+	CheckOrigin: func(r *http.Request) bool {
+		return true // P5 骨架，生产环境需校验 origin
+	},
+}
+
+type Handler struct {
+	hub       *hub.Hub
+	jwtSecret string
+}
+
+func NewHandler(h *hub.Hub, jwtSecret string) *Handler {
+	return &Handler{hub: h, jwtSecret: jwtSecret}
+}
+
+func (h *Handler) HandleWebSocket(c *gin.Context) {
+	// 从 query 参数获取 token（WebSocket 无法设置 Authorization 头）
+	tokenStr := c.Query("token")
+	if tokenStr == "" {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "missing token"})
+		return
+	}
+
+	token, err := jwt.Parse(tokenStr, func(t *jwt.Token) (interface{}, error) {
+		if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
+			return nil, jwt.ErrSignatureInvalid
+		}
+		return []byte(h.jwtSecret), nil
+	})
+
+	if err != nil || !token.Valid {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid token"})
+		return
+	}
+
+	claims, ok := token.Claims.(jwt.MapClaims)
+	if !ok {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid claims"})
+		return
+	}
+
+	userID, ok := claims["sub"].(string)
+	if !ok {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "missing user id"})
+		return
+	}
+
+	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+
+	h.hub.Register(userID, conn)
+	defer h.hub.Unregister(userID, conn)
+
+	// 读取循环（保持连接，处理心跳）
+	for {
+		_, msg, err := conn.ReadMessage()
+		if err != nil {
+			break
+		}
+		// 处理心跳 ping
+		if strings.ToLower(string(msg)) == "ping" {
+			conn.WriteMessage(websocket.TextMessage, []byte("pong"))
+		}
+	}
+}
+
+// PushHandler 接收来自 Msg 服务的推送请求
+func (h *Handler) PushHandler(c *gin.Context) {
+	var req struct {
+		UserID  string `json:"user_id"`
+		Message string `json:"message"`
+	}
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"success": false, "error": gin.H{"code": "INVALID_REQUEST", "message": err.Error()}})
+		return
+	}
+
+	if err := h.hub.SendToUser(req.UserID, []byte(req.Message)); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"success": false, "error": gin.H{"code": "PUSH_FAILED", "message": err.Error()}})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{"success": true})
+}

services/push-gateway/main.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"context"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/edu-cloud/push-gateway/internal/config"
+	"github.com/edu-cloud/push-gateway/internal/hub"
+	"github.com/edu-cloud/push-gateway/internal/ws"
+	"github.com/gin-gonic/gin"
+)
+
+func main() {
+	cfg := config.Load()
+	gin.SetMode(gin.ReleaseMode)
+
+	h := hub.NewHub()
+	wsHandler := ws.NewHandler(h, cfg.JWTSecret)
+
+	r := gin.New()
+	r.Use(gin.Recovery())
+
+	r.GET("/healthz", func(c *gin.Context) {
+		c.JSON(200, gin.H{"status": "ok", "service": "push-gateway"})
+	})
+
+	// WebSocket 升级端点
+	r.GET("/ws", wsHandler.HandleWebSocket)
+
+	// 内部推送 API（Msg 服务调用）
+	api := r.Group("/internal")
+	api.POST("/push", wsHandler.PushHandler)
+
+	srv := &http.Server{
+		Addr:         ":" + cfg.Port,
+		Handler:      r,
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 60 * time.Second, // WebSocket 长连接
+	}
+
+	go func() {
+		log.Printf("Push Gateway listening on :%s", cfg.Port)
+		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			log.Fatalf("listen: %s\n", err)
+		}
+	}()
+
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
+	log.Println("Shutting down Push Gateway...")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := srv.Shutdown(ctx); err != nil {
+		log.Fatal("Server forced to shutdown:", err)
+	}
+}