fix: enforce application limits and signature format

2026-07-22 16:29:22 +08:00
parent cd085d2712
commit a09036c67b
20 changed files with 479 additions and 79 deletions
@@ -300,6 +300,7 @@ function createPrismaMock() {
    globalBlacklist: {
      findMany: jest.fn().mockResolvedValue([]),
    },
+    $queryRaw: jest.fn().mockResolvedValue([{ dailyLimit: 100000, usedCount: 2 }]),
    $transaction: jest.fn((operations) => Promise.all(operations)),
  };
 }
@@ -364,6 +365,23 @@ describe('SendChainService', () => {
    expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
  });

+  it('rejects the whole batch atomically when the application daily send limit would be exceeded', async () => {
+    const { service, prisma, billing } = createService();
+    prisma.$queryRaw.mockResolvedValueOnce([{ dailyLimit: 1, usedCount: null }]);
+
+    await expect(service.createBatchTask({
+      tenantId: 'tenant-1',
+      applicationId: 'app-1',
+      templateId: 'tpl-1',
+      content: 'hello',
+      phones: ['13800000001', '13800000002'],
+    })).rejects.toThrow('应用当日发送上限1条');
+
+    expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
+    expect(prisma.smsMessageRecord.createMany).not.toHaveBeenCalled();
+    expect(billing.freeze).not.toHaveBeenCalled();
+  });
+
  it('recognizes an approved template for public HTTP content and reads back the api task', async () => {
    const { service, prisma, riskReview } = createService();
    service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
@@ -832,6 +850,31 @@ describe('SendChainService', () => {
    expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(2);
  });

+  it('rejects every destination in one CMPP Submit with auditable receipts when the daily limit is exceeded', async () => {
+    const { service, prisma, billing } = createService();
+    prisma.$queryRaw.mockResolvedValueOnce([{ dailyLimit: 1, usedCount: null }]);
+    let taskIndex = 0;
+    prisma.smsBatchTask.create.mockImplementation(({ data }) => Promise.resolve({ id: `task-${++taskIndex}`, ...data }));
+    let messageIndex = 0;
+    prisma.smsMessageRecord.create.mockImplementation(({ data }) => Promise.resolve({ id: `record-${++messageIndex}`, ...data }));
+
+    const result = await service.submitInboundMessage({
+      account: '100001',
+      phoneNumbers: ['13800000001', '13900000002'],
+      content: 'hello',
+      sequenceId: 88,
+      remoteIp: '127.0.0.1',
+    });
+
+    expect(result).toEqual(expect.objectContaining({ accepted: true, phoneCount: 2 }));
+    expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(2);
+    expect(prisma.smsReceiptRecord.create).toHaveBeenCalledTimes(2);
+    expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
+      data: expect.objectContaining({ errorCode: 'DAILY_LIMIT', receiptStatus: 'undelivered' }),
+    });
+    expect(billing.freeze).not.toHaveBeenCalled();
+  });
+
  it('rejects a multi-destination CMPP Submit before persistence when any destination is invalid', async () => {
    const { service, prisma } = createService();

@@ -1,4 +1,4 @@
-import { BadRequestException, forwardRef, Inject, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common';
+import { BadRequestException, forwardRef, HttpException, HttpStatus, Inject, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common';
 import { Prisma } from '@prisma/client';
 import { Queue, Worker } from 'bullmq';
 import IORedis from 'ioredis';
@@ -366,6 +366,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        throw new BadRequestException('企业账户余额不足');
      }
    }
+    if (data.applicationId && risk.status !== 'rejected') {
+      await this.reserveDailySendQuota(data.applicationId, phones.length);
+    }
    const task = await this.prisma.smsBatchTask.create({
      data: {
        tenantId: data.tenantId,
@@ -2020,6 +2023,15 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      throw new BadRequestException('CMPP submit phone number is invalid');
    }

+    const application = await this.findInboundApplication(data.account);
+    if (!application) {
+      throw new BadRequestException('CMPP account is invalid');
+    }
+    const dailyQuota = await this.tryReserveDailySendQuota(application.id, phoneNumbers.length);
+    const dailyLimitFailure = dailyQuota.reserved
+      ? undefined
+      : `应用当日发送上限${dailyQuota.dailyLimit}条，本次${phoneNumbers.length}条超出剩余配额`;
+
    const submitGroupMessageId = `MSG-${randomUUID()}`;
    const submissions = phoneNumbers.map((phoneNumber, index) => ({
      phoneNumber,
@@ -2033,7 +2045,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
        ...data,
        phoneNumber: submission.phoneNumber,
        phoneNumbers: undefined,
-      }, submission.messageId, submitGroupMessageId))));
+      }, submission.messageId, submitGroupMessageId, dailyLimitFailure))));
    }
    const first = results[0];
    return {
@@ -2053,6 +2065,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    data: GatewayInboundSubmitDto & { phoneNumber: string },
    messageId: string,
    submitGroupMessageId: string,
+    dailyLimitFailure?: string,
  ) {
    const application = await this.findInboundApplication(data.account);
    if (!application) {
@@ -2190,7 +2203,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      });
      await this.enqueueBatchTask(task.id);
    };
-    if (application.status !== 'active' || application.tenant.status !== 'active') {
+    if (dailyLimitFailure) {
+      await reject('DAILY_LIMIT', dailyLimitFailure);
+    } else if (application.status !== 'active' || application.tenant.status !== 'active') {
      await reject('ACCOUNT', '企业或短信应用已停用');
    } else if (!application.interfaceEnabled) {
      await reject('INTERFACE', '短信应用 CMPP 接口已停用');
@@ -2908,6 +2923,58 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    }
  }

+  private async reserveDailySendQuota(applicationId: string, requestedCount: number) {
+    const result = await this.tryReserveDailySendQuota(applicationId, requestedCount);
+    if (!result.reserved) {
+      throw new HttpException({
+        code: 'DAILY_SEND_LIMIT_EXCEEDED',
+        message: `应用当日发送上限${result.dailyLimit}条，本次${requestedCount}条超出剩余配额`,
+        dailyLimit: result.dailyLimit,
+        requestedCount,
+      }, HttpStatus.TOO_MANY_REQUESTS);
+    }
+    return result;
+  }
+
+  private async tryReserveDailySendQuota(applicationId: string, requestedCount: number) {
+    if (!Number.isInteger(requestedCount) || requestedCount <= 0) {
+      throw new BadRequestException('发送号码数量必须为正整数');
+    }
+    const usageDate = shanghaiDateKey();
+    const reservationId = randomUUID();
+    const rows = await this.prisma.$queryRaw<Array<{ dailyLimit: number; usedCount: number | null }>>(Prisma.sql`
+      WITH application_limit AS (
+        SELECT id, COALESCE("dailyLimit", 100000)::integer AS "dailyLimit"
+        FROM "SmsApplication"
+        WHERE id = ${applicationId}
+      ), reservation AS (
+        INSERT INTO "SmsApplicationDailyUsage" (
+          id, "applicationId", "usageDate", "usedCount", "createdAt", "updatedAt"
+        )
+        SELECT ${reservationId}, id, ${usageDate}::date, ${requestedCount}, NOW(), NOW()
+        FROM application_limit
+        WHERE ${requestedCount} <= "dailyLimit"
+        ON CONFLICT ("applicationId", "usageDate") DO UPDATE
+          SET "usedCount" = "SmsApplicationDailyUsage"."usedCount" + EXCLUDED."usedCount",
+              "updatedAt" = NOW()
+          WHERE "SmsApplicationDailyUsage"."usedCount" + EXCLUDED."usedCount"
+                <= (SELECT "dailyLimit" FROM application_limit)
+        RETURNING "usedCount"
+      )
+      SELECT application_limit."dailyLimit", reservation."usedCount"
+      FROM application_limit
+      LEFT JOIN reservation ON TRUE
+    `);
+    if (rows.length === 0) {
+      throw new NotFoundException('短信应用不存在');
+    }
+    return {
+      dailyLimit: Number(rows[0].dailyLimit),
+      usedCount: rows[0].usedCount == null ? null : Number(rows[0].usedCount),
+      reserved: rows[0].usedCount != null,
+    };
+  }
+
  private async chargeAcceptedMessage(message: {
    tenantId: string;
    applicationId?: string | null;
@@ -3697,6 +3764,17 @@ function positiveInteger(value: string | undefined, fallback: number) {
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
 }

+function shanghaiDateKey(now = new Date()) {
+  const parts = new Intl.DateTimeFormat('en-CA', {
+    timeZone: 'Asia/Shanghai',
+    year: 'numeric',
+    month: '2-digit',
+    day: '2-digit',
+  }).formatToParts(now);
+  const values = Object.fromEntries(parts.map((part) => [part.type, part.value]));
+  return `${values.year}-${values.month}-${values.day}`;
+}
+
 function bullmqConnection() {
  const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
  return {