From b29576fcd118bea04416be0c9fc1bc2a4213d830 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Thu, 23 Jul 2026 21:04:35 +0800 Subject: [PATCH] fix: reassemble inbound CMPP long messages --- .../migration.sql | 61 +++ api/prisma/schema.prisma | 45 +++ api/src/send-chain/send-chain.service.spec.ts | 363 ++++++++++++++++- api/src/send-chain/send-chain.service.ts | 369 +++++++++++++++++- .../first-version-development-requirements.md | 3 + docs/system-functional-test-cases.md | 34 ++ docs/testing-progress.md | 18 + gateway/internal/inbound/server.go | 93 ++++- gateway/internal/inbound/server_test.go | 138 +++++++ 9 files changed, 1102 insertions(+), 22 deletions(-) create mode 100644 api/prisma/migrations/20260723120000_add_cmpp_inbound_long_message_reassembly/migration.sql diff --git a/api/prisma/migrations/20260723120000_add_cmpp_inbound_long_message_reassembly/migration.sql b/api/prisma/migrations/20260723120000_add_cmpp_inbound_long_message_reassembly/migration.sql new file mode 100644 index 0000000..bcb19c4 --- /dev/null +++ b/api/prisma/migrations/20260723120000_add_cmpp_inbound_long_message_reassembly/migration.sql @@ -0,0 +1,61 @@ +CREATE TABLE "CmppInboundLongMessage" ( + "id" TEXT NOT NULL, + "tenantId" TEXT NOT NULL, + "applicationId" TEXT NOT NULL, + "groupKey" TEXT NOT NULL, + "account" TEXT NOT NULL, + "srcId" TEXT, + "phoneNumbers" JSONB NOT NULL, + "concatReference" INTEGER NOT NULL, + "segmentTotal" INTEGER NOT NULL, + "msgFmt" INTEGER NOT NULL, + "messageId" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'collecting', + "response" JSONB, + "expiresAt" TIMESTAMP(3) NOT NULL, + "completedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "CmppInboundLongMessage_pkey" PRIMARY KEY ("id") +); + +CREATE TABLE "CmppInboundLongMessageSegment" ( + "id" TEXT NOT NULL, + "groupId" TEXT NOT NULL, + "segmentIndex" INTEGER NOT NULL, + "sequenceId" TEXT, + "content" TEXT NOT NULL, + "contentHash" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "CmppInboundLongMessageSegment_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "CmppInboundLongMessage_messageId_key" +ON "CmppInboundLongMessage"("messageId"); + +CREATE INDEX "CmppInboundLongMessage_groupKey_status_createdAt_idx" +ON "CmppInboundLongMessage"("groupKey", "status", "createdAt"); + +CREATE INDEX "CmppInboundLongMessage_applicationId_status_expiresAt_idx" +ON "CmppInboundLongMessage"("applicationId", "status", "expiresAt"); + +CREATE UNIQUE INDEX "CmppInboundLongMessageSegment_groupId_segmentIndex_key" +ON "CmppInboundLongMessageSegment"("groupId", "segmentIndex"); + +CREATE INDEX "CmppInboundLongMessageSegment_sequenceId_idx" +ON "CmppInboundLongMessageSegment"("sequenceId"); + +ALTER TABLE "CmppInboundLongMessage" +ADD CONSTRAINT "CmppInboundLongMessage_tenantId_fkey" +FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +ALTER TABLE "CmppInboundLongMessage" +ADD CONSTRAINT "CmppInboundLongMessage_applicationId_fkey" +FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +ALTER TABLE "CmppInboundLongMessageSegment" +ADD CONSTRAINT "CmppInboundLongMessageSegment_groupId_fkey" +FOREIGN KEY ("groupId") REFERENCES "CmppInboundLongMessage"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 69b3071..fa4bf90 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -47,6 +47,7 @@ model Tenant { gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] openApiRequests OpenApiRequest[] httpWebhookEvents HttpWebhookEvent[] + cmppInboundLongMessages CmppInboundLongMessage[] } model EnterpriseCertification { @@ -400,6 +401,7 @@ model SmsApplication { httpWebhookEndpoints HttpWebhookEndpoint[] httpWebhookEvents HttpWebhookEvent[] dailyUsages SmsApplicationDailyUsage[] + inboundLongMessages CmppInboundLongMessage[] @@index([tenantId, status]) } @@ -1541,6 +1543,49 @@ model SmsMessageSegmentAudit { @@index([submitStatus, receiptStatus]) } +model CmppInboundLongMessage { + id String @id @default(cuid()) + tenantId String + applicationId String + groupKey String + account String + srcId String? + phoneNumbers Json + concatReference Int + segmentTotal Int + msgFmt Int + messageId String @unique + status String @default("collecting") + response Json? + expiresAt DateTime + completedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) + segments CmppInboundLongMessageSegment[] + + @@index([groupKey, status, createdAt]) + @@index([applicationId, status, expiresAt]) +} + +model CmppInboundLongMessageSegment { + id String @id @default(cuid()) + groupId String + segmentIndex Int + sequenceId String? + content String + contentHash String + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + group CmppInboundLongMessage @relation(fields: [groupId], references: [id], onDelete: Cascade) + + @@unique([groupId, segmentIndex]) + @@index([sequenceId]) +} + model SmsReceiptRecord { id String @id @default(cuid()) tenantId String? diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 212d02c..003ce3a 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { BillingService } from '../billing/billing.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { SendChainService } from './send-chain.service'; @@ -58,7 +59,7 @@ function createPrismaMock() { items: [{ id: 'item-1', groupId: 'group-1', channelId: 'channel-1', carrier: 'mobile', priority: 1, province: null, channel }], }, }; - return { + const prisma = { tenant: { findUnique: jest.fn().mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'approved' }), }, @@ -160,6 +161,16 @@ function createPrismaMock() { updateMany: jest.fn().mockResolvedValue({ count: 1 }), findFirst: jest.fn().mockResolvedValue(null), }, + cmppInboundLongMessage: { + create: jest.fn(), + findFirst: jest.fn().mockResolvedValue(null), + update: jest.fn(), + updateMany: jest.fn().mockResolvedValue({ count: 0 }), + }, + cmppInboundLongMessageSegment: { + create: jest.fn(), + findMany: jest.fn().mockResolvedValue([]), + }, channelSignatureReportTask: { findFirst: jest.fn().mockResolvedValue({ id: 'report-task-1' }), findMany: jest.fn().mockImplementation(({ where }) => Promise.resolve((where.channelId?.in ?? []).map((channelId: string) => ({ channelId })))), @@ -301,8 +312,13 @@ function createPrismaMock() { findMany: jest.fn().mockResolvedValue([]), }, $queryRaw: jest.fn().mockResolvedValue([{ dailyLimit: 100000, usedCount: 2 }]), - $transaction: jest.fn((operations) => Promise.all(operations)), + $executeRaw: jest.fn().mockResolvedValue(1), + $transaction: jest.fn(), }; + prisma.$transaction.mockImplementation((operations: any) => typeof operations === 'function' + ? operations(prisma) + : Promise.all(operations)); + return prisma; } function createService(prisma = createPrismaMock()) { @@ -850,6 +866,349 @@ describe('SendChainService', () => { expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(2); }); + it('persists inbound CMPP long-message fragments and creates one complete main record after reassembly', async () => { + const { service, prisma, billing } = createService(); + service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 }); + const segments: Array> = []; + const group = { + id: 'long-group-1', + tenantId: 'tenant-1', + applicationId: 'app-1', + groupKey: 'group-key', + account: '100001', + srcId: null, + phoneNumbers: ['13800000001'], + concatReference: 16, + segmentTotal: 2, + msgFmt: 8, + messageId: 'MSG-LONG-1', + status: 'collecting', + response: null, + expiresAt: new Date(Date.now() + 300_000), + createdAt: new Date(), + segments, + }; + prisma.cmppInboundLongMessage.findFirst.mockImplementation(() => Promise.resolve( + segments.length ? { ...group, segments: [...segments] } : null, + )); + prisma.cmppInboundLongMessage.create.mockResolvedValue(group); + prisma.cmppInboundLongMessageSegment.create.mockImplementation(({ data }: { data: any }) => { + const segment = { id: `segment-${data.segmentIndex}`, ...data }; + segments.push(segment); + return Promise.resolve(segment); + }); + prisma.cmppInboundLongMessageSegment.findMany.mockImplementation(() => Promise.resolve( + [...segments].sort((a, b) => a.segmentIndex - b.segmentIndex), + )); + prisma.cmppInboundLongMessage.update.mockImplementation(({ data }: { data: any }) => { + Object.assign(group, data); + return Promise.resolve({ ...group }); + }); + + const first = await service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: '【签名】第一片', + sequenceId: 101, + remoteIp: '127.0.0.1', + longMessage: { reference: 16, total: 2, index: 1, format: 8 }, + }); + expect(first).toEqual(expect.objectContaining({ + accepted: true, + fragmentPending: true, + messageId: 'MSG-LONG-1', + receivedSegments: 1, + })); + expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + + const second = await service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: '第二片正文', + sequenceId: 102, + remoteIp: '127.0.0.1', + longMessage: { reference: 16, total: 2, index: 2, format: 8 }, + }); + expect(second).toEqual(expect.objectContaining({ + accepted: true, + messageId: 'MSG-LONG-1', + messageRecordId: 'record-1', + })); + expect(prisma.smsBatchTask.create).toHaveBeenCalledTimes(1); + expect(prisma.smsBatchTask.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ content: '【签名】第一片第二片正文', phoneTotal: 1 }), + }); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(1); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + content: '【签名】第一片第二片正文', + cmppSubmitSequenceId: '101', + cmppSubmitGroupMessageId: 'MSG-LONG-1', + }), + }); + expect(billing.estimateSmsCost).toHaveBeenCalledWith(expect.objectContaining({ + content: '【签名】第一片第二片正文', + phoneCount: 1, + })); + expect(prisma.cmppInboundLongMessage.update).toHaveBeenCalledWith({ + where: { id: 'long-group-1' }, + data: expect.objectContaining({ status: 'completed', response: expect.any(Object) }), + }); + }); + + it('accepts out-of-order and duplicate CMPP long-message fragments but rejects conflicting duplicates', async () => { + const { service, prisma } = createService(); + service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 }); + const segments: Array> = []; + const group = { + id: 'long-group-2', + tenantId: 'tenant-1', + applicationId: 'app-1', + groupKey: 'group-key-2', + account: '100001', + srcId: null, + phoneNumbers: ['13800000001'], + concatReference: 17, + segmentTotal: 2, + msgFmt: 8, + messageId: 'MSG-LONG-2', + status: 'collecting', + response: null, + expiresAt: new Date(Date.now() + 300_000), + createdAt: new Date(), + segments, + }; + prisma.cmppInboundLongMessage.findFirst.mockImplementation(() => Promise.resolve( + segments.length ? { ...group, segments: [...segments] } : null, + )); + prisma.cmppInboundLongMessage.create.mockResolvedValue(group); + prisma.cmppInboundLongMessageSegment.create.mockImplementation(({ data }: { data: any }) => { + const segment = { id: `segment-${data.segmentIndex}`, ...data }; + segments.push(segment); + return Promise.resolve(segment); + }); + prisma.cmppInboundLongMessageSegment.findMany.mockImplementation(() => Promise.resolve( + [...segments].sort((a, b) => a.segmentIndex - b.segmentIndex), + )); + prisma.cmppInboundLongMessage.update.mockImplementation(({ data }: { data: any }) => { + Object.assign(group, data); + return Promise.resolve({ ...group }); + }); + + const secondFragment = { + account: '100001', + phoneNumber: '13800000001', + content: '第二片正文', + sequenceId: 202, + remoteIp: '127.0.0.1', + longMessage: { reference: 17, total: 2, index: 2, format: 8 }, + }; + await expect(service.submitInboundMessage(secondFragment)).resolves.toEqual(expect.objectContaining({ + fragmentPending: true, + receivedSegments: 1, + })); + await expect(service.submitInboundMessage(secondFragment)).resolves.toEqual(expect.objectContaining({ + fragmentPending: true, + receivedSegments: 1, + })); + expect(prisma.cmppInboundLongMessageSegment.create).toHaveBeenCalledTimes(1); + + await expect(service.submitInboundMessage({ + ...secondFragment, + content: '冲突的第二片', + })).rejects.toThrow('fragment 2 conflicts'); + + await expect(service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: '【签名】第一片', + sequenceId: 201, + remoteIp: '127.0.0.1', + longMessage: { reference: 17, total: 2, index: 1, format: 8 }, + })).resolves.toEqual(expect.objectContaining({ + accepted: true, + messageId: 'MSG-LONG-2', + messageRecordId: 'record-1', + })); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(1); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + content: '【签名】第一片第二片正文', + cmppSubmitSequenceId: '201', + }), + }); + }); + + it('resumes a persistently complete CMPP long message after processing is interrupted by a restart', async () => { + const { service, prisma } = createService(); + service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 }); + prisma.smsMessageRecord.findMany.mockResolvedValue([]); + const staleAt = new Date(Date.now() - 60_000); + const group = { + id: 'long-group-restart', + tenantId: 'tenant-1', + applicationId: 'app-1', + groupKey: 'group-key-restart', + account: '100001', + srcId: null, + phoneNumbers: ['13800000001'], + concatReference: 18, + segmentTotal: 2, + msgFmt: 8, + messageId: 'MSG-LONG-RESTART', + status: 'processing', + response: null, + expiresAt: new Date(Date.now() + 300_000), + createdAt: staleAt, + updatedAt: staleAt, + segments: [ + { + id: 'segment-restart-1', + groupId: 'long-group-restart', + segmentIndex: 1, + sequenceId: '301', + content: '【签名】第一片', + contentHash: createHash('sha256').update('【签名】第一片').digest('hex'), + }, + { + id: 'segment-restart-2', + groupId: 'long-group-restart', + segmentIndex: 2, + sequenceId: '302', + content: '第二片正文', + contentHash: createHash('sha256').update('第二片正文').digest('hex'), + }, + ], + }; + prisma.cmppInboundLongMessage.findFirst.mockResolvedValue(group); + prisma.cmppInboundLongMessage.update.mockImplementation(({ data }: { data: any }) => { + Object.assign(group, data, { updatedAt: new Date() }); + return Promise.resolve({ ...group }); + }); + + await expect(service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: '第二片正文', + sequenceId: 302, + remoteIp: '127.0.0.1', + longMessage: { reference: 18, total: 2, index: 2, format: 8 }, + })).resolves.toEqual(expect.objectContaining({ + accepted: true, + messageId: 'MSG-LONG-RESTART', + messageRecordId: 'record-1', + })); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(1); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + content: '【签名】第一片第二片正文', + cmppSubmitSequenceId: '301', + cmppSubmitGroupMessageId: 'MSG-LONG-RESTART', + }), + }); + expect(prisma.cmppInboundLongMessage.update).toHaveBeenCalledWith({ + where: { id: 'long-group-restart' }, + data: expect.objectContaining({ status: 'completed', response: expect.any(Object) }), + }); + }); + + it('recovers the persisted SMS result after a restart without creating a duplicate main record', async () => { + const { service, prisma } = createService(); + const staleAt = new Date(Date.now() - 60_000); + const group = { + id: 'long-group-after-record', + tenantId: 'tenant-1', + applicationId: 'app-1', + groupKey: 'group-key-after-record', + account: '100001', + srcId: null, + phoneNumbers: ['13800000001'], + concatReference: 19, + segmentTotal: 2, + msgFmt: 8, + messageId: 'MSG-LONG-AFTER-RECORD', + status: 'processing', + response: null, + expiresAt: new Date(Date.now() + 300_000), + createdAt: staleAt, + updatedAt: staleAt, + segments: [ + { + id: 'segment-after-record-1', + groupId: 'long-group-after-record', + segmentIndex: 1, + sequenceId: '401', + content: '【签名】第一片', + contentHash: createHash('sha256').update('【签名】第一片').digest('hex'), + }, + { + id: 'segment-after-record-2', + groupId: 'long-group-after-record', + segmentIndex: 2, + sequenceId: '402', + content: '第二片正文', + contentHash: createHash('sha256').update('第二片正文').digest('hex'), + }, + ], + }; + prisma.cmppInboundLongMessage.findFirst.mockResolvedValue(group); + prisma.cmppInboundLongMessage.update.mockImplementation(({ data }: { data: any }) => { + Object.assign(group, data, { updatedAt: new Date() }); + return Promise.resolve({ ...group }); + }); + prisma.smsMessageRecord.findMany.mockResolvedValue([{ + id: 'persisted-record-1', + tenantId: 'tenant-1', + applicationId: 'app-1', + batchTaskId: 'persisted-task-1', + messageId: 'MSG-LONG-AFTER-RECORD', + phoneNumber: '13800000001', + status: 'failed', + errorCode: 'SIGNATURE', + }]); + + await expect(service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: '第二片正文', + sequenceId: 402, + remoteIp: '127.0.0.1', + longMessage: { reference: 19, total: 2, index: 2, format: 8 }, + })).resolves.toEqual(expect.objectContaining({ + accepted: true, + messageId: 'MSG-LONG-AFTER-RECORD', + messageRecordId: 'persisted-record-1', + taskId: 'persisted-task-1', + })); + expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + expect(prisma.cmppInboundLongMessage.update).toHaveBeenCalledWith({ + where: { id: 'long-group-after-record' }, + data: expect.objectContaining({ status: 'completed', response: expect.any(Object) }), + }); + }); + + it('expires incomplete or interrupted CMPP long-message groups without creating SMS records', async () => { + const { service, prisma } = createService(); + const now = new Date('2026-07-23T12:00:00.000Z'); + prisma.cmppInboundLongMessage.updateMany.mockResolvedValue({ count: 2 }); + + await expect(service.expireInboundLongMessages(now)).resolves.toEqual({ count: 2 }); + expect(prisma.cmppInboundLongMessage.updateMany).toHaveBeenCalledWith({ + where: { + status: { in: ['collecting', 'processing'] }, + expiresAt: { lte: now }, + }, + data: { + status: 'expired', + completedAt: now, + }, + }); + expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + }); + it('rejects the whole CMPP Submit synchronously while keeping per-destination audit records when the daily limit is exceeded', async () => { const { service, prisma, billing } = createService(); prisma.$queryRaw.mockResolvedValueOnce([{ dailyLimit: 1, usedCount: null }]); diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 9116374..7187bcc 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -48,6 +48,12 @@ export interface GatewayInboundSubmitDto { destId?: string; sequenceId?: number; remoteIp?: string; + longMessage?: { + reference: number; + total: number; + index: number; + format: number; + }; } interface GatewayInboundSingleSubmitResult { @@ -260,6 +266,9 @@ const DEFAULT_SCHEDULED_DISPATCH_STALE_MS = 2 * 60_000; const SCHEDULED_DISPATCH_INITIAL_DELAY_MS = 1_000; const DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS = 2 * 60_000; const DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS = 2 * 60_000; +const DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS = 60_000; +const INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS = 10_000; +const DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS = 30; const GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS = 30 * 24 * 60 * 60; const BULLMQ_PRIORITY: Record = { priority: 1, @@ -279,6 +288,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { private scheduledDispatchInitialTimer?: ReturnType; private scheduledDispatchIntervalTimer?: ReturnType; private scheduledDispatchScanRunning = false; + private inboundLongMessageInitialTimer?: ReturnType; + private inboundLongMessageIntervalTimer?: ReturnType; constructor( private readonly prisma: PrismaService, @@ -312,6 +323,25 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ); this.scheduledDispatchIntervalTimer.unref?.(); } + if (process.env.CMPP_INBOUND_LONG_MESSAGE_SCAN_ENABLED !== 'false') { + this.inboundLongMessageInitialTimer = setTimeout( + () => void this.expireInboundLongMessages().catch((error) => { + this.logger.error(`Failed to expire inbound CMPP long messages: ${String(error)}`); + }), + INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS, + ); + this.inboundLongMessageInitialTimer.unref?.(); + this.inboundLongMessageIntervalTimer = setInterval( + () => void this.expireInboundLongMessages().catch((error) => { + this.logger.error(`Failed to expire inbound CMPP long messages: ${String(error)}`); + }), + positiveInteger( + process.env.CMPP_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS, + DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS, + ), + ); + this.inboundLongMessageIntervalTimer.unref?.(); + } } async onModuleDestroy() { @@ -319,6 +349,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (this.receiptTimeoutIntervalTimer) clearInterval(this.receiptTimeoutIntervalTimer); if (this.scheduledDispatchInitialTimer) clearTimeout(this.scheduledDispatchInitialTimer); if (this.scheduledDispatchIntervalTimer) clearInterval(this.scheduledDispatchIntervalTimer); + if (this.inboundLongMessageInitialTimer) clearTimeout(this.inboundLongMessageInitialTimer); + if (this.inboundLongMessageIntervalTimer) clearInterval(this.inboundLongMessageIntervalTimer); await this.worker?.close(); await this.sendQueue?.close(); await this.gatewayQueue?.close(); @@ -2027,28 +2059,174 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (!application) { throw new BadRequestException('CMPP account is invalid'); } - const dailyQuota = await this.tryReserveDailySendQuota(application.id, phoneNumbers.length); + 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'); + } + validateInboundApplicationSrcId(data.srcId, application); + const collection = await this.collectInboundLongMessageFragment(data, application, phoneNumbers); + if (collection.response) { + return collection.response; + } + if (!collection.complete) { + return { + accepted: true, + messageId: collection.messageId, + status: 'fragment_pending', + fragmentPending: true, + receivedSegments: collection.receivedSegments, + segmentTotal: data.longMessage.total, + phoneCount: phoneNumbers.length, + messages: phoneNumbers.map((phoneNumber) => ({ + phoneNumber, + messageId: collection.messageId, + status: 'fragment_pending', + })), + }; + } + try { + const response = await this.recoverCompletedInboundLongMessageResponse( + collection.messageId, + phoneNumbers, + ) ?? await this.submitCompleteInboundMessage({ + ...data, + content: collection.content, + sequenceId: collection.sequenceId, + longMessage: undefined, + }, phoneNumbers, application, collection.messageId); + await this.prisma.cmppInboundLongMessage.update({ + where: { id: collection.groupId }, + data: { + status: 'completed', + response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue, + completedAt: new Date(), + }, + }); + return response; + } catch (error) { + await this.prisma.cmppInboundLongMessage.update({ + where: { id: collection.groupId }, + data: { + status: 'rejected', + completedAt: new Date(), + }, + }).catch(() => undefined); + throw error; + } + } + return this.submitCompleteInboundMessage(data, phoneNumbers, application); + } + + private async recoverCompletedInboundLongMessageResponse(messageId: string, phoneNumbers: string[]) { + const existing = await this.prisma.smsMessageRecord.findMany({ + where: { + cmppSubmitGroupMessageId: messageId, + phoneNumber: { in: phoneNumbers }, + }, + select: { + id: true, + tenantId: true, + applicationId: true, + batchTaskId: true, + messageId: true, + phoneNumber: true, + status: true, + errorCode: true, + }, + }); + const byPhone = new Map(existing.map((item) => [item.phoneNumber, item])); + const ordered = phoneNumbers.map((phoneNumber) => byPhone.get(phoneNumber)); + if (ordered.some((item) => !item)) { + return null; + } + const messages = ordered.map((item, index) => ({ + phoneNumber: phoneNumbers[index], + messageId: item!.messageId, + messageRecordId: item!.id, + taskId: item!.batchTaskId ?? '', + status: item!.status, + })); + const first = ordered[0]!; + const dailyLimitRejected = ordered.every((item) => item!.errorCode === 'DAILY_LIMIT'); + return { + accepted: !dailyLimitRejected, + tenantId: first.tenantId ?? '', + applicationId: first.applicationId ?? '', + taskId: first.batchTaskId ?? '', + messageId: first.messageId, + messageRecordId: first.id, + status: dailyLimitRejected ? 'rejected' : 'accepted', + result: dailyLimitRejected ? 8 : undefined, + phoneCount: messages.length, + messages, + }; + } + + private async submitCompleteInboundMessage( + data: GatewayInboundSubmitDto, + phoneNumbers: string[], + application: Awaited>, + requestedGroupMessageId?: string, + ) { + if (!application) { + throw new BadRequestException('CMPP account is invalid'); + } + const persisted = requestedGroupMessageId + ? await this.prisma.smsMessageRecord.findMany({ + where: { + cmppSubmitGroupMessageId: requestedGroupMessageId, + phoneNumber: { in: phoneNumbers }, + }, + select: { + id: true, + tenantId: true, + applicationId: true, + batchTaskId: true, + messageId: true, + phoneNumber: true, + status: true, + errorCode: true, + }, + }) + : []; + const persistedByPhone = new Map(persisted.map((item) => [item.phoneNumber, item])); + const missingPhoneCount = phoneNumbers.filter((phoneNumber) => !persistedByPhone.has(phoneNumber)).length; + const dailyQuota = missingPhoneCount > 0 + ? await this.tryReserveDailySendQuota(application.id, missingPhoneCount) + : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 }; const dailyLimitRejection = dailyQuota.reserved ? undefined : { code: 'DAILY_LIMIT', - reason: `应用当日发送上限${dailyQuota.dailyLimit}条，本次${phoneNumbers.length}条超出剩余配额`, + reason: `应用当日发送上限${dailyQuota.dailyLimit}条，本次${missingPhoneCount}条超出剩余配额`, }; - const submitGroupMessageId = `MSG-${randomUUID()}`; + const submitGroupMessageId = requestedGroupMessageId ?? `MSG-${randomUUID()}`; const submissions = phoneNumbers.map((phoneNumber, index) => ({ phoneNumber, - messageId: index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`, + persisted: persistedByPhone.get(phoneNumber), + messageId: persistedByPhone.get(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, dailyLimitRejection)))); + results.push(...await Promise.all(batch.map((submission) => submission.persisted + ? Promise.resolve({ + accepted: submission.persisted.errorCode !== 'DAILY_LIMIT', + tenantId: submission.persisted.tenantId ?? application.tenantId, + applicationId: submission.persisted.applicationId ?? application.id, + taskId: submission.persisted.batchTaskId ?? '', + messageId: submission.persisted.messageId, + messageRecordId: submission.persisted.id, + status: submission.persisted.status, + }) + : this.submitInboundSingleMessage({ + ...data, + phoneNumber: submission.phoneNumber, + phoneNumbers: undefined, + }, submission.messageId, submitGroupMessageId, dailyLimitRejection)))); } const first = results[0]; return { @@ -2065,6 +2243,173 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }; } + private async collectInboundLongMessageFragment( + data: GatewayInboundSubmitDto, + application: NonNullable>>, + phoneNumbers: string[], + ) { + const fragment = data.longMessage; + if (!fragment || !Number.isInteger(fragment.reference) || fragment.reference < 0 || fragment.reference > 65535 + || !Number.isInteger(fragment.total) || fragment.total < 2 || fragment.total > 255 + || !Number.isInteger(fragment.index) || fragment.index < 1 || fragment.index > fragment.total + || !Number.isInteger(fragment.format) || fragment.format < 0 || fragment.format > 255) { + throw new BadRequestException('CMPP long message fragment metadata is invalid'); + } + const groupKey = createHash('sha256').update(JSON.stringify({ + applicationId: application.id, + account: data.account, + srcId: data.srcId?.trim() ?? '', + phoneNumbers, + reference: fragment.reference, + total: fragment.total, + format: fragment.format, + })).digest('hex'); + const contentHash = createHash('sha256').update(data.content).digest('hex'); + const now = new Date(); + const expiresAt = new Date(now.getTime() + positiveInteger( + process.env.CMPP_INBOUND_LONG_MESSAGE_TTL_SECONDS, + 300, + ) * 1000); + + return this.prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${groupKey}, 0))`; + await tx.cmppInboundLongMessage.updateMany({ + where: { + groupKey, + status: { in: ['collecting', 'processing'] }, + expiresAt: { lte: now }, + }, + data: { status: 'expired', completedAt: now }, + }); + + const recent = await tx.cmppInboundLongMessage.findFirst({ + where: { + groupKey, + expiresAt: { gt: now }, + }, + include: { segments: { orderBy: { segmentIndex: 'asc' } } }, + orderBy: { createdAt: 'desc' }, + }); + const matchingRecentSegment = recent?.segments.find((item) => item.segmentIndex === fragment.index); + if (recent && ['completed', 'rejected'].includes(recent.status) + && matchingRecentSegment?.contentHash === contentHash + && matchingRecentSegment.sequenceId === (data.sequenceId == null ? null : String(data.sequenceId))) { + return { + complete: recent.status === 'completed', + groupId: recent.id, + messageId: recent.messageId, + receivedSegments: recent.segments.length, + response: recent.response as any, + content: recent.segments.map((item) => item.content).join(''), + sequenceId: parseOptionalSequenceId(recent.segments[0]?.sequenceId), + }; + } + + let group = recent && ['collecting', 'processing'].includes(recent.status) ? recent : null; + if (!group) { + group = await tx.cmppInboundLongMessage.create({ + data: { + tenantId: application.tenantId, + applicationId: application.id, + groupKey, + account: data.account, + srcId: data.srcId?.trim() || null, + phoneNumbers, + concatReference: fragment.reference, + segmentTotal: fragment.total, + msgFmt: fragment.format, + messageId: `MSG-${randomUUID()}`, + expiresAt, + }, + include: { segments: { orderBy: { segmentIndex: 'asc' } } }, + }); + } + if (group.status === 'processing') { + const processingStaleMs = positiveInteger( + process.env.CMPP_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, + DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, + ) * 1000; + const complete = group.segments.length === fragment.total + && group.segments.every((item, index) => item.segmentIndex === index + 1); + if (complete && now.getTime() - group.updatedAt.getTime() >= processingStaleMs) { + await tx.cmppInboundLongMessage.update({ + where: { id: group.id }, + data: { status: 'processing', expiresAt }, + }); + return { + complete: true, + groupId: group.id, + messageId: group.messageId, + receivedSegments: group.segments.length, + response: null, + content: group.segments.map((item) => item.content).join(''), + sequenceId: parseOptionalSequenceId(group.segments[0]?.sequenceId), + }; + } + return { + complete: false, + groupId: group.id, + messageId: group.messageId, + receivedSegments: group.segments.length, + response: group.response as any, + content: '', + sequenceId: undefined, + }; + } + + const existing = group.segments.find((item) => item.segmentIndex === fragment.index); + if (existing && (existing.contentHash !== contentHash + || existing.sequenceId !== (data.sequenceId == null ? null : String(data.sequenceId)))) { + throw new BadRequestException(`CMPP long message fragment ${fragment.index} conflicts with the stored fragment`); + } + if (!existing) { + await tx.cmppInboundLongMessageSegment.create({ + data: { + groupId: group.id, + segmentIndex: fragment.index, + sequenceId: data.sequenceId == null ? null : String(data.sequenceId), + content: data.content, + contentHash, + }, + }); + } + const segments = await tx.cmppInboundLongMessageSegment.findMany({ + where: { groupId: group.id }, + orderBy: { segmentIndex: 'asc' }, + }); + const complete = segments.length === fragment.total + && segments.every((item, index) => item.segmentIndex === index + 1); + if (complete) { + await tx.cmppInboundLongMessage.update({ + where: { id: group.id }, + data: { status: 'processing', expiresAt }, + }); + } + return { + complete, + groupId: group.id, + messageId: group.messageId, + receivedSegments: segments.length, + response: null, + content: complete ? segments.map((item) => item.content).join('') : '', + sequenceId: parseOptionalSequenceId(segments[0]?.sequenceId), + }; + }); + } + + async expireInboundLongMessages(now = new Date()) { + return this.prisma.cmppInboundLongMessage.updateMany({ + where: { + status: { in: ['collecting', 'processing'] }, + expiresAt: { lte: now }, + }, + data: { + status: 'expired', + completedAt: now, + }, + }); + } + private async submitInboundSingleMessage( data: GatewayInboundSubmitDto & { phoneNumber: string }, messageId: string, @@ -3782,6 +4127,12 @@ function positiveInteger(value: string | undefined, fallback: number) { return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; } +function parseOptionalSequenceId(value: string | null | undefined) { + if (!value) return undefined; + const parsed = Number(value); + return Number.isInteger(parsed) && parsed >= 0 && parsed <= 0xffffffff ? parsed : undefined; +} + function shanghaiDateKey(now = new Date()) { const parts = new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai', diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index b1a7feb..abc94d7 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -180,6 +180,8 @@ 6. 一个企业应用可以分别绑定移动、联通、电信通道组；可以只绑定其中一类或两类，但不能一个都不绑定。 7. 发送时先按运营商识别结果分流：移动短信走移动通道组，联通短信走联通通道组，电信短信走电信通道组；识别不出的号码走移动通道组。 8. 运营商识别以可配置的号码前缀正则表达式为准，通常匹配手机号前 3 到 4 位；手机号段库只提供省份/城市识别，其 carrier 字段仅作后台校验或提示，不参与发送运营商判定。 + - 预发布运营商规则按公开码号资料覆盖中国移动、中国联通、中国电信及其移动转售号段；中国广电 `192` 号段按当前业务约定归入中国移动路由。 + - 号码前缀表示原始码号分配关系；携号转网号码无法仅凭前缀识别当前签约运营商，如后续要求按实时在网运营商路由，必须接入可信的携号转网/HLR 查询能力。 9. 路由规则只能表达应用到通道组的绑定关系，不能直接绑定单个通道，也不在规则层配置省份；省份和全国路由在通道组内部处理。 10. 发送服务必须使用手机号段库识别手机号省份和城市；无法识别省份时走对应运营商的全国通道组路由。 11. 通道组内必须先匹配省网路由；省网未匹配时走同一运营商通道组内的全国通道。省网发送失败后，当前版本立即跳到该通道组第一个全国通道补发，不再尝试同省第二省网通道。 @@ -254,6 +256,7 @@ 11. Gateway 必须支持下游客户上行接入场景：收到运营商上行后，按接入号、手机号、应用、时间窗口匹配并向客户连接推送 Deliver，上行同时入库。 12. 下游客户连接与上游通道连接必须隔离管理：客户侧账号密码不能用于连接上游通道，上游通道账号密码也不能作为客户接入凭据。 13. Gateway 必须为客户侧 CMPP2.0/3.0 Submit 记录可检索日志：收包时记录协议版本、账号、客户 IP、sequenceId、号码、srcId、编码、分片序号和内容长度；响应时记录 result、平台 messageId、CMPP Msg_Id、耗时和失败阶段。NestJS 拒绝 Submit 时，Gateway 日志必须保留 API 返回的真实业务原因，不能只记录 HTTP 状态码；短信正文不得明文写入 Gateway 日志，仅记录字符数和哈希。 +14. 客户已按 CMPP 标准 UDH 拆分的下游长短信必须先重组再进入业务发送链。Gateway 应识别 8 位 `05 00 03` 和 16 位 `06 08 04` 拼接头，去除 UDH 后按 `MsgFmt` 解码正文，并将引用号、总片数、片序号和编码传给 NestJS；NestJS 必须在 PostgreSQL 持久化分组和分片，支持乱序、同片幂等、冲突片拒绝、进程重启恢复和超时终止。分片未齐全前不得创建内部批次、`SmsMessageRecord`、计费或路由；齐全后只创建一条完整正文主记录，并使用第一片 `Sequence_Id` 建立客户消息映射。每个合法分片仍应分别取得一个 `CMPP_SUBMIT_RESP`，但不能把 UDH 字节作为正文、不能按分片生成多条短信记录。 #### 4.8.3 回执、上行与幂等 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 11e1c0e..d36248b 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -1011,6 +1011,21 @@ - 放入联通通道组的三网通道不能被移动号码选中。 - route/trace 标明实际命中的三网通道。 +### TC-SEND-016A 预发布运营商号段规则完整性 + +- 优先级：P0 +- 前置条件：运营商区分规则已按当前公开码号资料同步。 +- 步骤： + 1. 读取真实 `GET /api/admin/dictionaries/phone-carrier-rules?page=1&pageSize=100`。 + 2. 使用移动、联通、电信基础号段及 `162/165/167/170/171` 等移动转售号段生成代表号码。 + 3. 使用 `190/191/192/193/195/196/197/198/199` 生成代表号码。 + 4. 对每个代表号码按发送服务相同的优先级和 JavaScript 正则逐条匹配。 +- 预期结果： + - 每个代表号码唯一命中一条 active 规则，不得重叠或漏配。 + - `190/191/193/199` 归电信，`196` 归联通，`195/197/198` 归移动。 + - 中国广电 `192` 按当前业务约定唯一归入移动。 + - 测试结论仅代表原始号段分配规则，不把携号转网号码误宣称为实时运营商识别。 + ### TC-SEND-018 失败补发停止条件 - 优先级：P0 @@ -1205,6 +1220,25 @@ - CMPP 内部批次的详情、短信明细和取消请求均返回不可见/不存在，不泄露内部任务。 - CMPP 短信仍完整出现在短信记录、提交、回执和账务链路。 +### TC-GW-006A 下游 CMPP UDH 长短信持久化重组 + +- 优先级：P0 +- 前置条件：企业应用可正常 bind；已配置审核通过的完整签名和模板；NestJS、PostgreSQL、Redis 与 Go Gateway 使用真实本地或预发布链路。 +- 步骤： + 1. 使用 CMPP2.0 和 CMPP3.0 分别提交一条两片长短信，分片正文使用 UCS2，首片携带签名；覆盖 8 位 `05 00 03` 和 16 位 `06 08 04` UDH。 + 2. 第一片提交后查询 `CmppInboundLongMessage`、`CmppInboundLongMessageSegment`、`SmsBatchTask` 和 `SmsMessageRecord`。 + 3. 乱序提交第二片，再重复提交内容和 `Sequence_Id` 完全相同的分片。 + 4. 使用相同引用号和片序号提交内容不同或 `Sequence_Id` 不同的冲突片。 + 5. 在全部分片已持久化、业务处理尚未完成时重启 API，再重放任一已存分片。 + 6. 只提交部分分片并等待超过 `CMPP_INBOUND_LONG_MESSAGE_TTL_SECONDS`，执行超时扫描。 +- 预期结果： + - Gateway 去除 UDH 后才按 `MsgFmt` 解码，NestJS 收到的每片正文不含 `05 00 03`/`06 08 04` 控制字节；每个合法分片均返回 `SUBMIT_RESP status=0`。 + - 分片未齐全时只存在一条 collecting 分组和已收分片，API 返回稳定的分组 `messageId`，不得创建内部批次、短信主记录、计费或路由。 + - 分片齐全后按 `segmentIndex` 唯一排序拼接，只创建一条完整正文 `SmsMessageRecord` 和一个内部批次；正文签名/模板识别针对拼接后的完整内容执行，`cmppSubmitSequenceId` 使用第一片 `Sequence_Id`。 + - 乱序可完成；完全相同的重复片幂等复用原结果；冲突片返回明确 4xx/非零 SubmitResp，数据库唯一约束禁止同组同片序号出现两行。 + - API 重启后从 PostgreSQL 恢复分组、分片、原 `messageId` 和第一片 `Sequence_Id`，不得重复创建主记录。 + - 超时未齐分组转为 expired，保留分片审计但不创建短信主记录；后续相同引用号的新消息可建立新分组。 + ### TC-SEND-039 CMPP 模板不匹配短窗口聚合人工审核 - 优先级：P0 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 3006622..932667d 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -2244,3 +2244,21 @@ git diff --check - 发布后以字节级扫描复核49条签名，非法UTF-8为0；两个目标ID均精确恢复为hex `e38090e888aae5a4a9e4bfa1e681afe4bfa1e8afbae7bd91e38091`（`【航天信息信诺网】`）。生产Prisma真实执行签名、报备任务、报备记录、待报备资料四类关联查询均成功，分别返回49、34、82、4条，不再触发P2039/22021。 - API/Gateway health、外部首页、运营登录页、客户端登录页和外部API health均返回HTTP 200；Redis PONG，`gateway.submit.commands`为`pending=0、lag=0`，7个通道TPS权威配置存在。两个active上游通道均为`connected/currentConnections=1`；disabled通道的一条`connected/1`仍是2026-07-09历史状态残留，本次启动没有把它作为活动通道恢复。 - 部署后API日志新增P2039/22021为0，Nginx中企业签名及相关报备接口新增5xx为0。应用内Browser因Chrome标签被另一Codex会话占用且控制连接超时，未完成登录态页面交互验收；本次以真实NestJS所用Prisma关联查询、PostgreSQL字节扫描和Nginx/API日志作为后端修复证据，不虚报浏览器交互通过。未发送短信，未审核、删除、充值、改密或修改其他生产业务数据。 + +## 2026-07-23 预发布运营商区分规则同步（配置变更，未提交、未部署代码） + +- 在线证据以华为云2026年2月《消息&短信》号码规则为主，并以工信部关于`190/197/196/192`公众移动通信网网号核发信息交叉核对。规则覆盖三大基础运营商、移动转售号段及必要的上网卡/物联网/卫星前缀；按用户明确要求将中国广电`192`归入中国移动路由。 +- 变更前预发布`PhoneCarrierRule`有30条，存在`190`重复、移动`195`仅覆盖`1951—1952`、电信`191/193`等缺失。完整PostgreSQL及原规则CSV已备份至`/opt/cmpp-platform/backups/config/20260723-164357-phone-carrier-rules`；数据库备份SHA-256为`239c4c669d55e8ccf6bfc5458e92fab507e1af8cb928e7f4ace5cd7f53c1e6f1`，规则CSV SHA-256为`8e437138f0e1a526519b01c6ddc4e95ce58e1a76fb623dc2a3ab0f6616630e86`，两者权限600且gzip校验通过。 +- 使用单个PostgreSQL事务锁定并原子替换规则，最终30条全部active：移动12条、联通9条、电信9条。`19200000000`唯一命中`mobile / ^19[2578]`，备注明确“含中国广电192，按业务要求归中国移动”。 +- 真实管理员验证码登录成功（201），`GET /api/admin/dictionaries/phone-carrier-rules?page=1&pageSize=100`返回200和30条；对68个基础、转售及新号段代表号码按发送服务同款JavaScript正则验证，全部唯一命中、0个错配，随后真实登出成功。首次验证误把凭据文件的`password=unchanged`当作密码产生一次401，未锁定账号、未修改规则，修正为已授权密码后通过。 +- 本次只修改预发布字典配置，不改代码、不重启服务、不发送短信，也不回写历史短信运营商。号段规则反映原始码号分配；携号转网后的当前签约运营商无法只靠前缀判断，若业务要求实时识别需另接MNP/HLR能力。 + +## 2026-07-23 下游 CMPP UDH 长短信持久化重组修复（发布前验证） + +- 根因确认不是近期回归，而是既有下游入站链路从未实现重组：Gateway 将每个 CMPP Submit 的完整 `MsgContent`（包含 UDH）直接按 UCS2/GB18030 解码并逐片调用 NestJS，NestJS 因而把两片当成两条独立短信，控制字节污染首部签名匹配。此前台账通过的是“平台完整正文向上游拆分”和“上游 Deliver 长上行重组”，未覆盖“企业客户端已拆分的下游 Submit 重组”。 +- Gateway 新增标准 8 位 `05 00 03`、16 位 `06 08 04` UDH 解析，校验 `PkTotal/PkNumber` 与 UDH 总片数/片序号一致，先剥离 UDH 再按 `MsgFmt` 解码，并向 NestJS 传递引用号、总片数、片序号和编码。真实 CMPP2.0 TCP 回归确认两片均获得成功 SubmitResp，API 收到的片正文不含 UDH。 +- NestJS/Prisma 新增 `CmppInboundLongMessage`、`CmppInboundLongMessageSegment` 及 migration `20260723120000_add_cmpp_inbound_long_message_reassembly`。分组键包含应用、账号、Src_Id、目标号码、引用号、总片数和编码；使用 PostgreSQL advisory transaction lock 与同组同片唯一索引保证并发幂等。分片齐全前不创建批次/短信，齐全后按片序合并并只创建一条完整正文记录；持久化稳定 `messageId` 和第一片 `Sequence_Id`，支持乱序、重复片、冲突拒绝、进程重启恢复及超时转 expired。 +- 本地真实 PostgreSQL 16 已应用 62 条 migration，schema 最新。事务验证成功写入2片并按序拼成 `【测试】第一片第二片正文`，同组同片重复索引被唯一约束拒绝，验证事务最终回滚为0条残留。真实本地 NestJS API + PostgreSQL + Redis 调用两次入站接口后，分组为 completed、持久化2片、只创建1条主记录，完整正文和第一片 `Sequence_Id` 均正确；未启动 Go Gateway 上游连接，未发送真实短信。 +- 新增长短信相关 API 回归5项（合并、乱序/重复/冲突、处理中断恢复、主记录已落库后的幂等恢复、超时终止）和 Gateway 回归3项（8位UDH、16位UDH、真实CMPP2.0两片转发）。API发送链83/83、API全量24 suites/295项、Gateway `go test ./...`、API build、前端build、Prisma generate/validate/status均通过。`verify:phase8`首次与本地API并发时BullMQ为438.93 TPS而失败；关闭仅由本轮启动的API后单测为909.56 TPS，完整重跑为872.43 TPS并通过。前端仅保留既有约1.92MB单chunk告警。 +- 预发布两次测试正文使用的 `【深圳市合正物业服务有限公司】` 在该应用签名库中不存在；本次修复能消除UDH污染并完整重组，但不会绕过签名审核。部署后复测前需先按正常产品流程为应用配置并审核该签名/模板，或改用应用已有的审核通过签名。数据库升级只新增两张重组表和外键/索引；如必须回滚，应先停止新版本 API/Gateway，再删除子表和父表，未完成分片审计会丢失，既有短信主记录不受影响。 +- 发布前功能代码、迁移、回归测试和文档已完成并获用户授权提交、推送和部署；本节先保留发布前验证证据，实际提交、备份、migration、服务重启和发布后验收结果在部署完成后追加记录。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index a378e2f..d7af3e9 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -44,14 +44,22 @@ type authRequest struct { } type submitRequest struct { - Account string `json:"account"` - PhoneNumber string `json:"phoneNumber,omitempty"` - PhoneNumbers []string `json:"phoneNumbers,omitempty"` - Content string `json:"content"` - SrcID string `json:"srcId,omitempty"` - DestID string `json:"destId,omitempty"` - SequenceID uint32 `json:"sequenceId,omitempty"` - RemoteIP string `json:"remoteIp,omitempty"` + Account string `json:"account"` + PhoneNumber string `json:"phoneNumber,omitempty"` + PhoneNumbers []string `json:"phoneNumbers,omitempty"` + Content string `json:"content"` + SrcID string `json:"srcId,omitempty"` + DestID string `json:"destId,omitempty"` + SequenceID uint32 `json:"sequenceId,omitempty"` + RemoteIP string `json:"remoteIp,omitempty"` + LongMessage *inboundLongMessageFragment `json:"longMessage,omitempty"` +} + +type inboundLongMessageFragment struct { + Reference int `json:"reference"` + Total int `json:"total"` + Index int `json:"index"` + Format int `json:"format"` } type submitResponseMessage struct { @@ -291,7 +299,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge clientProtocol, req.protocol, account, enterpriseCode, remote, req.sequenceID, phone, strings.TrimSpace(req.srcID), req.msgFmt, req.pkNumber, req.pkTotal, len(req.destTerminalIDs), len(req.msgContent), ) - content, err := decodeContent(req.msgFmt, req.msgContent) + content, longMessage, err := decodeInboundSubmitContent(req) if err != nil { logger.Printf( "cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=decode reason=%q", @@ -311,6 +319,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge DestID: phone, SequenceID: req.sequenceID, RemoteIP: remoteIP(remote), + LongMessage: longMessage, }) if err != nil || !result.Accepted { reason := "api returned accepted=false" @@ -465,6 +474,7 @@ type inboundSubmitPacket struct { protocol string pkTotal uint8 pkNumber uint8 + tpUdhi uint8 msgFmt uint8 msgSrc string srcID string @@ -477,13 +487,13 @@ func normalizeInboundSubmit(packet any) (inboundSubmitPacket, bool) { switch req := packet.(type) { case *cmpp.Cmpp2SubmitReqPkt: return inboundSubmitPacket{ - protocol: "cmpp20", pkTotal: req.PkTotal, pkNumber: req.PkNumber, msgFmt: req.MsgFmt, + protocol: "cmpp20", pkTotal: req.PkTotal, pkNumber: req.PkNumber, tpUdhi: req.TpUdhi, msgFmt: req.MsgFmt, msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId, msgContent: req.MsgContent, sequenceID: req.SeqId, }, true case *cmpp.Cmpp3SubmitReqPkt: return inboundSubmitPacket{ - protocol: "cmpp30", pkTotal: req.PkTotal, pkNumber: req.PkNumber, msgFmt: req.MsgFmt, + protocol: "cmpp30", pkTotal: req.PkTotal, pkNumber: req.PkNumber, tpUdhi: req.TpUdhi, msgFmt: req.MsgFmt, msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId, msgContent: req.MsgContent, sequenceID: req.SeqId, }, true @@ -697,6 +707,60 @@ func decodeContent(format uint8, content string) (string, error) { } } +func decodeInboundSubmitContent(req inboundSubmitPacket) (string, *inboundLongMessageFragment, error) { + raw := []byte(req.msgContent) + if req.tpUdhi == 0 && req.pkTotal <= 1 { + content, err := decodeContent(req.msgFmt, req.msgContent) + return content, nil, err + } + if len(raw) == 0 { + return "", nil, errors.New("UDH message content is empty") + } + + headerLength := int(raw[0]) + 1 + if headerLength > len(raw) { + return "", nil, fmt.Errorf("UDH length %d exceeds message content length %d", headerLength, len(raw)) + } + + var reference, total, index int + switch { + case len(raw) >= 6 && raw[0] == 0x05 && raw[1] == 0x00 && raw[2] == 0x03: + reference = int(raw[3]) + total = int(raw[4]) + index = int(raw[5]) + case len(raw) >= 7 && raw[0] == 0x06 && raw[1] == 0x08 && raw[2] == 0x04: + reference = int(raw[3])<<8 | int(raw[4]) + total = int(raw[5]) + index = int(raw[6]) + default: + if req.pkTotal > 1 { + return "", nil, errors.New("concatenated CMPP submit is missing a supported 8-bit or 16-bit UDH") + } + content, err := decodeContent(req.msgFmt, string(raw[headerLength:])) + return content, nil, err + } + if total < 2 || index < 1 || index > total { + return "", nil, fmt.Errorf("invalid concatenated UDH total/index %d/%d", index, total) + } + if req.pkTotal > 0 && int(req.pkTotal) != total { + return "", nil, fmt.Errorf("PkTotal %d does not match UDH total %d", req.pkTotal, total) + } + if req.pkNumber > 0 && int(req.pkNumber) != index { + return "", nil, fmt.Errorf("PkNumber %d does not match UDH index %d", req.pkNumber, index) + } + + content, err := decodeContent(req.msgFmt, string(raw[headerLength:])) + if err != nil { + return "", nil, err + } + return content, &inboundLongMessageFragment{ + Reference: reference, + Total: total, + Index: index, + Format: int(req.msgFmt), + }, nil +} + func apiBaseURL(value string) string { if value == "" { return "http://127.0.0.1:3000/api" @@ -733,6 +797,13 @@ func rememberDownstream(session downstreamSession) { } session.touchPresence("connected", true, false) downstreamRegistry.Lock() + if existing := downstreamRegistry.byMessageID[session.messageID]; existing != nil && existing.conn == session.conn { + // A downstream long message returns one SUBMIT_RESP per fragment but is + // persisted as one platform message. Keep the first fragment Msg_Id so + // online delivery and restart recovery (which persists the first + // Sequence_Id) address the same client-side message. + session.gatewayMsgID = existing.gatewayMsgID + } downstreamRegistry.byMessageID[session.messageID] = &session downstreamRegistry.byConn[session.conn] = &session if session.account != "" { diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index a7049ba..36d7fc5 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -235,6 +235,144 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { } } +func TestDecodeInboundLongMessageStripsConcatUDHBeforeUCS2Decode(t *testing.T) { + payload, err := cmpputils.Utf8ToUcs2("【深圳市合正物业服务有限公司】第一片正文") + if err != nil { + t.Fatalf("encode content: %v", err) + } + raw := append([]byte{0x05, 0x00, 0x03, 0x10, 0x02, 0x01}, []byte(payload)...) + + content, fragment, err := decodeInboundSubmitContent(inboundSubmitPacket{ + pkTotal: 2, + pkNumber: 1, + tpUdhi: 1, + msgFmt: 8, + msgContent: string(raw), + }) + if err != nil { + t.Fatalf("decode long message: %v", err) + } + if content != "【深圳市合正物业服务有限公司】第一片正文" { + t.Fatalf("decoded content = %q", content) + } + if fragment == nil || fragment.Reference != 0x10 || fragment.Total != 2 || fragment.Index != 1 { + t.Fatalf("unexpected fragment metadata: %+v", fragment) + } +} + +func TestDecodeInboundLongMessageSupports16BitConcatReference(t *testing.T) { + payload, err := cmpputils.Utf8ToUcs2("第二片正文") + if err != nil { + t.Fatalf("encode content: %v", err) + } + raw := append([]byte{0x06, 0x08, 0x04, 0x12, 0x34, 0x02, 0x02}, []byte(payload)...) + + content, fragment, err := decodeInboundSubmitContent(inboundSubmitPacket{ + pkTotal: 2, + pkNumber: 2, + tpUdhi: 1, + msgFmt: 8, + msgContent: string(raw), + }) + if err != nil { + t.Fatalf("decode long message: %v", err) + } + if content != "第二片正文" { + t.Fatalf("decoded content = %q", content) + } + if fragment == nil || fragment.Reference != 0x1234 || fragment.Total != 2 || fragment.Index != 2 { + t.Fatalf("unexpected fragment metadata: %+v", fragment) + } +} + +func TestInboundServerForwardsLongMessageFragmentsWithoutUDHAndAcknowledgesEachSubmit(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + account := "100001" + password := "secret-hash" + var mu sync.Mutex + submits := make([]submitRequest, 0, 2) + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/gateway/events/inbound/authenticate": + _ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account}) + case "/api/gateway/events/inbound/submit": + var submit submitRequest + if err := json.NewDecoder(r.Body).Decode(&submit); err != nil { + t.Fatalf("decode submit: %v", err) + } + mu.Lock() + submits = append(submits, submit) + count := len(submits) + mu.Unlock() + status := "fragment_pending" + if count == 2 { + status = "accepted" + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "accepted": true, "messageId": "MSG-LONG-1", "status": status, + }) + case "/api/gateway/events/downstream/pending": + _ = json.NewEncoder(w).Encode([]pendingDelivery{}) + case "/api/gateway/events/inbound/connection": + w.WriteHeader(http.StatusOK) + default: + t.Fatalf("unexpected api path: %s", r.URL.Path) + } + })) + defer api.Close() + + addr := reserveTCPAddr(t) + go func() { _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe() }() + time.Sleep(300 * time.Millisecond) + + client := cmpp.NewClient(cmpp.V20) + defer client.Disconnect() + if err := client.Connect(addr, account, password, 2*time.Second); err != nil { + t.Fatalf("connect CMPP2 inbound: %v", err) + } + parts := []string{"【签名】第一片", "第二片正文"} + var firstResponseMsgID uint64 + for index, text := range parts { + payload, _ := cmpputils.Utf8ToUcs2(text) + raw := append([]byte{0x05, 0x00, 0x03, 0x22, 0x02, byte(index + 1)}, []byte(payload)...) + if _, err := client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{ + PkTotal: 2, PkNumber: uint8(index + 1), TpUdhi: 1, RegisteredDelivery: 1, MsgLevel: 1, + ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696", + MsgFmt: 8, MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000", + DestUsrTl: 1, DestTerminalId: []string{"13500002696"}, MsgLength: uint8(len(raw)), MsgContent: string(raw), + }); err != nil { + t.Fatalf("send long-message fragment %d: %v", index+1, err) + } + if rsp := recvSubmitRsp20(t, client); rsp.Result != 0 || rsp.MsgId == 0 { + t.Fatalf("unexpected fragment %d response: %+v", index+1, rsp) + } else if index == 0 { + firstResponseMsgID = rsp.MsgId + } + } + + mu.Lock() + defer mu.Unlock() + if len(submits) != 2 { + t.Fatalf("submit API calls = %d, want 2", len(submits)) + } + for index, submit := range submits { + if submit.Content != parts[index] { + t.Fatalf("fragment %d content = %q", index+1, submit.Content) + } + if submit.LongMessage == nil || submit.LongMessage.Reference != 0x22 || + submit.LongMessage.Total != 2 || submit.LongMessage.Index != index+1 || submit.LongMessage.Format != 8 { + t.Fatalf("fragment %d metadata = %+v", index+1, submit.LongMessage) + } + } + downstreamRegistry.RLock() + session := downstreamRegistry.byMessageID["MSG-LONG-1"] + downstreamRegistry.RUnlock() + if session == nil || session.gatewayMsgID != firstResponseMsgID { + t.Fatalf("stored long-message Msg_Id = %v, want first fragment Msg_Id %v", session, firstResponseMsgID) + } +} + func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry()