Add jobs phase 2 learning console with polling list sync

apps/server/.env.example

@@ -40,7 +40,7 @@ JWT_SECRET="your-super-secret-and-long-string"
 
 # Server
 # 应用程序监听的端口
-# PORT=3000
+# PORT=3174
 
 # Swagger
 # 是否启用 Swagger UI (API 文档)

apps/server/package.json

@@ -40,6 +40,8 @@
     "url": "https://github.com/fengzai6/my-first-nest/issues"
   },
   "dependencies": {
+    "@bull-board/api": "^8.3.0",
+    "@bull-board/express": "^8.3.0",
     "@keyv/redis": "^5.1.6",
     "@nestjs/bullmq": "^11.0.4",
     "@nestjs/cache-manager": "^3.1.2",

apps/server/src/common/decorators/skip-timeout.decorator.ts

@@ -0,0 +1,5 @@
+import { SetMetadata } from '@nestjs/common';
+
+export const SKIP_TIMEOUT_KEY = 'skipTimeout';
+
+export const SkipTimeout = () => SetMetadata(SKIP_TIMEOUT_KEY, true);

apps/server/src/common/interceptors/timeout.interceptor.ts

@@ -10,13 +10,22 @@ import {
 } from '@nestjs/common';
 import { Observable, throwError, TimeoutError } from 'rxjs';
 import { catchError, timeout } from 'rxjs/operators';
+import { SKIP_TIMEOUT_KEY } from '@/common/decorators/skip-timeout.decorator';
 
 @Injectable()
 export class TimeoutInterceptor implements NestInterceptor {
   constructor(private readonly app: INestApplication) {}
 
   intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
     const { server } = getAppConfig(this.app);
+    const handler =
+      typeof context.getHandler === 'function'
+        ? context.getHandler()
+        : undefined;
+
+    if (handler && Reflect.getMetadata(SKIP_TIMEOUT_KEY, handler) === true) {
+      return next.handle();
+    }
 
     return next.handle().pipe(
       timeout(server.timeout * 1000),

apps/server/src/shared/jobs/README.md

@@ -36,8 +36,52 @@ export class CleanupHandler implements IJobHandler {
 
 - `GET /api/jobs` 任务列表
 - `GET /api/jobs/:id` 单任务轮询
+- `GET /api/jobs/:id/events` 单任务 SSE 事件流
 - `POST /api/jobs/:id/cancel` 取消 queued/delayed
 
+## SSE 事件流
+
+`GET /api/jobs/:id/events` 是轮询之外的第二种学习示例，本期先提供服务端能力，前端暂不接入。
+
+- 连接后先发送 `job.snapshot`
+- 状态或进度变化发送 `job.updated`
+- `completed` / `failed` / `cancelled` 会发送终态事件并结束流
+- 需要 JWT，与其它 Jobs API 一样不对外裸奔
+
+事件格式：
+
+```text
+event: job.updated
+id: 1
+data: {"id":"...","status":"active","progress":50}
+
+```
+
+## 任务中心 vs Bull Board
+
+| 视图 | 数据源 | 用途 |
+|------|--------|------|
+| 前端任务中心 | PostgreSQL `job_runs` | 展示业务任务生命周期、payload、result、errorMessage、触发类型 |
+| Bull Board | BullMQ queue | 观察队列内部 waiting/active/completed/failed 状态 |
+
+Bull Board 挂载在 `/admin/queues`，用于队列可观测性学习，不能替代业务任务中心。
+
+## 轮询 vs SSE
+
+| 方式 | 优点 | 缺点 | 适用 |
+|------|------|------|------|
+| 轮询 | 实现简单、兼容性好 | 有延迟、多余请求 | 通用默认 |
+| SSE | 实时、服务端推送 | 连接管理更复杂 | 进度场景 |
+
+前端任务中心本期使用 `GET /api/jobs/:id` 轮询。SSE 后端接口保留，可用 curl 单独验证：
+
+```bash
+curl -N \
+  -H "Accept: text/event-stream" \
+  -H "Authorization: Bearer <access-token>" \
+  http://localhost:3174/api/jobs/<job-id>/events
+```
+
 ## 与 @nestjs/schedule 的边界
 
 | | scheduled-tasks | shared/jobs |

apps/server/src/shared/jobs/board/job-board.auth.middleware.ts

@@ -0,0 +1,63 @@
+import { REFRESH_TOKEN_KEY, TokenType } from '@/common/constants/auth';
+import { JwtPayload } from '@/modules/auth/strategies/jwt-auth.strategy';
+import { UsersService } from '@/modules/users/users.service';
+import { Injectable, NestMiddleware } from '@nestjs/common';
+import { JwtService } from '@nestjs/jwt';
+import { NextFunction, Request, Response } from 'express';
+
+@Injectable()
+export class JobBoardAuthMiddleware implements NestMiddleware {
+  constructor(
+    private readonly jwtService: JwtService,
+    private readonly usersService: UsersService,
+  ) {}
+
+  async use(req: Request, res: Response, next: NextFunction) {
+    const tokenResult = this.extractToken(req);
+    if (!tokenResult) {
+      this.reject(res);
+      return;
+    }
+
+    try {
+      const payload = this.jwtService.verify<JwtPayload>(tokenResult.token);
+      if (payload.type !== tokenResult.tokenType) {
+        this.reject(res);
+        return;
+      }
+
+      await this.usersService.findOne({ id: payload.sub });
+      next();
+    } catch {
+      this.reject(res);
+    }
+  }
+
+  private extractToken(req: Request) {
+    const bearerToken = this.extractBearerToken(req.headers.authorization);
+    if (bearerToken) {
+      return { token: bearerToken, tokenType: TokenType.ACCESS };
+    }
+
+    const refreshToken = req.cookies?.[REFRESH_TOKEN_KEY] as string | undefined;
+    if (refreshToken) {
+      return { token: refreshToken, tokenType: TokenType.REFRESH };
+    }
+
+    return null;
+  }
+
+  private extractBearerToken(authorization?: string): string | null {
+    if (!authorization?.startsWith('Bearer ')) return null;
+    const token = authorization.slice('Bearer '.length).trim();
+    return token || null;
+  }
+
+  private reject(res: Response) {
+    res.status(401).json({
+      statusCode: 401,
+      message: 'Unauthorized',
+      code: 'UNAUTHORIZED',
+    });
+  }
+}

apps/server/src/shared/jobs/board/job-board.module.ts

@@ -0,0 +1,61 @@
+import { AppConfigModule } from '@/config/config.module';
+import { getConfig } from '@/config/configuration';
+import { UsersModule } from '@/modules/users/users.module';
+import { InjectQueue } from '@nestjs/bullmq';
+import { Module, OnModuleInit } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { HttpAdapterHost } from '@nestjs/core';
+import { JwtModule } from '@nestjs/jwt';
+import { createBullBoard } from '@bull-board/api';
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
+import { ExpressAdapter } from '@bull-board/express';
+import { Queue } from 'bullmq';
+import { Express } from 'express';
+import { DEFAULT_JOB_QUEUE } from '../constants/job.constants';
+import { JobQueueModule } from '../queue/job-queue.module';
+import { IBullJobData } from '../types/job.types';
+import { JobBoardAuthMiddleware } from './job-board.auth.middleware';
+
+const JOB_BOARD_PATH = '/admin/queues';
+
+@Module({
+  imports: [
+    UsersModule,
+    JobQueueModule,
+    JwtModule.registerAsync({
+      imports: [AppConfigModule],
+      inject: [ConfigService],
+      useFactory: (configService: ConfigService) => {
+        const { jwt } = getConfig(configService);
+        return {
+          secret: jwt.secret,
+          signOptions: { expiresIn: jwt.accessExpiresIn },
+        };
+      },
+    }),
+  ],
+  providers: [JobBoardAuthMiddleware],
+})
+export class JobBoardModule implements OnModuleInit {
+  constructor(
+    private readonly httpAdapterHost: HttpAdapterHost,
+    private readonly authMiddleware: JobBoardAuthMiddleware,
+    @InjectQueue(DEFAULT_JOB_QUEUE)
+    private readonly defaultQueue: Queue<IBullJobData>,
+  ) {}
+
+  onModuleInit() {
+    const serverAdapter = new ExpressAdapter();
+    serverAdapter.setBasePath(JOB_BOARD_PATH);
+
+    createBullBoard({
+      queues: [new BullMQAdapter(this.defaultQueue)],
+      serverAdapter,
+    });
+
+    const app: Express =
+      this.httpAdapterHost.httpAdapter.getInstance<Express>();
+    app.use(JOB_BOARD_PATH, this.authMiddleware.use.bind(this.authMiddleware));
+    app.use(JOB_BOARD_PATH, serverAdapter.getRouter());
+  }
+}

apps/server/src/shared/jobs/constants/job.constants.ts

@@ -30,5 +30,16 @@ export const JOB_NAMES = {
 
 export type JobName = (typeof JOB_NAMES)[keyof typeof JOB_NAMES];
 
+export const JOB_SSE_EVENT = {
+  SNAPSHOT: 'job.snapshot',
+  UPDATED: 'job.updated',
+  COMPLETED: 'job.completed',
+  FAILED: 'job.failed',
+  CANCELLED: 'job.cancelled',
+} as const;
+
+export type JobSseEventName =
+  (typeof JOB_SSE_EVENT)[keyof typeof JOB_SSE_EVENT];
+
 /** BullMQ 默认队列名 */
 export const DEFAULT_JOB_QUEUE = JOB_QUEUE_NAME.DEFAULT;

apps/server/src/shared/jobs/events/job-events.service.ts

@@ -0,0 +1,27 @@
+import { Injectable } from '@nestjs/common';
+import { Observable, Subject, filter } from 'rxjs';
+import { IJobSseEvent } from '../types/job.types';
+
+@Injectable()
+export class JobEventsService {
+  private readonly events$ = new Subject<IJobSseEvent>();
+  private sequence = 0;
+
+  publish(event: Omit<IJobSseEvent, 'id'>): IJobSseEvent {
+    const nextEvent: IJobSseEvent = {
+      ...event,
+      id: String(++this.sequence),
+    };
+
+    this.events$.next(nextEvent);
+    return nextEvent;
+  }
+
+  subscribe(jobId: string): Observable<IJobSseEvent> {
+    return this.events$.asObservable().pipe(
+      filter((event) => {
+        return event.data.id === jobId;
+      }),
+    );
+  }
+}

apps/server/src/shared/jobs/events/job-sse.util.ts

@@ -0,0 +1,23 @@
+import {
+  JOB_SSE_EVENT,
+  JOB_STATUS,
+  JobStatus,
+} from '../constants/job.constants';
+import { IJobSseEvent } from '../types/job.types';
+
+export const resolveJobSseEventName = (status: JobStatus) => {
+  if (status === JOB_STATUS.COMPLETED) return JOB_SSE_EVENT.COMPLETED;
+  if (status === JOB_STATUS.FAILED) return JOB_SSE_EVENT.FAILED;
+  if (status === JOB_STATUS.CANCELLED) return JOB_SSE_EVENT.CANCELLED;
+  return JOB_SSE_EVENT.UPDATED;
+};
+
+export const formatSseEvent = (event: IJobSseEvent) => {
+  return [
+    `event: ${event.event}`,
+    `id: ${event.id}`,
+    `data: ${JSON.stringify(event.data)}`,
+    '',
+    '',
+  ].join('\n');
+};

apps/server/src/shared/jobs/jobs.controller.ts

@@ -6,28 +6,87 @@ import {
   Param,
   Post,
   Query,
+  Req,
+  Res,
 } from '@nestjs/common';
+import { Request, Response } from 'express';
+import { JOB_SSE_EVENT } from './constants/job.constants';
 import {
   ApiBearerAuth,
   ApiOperation,
   ApiParam,
   ApiTags,
 } from '@nestjs/swagger';
+import { JobEventsService } from './events/job-events.service';
+import { formatSseEvent } from './events/job-sse.util';
 import { ListJobsDto } from './dto/list-jobs.dto';
 import { JobService } from './services/job.service';
+import { JOB_TERMINAL_STATUSES } from './types/job.types';
+import { SkipTimeout } from '@/common/decorators/skip-timeout.decorator';
 
 @ApiTags('Jobs - 任务中心')
 @ApiBearerAuth()
 @Controller('jobs')
 export class JobsController {
-  constructor(private readonly jobService: JobService) {}
+  constructor(
+    private readonly jobService: JobService,
+    private readonly jobEvents: JobEventsService,
+  ) {}
 
   @Get()
   @ApiOperation({ summary: '分页查询任务执行记录' })
   list(@Query() query: ListJobsDto) {
     return this.jobService.list(query);
   }
 
+  @Get(':id/events')
+  @ApiOperation({ summary: '订阅单个任务状态事件（SSE）' })
+  @ApiParam({ name: 'id', description: 'job_runs.id' })
+  @SkipTimeout()
+  async getEvents(
+    @Param('id') id: string,
+    @Req() req: Request,
+    @Res() res: Response,
+  ) {
+    res.setHeader('Content-Type', 'text/event-stream');
+    res.setHeader('Cache-Control', 'no-cache, no-transform');
+    res.setHeader('Connection', 'keep-alive');
+    res.flushHeaders?.();
+
+    const endStream = () => {
+      if (!res.writableEnded) {
+        res.end();
+      }
+    };
+
+    const subscription = this.jobEvents.subscribe(id).subscribe((event) => {
+      res.write(formatSseEvent(event));
+      if (JOB_TERMINAL_STATUSES.includes(event.data.status)) {
+        subscription.unsubscribe();
+        endStream();
+      }
+    });
+
+    req.on('close', () => {
+      subscription.unsubscribe();
+    });
+
+    const snapshot = await this.jobService.getById(id);
+
+    res.write(
+      formatSseEvent({
+        id: 'snapshot',
+        event: JOB_SSE_EVENT.SNAPSHOT,
+        data: snapshot,
+      }),
+    );
+
+    if (JOB_TERMINAL_STATUSES.includes(snapshot.status)) {
+      subscription.unsubscribe();
+      endStream();
+    }
+  }
+
   @Get(':id')
   @ApiOperation({ summary: '查询单个任务状态（轮询）' })
   @ApiParam({ name: 'id', description: 'job_runs.id' })