From ca12f14b0007ee75f727d18e66791a669838b66e Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Fri, 24 Jul 2026 12:52:06 +0800 Subject: [PATCH] fix: reconcile shared-channel receipts and protocol logs --- .../gateway-events.controller.spec.ts | 119 +++++++++ .../send-chain/gateway-events.controller.ts | 38 ++- api/src/send-chain/send-chain.service.spec.ts | 180 ++++++++++++++ api/src/send-chain/send-chain.service.ts | 227 ++++++++++++++++-- .../first-version-development-requirements.md | 8 +- docs/system-functional-test-cases.md | 9 +- docs/testing-progress.md | 20 ++ gateway/internal/inbound/protocol_log_test.go | 52 ++++ gateway/internal/inbound/server.go | 100 +++++++- gateway/internal/inbound/server_test.go | 10 + gateway/internal/upstream/manager.go | 133 +++++++++- .../internal/upstream/protocol_log_test.go | 54 +++++ src/apps/admin/AdminSystemLogsPage.tsx | 34 ++- .../smoke/receipt-cross-connection-smoke.mjs | 169 +++++++++++++ 14 files changed, 1110 insertions(+), 43 deletions(-) create mode 100644 api/src/send-chain/gateway-events.controller.spec.ts create mode 100644 gateway/internal/inbound/protocol_log_test.go create mode 100644 gateway/internal/upstream/protocol_log_test.go create mode 100644 tools/smoke/receipt-cross-connection-smoke.mjs diff --git a/api/src/send-chain/gateway-events.controller.spec.ts b/api/src/send-chain/gateway-events.controller.spec.ts new file mode 100644 index 0000000..fb1dede --- /dev/null +++ b/api/src/send-chain/gateway-events.controller.spec.ts @@ -0,0 +1,119 @@ +import { BadRequestException } from '@nestjs/common'; +import { GatewayEventsController } from './gateway-events.controller'; + +describe('GatewayEventsController protocol logging', () => { + const sendChain = { + handleSubmitResult: jest.fn(), + handleReceipt: jest.fn(), + }; + const protocolLogs = { + record: jest.fn(), + }; + const controller = new GatewayEventsController(sendChain as never, {} as never, protocolLogs as never); + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('does not duplicate a supplier SubmitResp from the internal aggregate callback', async () => { + sendChain.handleSubmitResult.mockResolvedValue({ accepted: true }); + const body = { + messageId: 'MSG-1', + channelId: 'channel-1', + gatewayMessageId: '123', + sequenceId: 7, + submitStatus: 'accepted' as const, + }; + + await controller.submitResult(body); + + expect(protocolLogs.record).not.toHaveBeenCalled(); + }); + + it('accepts only safe outbound Gateway packet events', () => { + expect(controller.protocolLog({ + protocol: 'cmpp', + direction: 'platform_to_channel', + eventType: 'submit', + status: 'success', + messageId: 'MSG-1', + })).toEqual({ accepted: true }); + expect(protocolLogs.record).toHaveBeenCalledTimes(1); + + expect(() => controller.protocolLog({ + protocol: 'cmpp', + direction: 'client_to_platform', + eventType: 'submit', + status: 'success', + })).toThrow(BadRequestException); + + expect(controller.protocolLog({ + protocol: 'cmpp', + direction: 'channel_to_platform', + eventType: 'submit_resp', + status: 'success', + messageId: 'MSG-1', + })).toEqual({ accepted: true }); + expect(controller.protocolLog({ + protocol: 'cmpp', + direction: 'platform_to_client', + eventType: 'submit_resp', + status: 'success', + messageId: 'MSG-1', + })).toEqual({ accepted: true }); + }); + + it('enriches an enterprise Submit packet with identifiers returned by the real service', async () => { + (sendChain as Record).submitInboundMessage = jest.fn().mockResolvedValue({ + accepted: true, + tenantId: 'tenant-1', + applicationId: 'app-1', + messageId: 'MSG-1', + status: 'fragment_pending', + }); + + await controller.submitInbound({ + account: '607532', + phoneNumber: '13127620092', + content: 'fragment', + sequenceId: 141, + }); + + expect(protocolLogs.record).toHaveBeenCalledWith(expect.objectContaining({ + direction: 'client_to_platform', + eventType: 'submit', + tenantId: 'tenant-1', + applicationId: 'app-1', + messageId: 'MSG-1', + status: 'success', + })); + }); + + it('replaces a fallback receipt identifier with the resolved main message identifier', async () => { + sendChain.handleReceipt.mockResolvedValue({ + tenantId: 'tenant-1', + applicationId: 'app-1', + messageId: 'MSG-LONG-1', + }); + + await controller.receipt({ + messageId: 'receipt-736070230367350788', + channelId: 'channel-copy', + gatewayMessageId: '736070230367350788', + phoneNumber: '13127620092', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + }); + + expect(protocolLogs.record).toHaveBeenCalledWith(expect.objectContaining({ + direction: 'channel_to_platform', + eventType: 'deliver_receipt', + channelId: 'channel-copy', + messageId: 'MSG-LONG-1', + gatewayMessageId: '736070230367350788', + tenantId: 'tenant-1', + applicationId: 'app-1', + status: 'success', + })); + }); +}); diff --git a/api/src/send-chain/gateway-events.controller.ts b/api/src/send-chain/gateway-events.controller.ts index deb8f52..76d8bfc 100644 --- a/api/src/send-chain/gateway-events.controller.ts +++ b/api/src/send-chain/gateway-events.controller.ts @@ -1,4 +1,4 @@ -import { Body, Controller, Post } from '@nestjs/common'; +import { BadRequestException, Body, Controller, Post } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { GatewayInboundAuthDto, @@ -28,7 +28,7 @@ export class GatewayEventsController { @Post('submit-result') submitResult(@Body() body: GatewaySubmitResultDto) { - return this.trackGatewayEvent('submit_resp', body, () => this.sendChain.handleSubmitResult(body)); + return this.sendChain.handleSubmitResult(body); } @Post('receipt') @@ -41,6 +41,25 @@ export class GatewayEventsController { return this.trackGatewayEvent('deliver_uplink', body, () => this.sendChain.handleUplink(body)); } + @Post('protocol-log') + protocolLog(@Body() body: ProtocolLogInput) { + const allowedPacket = ( + body.direction === 'platform_to_channel' + && ['submit', 'deliver_resp'].includes(body.eventType) + ) || ( + body.direction === 'channel_to_platform' + && body.eventType === 'submit_resp' + ) || ( + body.direction === 'platform_to_client' + && body.eventType === 'submit_resp' + ); + if (body.protocol !== 'cmpp' || !allowedPacket || !['success', 'failed'].includes(body.status)) { + throw new BadRequestException('Unsupported Gateway protocol log event'); + } + this.protocolLogs.record(body); + return { accepted: true }; + } + @Post('dead-letter') deadLetter(@Body() body: GatewaySubmitDeadLetterDto) { return this.sendChain.recordGatewaySubmitDeadLetter(body); @@ -110,14 +129,25 @@ export class GatewayEventsController { messageId: (value.messageId ?? value.platformMessageId) as string, gatewayMessageId: (value.gatewayMessageId ?? value.msgId ?? value.upstreamMessageId) as string, phone: (value.phoneNumber ?? value.srcTerminalId ?? value.destinationId) as string, - resultCode: (value.result ?? value.status ?? value.stat) as string, + resultCode: (value.result ?? value.submitStatus ?? value.rawStatus ?? value.status ?? value.stat) as string, detail: { sequenceId: value.sequenceId, connectionId: value.connectionId }, }; - this.protocolLogs.record({ ...common, status: 'received', durationMs: 0 }); try { const result = await action(); + const resultValue = result && typeof result === 'object' + ? result as Record + : {}; this.protocolLogs.record({ ...common, + tenantId: (resultValue.tenantId ?? common.tenantId) as string, + applicationId: (resultValue.applicationId ?? common.applicationId) as string, + channelId: common.channelId ?? resultValue.channelId as string, + account: common.account ?? resultValue.account as string, + messageId: (resultValue.messageId ?? common.messageId) as string, + gatewayMessageId: common.gatewayMessageId + ?? (resultValue.gatewayMessageId ?? resultValue.msgId) as string, + resultCode: common.resultCode + ?? (resultValue.result ?? resultValue.status) as string, status: 'success', durationMs: Date.now() - startedAt, }); diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 003ce3a..632634b 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -160,6 +160,7 @@ function createPrismaMock() { upsert: jest.fn().mockResolvedValue({ id: 'segment-1' }), updateMany: jest.fn().mockResolvedValue({ count: 1 }), findFirst: jest.fn().mockResolvedValue(null), + findMany: jest.fn().mockResolvedValue([]), }, cmppInboundLongMessage: { create: jest.fn(), @@ -1917,6 +1918,7 @@ describe('SendChainService', () => { const { service, prisma } = createService(); prisma.smsMessageRecord.findUnique.mockResolvedValue(null); prisma.smsSubmitRecord.findMany + .mockResolvedValueOnce([]) .mockResolvedValueOnce([]) .mockResolvedValueOnce([ { @@ -2020,6 +2022,183 @@ describe('SendChainService', () => { }); }); + it('matches a receipt from another connection only when it is the unique channel of the same supplier', async () => { + const { service, prisma } = createService(); + prisma.smsMessageRecord.findUnique.mockResolvedValue(null); + prisma.smsChannel.findUnique.mockResolvedValue({ + id: 'channel-copy', + account: 'C59748', + gatewayHost: 'supplier.example.com', + gatewayPort: 7890, + protocol: 'CMPP', + cmppVersion: '2.0', + }); + prisma.smsSubmitRecord.findMany + .mockResolvedValueOnce([]); + prisma.smsMessageSegmentAudit.findMany + .mockResolvedValueOnce([ + { + id: 'segment-2', + submitId: 'SUB-LONG-1', + submitRecordId: 'submit-original', + channelId: 'channel-original', + gatewayMessageId: '736070230367350788', + submitRecord: { id: 'submit-original', submitId: 'SUB-LONG-1' }, + channel: { + id: 'channel-original', + account: 'C59748', + gatewayHost: 'supplier.example.com', + gatewayPort: 7890, + protocol: 'CMPP', + cmppVersion: '2.0', + }, + messageRecord: { + id: 'record-long', + tenantId: 'tenant-1', + batchTaskId: 'task-1', + applicationId: 'app-1', + messageId: 'MSG-LONG-1', + submitId: 'SUB-LONG-1', + phoneNumber: '13127620092', + channelId: 'channel-original', + gatewayMessageId: '736070227905294338', + status: 'submitted', + billingUnits: 2, + }, + }, + ]) + .mockResolvedValueOnce([]); + + await service.handleReceipt({ + messageId: 'receipt-736070230367350788', + channelId: 'channel-copy', + gatewayMessageId: '736070230367350788', + phoneNumber: '13127620092', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + }); + + expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + channelId: 'channel-original', + messageRecordId: 'record-long', + messageId: 'MSG-LONG-1', + }), + }); + }); + + it('does not match the same Msg_Id across channels belonging to different suppliers', async () => { + const { service, prisma } = createService(); + prisma.smsMessageRecord.findUnique.mockResolvedValue(null); + prisma.smsChannel.findUnique.mockResolvedValue({ + id: 'channel-other', + account: 'OTHER', + gatewayHost: 'other.example.com', + gatewayPort: 7890, + protocol: 'CMPP', + cmppVersion: '2.0', + }); + prisma.smsSubmitRecord.findMany + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]); + prisma.smsMessageSegmentAudit.findMany.mockResolvedValueOnce([ + { + id: 'segment-original', + submitRecordId: 'submit-original', + submitId: 'SUB-ORIGINAL', + channelId: 'channel-original', + gatewayMessageId: 'SHARED-ID', + submitRecord: { id: 'submit-original', submitId: 'SUB-ORIGINAL' }, + channel: { + id: 'channel-original', + account: 'C59748', + gatewayHost: 'supplier.example.com', + gatewayPort: 7890, + protocol: 'CMPP', + cmppVersion: '2.0', + }, + messageRecord: { + id: 'record-original', + messageId: 'MSG-ORIGINAL', + phoneNumber: '13127620092', + }, + }, + ]); + + await expect(service.handleReceipt({ + messageId: 'receipt-SHARED-ID', + channelId: 'channel-other', + gatewayMessageId: 'SHARED-ID', + phoneNumber: '13127620092', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + })).rejects.toThrow('SMS message record not found'); + + expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled(); + }); + + it('waits for every long-message segment before marking the main message delivered', async () => { + const { service, prisma } = createService(); + prisma.smsMessageRecord.findUnique.mockResolvedValue({ + id: 'record-long', + tenantId: 'tenant-1', + batchTaskId: 'task-1', + applicationId: 'app-1', + messageId: 'MSG-LONG-1', + submitId: 'SUB-LONG-1', + phoneNumber: '13127620092', + channelId: 'channel-1', + gatewayMessageId: 'GW-SEG-1', + status: 'submitted', + billingUnits: 2, + }); + prisma.smsSubmitRecord.findFirst.mockResolvedValue({ + id: 'submit-long', + submitId: 'SUB-LONG-1', + channelId: 'channel-1', + gatewayMessageId: 'GW-SEG-1', + }); + prisma.smsMessageSegmentAudit.findMany + .mockResolvedValueOnce([ + { segmentIndex: 1, segmentTotal: 2, receiptStatus: 'delivered', rawStatus: 'DELIVRD', deliveredAt: new Date() }, + { segmentIndex: 2, segmentTotal: 2, receiptStatus: null, rawStatus: null, deliveredAt: null }, + ]) + .mockResolvedValueOnce([ + { segmentIndex: 1, segmentTotal: 2, receiptStatus: 'delivered', rawStatus: 'DELIVRD', deliveredAt: new Date() }, + { segmentIndex: 2, segmentTotal: 2, receiptStatus: 'delivered', rawStatus: 'DELIVRD', deliveredAt: new Date() }, + ]); + + await service.handleReceipt({ + messageId: 'MSG-LONG-1', + channelId: 'channel-1', + gatewayMessageId: 'GW-SEG-1', + phoneNumber: '13127620092', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + }); + + expect(prisma.smsMessageRecord.update).not.toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ status: 'delivered' }), + })); + expect(prisma.cmppDownstreamDelivery.create).not.toHaveBeenCalled(); + + prisma.smsReceiptRecord.findUnique.mockResolvedValue(null); + await service.handleReceipt({ + messageId: 'MSG-LONG-1', + channelId: 'channel-1', + gatewayMessageId: 'GW-SEG-2', + phoneNumber: '13127620092', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + }); + + expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith(expect.objectContaining({ + where: { id: 'record-long' }, + data: expect.objectContaining({ status: 'delivered', receiptStatus: 'delivered' }), + })); + expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(1); + }); + it('treats a repeated DELIVRD event as idempotent and does not redeliver it downstream', async () => { const { service, prisma } = createService(); prisma.smsReceiptRecord.findUnique @@ -2050,6 +2229,7 @@ describe('SendChainService', () => { const { service, prisma } = createService(); prisma.smsMessageRecord.findUnique.mockResolvedValue(null); prisma.smsSubmitRecord.findMany + .mockResolvedValueOnce([]) .mockResolvedValueOnce([]) .mockResolvedValueOnce([ { diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 1db9397..e99fa92 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -982,7 +982,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } async handleReceipt(data: GatewayReceiptEventDto) { - const receiptKey = this.receiptEventKey(data); + const resolved = await this.resolveReceiptMessage(data); + const logicalChannelId = resolved.channelId ?? data.channelId; + const receiptKey = this.receiptEventKey(data, logicalChannelId); const existingReceipt = await this.prisma.smsReceiptRecord.findUnique({ where: { receiptKey }, include: { messageRecord: true }, @@ -990,11 +992,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { 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(); - const status = - data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed'; if (resolved.submitRecordId) { await this.prisma.smsSubmitRecord.updateMany({ where: { @@ -1014,7 +1013,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { batchTaskId: message.batchTaskId, messageRecordId: message.id, receiptKey, - channelId: data.channelId, + channelId: logicalChannelId, messageId: resolved.messageId, gatewayMessageId: data.gatewayMessageId, phoneNumber: data.phoneNumber?.trim() || message.phoneNumber, @@ -1036,10 +1035,26 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } throw error; } - await this.recordReceiptSegment(message, data, deliveredAt, resolved.submitRecordId); + const logicalReceipt = { ...data, channelId: logicalChannelId }; + await this.recordReceiptSegment(message, logicalReceipt, deliveredAt, resolved.submitRecordId); + const aggregate = await this.aggregateReceiptSegments( + message, + logicalReceipt, + deliveredAt, + resolved.submitRecordId, + resolved.submitId, + ); + if (!aggregate.terminal) { + return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); + } + const status = aggregate.status; const isCurrentAttempt = - (!message.channelId || message.channelId === data.channelId) - && (!message.gatewayMessageId || message.gatewayMessageId === data.gatewayMessageId); + (!message.channelId || message.channelId === logicalChannelId) + && ( + !message.gatewayMessageId + || message.gatewayMessageId === data.gatewayMessageId + || (aggregate.segmentTotal > 1 && (!message.submitId || message.submitId === resolved.submitId)) + ); if (!isCurrentAttempt || (status === 'failed' && message.status === 'delivered')) { return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } @@ -1056,14 +1071,14 @@ 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, + channelId: logicalChannelId, + gatewayMessageId: message.gatewayMessageId ?? data.gatewayMessageId, + receiptStatus: aggregate.receiptStatus, + receiptRawStatus: aggregate.rawStatus, status, - errorCode: data.errorCode, - errorMessage: data.errorMessage ?? (status === 'delivered' ? null : data.rawStatus), - deliveredAt, + errorCode: aggregate.errorCode, + errorMessage: aggregate.errorMessage ?? (status === 'delivered' ? null : aggregate.rawStatus), + deliveredAt: aggregate.deliveredAt, }, }); if (!isStandaloneChannelTest && message.tenantId && message.applicationId) { @@ -1077,12 +1092,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { messageId: message.messageId, gatewayMessageId: data.gatewayMessageId, phoneNumber: message.phoneNumber, - receiptStatus: data.receiptStatus, - rawStatus: data.rawStatus, - errorCode: data.errorCode, + receiptStatus: aggregate.receiptStatus, + rawStatus: aggregate.rawStatus, + errorCode: aggregate.errorCode, submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined, submitGroupMessageId: message.cmppSubmitGroupMessageId ?? undefined, - deliveredAt: deliveredAt.toISOString(), + deliveredAt: aggregate.deliveredAt.toISOString(), }, }); } @@ -2071,6 +2086,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (!collection.complete) { return { accepted: true, + tenantId: application.tenantId, + applicationId: application.id, messageId: collection.messageId, status: 'fragment_pending', fragmentPending: true, @@ -3522,6 +3539,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { upsert: (args: Record) => Promise; updateMany: (args: Record) => Promise<{ count: number }>; findFirst: (args: Record) => Promise; + findMany: (args: Record) => Promise; }; }).smsMessageSegmentAudit; } @@ -3690,6 +3708,101 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }); } + private async aggregateReceiptSegments( + message: { + id: string; + billingUnits?: number | null; + }, + data: GatewayReceiptEventDto, + deliveredAt: Date, + submitRecordId?: string, + submitId?: string, + ) { + const audits = await this.smsMessageSegmentAuditDelegate().findMany({ + where: submitRecordId + ? { messageRecordId: message.id, submitRecordId } + : submitId + ? { messageRecordId: message.id, submitId } + : { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId }, + orderBy: { segmentIndex: 'asc' }, + }); + if (audits.length === 0) { + const status = data.receiptStatus === 'delivered' + ? 'delivered' + : data.receiptStatus === 'unknown' + ? 'unknown' + : 'failed'; + return { + terminal: true, + segmentTotal: 1, + status, + receiptStatus: data.receiptStatus, + rawStatus: data.rawStatus, + errorCode: data.errorCode, + errorMessage: data.errorMessage, + deliveredAt, + }; + } + + const segmentTotal = Math.max( + 1, + Number(message.billingUnits ?? 1), + ...audits.map((audit) => Number(audit.segmentTotal ?? 1)), + ); + const received = audits.filter((audit) => Boolean(audit.receiptStatus)); + const failed = received.find((audit) => !['delivered', 'unknown'].includes(audit.receiptStatus ?? '')); + if (failed) { + return { + terminal: true, + segmentTotal, + status: 'failed', + receiptStatus: failed.receiptStatus ?? 'undelivered', + rawStatus: failed.rawStatus ?? data.rawStatus, + errorCode: failed.errorCode ?? data.errorCode, + errorMessage: failed.errorMessage ?? data.errorMessage, + deliveredAt: failed.deliveredAt ?? deliveredAt, + }; + } + const delivered = received.filter((audit) => audit.receiptStatus === 'delivered'); + if (delivered.length >= segmentTotal) { + const latest = delivered.reduce((current, audit) => + (audit.deliveredAt?.getTime() ?? 0) > (current.deliveredAt?.getTime() ?? 0) ? audit : current); + return { + terminal: true, + segmentTotal, + status: 'delivered', + receiptStatus: 'delivered', + rawStatus: latest.rawStatus ?? data.rawStatus, + errorCode: latest.errorCode ?? undefined, + errorMessage: undefined, + deliveredAt: latest.deliveredAt ?? deliveredAt, + }; + } + if (received.length >= segmentTotal) { + const latest = received[received.length - 1]; + return { + terminal: true, + segmentTotal, + status: 'unknown', + receiptStatus: 'unknown', + rawStatus: latest.rawStatus ?? data.rawStatus, + errorCode: latest.errorCode ?? data.errorCode, + errorMessage: latest.errorMessage ?? data.errorMessage, + deliveredAt: latest.deliveredAt ?? deliveredAt, + }; + } + return { + terminal: false, + segmentTotal, + status: 'submitted', + receiptStatus: data.receiptStatus, + rawStatus: data.rawStatus, + errorCode: data.errorCode, + errorMessage: data.errorMessage, + deliveredAt, + }; + } + private async findMessageByGatewayEvent(messageId?: string, gatewayMessageId?: string) { const conditions = [{ messageId }, gatewayMessageId ? { gatewayMessageId } : undefined].filter( Boolean, @@ -3732,6 +3845,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { message: exactMessage, messageId: exactMessage.messageId, submitRecordId: submitRecord?.id, + submitId: submitRecord?.submitId, + channelId: submitRecord?.channelId ?? data.channelId, }; } @@ -3751,6 +3866,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { message: exactSubmits[0].messageRecord, messageId: exactSubmits[0].messageRecord.messageId, submitRecordId: exactSubmits[0].id, + submitId: exactSubmits[0].submitId, + channelId: exactSubmits[0].channelId, }; } @@ -3758,6 +3875,61 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { throw new NotFoundException('SMS message record not found'); } + const incomingChannel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } }); + if (!incomingChannel) { + throw new NotFoundException('SMS message record not found'); + } + const segmentMatches = await this.smsMessageSegmentAuditDelegate().findMany({ + where: { + gatewayMessageId: data.gatewayMessageId, + messageRecord: { phoneNumber }, + }, + include: { messageRecord: true, submitRecord: true, channel: true }, + orderBy: { createdAt: 'desc' }, + take: 10, + }); + const exactSegmentMatches = segmentMatches.filter((candidate) => candidate.channelId === data.channelId); + if (exactSegmentMatches.length === 1 && exactSegmentMatches[0]?.messageRecord) { + return { + message: exactSegmentMatches[0].messageRecord, + messageId: exactSegmentMatches[0].messageRecord.messageId, + submitRecordId: exactSegmentMatches[0].submitRecordId ?? undefined, + submitId: exactSegmentMatches[0].submitRecord?.submitId ?? exactSegmentMatches[0].submitId, + channelId: exactSegmentMatches[0].channelId, + }; + } + const sameSupplierSegments = segmentMatches.filter((candidate) => + candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); + if (sameSupplierSegments.length === 1 && sameSupplierSegments[0]?.messageRecord) { + return { + message: sameSupplierSegments[0].messageRecord, + messageId: sameSupplierSegments[0].messageRecord.messageId, + submitRecordId: sameSupplierSegments[0].submitRecordId ?? undefined, + submitId: sameSupplierSegments[0].submitRecord?.submitId ?? sameSupplierSegments[0].submitId, + channelId: sameSupplierSegments[0].channelId, + }; + } + const crossConnectionSubmits = await this.prisma.smsSubmitRecord.findMany({ + where: { + gatewayMessageId: data.gatewayMessageId, + messageRecord: { phoneNumber }, + }, + include: { messageRecord: true, channel: true }, + orderBy: { createdAt: 'desc' }, + take: 10, + }); + const sameSupplierSubmits = crossConnectionSubmits.filter((candidate) => + candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); + if (sameSupplierSubmits.length === 1 && sameSupplierSubmits[0]?.messageRecord) { + return { + message: sameSupplierSubmits[0].messageRecord, + messageId: sameSupplierSubmits[0].messageRecord.messageId, + submitRecordId: sameSupplierSubmits[0].id, + submitId: sameSupplierSubmits[0].submitId, + channelId: sameSupplierSubmits[0].channelId, + }; + } + const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date(); const submittedAfter = new Date(deliveredAt.getTime() - 72 * 60 * 60 * 1000); const candidates = await this.prisma.smsSubmitRecord.findMany({ @@ -3790,12 +3962,25 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { message: candidates[0].messageRecord, messageId: candidates[0].messageRecord.messageId, submitRecordId: candidates[0].id, + submitId: candidates[0].submitId, + channelId: candidates[0].channelId, }; } - private receiptEventKey(data: GatewayReceiptEventDto) { + private isSameSupplierConnection( + left: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, + right: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, + ) { + return left.account.trim() === right.account.trim() + && left.gatewayHost.trim().toLowerCase() === right.gatewayHost.trim().toLowerCase() + && left.gatewayPort === right.gatewayPort + && left.protocol.trim().toUpperCase() === right.protocol.trim().toUpperCase() + && left.cmppVersion.trim() === right.cmppVersion.trim(); + } + + private receiptEventKey(data: GatewayReceiptEventDto, channelId = data.channelId) { return createHash('sha256').update([ - data.channelId, + channelId, data.gatewayMessageId, data.phoneNumber?.trim() ?? '', data.receiptStatus, diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index a814ad2..2a5da31 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -1564,7 +1564,7 @@ ## 2026-07-20 回执、报表、安全与公开 HTTP 接口收口要求 -1. 上游回执按内部消息 ID，或 `channelId + gatewayMessageId + DestTerminalId` 对提交记录作唯一匹配；同账号多通道不得互相认领。每个回执事件必须有数据库唯一键，并发或重复事件不得重复生成记录、退款、计费或下游投递。 +1. 供应商回执优先按内部消息 ID，或 `channelId + gatewayMessageId + DestTerminalId` 对提交记录/长短信分片审计作唯一匹配。若同一供应商账号配置了多个物理通道连接，回执可能从非原提交连接返回，此时仅允许在“账号、Gateway 主机、端口、协议及 CMPP 版本全部一致，且 `gatewayMessageId + DestTerminalId` 只有一个候选提交/分片”时跨连接认领，并仍归属原提交的逻辑通道；任一字段不同或候选不唯一必须拒绝自动匹配。每个回执事件必须以逻辑通道生成数据库唯一键，并发或重复事件不得重复生成记录、退款、计费或企业应用投递。 2. 成功回执统一更新短信主记录状态、回执状态、到达时间、真实通道、通道消息号、原始回执码和文本；失败、超时和未知回执保留可解释状态，最终失败只退款一次。 3. 对账、利润和质量报表统一以短信记录计费条数为发送量，以唯一主记录终态统计成功/失败；收入取扣费流水，返还取退款流水，成本取真实提交通道单价，利润等于收入减成本。到达时长按提交至成功回执计算，延迟回执由 T+1 和 T-4 至 T-1 重算覆盖；查询、页面和 CSV 共用同一聚合表。 4. 运营端与客户端用户接口使用各自安全 DTO，禁止输出密码散列、会话版本、登录失败内部计数和密钥字段。后端禁止自删除/自停用、禁止删除或降权最后一个平台管理员及企业管理员，并强制校验跨租户操作；唯一冲突返回 HTTP 409 和明确字段。 @@ -1666,7 +1666,9 @@ ## 2026-07-24 CMPP/HTTP 通讯交互日志要求 1. 运营端“系统日志”必须将人员操作审计与协议通讯日志分成两个独立页签。通讯日志至少支持协议、交互方向、事件类型、结果、关键字和时间范围过滤，并展示平台消息号、上游消息号或HTTP请求号、脱敏对象、结果码、耗时和安全详情。 -2. CMPP应覆盖客户登录/Submit、供应商SubmitResp、状态报告Deliver、上行Deliver及平台下游投递；HTTP应覆盖客户发送请求和平台回执/上行Webhook。日志状态至少区分已收到、已受理、成功、重试和失败，不能只记录最终成功。 -3. Gateway收到状态报告或上行后，必须对解包/解码失败及转发NestJS失败输出结构化安全日志；NestJS入口应分别记录到达和业务处理结果，以便区分“上游未发”“Gateway未收到”“Gateway转发失败”和“API落库失败”。 +2. CMPP应覆盖客户登录/Submit、供应商SubmitResp、状态报告Deliver、上行Deliver及平台下游投递；HTTP应覆盖客户发送请求和平台回执/上行Webhook。数据库中一条记录必须对应一个真实业务报文，不得把同一报文的“入口收到”和“处理成功”拆成两条记录；处理结果、结果码和耗时写在该报文同一条记录中，失败、重试等后续真实交互另行记录。 +3. Gateway收到状态报告或上行后，必须对解包/解码失败及转发NestJS失败输出结构化安全日志；NestJS入口把业务处理结果合并回同一报文记录，以便区分“上游未发”“Gateway未收到”“Gateway转发失败”和“API落库失败”。一条正常短短信的供应商侧完整成功闭环应依次展示四个真实报文：平台→通道 `CMPP_SUBMIT`、通道→平台 `CMPP_SUBMIT_RESP`、通道→平台 `CMPP_DELIVER`、平台→通道 `CMPP_DELIVER_RESP`；箭头只表达报文实际传输方向。 4. 通讯日志不得保存短信正文、密码、密钥、Token、签名鉴权值或完整HTTP请求体；手机号只保存脱敏值。CMPP心跳不得逐包写入数据库，连接健康仍使用连接状态和聚合指标。 5. 通讯日志写入不能阻塞短信主链路，默认批量异步写入，缓冲区应有上限和溢出告警；热数据默认保留30天，保留期允许通过环境变量配置。 +6. 通讯日志方向固定使用“企业应用 → 平台、平台 → 供应商通道、供应商通道 → 平台、平台 → 企业应用”。供应商长短信每个真实 `SUBMIT` 和 `SUBMIT_RESP` 分片各记一条，企业应用每个真实 `SUBMIT_RESP` 也必须记录；内部 `submit-result` 聚合回调不是协议报文，不得重复生成通讯日志。 +7. 供应商长短信回执必须先写入对应 `SmsMessageSegmentAudit`。仅当同一提交尝试的全部分片均为 `delivered` 时，主记录才转 `delivered` 并向企业应用投递一次最终回执；任一分片明确失败可进入最终失败/补发状态，分片尚未齐全时主记录保持 `submitted`，不得由首片成功提前聚合。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 1512ac7..c315861 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3786,9 +3786,14 @@ npm run verify:phase8 ## 2026-07-24 CMPP/HTTP 通讯交互日志用例 - `TC-PROTOCOL-LOG-001`：向Gateway客户认证入口提交不存在的CMPP账号；真实接口返回业务4xx，通讯日志分别出现`received`和`failed`事件，账号可检索、耗时和安全错误可见，数据库无短信业务记录。 -- `TC-PROTOCOL-LOG-002`：真实CMPP Submit获得供应商SubmitResp；通讯日志可按CMPP、通道到平台、SubmitResp及平台消息号筛选，展示上游消息号和结果码，不包含短信正文或通道密码。 -- `TC-PROTOCOL-LOG-003`：供应商发送DELIVER状态报告；Gateway结构化日志出现收到事件，NestJS通讯日志依次出现入口收到和落库成功。构造解包失败或API拒绝时必须出现对应失败证据，不能静默返回。 +- `TC-PROTOCOL-LOG-002`：真实CMPP Submit获得供应商SubmitResp；通讯日志可按CMPP、通道到平台、SubmitResp及平台消息号筛选，展示上游消息号和结果码，不包含短信正文或通道密码；同一个SubmitResp只能落一条最终处理结果，不得同时出现`received`和`success`重复行。 +- `TC-PROTOCOL-LOG-003`：供应商发送DELIVER状态报告；Gateway结构化日志出现收到事件，NestJS通讯日志以一条记录展示该报文及最终处理结果。构造解包失败或API拒绝时必须出现对应失败证据，不能静默返回。 - `TC-PROTOCOL-LOG-004`：通过公开HTTP API提交合法和非法请求；通讯日志展示客户到平台的受理或失败状态、请求号、脱敏手机号、业务码和耗时，鉴权头、密钥和正文不得入库。 - `TC-PROTOCOL-LOG-005`：平台向客户投递回执或上行Webhook并触发成功、网络失败和重试；通讯日志展示事件ID、HTTP状态或网络错误、耗时、尝试次数及最终状态，真实`HttpWebhookAttempt`状态一致。 - `TC-PROTOCOL-LOG-006`：连续运行CMPP心跳；`ProtocolInteractionLog`行数不随每个ACTIVE_TEST增长，连接状态中的最近心跳仍更新。超过配置保留期的数据被清理，业务表及操作审计不受影响。 - `TC-PROTOCOL-LOG-007`：运营端真实登录后打开系统日志，键盘切换“系统与操作日志/通讯交互日志”，筛选、分页、详情及固定操作列可用；桌面和平板/手机不产生页面级横向溢出，宽表允许容器内滚动，控制台无error/warn。 +- `TC-PROTOCOL-LOG-008`：一条真实短短信取得成功状态报告后，按同一平台消息号查询应恰好看到四个供应商侧真实业务报文：`平台→通道/CMPP_SUBMIT`、`通道→平台/CMPP_SUBMIT_RESP`、`通道→平台/CMPP_DELIVER`、`平台→通道/CMPP_DELIVER_RESP`；每个报文只出现一条，箭头与抓包传输方向一致，长短信则按实际分片分别记录Submit/SubmitResp。 +- `TC-PROTOCOL-LOG-009`：企业应用提交短信时，入站Submit显示“企业应用→平台”，每个实际返回的SubmitResp显示“平台→企业应用”；供应商侧统一显示“平台→供应商通道/供应商通道→平台”，不得再使用含义模糊的客户/通道箭头。 +- `TC-RECEIPT-SHARED-010`：供应商账号、Gateway主机、端口、协议和CMPP版本均相同的两个物理通道连接中，回执从副连接进入、原连接存在唯一`gatewayMessageId + DestTerminalId`分片候选时，应写入原提交逻辑通道；账号或端点任一不同、或候选超过一条时不得自动匹配。 +- `TC-RECEIPT-LONG-011`：两分片长短信仅收到第一片`DELIVRD`时，`SmsMessageRecord`保持`submitted`且不创建企业应用最终回执；第二片到达后两条分片审计均为`delivered`，主记录只聚合一次为`delivered`，重复回执不得重复投递、扣费或退款。 +- `TC-PROTOCOL-LOG-012`：供应商长短信每个真实分片分别产生一条`平台→供应商通道/CMPP_SUBMIT`和一条`供应商通道→平台/CMPP_SUBMIT_RESP`；内部`submit-result`聚合回调不得额外落协议日志。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index e385918..bbfed8d 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -2341,3 +2341,23 @@ git diff --check - Gateway重启后5条启用上游通道中3条立即连接，“富泷物业-联通/电信”共享账号`C59748`首次被供应商返回`auth failed`；自动重连按`nextReconnectAt=11:00:58`执行后两条均恢复。最终5/5通道全部`connected/currentConnections=1/desiredConnections=1`，最近心跳持续刷新，`lastError`和`nextReconnectAt`清空。 - 公网首页、运营登录、客户端登录和API health均HTTP 200，公网CMPP 17890 TCP连接成功。真实运营账号通过算术验证码登录，在“系统日志→通讯交互日志”看到预发布CMPP连接的`received/success`事件、耗时和固定详情列；浏览器控制台0条error/warn。 - 发布后API/Gateway日志未出现panic、fatal、unhandled、exception、通讯日志批量写入失败、回执解包失败或API转发失败；Nginx无error/crit/emerg。未发送真实短信、未充值、未审核、未删除或修改业务对象。 + +## 2026-07-24 通讯日志方向与重复记录修正（本地未提交、未部署） + +- 复核预发布号码`18821203795`对应消息`MSG-TEST-1784863949078-b28e828d`确认：原页面四条记录全部显示“通道→平台”，不是四个完整协议报文，而是仅采集了Gateway转入NestJS的`CMPP_SUBMIT_RESP`和`CMPP_DELIVER`，并把每个报文各拆成`received/success`两行；真实下行`CMPP_SUBMIT`与`CMPP_DELIVER_RESP`此前未采集，因此箭头虽符合现有事件入口方向，但不足以表达完整交互并造成重复观感。 +- 改为“一条数据库记录对应一个真实业务报文”：NestJS不再为同一入站报文分别写`received`和`success`，只落最终成功或失败结果；Go Gateway在真实`SendReqPkt(CMPP_SUBMIT)`及`SendRspPkt(CMPP_DELIVER_RESP)`写包完成后异步上报平台→通道事件。正常短短信完整成功闭环将显示`CMPP_SUBMIT → CMPP_SUBMIT_RESP → CMPP_DELIVER → CMPP_DELIVER_RESP`四条真实报文及实际传输方向；历史`received`行保留并标记为历史，不修改既有预发布数据。 +- Gateway遥测使用独立goroutine和既有HTTP超时，不阻塞发送/回执读循环；API仅接受`cmpp + platform_to_channel + submit/deliver_resp + success/failed`白名单事件，不接收正文、密码或密钥。 +- 本地使用真实NestJS API、真实算术验证码管理员会话和真实PostgreSQL写入两条安全的合成出站协议事件（未发送短信），再从运营端查询`MSG-DIRECTION-SMOKE`，页面正确显示“平台→通道 / CMPP_SUBMIT / 已发送”和“平台→通道 / CMPP_DELIVER_RESP / 已发送”；浏览器控制台0条error/warn。 +- 自动化验证：新增Controller单元测试覆盖入站单报文单记录和Gateway白名单，新增Go测试覆盖出站遥测payload；API全量、API/前端构建、Prisma validate、Gateway全量测试及`git diff --check`结果见本次会话最终记录。 +- 当前修改保持未提交、未推送、未部署；预发布仍运行`0bfeb0839ee6b67611e13796f46c19fc05b88de9`，发布前不得把本地验证结果误认为预发布已生效。 + +## 2026-07-24 跨连接回执、长短信聚合与通讯日志闭环修复（发布前） + +- 对号码`13127620092`的预发布只读证据确认：两分片均已由供应商受理，第二片回执从同账号复制通道连接返回。Gateway内存映射按物理连接隔离，未找到原提交后上报了`receipt-736070230367350788`和当前连接通道；NestJS又要求`channelId + gatewayMessageId`严格一致，最终通讯日志显示`SMS message record not found`。首片`DELIVRD`同时提前把主记录改成`delivered`，没有等待第二片，是独立的长短信聚合缺陷。 +- 回执匹配增加分片审计路径，并把“同供应商”固定为账号、Gateway主机、端口、协议和CMPP版本全部一致。只有`gatewayMessageId + DestTerminalId`在该供应商范围内唯一时才允许跨物理连接认领；回执、幂等键和主记录仍归属原提交逻辑通道。不同供应商或候选不唯一继续拒绝，避免串单。 +- 长短信每片回执先写`SmsMessageSegmentAudit`。部分成功时主记录保持`submitted`且不向企业应用投递最终回执；全部分片成功后才聚合为`delivered`并投递一次。任一明确失败进入既有最终失败/补发路径，重复回执继续由逻辑通道回执键幂等。 +- 通讯日志改为一条真实业务报文一条记录：供应商长短信每片真实`SUBMIT/SUBMIT_RESP`分别采集，内部`submit-result`聚合回调不重复落协议日志；Gateway补充供应商`SUBMIT`、`SUBMIT_RESP`、`DELIVER_RESP`和企业应用`SUBMIT_RESP`出入方向。运营端方向名称统一为“企业应用→平台、平台→供应商通道、供应商通道→平台、平台→企业应用”，回执成功处理后用真实主消息ID替换临时`receipt-*`标识。 +- 新增回归覆盖：同供应商副连接唯一匹配、不同供应商拒绝、两分片未齐不提前送达、全部到齐一次聚合、内部回调不重复记录、入站服务真实SubmitResp遥测及安全白名单。API定向2 suites/90项、API全量26 suites/313项、Gateway全量`go test ./...`和`go vet ./...`、API/前端build、Prisma generate/validate/status（本地67条migration最新）、`git diff --check`均通过。 +- 使用当前构建启动独立本地NestJS API，连接真实PostgreSQL和Redis执行`tools/smoke/receipt-cross-connection-smoke.mjs`：第二片从副连接进入后回执行的逻辑通道为原通道、主记录保持`submitted`；第一片补齐后主记录为`delivered`，恰好2条回执和2个已送达分片，脚本最终清理全部合成业务数据。 +- `verify:phase8`首次使用共享Redis且本机已有API争用时为373.05 TPS，低于500阈值，明确记为失败；改用独立临时Redis完整重跑为764.69 TPS并通过，临时Redis随后停止。前端仅保留既有约1.93MB单chunk警告。 +- 本次不新增Prisma模型或migration。发布后仍需只读确认服务、67条migration、Redis Stream、5条供应商通道重连、通讯日志新方向和近期错误；未经单独授权不发送真实短信，也不改写`13127620092`历史业务记录。 diff --git a/gateway/internal/inbound/protocol_log_test.go b/gateway/internal/inbound/protocol_log_test.go new file mode 100644 index 0000000..722d887 --- /dev/null +++ b/gateway/internal/inbound/protocol_log_test.go @@ -0,0 +1,52 @@ +package inbound + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestSubmitResponseProtocolLoggerEmitsActualPacketDirection(t *testing.T) { + events := make(chan protocolLogEvent, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/gateway/events/protocol-log" { + http.NotFound(w, r) + return + } + var event protocolLogEvent + if err := json.NewDecoder(r.Body).Decode(&event); err != nil { + t.Fatalf("decode event: %v", err) + } + events <- event + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) + })) + defer server.Close() + + gateway := Server{APIBaseURL: server.URL + "/api", HTTPClient: server.Client()} + gateway.submitResponseProtocolLogger( + "607532", + "cmpp20", + 141, + "13127620092", + "MSG-LONG-1", + 736070227905294338, + 0, + )(nil) + + select { + case event := <-events: + if event.Protocol != "cmpp" || event.Direction != "platform_to_client" || event.EventType != "submit_resp" { + t.Fatalf("unexpected protocol event: %+v", event) + } + if event.Account != "607532" || event.MessageID != "MSG-LONG-1" || event.ResultCode != "0" { + t.Fatalf("unexpected identifiers: %+v", event) + } + if event.Detail["sequenceId"] != float64(141) && event.Detail["sequenceId"] != uint32(141) { + t.Fatalf("sequenceId = %#v", event.Detail["sequenceId"]) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for protocol event") + } +} diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index d7af3e9..26a0cb6 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -68,10 +68,27 @@ type submitResponseMessage struct { } type submitResponse struct { - Accepted bool `json:"accepted"` - Result uint32 `json:"result,omitempty"` - MessageID string `json:"messageId"` - Messages []submitResponseMessage `json:"messages,omitempty"` + Accepted bool `json:"accepted"` + Result uint32 `json:"result,omitempty"` + TenantID string `json:"tenantId,omitempty"` + ApplicationID string `json:"applicationId,omitempty"` + MessageID string `json:"messageId"` + Messages []submitResponseMessage `json:"messages,omitempty"` +} + +type protocolLogEvent struct { + Protocol string `json:"protocol"` + Direction string `json:"direction"` + EventType string `json:"eventType"` + Status string `json:"status"` + TenantID string `json:"tenantId,omitempty"` + ApplicationID string `json:"applicationId,omitempty"` + Account string `json:"account,omitempty"` + MessageID string `json:"messageId,omitempty"` + GatewayMessageID string `json:"gatewayMessageId,omitempty"` + Phone string `json:"phone,omitempty"` + ResultCode string `json:"resultCode,omitempty"` + Detail map[string]any `json:"detail,omitempty"` } type authResponse struct { @@ -271,6 +288,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge req.protocol, req.protocol, packet.Conn.Conn.RemoteAddr(), req.sequenceID, "authenticated connection session not found", ) setInboundSubmitResponse(response.Packer, 0, 9) + response.AfterSend = s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 0, 9) return false, nil } account := session.account @@ -282,6 +300,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge fmt.Sprintf("enterprise code mismatch: expected %s", session.enterpriseCode), ) setInboundSubmitResponse(response.Packer, 0, 9) + response.AfterSend = s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9) return false, nil } phones := make([]string, len(req.destTerminalIDs)) @@ -306,6 +325,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge clientProtocol, req.protocol, account, remote, req.sequenceID, phone, err, ) setInboundSubmitResponse(response.Packer, 0, 9) + response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9) return false, nil } contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content))) @@ -335,6 +355,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason, ) setInboundSubmitResponse(response.Packer, 0, responseResult) + response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult) return false, nil } gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID) @@ -371,6 +392,20 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge go current.report(current, "submit", "") } response.AfterSend = func(sendErr error) { + s.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_client", + EventType: "submit_resp", + Status: protocolSendStatus(sendErr), + TenantID: result.TenantID, + ApplicationID: result.ApplicationID, + Account: account, + MessageID: result.MessageID, + GatewayMessageID: fmt.Sprint(gatewayMsgID), + Phone: phone, + ResultCode: protocolSendResultCode(sendErr, 0), + Detail: protocolSubmitResponseDetail(req.sequenceID, sendErr), + }) if sendErr != nil { return } @@ -387,6 +422,63 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge return false, nil } +func (s Server) submitResponseProtocolLogger( + account string, + protocol string, + sequenceID uint32, + phone string, + messageID string, + gatewayMessageID uint64, + result uint32, +) func(error) { + return func(sendErr error) { + s.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_client", + EventType: "submit_resp", + Status: protocolSendStatus(sendErr), + Account: account, + MessageID: messageID, + GatewayMessageID: fmt.Sprint(gatewayMessageID), + Phone: phone, + ResultCode: protocolSendResultCode(sendErr, result), + Detail: protocolSubmitResponseDetail(sequenceID, sendErr), + }) + } +} + +func protocolSendStatus(sendErr error) string { + if sendErr != nil { + return "failed" + } + return "success" +} + +func protocolSendResultCode(sendErr error, result uint32) string { + if sendErr != nil { + return "SEND_FAILED" + } + return fmt.Sprint(result) +} + +func protocolSubmitResponseDetail(sequenceID uint32, sendErr error) map[string]any { + detail := map[string]any{"sequenceId": sequenceID} + if sendErr != nil { + detail["error"] = sendErr.Error() + } + return detail +} + +func (s Server) emitProtocolLog(event protocolLogEvent) { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout) + defer cancel() + if err := s.post(ctx, "/gateway/events/protocol-log", event, nil); err != nil { + log.Printf("cmpp inbound protocol_event direction=%s event=%s status=telemetry_failed account=%s message_id=%s error=%q", event.Direction, event.EventType, event.Account, event.MessageID, err) + } + }() +} + func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) { session := findSessionByConn(packet.Conn) if session == nil { diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index 36d7fc5..307280f 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -139,6 +139,8 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { } acknowledgements <- event w.WriteHeader(http.StatusOK) + case "/api/gateway/events/protocol-log": + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } @@ -316,6 +318,8 @@ func TestInboundServerForwardsLongMessageFragmentsWithoutUDHAndAcknowledgesEachS _ = json.NewEncoder(w).Encode([]pendingDelivery{}) case "/api/gateway/events/inbound/connection": w.WriteHeader(http.StatusOK) + case "/api/gateway/events/protocol-log": + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } @@ -410,6 +414,8 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) { }}) case "/api/gateway/events/inbound/connection", "/api/gateway/events/downstream/sent": w.WriteHeader(http.StatusOK) + case "/api/gateway/events/protocol-log": + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } @@ -467,6 +473,8 @@ func TestDailyLimitRejectsSubmitSynchronouslyWithoutPendingReceipt(t *testing.T) _ = json.NewEncoder(w).Encode([]pendingDelivery{}) case "/api/gateway/events/inbound/connection": w.WriteHeader(http.StatusOK) + case "/api/gateway/events/protocol-log": + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } @@ -562,6 +570,8 @@ func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *tes _ = json.NewEncoder(w).Encode([]pendingDelivery{}) case "/api/gateway/events/inbound/connection": w.WriteHeader(http.StatusOK) + case "/api/gateway/events/protocol-log": + _ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true}) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } diff --git a/gateway/internal/upstream/manager.go b/gateway/internal/upstream/manager.go index 15a78f6..6c632e4 100644 --- a/gateway/internal/upstream/manager.go +++ b/gateway/internal/upstream/manager.go @@ -637,10 +637,46 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa seq, err := client.SendReqPkt(pkt) c.sendMu.Unlock() if err != nil { + c.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_channel", + EventType: "submit", + Status: "failed", + TenantID: cmd.TenantID, + ApplicationID: cmd.ApplicationID, + ChannelID: cmd.ChannelID, + Account: c.config.Account, + MessageID: cmd.MessageID, + Phone: cmd.PhoneNumber, + ResultCode: "SEND_FAILED", + PayloadBytes: len(part.MsgContent), + Detail: map[string]any{ + "segmentTotal": part.PkTotal, + "segmentIndex": part.PkNumber, + }, + }) c.close() result := submitResult(cmd, 0, "", "timeout", "SEND_FAILED", err.Error()) return 0, "", result, err } + c.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_channel", + EventType: "submit", + Status: "success", + TenantID: cmd.TenantID, + ApplicationID: cmd.ApplicationID, + ChannelID: cmd.ChannelID, + Account: c.config.Account, + MessageID: cmd.MessageID, + Phone: cmd.PhoneNumber, + PayloadBytes: len(part.MsgContent), + Detail: map[string]any{ + "sequenceId": seq, + "segmentTotal": part.PkTotal, + "segmentIndex": part.PkNumber, + }, + }) c.mu.Lock() c.pending[seq] = rspCh @@ -671,6 +707,25 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa errorCode = fmt.Sprint(rsp.result) errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.result) } + c.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "channel_to_platform", + EventType: "submit_resp", + Status: "success", + TenantID: cmd.TenantID, + ApplicationID: cmd.ApplicationID, + ChannelID: cmd.ChannelID, + Account: c.config.Account, + MessageID: cmd.MessageID, + GatewayMessageID: gatewayMessageID, + Phone: cmd.PhoneNumber, + ResultCode: fmt.Sprint(rsp.result), + Detail: map[string]any{ + "sequenceId": rsp.seqID, + "segmentTotal": part.PkTotal, + "segmentIndex": part.PkNumber, + }, + }) if rsp.result == 0 { c.mu.Lock() c.tracker[rsp.msgID] = cmd @@ -823,10 +878,12 @@ func (c *connection) readLoop() { ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result} } case *cmpp.Cmpp2DeliverReqPkt: - _ = c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliverPacketFromCMPP2(p), responseErr) c.handleDeliver(deliverPacketFromCMPP2(p)) case *cmpp.Cmpp3DeliverReqPkt: - _ = c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliverPacketFromCMPP3(p), responseErr) c.handleDeliver(deliverPacketFromCMPP3(p)) case *cmpp.CmppActiveTestReqPkt: _ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId) @@ -914,6 +971,78 @@ type deliverPacket struct { msgContent string } +type protocolLogEvent struct { + Protocol string `json:"protocol"` + Direction string `json:"direction"` + EventType string `json:"eventType"` + Status string `json:"status"` + TenantID string `json:"tenantId,omitempty"` + ApplicationID string `json:"applicationId,omitempty"` + ChannelID string `json:"channelId,omitempty"` + Account string `json:"account,omitempty"` + MessageID string `json:"messageId,omitempty"` + GatewayMessageID string `json:"gatewayMessageId,omitempty"` + Phone string `json:"phone,omitempty"` + ResultCode string `json:"resultCode,omitempty"` + PayloadBytes int `json:"payloadBytes,omitempty"` + Detail map[string]any `json:"detail,omitempty"` +} + +func (c *connection) emitDeliverResponse(pkt deliverPacket, responseErr error) { + status := "success" + resultCode := "0" + detail := map[string]any{"sequenceId": pkt.seqID} + gatewayMessageID := fmt.Sprint(pkt.msgID) + messageID := "" + phone := "" + tenantID := "" + applicationID := "" + channelID := c.channelID + if pkt.registerDelivery == 1 { + var receipt cmpp.CmppReceiptPkt + if err := receipt.Unpack([]byte(pkt.msgContent)); err == nil { + gatewayMessageID = fmt.Sprint(receipt.MsgId) + phone = strings.TrimSpace(receipt.DestTerminalId) + if cmd, ok := c.commandFor(receipt.MsgId); ok { + messageID = cmd.MessageID + tenantID = cmd.TenantID + applicationID = cmd.ApplicationID + channelID = cmd.ChannelID + } + } + } + if responseErr != nil { + status = "failed" + resultCode = "SEND_FAILED" + detail["error"] = responseErr.Error() + } + c.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_channel", + EventType: "deliver_resp", + Status: status, + TenantID: tenantID, + ApplicationID: applicationID, + ChannelID: channelID, + Account: c.config.Account, + MessageID: messageID, + GatewayMessageID: gatewayMessageID, + Phone: phone, + ResultCode: resultCode, + Detail: detail, + }) +} + +func (c *connection) emitProtocolLog(event protocolLogEvent) { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout) + defer cancel() + if err := postJSON(ctx, c.httpClient, c.apiBaseURL, "/gateway/events/protocol-log", event); err != nil { + log.Printf("protocol_event protocol=%s direction=%s event=%s status=telemetry_failed channel_id=%s message_id=%s error=%q", event.Protocol, event.Direction, event.EventType, event.ChannelID, event.MessageID, err) + } + }() +} + func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket { return deliverPacket{ seqID: pkt.SeqId, diff --git a/gateway/internal/upstream/protocol_log_test.go b/gateway/internal/upstream/protocol_log_test.go new file mode 100644 index 0000000..2346333 --- /dev/null +++ b/gateway/internal/upstream/protocol_log_test.go @@ -0,0 +1,54 @@ +package upstream + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestEmitProtocolLogPostsSafeOutboundPacketEvent(t *testing.T) { + events := make(chan protocolLogEvent, 1) + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/gateway/events/protocol-log" { + t.Errorf("unexpected path: %s", r.URL.Path) + http.Error(w, "unexpected path", http.StatusNotFound) + return + } + var event protocolLogEvent + if err := json.NewDecoder(r.Body).Decode(&event); err != nil { + t.Errorf("decode protocol event: %v", err) + http.Error(w, "invalid event", http.StatusBadRequest) + return + } + events <- event + w.WriteHeader(http.StatusOK) + })) + defer api.Close() + + conn := &connection{apiBaseURL: api.URL, httpClient: api.Client()} + conn.emitProtocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "platform_to_channel", + EventType: "submit", + Status: "success", + ChannelID: "channel-1", + MessageID: "MSG-1", + Phone: "18821203795", + PayloadBytes: 32, + Detail: map[string]any{"sequenceId": 7}, + }) + + select { + case event := <-events: + if event.Direction != "platform_to_channel" || event.EventType != "submit" || event.Status != "success" { + t.Fatalf("unexpected protocol event: %+v", event) + } + if event.MessageID != "MSG-1" || event.Phone != "18821203795" || event.PayloadBytes != 32 { + t.Fatalf("unexpected event identity: %+v", event) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for protocol event") + } +} diff --git a/src/apps/admin/AdminSystemLogsPage.tsx b/src/apps/admin/AdminSystemLogsPage.tsx index b0de47a..89adc6b 100644 --- a/src/apps/admin/AdminSystemLogsPage.tsx +++ b/src/apps/admin/AdminSystemLogsPage.tsx @@ -160,10 +160,10 @@ export function AdminSystemLogsPage() { } const directionLabels: Record = { - client_to_platform: '客户 → 平台', - platform_to_channel: '平台 → 通道', - channel_to_platform: '通道 → 平台', - platform_to_client: '平台 → 客户', + client_to_platform: '企业应用 → 平台', + platform_to_channel: '平台 → 供应商通道', + channel_to_platform: '供应商通道 → 平台', + platform_to_client: '平台 → 企业应用', }; const protocolStatusTone: Record = { @@ -174,6 +174,26 @@ const protocolStatusTone: Record = { + submit: 'CMPP_SUBMIT', + submit_resp: 'CMPP_SUBMIT_RESP', + deliver_receipt: 'CMPP_DELIVER（状态报告）', + deliver_uplink: 'CMPP_DELIVER（上行）', + deliver_resp: 'CMPP_DELIVER_RESP', + connect: 'CMPP_CONNECT', + send_request: 'HTTP发送请求', + receipt_webhook: 'HTTP回执Webhook', + uplink_webhook: 'HTTP上行Webhook', +}; + +function protocolStatusLabel(record: ProtocolInteractionLogItem) { + if (record.status === 'failed') return '失败'; + if (record.status === 'retrying') return '重试中'; + if (record.status === 'accepted') return '已受理'; + if (record.status === 'received') return '已收到（历史）'; + return record.direction === 'platform_to_channel' || record.direction === 'platform_to_client' ? '已发送' : '已接收并处理'; +} + function ProtocolInteractionPanel({ active }: { active: boolean }) { const [inputs, setInputs] = useState({ keyword: '', protocol: 'all', direction: 'all', eventType: 'all', status: 'all', range: 'today' }); const [filters, setFilters] = useState(inputs); @@ -210,10 +230,10 @@ function ProtocolInteractionPanel({ active }: { active: boolean }) { { key: 'createdAt', title: '时间', width: '170px', render: (record) => {formatDateTime(record.createdAt)} }, { key: 'protocol', title: '协议', width: '90px', render: (record) => {record.protocol.toUpperCase()} }, { key: 'direction', title: '方向', width: '150px', render: (record) => directionLabels[record.direction] }, - { key: 'eventType', title: '事件', width: '150px', render: (record) => {record.eventType} }, + { key: 'eventType', title: '协议报文', width: '190px', render: (record) => {protocolEventLabels[record.eventType] ?? record.eventType} }, { key: 'messageId', title: '消息标识', width: '220px', render: (record) =>
{record.messageId || '-'}{record.gatewayMessageId || record.requestId || ''}
}, { key: 'target', title: '对象', width: '160px', render: (record) =>
{record.phoneMasked || record.account || '-'}{record.channelId || record.applicationId || ''}
}, - { key: 'status', title: '结果', width: '130px', render: (record) =>
{record.status}{record.resultCode || ''}
}, + { key: 'status', title: '处理结果', width: '150px', render: (record) =>
{protocolStatusLabel(record)}{record.resultCode || ''}
}, { key: 'durationMs', title: '耗时', width: '90px', render: (record) => record.durationMs == null ? '-' : `${record.durationMs} ms` }, { key: 'detail', title: '详情', width: '90px', render: (record) => }, ], []); @@ -232,7 +252,7 @@ function ProtocolInteractionPanel({ active }: { active: boolean }) { return (
-
仅记录业务交互和异常，不逐包记录 CMPP 心跳；手机号已脱敏，短信内容、密钥和鉴权头不会入库。
+
一条记录对应一个真实业务报文，箭头表示报文实际传输方向；不逐包记录 CMPP 心跳，手机号已脱敏，短信内容、密钥和鉴权头不会入库。
setInputs((value) => ({ ...value, keyword: event.target.value }))} placeholder="消息ID、请求ID、账号、脱敏手机号或结果码" prefix={} value={inputs.keyword} />