feat: strengthen risk controls and review workflows

2026-07-26 13:08:12 +08:00
parent b461532075
commit 2ce682c3fc
34 changed files with 2167 additions and 390 deletions
@@ -402,6 +402,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    const phones = [...new Set(data.phones ?? [])];
    const schedule = parseSchedule(data);
    await this.validateSendResources(data.tenantId, data.applicationId, data.templateId);
+    const phoneRejections = await this.classifyRejectedPhones(data.tenantId, data.applicationId, phones);
+    const sendablePhones = phones.filter((phone) => !phoneRejections.has(phone));
    const [messageClassification, unitPrice, queuePriority, accessNumber] = await Promise.all([
      this.resolveTemplateMessageClassification(data.tenantId, data.applicationId, data.templateId, data.content),
      this.resolveUnitPrice(data.tenantId, data.applicationId),
@@ -419,16 +421,19 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        phones,
        variables: messageClassification.variables ?? data.variables,
        createdById: data.createdById,
+        sourceType: data.sourceType ?? 'client',
      });
    const billing = this.billing.estimateSmsCost({
      tenantId: data.tenantId,
      applicationId: data.applicationId,
      taskId: risk.task?.id,
      content: data.content,
-      phoneCount: phones.length,
+      phoneCount: sendablePhones.length,
      unitPrice,
    });
-    const batchStatus = statusFromRisk(risk.status, Boolean(schedule.scheduledAt));
+    const batchStatus = risk.status === 'approved' && sendablePhones.length === 0
+      ? 'failed'
+      : statusFromRisk(risk.status, Boolean(schedule.scheduledAt));
    const shouldReserveBalance = batchStatus === 'ready';
    if (risk.status === 'approved') {
      const accountCheck = await this.billing.checkAccount({
@@ -439,8 +444,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        throw new BadRequestException('企业账户余额不足');
      }
    }
-    if (data.applicationId && risk.status !== 'rejected') {
-      await this.reserveDailySendQuota(data.applicationId, phones.length);
+    if (data.applicationId && risk.status !== 'rejected' && sendablePhones.length > 0) {
+      await this.reserveDailySendQuota(data.applicationId, sendablePhones.length);
    }
    const task = await this.prisma.smsBatchTask.create({
      data: {
@@ -485,35 +490,50 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
          sendMode: schedule.scheduledAt ? 'scheduled' : 'immediate',
          scheduledAt: schedule.scheduledAt?.toISOString(),
        },
-        status: batchStatus === 'rejected' ? 'rejected' : 'accepted',
+        status: ['rejected', 'failed'].includes(batchStatus) ? 'rejected' : 'accepted',
      },
    });
    if (phones.length > 0) {
      await this.prisma.smsMessageRecord.createMany({
-        data: phones.map((phone) => ({
+        data: phones.map((phone) => {
+          const rejection = phoneRejections.get(phone);
+          const status = rejection
+            ? 'submit_failed'
+            : batchStatus === 'ready'
+              ? 'queued'
+              : batchStatus === 'scheduled'
+                ? 'scheduled'
+                : batchStatus;
+          return {
          tenantId: data.tenantId,
          batchTaskId: task.id,
          applicationId: data.applicationId,
          templateId: data.templateId,
          signatureId: messageClassification.signatureId,
          drainageInfoId: messageClassification.drainageInfoId,
+          reviewTaskId: !rejection && risk.status === 'pending_review' ? risk.task?.id : undefined,
          messageId: `MSG-${randomUUID()}`,
          clientMessageId: data.clientMessageId,
          phoneNumber: phone,
          content: data.content,
          billingUnits: billing.billingUnitsPerMessage,
-          unitPrice: billing.unitPrice,
-          amountCents: billing.billingUnitsPerMessage * billing.unitPrice,
+          unitPrice: rejection ? 0 : billing.unitPrice,
+          amountCents: rejection ? 0 : billing.billingUnitsPerMessage * billing.unitPrice,
          queuePriority,
          clientSrcId: accessNumber.clientSrcId,
          applicationExtension: accessNumber.applicationExtension,
-          status: batchStatus === 'ready' ? 'queued' : batchStatus === 'scheduled' ? 'scheduled' : batchStatus,
-          errorMessage: risk.status === 'rejected' ? risk.reason ?? undefined : undefined,
-        })),
+          status,
+          submitStatus: rejection ? 'rejected' : undefined,
+          errorCode: rejection?.code,
+          errorMessage: rejection?.reason ?? (risk.status === 'rejected' ? risk.reason ?? undefined : undefined),
+        };
+        }),
      });
    }
-    if (batchStatus === 'ready') {
+    if (batchStatus === 'ready' && sendablePhones.length > 0) {
      await this.enqueueBatchTask(task.id);
+    } else if (batchStatus === 'failed') {
+      await this.refreshTaskProgress(task.id);
    }
    return this.getBatchTask(task.id, undefined, data.sourceType ?? 'client');
  }
@@ -743,18 +763,25 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
  async handleReviewDecision(reviewTaskId: string, decision: 'approved' | 'rejected', reason: string) {
    const reviewTask = await this.prisma.smsSendTask.findUnique({
      where: { id: reviewTaskId },
-      include: {
-        messageRecords: {
-          where: { status: 'pending_review' },
-          include: { batchTask: true },
-        },
-      },
    });
-    if (!reviewTask || reviewTask.messageRecords.length === 0) {
+    if (!reviewTask) {
+      return { reviewTaskId, decision, affected: 0 };
+    }
+    const messageRecords = await this.prisma.smsMessageRecord.findMany({
+      where: {
+        status: 'pending_review',
+        OR: [
+          { reviewTaskId },
+          { batchTask: { riskTaskId: reviewTaskId } },
+        ],
+      },
+      include: { batchTask: true },
+    });
+    if (messageRecords.length === 0) {
      return { reviewTaskId, decision, affected: 0 };
    }
    const batchTaskIds = new Set<string>();
-    for (const message of reviewTask.messageRecords) {
+    for (const message of messageRecords) {
      if (!message.tenantId || !message.applicationId || !message.batchTaskId) continue;
      if (decision === 'approved') {
        await this.prisma.smsMessageRecord.update({
@@ -781,7 +808,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    for (const batchTaskId of batchTaskIds) {
      await this.enqueueBatchTask(batchTaskId);
    }
-    return { reviewTaskId, decision, affected: reviewTask.messageRecords.length };
+    return { reviewTaskId, decision, affected: messageRecords.length };
  }

  async terminateBatchTask(taskId: string) {
@@ -1440,8 +1467,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {

  async listPendingDownstreamDeliveries(data: GatewayPendingDeliveryQueryDto) {
    const application = await this.findInboundApplication(data.account);
-    if (!application || application.status !== 'active' || application.tenant.status !== 'active') {
-      throw new BadRequestException('CMPP account is invalid or disabled');
+    if (!application) {
+      throw new BadRequestException('CMPP account is invalid');
    }
    const expiredAcknowledgements = await this.prisma.cmppDownstreamDelivery.findMany({
      where: { applicationId: application.id, status: 'awaiting_ack', ackDeadlineAt: { lte: new Date() } },
@@ -2244,23 +2271,27 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      select: {
        cmppAccount: true,
        interfaceEnabled: true,
+        status: true,
        downstreamReceiptRetryEnabled: true,
        downstreamUplinkRetryEnabled: true,
        httpConfig: true,
      },
    });
-    try {
-      await this.openApi?.queueWebhookEvent({
-        tenantId: data.tenantId,
-        applicationId: data.applicationId,
-        messageRecordId: data.messageRecordId,
-        messageId: data.messageId,
-        uplinkMessageId: typeof data.payload.uplinkMessageId === 'string' ? data.payload.uplinkMessageId : undefined,
-        eventType: data.deliveryType,
-        payload: data.payload,
-      });
-    } catch (error) {
-      this.logger.error(`HTTP webhook queue failed for ${data.deliveryType}/${data.messageId ?? '-'}: ${error instanceof Error ? error.message : String(error)}`);
+    const deliveryAllowed = application?.status === 'active' || application?.status === 'disabling';
+    if (deliveryAllowed) {
+      try {
+        await this.openApi?.queueWebhookEvent({
+          tenantId: data.tenantId,
+          applicationId: data.applicationId,
+          messageRecordId: data.messageRecordId,
+          messageId: data.messageId,
+          uplinkMessageId: typeof data.payload.uplinkMessageId === 'string' ? data.payload.uplinkMessageId : undefined,
+          eventType: data.deliveryType,
+          payload: data.payload,
+        });
+      } catch (error) {
+        this.logger.error(`HTTP webhook queue failed for ${data.deliveryType}/${data.messageId ?? '-'}: ${error instanceof Error ? error.message : String(error)}`);
+      }
    }
    if (application?.interfaceEnabled !== true) {
      return null;
@@ -2274,12 +2305,16 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        messageId: data.messageId,
        deliveryType: data.deliveryType,
        payload,
-        retryEnabled: data.deliveryType === 'uplink'
+        retryEnabled: deliveryAllowed && (data.deliveryType === 'uplink'
          ? application?.downstreamUplinkRetryEnabled ?? true
-          : application?.downstreamReceiptRetryEnabled ?? true,
-        status: 'pending',
+          : application?.downstreamReceiptRetryEnabled ?? true),
+        status: deliveryAllowed ? 'pending' : 'abandoned',
+        lastError: deliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
      },
    });
+    if (!deliveryAllowed) {
+      return delivery;
+    }
    try {
      const result = await this.postGatewayControl(
        data.deliveryType === 'receipt' ? '/downstream/receipt' : '/downstream/uplink',
@@ -2413,7 +2448,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {

  async authenticateInboundApplication(data: GatewayInboundAuthDto) {
    const application = await this.findInboundApplication(data.account);
-    if (!application || application.status !== 'active' || application.tenant.status !== 'active') {
+    if (!application || !['active', 'disabling'].includes(application.status) || application.tenant.status !== 'active') {
      throw new BadRequestException('CMPP account is invalid or disabled');
    }
    if (!application.interfaceEnabled) {
@@ -2445,7 +2480,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      : data.phoneNumber
        ? [data.phoneNumber.trim()]
        : [];
-    if (phoneNumbers.length === 0 || phoneNumbers.some((phoneNumber) => !/^1[3-9]\d{9}$/.test(phoneNumber))) {
+    if (phoneNumbers.length === 0) {
      throw new BadRequestException('CMPP submit phone number is invalid');
    }

@@ -2453,6 +2488,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    if (!application) {
      throw new BadRequestException('CMPP account is invalid');
    }
+    if (application.status !== 'active' || application.tenant.status !== 'active' || !application.interfaceEnabled) {
+      throw new BadRequestException('CMPP account is disabled for new submissions');
+    }
    if (data.longMessage) {
      if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
        throw new BadRequestException('CMPP source IP is not in application allowlist');
@@ -2586,7 +2624,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        })
      : [];
    const persistedByPhone = new Map(persisted.map((item) => [item.phoneNumber, item]));
-    const missingPhoneCount = phoneNumbers.filter((phoneNumber) => !persistedByPhone.has(phoneNumber)).length;
+    const phoneRejections = await this.classifyRejectedPhones(application.tenantId, application.id, phoneNumbers);
+    const missingPhoneCount = phoneNumbers.filter((phoneNumber) => (
+      !persistedByPhone.has(phoneNumber) && !phoneRejections.has(phoneNumber)
+    )).length;
    const dailyQuota = missingPhoneCount > 0
      ? await this.tryReserveDailySendQuota(application.id, missingPhoneCount)
      : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 };
@@ -2601,6 +2642,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    const submissions = phoneNumbers.map((phoneNumber, index) => ({
      phoneNumber,
      persisted: persistedByPhone.get(phoneNumber),
+      receiptRejection: phoneRejections.get(phoneNumber),
      messageId: persistedByPhone.get(phoneNumber)?.messageId
        ?? (index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`),
    }));
@@ -2622,7 +2664,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
            ...data,
            phoneNumber: submission.phoneNumber,
            phoneNumbers: undefined,
-          }, submission.messageId, submitGroupMessageId, dailyLimitRejection))));
+          }, submission.messageId, submitGroupMessageId, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection))));
    }
    const first = results[0];
    return {
@@ -2811,6 +2853,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    messageId: string,
    submitGroupMessageId: string,
    synchronousRejection?: { code: string; reason: string },
+    receiptRejection?: { code: string; reason: string },
  ) {
    const application = await this.findInboundApplication(data.account);
    if (!application) {
@@ -2819,9 +2862,6 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
      throw new BadRequestException('CMPP source IP is not in application allowlist');
    }
-    if (!/^1[3-9]\d{9}$/.test(data.phoneNumber)) {
-      throw new BadRequestException('CMPP submit phone number is invalid');
-    }
    const clientSrcId = validateInboundApplicationSrcId(data.srcId, application);
    const template = await this.resolveInboundTemplateCandidate(application.id, data.content);
    const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : {};
@@ -2870,8 +2910,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        phoneNumber: data.phoneNumber,
        content: data.content,
        billingUnits: billing.billingUnitsPerMessage,
-        unitPrice: billing.unitPrice,
-        amountCents: billing.amountCents,
+        unitPrice: receiptRejection ? 0 : billing.unitPrice,
+        amountCents: receiptRejection ? 0 : billing.amountCents,
        queuePriority,
        cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId),
        cmppSubmitGroupMessageId: submitGroupMessageId,
@@ -2921,6 +2961,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        content: data.content,
        variables: options.templateId ? templateVariables : undefined,
        phones: [data.phoneNumber],
+        sourceType: 'cmpp',
      });
      if (risk.status === 'rejected') {
        await reject('RISK', risk.reason || '短信被风控拒绝');
@@ -2929,7 +2970,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      if (risk.status === 'pending_review') {
        await this.prisma.smsMessageRecord.update({
          where: { id: message.id },
-          data: { status: 'pending_review', signatureId: options.signatureId, drainageInfoId },
+          data: {
+            status: 'pending_review',
+            reviewTaskId: risk.task?.id,
+            signatureId: options.signatureId,
+            drainageInfoId,
+          },
        });
        await this.prisma.smsBatchTask.update({
          where: { id: task.id },
@@ -2964,7 +3010,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      });
      await this.enqueueBatchTask(task.id);
    };
-    if (application.status !== 'active' || application.tenant.status !== 'active') {
+    if (receiptRejection) {
+      await reject(receiptRejection.code, receiptRejection.reason);
+    } else if (application.status !== 'active' || application.tenant.status !== 'active') {
      await reject('ACCOUNT', '企业或短信应用已停用');
    } else if (!application.interfaceEnabled) {
      await reject('INTERFACE', '短信应用 CMPP 接口已停用');
@@ -2998,6 +3046,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
          applicationId: application.id,
          content: data.content,
          phones: [data.phoneNumber],
+          sourceType: 'cmpp',
        });
        if (risk.status === 'rejected') {
          await reject('RISK', risk.reason || '短信被风控拒绝');
@@ -3651,6 +3700,44 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    return receipt;
  }

+  private async classifyRejectedPhones(tenantId: string, applicationId: string | undefined, phones: string[]) {
+    const rejected = new Map<string, { code: string; reason: string }>();
+    for (const phone of phones) {
+      if (!/^1\d{10}$/.test(phone)) {
+        rejected.set(phone, { code: 'INVALID_PHONE', reason: '手机号码必须是1开头的11位数字' });
+      }
+    }
+    const validPhones = phones.filter((phone) => !rejected.has(phone));
+    if (validPhones.length === 0) {
+      return rejected;
+    }
+    const [globalHits, enterpriseHits] = await Promise.all([
+      this.prisma.globalBlacklist.findMany({
+        where: { phoneNumber: { in: validPhones }, status: 'active' },
+        select: { phoneNumber: true, reason: true },
+      }),
+      applicationId
+        ? this.prisma.enterpriseBlacklist.findMany({
+            where: { tenantId, applicationId, phoneNumber: { in: validPhones }, status: 'active' },
+            select: { phoneNumber: true, reason: true },
+          })
+        : Promise.resolve([]),
+    ]);
+    for (const hit of globalHits) {
+      rejected.set(hit.phoneNumber, {
+        code: 'GLOBAL_BLACKLIST',
+        reason: hit.reason?.trim() || '号码命中平台黑名单',
+      });
+    }
+    for (const hit of enterpriseHits) {
+      rejected.set(hit.phoneNumber, {
+        code: 'ENTERPRISE_BLACKLIST',
+        reason: hit.reason?.trim() || '号码命中企业应用黑名单',
+      });
+    }
+    return rejected;
+  }
+
  private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) {
    const tenant = await this.prisma.tenant.findUnique({ where: { id: tenantId } });
    if (!tenant || tenant.status !== 'active') {