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 方法）

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"]
+}