hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: harden CMPP delivery and platform workflows

Browse Source
This commit is contained in:
hectorzhao
2026-07-20 18:07:29 +08:00
parent 80fb5a8f53
commit f02c33cbb7
61 changed files with 1834 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+179 -7
View File
@@ -163,6 +163,8 @@ function createPrismaMock() {
},
smsReceiptRecord: {
create: jest.fn().mockResolvedValue({ id: 'receipt-1' }),
upsert: jest.fn().mockResolvedValue({ id: 'receipt-1', createdAt: new Date('2026-07-01T10:01:00.000Z') }),
findUnique: jest.fn().mockResolvedValue(null),
findFirst: jest.fn().mockResolvedValue(null),
findMany: jest.fn(),
},
@@ -355,6 +357,28 @@ describe('SendChainService', () => {
expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
});
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 });
prisma.smsTemplate.findFirst.mockResolvedValue({
id: 'tpl-http', tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】验证码${code}',
auditStatus: 'approved', signatureId: 'sig-1', signature: { id: 'sig-1', auditStatus: 'approved' },
});
await service.createHttpBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】验证码123456', phones: ['13800000001'],
sourceIp: '127.0.0.1', clientMessageId: 'client-http-1',
});
expect(riskReview.evaluateTask).toHaveBeenCalledWith(expect.objectContaining({
templateId: 'tpl-http',
variables: { code: '123456' },
}));
expect(prisma.smsBatchTask.findFirst).toHaveBeenCalledWith(expect.objectContaining({
where: expect.objectContaining({ id: 'task-1', sourceType: 'api' }),
}));
});
it('persists the unique longest approved drainage URL match on new message records', async () => {
const { service, prisma } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
@@ -576,6 +600,70 @@ describe('SendChainService', () => {
});
});
it('splits every destination in one inbound CMPP Submit into an independent real message record', async () => {
const { service, prisma } = createService();
prisma.smsApplication.findFirst.mockResolvedValue({
id: 'app-1',
tenantId: 'tenant-1',
cmppAccount: '100001',
secretHash: 'secret-hash',
status: 'active',
interfaceEnabled: false,
queuePriority: 'normal',
ipAllowlist: [{ ipCidr: '127.0.0.1/32' }],
tenant: { id: 'tenant-1', status: 'active', certificationStatus: 'approved' },
});
let taskIndex = 0;
prisma.smsBatchTask.create.mockImplementation(({ data }) => {
taskIndex += 1;
return Promise.resolve({ id: `task-${taskIndex}`, ...data });
});
let messageIndex = 0;
prisma.smsMessageRecord.create.mockImplementation(({ data }) => {
messageIndex += 1;
return Promise.resolve({ id: `record-${messageIndex}`, messageId: data.messageId, ...data });
});
const result = await service.submitInboundMessage({
account: '100001',
phoneNumbers: ['13800000001', '13900000002'],
content: 'hello',
sequenceId: 777823876,
remoteIp: '127.0.0.1',
});
expect(result).toEqual(expect.objectContaining({
accepted: true,
phoneCount: 2,
messages: [
expect.objectContaining({ phoneNumber: '13800000001', messageRecordId: 'record-1' }),
expect.objectContaining({ phoneNumber: '13900000002', messageRecordId: 'record-2' }),
],
}));
expect(result.messageId).toBe(result.messages[0].messageId);
expect(prisma.smsBatchTask.create).toHaveBeenCalledTimes(2);
expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(2);
expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ data: expect.objectContaining({ phoneNumber: '13800000001', cmppSubmitSequenceId: '777823876', cmppSubmitGroupMessageId: result.messageId }) });
expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ data: expect.objectContaining({ phoneNumber: '13900000002', cmppSubmitSequenceId: '777823876', cmppSubmitGroupMessageId: result.messageId }) });
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledTimes(2);
expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(2);
});
it('rejects a multi-destination CMPP Submit before persistence when any destination is invalid', async () => {
const { service, prisma } = createService();
await expect(service.submitInboundMessage({
account: '100001',
phoneNumbers: ['13800000001', 'invalid'],
content: 'hello',
remoteIp: '127.0.0.1',
})).rejects.toThrow('CMPP submit phone number is invalid');
expect(prisma.smsApplication.findFirst).not.toHaveBeenCalled();
expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled();
});
it('accepts only the filled client Src_Id and snapshots the real application extension', async () => {
const { service, prisma } = createService();
prisma.smsApplication.findFirst.mockResolvedValue({
@@ -684,7 +772,10 @@ describe('SendChainService', () => {
})).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' }));
expect(prisma.smsTemplate.findMany).toHaveBeenCalledWith({
where: { applicationId: 'app-1', content: { contains: '${' } },
where: {
applicationId: 'app-1', content: { contains: '${' }, auditStatus: 'approved',
signature: { auditStatus: 'approved' },
},
include: { signature: true },
orderBy: { updatedAt: 'desc' },
});
@@ -1238,8 +1329,10 @@ describe('SendChainService', () => {
it('matches receipt to a unique timed-out submit attempt when the upstream submit response was lost', async () => {
const { service, prisma } = createService();
prisma.smsMessageRecord.findFirst.mockResolvedValue(null);
prisma.smsSubmitRecord.findMany.mockResolvedValue([
prisma.smsMessageRecord.findUnique.mockResolvedValue(null);
prisma.smsSubmitRecord.findMany
.mockResolvedValueOnce([])
.mockResolvedValueOnce([
{
id: 'submit-timeout-1',
channelId: 'channel-1',
@@ -1258,7 +1351,7 @@ describe('SendChainService', () => {
status: 'timeout',
},
},
]);
]);
await service.handleReceipt({
messageId: 'receipt-123456789',
@@ -1290,10 +1383,89 @@ describe('SendChainService', () => {
});
});
it('matches identical upstream Msg_Id values by channel and destination instead of another channel record', async () => {
const { service, prisma } = createService();
prisma.smsMessageRecord.findUnique.mockResolvedValue(null);
prisma.smsSubmitRecord.findMany.mockResolvedValueOnce([
{
id: 'submit-channel-b',
channelId: 'channel-b',
gatewayMessageId: 'SHARED-UPSTREAM-ID',
messageRecord: {
id: 'record-channel-b',
tenantId: 'tenant-1',
batchTaskId: 'task-1',
applicationId: 'app-1',
messageId: 'MSG-B',
phoneNumber: '15601992925',
channelId: 'channel-b',
gatewayMessageId: 'SHARED-UPSTREAM-ID',
status: 'submitted',
},
},
]);
await service.handleReceipt({
messageId: 'receipt-SHARED-UPSTREAM-ID',
channelId: 'channel-b',
gatewayMessageId: 'SHARED-UPSTREAM-ID',
phoneNumber: '15601992925',
receiptStatus: 'delivered',
rawStatus: 'DELIVRD',
deliveredAt: '2026-07-01T10:01:00.000Z',
});
expect(prisma.smsSubmitRecord.findMany).toHaveBeenCalledWith(expect.objectContaining({
where: expect.objectContaining({
channelId: 'channel-b',
gatewayMessageId: 'SHARED-UPSTREAM-ID',
messageRecord: { phoneNumber: '15601992925' },
}),
}));
expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({
where: { id: 'record-channel-b' },
data: expect.objectContaining({
status: 'delivered',
receiptStatus: 'delivered',
channelId: 'channel-b',
gatewayMessageId: 'SHARED-UPSTREAM-ID',
receiptRawStatus: 'DELIVRD',
}),
});
});
it('treats a repeated DELIVRD event as idempotent and does not redeliver it downstream', async () => {
const { service, prisma } = createService();
prisma.smsReceiptRecord.findUnique
.mockResolvedValueOnce(null)
.mockResolvedValueOnce({
id: 'receipt-existing',
messageRecordId: 'record-1',
messageRecord: { id: 'record-1', messageId: 'MSG-1', status: 'delivered' },
});
const receipt = {
messageId: 'MSG-1',
channelId: 'channel-1',
gatewayMessageId: 'GW-1',
phoneNumber: '13800000001',
receiptStatus: 'delivered' as const,
rawStatus: 'DELIVRD',
deliveredAt: '2026-07-01T10:01:00.000Z',
};
await service.handleReceipt(receipt);
await service.handleReceipt(receipt);
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledTimes(1);
expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(1);
});
it('rejects ambiguous receipt heuristic matches to avoid binding to the wrong message', async () => {
const { service, prisma } = createService();
prisma.smsMessageRecord.findFirst.mockResolvedValue(null);
prisma.smsSubmitRecord.findMany.mockResolvedValue([
prisma.smsMessageRecord.findUnique.mockResolvedValue(null);
prisma.smsSubmitRecord.findMany
.mockResolvedValueOnce([])
.mockResolvedValueOnce([
{
id: 'submit-timeout-1',
messageRecord: { id: 'record-1', messageId: 'MSG-1', phoneNumber: '13800000001' },
@@ -1302,7 +1474,7 @@ describe('SendChainService', () => {
id: 'submit-timeout-2',
messageRecord: { id: 'record-2', messageId: 'MSG-2', phoneNumber: '13800000001' },
},
]);
]);
await expect(
service.handleReceipt({
api/src/send-chain/send-chain.service.ts
+176 -21
View File
@@ -29,6 +29,8 @@ export interface CreateBatchTaskDto {
clientMessageId?: string;
}
export type CreateHttpBatchTaskDto = Omit<CreateBatchTaskDto, 'templateId' | 'variables' | 'sourceType'>;
export interface GatewayInboundAuthDto {
account: string;
password?: string;
@@ -39,7 +41,8 @@ export interface GatewayInboundAuthDto {
export interface GatewayInboundSubmitDto {
account: string;
phoneNumber: string;
phoneNumber?: string;
phoneNumbers?: string[];
content: string;
srcId?: string;
destId?: string;
@@ -47,6 +50,16 @@ export interface GatewayInboundSubmitDto {
remoteIp?: string;
}
interface GatewayInboundSingleSubmitResult {
accepted: boolean;
tenantId: string;
applicationId: string;
taskId: string;
messageId: string;
messageRecordId: string;
status: string;
}
export interface GatewaySubmitResultDto {
traceId?: string;
messageId: string;
@@ -80,6 +93,7 @@ export interface GatewayReceiptEventDto {
receiptStatus: 'delivered' | 'undelivered' | 'unknown';
rawStatus: string;
errorCode?: string;
errorMessage?: string;
deliveredAt?: string;
}
@@ -400,7 +414,27 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (batchStatus === 'ready') {
await this.enqueueBatchTask(task.id);
}
return this.getBatchTask(task.id);
return this.getBatchTask(task.id, undefined, data.sourceType ?? 'client');
}
async createHttpBatchTask(data: CreateHttpBatchTaskDto) {
if (!data.applicationId) {
throw new BadRequestException('公开 HTTP 发送必须关联企业应用');
}
const template = await this.resolveInboundTemplateCandidate(data.applicationId, data.content);
if (!template || template.auditStatus !== 'approved' || template.signature?.auditStatus !== 'approved') {
throw new BadRequestException('短信内容未匹配当前应用已审核通过的签名和模板');
}
const variables = matchTemplateContent(template.content, data.content);
if (variables === null) {
throw new BadRequestException('短信内容与已审核模板不匹配');
}
return this.createBatchTask({
...data,
templateId: template.id,
variables,
sourceType: 'api',
});
}
async listBatchTasks(tenantId?: string, status?: string, sourceType = 'client') {
@@ -837,6 +871,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
async handleReceipt(data: GatewayReceiptEventDto) {
const receiptKey = this.receiptEventKey(data);
const existingReceipt = await this.prisma.smsReceiptRecord.findUnique({
where: { receiptKey },
include: { messageRecord: true },
});
if (existingReceipt?.messageRecord) {
return existingReceipt.messageRecord;
}
const resolved = await this.resolveReceiptMessage(data);
const message = resolved.message;
const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
@@ -854,21 +896,35 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
},
});
}
await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
channelId: data.channelId,
messageId: resolved.messageId,
gatewayMessageId: data.gatewayMessageId,
sequenceId: data.sequenceId,
receiptStatus: data.receiptStatus,
rawStatus: data.rawStatus,
errorCode: data.errorCode,
deliveredAt,
},
});
try {
await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
receiptKey,
channelId: data.channelId,
messageId: resolved.messageId,
gatewayMessageId: data.gatewayMessageId,
phoneNumber: data.phoneNumber?.trim() || message.phoneNumber,
sequenceId: data.sequenceId,
receiptStatus: data.receiptStatus,
rawStatus: data.rawStatus,
errorCode: data.errorCode,
errorMessage: data.errorMessage,
deliveredAt,
},
});
} catch (error) {
if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002') {
const duplicate = await this.prisma.smsReceiptRecord.findUnique({
where: { receiptKey },
include: { messageRecord: true },
});
if (duplicate?.messageRecord) return duplicate.messageRecord;
}
throw error;
}
await this.recordReceiptSegment(message, data, deliveredAt, resolved.submitRecordId);
const isCurrentAttempt =
(!message.channelId || message.channelId === data.channelId)
@@ -889,9 +945,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: {
channelId: data.channelId,
gatewayMessageId: data.gatewayMessageId,
receiptStatus: data.receiptStatus,
receiptRawStatus: data.rawStatus,
status,
errorCode: data.errorCode,
errorMessage: data.errorMessage ?? (status === 'delivered' ? null : data.rawStatus),
deliveredAt,
},
});
@@ -910,6 +970,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
rawStatus: data.rawStatus,
errorCode: data.errorCode,
submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined,
submitGroupMessageId: message.cmppSubmitGroupMessageId ?? undefined,
deliveredAt: deliveredAt.toISOString(),
},
});
@@ -1770,6 +1831,49 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
async submitInboundMessage(data: GatewayInboundSubmitDto) {
const phoneNumbers = data.phoneNumbers?.length
? data.phoneNumbers.map((phoneNumber) => phoneNumber.trim())
: data.phoneNumber
? [data.phoneNumber.trim()]
: [];
if (phoneNumbers.length === 0 || phoneNumbers.some((phoneNumber) => !/^1[3-9]\d{9}$/.test(phoneNumber))) {
throw new BadRequestException('CMPP submit phone number is invalid');
}
const submitGroupMessageId = `MSG-${randomUUID()}`;
const submissions = phoneNumbers.map((phoneNumber, index) => ({
phoneNumber,
messageId: index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`,
}));
const results: GatewayInboundSingleSubmitResult[] = [];
const concurrency = 10;
for (let offset = 0; offset < submissions.length; offset += concurrency) {
const batch = submissions.slice(offset, offset + concurrency);
results.push(...await Promise.all(batch.map((submission) => this.submitInboundSingleMessage({
...data,
phoneNumber: submission.phoneNumber,
phoneNumbers: undefined,
}, submission.messageId, submitGroupMessageId))));
}
const first = results[0];
return {
...first,
phoneCount: results.length,
messages: results.map((result, index) => ({
phoneNumber: phoneNumbers[index],
messageId: result.messageId,
messageRecordId: result.messageRecordId,
taskId: result.taskId,
status: result.status,
})),
};
}
private async submitInboundSingleMessage(
data: GatewayInboundSubmitDto & { phoneNumber: string },
messageId: string,
submitGroupMessageId: string,
) {
const application = await this.findInboundApplication(data.account);
if (!application) {
throw new BadRequestException('CMPP account is invalid');
@@ -1822,7 +1926,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
batchTaskId: task.id,
applicationId: application.id,
templateId: template?.id,
messageId: `MSG-${randomUUID()}`,
messageId,
phoneNumber: data.phoneNumber,
content: data.content,
billingUnits: billing.billingUnitsPerMessage,
@@ -1830,6 +1934,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
amountCents: billing.amountCents,
queuePriority,
cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId),
cmppSubmitGroupMessageId: submitGroupMessageId,
clientSrcId,
applicationExtension: application.cmppApplicationExtension,
status: 'validating',
@@ -2372,6 +2477,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
where: {
applicationId,
content,
auditStatus: 'approved',
signature: { auditStatus: 'approved' },
},
include: { signature: true },
orderBy: { updatedAt: 'desc' },
@@ -2381,6 +2488,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
where: {
applicationId,
content: { contains: '${' },
auditStatus: 'approved',
signature: { auditStatus: 'approved' },
},
include: { signature: true },
orderBy: { updatedAt: 'desc' },
@@ -2445,6 +2554,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
messageId: string;
phoneNumber: string;
cmppSubmitSequenceId?: string | null;
cmppSubmitGroupMessageId?: string | null;
},
errorCode: string,
reason: string,
@@ -2457,18 +2567,22 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
const deliveredAt = new Date();
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { status: 'failed', receiptStatus: 'undelivered', errorCode, errorMessage: reason, deliveredAt },
data: { status: 'failed', receiptStatus: 'undelivered', receiptRawStatus: 'REJECTD', errorCode, errorMessage: reason, deliveredAt },
});
const gatewayMessageId = `PLATFORM:${message.messageId}`;
const receipt = await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
receiptKey: createHash('sha256').update(`platform\u0000${gatewayMessageId}\u0000${message.phoneNumber}\u0000undelivered\u0000REJECTD\u0000${errorCode}`).digest('hex'),
messageId: message.messageId,
gatewayMessageId: `PLATFORM:${message.messageId}`,
gatewayMessageId,
phoneNumber: message.phoneNumber,
receiptStatus: 'undelivered',
rawStatus: 'REJECTD',
errorCode,
errorMessage: reason,
deliveredAt,
},
});
@@ -2487,6 +2601,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
errorCode,
errorMessage: reason,
submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined,
submitGroupMessageId: message.cmppSubmitGroupMessageId ?? undefined,
deliveredAt: deliveredAt.toISOString(),
},
});
@@ -2903,15 +3018,44 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
private async resolveReceiptMessage(data: GatewayReceiptEventDto) {
const exactMessage = await this.findMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
const exactMessage = data.messageId
? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } })
: null;
if (exactMessage) {
const submitRecord = await this.prisma.smsSubmitRecord.findFirst({
where: {
messageRecordId: exactMessage.id,
channelId: data.channelId,
gatewayMessageId: data.gatewayMessageId,
},
orderBy: { createdAt: 'desc' },
});
return {
message: exactMessage,
messageId: exactMessage.messageId,
submitRecordId: submitRecord?.id,
};
}
const phoneNumber = data.phoneNumber?.trim();
const exactSubmits = await this.prisma.smsSubmitRecord.findMany({
where: {
channelId: data.channelId,
gatewayMessageId: data.gatewayMessageId,
...(phoneNumber ? { messageRecord: { phoneNumber } } : {}),
},
include: { messageRecord: true },
orderBy: { createdAt: 'desc' },
take: 2,
});
if (exactSubmits.length === 1 && exactSubmits[0]?.messageRecord) {
return {
message: exactSubmits[0].messageRecord,
messageId: exactSubmits[0].messageRecord.messageId,
submitRecordId: exactSubmits[0].id,
};
}
if (!phoneNumber) {
throw new NotFoundException('SMS message record not found');
}
@@ -2951,6 +3095,17 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
};
}
private receiptEventKey(data: GatewayReceiptEventDto) {
return createHash('sha256').update([
data.channelId,
data.gatewayMessageId,
data.phoneNumber?.trim() ?? '',
data.receiptStatus,
data.rawStatus.trim(),
data.errorCode ?? '',
].join('\u0000')).digest('hex');
}
private getSendQueue(): Queue<SendJob, unknown, 'send-message'> {
if (!this.sendQueue) {
this.sendQueue = new Queue<SendJob, unknown, 'send-message'>(SEND_QUEUE, { connection: bullmqConnection() });