feat: aggregate cmpp template mismatch reviews

This commit is contained in:
hectorzhao
2026-07-12 13:25:19 +08:00
parent c8b6d0aa7c
commit 8b6ec92f4f
17 changed files with 523 additions and 38 deletions
@@ -0,0 +1,21 @@
ALTER TABLE "SmsSendTask"
ADD COLUMN "sourceType" TEXT NOT NULL DEFAULT 'risk',
ADD COLUMN "aggregationKey" TEXT,
ADD COLUMN "contentHash" TEXT,
ADD COLUMN "windowStartedAt" TIMESTAMP(3),
ADD COLUMN "windowEndsAt" TIMESTAMP(3);
ALTER TABLE "SmsMessageRecord"
ADD COLUMN "signatureId" TEXT,
ADD COLUMN "reviewTaskId" TEXT;
CREATE UNIQUE INDEX "SmsSendTask_aggregationKey_key" ON "SmsSendTask"("aggregationKey");
CREATE INDEX "SmsMessageRecord_reviewTaskId_status_idx" ON "SmsMessageRecord"("reviewTaskId", "status");
ALTER TABLE "SmsMessageRecord"
ADD CONSTRAINT "SmsMessageRecord_signatureId_fkey"
FOREIGN KEY ("signatureId") REFERENCES "SmsSignature"("id") ON DELETE SET NULL ON UPDATE CASCADE;
ALTER TABLE "SmsMessageRecord"
ADD CONSTRAINT "SmsMessageRecord_reviewTaskId_fkey"
FOREIGN KEY ("reviewTaskId") REFERENCES "SmsSendTask"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+12
View File
@@ -411,6 +411,7 @@ model SmsSignature {
templates SmsTemplate[]
reportMaterials SignatureReportMaterial[]
reportTasks ChannelSignatureReportTask[]
messageRecords SmsMessageRecord[]
@@index([tenantId, auditStatus])
@@index([tenantId, reportStatus])
@@ -782,6 +783,11 @@ model SmsSendTask {
applicationId String?
templateId String?
taskNo String @unique
sourceType String @default("risk")
aggregationKey String? @unique
contentHash String?
windowStartedAt DateTime?
windowEndsAt DateTime?
content String
category String?
phoneTotal Int
@@ -806,6 +812,7 @@ model SmsSendTask {
createdBy User? @relation("SmsSendTaskCreator", fields: [createdById], references: [id])
reviewedBy User? @relation("SmsSendTaskReviewer", fields: [reviewedById], references: [id])
riskHits RiskHitRecord[]
messageRecords SmsMessageRecord[]
@@index([tenantId, status, createdAt])
@@index([applicationId, createdAt])
@@ -899,6 +906,8 @@ model SmsMessageRecord {
batchTaskId String?
applicationId String?
templateId String?
signatureId String?
reviewTaskId String?
messageId String @unique
phoneNumber String
carrier String?
@@ -926,6 +935,8 @@ model SmsMessageRecord {
batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade)
application SmsApplication? @relation(fields: [applicationId], references: [id])
template SmsTemplate? @relation(fields: [templateId], references: [id])
signature SmsSignature? @relation(fields: [signatureId], references: [id])
reviewTask SmsSendTask? @relation(fields: [reviewTaskId], references: [id])
channel SmsChannel? @relation(fields: [channelId], references: [id])
submitRecords SmsSubmitRecord[]
receiptRecords SmsReceiptRecord[]
@@ -936,6 +947,7 @@ model SmsMessageRecord {
@@index([tenantId, status, queuedAt])
@@index([batchTaskId, status])
@@index([reviewTaskId, status])
@@index([phoneNumber])
@@index([gatewayMessageId])
}
@@ -166,6 +166,17 @@ function createPrismaMock() {
}
describe('OperationsService', () => {
it('keeps task progress limited to client-created batch tasks', async () => {
const prisma = createPrismaMock();
const service = new OperationsService(prisma as never);
await service.listBatchTasks({ tenantId: 'tenant-1', status: 'queued' });
expect(prisma.smsBatchTask.findMany).toHaveBeenCalledWith(expect.objectContaining({
where: { tenantId: 'tenant-1', status: 'queued', sourceType: 'client' },
}));
});
it('filters send-chain messages by tenant, application, channel, content, date, task, phone, and status', async () => {
const prisma = createPrismaMock();
const service = new OperationsService(prisma as never);
+1 -1
View File
@@ -78,7 +78,7 @@ export class OperationsService {
listBatchTasks(query: { tenantId?: string; status?: string }) {
return this.prisma.smsBatchTask.findMany({
where: { tenantId: query.tenantId, status: query.status },
where: { tenantId: query.tenantId, status: query.status, sourceType: 'client' },
include: { apiRequests: true },
orderBy: { createdAt: 'desc' },
});
@@ -1,5 +1,6 @@
import { Body, Controller, Get, Param, Post, Query } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { SendChainService } from '../send-chain/send-chain.service';
import {
CreateRiskRuleDto,
ReviewSmsTaskDto,
@@ -9,7 +10,7 @@ import {
@ApiTags('risk-review')
@Controller('admin/risk-review')
export class AdminRiskReviewController {
constructor(private readonly riskReview: RiskReviewService) {}
constructor(private readonly riskReview: RiskReviewService, private readonly sendChain: SendChainService) {}
@Get('rules')
listRules(@Query('tenantId') tenantId?: string) {
@@ -37,13 +38,16 @@ export class AdminRiskReviewController {
}
@Post('tasks/:id/approve')
approveTask(@Param('id') taskId: string, @Body() body: ReviewSmsTaskDto) {
return this.riskReview.approveTask(taskId, body);
async approveTask(@Param('id') taskId: string, @Body() body: ReviewSmsTaskDto) {
const task = await this.riskReview.approveTask(taskId, body);
await this.sendChain.handleReviewDecision(taskId, 'approved', body.reason ?? '运营审核通过');
return task;
}
@Post('tasks/:id/reject')
rejectTask(@Param('id') taskId: string, @Body() body: ReviewSmsTaskDto) {
return this.riskReview.rejectTask(taskId, body);
async rejectTask(@Param('id') taskId: string, @Body() body: ReviewSmsTaskDto) {
const task = await this.riskReview.rejectTask(taskId, body);
await this.sendChain.handleReviewDecision(taskId, 'rejected', body.reason ?? task.rejectReason ?? '运营审核驳回');
return task;
}
}
+3 -3
View File
@@ -1,14 +1,14 @@
import { Module } from '@nestjs/common';
import { forwardRef, Module } from '@nestjs/common';
import { PrismaModule } from '../prisma/prisma.module';
import { AdminRiskReviewController } from './admin-risk-review.controller';
import { ClientRiskReviewController } from './client-risk-review.controller';
import { RiskReviewService } from './risk-review.service';
import { SendChainModule } from '../send-chain/send-chain.module';
@Module({
imports: [PrismaModule],
imports: [PrismaModule, forwardRef(() => SendChainModule)],
controllers: [AdminRiskReviewController, ClientRiskReviewController],
providers: [RiskReviewService],
exports: [RiskReviewService],
})
export class RiskReviewModule {}
@@ -33,16 +33,65 @@ function createPrismaMock(overrides: Record<string, unknown> = {}) {
findUnique: jest.fn().mockResolvedValue({ id: 'risk-task-1', riskHits: [] }),
update: jest.fn(),
findMany: jest.fn(),
upsert: jest.fn().mockImplementation(({ create }: { create: Record<string, unknown> }) => Promise.resolve({ id: 'review-task-1', ...create })),
},
smsMessageRecord: {
update: jest.fn().mockResolvedValue({ id: 'message-1' }),
},
riskHitRecord: {
createMany: jest.fn(),
findMany: jest.fn(),
},
$transaction: jest.fn(async (callback) => callback({
smsSendTask: {
upsert: jest.fn().mockImplementation(({ create }: { create: Record<string, unknown> }) => Promise.resolve({ id: 'review-task-1', ...create })),
},
smsMessageRecord: {
update: jest.fn().mockResolvedValue({ id: 'message-1' }),
},
})),
...overrides,
};
}
describe('RiskReviewService', () => {
it('groups identical CMPP template mismatches into a deterministic short review window', async () => {
const prisma = createPrismaMock();
prisma.smsSendTask.findUnique.mockResolvedValue({
id: 'review-task-1',
sourceType: 'cmpp_template_mismatch',
phoneTotal: 2,
_count: { messageRecords: 2 },
riskHits: [],
});
const service = new RiskReviewService(prisma as never);
const result = await service.aggregateTemplateMismatch({
tenantId: 'tenant-1',
applicationId: 'app-1',
account: '100001',
messageRecordId: 'message-1',
signatureId: 'sig-1',
content: '【签名】同一审核内容',
});
expect(prisma.$transaction).toHaveBeenCalledTimes(1);
expect(result).toEqual(expect.objectContaining({ sourceType: 'cmpp_template_mismatch', phoneTotal: 2 }));
});
it('does not allow an aggregated review task to be decided before its window closes', async () => {
const prisma = createPrismaMock();
prisma.smsSendTask.findUnique.mockResolvedValue({
id: 'review-task-1',
sourceType: 'cmpp_template_mismatch',
windowEndsAt: new Date(Date.now() + 10_000),
});
const service = new RiskReviewService(prisma as never);
await expect(service.approveTask('review-task-1', { reason: '通过' })).rejects.toThrow('聚合窗口尚未关闭');
expect(prisma.smsSendTask.update).not.toHaveBeenCalled();
});
it('rejects tasks over the application max phone threshold', async () => {
const prisma = createPrismaMock();
prisma.smsApplication.findUnique.mockResolvedValue({ id: 'app-1', maxPhonesPerTask: 2 });
+80 -2
View File
@@ -1,6 +1,6 @@
import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { randomUUID } from 'node:crypto';
import { createHash, randomUUID } from 'node:crypto';
import { PrismaService } from '../prisma/prisma.service';
export interface CreateRiskRuleDto {
@@ -33,6 +33,15 @@ export interface ReviewSmsTaskDto {
reason?: string;
}
export interface AggregateTemplateMismatchDto {
tenantId: string;
applicationId: string;
account: string;
messageRecordId: string;
signatureId: string;
content: string;
}
interface RuleEvaluation {
ruleId?: string;
ruleCode: string;
@@ -153,8 +162,14 @@ export class RiskReviewService {
where: {
tenantId,
status,
...(status === 'pending_review' ? {
OR: [
{ sourceType: { not: 'cmpp_template_mismatch' } },
{ windowEndsAt: { lte: new Date() } },
],
} : {}),
},
include: { riskHits: true },
include: { riskHits: true, _count: { select: { messageRecords: true } } },
orderBy: { createdAt: 'desc' },
});
}
@@ -163,6 +178,61 @@ export class RiskReviewService {
return this.listTasks(undefined, 'pending_review');
}
async aggregateTemplateMismatch(data: AggregateTemplateMismatchDto) {
const normalizedContent = data.content.replace(/\r\n/g, '\n').trim();
const contentHash = createHash('sha256').update(normalizedContent, 'utf8').digest('hex');
const windowMs = Math.max(1_000, Number(process.env.CMPP_TEMPLATE_REVIEW_WINDOW_MS ?? 10_000));
const now = new Date();
const windowStartedAt = new Date(Math.floor(now.getTime() / windowMs) * windowMs);
const windowEndsAt = new Date(windowStartedAt.getTime() + windowMs);
const aggregationKey = createHash('sha256')
.update(`${data.applicationId}|${data.account}|${contentHash}|${windowStartedAt.toISOString()}`, 'utf8')
.digest('hex');
const task = await this.prisma.$transaction(async (tx) => {
const aggregated = await tx.smsSendTask.upsert({
where: { aggregationKey },
update: {
phoneTotal: { increment: 1 },
uniquePhoneTotal: { increment: 1 },
},
create: {
tenantId: data.tenantId,
applicationId: data.applicationId,
taskNo: `CMPP-REVIEW-${windowStartedAt.getTime()}-${aggregationKey.slice(0, 8)}`,
sourceType: 'cmpp_template_mismatch',
aggregationKey,
contentHash,
windowStartedAt,
windowEndsAt,
content: normalizedContent,
phoneTotal: 1,
uniquePhoneTotal: 1,
status: 'pending_review',
riskDecision: 'manual_review',
reviewReason: '企业应用已配置模板不匹配进入人工审核',
variableIssues: {
sourceType: 'cmpp',
account: data.account,
aggregationWindowMs: windowMs,
} as Prisma.InputJsonValue,
},
});
await tx.smsMessageRecord.update({
where: { id: data.messageRecordId },
data: {
reviewTaskId: aggregated.id,
signatureId: data.signatureId,
status: 'pending_review',
},
});
return aggregated;
});
return this.prisma.smsSendTask.findUnique({
where: { id: task.id },
include: { riskHits: true, _count: { select: { messageRecords: true } } },
});
}
async evaluateTask(data: EvaluateSmsTaskDto) {
await this.ensureDefaultRules();
if (data.createdById) {
@@ -261,6 +331,7 @@ export class RiskReviewService {
if (!task) {
throw new NotFoundException('SMS send task not found');
}
this.assertAggregationWindowClosed(task);
return this.prisma.smsSendTask.update({
where: { id: taskId },
data: {
@@ -280,6 +351,7 @@ export class RiskReviewService {
if (!task) {
throw new NotFoundException('SMS send task not found');
}
this.assertAggregationWindowClosed(task);
const reason = data.reason ?? task.reviewReason ?? '审核拒绝';
return this.prisma.smsSendTask.update({
where: { id: taskId },
@@ -306,6 +378,12 @@ export class RiskReviewService {
}
}
private assertAggregationWindowClosed(task: { sourceType?: string | null; windowEndsAt?: Date | null }) {
if (task.sourceType === 'cmpp_template_mismatch' && task.windowEndsAt && task.windowEndsAt.getTime() > Date.now()) {
throw new BadRequestException('聚合窗口尚未关闭,请在窗口结束后审核');
}
}
private async effectiveRules(tenantId: string) {
const rules = await this.prisma.riskRule.findMany({
where: {
@@ -1,4 +1,4 @@
import { Body, Controller, Get, Param, Post, Query } from '@nestjs/common';
import { BadRequestException, Body, Controller, Get, Param, Post, Query } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { TenantId } from '../common/tenant-id.decorator';
import { ConfirmImportDto, CreateBatchTaskDto, ImportPreviewDto, SendChainService } from './send-chain.service';
@@ -25,21 +25,28 @@ export class ClientSendChainController {
@Get('batch-tasks')
listBatchTasks(@TenantId() tenantId?: string, @Query('status') status?: string) {
return this.sendChain.listBatchTasks(tenantId, status);
return this.sendChain.listBatchTasks(requireTenantId(tenantId), status, 'client');
}
@Get('batch-tasks/:id')
getBatchTask(@Param('id') taskId: string) {
return this.sendChain.getBatchTask(taskId);
getBatchTask(@TenantId() tenantId: string | undefined, @Param('id') taskId: string) {
return this.sendChain.getBatchTask(taskId, requireTenantId(tenantId), 'client');
}
@Get('batch-tasks/:id/messages')
listTaskMessages(@Param('id') taskId: string) {
return this.sendChain.listMessages({ taskId });
listTaskMessages(@TenantId() tenantId: string | undefined, @Param('id') taskId: string) {
return this.sendChain.listClientTaskMessages(taskId, requireTenantId(tenantId));
}
@Post('batch-tasks/:id/cancel')
cancelBatchTask(@Param('id') taskId: string) {
return this.sendChain.cancelBatchTask(taskId);
cancelBatchTask(@TenantId() tenantId: string | undefined, @Param('id') taskId: string) {
return this.sendChain.cancelBatchTask(taskId, requireTenantId(tenantId), 'client');
}
}
function requireTenantId(tenantId?: string) {
if (!tenantId) {
throw new BadRequestException('Tenant context is required');
}
return tenantId;
}
+2 -2
View File
@@ -1,4 +1,4 @@
import { Module } from '@nestjs/common';
import { forwardRef, Module } from '@nestjs/common';
import { BillingModule } from '../billing/billing.module';
import { PrismaModule } from '../prisma/prisma.module';
import { RiskReviewModule } from '../risk-review/risk-review.module';
@@ -9,7 +9,7 @@ import { GatewayEventsController } from './gateway-events.controller';
import { SendChainService } from './send-chain.service';
@Module({
imports: [PrismaModule, BillingModule, RiskReviewModule, SmsConfigModule],
imports: [PrismaModule, BillingModule, forwardRef(() => RiskReviewModule), SmsConfigModule],
controllers: [AdminSendChainController, ClientSendChainController, GatewayEventsController],
providers: [SendChainService],
exports: [SendChainService],
+105 -2
View File
@@ -95,9 +95,16 @@ function createPrismaMock() {
signature: { auditStatus: 'approved', reportStatus: 'approved' },
}),
},
smsSignature: {
findFirst: jest.fn().mockResolvedValue({ id: 'sig-1', name: '签名', auditStatus: 'approved', reportStatus: 'approved' }),
},
smsSendTask: {
findUnique: jest.fn().mockResolvedValue(null),
},
smsBatchTask: {
create: jest.fn().mockResolvedValue(task),
findUnique: jest.fn().mockResolvedValue(task),
findFirst: jest.fn().mockResolvedValue(task),
findMany: jest.fn(),
update: jest.fn().mockResolvedValue(task),
},
@@ -296,6 +303,10 @@ function createService(prisma = createPrismaMock()) {
reason: null,
task: { id: 'risk-task-1' },
}),
aggregateTemplateMismatch: jest.fn().mockResolvedValue({
id: 'review-task-1',
reviewReason: '企业应用已配置模板不匹配进入人工审核',
}),
} as unknown as RiskReviewService;
const service = new SendChainService(prisma as never, billing, riskReview);
service['postGatewayControl'] = jest.fn().mockResolvedValue({ delivered: true });
@@ -372,7 +383,7 @@ describe('SendChainService', () => {
it('cancels scheduled tasks before dispatch', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findUnique.mockResolvedValue({ id: 'task-1', status: 'scheduled' });
prisma.smsBatchTask.findFirst.mockResolvedValue({ id: 'task-1', status: 'scheduled' });
await service.cancelBatchTask('task-1');
@@ -386,6 +397,30 @@ describe('SendChainService', () => {
});
});
it('lists only client-created batch tasks for task progress', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findMany.mockResolvedValue([{ id: 'task-client', tenantId: 'tenant-1', sourceType: 'client' }]);
prisma.smsMessageRecord.groupBy.mockResolvedValue([]);
await service.listBatchTasks('tenant-1', 'queued');
expect(prisma.smsBatchTask.findMany).toHaveBeenCalledWith(expect.objectContaining({
where: { tenantId: 'tenant-1', status: 'queued', sourceType: 'client' },
}));
});
it('does not expose CMPP internal tasks through client task detail or messages', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findFirst.mockResolvedValue(null);
await expect(service.getBatchTask('task-cmpp', 'tenant-1', 'client')).rejects.toThrow('SMS batch task not found');
await expect(service.listClientTaskMessages('task-cmpp', 'tenant-1')).rejects.toThrow('SMS batch task not found');
expect(prisma.smsBatchTask.findFirst).toHaveBeenCalledWith(expect.objectContaining({
where: { id: 'task-cmpp', tenantId: 'tenant-1', sourceType: 'client' },
}));
expect(prisma.smsMessageRecord.findMany).not.toHaveBeenCalled();
});
it('terminates non-final tasks by canceling unsubmitted messages', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findUnique.mockResolvedValue({ id: 'task-1', status: 'sending' });
@@ -507,7 +542,7 @@ describe('SendChainService', () => {
});
it('records an unreported CMPP message and returns success before delivering the template failure receipt', async () => {
const { service, prisma } = createService();
const { service, prisma, riskReview } = createService();
prisma.smsTemplate.findFirst.mockResolvedValue(null);
await expect(service.submitInboundMessage({
@@ -524,6 +559,74 @@ describe('SendChainService', () => {
'/downstream/receipt',
expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }),
);
expect(riskReview.aggregateTemplateMismatch).not.toHaveBeenCalled();
});
it('aggregates template-mismatched CMPP messages only when the application uses manual review', async () => {
const { service, prisma, riskReview } = createService();
prisma.smsApplication.findFirst.mockResolvedValue({
id: 'app-1',
tenantId: 'tenant-1',
cmppAccount: '100001',
status: 'active',
interfaceEnabled: true,
templateMismatchMode: 'manual_review',
customerUnitPrice: 3,
queuePriority: 'normal',
ipAllowlist: [{ ipCidr: '127.0.0.1/32' }],
tenant: { id: 'tenant-1', status: 'active', certificationStatus: 'approved' },
});
prisma.smsTemplate.findFirst.mockResolvedValue(null);
await expect(service.submitInboundMessage({
account: '100001',
phoneNumber: '13800000001',
content: '【签名】未匹配模板的内容',
remoteIp: '127.0.0.1',
})).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' }));
expect(riskReview.aggregateTemplateMismatch).toHaveBeenCalledWith(expect.objectContaining({
applicationId: 'app-1',
account: '100001',
messageRecordId: 'record-1',
signatureId: 'sig-1',
}));
expect(prisma.smsBatchTask.update).toHaveBeenCalledWith({
where: { id: 'task-1' },
data: expect.objectContaining({ status: 'pending_review', riskTaskId: 'review-task-1', auditStatus: 'pending' }),
});
expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled();
});
it('fans an approved aggregated review task back into each internal CMPP batch', async () => {
const { service, prisma } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
prisma.smsSendTask.findUnique.mockResolvedValue({
id: 'review-task-1',
messageRecords: [{
id: 'record-1',
tenantId: 'tenant-1',
applicationId: 'app-1',
batchTaskId: 'task-1',
messageId: 'MSG-1',
phoneNumber: '13800000001',
amountCents: 3,
billingUnits: 1,
batchTask: { id: 'task-1', sourceType: 'cmpp' },
}],
});
await expect(service.handleReviewDecision('review-task-1', 'approved', '审核通过')).resolves.toEqual({
reviewTaskId: 'review-task-1',
decision: 'approved',
affected: 1,
});
expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({
where: { id: 'record-1' },
data: { status: 'queued', errorCode: null, errorMessage: null },
});
expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
});
it('previews imported phone files with duplicate, invalid, blacklist, and variable errors', async () => {
+141 -10
View File
@@ -331,9 +331,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return this.getBatchTask(task.id);
}
async listBatchTasks(tenantId?: string, status?: string) {
async listBatchTasks(tenantId?: string, status?: string, sourceType = 'client') {
const tasks = await this.prisma.smsBatchTask.findMany({
where: { tenantId, status },
where: { tenantId, status, sourceType },
include: {
tenant: true,
application: true,
@@ -355,11 +355,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}));
}
getBatchTask(taskId: string) {
return this.prisma.smsBatchTask.findUnique({
where: { id: taskId },
async getBatchTask(taskId: string, tenantId?: string, sourceType = 'client') {
const task = await this.prisma.smsBatchTask.findFirst({
where: { id: taskId, tenantId, sourceType },
include: { apiRequests: true, messages: { take: 20, orderBy: { queuedAt: 'asc' } } },
});
if (!task) {
throw new NotFoundException('SMS batch task not found');
}
return task;
}
async listClientTaskMessages(taskId: string, tenantId: string) {
await this.getBatchTask(taskId, tenantId, 'client');
return this.listMessages({ tenantId, taskId });
}
listMessages(query: {
@@ -506,8 +515,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return { taskId, enqueued: messages.length };
}
async cancelBatchTask(taskId: string) {
const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
async cancelBatchTask(taskId: string, tenantId?: string, sourceType = 'client') {
const task = await this.prisma.smsBatchTask.findFirst({ where: { id: taskId, tenantId, sourceType } });
if (!task) {
throw new NotFoundException('SMS batch task not found');
}
@@ -524,6 +533,50 @@ 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) {
return { reviewTaskId, decision, affected: 0 };
}
const batchTaskIds = new Set<string>();
for (const message of reviewTask.messageRecords) {
if (!message.tenantId || !message.applicationId || !message.batchTaskId) continue;
if (decision === 'approved') {
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { status: 'queued', errorCode: null, errorMessage: null },
});
await this.prisma.smsBatchTask.update({
where: { id: message.batchTaskId },
data: { status: 'ready', auditStatus: 'approved', reviewReason: reason, rejectReason: null },
});
batchTaskIds.add(message.batchTaskId);
} else {
await this.releaseMessageReservation(
message as typeof message & { tenantId: string; batchTaskId: string },
'模板不匹配人工审核驳回释放冻结',
);
await this.prisma.smsBatchTask.update({
where: { id: message.batchTaskId },
data: { status: 'rejected', auditStatus: 'rejected', rejectReason: reason },
});
await this.recordCmppFailureReceipt(message, 'REVIEW_REJECTED', reason);
}
}
for (const batchTaskId of batchTaskIds) {
await this.enqueueBatchTask(batchTaskId);
}
return { reviewTaskId, decision, affected: reviewTask.messageRecords.length };
}
async terminateBatchTask(taskId: string) {
const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
if (!task) {
@@ -611,7 +664,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
async processSendJob(job: SendJob) {
const message = await this.prisma.smsMessageRecord.findUnique({
where: { id: job.messageRecordId },
include: { batchTask: true, template: { include: { signature: true } } },
include: { batchTask: true, template: { include: { signature: true } }, signature: true },
});
if (!message || message.status !== 'queued') {
return { skipped: true };
@@ -1513,6 +1566,60 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
await reject('INTERFACE', '短信应用 CMPP 接口已停用');
} else if (application.tenant.certificationStatus !== 'approved') {
await reject('CERT', '企业认证未通过');
} else if (!template && application.templateMismatchMode === 'manual_review') {
const signature = await this.resolveInboundSignatureCandidate(application.id, data.content);
if (!signature) {
await reject('SIGNATURE', '短信内容未识别到已审核且已报备的签名');
} else {
const risk = await this.riskReview.evaluateTask({
tenantId: application.tenantId,
applicationId: application.id,
content: data.content,
phones: [data.phoneNumber],
});
if (risk.status === 'rejected') {
await reject('RISK', risk.reason || '短信被风控拒绝');
} else {
const accountCheck = await this.billing.checkAccount({
tenantId: application.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
});
if (!accountCheck.canSend) {
await reject('BALANCE', '企业账户余额、套餐余量或授信额度不足');
} else {
if (billing.amountCents + billing.totalBillingUnits > 0) {
await this.billing.freeze({
tenantId: application.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: 'CMPP 模板不匹配待审核短信冻结',
});
}
const reviewTask = risk.status === 'pending_review' && risk.task
? await this.attachMessageToReviewTask(risk.task.id, message.id, signature.id)
: await this.riskReview.aggregateTemplateMismatch({
tenantId: application.tenantId,
applicationId: application.id,
account: data.account,
messageRecordId: message.id,
signatureId: signature.id,
content: data.content,
});
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: {
status: 'pending_review',
riskTaskId: reviewTask?.id,
auditStatus: 'pending',
reviewReason: reviewTask?.reviewReason ?? '模板不匹配,等待人工审核',
},
});
}
}
}
} else if (!template) {
await reject('TEMPLATE', '短信内容未匹配到已报备模板');
} else if (template.auditStatus !== 'approved') {
@@ -1616,6 +1723,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
billingUnits: number;
queuePriority?: string | null;
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
signature?: { id?: string | null; name?: string | null } | null;
},
routed: RoutedChannel,
attempt: number,
@@ -1668,7 +1776,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
queuePriority: normalizeQueuePriority(message.queuePriority),
phoneNumber: message.phoneNumber,
content: message.content,
signature: message.template?.signature?.name ?? 'SMS',
signature: message.template?.signature?.name ?? message.signature?.name ?? 'SMS',
templateId: message.templateId ?? 'unknown',
billingUnits: message.billingUnits,
route: {
@@ -1897,6 +2005,28 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
private resolveInboundSignatureCandidate(applicationId: string, content: string) {
const match = content.match(/^【([^】]+)】/);
if (!match?.[1]) return null;
return this.prisma.smsSignature.findFirst({
where: {
applicationId,
name: match[1],
auditStatus: 'approved',
reportStatus: 'approved',
},
orderBy: { updatedAt: 'desc' },
});
}
private async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, signatureId: string) {
await this.prisma.smsMessageRecord.update({
where: { id: messageRecordId },
data: { reviewTaskId, signatureId, status: 'pending_review' },
});
return this.prisma.smsSendTask.findUnique({ where: { id: reviewTaskId } });
}
private async recordCmppFailureReceipt(
message: {
id: string;
@@ -2102,10 +2232,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
id: string;
templateId?: string | null;
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
signature?: { id?: string | null; name?: string | null } | null;
},
channelId: string,
) {
let signatureId = message.template?.signature?.id ?? null;
let signatureId = message.template?.signature?.id ?? message.signature?.id ?? null;
if (!signatureId && message.templateId) {
const template = await this.prisma.smsTemplate.findUnique({
where: { id: message.templateId },
+10 -3
View File
@@ -139,11 +139,18 @@
7. 若命中审核策略,运营端短信审核通过后进入发送队列,审核页面必须展示进入审核的原因。
8. 发送服务按通道组路由、通道限速、企业限速执行提交。
9. 平台批量任务只记录客户端创建的发送任务;API 调用和 CMPP 对接发送不进入批量任务。
- 运营端和客户端“短信任务进度”只查询 `SmsBatchTask.sourceType=client`
- CMPP/API/通道测试可使用内部批次承载风控、计费、队列、重试和回执关联,但不得出现在客户批量任务列表。
- 客户端批量任务列表、详情、短信明细和取消操作必须同时校验当前企业和 `sourceType=client`
10. 所有来源的短信,包括平台批量任务、API 调用、CMPP 对接发送,全部按手机号维度进入短信记录。
11. 任务进度、发送详情和短信记录实时或准实时更新。
12. 发送入队必须按短信应用的队列等级分流到优先队列或普通队列;同等条件下优先队列消息必须先于普通队列消息被 Send Worker 消费并提交 Gateway
13. 优先队列只能改变待发送消息的调度顺序,不得绕过企业/应用状态、签名模板审核、通道报备、余额/授信、黑名单、风控、通道组路由、通道限速和 Gateway 连接可用性校验
14. 同一队列内部按创建时间、任务顺序和手机号拆分顺序保持 FIFO 或可解释的稳定排序;优先队列插队时必须可在 trace 或任务日志中追踪队列等级和入队时间
12. 企业应用“不符合模板的短信”配置为 `manual_review` 时,合法的 CMPP Submit 在模板不匹配后进入人工审核;配置为 `reject` 时仍直接拒绝并返回 `REJECTD` Deliver Receipt,其他模式不得被人工审核聚合逻辑误接管
13. CMPP 模板不匹配审核支持短窗口内容指纹聚合:只有同一企业应用、同一 CMPP 账号、规范化后内容 SHA-256 完全一致且位于同一时间窗口的短信才能合并为一个审核任务。默认窗口 10 秒,可通过 `CMPP_TEMPLATE_REVIEW_WINDOW_MS` 调整
14. 聚合审核不合并短信记录、计费或回执:每个手机号仍有独立 `SmsMessageRecord/messageId/sequenceId`。审核通过后逐条进入真实路由和上游提交;审核驳回后逐条释放冻结并产生客户侧 `REJECTD` 回执
15. 人工审核只覆盖模板不匹配;签名必须能从短信前缀识别且已审核/报备通过。签名不合法、风控直接拒绝或余额不足不得因内容聚合而绕过。
16. 发送入队必须按短信应用的队列等级分流到优先队列或普通队列;同等条件下优先队列消息必须先于普通队列消息被 Send Worker 消费并提交 Gateway。
17. 优先队列只能改变待发送消息的调度顺序,不得绕过企业/应用状态、签名模板审核、通道报备、余额/授信、黑名单、风控、通道组路由、通道限速和 Gateway 连接可用性校验。
18. 同一队列内部按创建时间、任务顺序和手机号拆分顺序保持 FIFO 或可解释的稳定排序;优先队列插队时必须可在 trace 或任务日志中追踪队列等级和入队时间。
### 4.6 通道配置与路由
+31
View File
@@ -1008,6 +1008,7 @@
- CMPP2.0 和 CMPP3.0 连接分别返回对应版本 ConnectResp,后续 Submit/Deliver 按该 TCP 连接协商版本解包和组包,不发生字段错位。
- 密码错误、应用停用、企业停用、短信接口关闭、IP 不在白名单时 connect/login 被拒绝。
- submit 被接受后返回 CMPP SubmitResp 成功,并在真实数据库创建 `sourceType=cmpp` 的发送记录,进入真实发送链路。
- `sourceType=cmpp` 的内部批次不出现在运营端或客户端“短信任务进度”;客户端不能通过任务 ID 读取该内部批次的详情、短信明细或执行取消。
- Submit 应用身份使用 bind 已鉴权账号;`MsgSrc` 使用应用级企业代码并独立校验。企业代码与登录账号不同时仍能正确定位应用,企业代码不匹配时返回失败。
- 鉴权失败、IP 白名单不符、手机号等协议参数不合法时返回非零 SubmitResp,且不创建短信记录。
- 已鉴权且参数合法的 Submit 必须先返回成功 SubmitResp 和平台 Msg_Id;内容不匹配审核模板、签名/报备未通过、余额不足、应用在 bind 后停用、无可用通道、上游 Submit 最终失败时,均须真实创建短信记录、`SmsReceiptRecord``CmppDownstreamDelivery`,并向客户下发 `undelivered/REJECTD` Deliver Receipt,不得仅以 SubmitResp 失败替代回执。
@@ -1015,6 +1016,36 @@
- CMPP 包在进入 handler 前因长度、命令字、读包或 Unpack 失败时,Gateway 记录 `read/unpack packet failed`、远端地址、协议模式、错误类型和原始错误,不得静默断开。
- 企业应用列表和连接详情展示真实下游 CMPP 会话:bind 后当前连接数加一,显示客户 IP、企业代码、CMPP 版本、连接建立时间与最后心跳;连接持续未响应 `ACTIVE_TEST` 超过阈值后转为心跳超时/断开,不能继续显示为正常连接。
### TC-SEND-038 批量任务与 CMPP 内部批次隔离
- 优先级:P0
- 前置条件:同一企业已存在一个客户端批量发送任务,并通过下游 CMPP 提交一条短信。
- 步骤:
1. 查询运营端短信任务进度。
2. 查询该企业的客户端批量任务列表。
3. 使用 CMPP 内部批次 ID 请求客户端任务详情、短信明细和取消接口。
4. 查询运营端短信记录。
- 预期结果:
- 两个任务进度列表只返回 `sourceType=client` 的客户端批量任务。
- CMPP 内部批次的详情、短信明细和取消请求均返回不可见/不存在,不泄露内部任务。
- CMPP 短信仍完整出现在短信记录、提交、回执和账务链路。
### TC-SEND-039 CMPP 模板不匹配短窗口聚合人工审核
- 优先级:P0
- 前置条件:应用 A 配置 `templateMismatchMode=manual_review`,应用 B 配置 `reject`;两个应用均已配置审核和报备通过的签名、余额和通道组。
- 步骤:
1. 在 10 秒内用应用 A 的同一 CMPP 账号向不同手机号提交多条规范化后内容完全一致、签名合法但不匹配模板的短信。
2. 用应用 A 提交内容不同、账号不同或跨越聚合窗口的短信。
3. 用应用 B 提交同样的模板不匹配短信。
4. 分别审核通过和驳回应用 A 的聚合任务。
- 预期结果:
- 应用 A 同账号、同内容指纹、同窗口的短信只创建一个 `sourceType=cmpp_template_mismatch` 审核任务,审核页展示真实聚合号码数。
- 不同应用、账号、内容指纹或窗口的短信不合并。
- 应用 B 不进入人工审核,继续逐条产生 `REJECTD` Deliver Receipt。
- 审核通过后每条成员独立进入路由、提交和计费;审核驳回后每条成员独立释放冻结并向客户下发 `REJECTD`
- 签名无法识别/未报备、风控直接拒绝或余额不足时不进入聚合审核。
### TC-GW-007 CMPP 客户到上游 SMSC 完整闭环
- 优先级:P0
+17
View File
@@ -1600,3 +1600,20 @@ git diff --check
- 保留的固定条数均属于明确的近期日志/监控窗口、超时扫描单批、导出保护、单消息重试尝试或唯一候选判定,这些响应不被页面展示为业务总数。
- 已执行 API 全量测试(13 suites、122 项通过),及受影响服务定向测试(5 suites、91 项通过)、API build、前端 build 与 `git diff --check`;前端仅有既有 chunk size warning。
- 已将 `390b9700` 部署生产,部署前完成 PostgreSQL 和当前发布源码备份,Prisma 确认 27 条 migration 均已应用。`cmpp-api``cmpp-gateway`、Nginx、MinIO 均为 active`12026/17890/8090/3000` 监听、API/Gateway health 和外部 `12026` HTTP 均通过。使用生产平台管理员会话调用真实受保护 API:企业管理返回 105 条未删除企业,企业应用返回 204 条,不再封顶 100。
## 2026-07-12 客户批量任务与 CMPP 内部批次隔离
- 短信任务进度的产品口径回归为“客户在客户端提交的批量发送任务”。运营端和客户端任务列表统一强制 `SmsBatchTask.sourceType=client`,不再展示每条 CMPP Submit 创建的 `sourceType=cmpp` 内部批次或通道测试批次。
- CMPP 内部批次仍保留在 PostgreSQL,继续承载模板/签名/报备校验、风控、冻结/计费、队列、补发和 Deliver Receipt 关联;所有 CMPP 号码仍进入短信记录。
- 客户端任务详情、任务短信明细和取消接口同时校验当前 `tenantId``sourceType=client`,不能通过内部批次 ID 读取或操作 CMPP 内部任务。
- 生产现状只读核对:`sourceType=cmpp` 2 个、`sourceType=admin_channel_test` 1 个、`sourceType=client` 0 个。部署后任务进度应显示 0 个客户批量任务,但不删除现有内部批次数据。
- 已执行 `send-chain.service.spec.ts + operations.service.spec.ts`2 suites、50 项通过)、API 全量测试(13 suites、125 项通过)、API build 和前端 build;前端仅有既有 chunk size warning。按用户要求本轮暂不部署生产。
## 2026-07-12 CMPP 模板不匹配短窗口聚合审核
- 只有企业应用配置 `templateMismatchMode=manual_review` 时,CMPP 模板不匹配短信才进入人工审核。`reject` 继续逐条失败并下发 `REJECTD`;其他模式不被聚合逻辑接管。
- 新增 `SmsSendTask.sourceType/aggregationKey/contentHash/windowStartedAt/windowEndsAt`,以应用、CMPP 账号、规范化内容 SHA-256 和默认 10 秒窗口生成唯一聚合键。`SmsMessageRecord.reviewTaskId/signatureId` 保留每条成员与审核任务、真实签名的关联。
- 聚合前仍校验签名审核/报备、风控直接拒绝和账户余额,并按每条短信独立冻结。审核通过后逐条恢复到各自 `sourceType=cmpp` 内部批次并入队;驳回后逐条释放冻结、写入失败记录并生成客户侧 Deliver Receipt。
- 运营端短信审核页新增“审核来源”和“聚合号码数”,区分 CMPP 模板不匹配聚合与普通风控审核。
- 聚合窗口关闭前,任务不返回到待审列表且审核接口拒绝提前操作,避免窗口内后到短信加入已完成任务。
- Prisma migration`20260712113000_add_cmpp_review_aggregation`。已执行 API 全量测试、API build、前端 build、Prisma validate 和 `git diff --check`;本轮按用户要求暂不部署生产。
+9
View File
@@ -305,6 +305,10 @@ export type SmsBatchTask = {
id: string;
tenantId: string;
taskNo: string;
sourceType?: string;
contentHash?: string | null;
windowStartedAt?: string | null;
windowEndsAt?: string | null;
applicationId?: string | null;
templateId?: string | null;
content: string;
@@ -559,6 +563,10 @@ export type RiskReviewTask = {
applicationId?: string | null;
templateId?: string | null;
taskNo: string;
sourceType?: string;
contentHash?: string | null;
windowStartedAt?: string | null;
windowEndsAt?: string | null;
content: string;
category?: string | null;
phoneTotal: number;
@@ -574,6 +582,7 @@ export type RiskReviewTask = {
createdAt: string;
reviewedAt?: string | null;
riskHits?: Array<{ id: string; ruleName: string; reason: string }>;
_count?: { messageRecords: number };
};
export type TenantAccount = {
+6 -1
View File
@@ -15,6 +15,10 @@ const statusTone: Record<string, 'warning' | 'success' | 'danger'> = {
rejected: 'danger',
};
function sourceLabel(sourceType?: string) {
return sourceType === 'cmpp_template_mismatch' ? 'CMPP模板不匹配聚合' : '风控审核';
}
export function AdminSmsAuditPage() {
const [records, setRecords] = useState<RiskReviewTask[]>([]);
const [keyword, setKeyword] = useState('');
@@ -69,8 +73,9 @@ export function AdminSmsAuditPage() {
const columns: Array<TableColumn<RiskReviewTask>> = [
{ key: 'taskNo', title: '任务编号', width: '180px', render: (record) => <strong>{record.taskNo}</strong> },
{ key: 'sourceType', title: '审核来源', width: '180px', render: (record) => <Tag tone={record.sourceType === 'cmpp_template_mismatch' ? 'warning' : 'info'}>{sourceLabel(record.sourceType)}</Tag> },
{ key: 'content', title: '短信内容', render: (record) => <span className="table-long-text">{record.content}</span> },
{ key: 'phoneTotal', title: '号码数', width: '130px', render: (record) => record.phoneTotal.toLocaleString('zh-CN') },
{ key: 'phoneTotal', title: '聚合号码数', width: '140px', render: (record) => (record._count?.messageRecords ?? record.phoneTotal).toLocaleString('zh-CN') },
{ key: 'createdAt', title: '提交时间', width: '190px', render: (record) => record.createdAt },
{ key: 'reason', title: '审核原因', render: (record) => record.reviewReason ?? record.rejectReason ?? record.riskHits?.map((item) => item.reason).join('') ?? '-' },
{