增强关键接口幂等与限流保护

apps/backend/src/middleware/rate-limit.test.ts

@@ -0,0 +1,116 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import type { NextFunction, Request, Response } from "express";
+
+import {
+  createRateLimitMiddleware,
+  resetRateLimitBucketsForTest,
+} from "./rate-limit.js";
+
+test("createRateLimitMiddleware allows requests within the window", () => {
+  resetRateLimitBucketsForTest();
+
+  const limiter = createRateLimitMiddleware({
+    scope: "test:allow",
+    windowMs: 60_000,
+    max: 2,
+  });
+  const req = createMockRequest();
+  const res = createMockResponse();
+  let nextCount = 0;
+  const next: NextFunction = () => {
+    nextCount += 1;
+  };
+
+  limiter(req, res, next);
+  limiter(req, res, next);
+
+  assert.equal(nextCount, 2);
+  assert.equal(res.statusCode, 200);
+});
+
+test("createRateLimitMiddleware blocks requests over the limit", () => {
+  resetRateLimitBucketsForTest();
+
+  const limiter = createRateLimitMiddleware({
+    scope: "test:block",
+    windowMs: 60_000,
+    max: 1,
+  });
+  const req = createMockRequest();
+  const res = createMockResponse();
+  let nextCount = 0;
+  const next: NextFunction = () => {
+    nextCount += 1;
+  };
+
+  limiter(req, res, next);
+  limiter(req, res, next);
+
+  assert.equal(nextCount, 1);
+  assert.equal(res.statusCode, 429);
+  assert.equal(res.headers["Retry-After"], "60");
+  assert.equal(res.body.errorCode, "rate_limited");
+});
+
+test("createRateLimitMiddleware supports custom limit responses", () => {
+  resetRateLimitBucketsForTest();
+
+  const limiter = createRateLimitMiddleware({
+    scope: "test:custom",
+    windowMs: 60_000,
+    max: 1,
+    onLimit: (_req, res) => {
+      res.status(200).json({ code: 429, message: "limited" });
+    },
+  });
+  const req = createMockRequest();
+  const res = createMockResponse();
+
+  limiter(req, res, () => {});
+  limiter(req, res, () => {});
+
+  assert.equal(res.statusCode, 200);
+  assert.deepEqual(res.body, { code: 429, message: "limited" });
+});
+
+function createMockRequest(): Request {
+  return {
+    ip: "127.0.0.1",
+    originalUrl: "/test",
+    headers: {},
+    socket: {
+      remoteAddress: "127.0.0.1",
+    },
+  } as Request;
+}
+
+function createMockResponse(): Response & {
+  statusCode: number;
+  headers: Record<string, string>;
+  body: any;
+} {
+  const res = {
+    statusCode: 200,
+    headers: {} as Record<string, string>,
+    body: null as any,
+    setHeader(name: string, value: string) {
+      this.headers[name] = value;
+      return this;
+    },
+    status(statusCode: number) {
+      this.statusCode = statusCode;
+      return this;
+    },
+    json(body: any) {
+      this.body = body;
+      return this;
+    },
+  };
+
+  return res as Response & {
+    statusCode: number;
+    headers: Record<string, string>;
+    body: any;
+  };
+}

apps/backend/src/middleware/rate-limit.ts

@@ -0,0 +1,122 @@
+import type { NextFunction, Request, Response } from "express";
+
+import { buildErrorPayload, createHttpError } from "../utils/http.js";
+import { logWarn } from "../utils/logger.js";
+
+type RateLimitKeyResolver = (req: Request) => string;
+
+type RateLimitExceededContext = {
+  scope: string;
+  retryAfterSeconds: number;
+};
+
+type RateLimitOptions = {
+  scope: string;
+  windowMs: number;
+  max: number;
+  key?: RateLimitKeyResolver;
+  onLimit?: (req: Request, res: Response, context: RateLimitExceededContext) => void;
+};
+
+type RateLimitBucket = {
+  count: number;
+  resetAt: number;
+};
+
+const buckets = new Map<string, RateLimitBucket>();
+let lastCleanupAt = 0;
+
+export function createRateLimitMiddleware({
+  scope,
+  windowMs,
+  max,
+  key = defaultRateLimitKey,
+  onLimit,
+}: RateLimitOptions) {
+  const normalizedScope = String(scope || "default").trim() || "default";
+  const normalizedWindowMs = Math.max(1000, Number(windowMs || 0));
+  const normalizedMax = Math.max(1, Number(max || 0));
+
+  return (req: Request, res: Response, next: NextFunction): void => {
+    const now = Date.now();
+    cleanupExpiredBuckets(now);
+
+    const bucketKey = `${normalizedScope}:${key(req)}`;
+    const current = buckets.get(bucketKey);
+    const bucket = current && current.resetAt > now
+      ? current
+      : { count: 0, resetAt: now + normalizedWindowMs };
+
+    bucket.count += 1;
+    buckets.set(bucketKey, bucket);
+
+    if (bucket.count <= normalizedMax) {
+      next();
+      return;
+    }
+
+    const retryAfterSeconds = Math.max(1, Math.ceil((bucket.resetAt - now) / 1000));
+    res.setHeader("Retry-After", String(retryAfterSeconds));
+
+    logWarn("[rate-limit]", "请求触发限流", {
+      scope: normalizedScope,
+      ip: req.ip,
+      originalUrl: req.originalUrl,
+      retryAfterSeconds,
+    });
+
+    if (onLimit) {
+      onLimit(req, res, {
+        scope: normalizedScope,
+        retryAfterSeconds,
+      });
+      return;
+    }
+
+    const error = createHttpError("请求过于频繁，请稍后再试", {
+      statusCode: 429,
+      errorCode: "rate_limited",
+    });
+    res.status(429).json(buildErrorPayload(error, "请求过于频繁，请稍后再试"));
+  };
+}
+
+export function resetRateLimitBucketsForTest(): void {
+  buckets.clear();
+  lastCleanupAt = 0;
+}
+
+export function getBodyFieldRateLimitKey(fieldName: string): RateLimitKeyResolver {
+  return (req) => [defaultRateLimitKey(req), normalizeKeyPart(req.body?.[fieldName])].join(":");
+}
+
+export function getParamRateLimitKey(paramName: string): RateLimitKeyResolver {
+  return (req) => [defaultRateLimitKey(req), normalizeKeyPart(req.params?.[paramName])].join(":");
+}
+
+function defaultRateLimitKey(req: Request): string {
+  return normalizeKeyPart(
+    req.ip ||
+      String(req.headers["x-forwarded-for"] || "").split(",")[0] ||
+      req.socket.remoteAddress ||
+      "unknown",
+  );
+}
+
+function normalizeKeyPart(value: unknown): string {
+  const normalized = String(value || "").trim().toLowerCase();
+  return normalized || "unknown";
+}
+
+function cleanupExpiredBuckets(now: number): void {
+  if (now - lastCleanupAt < 60_000) {
+    return;
+  }
+
+  lastCleanupAt = now;
+  for (const [key, bucket] of buckets.entries()) {
+    if (bucket.resetAt <= now) {
+      buckets.delete(key);
+    }
+  }
+}

apps/backend/src/repositories/task-repo.ts

@@ -213,6 +213,55 @@ export async function updateTask(taskId: number | string, patch: TaskUpdatePatch
   })
 }
 
+export async function updateTaskStatusIfCurrent(
+  taskId: number | string,
+  currentStatus: string,
+  patch: TaskUpdatePatch,
+): Promise<TaskRow | null> {
+  const now = String(patch.updated_at || '').trim()
+  const result = await query<{ id: number }>(
+    `
+      UPDATE fulfillment_tasks
+      SET
+        task_status = $1,
+        delivery_status = COALESCE($2, delivery_status),
+        result_code = COALESCE($3, result_code),
+        result_message = COALESCE($4, result_message),
+        user_action_status = COALESCE($5, user_action_status),
+        last_error = COALESCE($6, last_error),
+        context_json = COALESCE($7::jsonb, context_json),
+        redeemed_at = COALESCE($8, redeemed_at),
+        updated_at = COALESCE($9, updated_at)
+      WHERE id = $10
+        AND task_status = $11
+      RETURNING id
+    `,
+    [
+      patch.task_status,
+      patch.delivery_status ?? null,
+      patch.result_code ?? null,
+      patch.result_message ?? null,
+      patch.user_action_status ?? null,
+      patch.last_error ?? null,
+      typeof patch.context_json === 'undefined'
+        ? null
+        : typeof patch.context_json === 'string'
+          ? patch.context_json
+          : JSON.stringify(patch.context_json || {}),
+      patch.redeemed_at ?? null,
+      now || null,
+      Number(taskId),
+      currentStatus,
+    ],
+  )
+
+  if (!result.rows[0]) {
+    return null
+  }
+
+  return getTaskById(taskId)
+}
+
 export async function getTaskById(taskId: number | string): Promise<TaskRow | null> {
   return getTaskByIdWithExecutor(query, taskId)
 }

apps/backend/src/routes/admin/auth.ts

@@ -1,11 +1,17 @@
 import { Router } from 'express'
 
+import { createRateLimitMiddleware, getBodyFieldRateLimitKey } from '../../middleware/rate-limit.js'
 import { getAdminSessionSummary, loginAdmin } from '../../services/admin/admin-auth-service.js'
 import { createJsonHandler, extractBearerToken } from './shared.js'
 
 const router = Router()
 
-router.post('/auth/login', createJsonHandler(
+router.post('/auth/login', createRateLimitMiddleware({
+  scope: 'admin:login',
+  windowMs: 60_000,
+  max: 10,
+  key: getBodyFieldRateLimitKey('username'),
+}), createJsonHandler(
   (req) => loginAdmin(req.body?.username, req.body?.password),
   {
     successMessage: '登录成功',

apps/backend/src/routes/claims.ts

@@ -1,5 +1,6 @@
 import { Router } from "express";
 
+import { createRateLimitMiddleware, getParamRateLimitKey } from "../middleware/rate-limit.js";
 import {
   confirmKuaishouCloudClaimRole,
   getKuaishouCloudClaimDetail,
@@ -14,9 +15,22 @@ import {
 } from "../utils/http.js";
 
 const router = Router();
+const claimReadRateLimit = createRateLimitMiddleware({
+  scope: "claim:read",
+  windowMs: 60_000,
+  max: 120,
+  key: getParamRateLimitKey("token"),
+});
+const claimWriteRateLimit = createRateLimitMiddleware({
+  scope: "claim:write",
+  windowMs: 60_000,
+  max: 30,
+  key: getParamRateLimitKey("token"),
+});
 
 router.get(
   "/:token",
+  claimReadRateLimit,
   createRouteHandler((req) => getKuaishouCloudClaimDetail(req.params.token), {
     errorMessage: "查询快手领取详情失败",
     scope: "[claims/:token]",
@@ -25,6 +39,7 @@ router.get(
 
 router.post(
   "/:token/kuaishou-cloud/verify-ticket",
+  claimWriteRateLimit,
   createRouteHandler(
     (req) => verifyKuaishouCloudClaimTicket(req.params.token, req.body),
     {
@@ -37,6 +52,7 @@ router.post(
 
 router.post(
   "/:token/kuaishou-cloud/confirm-role",
+  claimWriteRateLimit,
   createRouteHandler((req) => confirmKuaishouCloudClaimRole(req.params.token), {
     successMessage: "角色已确认",
     errorMessage: "确认角色失败",
@@ -46,6 +62,7 @@ router.post(
 
 router.post(
   "/:token/kuaishou-cloud/redeem",
+  claimWriteRateLimit,
   createRouteHandler((req) => redeemKuaishouCloudClaim(req.params.token), {
     successMessage: "兑换请求已提交",
     errorMessage: "兑换失败",

apps/backend/src/routes/open-91.ts

@@ -1,13 +1,22 @@
 import { Router } from 'express'
 
+import { createRateLimitMiddleware } from '../middleware/rate-limit.js'
 import { createOpen91Order } from '../services/open-91/order-create-service.js'
 import { queryOpen91Order } from '../services/open-91/order-query-service.js'
 import { buildOpen91ErrorResponse } from '../services/open-91/shared.js'
 import { createRequestId, logIntegration } from '../utils/logger.js'
 
 const router = Router()
+const open91RateLimit = createRateLimitMiddleware({
+  scope: 'open91',
+  windowMs: 60_000,
+  max: 120,
+  onLimit: (_req, res) => {
+    res.status(200).json(buildOpen91ErrorResponse('请求过于频繁，请稍后再试', 429))
+  },
+})
 
-router.post('/orders/create', async (req, res) => {
+router.post('/orders/create', open91RateLimit, async (req, res) => {
   const requestId = createRequestId('91')
   const startedAt = Date.now()
 
@@ -39,7 +48,7 @@ router.post('/orders/create', async (req, res) => {
   }
 })
 
-router.post('/orders/query', async (req, res) => {
+router.post('/orders/query', open91RateLimit, async (req, res) => {
   const requestId = createRequestId('91')
   const startedAt = Date.now()
 

apps/backend/src/services/claim/kuaishou-cloud-claim-service.ts

@@ -3,7 +3,7 @@ import path from 'node:path'
 
 import { PROJECT_ROOT } from '../../config/runtime.js'
 import { createTaskEvent } from '../../repositories/task-event-repo.js'
-import { updateTask } from '../../repositories/task-repo.js'
+import { getTaskById, updateTask, updateTaskStatusIfCurrent } from '../../repositories/task-repo.js'
 import { createHttpError } from '../../utils/http.js'
 import { nowIso } from '../../utils/time.js'
 import {
@@ -201,7 +201,7 @@ export async function confirmKuaishouCloudClaimRole(token: unknown) {
     })
   }
 
-  if (['role_confirmed', 'dispatched_pending_return', 'completed'].includes(String(context.task.task_status || '').trim())) {
+  if (['role_confirmed', 'redeeming', 'dispatched_pending_return', 'completed', 'manual_review'].includes(String(context.task.task_status || '').trim())) {
     return getKuaishouCloudClaimDetail(token)
   }
 
@@ -257,18 +257,58 @@ export async function redeemKuaishouCloudClaim(token: unknown) {
     })
   }
 
-  if (context.task.task_status !== 'role_confirmed') {
+  const currentStatus = String(context.task.task_status || '').trim()
+  if (['redeeming', 'dispatched_pending_return', 'completed', 'manual_review'].includes(currentStatus)) {
+    return getKuaishouCloudClaimDetail(token)
+  }
+
+  if (currentStatus !== 'role_confirmed') {
     throw createHttpError('请先确认角色信息', {
       statusCode: 409,
       errorCode: 'claim_kuaishou_cloud_role_not_confirmed',
     })
   }
 
-  await dispatchKuaishouCloudFulfillmentTask(context.task, {
-    source: 'claim_page_redeem',
-    actor: { source: 'claim_page' },
-    autoFinalize: true,
+  const lockedTask = await updateTaskStatusIfCurrent(context.task.id, 'role_confirmed', {
+    task_status: 'redeeming',
+    user_action_status: 'not_required',
+    last_error: '',
+    updated_at: now,
   })
 
+  if (!lockedTask) {
+    return getKuaishouCloudClaimDetail(token)
+  }
+
+  try {
+    await dispatchKuaishouCloudFulfillmentTask(lockedTask, {
+      source: 'claim_page_redeem',
+      actor: { source: 'claim_page' },
+      autoFinalize: true,
+    })
+  } catch (error) {
+    const latestTask = await getTaskById(lockedTask.id)
+    if (latestTask && String(latestTask.task_status || '').trim() !== 'redeeming') {
+      return getKuaishouCloudClaimDetail(token)
+    }
+
+    const message = error instanceof Error ? error.message : '兑换请求提交失败，请联系客服处理'
+    await updateTask(lockedTask.id, {
+      task_status: 'manual_review',
+      user_action_status: 'not_required',
+      last_error: message,
+      result_code: 'kuaishou_cloud_redeem_failed',
+      result_message: message,
+      updated_at: nowIso(),
+    })
+
+    await createTaskEvent(lockedTask.id, 'kuaishou_cloud_redeem_failed', {
+      source: 'claim_page_redeem',
+      errorMessage: message,
+    }, nowIso())
+
+    throw error
+  }
+
   return getKuaishouCloudClaimDetail(token)
 }