From a758672436b3ebcd78aab91ba4a8e3cf68cbdfb0 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Wed, 15 Jul 2026 12:05:24 +0800 Subject: [PATCH] fix: harden dependencies and downstream delivery --- api/package-lock.json | 38 +++---- api/package.json | 9 +- api/src/files/files.controller.ts | 12 ++- api/src/operations/operations.service.spec.ts | 33 ++++++- api/src/operations/operations.service.ts | 32 ++++-- api/src/send-chain/send-chain.service.spec.ts | 98 +++++++++++++++++++ api/src/send-chain/send-chain.service.ts | 77 ++++++++++++++- api/src/tenants/tenants.service.spec.ts | 7 +- api/src/tenants/tenants.service.ts | 12 ++- .../first-version-development-requirements.md | 7 +- docs/system-functional-test-cases.md | 14 ++- docs/testing-progress.md | 22 +++++ gateway/internal/inbound/server.go | 62 +++++++++++- gateway/internal/inbound/server_test.go | 41 +++++++- src/api/adminApi.ts | 2 +- src/apps/client/ClientHome.tsx | 5 +- src/layouts/AdminLayout.tsx | 4 - 17 files changed, 418 insertions(+), 57 deletions(-) diff --git a/api/package-lock.json b/api/package-lock.json index b9aa37e..c3eb4eb 100644 --- a/api/package-lock.json +++ b/api/package-lock.json @@ -8,10 +8,10 @@ "name": "cmpp-platform-api", "version": "0.1.0", "dependencies": { - "@nestjs/common": "^11.1.9", + "@nestjs/common": "^11.1.28", "@nestjs/config": "^4.0.2", - "@nestjs/core": "^11.1.9", - "@nestjs/platform-express": "^11.1.9", + "@nestjs/core": "^11.1.28", + "@nestjs/platform-express": "^11.1.28", "@nestjs/swagger": "^11.2.3", "@prisma/adapter-pg": "^7.8.0", "@prisma/client": "^7.0.1", @@ -649,9 +649,9 @@ } }, "node_modules/@hono/node-server": { - "version": "1.19.11", - "resolved": "https://registry.npmjs.org/@hono/node-server/-/node-server-1.19.11.tgz", - "integrity": "sha512-dr8/3zEaB+p0D2n/IUrlPF1HZm586qgJNXK1a9fhg/PzdtkK7Ksd5l312tJX2yBuALqDYBlG20QEbayqPyxn+g==", + "version": "1.19.13", + "resolved": "https://registry.npmjs.org/@hono/node-server/-/node-server-1.19.13.tgz", + "integrity": "sha512-TsQLe4i2gvoTtrHje625ngThGBySOgSK3Xo2XRYOdqGN1teR8+I7vchQC46uLJi8OF62YTYA3AhSpumtkhsaKQ==", "devOptional": true, "license": "MIT", "engines": { @@ -1302,9 +1302,9 @@ } }, "node_modules/@nestjs/common": { - "version": "11.1.27", - "resolved": "https://registry.npmjs.org/@nestjs/common/-/common-11.1.27.tgz", - "integrity": "sha512-kEGSzqM2lWr4whh4Ubflw+oPZSEzxvRMu9WL+LveZploJWTjec5bBlCiRVlVzTPg2kIwBiLwWSvCCW7Wnin1gg==", + "version": "11.1.28", + "resolved": "https://registry.npmjs.org/@nestjs/common/-/common-11.1.28.tgz", + "integrity": "sha512-bRImsxibie+AM7xjdwcrm/gr5YeacI65kSBNzTufa1Ib5iwziaY/lqMtRh9THq6pbV4e1HP9aI2ZxGUumnmaoQ==", "license": "MIT", "dependencies": { "file-type": "21.3.4", @@ -1348,9 +1348,9 @@ } }, "node_modules/@nestjs/core": { - "version": "11.1.27", - "resolved": "https://registry.npmjs.org/@nestjs/core/-/core-11.1.27.tgz", - "integrity": "sha512-K6DX7hcqmZdeXkv7tsPakKBRCgqL19a4mtbX4FluY0hWtFdtPKp6lbe+lb8gWPfvLdbOWr/CPScn7BSjBX+Ecg==", + "version": "11.1.28", + "resolved": "https://registry.npmjs.org/@nestjs/core/-/core-11.1.28.tgz", + "integrity": "sha512-06m63xIRj8+l8uOeh/8LnYupGubkyu4f+bPKIadaSui6vK9KpXgoz7HveT1yOVLcEt0M0oCOEW5EuEXZkEmBBQ==", "license": "MIT", "dependencies": { "fast-safe-stringify": "2.1.1", @@ -1407,14 +1407,14 @@ } }, "node_modules/@nestjs/platform-express": { - "version": "11.1.27", - "resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-11.1.27.tgz", - "integrity": "sha512-0ZFhz6H6EdGh4xQVbUNwjoAwBuz73P7FvUAl67h9CTdMqQlJDaQYJApBv8pKfVZ1fGjMCbl0m9DcC6pXaZPWSQ==", + "version": "11.1.28", + "resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-11.1.28.tgz", + "integrity": "sha512-hU+9Sz4m+onHrR5AmelI59QKmY/Re546bPnygnpqqeQdHDiJpBgjWbL4t6Jr73CBpS60cpyng7WzjgphNB9iwA==", "license": "MIT", "dependencies": { "cors": "2.8.6", "express": "5.2.1", - "multer": "2.1.1", + "multer": "2.2.0", "path-to-regexp": "8.4.2", "tslib": "2.8.1" }, @@ -5712,9 +5712,9 @@ } }, "node_modules/multer": { - "version": "2.1.1", - "resolved": "https://registry.npmjs.org/multer/-/multer-2.1.1.tgz", - "integrity": "sha512-mo+QTzKlx8R7E5ylSXxWzGoXoZbOsRMpyitcht8By2KHvMbf3tjwosZ/Mu/XYU6UuJ3VZnODIrak5ZrPiPyB6A==", + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/multer/-/multer-2.2.0.tgz", + "integrity": "sha512-6rdyFg2kLrMh9Jee7/BMPuV9lEAd7lLW2YUpF9/YxR7njyoUwwQ0ZPh3TaIY50Sw6vlyD2HW3wGOkTS4P79xrQ==", "license": "MIT", "dependencies": { "append-field": "^1.0.0", diff --git a/api/package.json b/api/package.json index 68adee1..87b5058 100644 --- a/api/package.json +++ b/api/package.json @@ -14,10 +14,10 @@ "prisma:migrate:deploy": "prisma migrate deploy" }, "dependencies": { - "@nestjs/common": "^11.1.9", + "@nestjs/common": "^11.1.28", "@nestjs/config": "^4.0.2", - "@nestjs/core": "^11.1.9", - "@nestjs/platform-express": "^11.1.9", + "@nestjs/core": "^11.1.28", + "@nestjs/platform-express": "^11.1.28", "@nestjs/swagger": "^11.2.3", "@prisma/adapter-pg": "^7.8.0", "@prisma/client": "^7.0.1", @@ -38,5 +38,8 @@ "ts-jest": "^29.4.11", "ts-node": "^10.9.2", "typescript": "^6.0.3" + }, + "overrides": { + "@hono/node-server": "1.19.13" } } diff --git a/api/src/files/files.controller.ts b/api/src/files/files.controller.ts index e8a777e..9209e5a 100644 --- a/api/src/files/files.controller.ts +++ b/api/src/files/files.controller.ts @@ -48,7 +48,17 @@ export class FilesController { } @Post('upload') - @UseInterceptors(FileInterceptor('file', { limits: { fileSize: 20 * 1024 * 1024 } })) + @UseInterceptors(FileInterceptor('file', { + limits: { + fileSize: 20 * 1024 * 1024, + files: 1, + fields: 4, + parts: 5, + fieldNameSize: 100, + fieldSize: 1024, + headerPairs: 100, + }, + })) upload(@UploadedFile() file: UploadedMultipartFile, @Body('purpose') purpose: string, @Body('prefix') prefix?: string, @TenantId() tenantId?: string) { if (!file) { throw new BadRequestException('Upload file is required'); diff --git a/api/src/operations/operations.service.spec.ts b/api/src/operations/operations.service.spec.ts index 605fa5c..27c3733 100644 --- a/api/src/operations/operations.service.spec.ts +++ b/api/src/operations/operations.service.spec.ts @@ -254,6 +254,7 @@ describe('OperationsService', () => { .mockResolvedValueOnce(1) .mockResolvedValueOnce(1) .mockResolvedValueOnce(2); + prisma.accountTransaction.aggregate.mockResolvedValueOnce({ _count: { _all: 2 }, _sum: { amountCents: 10 } }); const service = new OperationsService(prisma as never); await expect(service.dashboard({ tenantId: 'tenant-1' })).resolves.toEqual( @@ -269,6 +270,7 @@ describe('OperationsService', () => { templates: 1, total: 5, }, + today: expect.objectContaining({ returnedCents: 10 }), gatewayConnections: [{ status: 'connected', _count: { _all: 1 }, _sum: { currentConnections: 2, desiredConnections: 2 } }], downstreamDeliverySummary: expect.objectContaining({ pending: 3, @@ -282,7 +284,14 @@ describe('OperationsService', () => { }), ); expect(prisma.cmppDownstreamDelivery.count).toHaveBeenNthCalledWith(4, { - where: { tenantId: 'tenant-1', status: 'pending', createdAt: { lte: expect.any(Date) } }, + where: { + tenantId: 'tenant-1', + status: 'pending', + OR: [ + { lastRetriedAt: null, createdAt: { lte: expect.any(Date) } }, + { lastRetriedAt: { lte: expect.any(Date) } }, + ], + }, }); expect(prisma.cmppDownstreamDelivery.count).toHaveBeenNthCalledWith(5, { where: { tenantId: 'tenant-1', status: 'awaiting_ack', ackDeadlineAt: { lte: expect.any(Date) } }, @@ -294,6 +303,18 @@ describe('OperationsService', () => { updatedAt: { gte: expect.any(Date) }, }, }); + expect(prisma.accountTransaction.aggregate).toHaveBeenCalledWith({ + where: { + tenantId: 'tenant-1', + createdAt: { gte: expect.any(Date) }, + OR: [ + { transactionType: 'refunded' }, + { transactionType: 'released', relatedType: 'sms_message_record' }, + ], + }, + _sum: { amountCents: true }, + _count: { _all: true }, + }); await service.statistics({ tenantId: 'tenant-1', groupBy: 'application' }); expect(prisma.smsMessageRecord.groupBy).toHaveBeenCalledWith({ @@ -516,10 +537,16 @@ describe('OperationsService', () => { by: ['applicationId'], where: { AND: [ - { tenantId: 'tenant-1', applicationId: 'app-1', deliveryType: undefined }, + { tenantId: 'tenant-1', applicationId: 'app-1', deliveryType: undefined, createdAt: undefined }, { OR: [ - { status: 'pending', createdAt: { lte: expect.any(Date) } }, + { + status: 'pending', + OR: [ + { lastRetriedAt: null, createdAt: { lte: expect.any(Date) } }, + { lastRetriedAt: { lte: expect.any(Date) } }, + ], + }, { status: 'awaiting_ack', ackDeadlineAt: { lte: expect.any(Date) } }, { status: { in: ['failed', 'unconfirmed', 'rejected'] }, updatedAt: { gte: expect.any(Date) } }, ], diff --git a/api/src/operations/operations.service.ts b/api/src/operations/operations.service.ts index ef404e5..1745797 100644 --- a/api/src/operations/operations.service.ts +++ b/api/src/operations/operations.service.ts @@ -186,7 +186,7 @@ export class OperationsService { _count: { _all: true }, }), this.prisma.accountTransaction.aggregate({ - where: { tenantId: query.tenantId, transactionType: 'refunded', createdAt: { gte: sinceToday } }, + where: returnedTransactionWhere(sinceToday, query.tenantId), _sum: { amountCents: true }, _count: { _all: true }, }), @@ -230,8 +230,7 @@ export class OperationsService { this.prisma.cmppDownstreamDelivery.count({ where: { tenantId: query.tenantId, - status: 'pending', - createdAt: { lte: downstreamAlertWindow.stalledPendingAt }, + ...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt), }, }), this.prisma.cmppDownstreamDelivery.count({ @@ -261,6 +260,7 @@ export class OperationsService { unknown: todayTotals.unknown, successRate: todayTotals.total > 0 ? Number(((todayTotals.delivered / todayTotals.total) * 100).toFixed(1)) : 0, spendCents: todayTotals.amountCents, + returnedCents: transactionAggregate._sum.amountCents ?? 0, billingUnits: todayTotals.billingUnits, }, uplinkCount, @@ -439,8 +439,7 @@ export class OperationsService { this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, - status: 'pending', - createdAt: { lte: downstreamAlertWindow.stalledPendingAt }, + ...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt), }, }), this.prisma.cmppDownstreamDelivery.count({ @@ -841,6 +840,17 @@ function startOfToday() { return date; } +function returnedTransactionWhere(since: Date, tenantId?: string): Prisma.AccountTransactionWhereInput { + return { + tenantId, + createdAt: { gte: since }, + OR: [ + { transactionType: 'refunded' }, + { transactionType: 'released', relatedType: 'sms_message_record' }, + ], + }; +} + function createdAtRange(range?: string): Prisma.DateTimeFilter | undefined { if (!range || range === 'all') { return undefined; @@ -882,7 +892,7 @@ function downstreamAlertWhere( scopedWhere, { OR: [ - { status: 'pending', createdAt: { lte: window.stalledPendingAt } }, + stalledPendingWhere(window.stalledPendingAt), { status: 'awaiting_ack', ackDeadlineAt: { lte: window.now } }, { status: { in: ['failed', 'unconfirmed', 'rejected'] }, updatedAt: { gte: window.recentFailedAt } }, ], @@ -891,6 +901,16 @@ function downstreamAlertWhere( }; } +function stalledPendingWhere(cutoff: Date): Prisma.CmppDownstreamDeliveryWhereInput { + return { + status: 'pending', + OR: [ + { lastRetriedAt: null, createdAt: { lte: cutoff } }, + { lastRetriedAt: { lte: cutoff } }, + ], + }; +} + function downstreamDeliveryScopedWhere(query: DownstreamDeliveryDashboardQuery): Prisma.CmppDownstreamDeliveryWhereInput { const createdAtFrom = parseDateBoundary(query.createdAtFrom, false); const createdAtTo = parseDateBoundary(query.createdAtTo, true); diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index c288c3a..bb43a21 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -1574,6 +1574,50 @@ describe('SendChainService', () => { }); }); + it('immediately terminates an unrecoverable downstream delivery', async () => { + const { service, prisma } = createService(); + prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue({ + id: 'delivery-1', + tenantId: 'tenant-1', + applicationId: 'app-1', + messageId: 'MSG-1', + deliveryType: 'receipt', + retryCount: 0, + retryEnabled: true, + }); + + await service.markDownstreamDeliveryFailed( + 'delivery-1', + '历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id', + 'unrecoverable', + ); + + expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ + status: 'failed', + retryCount: 1, + nextRetryAt: null, + }), + })); + }); + + it('does not let the pending timeout scan overwrite a delivery that is already awaiting acknowledgement', async () => { + const { service, prisma } = createService(); + const awaitingAck = { + id: 'delivery-1', status: 'awaiting_ack', tenantId: 'tenant-1', applicationId: 'app-1', + messageId: 'MSG-1', deliveryType: 'receipt', retryCount: 0, + }; + prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue(awaitingAck); + + await expect(service.markDownstreamDeliveryFailed( + 'delivery-1', + '下游投递排队超过 72 小时，系统自动终止重试', + 'queue_timeout', + )).resolves.toEqual(awaitingAck); + + expect(prisma.cmppDownstreamDelivery.update).not.toHaveBeenCalled(); + }); + it('only marks downstream delivery delivered after a successful CMPP_DELIVER_RESP', async () => { const { service, prisma } = createService(); @@ -1769,6 +1813,35 @@ describe('SendChainService', () => { expect(service['postGatewayControl']).not.toHaveBeenCalled(); }); + it('terminates a manual requeue when gateway reports it is unrecoverable', async () => { + const { service, prisma } = createService(); + prisma.cmppDownstreamDelivery.findUnique.mockResolvedValueOnce({ + id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1', + deliveryType: 'receipt', status: 'failed', retryCount: 3, manualRetryCount: 0, + payload: { account: '100001', messageId: 'MSG-1', receiptStatus: 'delivered' }, + application: { cmppAccount: '100001' }, + }).mockResolvedValueOnce({ + id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1', + deliveryType: 'receipt', status: 'pending', retryCount: 0, retryEnabled: true, + }); + service['postGatewayControl'] = jest.fn().mockResolvedValue({ + sent: false, + retryable: false, + reasonCode: 'MISSING_SUBMIT_SEQUENCE_ID', + errorMessage: '历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投', + }); + + await service.requeueDownstreamDelivery('delivery-1'); + + expect(prisma.cmppDownstreamDelivery.update).toHaveBeenLastCalledWith(expect.objectContaining({ + data: expect.objectContaining({ + status: 'failed', + nextRetryAt: null, + lastError: expect.stringContaining('MISSING_SUBMIT_SEQUENCE_ID'), + }), + })); + }); + it('supports batch requeue of downstream deliveries', async () => { const { service } = createService(); service.requeueDownstreamDelivery = jest.fn() @@ -1819,16 +1892,41 @@ describe('SendChainService', () => { expect(prisma.smsBatchTask.update).toHaveBeenCalled(); }); + it('terminates downstream deliveries that remain pending for 72 hours after the latest manual retry', async () => { + const { service, prisma } = createService(); + prisma.cmppDownstreamDelivery.findMany.mockResolvedValue([{ id: 'delivery-expired' }]); + service.markDownstreamDeliveryFailed = jest.fn().mockResolvedValue({ id: 'delivery-expired', status: 'failed' }); + + await expect(service.markExpiredDownstreamDeliveries(72)).resolves.toEqual({ failed: 1 }); + + expect(prisma.cmppDownstreamDelivery.findMany).toHaveBeenCalledWith(expect.objectContaining({ + where: { + status: 'pending', + OR: [ + { lastRetriedAt: null, createdAt: { lte: expect.any(Date) } }, + { lastRetriedAt: { lte: expect.any(Date) } }, + ], + }, + })); + expect(service.markDownstreamDeliveryFailed).toHaveBeenCalledWith( + 'delivery-expired', + '下游投递排队超过 72 小时，系统自动终止重试', + 'queue_timeout', + ); + }); + it('starts the automatic receipt-timeout scan after application startup', async () => { jest.useFakeTimers(); const previousEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED; const { service } = createService(); const scan = jest.spyOn(service, 'markUnknownTimeout').mockResolvedValue({ timeout: 0 }); + const downstreamScan = jest.spyOn(service, 'markExpiredDownstreamDeliveries').mockResolvedValue({ failed: 0 }); try { process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = 'true'; service.onModuleInit(); await jest.advanceTimersByTimeAsync(60_000); expect(scan).toHaveBeenCalledWith({}); + expect(downstreamScan).toHaveBeenCalledWith(); await service.onModuleDestroy(); } finally { if (previousEnabled === undefined) delete process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED; diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 24525e1..4611cb5 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -119,11 +119,21 @@ export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentD acknowledgedAt?: string; } -export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'ack_invalid' | 'connection_lost'; +export type GatewayDownstreamFailureType = + | 'send_failed' + | 'ack_timeout' + | 'ack_rejected' + | 'ack_invalid' + | 'connection_lost' + | 'unrecoverable' + | 'queue_timeout'; type GatewayControlDeliveryResult = { sent?: boolean; delivered?: boolean; + retryable?: boolean; + reasonCode?: string; + errorMessage?: string; connectionId?: string; sequenceId?: string; messageId?: string; @@ -218,6 +228,7 @@ const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands'; const DEFAULT_DOWNSTREAM_RETRY_DELAY_MS = 60_000; const DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS = 30 * 60_000; const DEFAULT_DOWNSTREAM_MAX_RETRIES = 10; +const DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS = 72; const DEFAULT_RECEIPT_TIMEOUT_HOURS = 72; const DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS = 5 * 60_000; const RECEIPT_TIMEOUT_INITIAL_DELAY_MS = 60_000; @@ -879,6 +890,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { receiptStatus: data.receiptStatus, rawStatus: data.rawStatus, errorCode: data.errorCode, + submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined, deliveredAt: deliveredAt.toISOString(), }, }); @@ -1044,10 +1056,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (delivery.status === 'delivered') { return delivery; } + if (failureType === 'queue_timeout' && delivery.status !== 'pending') { + return delivery; + } const retryCount = (delivery.retryCount ?? 0) + 1; const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'ack_invalid' || failureType === 'connection_lost'; const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false; - const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries(); + const nonRetryableFailure = failureType === 'unrecoverable' || failureType === 'queue_timeout'; + const finalFailure = nonRetryableFailure || !retryAllowed || retryCount >= downstreamMaxRetries(); const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed'; const updated = await this.prisma.cmppDownstreamDelivery.update({ where: { id }, @@ -1327,7 +1343,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (result.sent || result.delivered) { return this.markDownstreamDeliverySent({ id: delivery.id, ...result }); } - return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } }); + return this.markDownstreamDeliveryFailed( + delivery.id, + downstreamControlFailureMessage(result), + result.retryable === false ? 'unrecoverable' : 'send_failed', + ); } catch (error) { return this.markDownstreamDeliveryFailed( delivery.id, @@ -1504,6 +1524,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ) as GatewayControlDeliveryResult; if (result.sent || result.delivered) { await this.markDownstreamDeliverySent({ id: delivery.id, ...result }); + } else { + await this.markDownstreamDeliveryFailed( + delivery.id, + downstreamControlFailureMessage(result), + result.retryable === false ? 'unrecoverable' : 'send_failed', + ); } } catch (error) { await this.markDownstreamDeliveryFailed(delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed'); @@ -1884,12 +1910,39 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return { timeout }; } + async markExpiredDownstreamDeliveries(olderThanHours = downstreamPendingTimeoutHours()) { + const cutoff = new Date(Date.now() - olderThanHours * 60 * 60_000); + const expired = await this.prisma.cmppDownstreamDelivery.findMany({ + where: { + status: 'pending', + OR: [ + { lastRetriedAt: null, createdAt: { lte: cutoff } }, + { lastRetriedAt: { lte: cutoff } }, + ], + }, + select: { id: true }, + take: 500, + }); + for (const delivery of expired) { + await this.markDownstreamDeliveryFailed( + delivery.id, + `下游投递排队超过 ${olderThanHours} 小时，系统自动终止重试`, + 'queue_timeout', + ); + } + return { failed: expired.length }; + } + private async runReceiptTimeoutScan() { if (this.receiptTimeoutScanRunning) return; this.receiptTimeoutScanRunning = true; try { - const result = await this.markUnknownTimeout({}); - if (result.timeout > 0) this.logger.log(`Marked ${result.timeout} messages as receipt timeout and refunded charged messages`); + const [receiptResult, downstreamResult] = await Promise.all([ + this.markUnknownTimeout({}), + this.markExpiredDownstreamDeliveries(), + ]); + if (receiptResult.timeout > 0) this.logger.log(`Marked ${receiptResult.timeout} messages as receipt timeout and refunded charged messages`); + if (downstreamResult.failed > 0) this.logger.log(`Terminated ${downstreamResult.failed} expired downstream deliveries`); } catch (error) { this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error)); } finally { @@ -2863,6 +2916,20 @@ function downstreamMaxRetries() { return Number.isFinite(value) && value > 0 ? Math.floor(value) : DEFAULT_DOWNSTREAM_MAX_RETRIES; } +function downstreamPendingTimeoutHours() { + const value = Number(process.env.CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS ?? DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS); + return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS; +} + +function downstreamControlFailureMessage(result: GatewayControlDeliveryResult) { + const reason = String(result.errorMessage ?? '').trim(); + const code = String(result.reasonCode ?? '').trim(); + if (reason && code) return `${reason} (${code})`; + if (reason) return reason; + if (code) return `Gateway 未完成下游投递 (${code})`; + return 'Gateway 未完成下游投递，等待自动重试'; +} + function parseImportRows(content: string, delimiter?: ',' | '\t') { const normalized = content.replace(/^\uFEFF/, ''); const lines = normalized.split(/\r?\n/).filter((line) => line.trim().length > 0); diff --git a/api/src/tenants/tenants.service.spec.ts b/api/src/tenants/tenants.service.spec.ts index 6327336..ce2c554 100644 --- a/api/src/tenants/tenants.service.spec.ts +++ b/api/src/tenants/tenants.service.spec.ts @@ -105,7 +105,12 @@ describe('TenantsService', () => { })); expect(prisma.accountTransaction.groupBy).toHaveBeenCalledWith(expect.objectContaining({ by: ['tenantId'], - where: expect.objectContaining({ transactionType: 'refunded' }), + where: expect.objectContaining({ + OR: [ + { transactionType: 'refunded' }, + { transactionType: 'released', relatedType: 'sms_message_record' }, + ], + }), _sum: { amountCents: true }, })); }); diff --git a/api/src/tenants/tenants.service.ts b/api/src/tenants/tenants.service.ts index 0e916bf..3a00bdb 100644 --- a/api/src/tenants/tenants.service.ts +++ b/api/src/tenants/tenants.service.ts @@ -59,7 +59,7 @@ export class TenantsService { }), this.prisma.accountTransaction.groupBy({ by: ['tenantId'], - where: { transactionType: 'refunded', createdAt: { gte: sinceToday } }, + where: returnedTransactionWhere(sinceToday), _sum: { amountCents: true }, }), ]); @@ -201,6 +201,16 @@ function startOfToday() { return date; } +function returnedTransactionWhere(since: Date): Prisma.AccountTransactionWhereInput { + return { + createdAt: { gte: since }, + OR: [ + { transactionType: 'refunded' }, + { transactionType: 'released', relatedType: 'sms_message_record' }, + ], + }; +} + function generateTenantCode(data: CreateTenantDto) { const source = data.creditCode?.trim() || data.name.trim(); const normalized = source.replace(/[^\da-zA-Z]/g, '').toLowerCase(); diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index b0b41a7..10c9231 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -279,7 +279,10 @@ - 已实现下游投递失败审计与人工重投第一版：运营端后端与页面可分页查看 `CmppDownstreamDelivery` 的 pending/awaiting_ack/failed/unconfirmed/rejected/delivered 记录，支持按状态、类型、应用和关键字筛选，并可对非 `awaiting_ack` 记录执行人工重投，真实调用 Gateway `/downstream/receipt` 或 `/downstream/uplink`。主记录必须分开保存自动重试次数 `retryCount`、人工重投次数 `manualRetryCount` 和最近人工重投时间 `lastRetriedAt`，操作日志保留重投前状态与自动重试次数。 - 已实现下游投递批量重投第一版：运营端可在当前页勾选多条 `pending/failed` 下游投递记录，调用真实批量接口逐条重投并返回成功/失败汇总，不允许用前端循环假装成功。 - 下游投递的 `pending` 展示必须结合真实尝试字段：自动与人工次数均为 0 时显示“待首次投递”，`retryCount > 0` 时显示“等待自动重试”，`manualRetryCount > 0` 时显示“人工重投排队中”。人工重投可重置新一轮自动重试预算，但不得把记录伪装成从未投递。 -- 已实现下游投递告警统一口径：运营看板、侧栏通知、下游投递 Dashboard 和应用告警排行必须基于同一组真实 `CmppDownstreamDelivery` 条件统计：`pending` 超过积压阈值、`awaiting_ack` 超过 `ackDeadlineAt`，以及最近失败窗口内的 `failed/unconfirmed/rejected`。默认积压阈值为 10 分钟，最近失败窗口为 1 小时，可分别通过 `CMPP_DOWNSTREAM_ALERT_PENDING_MINUTES` 和 `CMPP_DOWNSTREAM_ALERT_RECENT_FAILED_HOURS` 覆盖。 +- 下游投递不得无限停留在 `pending`：Gateway 对未写出的结果必须返回明确的 `retryable/reasonCode/errorMessage`。状态回执在原消息映射已经丢失且缺少 `submitSequenceId` 时属于不可恢复错误，立即转为 `failed` 并保留失败原因和操作日志；客户端暂时离线属于可恢复错误，按既有重试次数与指数退避处理。所有 pending 从创建时间或最近人工重投时间起最多保留 72 小时，可通过 `CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS` 调整，超时后自动终结为 `failed`。 +- 新产生的客户侧状态回执投递 payload 必须保存短信原始 `cmppSubmitSequenceId`，使 Gateway 重启后仍能结合账号和平台消息 ID 重建与原 `CMPP_SUBMIT_RESP` 一致的非 0 `Msg_Id`；不得用伪造或为 0 的 `Msg_Id` 冒充成功投递。 +- 已实现下游投递告警统一口径：运营看板、下游投递 Dashboard 和应用告警排行必须基于同一组真实 `CmppDownstreamDelivery` 条件统计：`pending` 超过积压阈值、`awaiting_ack` 超过 `ackDeadlineAt`，以及最近失败窗口内的 `failed/unconfirmed/rejected`。默认积压阈值为 10 分钟，最近失败窗口为 1 小时，可分别通过 `CMPP_DOWNSTREAM_ALERT_PENDING_MINUTES` 和 `CMPP_DOWNSTREAM_ALERT_RECENT_FAILED_HOURS` 覆盖。右上角“待审核任务”仅汇总审核业务，不显示下游投递告警菜单或数字。 +- pending 积压告警必须从本轮排队开始时间计算：从未人工重投的记录使用 `createdAt`，人工重投记录使用 `lastRetriedAt`；刚执行人工重投的记录不得因旧 `createdAt` 立即重新计入积压告警。 - 已实现下游投递 Dashboard 第一版：运营端“下游投递记录”页面顶部新增真实聚合总览，直接按 `tenantId/applicationId/deliveryType` 统计投递总量、pending/awaiting_ack/delivered/failed/unconfirmed/rejected、积压告警、ACK 超时告警、按类型分布、重试压力分布和应用告警排行，数据源必须来自 `CmppDownstreamDelivery`，不能靠前端本地汇总。应用告警排行只统计满足统一告警时间窗的记录，不得将新创建的 `pending` 或超出最近窗口的历史失败永久累加为告警。 - 已实现下游连接映射持久化第一步：Gateway 在客户 CMPP 账号 bind 成功、下游 submit 建链和回执/上行下发时，会把账号在线状态、实例标识、最近活跃时间写入 Redis presence；该状态不再只保留在 Gateway 进程内存中，为后续“Gateway 重启后的 pending 恢复”提供外部状态基础。 - 已实现下游连接映射持久化第二步：Gateway 启动时会读取 Redis presence 与当前内存在线账号，形成“恢复候选视图”，并通过控制面 `GET /downstream/recovery-candidates` 暴露候选账号列表，供后续恢复逻辑与运维排查使用；本阶段仍不等同于自动恢复 pending 投递。 @@ -521,6 +524,7 @@ - 计费规则：支持按短信计费条数和企业应用单价计算费用；发送准入额度为现金余额加授信额度。 - 账务流水必须与短信记录形成可追溯关系，支持对账导出。 - 最终失败、超时失败需要退费。 +- “今日返还金额”按当天实际恢复到企业可用余额的消息级流水汇总：包含已扣费后的 `refunded`，以及提交前失败后按 `sms_message_record` 释放的 `released`；任务冻结转扣费过程中按 `sms_batch_task` 产生的内部释放不得计入返还。 - 计费口径可配置为按提交成功计费或按回执成功计费。 ### 5.20 数据保存与清理 @@ -1346,6 +1350,7 @@ - 原“剩余条数”改为“剩余余额”。 - 原“近24小时成功率”改为“今日发送条数和今日成功率”。 - 新增“今日消费金额”和“今日返还金额”展示。 + - “今日返还金额”展示当天真实返还到可用余额的金额，包含消息级冻结释放和扣费后退款，排除任务冻结转扣费产生的内部释放流水，金额统一保留三位小数。 3. 运营端增加系统日志。 - 运营端可查询全平台系统日志。 - 日志需支持按企业、模块、级别、操作人、资源 ID 或详情定位。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index e0de801..01aa82e 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -36,6 +36,7 @@ | TC-ADMIN-TEMPLATE-0715-06 | P1 | 将光标置于模板中间并插入推荐/自定义变量 | 变量在光标或选区处插入，原选区被替换，光标停在变量后；运营端和客户端一致。 | | TC-ADMIN-RECORD-0715-07 | P1 | 查看桌面/窄屏短信记录及失败详情 | 卡片不产生页面横向滚动，信息分组清晰；失败原因独立突出；详情仍读取真实 submit、receipt 和分片审计 API。 | | TC-ADMIN-MISC-0715-08 | P2 | 查看签名引流信息、零待审核通知和充值弹窗 | 使用“引流信息”标题且无提交时间；0 为黑字灰底；充值弹窗无操作人字段。 | +| TC-ADMIN-NOTICE-0715-09 | P1 | 制造下游投递告警后打开右上角“待审核任务” | 待审核总数和菜单仅包含五类真实审核任务，不出现“下游投递告警”；运营看板和下游投递页仍展示下游告警。 | | 客户 | 正常客户、停用客户、欠费客户、未认证客户、跨租户客户、客户联系人和开票资料。 | | 导入文件 | UTF-8 CSV、GBK CSV、TXT、超 20 MB 文件、含空行/重复/非法号码/非法字符文件。 | | 非法内容 | 控制字符、emoji、换行、不可见字符、超长变量、签名外置内容、敏感词内容。 | @@ -1361,14 +1362,14 @@ - 优先级：P1 - 前置条件：真实 `CmppDownstreamDelivery` 中准备阈值内/外的 `pending`、未超时/已超过 `ackDeadlineAt` 的 `awaiting_ack`、最近窗口内/外的 `failed/unconfirmed/rejected` 及正常 `delivered` 记录，且覆盖多个应用。 - 步骤： - 1. 访问运营端 Dashboard 和右上角通知区域。 + 1. 访问运营端 Dashboard、下游投递记录页和右上角待审核通知区域。 2. 调用真实 `/api/admin/operations/dashboard/statistics`，核对返回的下游投递告警聚合。 - 3. 点击“下游投递告警”通知，跳转到下游投递记录页进一步筛查。 + 3. 打开右上角“待审核任务”，核对菜单和总数只包含审核业务。 - 预期结果： - Dashboard 返回真实 `downstreamDeliverySummary`，至少包含 `pending/failed/delivered/stalledPending/stalledAck/recentFailed/alertCount`。 - `alertCount` 精确等于“超阈值 pending + 超时 awaiting_ack + 最近窗口内 failed/unconfirmed/rejected”，阈值内 pending、未超时 awaiting_ack、历史失败和 delivered 不计入。 - - 侧栏/首页的“下游投递告警”数量与下游投递 Dashboard 在同一筛选范围下一致，不是前端写死值。 - - 点击通知后可以进入真实下游投递记录页继续处理。 + - 运营看板的“下游投递告警”数量与下游投递 Dashboard 在同一筛选范围下一致，不是前端写死值。 + - 右上角“待审核任务”不出现下游投递告警菜单或数字；下游异常从运营看板或下游投递记录页进入处理。 ### TC-GW-018 下游投递 Dashboard 聚合视图 @@ -3254,8 +3255,9 @@ npm run verify:phase8 | TC-BILLING-009 | 无权限用户、审核员、管理员分别执行充值；大额人工充值不走审批。 | 权限不足被拒绝并写失败日志；有权限用户确认后立即入账；不产生 pending 审批态；充值订单、账户余额、流水和日志同步完成。 | | TC-BILLING-010 | 余额不足发送失败，人工充值后重试发送并模拟 delivered。 | 充值前不扣费；充值后发送成功；冻结、扣费、短信计费记录完整；reconciliation diff 为 0。 | | TC-BILLING-011 | 分别准备 `余额+授信` 为正数、0 和负数的账户，使用相同短信费用发起发送。 | 和为正数时允许发送；和为 0 或负数时提示余额不足。判断公式为 `balanceCents + creditCents > 0`，与本次费用和套餐无关。 | -| TC-BILLING-012 | 已扣费短信收到最终失败回执；另一个未提交成功任务只释放冻结。 | 前者只生成一条 refunded 流水并计入“今日返还”；重复回执不重复退款；后者的 released 流水不计入“今日返还”。 | +| TC-BILLING-012 | 已扣费短信收到最终失败回执；另一个消息在提交前失败并释放冻结；另准备一笔任务冻结转扣费时的批次级释放。 | 最终失败只生成一条 `refunded` 并计入“今日返还”，重复回执不重复退款；提交前失败生成 `released + relatedType=sms_message_record` 并计入“今日返还”；冻结转扣费的 `released + relatedType=sms_batch_task` 属于内部转换，不计入“今日返还”；客户端和运营端当日金额一致且保留三位小数。 | | TC-BILLING-013 | 准备已提交扣费但 72 小时完全无回执的 `submitted` 短信，以及有 `UNKNOWN` 回执且超过 72 小时的短信；启动 API 定时扫描并模拟重复扫描。 | 两类短信都转为 timeout 并退款；任务进度刷新；同一短信只退款一次；定时扫描默认启用且每 5 分钟执行。 | +| TC-SEC-006 | 安装 API 生产依赖并执行 `npm audit`；使用缺文件、多文件、超大文件、超量字段和正常单文件调用认证后的 multipart 上传接口。 | NestJS/Multer/Hono 已升级或锁定到修复版本，生产依赖 audit 为 0；接口只接受一个不超过 20MB 的文件，并限制字段、part、字段名、字段值和 header pair 数量；异常请求返回受控 4xx，正常文件仍写入真实 MinIO 和 `FileObject`。 | ### 17.6 系统日志细化 @@ -3370,3 +3372,5 @@ npm run verify:phase8 | TC-GW-ACK-003 | 在企业应用中分别关闭“回执自动重试”和“上行短信自动重试”，各制造一次 ACK 超时，再重新开启并创建新投递。 | 关闭只影响对应类型的新投递策略快照；离线后的首次投递仍会在重连时执行；已写出未确认的记录不自动重发；重新开启后新记录按退避策略重试。 | | TC-GW-ACK-004 | 对 `unconfirmed/rejected/failed/delivered` 记录执行单条和批量手工重投。 | 页面提示重复处理风险并二次确认；真实调用 Gateway；`awaiting_ack` 不允许并发重投；重发复用同一业务 Msg_Id。 | | TC-GW-ACK-005 | 客户 Submit 后由业务校验立即生成失败回执，并覆盖在线即时投递、Gateway 重启后恢复投递；另模拟客户端对 `Msg_Id=0` 返回 Result=0。 | 客户收到的第一个响应包必须是对应 `CMPP_SUBMIT_RESP`，之后 Deliver 的 `Msg_Id` 非 0 且与 SubmitResp 完全一致；重启后根据持久化 Submit Sequence_Id 重建同一 Msg_Id；Result=0/Msg_Id=0 不得写为 delivered。 | +| TC-GW-ACK-006 | 对一条 Gateway 已丢失原消息映射且 payload 缺少 `submitSequenceId` 的历史状态回执执行人工重投；另对字段完整但客户离线的记录重投。 | 缺少序列号的记录由 Gateway 返回 `retryable=false/MISSING_SUBMIT_SEQUENCE_ID`，NestJS 立即终结为 `failed` 并保存明确原因；客户暂时离线的记录返回 `retryable=true/CLIENT_DISCONNECTED`，按次数上限和指数退避继续处理，不得无限保持 `retryCount=0`。 | +| TC-GW-ACK-007 | 构造一条从未重投且 pending 超过 72 小时的记录，以及一条人工重投后尚未满 72 小时但原 `createdAt` 很早的记录，执行自动扫描并查看告警。 | 第一条自动转为 `failed/queue_timeout`；第二条仍保持 pending，终结时间和 10 分钟积压告警均从 `lastRetriedAt` 重新计算，刚重投后不立即告警。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index e9807bb..f10e0f0 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1,5 +1,27 @@ # 第一版系统化测试进度 +## 2026-07-15 依赖安全与今日返还修复（未提交、未部署） + +- 生产数据库只读核查确认目标企业当天共有两笔真实返还：提交前路由失败产生消息级 `released=5` 分，最终失败产生 `refunded=5` 分，正确合计为 10 分（页面应显示 `¥0.100`）。原 Dashboard 和企业列表只聚合 `refunded`，因此少算前一笔并显示 5 分。 +- 客户端 Dashboard 与运营端企业列表统一按当天 `refunded` 加 `released + relatedType=sms_message_record` 汇总；排除任务冻结转扣费时的 `released + relatedType=sms_batch_task`，避免把内部账务转换误当成返还。 +- API 依赖升级到 NestJS `11.1.28`、Multer `2.2.0`，并通过 override 将 `@hono/node-server` 固定为 `1.19.13`；`npm --prefix api audit --json` 已由 3 个 moderate、2 个 high 降为 0。 +- multipart 文件上传补充单文件 20MB、字段数、part 数、字段名/值长度和 header pair 限制，继续使用认证后的真实 NestJS API、MinIO 与 `FileObject` 链路。定向回归 3 suites/21 项通过；完整 API 17 suites/173 项、API build、前端 build、Gateway 全量 Go 测试、Prisma validate/migrate status 和依赖 audit 均通过，43 条 migration 全部齐全。前端仅有既有 Vite chunk size warning。 + +## 2026-07-15 下游人工重投无限排队修复（未提交、未部署） + +- 根因是 Gateway 以空的 `DownstreamSendResult` 同时表达“客户暂时离线”和“原消息映射丢失且缺少 Submit Sequence_Id”，NestJS 收到 `sent=false` 后又不登记失败，导致历史状态回执永久保持 `pending/retryCount=0`。 +- Gateway 控制结果新增 `retryable/reasonCode/errorMessage`：缺少原 `submitSequenceId` 且无法命中内存映射时返回不可恢复的 `MISSING_SUBMIT_SEQUENCE_ID`，NestJS 立即终结为 `failed`；字段完整但客户离线时返回可恢复的 `CLIENT_DISCONNECTED`，失败回调进入真实次数上限与指数退避。 +- 新状态回执入队补存 `SmsMessageRecord.cmppSubmitSequenceId`，Gateway 控制面允许在当前连接存在时立即恢复原 `Msg_Id`；不生成伪造或为 0 的 Msg_Id。 +- 自动扫描为 pending 增加默认 72 小时绝对终点（`CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS` 可覆盖），从 `createdAt` 或最近 `lastRetriedAt` 计算，超时写 `failed/queue_timeout` 和失败操作日志。积压告警同样以最近人工重投时间重新计时，避免刚重投就因旧创建时间立即告警。 +- 回归已通过：API 定向 `send-chain/operations` 2 suites、68 项，API 全量 17 suites、173 项，Gateway 全量 Go 测试、Prisma validate、API build、前端 build 和 `git diff --check`；前端仅有既有 Vite chunk size warning。尚未提交、push 或部署，生产原 3 条历史 pending 记录未修改。 + +## 2026-07-15 待审核通知移除下游投递告警（未提交、未部署） + +- 生产只读核查确认当前“下游投递告警 3”全部为历史状态回执记录：三条均在人工重投后成为 `pending/manualRetryCount=1/retryCount=0`，因为告警仍按原 `createdAt` 超过 10 分钟判断，立即计入 `stalled_pending`；它们不是待审核任务。 +- 右上角“待审核任务”移除下游投递告警菜单项及其数字，通知总数只汇总企业认证、短信、模板、签名和引流信息五类审核。运营看板和下游投递记录页面继续保留独立告警展示。 +- “人工重投排队中”表示记录已由人工重投重置为 `pending` 并累计人工重投次数，正在等待 Gateway 对在线客户写出 Deliver 并取得 `CMPP_DELIVER_RESP`，不代表重投已经成功。 +- 进一步核查三条历史记录的 `payload.submitSequenceId` 均为空；虽然账号 `910887` 当前 connected 且持续心跳，Gateway 重启后没有原 Submit 的消息映射，也无法安全重建非 0 Msg_Id，因此人工重投没有写出 Deliver，继续保持 pending。前端 build 与 `git diff --check` 通过；应用内 Browser 连接因插件运行时初始化失败，未使用独立 Playwright 或 mock 页面替代。 + ## 2026-07-15 当前工作区汇总发布 - 当前各对话产生的 34 个文件变更已统一提交为 `e47432bc631bf37f4c6a6dcb3576ce3c86370b45` 并 push 到 `origin/main`；提交明确排除 `api/tsconfig.build.tsbuildinfo` 和 `logs/`。发布前确认本地 `main` 与最新 `origin/main` 无分叉，API 全量 17 suites/169 项、API build、前端 build、Gateway 全量 Go 测试、Prisma validate/migrate status 和 `git diff --check` 全部通过，前端仅有既有 Vite chunk size warning。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 00e59a1..723b537 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -93,6 +93,9 @@ type DownstreamUplink struct { type DownstreamSendResult struct { Sent bool `json:"sent"` + Retryable bool `json:"retryable"` + ReasonCode string `json:"reasonCode,omitempty"` + ErrorMessage string `json:"errorMessage,omitempty"` ConnectionID string `json:"connectionId,omitempty"` SequenceID string `json:"sequenceId,omitempty"` MessageID string `json:"messageId,omitempty"` @@ -544,7 +547,20 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe result.DeliveredCount++ continue } - result.WaitingCount++ + errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery") + failureType := "unrecoverable" + if sendResult.Retryable { + failureType = "send_failed" + result.WaitingCount++ + } else { + result.FailedCount++ + } + result.LastError = errorMessage + _ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{ + "id": delivery.ID, + "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode), + "failureType": failureType, + }, nil) } return result, nil } @@ -776,6 +792,10 @@ func (s Server) recoverPendingCandidates(logger *log.Logger) { status.State = "partial" status.LastError = result.LastError status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason) + case result.FailedCount > 0: + status.State = "failed" + status.LastError = result.LastError + status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason) default: status.State = "success" status.LastError = "" @@ -861,7 +881,7 @@ func PushReceipt(event DownstreamReceipt) (bool, error) { } func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) { - return pushReceiptWithResult(event, false) + return pushReceiptWithResult(event, true) } func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) { @@ -870,7 +890,25 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr session = recoverReceiptSession(event) } if session == nil { - return DownstreamSendResult{}, nil + if event.SubmitSequenceID == 0 { + return DownstreamSendResult{ + Retryable: false, + ReasonCode: "MISSING_SUBMIT_SEQUENCE_ID", + ErrorMessage: "历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投", + }, nil + } + if strings.TrimSpace(event.MessageID) == "" || strings.TrimSpace(event.Account) == "" { + return DownstreamSendResult{ + Retryable: false, + ReasonCode: "INVALID_RECEIPT_PAYLOAD", + ErrorMessage: "状态回执缺少平台消息 ID 或客户账号，系统已终止重投", + }, nil + } + return DownstreamSendResult{ + Retryable: true, + ReasonCode: "CLIENT_DISCONNECTED", + ErrorMessage: "下游客户端当前未连接，等待自动重试", + }, nil } stat := strings.TrimSpace(event.RawStatus) if stat == "" { @@ -934,7 +972,11 @@ func PushUplink(event DownstreamUplink) (bool, error) { func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) { session := findSession(event.MessageID, event.Account) if session == nil { - return DownstreamSendResult{}, nil + return DownstreamSendResult{ + Retryable: true, + ReasonCode: "CLIENT_DISCONNECTED", + ErrorMessage: "下游客户端当前未连接，等待自动重试", + }, nil } content, err := cmpputils.Utf8ToUcs2(event.Content) if err != nil { @@ -952,6 +994,18 @@ func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) return sendDownstream(session, deliver, event.DeliveryID) } +func errorMessageWithCode(message string, code string) string { + message = strings.TrimSpace(message) + code = strings.TrimSpace(code) + if message == "" { + message = "gateway did not complete downstream delivery" + } + if code == "" { + return message + } + return fmt.Sprintf("%s (%s)", message, code) +} + func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer { if session != nil && (session.protocol == "cmpp20" || session.protocol == "cmpp21") { return &cmpp.Cmpp2DeliverReqPkt{ diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index 521b920..0e5f079 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -302,6 +302,43 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) { } } +func TestReceiptWithoutOriginalSequenceIsUnrecoverable(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + result, err := PushReceiptWithResult(DownstreamReceipt{ + DeliveryID: "delivery-history", + Account: "100001", + MessageID: "MSG-HISTORY", + ReceiptStatus: "undelivered", + }) + if err != nil { + t.Fatalf("push receipt: %v", err) + } + if result.Sent || result.Retryable || result.ReasonCode != "MISSING_SUBMIT_SEQUENCE_ID" { + t.Fatalf("unexpected result: %+v", result) + } +} + +func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + result, err := PushReceiptWithResult(DownstreamReceipt{ + DeliveryID: "delivery-retry", + Account: "100001", + MessageID: "MSG-RETRY", + SubmitSequenceID: 77, + ReceiptStatus: "delivered", + }) + if err != nil { + t.Fatalf("push receipt: %v", err) + } + if result.Sent || !result.Retryable || result.ReasonCode != "CLIENT_DISCONNECTED" { + t.Fatalf("unexpected result: %+v", result) + } +} + func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry() @@ -571,8 +608,10 @@ func TestRecoverPendingCandidatesWritesWaitingConnectionStatus(t *testing.T) { _ = json.NewEncoder(w).Encode([]pendingDelivery{{ ID: "delivery-1", DeliveryType: "receipt", - Payload: json.RawMessage(`{"messageId":"MSG-404","phoneNumber":"13800000001","receiptStatus":"delivered"}`), + Payload: json.RawMessage(`{"account":"100010","messageId":"MSG-404","submitSequenceId":77,"phoneNumber":"13800000001","receiptStatus":"delivered"}`), }}) + case "/api/gateway/events/downstream/failed": + w.WriteHeader(http.StatusOK) case "/api/gateway/events/downstream/recovery-status": w.WriteHeader(http.StatusOK) default: diff --git a/src/api/adminApi.ts b/src/api/adminApi.ts index ec3575a..7f88190 100644 --- a/src/api/adminApi.ts +++ b/src/api/adminApi.ts @@ -240,7 +240,7 @@ export type UserPayload = { export type DashboardResponse = { taskCount: number; messageStatus: Array<{ status: string; _count: { _all: number }; _sum: { amountCents?: number | null; billingUnits?: number | null } }>; - today: { sent: number; delivered: number; failed: number; unknown: number; successRate: number; spendCents: number; billingUnits: number }; + today: { sent: number; delivered: number; failed: number; unknown: number; successRate: number; spendCents: number; returnedCents: number; billingUnits: number }; uplinkCount: number; billing: { _count: { _all: number }; _sum: { amountCents?: number | null; billingUnits?: number | null } }; transactions: { _count: { _all: number }; _sum: { amountCents?: number | null } }; diff --git a/src/apps/client/ClientHome.tsx b/src/apps/client/ClientHome.tsx index d231460..3e9ba42 100644 --- a/src/apps/client/ClientHome.tsx +++ b/src/apps/client/ClientHome.tsx @@ -52,7 +52,8 @@ export function ClientHome() { const account = dashboard?.accounts[0]; const availableBalance = ((account?.balanceCents ?? 0) + (account?.creditCents ?? 0)) / 100; const todaySpend = (dashboard?.today.spendCents ?? 0) / 100; - const todayRefund = Math.abs((dashboard?.transactions._sum.amountCents ?? 0) < 0 ? 0 : dashboard?.transactions._sum.amountCents ?? 0) / 100; + const todayRefundCents = Math.max(0, dashboard?.today.returnedCents ?? 0); + const todayRefund = todayRefundCents / 100; const balanceBaseline = Math.max(availableBalance + todaySpend - todayRefund, availableBalance, 1); const balancePercent = Math.min(100, Math.round((availableBalance / balanceBaseline) * 100)); const recentMessages = useMemo(() => (dashboard?.recentTasks ?? []).map((task) => ({ @@ -113,7 +114,7 @@ export function ClientHome() {
今日返还金额 - ¥{formatAmount(todayRefund)} + ¥{formatCents(todayRefundCents)} 异常回执与退费返还
diff --git a/src/layouts/AdminLayout.tsx b/src/layouts/AdminLayout.tsx index 6e0e4ea..150d00f 100644 --- a/src/layouts/AdminLayout.tsx +++ b/src/layouts/AdminLayout.tsx @@ -38,16 +38,13 @@ import { AppShell } from '@/layouts/AppShell'; export function AdminLayout() { const session = readSession(); const [pendingAudits, setPendingAudits] = useState({ enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 }); - const [downstreamAlertCount, setDownstreamAlertCount] = useState(0); const loadPendingAuditCount = useCallback(() => { adminApi.getDashboard() .then((dashboard) => { setPendingAudits(dashboard.pendingAudits ?? { enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 }); - setDownstreamAlertCount(dashboard.downstreamDeliverySummary?.alertCount ?? 0); }) .catch(() => { setPendingAudits({ enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 }); - setDownstreamAlertCount(0); }); }, []); @@ -83,7 +80,6 @@ export function AdminLayout() { { label: '模板待审', count: pendingAudits.templates, to: '/admin/templates' }, { label: '签名待审', count: pendingAudits.signatures, to: '/admin/signatures' }, { label: '引流信息待审', count: pendingAudits.drainageInfos, to: '/admin/drainage-audits' }, - { label: '下游投递告警', count: downstreamAlertCount, to: '/admin/downstream-deliveries' }, ]} navSections={[ {