perf(cmpp): reduce inbound database round trips

2026-08-20 15:29:45 +08:00
parent 67b760a599
commit 14f993c1f8
12 changed files with 153 additions and 24 deletions
@@ -3,6 +3,7 @@ import { RiskReviewService } from './risk-review.service';
 function createPrismaMock(overrides: Record<string, unknown> = {}) {
  return {
    riskRule: {
+      count: jest.fn().mockResolvedValue(5),
      findFirst: jest.fn().mockResolvedValue({ id: 'default-rule' }),
      findUnique: jest.fn(),
      create: jest.fn(),
@@ -61,6 +62,51 @@ function createPrismaMock(overrides: Record<string, unknown> = {}) {
 }

 describe('RiskReviewService', () => {
+  it('coalesces concurrent default-rule checks and reuses the short completeness cache', async () => {
+    const prisma = createPrismaMock();
+    let releaseCount: ((count: number) => void) | undefined;
+    prisma.riskRule.count.mockReturnValue(new Promise((resolve) => { releaseCount = resolve; }));
+    const service = new RiskReviewService(prisma as never);
+
+    const first = service.ensureDefaultRules();
+    const second = service.ensureDefaultRules();
+    releaseCount?.(5);
+    await Promise.all([first, second]);
+    await service.ensureDefaultRules();
+
+    expect(prisma.riskRule.count).toHaveBeenCalledTimes(1);
+    expect(prisma.riskRule.findFirst).not.toHaveBeenCalled();
+  });
+
+  it('clears a failed default-rule check so the next request can retry', async () => {
+    const prisma = createPrismaMock();
+    prisma.riskRule.count
+      .mockRejectedValueOnce(new Error('database unavailable'))
+      .mockResolvedValueOnce(5);
+    const service = new RiskReviewService(prisma as never);
+
+    await expect(service.ensureDefaultRules()).rejects.toThrow('database unavailable');
+    await expect(service.ensureDefaultRules()).resolves.toBeUndefined();
+
+    expect(prisma.riskRule.count).toHaveBeenCalledTimes(2);
+  });
+
+  it('falls back to per-rule recovery when the completeness count finds a missing default', async () => {
+    const prisma = createPrismaMock();
+    prisma.riskRule.count.mockResolvedValue(4);
+    prisma.riskRule.findFirst.mockImplementation(({ where }: { where: { code: string } }) => (
+      Promise.resolve(where.code === 'PHONE_FREQUENCY_5M' ? null : { id: `rule-${where.code}` })
+    ));
+    const service = new RiskReviewService(prisma as never);
+    service.createRule = jest.fn().mockResolvedValue({ id: 'restored-rule' }) as never;
+
+    await service.ensureDefaultRules();
+
+    expect(prisma.riskRule.findFirst).toHaveBeenCalledTimes(5);
+    expect(service.createRule).toHaveBeenCalledTimes(1);
+    expect(service.createRule).toHaveBeenCalledWith(expect.objectContaining({ code: 'PHONE_FREQUENCY_5M' }));
+  });
+
  it('keeps phone-frequency periods fixed and rejects manual-review actions', () => {
    const service = new RiskReviewService(createPrismaMock() as never);

@@ -114,6 +114,8 @@ const RULE_DEFINITIONS = new Map(DEFAULT_RULES.map((rule) => [rule.code, rule]))

@Injectable()
 export class RiskReviewService {
+  private defaultRulesCheck?: { expiresAt: number; promise: Promise<void> };
+
  constructor(private readonly prisma: PrismaService) {}

  async listRules(applicationId?: string) {
@@ -491,14 +493,39 @@ export class RiskReviewService {
  }

  async ensureDefaultRules() {
+    const now = Date.now();
+    if (this.defaultRulesCheck && this.defaultRulesCheck.expiresAt > now) {
+      return this.defaultRulesCheck.promise;
+    }
+
+    // Submit高并发时，任务风控和号码频控都会确认默认规则。短TTL只缓存“规则是否齐全”，
+    // 实际生效规则仍逐次查询；并发单飞避免每条短信重复执行5次存在性SQL，同时允许删除后自动恢复。
+    const promise = this.ensureDefaultRulesFromDatabase();
+    this.defaultRulesCheck = { expiresAt: now + 30_000, promise };
+    try {
+      await promise;
+    } catch (error) {
+      if (this.defaultRulesCheck?.promise === promise) this.defaultRulesCheck = undefined;
+      throw error;
+    }
+  }
+
+  private async ensureDefaultRulesFromDatabase() {
+    const existingCount = await this.prisma.riskRule.count({
+      where: {
+        applicationId: null,
+        code: { in: DEFAULT_RULES.map((rule) => rule.code) },
+        status: { not: 'deleted' },
+      },
+    });
+    if (existingCount === DEFAULT_RULES.length) return;
+
    for (const rule of DEFAULT_RULES) {
      const exists = await this.prisma.riskRule.findFirst({
        where: { applicationId: null, code: rule.code, status: { not: 'deleted' } },
        select: { id: true },
      });
-      if (!exists) {
-        await this.createRule(rule);
-      }
+      if (!exists) await this.createRule(rule);
    }
  }

@@ -397,6 +397,7 @@ function createService(
  );
  service['postGatewayControl'] = jest.fn().mockResolvedValue({ delivered: true });
  service['publishGatewaySubmitCommand'] = jest.fn().mockResolvedValue(undefined);
+  service['getSendQueue'] = jest.fn().mockReturnValue({ add: jest.fn().mockResolvedValue(undefined) });
  return { service, prisma, billing, riskReview, phoneFrequency };
 }

@@ -1627,6 +1628,7 @@ describe('SendChainService', () => {
    })).resolves.toEqual(expect.objectContaining({ accepted: true, phoneCount: 2 }));

    expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(2);
+    expect(prisma.smsApplication.findFirst).toHaveBeenCalledTimes(1);
    expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({
      where: { id: 'record-2' },
      data: expect.objectContaining({
@@ -1766,7 +1768,10 @@ describe('SendChainService', () => {
      templateId: 'tpl-code',
      variables: { code: '715021' },
    }));
-    expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
+    expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1', {
+      messageRecordId: 'record-1',
+      queuePriority: 'normal',
+    });
    expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled();
  });

@@ -1814,7 +1819,10 @@ describe('SendChainService', () => {
      where: { id: 'record-1' },
      data: { status: 'queued', signatureId: 'sig-1' },
    });
-    expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
+    expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1', {
+      messageRecordId: 'record-1',
+      queuePriority: 'normal',
+    });
    expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled();
  });

@@ -2090,6 +2098,26 @@ describe('SendChainService', () => {
    });
  });

+  it('enqueues a freshly persisted inbound message without querying the task and message again', async () => {
+    const { service, prisma } = createService();
+    const add = jest.fn().mockResolvedValue(undefined);
+    service['getSendQueue'] = jest.fn().mockReturnValue({ add });
+
+    await expect(service.enqueueBatchTask('task-1', {
+      messageRecordId: 'record-1',
+      queuePriority: 'priority',
+    })).resolves.toEqual({ taskId: 'task-1', enqueued: 1 });
+
+    expect(prisma.smsBatchTask.findUnique).not.toHaveBeenCalled();
+    expect(prisma.smsMessageRecord.findMany).not.toHaveBeenCalled();
+    expect(add).toHaveBeenCalledWith('send-message', { messageRecordId: 'record-1' }, {
+      jobId: 'record-1',
+      attempts: 3,
+      priority: 1,
+    });
+    expect(prisma.smsBatchTask.update).toHaveBeenCalledWith({ where: { id: 'task-1' }, data: { status: 'queued' } });
+  });
+
  it('reuses persisted carrier and province without querying routing dictionaries again', async () => {
    const { service, prisma } = createService();
    service['identifyCarrier'] = jest.fn();
@@ -352,8 +352,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    return this.submission.confirmImport(data);
  }

-  async enqueueBatchTask(taskId: string) {
-    return this.submission.enqueueBatchTask(taskId);
+  async enqueueBatchTask(taskId: string, preparedMessage?: { messageRecordId: string; queuePriority?: string | null }) {
+    return this.submission.enqueueBatchTask(taskId, preparedMessage);
  }

  async cancelBatchTask(taskId: string, tenantId?: string, sourceType = 'client') {
@@ -599,10 +599,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    data: GatewayInboundSubmitDto & { phoneNumber: string },
    messageId: string,
    submitGroupMessageId: string,
+    application: NonNullable<Awaited<ReturnType<SendChainService['findInboundApplication']>>>,
    synchronousRejection?: { code: string; reason: string },
    receiptRejection?: { code: string; reason: string },
  ) {
-    return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, synchronousRejection, receiptRejection);
+    return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection);
  }

  /**
@@ -67,7 +67,19 @@ export class SendGatewaySubmitService {
  }


-async enqueueBatchTask(taskId: string) {
+async enqueueBatchTask(taskId: string, preparedMessage?: { messageRecordId: string; queuePriority?: string | null }) {
+    if (preparedMessage) {
+      // CMPP内部任务在当前请求内刚完成持久化且不暴露取消入口，可安全复用已知ID；
+      // 普通批量任务仍走下方查询路径，以保留取消检查和多消息枚举语义。
+      const queuePriority = normalizeQueuePriority(preparedMessage.queuePriority);
+      await this.facade.getSendQueue().add('send-message', { messageRecordId: preparedMessage.messageRecordId }, {
+        jobId: preparedMessage.messageRecordId,
+        attempts: 3,
+        priority: BULLMQ_PRIORITY[queuePriority],
+      });
+      await this.prisma.smsBatchTask.update({ where: { id: taskId }, data: { status: 'queued' } });
+      return { taskId, enqueued: 1 };
+    }
    const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
    if (!task) {
      throw new NotFoundException('SMS batch task not found');
@@ -356,7 +356,7 @@ async submitCompleteInboundMessage(
            ...data,
            phoneNumber: submission.phoneNumber,
            phoneNumbers: undefined,
-          }, submission.messageId, submitGroupMessageId, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection))));
+          }, submission.messageId, submitGroupMessageId, application, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection))));
    }
    const first = results[0];
    return {
@@ -550,16 +550,11 @@ async submitInboundSingleMessage(
    data: GatewayInboundSubmitDto & { phoneNumber: string },
    messageId: string,
    submitGroupMessageId: string,
+    application: NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>,
    synchronousRejection?: { code: string; reason: string },
    receiptRejection?: { code: string; reason: string },
  ) {
-    const application = await this.measureInboundStage(
-      'application_lookup',
-      () => this.facade.findInboundApplication(data.account),
-    );
-    if (!application) {
-      throw new BadRequestException('CMPP account is invalid');
-    }
+    // 入口已按账号取得并校验同一个应用快照；复用它可避免每个目标号码再次查询应用、企业和IP白名单。
    if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
      throw new BadRequestException('CMPP source IP is not in application allowlist');
    }
@@ -715,7 +710,10 @@ async submitInboundSingleMessage(
          where: { id: task.id },
          data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' },
        });
-        await this.facade.enqueueBatchTask(task.id);
+        await this.facade.enqueueBatchTask(task.id, {
+          messageRecordId: message.id,
+          queuePriority,
+        });
      });
    };
    if (receiptRejection) {
@@ -164,10 +164,11 @@ async submitInboundSingleMessage(
    data: GatewayInboundSubmitDto & { phoneNumber: string },
    messageId: string,
    submitGroupMessageId: string,
+    application: NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>,
    synchronousRejection?: { code: string; reason: string },
    receiptRejection?: { code: string; reason: string },
  ) {
-    return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, synchronousRejection, receiptRejection);
+    return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection);
  }

 async evaluateRiskWithPhoneFrequency(input: {
@@ -214,8 +215,8 @@ async runScheduledDispatchScan() {
    return this.scheduledDispatch.runScheduledDispatchScan();
  }

-async enqueueBatchTask(taskId: string) {
-    return this.gatewaySubmit.enqueueBatchTask(taskId);
+async enqueueBatchTask(taskId: string, preparedMessage?: { messageRecordId: string; queuePriority?: string | null }) {
+    return this.gatewaySubmit.enqueueBatchTask(taskId, preparedMessage);
  }

 startWorker() {