feat(M08-B): 补保洁微信转账回调与轮询

backend/src/routes/cleaning.ts

@@ -1,4 +1,7 @@
-import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
+import { Transform } from 'node:stream';
+import type {
+  FastifyInstance, FastifyReply, FastifyRequest, preParsingHookHandler
+} from 'fastify';
 import { z } from 'zod';
 import type { AuthRepository } from '../auth/auth-repository.js';
 import { authenticateAccessToken } from '../auth/authenticate.js';
@@ -14,7 +17,7 @@ import {
   CleaningPayoutError,
   type CleaningPayoutService
 } from '../cleaning/cleaning-payout-service.js';
-import { WechatPayError } from '../payments/wechat-pay-client.js';
+import { WechatPayError, type WechatNotificationHeaders } from '../payments/wechat-pay-client.js';
 
 const listSchema = z.object({
   page: z.coerce.number().int().min(1).default(1),
@@ -74,6 +77,9 @@ const settlementWechatTransferSchema = z.object({
   mode: z.enum(['API', 'MOCK']).default('API'),
   note: z.string().trim().max(512).optional()
 }).strict();
+const settlementWechatSyncSchema = z.object({
+  note: z.string().trim().max(512).optional()
+}).strict();
 const reclaimSchema = z.object({
   olderThanMinutes: z.coerce.number().int().min(5).max(1440).default(60),
   limit: z.coerce.number().int().min(1).max(100).default(20)
@@ -87,7 +93,8 @@ export interface CleaningRouteOptions {
     | 'recordSettlementPayoutFailure' | 'reclaimTimeouts'
     | 'assertCanUploadPhoto' | 'stats'>;
   mediaStorage?: MediaStorage;
-  payoutService?: Pick<CleaningPayoutService, 'executeWechatTransfer'>;
+  payoutService?: Pick<CleaningPayoutService,
+    'executeWechatTransfer' | 'syncWechatTransfer' | 'processWechatTransferNotification'>;
   authRepository: Pick<AuthRepository, 'validateSession'>;
   accessControl: { getAccessProfile(tenantId: string, userId: string): Promise<AccessProfile> };
   jwtSecret: string;
@@ -457,6 +464,48 @@ export async function registerCleaningRoutes(
       traceId: request.traceId
     }));
   });
+
+  app.post('/admin-api/cleaning/settlements/:settlementId/wechat-transfer/sync', async (request, reply) => {
+    if (!options.payoutService) return reply.status(501).send({
+      code: 'CLEANING_PAYOUT_UNAVAILABLE',
+      message: 'Cleaning payout service is not configured.',
+      traceId: request.traceId
+    });
+    const actor = await requireActor(request, reply, options, 'write');
+    if (!actor) return;
+    const params = settlementParamsSchema.safeParse(request.params);
+    const body = settlementWechatSyncSchema.safeParse(request.body ?? {});
+    if (!params.success || !body.success) return invalid(reply, request.traceId);
+    return handle(reply, request.traceId, async () => ({
+      code: 0,
+      data: await options.payoutService!.syncWechatTransfer({
+        ...actor,
+        settlementId: params.data.settlementId,
+        note: body.data.note
+      }),
+      traceId: request.traceId
+    }));
+  });
+
+  app.post('/app-api/cleaning/wechat-transfer/notify', {
+    preParsing: captureRawBody
+  }, async (request, reply) => {
+    if (!options.payoutService) return reply.status(501).send({
+      code: 'CLEANING_PAYOUT_UNAVAILABLE',
+      message: 'Cleaning payout service is not configured.',
+      traceId: request.traceId
+    });
+    const headers = wechatHeaders(request);
+    if (!headers) return invalid(reply, request.traceId);
+    return handle(reply, request.traceId, async () => {
+      await options.payoutService!.processWechatTransferNotification(
+        headers,
+        request.rawBody,
+        request.traceId
+      );
+      return { code: 'SUCCESS', message: '成功', traceId: request.traceId };
+    });
+  });
 }
 
 async function requireActor(
@@ -532,3 +581,33 @@ function invalid(reply: FastifyReply, traceId: string) {
     traceId
   });
 }
+
+function wechatHeaders(request: FastifyRequest): WechatNotificationHeaders | null {
+  const read = (key: string) => singleHeader(request.headers[key]);
+  const headers = {
+    timestamp: read('wechatpay-timestamp'),
+    nonce: read('wechatpay-nonce'),
+    serial: read('wechatpay-serial'),
+    signature: read('wechatpay-signature')
+  };
+  return headers.timestamp && headers.nonce && headers.serial && headers.signature
+    ? headers as WechatNotificationHeaders
+    : null;
+}
+
+const captureRawBody: preParsingHookHandler = (request, _reply, payload, done) => {
+  const chunks: Buffer[] = [];
+  const capture = new Transform({
+    transform(chunk, _encoding, callback) {
+      chunks.push(Buffer.from(chunk));
+      callback(null, chunk);
+    },
+    flush(callback) {
+      request.rawBody = Buffer.concat(chunks).toString('utf8');
+      callback();
+    }
+  });
+  const transformed = payload.pipe(capture) as typeof payload;
+  transformed.receivedEncodedLength = payload.receivedEncodedLength;
+  done(null, transformed);
+};