fix: make SMS retry side effects idempotent

2026-07-26 22:31:30 +08:00
parent 04f78695ed
commit e0f6eed0d4
11 changed files with 516 additions and 105 deletions
@@ -1,4 +1,5 @@
 import { createHash } from 'node:crypto';
+import { Prisma } from '@prisma/client';
 import { BillingService } from '../billing/billing.service';
 import { RiskReviewService } from '../risk-review/risk-review.service';
 import { SendChainService } from './send-chain.service';
@@ -153,7 +154,11 @@ function createPrismaMock() {
      create: jest.fn().mockResolvedValue({ id: 'submit-1' }),
      updateMany: jest.fn().mockResolvedValue({ count: 1 }),
      findFirst: jest.fn().mockResolvedValue({ id: 'submit-1', submitId: 'SUB-1', submitStatus: 'accepted', createdAt: new Date('2026-07-01T10:00:00.000Z') }),
-      findUnique: jest.fn().mockResolvedValue({ id: 'submit-1', submitId: 'SUB-1', submitStatus: 'accepted' }),
+      findUnique: jest.fn().mockImplementation(({ where }) => Promise.resolve(
+        where.retryOfSubmitRecordId
+          ? null
+          : { id: 'submit-1', messageRecordId: 'record-1', channelId: 'channel-1', submitId: 'SUB-1', submitStatus: 'accepted' },
+      )),
      count: jest.fn().mockResolvedValue(1),
      findMany: jest.fn().mockResolvedValue([]),
    },
@@ -1971,7 +1976,7 @@ describe('SendChainService', () => {
      },
    });
    prisma.smsSubmitRecord.findMany.mockResolvedValue([
-      { id: 'submit-1', submitId: 'SUB-1', channelId: primary.id, createdAt: new Date() },
+      { id: 'submit-1', submitId: 'SUB-1', messageRecordId: 'record-1', channelId: primary.id, createdAt: new Date() },
    ]);
    const submitMessageToGateway = jest.spyOn(service as any, 'submitMessageToGateway')
      .mockResolvedValue({ submitted: true, messageRecordId: 'record-1', channelId: backup.id, attempt: 1 });
@@ -1997,9 +2002,67 @@ describe('SendChainService', () => {
      expect.objectContaining({ signatureId: 'sig-direct' }),
      expect.objectContaining({ channel: expect.objectContaining({ id: backup.id }) }),
      1,
+      'submit-1',
    );
  });

+  it('allows only one retry submit when three long-message failure receipts race', async () => {
+    const { service, prisma } = createService();
+    const channel = await prisma.smsChannel.findUnique();
+    const queueAdd = jest.fn().mockResolvedValue(undefined);
+    jest.spyOn(service as any, 'getGatewayQueue').mockReturnValue({ add: queueAdd });
+    jest.spyOn(service as any, 'waitForChannelRateLimit').mockResolvedValue(undefined);
+    let claimedRetry: Record<string, unknown> | null = null;
+    prisma.smsSubmitRecord.create.mockImplementation(async ({ data }) => {
+      if (data.retryOfSubmitRecordId) {
+        if (claimedRetry) {
+          throw new Prisma.PrismaClientKnownRequestError('duplicate retry claim', {
+            code: 'P2002',
+            clientVersion: '7.9.0',
+            meta: { target: ['retryOfSubmitRecordId'] },
+          });
+        }
+        claimedRetry = { id: 'retry-submit-1', ...data };
+        return claimedRetry;
+      }
+      return { id: 'submit-1', ...data };
+    });
+    prisma.smsSubmitRecord.findUnique.mockImplementation(({ where }) => Promise.resolve(
+      where.retryOfSubmitRecordId ? claimedRetry : null,
+    ));
+    const message = {
+      id: 'record-long-race',
+      tenantId: 'tenant-1',
+      batchTaskId: 'task-1',
+      applicationId: 'app-1',
+      templateId: 'tpl-1',
+      messageId: 'MSG-LONG-RACE',
+      phoneNumber: '13800000001',
+      content: '长短信'.repeat(136),
+      billingUnits: 3,
+      template: { signature: { id: 'sig-1', name: '签名' } },
+    };
+    const routed = {
+      channel,
+      groupId: 'group-1',
+      carrier: 'mobile',
+      province: '山东',
+      routeScope: 'national',
+    };
+
+    const results = await Promise.all([
+      (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'),
+      (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'),
+      (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'),
+    ]);
+
+    expect(results.filter((result) => result.submitted)).toHaveLength(1);
+    expect(results.filter((result) => result.duplicateRetry)).toHaveLength(2);
+    expect(queueAdd).toHaveBeenCalledTimes(1);
+    expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledTimes(1);
+    expect(prisma.smsMessageRecord.update).toHaveBeenCalledTimes(1);
+  });
+
  it('releases reservation for rejected submit result and refunds failed receipts', async () => {
    const { service, prisma, billing } = createService();
    prisma.smsBillingRecord.findFirst
@@ -2031,7 +2094,10 @@ describe('SendChainService', () => {
      receiptStatus: 'undelivered',
      rawStatus: 'UNDELIV',
    });
-    expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({ remark: '最终失败退款' }));
+    expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({
+      idempotencyKey: 'sms-refund:MSG-1',
+      remark: '最终失败退款',
+    }));
  });

  it('stops failed receipt retry after the configured minute limit', async () => {
@@ -2403,6 +2469,56 @@ describe('SendChainService', () => {
    expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(1);
  });

+  it('creates and sends only one downstream final receipt under concurrent completion', async () => {
+    const { service, prisma } = createService();
+    let claimedDelivery: Record<string, unknown> | null = null;
+    prisma.cmppDownstreamDelivery.create.mockImplementation(async ({ data }) => {
+      if (claimedDelivery) {
+        throw new Prisma.PrismaClientKnownRequestError('duplicate downstream receipt', {
+          code: 'P2002',
+          clientVersion: '7.9.0',
+          meta: { target: ['dedupeKey'] },
+        });
+      }
+      claimedDelivery = {
+        id: 'delivery-once',
+        ...data,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      };
+      return claimedDelivery;
+    });
+    prisma.cmppDownstreamDelivery.findUnique.mockImplementation(({ where }) => Promise.resolve(
+      where.dedupeKey || where.id === 'delivery-once' ? claimedDelivery : null,
+    ));
+    const payload = {
+      tenantId: 'tenant-1',
+      applicationId: 'app-1',
+      messageRecordId: 'record-long-race',
+      messageId: 'MSG-LONG-RACE',
+      deliveryType: 'receipt' as const,
+      payload: {
+        messageId: 'MSG-LONG-RACE',
+        receiptStatus: 'undelivered',
+        rawStatus: 'FLNIGLK',
+      },
+    };
+
+    const results = await Promise.all([
+      (service as any).queueAndTryDownstreamDelivery(payload),
+      (service as any).queueAndTryDownstreamDelivery(payload),
+      (service as any).queueAndTryDownstreamDelivery(payload),
+    ]);
+
+    expect(results.map((result) => result.id)).toEqual([
+      'delivery-once',
+      'delivery-once',
+      'delivery-once',
+    ]);
+    expect(service['postGatewayControl']).toHaveBeenCalledTimes(1);
+    expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(3);
+  });
+
  it('marks a long message failed when a non-primary segment returns an explicit failure', async () => {
    const { service, prisma, billing } = createService();
    prisma.smsMessageRecord.findUnique.mockResolvedValue({
@@ -2434,6 +2550,17 @@ describe('SendChainService', () => {
      { segmentIndex: 1, segmentTotal: 2, receiptStatus: null, rawStatus: null, deliveredAt: null },
      { segmentIndex: 2, segmentTotal: 2, receiptStatus: 'undelivered', rawStatus: 'YL:1014', deliveredAt: new Date() },
    ]);
+    prisma.smsSubmitRecord.findUnique.mockImplementation(({ where }) => Promise.resolve(
+      where.id === 'submit-long'
+        ? {
+          id: 'submit-long',
+          messageRecordId: 'record-long',
+          channelId: 'channel-1',
+          submitId: 'SUB-LONG-FAIL',
+          submitStatus: 'accepted',
+        }
+        : null,
+    ));
    prisma.smsBillingRecord.findFirst
      .mockResolvedValueOnce(null)
      .mockResolvedValueOnce({ id: 'billing-charged', billingStatus: 'charged' });
@@ -1108,7 +1108,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      }
    } else if (data.submitStatus !== 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
      const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
-      const retried = await this.retryMessageIfAllowed(businessMessage, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发');
+      const retried = await this.retryMessageIfAllowed(
+        businessMessage,
+        data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发',
+        submitRecord.id,
+      );
      if (retried) {
        await this.refreshTaskProgress(businessMessage.batchTaskId);
        return retried;
@@ -1465,7 +1469,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId;
    if (status === 'failed' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
      const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
-      const retried = await this.retryMessageIfAllowed(businessMessage, '回执失败补发');
+      const retried = await this.retryMessageIfAllowed(
+        businessMessage,
+        '回执失败补发',
+        resolved.submitRecordId,
+      );
      if (retried) {
        await this.refreshTaskProgress(businessMessage.batchTaskId);
        return retried;
@@ -2400,21 +2408,51 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      return null;
    }
    const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload };
-    const delivery = await this.prisma.cmppDownstreamDelivery.create({
-      data: {
-        tenantId: data.tenantId,
-        applicationId: data.applicationId,
-        messageRecordId: data.messageRecordId,
-        messageId: data.messageId,
-        deliveryType: data.deliveryType,
-        payload,
-        retryEnabled: deliveryAllowed && (data.deliveryType === 'uplink'
-          ? application?.downstreamUplinkRetryEnabled ?? true
-          : application?.downstreamReceiptRetryEnabled ?? true),
-        status: deliveryAllowed ? 'pending' : 'abandoned',
-        lastError: deliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
-      },
-    });
+    const dedupeKey = data.deliveryType === 'receipt' && data.messageRecordId
+      ? `receipt:${data.messageRecordId}`
+      : data.deliveryType === 'uplink' && typeof data.payload.uplinkMessageId === 'string'
+        ? `uplink:${data.payload.uplinkMessageId}`
+        : null;
+    let delivery;
+    try {
+      delivery = await this.prisma.cmppDownstreamDelivery.create({
+        data: {
+          tenantId: data.tenantId,
+          applicationId: data.applicationId,
+          messageRecordId: data.messageRecordId,
+          messageId: data.messageId,
+          dedupeKey,
+          deliveryType: data.deliveryType,
+          payload,
+          retryEnabled: deliveryAllowed && (data.deliveryType === 'uplink'
+            ? application?.downstreamUplinkRetryEnabled ?? true
+            : application?.downstreamReceiptRetryEnabled ?? true),
+          status: deliveryAllowed ? 'pending' : 'abandoned',
+          lastError: deliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
+        },
+      });
+    } catch (error) {
+      if (
+        dedupeKey
+        && error instanceof Prisma.PrismaClientKnownRequestError
+        && error.code === 'P2002'
+      ) {
+        const existing = await this.prisma.cmppDownstreamDelivery.findUnique({
+          where: { dedupeKey },
+        });
+        if (existing) {
+          this.logger.warn(`downstream_delivery_deduplicated ${JSON.stringify({
+            deliveryType: data.deliveryType,
+            messageRecordId: data.messageRecordId,
+            messageId: data.messageId,
+            dedupeKey,
+            deliveryId: existing.id,
+          })}`);
+          return existing;
+        }
+      }
+      throw error;
+    }
    if (!deliveryAllowed) {
      return delivery;
    }
@@ -3302,6 +3340,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      applicationId?: string | null;
      templateId?: string | null;
      signatureId?: string | null;
+      submitId?: string | null;
      messageId: string;
      phoneNumber: string;
      content: string;
@@ -3314,44 +3353,87 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    },
    routed: RoutedChannel,
    attempt: number,
+    retryOfSubmitRecordId?: string,
  ) {
    const channel = routed.channel;
    const upstreamSrcId = composeUpstreamSrcId(channel.srcId, message.applicationExtension);
    await this.ensureSignatureReportedForChannel(message, channel.id);
    await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
    const submitId = `SUB-${randomUUID()}`;
-    const session = await this.prisma.cmppSubmitSession.upsert({
-      where: { sessionNo: `OPEN-${channel.id}` },
-      update: { submitTotal: { increment: 1 } },
-      create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
-    });
-    await this.prisma.smsSubmitRecord.create({
-      data: {
-        tenantId: message.tenantId,
-        batchTaskId: message.batchTaskId,
-        messageRecordId: message.id,
-        channelId: channel.id,
-        sessionId: session.id,
-        submitId,
-        submitStatus: 'queued',
-        costUnitPrice: channel.unitPrice ?? 0,
-        costAmountCents: moneyToNumber(channel.unitPrice) * Math.max(1, message.billingUnits ?? 1),
-      },
-    });
-    await this.prisma.smsMessageRecord.update({
-      where: { id: message.id },
-      data: {
-        channelId: channel.id,
-        carrier: routed.carrier,
-        province: routed.province,
-        submitId,
-        status: 'submit_queued',
-        submitStatus: 'queued',
-        receiptStatus: null,
-        errorCode: null,
-        errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined,
-      },
-    });
+    try {
+      await this.prisma.$transaction(async (tx) => {
+        const session = await tx.cmppSubmitSession.upsert({
+          where: { sessionNo: `OPEN-${channel.id}` },
+          update: { submitTotal: { increment: 1 } },
+          create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
+        });
+        await tx.smsSubmitRecord.create({
+          data: {
+            tenantId: message.tenantId,
+            batchTaskId: message.batchTaskId,
+            messageRecordId: message.id,
+            channelId: channel.id,
+            sessionId: session.id,
+            retryOfSubmitRecordId,
+            submitId,
+            submitStatus: 'queued',
+            costUnitPrice: channel.unitPrice ?? 0,
+            costAmountCents: moneyToNumber(channel.unitPrice) * Math.max(1, message.billingUnits ?? 1),
+          },
+        });
+        await tx.smsMessageRecord.update({
+          where: { id: message.id },
+          data: {
+            channelId: channel.id,
+            carrier: routed.carrier,
+            province: routed.province,
+            submitId,
+            status: 'submit_queued',
+            submitStatus: 'queued',
+            receiptStatus: null,
+            errorCode: null,
+            errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined,
+          },
+        });
+      });
+      if (retryOfSubmitRecordId) {
+        this.logger.log(`sms_retry_claim_acquired ${JSON.stringify({
+          messageId: message.messageId,
+          messageRecordId: message.id,
+          retryOfSubmitRecordId,
+          submitId,
+          channelId: channel.id,
+        })}`);
+      }
+    } catch (error) {
+      if (
+        retryOfSubmitRecordId
+        && error instanceof Prisma.PrismaClientKnownRequestError
+        && error.code === 'P2002'
+      ) {
+        const existingRetry = await this.prisma.smsSubmitRecord.findUnique({
+          where: { retryOfSubmitRecordId },
+        });
+        if (existingRetry) {
+          this.logger.warn(`sms_retry_claim_reused ${JSON.stringify({
+            messageId: message.messageId,
+            messageRecordId: message.id,
+            retryOfSubmitRecordId,
+            submitId: existingRetry.submitId,
+            channelId: existingRetry.channelId,
+          })}`);
+          return {
+            submitted: false,
+            duplicateRetry: true,
+            messageRecordId: message.id,
+            channelId: existingRetry.channelId,
+            attempt,
+            submitId: existingRetry.submitId,
+          };
+        }
+      }
+      throw error;
+    }
    const command = {
      schemaVersion: 'v1',
      messageType: 'SubmitCommand',
@@ -3414,6 +3496,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      batchTaskId: string;
      applicationId?: string | null;
      templateId?: string | null;
+      signatureId?: string | null;
+      submitId?: string | null;
      messageId: string;
      phoneNumber: string;
      content: string;
@@ -3423,6 +3507,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      applicationExtension?: string | null;
    },
    reason: string,
+    sourceSubmitRecordId?: string,
  ) {
    const attempts = await this.prisma.smsSubmitRecord.findMany({
      where: { messageRecordId: message.id },
@@ -3430,6 +3515,38 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      take: 200,
    });
    const attemptedChannelIds = attempts.map((attempt) => attempt.channelId);
+    let sourceAttempt = sourceSubmitRecordId
+      ? attempts.find((attempt) => attempt.id === sourceSubmitRecordId)
+      : attempts.find((attempt) => attempt.submitId === message.submitId) ?? attempts[attempts.length - 1];
+    if (!sourceAttempt && sourceSubmitRecordId) {
+      sourceAttempt = await this.prisma.smsSubmitRecord.findUnique({
+        where: { id: sourceSubmitRecordId },
+      }) ?? undefined;
+    }
+    if (!sourceAttempt || sourceAttempt.messageRecordId !== message.id) {
+      this.logger.error(`sms_retry_route_failed ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        reason,
+        sourceSubmitRecordId,
+        sourceMessageRecordId: sourceAttempt?.messageRecordId,
+        error: sourceAttempt ? 'retry_source_submit_record_mismatch' : 'retry_source_submit_record_missing',
+      })}`);
+      return null;
+    }
+    const existingRetry = await this.prisma.smsSubmitRecord.findUnique({
+      where: { retryOfSubmitRecordId: sourceAttempt.id },
+    });
+    if (existingRetry) {
+      this.logger.warn(`sms_retry_claim_reused ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        retryOfSubmitRecordId: sourceAttempt.id,
+        submitId: existingRetry.submitId,
+        channelId: existingRetry.channelId,
+      })}`);
+      return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
+    }
    const ageMinutes = (Date.now() - new Date(message.queuedAt ?? Date.now()).getTime()) / 60_000;
    this.logger.log(`sms_retry_route_started ${JSON.stringify({
      messageId: message.messageId,
@@ -3469,7 +3586,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        where: { id: message.id },
        data: { errorMessage: reason },
      });
-      const retried = await this.submitMessageToGateway(message, routed, attempts.length);
+      const retried = await this.submitMessageToGateway(
+        message,
+        routed,
+        attempts.length,
+        sourceAttempt.id,
+      );
      this.logger.log(`sms_retry_route_selected ${JSON.stringify({
        messageId: message.messageId,
        messageRecordId: message.id,
@@ -3985,6 +4107,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      await this.billing.release({
        tenantId: message.tenantId,
        amountCents,
+        idempotencyKey: `sms-charge-release:${message.messageId}`,
        relatedType: 'sms_batch_task',
        relatedId: message.batchTaskId,
        remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`,
@@ -3993,6 +4116,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    const transaction = await this.billing.charge({
      tenantId: message.tenantId,
      amountCents,
+      idempotencyKey: `sms-charge:${message.messageId}`,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark: '提交成功扣费',
@@ -4038,6 +4162,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    await this.billing.release({
      tenantId: message.tenantId,
      amountCents,
+      idempotencyKey: `sms-reservation-release:${message.messageId}`,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark: `${remark}: ${message.messageId}`,
@@ -4063,6 +4188,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    const transaction = await this.billing.refund({
      tenantId: message.tenantId,
      amountCents,
+      idempotencyKey: `sms-refund:${message.messageId}`,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark,