feat(M08-A): 接入充值微信支付调起

backend/src/db/migration-runner.ts

@@ -43,7 +43,8 @@ const migrationFiles: Record<MigrationDirection, readonly string[]> = {
     'database/migrations/2026062220_m06c_iot_messages.up.sql',
     'database/migrations/2026062421_m07a_wallet_ledger.up.sql',
     'database/migrations/2026062422_m07b_recharge_plans.up.sql',
-    'database/migrations/2026062423_m07c_benefits.up.sql'
+    'database/migrations/2026062423_m07c_benefits.up.sql',
+    'database/migrations/2026062524_m08a_recharge_wechat.up.sql'
   ],
   verify: [
     'database/migrations/2026061601_m01b_core_schema.verify.sql',
@@ -68,9 +69,11 @@ const migrationFiles: Record<MigrationDirection, readonly string[]> = {
     'database/migrations/2026062220_m06c_iot_messages.verify.sql',
     'database/migrations/2026062421_m07a_wallet_ledger.verify.sql',
     'database/migrations/2026062422_m07b_recharge_plans.verify.sql',
-    'database/migrations/2026062423_m07c_benefits.verify.sql'
+    'database/migrations/2026062423_m07c_benefits.verify.sql',
+    'database/migrations/2026062524_m08a_recharge_wechat.verify.sql'
   ],
   down: [
+    'database/migrations/2026062524_m08a_recharge_wechat.down.sql',
     'database/migrations/2026062423_m07c_benefits.down.sql',
     'database/migrations/2026062422_m07b_recharge_plans.down.sql',
     'database/migrations/2026062421_m07a_wallet_ledger.down.sql',

backend/src/routes/recharge.ts

@@ -1,9 +1,11 @@
-import type { FastifyInstance, FastifyReply } from 'fastify';
+import type { FastifyInstance, FastifyReply, preParsingHookHandler } from 'fastify';
+import { Transform } from 'node:stream';
 import { z } from 'zod';
 import type { AuthRepository } from '../auth/auth-repository.js';
 import { authenticateAccessToken } from '../auth/authenticate.js';
 import type { AccessProfile } from '../auth/rbac-repository.js';
 import { RechargeError, type RechargeService } from '../wallets/recharge-service.js';
+import { WechatPayError } from '../payments/wechat-pay-client.js';
 
 const listSchema = z.object({
   storeId: z.string().regex(/^[1-9]\d{0,19}$/).optional()
@@ -13,9 +15,15 @@ const createSchema = z.object({
   storeId: z.string().regex(/^[1-9]\d{0,19}$/).nullable().optional(),
   clientRequestId: z.string().min(8).max(128)
 }).strict();
+const orderParams = z.object({
+  rechargeOrderId: z.string().regex(/^[1-9]\d{0,19}$/)
+});
 
 export interface RechargeRouteOptions {
-  service: Pick<RechargeService, 'listAvailablePlans' | 'createRechargeOrder'>;
+  service: Pick<
+    RechargeService,
+    'listAvailablePlans' | 'createRechargeOrder' | 'createWechatPrepay' | 'processWechatNotification'
+  >;
   authRepository: Pick<AuthRepository, 'validateSession'>;
   accessControl: { getAccessProfile(tenantId: string, userId: string): Promise<AccessProfile> };
   jwtSecret: string;
@@ -58,6 +66,36 @@ export async function registerRechargeRoutes(
       traceId: request.traceId
     }));
   });
+
+  app.post('/app-api/recharge/orders/:rechargeOrderId/wechat-prepay', async (request, reply) => {
+    const auth = await authenticate(request.headers.authorization, options);
+    const params = orderParams.safeParse(request.params);
+    if (!auth) return unauthorized(reply, request.traceId);
+    if (!params.success) return invalid(reply, request.traceId);
+    return handle(reply, request.traceId, async () => ({
+      code: 0,
+      data: await options.service.createWechatPrepay({
+        tenantId: auth.tenantId,
+        platformAppId: auth.platformAppId,
+        userId: auth.userId,
+        rechargeOrderId: params.data.rechargeOrderId
+      }),
+      traceId: request.traceId
+    }));
+  });
+
+  app.post('/app-api/recharge/wechat/notify', {
+    preParsing: captureRawBody
+  }, async (request, reply) => {
+    return handle(reply, request.traceId, async () => {
+      const data = await options.service.processWechatNotification(
+        notificationHeaders(request.headers),
+        request.rawBody,
+        request.traceId
+      );
+      return reply.send({ code: 'SUCCESS', message: '成功', data });
+    });
+  });
 }
 
 async function authenticate(
@@ -77,6 +115,7 @@ async function authenticate(
   if (!access.capabilities.includes('profile.read')) return null;
   return {
     tenantId: result.session.tenantId,
+    platformAppId: result.session.platformAppId,
     userId: result.session.user.id
   };
 }
@@ -85,11 +124,12 @@ async function handle(reply: FastifyReply, traceId: string, work: () => Promise<
   try {
     return await work();
   } catch (error) {
-    if (!(error instanceof RechargeError)) throw error;
+    if (!(error instanceof RechargeError) && !(error instanceof WechatPayError)) throw error;
     const status = error.code === 'RECHARGE_PLAN_NOT_FOUND'
       || error.code === 'RECHARGE_ORDER_NOT_FOUND'
       ? 404
-      : error.code === 'RECHARGE_LIMIT_REACHED' ? 409 : 400;
+      : error.code === 'RECHARGE_LIMIT_REACHED' ? 409
+        : error.code.includes('FORBIDDEN') ? 403 : 400;
     return reply.status(status).send({
       code: error.code,
       message: 'The recharge request is not available.',
@@ -98,6 +138,39 @@ async function handle(reply: FastifyReply, traceId: string, work: () => Promise<
   }
 }
 
+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);
+};
+
+function notificationHeaders(headers: Record<string, unknown>) {
+  const read = (name: string) => {
+    const value = headers[name];
+    if (typeof value !== 'string' || value.length === 0) {
+      throw new WechatPayError('WECHAT_NOTIFICATION_HEADER_INVALID');
+    }
+    return value;
+  };
+  return {
+    timestamp: read('wechatpay-timestamp'),
+    nonce: read('wechatpay-nonce'),
+    serial: read('wechatpay-serial'),
+    signature: read('wechatpay-signature')
+  };
+}
+
 function unauthorized(reply: FastifyReply, traceId: string) {
   return reply.status(401).send({
     code: 'AUTH_SESSION_INVALID',

backend/src/server.ts

@@ -174,7 +174,11 @@ const app = await buildApp({
     jwtSecret: config.auth.jwtSecret
   },
   recharge: {
-    service: new RechargeService(pool, walletLedgerService),
+    service: new RechargeService(pool, walletLedgerService, {
+      paymentRepository,
+      client: wechatPayClient,
+      credentials: wechatCredentials
+    }),
     authRepository,
     accessControl,
     jwtSecret: config.auth.jwtSecret

backend/src/wallets/recharge-service.ts

@@ -1,6 +1,13 @@
 import { randomBytes } from 'node:crypto';
 import type { PoolConnection, ResultSetHeader, RowDataPacket } from 'mysql2/promise';
 import type { MySqlPool } from '../db/mysql.js';
+import type { PaymentRepository } from '../payments/payment-repository.js';
+import {
+  WechatPayClient,
+  WechatPayError,
+  type WechatNotificationHeaders,
+  type WechatPayCredential
+} from '../payments/wechat-pay-client.js';
 import type { WalletLedgerService } from './wallet-ledger-service.js';
 
 interface PlanRow extends RowDataPacket {
@@ -26,6 +33,7 @@ interface RechargeOrderRow extends RowDataPacket {
   payAmountCents: number;
   giftAmountCents: number;
   status: string;
+  providerPaymentId?: string | null;
 }
 
 interface CountRow extends RowDataPacket { total: number }
@@ -37,7 +45,12 @@ export class RechargeError extends Error {
 export class RechargeService {
   constructor(
     private readonly pool: MySqlPool,
-    private readonly wallet: Pick<WalletLedgerService, 'credit'>
+    private readonly wallet: Pick<WalletLedgerService, 'credit'>,
+    private readonly wechat?: {
+      paymentRepository: Pick<PaymentRepository, 'resolveConfig'>;
+      client: WechatPayClient;
+      credentials: ReadonlyMap<string, WechatPayCredential>;
+    }
   ) {}
 
   async listAvailablePlans(input: {
@@ -158,6 +171,113 @@ export class RechargeService {
     });
   }
 
+  async createWechatPrepay(input: {
+    tenantId: string;
+    platformAppId: string;
+    userId: string;
+    rechargeOrderId: string;
+  }) {
+    if (!this.wechat) throw new WechatPayError('WECHAT_PAYMENT_NOT_CONFIGURED');
+    const order = await this.loadOwnedRechargeOrder(input, false);
+    if (order.status === 'CREDITED') throw new RechargeError('RECHARGE_ALREADY_CREDITED');
+    if (order.status !== 'PENDING_PAYMENT' && order.status !== 'PAID') {
+      throw new RechargeError('RECHARGE_STATUS_INVALID');
+    }
+    const config = await this.wechat.paymentRepository.resolveConfig(
+      this.pool,
+      input.tenantId,
+      input.platformAppId,
+      order.storeId ?? '0',
+      'WECHAT'
+    );
+    const credential = this.resolveWechatCredential(config.credentialRef);
+    const settings = config.settings as Record<string, unknown>;
+    const [identityRows] = await this.pool.execute<RowDataPacket[]>(
+      `SELECT openid FROM qipai_user_identities
+       WHERE tenant_id = ? AND platform_app_id = ? AND user_id = ?
+         AND provider = 'WECHAT' LIMIT 1`,
+      [input.tenantId, input.platformAppId, input.userId]
+    );
+    if (!identityRows[0]?.openid) throw new WechatPayError('WECHAT_OPENID_NOT_FOUND');
+    const result = await this.wechat.client.createJsapiPrepay(credential, {
+      description: typeof settings.rechargeDescription === 'string'
+        ? settings.rechargeDescription : `棋牌室余额充值 ${order.rechargeNo}`,
+      outTradeNo: order.rechargeNo,
+      notifyUrl: settingUrl(
+        settings, 'rechargeNotifyUrl', 'https://api.txyundm.cn/app-api/recharge/wechat/notify'
+      ),
+      amountCents: Number(order.payAmountCents),
+      payerOpenId: String(identityRows[0].openid)
+    });
+    await this.pool.execute(
+      `UPDATE qipai_recharge_orders
+       SET payment_provider = 'WECHAT', provider_prepay_id = ?
+       WHERE tenant_id = ? AND id = ? AND user_id = ?`,
+      [result.prepayId, input.tenantId, input.rechargeOrderId, input.userId]
+    );
+    return {
+      rechargeOrderId: input.rechargeOrderId,
+      rechargeNo: order.rechargeNo,
+      amountCents: Number(order.payAmountCents),
+      ...result.paymentParams
+    };
+  }
+
+  async processWechatNotification(
+    headers: WechatNotificationHeaders,
+    rawBody: string,
+    traceId: string
+  ) {
+    if (!this.wechat) throw new WechatPayError('WECHAT_PAYMENT_NOT_CONFIGURED');
+    const data = this.verifyWithConfiguredCredential(headers, rawBody);
+    const rechargeNo = requiredString(data, 'out_trade_no');
+    const transactionId = requiredString(data, 'transaction_id');
+    const tradeState = requiredString(data, 'trade_state');
+    const amount = objectValue(data.amount);
+    const total = requiredNumber(amount, 'total');
+    return this.transaction(async (connection) => {
+      const order = await this.loadRechargeOrderByNo(connection, rechargeNo, true);
+      const callbackId = `${headers.serial}:${transactionId}:${tradeState}`;
+      const [callback] = await connection.execute<ResultSetHeader>(
+        `INSERT IGNORE INTO qipai_payment_callbacks
+          (tenant_id, payment_id, provider, callback_id, callback_type,
+           verified, payload)
+         VALUES (?, NULL, 'WECHAT_RECHARGE', ?, 'PAYMENT_NOTIFICATION', 1, CAST(? AS JSON))`,
+        [order.tenantId, callbackId, JSON.stringify(redactNotification(data))]
+      );
+      if (callback.affectedRows === 0) {
+        return { rechargeOrderId: order.id, status: order.status, idempotent: true };
+      }
+      if (tradeState !== 'SUCCESS' || total !== Number(order.payAmountCents)) {
+        const code = tradeState !== 'SUCCESS'
+          ? `WECHAT_TRADE_${tradeState}` : 'PAYMENT_AMOUNT_MISMATCH';
+        await connection.execute(
+          `UPDATE qipai_payment_callbacks
+           SET processing_status = 'REJECTED', error_code = ?,
+               processed_at = UTC_TIMESTAMP(3)
+           WHERE tenant_id = ? AND provider = 'WECHAT_RECHARGE' AND callback_id = ?`,
+          [code, order.tenantId, callbackId]
+        );
+        return { rechargeOrderId: order.id, status: 'REJECTED', code, idempotent: false };
+      }
+      const credited = await this.creditRechargeOrder(connection, {
+        tenantId: String(order.tenantId),
+        rechargeOrderId: String(order.id),
+        providerTransactionId: transactionId,
+        providerCallbackId: callbackId,
+        rawNotify: redactNotification(data),
+        traceId
+      });
+      await connection.execute(
+        `UPDATE qipai_payment_callbacks
+         SET processing_status = 'PROCESSED', processed_at = UTC_TIMESTAMP(3)
+         WHERE tenant_id = ? AND provider = 'WECHAT_RECHARGE' AND callback_id = ?`,
+        [order.tenantId, callbackId]
+      );
+      return credited;
+    });
+  }
+
   private async loadPlan(connection: PoolConnection, tenantId: string, planId: string) {
     const [rows] = await connection.execute<PlanRow[]>(
       `SELECT id, tenant_id AS tenantId, store_id AS storeId, name,
@@ -217,6 +337,128 @@ export class RechargeService {
     };
   }
 
+  private async loadOwnedRechargeOrder(
+    input: { tenantId: string; userId: string; rechargeOrderId: string },
+    lock: boolean
+  ) {
+    const [rows] = await this.pool.execute<RechargeOrderRow[]>(
+      `SELECT id, user_id AS userId, store_id AS storeId, plan_id AS planId,
+              recharge_no AS rechargeNo, pay_amount_cents AS payAmountCents,
+              gift_amount_cents AS giftAmountCents, status,
+              provider_payment_id AS providerPaymentId
+       FROM qipai_recharge_orders
+       WHERE tenant_id = ? AND id = ? AND user_id = ?
+       ${lock ? 'FOR UPDATE' : ''}`,
+      [input.tenantId, input.rechargeOrderId, input.userId]
+    );
+    if (!rows[0]) throw new RechargeError('RECHARGE_ORDER_NOT_FOUND');
+    return normalizeRechargeOrder(rows[0]);
+  }
+
+  private async loadRechargeOrderByNo(
+    connection: PoolConnection,
+    rechargeNo: string,
+    lock: boolean
+  ) {
+    const [rows] = await connection.execute<Array<RechargeOrderRow & { tenantId: string }>>(
+      `SELECT id, tenant_id AS tenantId, user_id AS userId,
+              store_id AS storeId, plan_id AS planId, recharge_no AS rechargeNo,
+              pay_amount_cents AS payAmountCents,
+              gift_amount_cents AS giftAmountCents, status,
+              provider_payment_id AS providerPaymentId
+       FROM qipai_recharge_orders
+       WHERE recharge_no = ?
+       ${lock ? 'FOR UPDATE' : ''}`,
+      [rechargeNo]
+    );
+    if (!rows[0]) throw new RechargeError('RECHARGE_ORDER_NOT_FOUND');
+    return { ...normalizeRechargeOrder(rows[0]), tenantId: String(rows[0].tenantId) };
+  }
+
+  private async creditRechargeOrder(
+    connection: PoolConnection,
+    input: {
+      tenantId: string;
+      rechargeOrderId: string;
+      providerTransactionId: string;
+      providerCallbackId: string;
+      rawNotify: Record<string, unknown>;
+      traceId: string;
+    }
+  ) {
+    const order = await this.loadRechargeOrder(connection, input, true);
+    if (order.status === 'CREDITED') {
+      return { rechargeOrderId: order.id, status: 'CREDITED', idempotent: true };
+    }
+    if (order.status !== 'PENDING_PAYMENT' && order.status !== 'PAID') {
+      throw new RechargeError('RECHARGE_STATUS_INVALID');
+    }
+    await connection.execute(
+      `UPDATE qipai_recharge_orders
+       SET status = 'PAID',
+           provider_payment_id = ?,
+           provider_callback_id = ?,
+           raw_notify = CAST(? AS JSON),
+           paid_at = COALESCE(paid_at, UTC_TIMESTAMP(3))
+       WHERE tenant_id = ? AND id = ?`,
+      [
+        input.providerTransactionId,
+        input.providerCallbackId,
+        JSON.stringify(input.rawNotify),
+        input.tenantId,
+        order.id
+      ]
+    );
+    const plan = await this.loadPlan(connection, input.tenantId, order.planId);
+    await this.wallet.credit({
+      tenantId: input.tenantId,
+      userId: String(order.userId),
+      scopeType: plan.scopeType,
+      storeId: plan.scopeType === 'STORE' ? order.storeId : null,
+      businessType: 'RECHARGE',
+      businessId: order.id,
+      entryType: 'RECHARGE',
+      cashDeltaCents: Number(order.payAmountCents),
+      giftDeltaCents: Number(order.giftAmountCents),
+      traceId: input.traceId,
+      metadata: {
+        provider: 'WECHAT',
+        providerTransactionId: input.providerTransactionId,
+        rechargeNo: order.rechargeNo
+      }
+    });
+    await connection.execute(
+      `UPDATE qipai_recharge_orders
+       SET status = 'CREDITED', credited_at = UTC_TIMESTAMP(3)
+       WHERE tenant_id = ? AND id = ?`,
+      [input.tenantId, order.id]
+    );
+    return { rechargeOrderId: order.id, status: 'CREDITED', idempotent: false };
+  }
+
+  private resolveWechatCredential(reference: string) {
+    const credential = this.wechat?.credentials.get(reference)
+      ?? this.wechat?.credentials.get(reference.replace(/^env:/, ''));
+    if (!credential) throw new WechatPayError('WECHAT_CREDENTIAL_NOT_CONFIGURED');
+    return credential;
+  }
+
+  private verifyWithConfiguredCredential(
+    headers: WechatNotificationHeaders,
+    rawBody: string
+  ) {
+    let lastError: unknown;
+    for (const credential of this.wechat?.credentials.values() ?? []) {
+      if (!credential.platformCertificates[headers.serial]) continue;
+      try {
+        return this.wechat!.client.verifyAndDecrypt(credential, headers, rawBody);
+      } catch (error) {
+        lastError = error;
+      }
+    }
+    throw lastError ?? new WechatPayError('WECHAT_CERTIFICATE_NOT_FOUND');
+  }
+
   private async assertPurchaseLimit(
     connection: PoolConnection,
     tenantId: string,
@@ -276,3 +518,52 @@ function rechargeResponse(row: RechargeOrderRow, idempotent: boolean) {
     idempotent
   };
 }
+
+function normalizeRechargeOrder(row: RechargeOrderRow): RechargeOrderRow {
+  return {
+    ...row,
+    id: String(row.id),
+    userId: String(row.userId),
+    storeId: row.storeId === null ? null : String(row.storeId),
+    planId: String(row.planId),
+    payAmountCents: Number(row.payAmountCents),
+    giftAmountCents: Number(row.giftAmountCents)
+  };
+}
+
+function settingUrl(
+  settings: Record<string, unknown>, key: string, fallback: string
+) {
+  const value = settings[key];
+  return typeof value === 'string' && value.startsWith('https://') ? value : fallback;
+}
+
+function requiredString(value: Record<string, unknown>, key: string) {
+  const field = value[key];
+  if (typeof field !== 'string' || field.length === 0) {
+    throw new WechatPayError('WECHAT_RESOURCE_INVALID');
+  }
+  return field;
+}
+
+function requiredNumber(value: Record<string, unknown>, key: string) {
+  const field = value[key];
+  if (typeof field !== 'number' || !Number.isInteger(field)) {
+    throw new WechatPayError('WECHAT_RESOURCE_INVALID');
+  }
+  return field;
+}
+
+function objectValue(value: unknown) {
+  if (!value || typeof value !== 'object' || Array.isArray(value)) {
+    throw new WechatPayError('WECHAT_RESOURCE_INVALID');
+  }
+  return value as Record<string, unknown>;
+}
+
+function redactNotification(value: Record<string, unknown>) {
+  const copy = { ...value };
+  delete copy.payer;
+  delete copy.user_received_account;
+  return copy;
+}