fix: harden channel retry attribution and operations UI

2026-07-26 21:33:58 +08:00
parent 059b38e8fe
commit 0857de09d8
27 changed files with 888 additions and 117 deletions
@@ -1863,7 +1863,7 @@ describe('SendChainService', () => {
    });

    expect(prisma.smsSubmitRecord.updateMany).toHaveBeenCalledWith({
-      where: { submitId: 'SUB-1' },
+      where: { id: 'submit-1' },
      data: expect.objectContaining({ sequenceId: 7, gatewayMessageId: 'GW-1', submitStatus: 'accepted' }),
    });
    expect(prisma.smsMessageRecord.updateMany).toHaveBeenCalledWith({
@@ -1937,6 +1937,69 @@ describe('SendChainService', () => {
    });
  });

+  it('rejects a legacy aggregate SubmitResult when it cannot match one submit attempt uniquely', async () => {
+    const { service, prisma } = createService();
+    prisma.smsSubmitRecord.findMany.mockResolvedValue([
+      { id: 'submit-1', submitId: 'SUB-1', channelId: 'channel-1', gatewayMessageId: null },
+      { id: 'submit-2', submitId: 'SUB-2', channelId: 'channel-1', gatewayMessageId: null },
+    ]);
+
+    await expect(service.handleSubmitResult({
+      messageId: 'MSG-1',
+      channelId: 'channel-1',
+      gatewayMessageId: 'GW-LEGACY',
+      submitStatus: 'accepted',
+    })).rejects.toThrow('cannot be matched uniquely');
+
+    expect(prisma.smsSubmitRecord.updateMany).not.toHaveBeenCalled();
+    expect(prisma.smsMessageRecord.updateMany).not.toHaveBeenCalled();
+  });
+
+  it('retries a direct-signature CMPP message through the next approved channel', async () => {
+    const { service, prisma } = createService();
+    const route = await prisma.channelRouteRule.findFirst();
+    const primary = route.group.items[0].channel;
+    const backup = { ...primary, id: 'channel-backup', code: 'CMPP-B' };
+    prisma.channelRouteRule.findFirst.mockResolvedValue({
+      ...route,
+      group: {
+        ...route.group,
+        items: [
+          { ...route.group.items[0], channelId: primary.id, priority: 1, channel: primary },
+          { ...route.group.items[0], id: 'item-2', channelId: backup.id, priority: 2, channel: backup },
+        ],
+      },
+    });
+    prisma.smsSubmitRecord.findMany.mockResolvedValue([
+      { id: 'submit-1', submitId: 'SUB-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 });
+
+    await expect((service as any).retryMessageIfAllowed({
+      id: 'record-1',
+      tenantId: 'tenant-1',
+      batchTaskId: 'task-1',
+      applicationId: 'app-1',
+      templateId: null,
+      signatureId: 'sig-direct',
+      messageId: 'MSG-DIRECT-SIGNATURE',
+      phoneNumber: '13800000001',
+      content: '【签名】无模板内容',
+      billingUnits: 1,
+      queuedAt: new Date(),
+    }, '回执失败补发')).resolves.toEqual(expect.objectContaining({ channelId: backup.id }));
+
+    expect(prisma.channelSignatureReportTask.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      where: expect.objectContaining({ signatureId: 'sig-direct' }),
+    }));
+    expect(submitMessageToGateway).toHaveBeenCalledWith(
+      expect.objectContaining({ signatureId: 'sig-direct' }),
+      expect.objectContaining({ channel: expect.objectContaining({ id: backup.id }) }),
+      1,
+    );
+  });
+
  it('releases reservation for rejected submit result and refunds failed receipts', async () => {
    const { service, prisma, billing } = createService();
    prisma.smsBillingRecord.findFirst
@@ -1955,6 +2018,7 @@ describe('SendChainService', () => {
    await service.handleSubmitResult({
      messageId: 'MSG-1',
      channelId: 'channel-1',
+      submitId: 'SUB-1',
      gatewayMessageId: 'GW-1',
      submitStatus: 'rejected',
    });
@@ -2858,11 +2922,32 @@ describe('SendChainService', () => {
      }),
    }));
    expect(prisma.smsSubmitRecord.updateMany).toHaveBeenCalledWith(expect.objectContaining({
-      where: { submitId: 'SUB-1', gatewayMessageId: null },
+      where: { id: 'submit-1', gatewayMessageId: null },
      data: expect.objectContaining({ sequenceId: 71, gatewayMessageId: 'GW-SEG-1' }),
    }));
  });

+  it('rejects a legacy segment result when multiple channel attempts could match', async () => {
+    const { service, prisma } = createService();
+    prisma.smsSubmitRecord.findMany.mockResolvedValue([
+      { id: 'submit-2', submitId: 'SUB-2', messageRecordId: 'record-1', channelId: 'channel-1' },
+      { id: 'submit-1', submitId: 'SUB-1', messageRecordId: 'record-1', channelId: 'channel-1' },
+    ]);
+
+    await expect(service.handleSubmitSegmentResult({
+      messageId: 'MSG-1',
+      channelId: 'channel-1',
+      segmentTotal: 2,
+      segmentIndex: 1,
+      sequenceId: 72,
+      gatewayMessageId: 'GW-SEG-2',
+      submitStatus: 'accepted',
+    })).rejects.toThrow('cannot be matched uniquely');
+
+    expect(prisma.smsMessageSegmentAudit.upsert).not.toHaveBeenCalled();
+    expect(prisma.smsSubmitRecord.updateMany).not.toHaveBeenCalled();
+  });
+
  it('durably intakes an upstream receipt before asynchronous business matching', async () => {
    const { service, prisma } = createService();
    jest.spyOn(service as any, 'processUpstreamReceiptInboxRecord').mockResolvedValue(false);
@@ -977,11 +977,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {

  async handleSubmitSegmentResult(data: GatewaySubmitSegmentResultDto) {
    const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
+    const submitRecord = await this.resolveSubmitRecordForGatewaySegmentResult(message.id, data);
+    const effectiveSubmitId = submitRecord.submitId;
    const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
    await this.recordSubmitSegments(message, {
      messageId: data.messageId,
      channelId: data.channelId,
-      submitId: data.submitId,
+      submitId: effectiveSubmitId,
      sequenceId: data.sequenceId,
      gatewayMessageId: data.gatewayMessageId ?? '',
      submitStatus: normalizeSubmitStatus(data.submitStatus),
@@ -1002,7 +1004,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    if (data.gatewayMessageId) {
      await this.prisma.smsSubmitRecord.updateMany({
        where: {
-          ...(data.submitId ? { submitId: data.submitId } : { messageRecordId: message.id }),
+          id: submitRecord.id,
          gatewayMessageId: null,
        },
        data: {
@@ -1015,14 +1017,73 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    return { accepted: true };
  }

+  private async resolveSubmitRecordForGatewaySegmentResult(
+    messageRecordId: string,
+    data: GatewaySubmitSegmentResultDto,
+  ) {
+    if (data.submitId) {
+      const exact = await this.prisma.smsSubmitRecord.findUnique({
+        where: { submitId: data.submitId },
+      });
+      if (
+        !exact ||
+        (exact.messageRecordId && exact.messageRecordId !== messageRecordId) ||
+        (exact.channelId && exact.channelId !== data.channelId)
+      ) {
+        this.logger.error(`gateway_submit_segment_result_unmatched ${JSON.stringify({
+          messageId: data.messageId,
+          messageRecordId,
+          submitId: data.submitId,
+          channelId: data.channelId,
+          segmentIndex: data.segmentIndex,
+        })}`);
+        throw new BadRequestException(
+          'Gateway SubmitSegmentResult submitId does not match the SMS message and channel',
+        );
+      }
+      return exact;
+    }
+
+    const candidates = await this.prisma.smsSubmitRecord.findMany({
+      where: {
+        messageRecordId,
+        channelId: data.channelId,
+      },
+      orderBy: { createdAt: 'desc' },
+      take: 2,
+    });
+    if (candidates.length !== 1) {
+      this.logger.error(`gateway_submit_segment_result_unmatched ${JSON.stringify({
+        messageId: data.messageId,
+        messageRecordId,
+        channelId: data.channelId,
+        segmentIndex: data.segmentIndex,
+        candidateCount: candidates.length,
+      })}`);
+      throw new BadRequestException(
+        'Gateway SubmitSegmentResult without submitId cannot be matched uniquely',
+      );
+    }
+    this.logger.warn(`gateway_submit_segment_result_legacy_match ${JSON.stringify({
+      messageId: data.messageId,
+      messageRecordId,
+      channelId: data.channelId,
+      segmentIndex: data.segmentIndex,
+      submitId: candidates[0].submitId,
+    })}`);
+    return candidates[0];
+  }
+
  async handleSubmitResult(data: GatewaySubmitResultDto) {
    const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
+    const submitRecord = await this.resolveSubmitRecordForGatewayResult(message.id, data);
+    const effectiveData = { ...data, submitId: submitRecord.submitId };
    const batchTask = message.batchTaskId
      ? await this.prisma.smsBatchTask.findUnique({ where: { id: message.batchTaskId }, select: { sourceType: true } })
      : null;
    const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
    await this.prisma.smsSubmitRecord.updateMany({
-      where: data.submitId ? { submitId: data.submitId } : { messageRecordId: message.id },
+      where: { id: submitRecord.id },
      data: {
        sequenceId: data.sequenceId,
        gatewayMessageId: data.gatewayMessageId,
@@ -1032,9 +1093,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        submittedAt,
      },
    });
-    await this.recordSubmitSegments(message, data, submittedAt);
+    await this.recordSubmitSegments(message, effectiveData, submittedAt);
    const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId;
-    if (data.submitId && message.submitId && data.submitId !== message.submitId) {
+    if (message.submitId && effectiveData.submitId !== message.submitId) {
      return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
    }
    const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
@@ -1089,7 +1150,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      where: {
        status: { in: ['pending', 'requeueing', 'requeue_recovering', 'requeued'] },
        OR: [
-          data.submitId ? { submitId: data.submitId } : undefined,
+          { submitId: effectiveData.submitId },
          data.messageId ? { messageId: data.messageId } : undefined,
        ].filter(Boolean) as Array<{ submitId?: string; messageId?: string }>,
      },
@@ -1105,6 +1166,48 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
  }

+  private async resolveSubmitRecordForGatewayResult(messageRecordId: string, data: GatewaySubmitResultDto) {
+    if (data.submitId) {
+      const exact = await this.prisma.smsSubmitRecord.findUnique({ where: { submitId: data.submitId } });
+      if (!exact
+        || (exact.messageRecordId && exact.messageRecordId !== messageRecordId)
+        || (exact.channelId && exact.channelId !== data.channelId)) {
+        throw new BadRequestException('Gateway SubmitResult submitId does not match the SMS message and channel');
+      }
+      return exact;
+    }
+    const candidates = await this.prisma.smsSubmitRecord.findMany({
+      where: {
+        messageRecordId,
+        channelId: data.channelId,
+        OR: [
+          data.gatewayMessageId ? { gatewayMessageId: data.gatewayMessageId } : undefined,
+          { gatewayMessageId: null },
+        ].filter(Boolean) as Array<{ gatewayMessageId?: string | null }>,
+      },
+      orderBy: { createdAt: 'desc' },
+      take: 2,
+    });
+    if (candidates.length !== 1) {
+      this.logger.error(`gateway_submit_result_unmatched ${JSON.stringify({
+        messageId: data.messageId,
+        messageRecordId,
+        channelId: data.channelId,
+        gatewayMessageId: data.gatewayMessageId,
+        candidateCount: candidates.length,
+      })}`);
+      throw new BadRequestException('Gateway SubmitResult without submitId cannot be matched uniquely');
+    }
+    this.logger.warn(`gateway_submit_result_legacy_match ${JSON.stringify({
+      messageId: data.messageId,
+      messageRecordId,
+      channelId: data.channelId,
+      gatewayMessageId: data.gatewayMessageId,
+      submitId: candidates[0].submitId,
+    })}`);
+    return candidates[0];
+  }
+
  async intakeReceipt(data: GatewayReceiptEventDto) {
    const channel = await this.prisma.smsChannel.findUnique({
      where: { id: data.channelId },
@@ -3198,6 +3301,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      batchTaskId: string;
      applicationId?: string | null;
      templateId?: string | null;
+      signatureId?: string | null;
      messageId: string;
      phoneNumber: string;
      content: string;
@@ -3327,12 +3431,33 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    });
    const attemptedChannelIds = attempts.map((attempt) => attempt.channelId);
    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,
+      messageRecordId: message.id,
+      reason,
+      attemptedChannelIds,
+      ageMinutes: Math.round(ageMinutes * 100) / 100,
+    })}`);
    if (ageMinutes >= 72 * 60) {
+      this.logger.warn(`sms_retry_route_skipped ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        reason: 'maximum_message_age_exceeded',
+        ageMinutes: Math.round(ageMinutes * 100) / 100,
+      })}`);
      return null;
    }
    const route = await this.findApplicationRoute(message.tenantId, message.applicationId ?? undefined, await this.identifyCarrier(message.phoneNumber));
    const retryTimeLimitMinutes = Math.min(route.group.retryTimeLimitMinutes ?? route.group.retryTimeLimitHours * 60, 72 * 60);
    if (!route.group.retryEnabled || ageMinutes >= retryTimeLimitMinutes) {
+      this.logger.warn(`sms_retry_route_skipped ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        groupId: route.groupId,
+        reason: !route.group.retryEnabled ? 'group_retry_disabled' : 'group_retry_time_limit_exceeded',
+        ageMinutes: Math.round(ageMinutes * 100) / 100,
+        retryTimeLimitMinutes,
+      })}`);
      return null;
    }
    try {
@@ -3344,14 +3469,29 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        where: { id: message.id },
        data: { errorMessage: reason },
      });
-      return await this.submitMessageToGateway(message, routed, attempts.length);
-    } catch {
+      const retried = await this.submitMessageToGateway(message, routed, attempts.length);
+      this.logger.log(`sms_retry_route_selected ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        groupId: routed.groupId,
+        channelId: routed.channel.id,
+        attempt: attempts.length,
+      })}`);
+      return retried;
+    } catch (error) {
+      this.logger.error(`sms_retry_route_failed ${JSON.stringify({
+        messageId: message.messageId,
+        messageRecordId: message.id,
+        reason,
+        attemptedChannelIds,
+        error: error instanceof Error ? error.message : String(error),
+      })}`);
      return null;
    }
  }

  private async selectChannelForMessage(
-    message: { id: string; tenantId: string; applicationId?: string | null; templateId?: string | null; phoneNumber: string; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null },
+    message: { id: string; tenantId: string; applicationId?: string | null; templateId?: string | null; signatureId?: string | null; phoneNumber: string; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null },
    options: { forceNational?: boolean; excludeChannelIds?: string[] } = {},
  ): Promise<RoutedChannel> {
    if (!message.applicationId) {
@@ -3955,8 +4095,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    }
  }

-  private async resolveMessageSignatureId(message: { templateId?: string | null; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null }) {
-    const direct = message.template?.signature?.id ?? message.signature?.id ?? null;
+  private async resolveMessageSignatureId(message: { templateId?: string | null; signatureId?: string | null; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null }) {
+    const direct = message.signatureId ?? message.template?.signature?.id ?? message.signature?.id ?? null;
    if (direct || !message.templateId) return direct;
    const template = await this.prisma.smsTemplate.findUnique({ where: { id: message.templateId }, include: { signature: true } });
    return template?.signature?.id ?? null;