From 4994841709ff79ad03f22f8b907e7b997abb969f Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Wed, 12 Aug 2026 17:05:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E4=B8=8B=E6=B8=B8?= =?UTF-8?q?=E9=87=8D=E6=8A=95=E4=B8=8E=E7=AD=BE=E5=90=8D=E8=B4=A8=E9=87=8F?= =?UTF-8?q?=E6=A3=80=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../migration.sql | 57 +++++ api/prisma/schema.prisma | 106 ++++++-- .../operations/admin-operations.controller.ts | 46 ++++ api/src/send-chain/send-chain.service.spec.ts | 43 ++++ api/src/send-chain/send-chain.service.ts | 36 +++ api/src/send-chain/send-completion.service.ts | 4 + ...nd-downstream-requeue-task.service.spec.ts | 68 +++++ .../send-downstream-requeue-task.service.ts | 235 ++++++++++++++++++ api/src/send-chain/send-retry.service.ts | 41 +++ .../signature-retirement.service.spec.ts | 7 + .../signature-retirement.service.ts | 53 ++-- docs/contracts/send-chain-r10-completion.json | 2 +- .../first-version-development-requirements.md | 12 +- docs/signature-retirement-alert-design.md | 2 +- docs/system-functional-test-cases.md | 17 +- docs/testing-progress.md | 19 ++ src/api/admin/operations.api.ts | 13 +- src/api/types/operations.ts | 45 ++++ src/apps/admin/AdminAnalyticsPage.tsx | 4 +- .../admin/AdminDownstreamDeliveriesPage.tsx | 125 +++++++++- .../AdminGatewaySubmitExceptionsPage.tsx | 69 ++++- src/styles/global.css | 11 + tools/quality/verify-send-chain-r10.mjs | 12 +- 23 files changed, 960 insertions(+), 67 deletions(-) create mode 100644 api/prisma/migrations/20260812153000_add_downstream_requeue_tasks/migration.sql create mode 100644 api/src/send-chain/send-downstream-requeue-task.service.spec.ts create mode 100644 api/src/send-chain/send-downstream-requeue-task.service.ts diff --git a/api/prisma/migrations/20260812153000_add_downstream_requeue_tasks/migration.sql b/api/prisma/migrations/20260812153000_add_downstream_requeue_tasks/migration.sql new file mode 100644 index 0000000..c6bac0f --- /dev/null +++ b/api/prisma/migrations/20260812153000_add_downstream_requeue_tasks/migration.sql @@ -0,0 +1,57 @@ +CREATE TABLE "DownstreamRequeueTask" ( + "id" TEXT NOT NULL, + "taskNo" TEXT NOT NULL, + "tenantId" TEXT, + "applicationId" TEXT, + "status" TEXT NOT NULL DEFAULT 'queued', + "filterSnapshot" JSONB NOT NULL, + "snapshotAt" TIMESTAMP(3) NOT NULL, + "reason" TEXT NOT NULL, + "ratePerSecond" INTEGER NOT NULL DEFAULT 10, + "consecutiveFailureLimit" INTEGER NOT NULL DEFAULT 10, + "totalCount" INTEGER NOT NULL DEFAULT 0, + "successCount" INTEGER NOT NULL DEFAULT 0, + "failedCount" INTEGER NOT NULL DEFAULT 0, + "skippedCount" INTEGER NOT NULL DEFAULT 0, + "waitingCount" INTEGER NOT NULL DEFAULT 0, + "consecutiveFailures" INTEGER NOT NULL DEFAULT 0, + "lastError" TEXT, + "createdById" TEXT, + "startedAt" TIMESTAMP(3), + "pausedAt" TIMESTAMP(3), + "finishedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + CONSTRAINT "DownstreamRequeueTask_pkey" PRIMARY KEY ("id") +); + +CREATE TABLE "DownstreamRequeueTaskItem" ( + "id" TEXT NOT NULL, + "taskId" TEXT NOT NULL, + "deliveryId" TEXT NOT NULL, + "applicationId" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'queued', + "previousStatus" TEXT NOT NULL, + "skipReason" TEXT, + "errorMessage" TEXT, + "claimedAt" TIMESTAMP(3), + "completedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + CONSTRAINT "DownstreamRequeueTaskItem_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "DownstreamRequeueTask_taskNo_key" ON "DownstreamRequeueTask"("taskNo"); +CREATE INDEX "DownstreamRequeueTask_status_createdAt_idx" ON "DownstreamRequeueTask"("status", "createdAt"); +CREATE INDEX "DownstreamRequeueTask_applicationId_status_createdAt_idx" ON "DownstreamRequeueTask"("applicationId", "status", "createdAt"); +CREATE INDEX "DownstreamRequeueTask_tenantId_createdAt_idx" ON "DownstreamRequeueTask"("tenantId", "createdAt"); +CREATE UNIQUE INDEX "DownstreamRequeueTaskItem_taskId_deliveryId_key" ON "DownstreamRequeueTaskItem"("taskId", "deliveryId"); +CREATE INDEX "DownstreamRequeueTaskItem_taskId_status_createdAt_idx" ON "DownstreamRequeueTaskItem"("taskId", "status", "createdAt"); +CREATE INDEX "DownstreamRequeueTaskItem_applicationId_status_createdAt_idx" ON "DownstreamRequeueTaskItem"("applicationId", "status", "createdAt"); +CREATE INDEX "DownstreamRequeueTaskItem_deliveryId_status_idx" ON "DownstreamRequeueTaskItem"("deliveryId", "status"); + +ALTER TABLE "DownstreamRequeueTask" ADD CONSTRAINT "DownstreamRequeueTask_tenantId_fkey" FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE SET NULL ON UPDATE CASCADE; +ALTER TABLE "DownstreamRequeueTask" ADD CONSTRAINT "DownstreamRequeueTask_applicationId_fkey" FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE SET NULL ON UPDATE CASCADE; +ALTER TABLE "DownstreamRequeueTask" ADD CONSTRAINT "DownstreamRequeueTask_createdById_fkey" FOREIGN KEY ("createdById") REFERENCES "User"("id") ON DELETE SET NULL ON UPDATE CASCADE; +ALTER TABLE "DownstreamRequeueTaskItem" ADD CONSTRAINT "DownstreamRequeueTaskItem_taskId_fkey" FOREIGN KEY ("taskId") REFERENCES "DownstreamRequeueTask"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "DownstreamRequeueTaskItem" ADD CONSTRAINT "DownstreamRequeueTaskItem_deliveryId_fkey" FOREIGN KEY ("deliveryId") REFERENCES "CmppDownstreamDelivery"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 1137f85..aeacd92 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -43,6 +43,7 @@ model Tenant { smsUplinkMessages SmsUplinkMessage[] smsUplinkMatchCandidates SmsUplinkMatchCandidate[] cmppDownstreamDeliveries CmppDownstreamDelivery[] + downstreamRequeueTasks DownstreamRequeueTask[] cmppDownstreamConnections CmppDownstreamConnection[] cmppConnectionStates CmppConnectionState[] gatewayDownstreamRecoveryStatuses GatewayDownstreamRecoveryStatus[] @@ -98,6 +99,7 @@ model User { createdSmsSendTasks SmsSendTask[] @relation("SmsSendTaskCreator") reviewedSmsSendTasks SmsSendTask[] @relation("SmsSendTaskReviewer") createdSmsBatchTasks SmsBatchTask[] @relation("SmsBatchTaskCreator") + createdDownstreamRequeueTasks DownstreamRequeueTask[] @relation("DownstreamRequeueTaskCreator") releasedPhoneFrequencyHits PhoneFrequencyHit[] @relation("PhoneFrequencyHitReleaser") createdPhoneFrequencyWhitelistEntries PhoneFrequencyWhitelist[] @relation("PhoneFrequencyWhitelistCreator") updatedPhoneFrequencyWhitelistEntries PhoneFrequencyWhitelist[] @relation("PhoneFrequencyWhitelistUpdater") @@ -451,6 +453,7 @@ model SmsApplication { uplinkMessages SmsUplinkMessage[] uplinkMatchCandidates SmsUplinkMatchCandidate[] downstreamDeliveries CmppDownstreamDelivery[] + downstreamRequeueTasks DownstreamRequeueTask[] downstreamConnections CmppDownstreamConnection[] connectionStates CmppConnectionState[] gatewayDownstreamRecoveryStatuses GatewayDownstreamRecoveryStatus[] @@ -1056,20 +1059,20 @@ model DrainageReportMaterial { } model ChannelSignatureReportTask { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String signatureId String channelId String carrier String? approvedAt DateTime? - approvalScope String @default("legacy_channel") - reportType String @default("signature") + approvalScope String @default("legacy_channel") + reportType String @default("signature") drainageItemId String? - status String @default("pending") + status String @default("pending") reason String? createdById String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt signature SmsSignature @relation(fields: [signatureId], references: [id]) channel SmsChannel @relation(fields: [channelId], references: [id]) @@ -1088,22 +1091,22 @@ model ChannelSignatureReportTask { } model SignatureRetirementRule { - id String @id @default(cuid()) - ruleType String - targetId String? - targetKey String @default("") - enabled Boolean @default(true) - mobileWindowDays Int @default(30) - mobileThreshold Int @default(1) - unicomWindowDays Int @default(30) - unicomThreshold Int @default(1) - telecomWindowDays Int @default(30) - telecomThreshold Int @default(1) - messageTemplate String? - version Int @default(1) - createdById String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + ruleType String + targetId String? + targetKey String @default("") + enabled Boolean @default(true) + mobileWindowDays Int @default(30) + mobileThreshold Int @default(1) + unicomWindowDays Int @default(30) + unicomThreshold Int @default(1) + telecomWindowDays Int @default(30) + telecomThreshold Int @default(1) + messageTemplate String? + version Int @default(1) + createdById String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt @@unique([ruleType, targetKey]) @@index([ruleType, enabled]) @@ -2107,6 +2110,7 @@ model CmppDownstreamDelivery { application SmsApplication @relation(fields: [applicationId], references: [id]) messageRecord SmsMessageRecord? @relation(fields: [messageRecordId], references: [id]) attempts CmppDownstreamDeliveryAttempt[] + requeueItems DownstreamRequeueTaskItem[] @@index([tenantId, status, createdAt]) @@index([applicationId, status, createdAt]) @@ -2116,6 +2120,64 @@ model CmppDownstreamDelivery { @@index([status, ackDeadlineAt]) } +model DownstreamRequeueTask { + id String @id @default(cuid()) + taskNo String @unique + tenantId String? + applicationId String? + status String @default("queued") + filterSnapshot Json + snapshotAt DateTime + reason String + ratePerSecond Int @default(10) + consecutiveFailureLimit Int @default(10) + totalCount Int @default(0) + successCount Int @default(0) + failedCount Int @default(0) + skippedCount Int @default(0) + waitingCount Int @default(0) + consecutiveFailures Int @default(0) + lastError String? + createdById String? + startedAt DateTime? + pausedAt DateTime? + finishedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + tenant Tenant? @relation(fields: [tenantId], references: [id]) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + createdBy User? @relation("DownstreamRequeueTaskCreator", fields: [createdById], references: [id]) + items DownstreamRequeueTaskItem[] + + @@index([status, createdAt]) + @@index([applicationId, status, createdAt]) + @@index([tenantId, createdAt]) +} + +model DownstreamRequeueTaskItem { + id String @id @default(cuid()) + taskId String + deliveryId String + applicationId String + status String @default("queued") + previousStatus String + skipReason String? + errorMessage String? + claimedAt DateTime? + completedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + task DownstreamRequeueTask @relation(fields: [taskId], references: [id], onDelete: Cascade) + delivery CmppDownstreamDelivery @relation(fields: [deliveryId], references: [id], onDelete: Restrict) + + @@unique([taskId, deliveryId]) + @@index([taskId, status, createdAt]) + @@index([applicationId, status, createdAt]) + @@index([deliveryId, status]) +} + model CmppDownstreamDeliveryAttempt { id String @id @default(cuid()) deliveryId String diff --git a/api/src/operations/admin-operations.controller.ts b/api/src/operations/admin-operations.controller.ts index a765d41..5bb505a 100644 --- a/api/src/operations/admin-operations.controller.ts +++ b/api/src/operations/admin-operations.controller.ts @@ -225,6 +225,14 @@ export class AdminOperationsController { return this.sendChain.requeueGatewaySubmitDeadLetter(id, { ...body, operatorId }); } + @Post('gateway-submit-dead-letters/:id/resolve') + resolveGatewaySubmitDeadLetter( + @Param('id') id: string, + @CurrentSessionUserId() operatorId?: string, + ) { + return this.sendChain.resolveGatewaySubmitDeadLetter(id, operatorId); + } + @Get('receipt-anomalies') receiptAnomalies( @Query('tenantId') tenantId?: string, @@ -354,6 +362,44 @@ export class AdminOperationsController { batchRequeueDownstreamDeliveries(@Body() body: { ids?: string[] }) { return this.sendChain.batchRequeueDownstreamDeliveries(body.ids ?? []); } + + @Post('downstream-requeue-tasks/preview') + previewDownstreamRequeueTask(@Body() body: { filter?: Record }) { + return this.sendChain.previewDownstreamRequeueTask(body.filter ?? {}); + } + + @Post('downstream-requeue-tasks') + createDownstreamRequeueTask( + @Body() body: { filter?: Record; snapshotAt?: string; reason?: string; ratePerSecond?: number; consecutiveFailureLimit?: number }, + @CurrentSessionUserId() operatorId?: string, + ) { + return this.sendChain.createDownstreamRequeueTask({ + filter: body.filter ?? {}, + snapshotAt: body.snapshotAt ?? '', + reason: body.reason ?? '', + ratePerSecond: body.ratePerSecond, + consecutiveFailureLimit: body.consecutiveFailureLimit, + }, operatorId); + } + + @Get('downstream-requeue-tasks') + listDownstreamRequeueTasks(@Query('status') status?: string, @Query('page') page?: string, @Query('pageSize') pageSize?: string) { + return this.sendChain.listDownstreamRequeueTasks({ status, page: Number(page), pageSize: Number(pageSize) }); + } + + @Get('downstream-requeue-tasks/:id') + getDownstreamRequeueTask(@Param('id') id: string) { + return this.sendChain.getDownstreamRequeueTask(id); + } + + @Post('downstream-requeue-tasks/:id/:action') + changeDownstreamRequeueTaskStatus( + @Param('id') id: string, + @Param('action') action: 'pause' | 'resume' | 'terminate', + @CurrentSessionUserId() operatorId?: string, + ) { + return this.sendChain.changeDownstreamRequeueTaskStatus(id, action, operatorId); + } } @ApiTags('admin-system-logs') diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index c35e9c5..a0a40a3 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -853,12 +853,14 @@ describe('SendChainService', () => { const previousReceiptEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED; const previousScheduledEnabled = process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED; const previousScheduledInterval = process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS; + const previousDownstreamRequeueEnabled = process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED; const { service } = createService(); const dispatch = jest.spyOn(service, 'dispatchDueScheduledTasks').mockResolvedValue({ dispatched: 0, results: [] }); try { process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = 'false'; process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = 'true'; process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = '60000'; + process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED = 'false'; service.onModuleInit(); await jest.advanceTimersByTimeAsync(1_000); expect(dispatch).toHaveBeenCalledTimes(1); @@ -870,6 +872,8 @@ describe('SendChainService', () => { else process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = previousScheduledEnabled; if (previousScheduledInterval === undefined) delete process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS; else process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = previousScheduledInterval; + if (previousDownstreamRequeueEnabled === undefined) delete process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED; + else process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED = previousDownstreamRequeueEnabled; jest.useRealTimers(); } }); @@ -3376,6 +3380,41 @@ describe('SendChainService', () => { expect(service['publishGatewaySubmitCommand']).not.toHaveBeenCalled(); }); + it('marks a pending gateway submit exception as resolved without requeueing it', async () => { + const { service, prisma } = createService(); + prisma.gatewaySubmitDeadLetter.findUnique + .mockResolvedValueOnce({ + id: 'dead-1', + tenantId: 'tenant-1', + status: 'pending', + messageId: 'MSG-1', + submitId: 'SUB-1', + }) + .mockResolvedValueOnce({ id: 'dead-1', status: 'resolved', resolvedStatus: 'manually_resolved' }); + prisma.gatewaySubmitDeadLetter.updateMany.mockResolvedValueOnce({ count: 1 }); + + await expect(service.resolveGatewaySubmitDeadLetter('dead-1', 'user-1')).resolves.toEqual( + expect.objectContaining({ status: 'resolved', resolvedStatus: 'manually_resolved' }), + ); + + expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenCalledWith({ + where: { id: 'dead-1', status: 'pending' }, + data: expect.objectContaining({ + status: 'resolved', + resolvedAt: expect.any(Date), + resolvedStatus: 'manually_resolved', + }), + }); + expect(prisma.operationLog.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + userId: 'user-1', + action: 'gateway.submit_dead_letter_resolved', + resourceId: 'dead-1', + }), + }); + expect(service['publishGatewaySubmitCommand']).not.toHaveBeenCalled(); + }); + it('does not reset a resolved submit exception when Gateway repeats the same dead-letter report', async () => { const { service, prisma } = createService(); @@ -4208,12 +4247,14 @@ describe('SendChainService', () => { jest.useFakeTimers(); const previousEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED; const previousScheduledEnabled = process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED; + const previousDownstreamRequeueEnabled = process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_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'; process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = 'false'; + process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED = 'false'; service.onModuleInit(); await jest.advanceTimersByTimeAsync(60_000); expect(scan).toHaveBeenCalledWith({}); @@ -4224,6 +4265,8 @@ describe('SendChainService', () => { else process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = previousEnabled; if (previousScheduledEnabled === undefined) delete process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED; else process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = previousScheduledEnabled; + if (previousDownstreamRequeueEnabled === undefined) delete process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED; + else process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED = previousDownstreamRequeueEnabled; jest.useRealTimers(); } }); diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 7a9c39c..48a1073 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -19,6 +19,7 @@ import { aggregateReceiptSegmentState, isSameUpstreamEndpointIdentity, receiptEv import type { DownstreamDeliveryQueueRequest } from './downstream-receipt-targets'; import { SendSubmissionService, type SendResourceValidationOptions } from './send-submission.service'; import { SendCompletionService, type SendCompletionFacade } from './send-completion.service'; +import { SendDownstreamRequeueTaskService, type DownstreamRequeueFilter } from './send-downstream-requeue-task.service'; @Injectable() export class SendChainService implements OnModuleInit, OnModuleDestroy { @@ -37,8 +38,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { private inboundLongMessageIntervalTimer?: ReturnType; private upstreamReceiptInboxInitialTimer?: ReturnType; private upstreamReceiptInboxIntervalTimer?: ReturnType; + private downstreamRequeueTaskIntervalTimer?: ReturnType; private readonly submission: SendSubmissionService; private readonly completion: SendCompletionService; + private readonly downstreamRequeueTasks: SendDownstreamRequeueTaskService; constructor( private readonly prisma: PrismaService, @@ -68,6 +71,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { openApi, this as unknown as SendCompletionFacade, ); + this.downstreamRequeueTasks = new SendDownstreamRequeueTaskService(prisma, this); } onModuleInit() { @@ -129,6 +133,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ); this.upstreamReceiptInboxIntervalTimer.unref?.(); } + if (process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_ENABLED !== 'false') { + this.downstreamRequeueTaskIntervalTimer = setInterval( + () => void this.downstreamRequeueTasks.runScan().catch((error) => this.logger.error(`Downstream requeue task scan failed: ${String(error)}`)), + positiveInteger(process.env.CMPP_DOWNSTREAM_REQUEUE_TASK_SCAN_INTERVAL_MS, 1_000), + ); + this.downstreamRequeueTaskIntervalTimer.unref?.(); + } } async onModuleDestroy() { @@ -140,6 +151,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (this.inboundLongMessageIntervalTimer) clearInterval(this.inboundLongMessageIntervalTimer); if (this.upstreamReceiptInboxInitialTimer) clearTimeout(this.upstreamReceiptInboxInitialTimer); if (this.upstreamReceiptInboxIntervalTimer) clearInterval(this.upstreamReceiptInboxIntervalTimer); + if (this.downstreamRequeueTaskIntervalTimer) clearInterval(this.downstreamRequeueTaskIntervalTimer); await this.worker?.close(); await this.sendQueue?.close(); await this.gatewayQueue?.close(); @@ -481,6 +493,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return this.completion.requeueGatewaySubmitDeadLetter(id, data); } + async resolveGatewaySubmitDeadLetter(id: string, operatorId?: string) { + return this.completion.resolveGatewaySubmitDeadLetter(id, operatorId); + } + async recoverStaleGatewaySubmitRequeues(now = new Date()) { return this.completion.recoverStaleGatewaySubmitRequeues(now); } @@ -497,6 +513,26 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return this.completion.batchRequeueDownstreamDeliveries(ids); } + previewDownstreamRequeueTask(filter: DownstreamRequeueFilter) { + return this.downstreamRequeueTasks.preview(filter); + } + + createDownstreamRequeueTask(data: { filter: DownstreamRequeueFilter; snapshotAt: string; reason: string; ratePerSecond?: number; consecutiveFailureLimit?: number }, operatorId?: string) { + return this.downstreamRequeueTasks.create(data, operatorId); + } + + listDownstreamRequeueTasks(query: { status?: string; page?: number; pageSize?: number }) { + return this.downstreamRequeueTasks.list(query); + } + + getDownstreamRequeueTask(id: string) { + return this.downstreamRequeueTasks.get(id); + } + + changeDownstreamRequeueTaskStatus(id: string, action: 'pause' | 'resume' | 'terminate', operatorId?: string) { + return this.downstreamRequeueTasks.changeStatus(id, action, operatorId); + } + async claimUplinkMatchCandidate(uplinkMessageId: string, candidateId: string, operatorId?: string) { return this.completion.claimUplinkMatchCandidate(uplinkMessageId, candidateId, operatorId); } diff --git a/api/src/send-chain/send-completion.service.ts b/api/src/send-chain/send-completion.service.ts index bf1da4f..402d3c9 100644 --- a/api/src/send-chain/send-completion.service.ts +++ b/api/src/send-chain/send-completion.service.ts @@ -158,6 +158,10 @@ export class SendCompletionService { return this.retry.requeueGatewaySubmitDeadLetter(id, data); } + async resolveGatewaySubmitDeadLetter(id: string, operatorId?: string) { + return this.retry.resolveGatewaySubmitDeadLetter(id, operatorId); + } + async recoverStaleGatewaySubmitRequeues(now = new Date()) { return this.retry.recoverStaleGatewaySubmitRequeues(now); } diff --git a/api/src/send-chain/send-downstream-requeue-task.service.spec.ts b/api/src/send-chain/send-downstream-requeue-task.service.spec.ts new file mode 100644 index 0000000..e5cfa07 --- /dev/null +++ b/api/src/send-chain/send-downstream-requeue-task.service.spec.ts @@ -0,0 +1,68 @@ +import { BadRequestException } from '@nestjs/common'; +import { SendDownstreamRequeueTaskService } from './send-downstream-requeue-task.service'; + +function prismaMock(): Record { + const result: Record = { + cmppDownstreamDelivery: { + count: jest.fn(), groupBy: jest.fn(), findFirst: jest.fn(), findMany: jest.fn(), findUnique: jest.fn(), + }, + downstreamRequeueTask: { + findFirst: jest.fn(), findMany: jest.fn(), count: jest.fn(), findUnique: jest.fn(), create: jest.fn(), update: jest.fn(), + }, + downstreamRequeueTaskItem: { + createMany: jest.fn(), groupBy: jest.fn(), findMany: jest.fn(), findFirst: jest.fn(), updateMany: jest.fn(), update: jest.fn(), + }, + operationLog: { create: jest.fn() }, + }; + result.$transaction = jest.fn(async (callback: (tx: unknown) => unknown) => callback(result)); + return result; +} + +let mock: Record; + +describe('SendDownstreamRequeueTaskService', () => { + beforeEach(() => { mock = prismaMock(); }); + + it('previews all matches separately from replayable records', async () => { + mock.cmppDownstreamDelivery.count.mockResolvedValueOnce(12).mockResolvedValueOnce(8); + mock.cmppDownstreamDelivery.groupBy + .mockResolvedValueOnce([{ status: 'pending', _count: { _all: 8 } }, { status: 'delivered', _count: { _all: 4 } }]) + .mockResolvedValueOnce([{ applicationId: 'app-1', _count: { _all: 12 } }]); + mock.cmppDownstreamDelivery.findFirst.mockResolvedValue({ createdAt: new Date('2026-08-11T00:00:00Z') }); + const service = new SendDownstreamRequeueTaskService(mock as never, { requeueDownstreamDelivery: jest.fn() }); + const result = await service.preview({ status: 'all' }); + expect(result).toEqual(expect.objectContaining({ matchedCount: 12, replayableCount: 8, skippedCount: 4, applicationCount: 1 })); + }); + + it('rejects delivered filters and short reasons', async () => { + const service = new SendDownstreamRequeueTaskService(mock as never, { requeueDownstreamDelivery: jest.fn() }); + await expect(service.create({ filter: { status: 'delivered' }, snapshotAt: new Date().toISOString(), reason: '事故恢复' })).rejects.toBeInstanceOf(BadRequestException); + await expect(service.create({ filter: { status: 'pending' }, snapshotAt: new Date().toISOString(), reason: '短' })).rejects.toBeInstanceOf(BadRequestException); + }); + + it('rejects a new task when the same application scope already has an unfinished task', async () => { + mock.downstreamRequeueTask.findFirst.mockResolvedValue({ taskNo: 'DRT-EXISTING' }); + const service = new SendDownstreamRequeueTaskService(mock as never, { requeueDownstreamDelivery: jest.fn() }); + await expect(service.create({ filter: { applicationId: 'app-1', status: 'pending' }, snapshotAt: new Date().toISOString(), reason: '处理历史回执积压' })).rejects.toThrow('DRT-EXISTING'); + }); + + it('skips a delivery that automatic recovery already confirmed before task execution', async () => { + mock.downstreamRequeueTask.findMany.mockResolvedValue([{ id: 'task-1' }]); + mock.downstreamRequeueTask.findUnique + .mockResolvedValueOnce({ id: 'task-1', taskNo: 'DRT-1', status: 'queued', startedAt: null, ratePerSecond: 10, consecutiveFailures: 0, consecutiveFailureLimit: 10 }) + .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValue({ status: 'running' }); + mock.downstreamRequeueTaskItem.findMany + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ id: 'item-1', deliveryId: 'delivery-1' }]) + .mockResolvedValueOnce([]); + mock.downstreamRequeueTaskItem.updateMany.mockResolvedValue({ count: 1 }); + mock.cmppDownstreamDelivery.findUnique.mockResolvedValue({ status: 'delivered', payload: {}, deliveryType: 'receipt', application: { status: 'active', interfaceEnabled: true } }); + mock.downstreamRequeueTaskItem.groupBy.mockResolvedValue([{ status: 'skipped', _count: { _all: 1 } }]); + const requeue = jest.fn(); + const service = new SendDownstreamRequeueTaskService(mock as never, { requeueDownstreamDelivery: requeue }); + await service.runScan(); + expect(requeue).not.toHaveBeenCalled(); + expect(mock.downstreamRequeueTaskItem.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ status: 'skipped', skipReason: '已被客户确认' }) })); + }); +}); diff --git a/api/src/send-chain/send-downstream-requeue-task.service.ts b/api/src/send-chain/send-downstream-requeue-task.service.ts new file mode 100644 index 0000000..324e9cf --- /dev/null +++ b/api/src/send-chain/send-downstream-requeue-task.service.ts @@ -0,0 +1,235 @@ +import { BadRequestException, NotFoundException } from '@nestjs/common'; +import { Prisma } from '@prisma/client'; +import { PrismaService } from '../prisma/prisma.service'; +import { parseDateBoundary } from '../operations/operations.helpers'; + +export type DownstreamRequeueFilter = { + tenantId?: string; + applicationId?: string; + deliveryType?: string; + status?: string; + keyword?: string; + createdAtFrom?: string; + createdAtTo?: string; +}; + +type RequeueFacade = { requeueDownstreamDelivery(id: string): Promise }; +const REPLAYABLE_STATUSES = ['pending', 'failed', 'unconfirmed', 'rejected']; + +function taskWhere(filter: DownstreamRequeueFilter, snapshotAt: Date, replayableByDefault = true): Prisma.CmppDownstreamDeliveryWhereInput { + const from = parseDateBoundary(filter.createdAtFrom, false); + const to = parseDateBoundary(filter.createdAtTo, true); + return { + tenantId: filter.tenantId && filter.tenantId !== 'all' ? filter.tenantId : undefined, + applicationId: filter.applicationId && filter.applicationId !== 'all' ? filter.applicationId : undefined, + deliveryType: filter.deliveryType && filter.deliveryType !== 'all' ? filter.deliveryType : undefined, + status: filter.status && filter.status !== 'all' ? filter.status : replayableByDefault ? { in: REPLAYABLE_STATUSES } : undefined, + createdAt: { ...(from ? { gte: from } : {}), lte: to && to < snapshotAt ? to : snapshotAt }, + OR: filter.keyword ? [ + { messageId: { contains: filter.keyword } }, + { payload: { path: ['account'], string_contains: filter.keyword } }, + { payload: { path: ['phoneNumber'], string_contains: filter.keyword } }, + { lastError: { contains: filter.keyword } }, + { tenant: { name: { contains: filter.keyword } } }, + { application: { name: { contains: filter.keyword } } }, + ] : undefined, + }; +} + +export class SendDownstreamRequeueTaskService { + constructor(private readonly prisma: PrismaService, private readonly facade: RequeueFacade) {} + + async preview(filter: DownstreamRequeueFilter) { + const snapshotAt = new Date(); + const base = taskWhere({ ...filter, status: 'all' }, snapshotAt, false); + const where = taskWhere(filter, snapshotAt); + const [matchedCount, replayableCount, statusGroups, appGroups, oldest] = await Promise.all([ + this.prisma.cmppDownstreamDelivery.count({ where: base }), + this.prisma.cmppDownstreamDelivery.count({ where: { AND: [where, { status: { in: REPLAYABLE_STATUSES } }] } }), + this.prisma.cmppDownstreamDelivery.groupBy({ by: ['status'], where: base, _count: { _all: true } }), + this.prisma.cmppDownstreamDelivery.groupBy({ by: ['applicationId'], where: base, _count: { _all: true } }), + this.prisma.cmppDownstreamDelivery.findFirst({ where: base, orderBy: { createdAt: 'asc' }, select: { createdAt: true } }), + ]); + return { + snapshotAt, + matchedCount, + replayableCount, + skippedCount: matchedCount - replayableCount, + applicationCount: appGroups.length, + oldestCreatedAt: oldest?.createdAt ?? null, + statusCounts: Object.fromEntries(statusGroups.map((item) => [item.status, item._count._all])), + filter: { ...filter, status: filter.status ?? 'all' }, + }; + } + + async create(data: { filter: DownstreamRequeueFilter; snapshotAt: string; reason: string; ratePerSecond?: number; consecutiveFailureLimit?: number }, createdById?: string) { + const reason = data.reason?.trim(); + if (!reason || reason.length < 5) throw new BadRequestException('任务原因至少填写5个字'); + const snapshotAt = new Date(data.snapshotAt); + if (Number.isNaN(snapshotAt.getTime()) || snapshotAt.getTime() > Date.now() + 10_000) throw new BadRequestException('预检快照时间无效'); + if (data.filter.status === 'delivered' || data.filter.status === 'awaiting_ack') throw new BadRequestException('第一版后台任务不支持已确认或等待ACK记录'); + const activeTask = await this.prisma.downstreamRequeueTask.findFirst({ where: { + status: { in: ['queued', 'running', 'paused'] }, + ...(data.filter.applicationId && data.filter.applicationId !== 'all' + ? { OR: [{ applicationId: data.filter.applicationId }, { applicationId: null }] } + : {}), + }, select: { taskNo: true } }); + if (activeTask) throw new BadRequestException(`当前应用范围已有未结束任务 ${activeTask.taskNo}`); + const where = { AND: [taskWhere(data.filter, snapshotAt), { status: { in: REPLAYABLE_STATUSES } }] } as Prisma.CmppDownstreamDeliveryWhereInput; + const deliveries = await this.prisma.cmppDownstreamDelivery.findMany({ where, orderBy: [{ createdAt: 'asc' }, { id: 'asc' }], take: 100001, select: { id: true, tenantId: true, applicationId: true, status: true } }); + if (!deliveries.length) throw new BadRequestException('当前筛选条件下没有可重投记录'); + if (deliveries.length > 100000) throw new BadRequestException('单个任务最多处理100000条，请缩小日期范围'); + const ratePerSecond = Math.min(50, Math.max(1, Number(data.ratePerSecond ?? 10))); + const failureLimit = Math.min(100, Math.max(1, Number(data.consecutiveFailureLimit ?? 10))); + const taskNo = `DRT-${Date.now()}-${Math.floor(Math.random() * 1000).toString().padStart(3, '0')}`; + const task = await this.prisma.$transaction(async (tx) => { + const created = await tx.downstreamRequeueTask.create({ data: { + taskNo, + tenantId: data.filter.tenantId && data.filter.tenantId !== 'all' ? data.filter.tenantId : null, + applicationId: data.filter.applicationId && data.filter.applicationId !== 'all' ? data.filter.applicationId : null, + filterSnapshot: data.filter as Prisma.InputJsonValue, + snapshotAt, reason, ratePerSecond, consecutiveFailureLimit: failureLimit, + totalCount: deliveries.length, createdById, + } }); + await tx.downstreamRequeueTaskItem.createMany({ data: deliveries.map((item) => ({ taskId: created.id, deliveryId: item.id, applicationId: item.applicationId, previousStatus: item.status })) }); + await tx.operationLog.create({ data: { userId: createdById, action: 'gateway.downstream_requeue_task_created', resource: 'downstream_requeue_task', resourceId: created.id, detail: { taskNo, reason, totalCount: deliveries.length, snapshotAt, filter: data.filter, ratePerSecond } } }); + return created; + }); + return this.get(task.id); + } + + async list(query: { status?: string; page?: number; pageSize?: number }) { + const page = Math.max(1, Number(query.page ?? 1)); + const pageSize = Math.min(50, Math.max(1, Number(query.pageSize ?? 10))); + const where = { status: query.status && query.status !== 'all' ? query.status : undefined }; + const [items, total] = await Promise.all([ + this.prisma.downstreamRequeueTask.findMany({ where, include: { tenant: true, application: true, createdBy: { select: { id: true, displayName: true, username: true } } }, orderBy: { createdAt: 'desc' }, skip: (page - 1) * pageSize, take: pageSize }), + this.prisma.downstreamRequeueTask.count({ where }), + ]); + return { items, total, page, pageSize }; + } + + async get(id: string) { + const task = await this.prisma.downstreamRequeueTask.findUnique({ where: { id }, include: { tenant: true, application: true, createdBy: { select: { id: true, displayName: true, username: true } } } }); + if (!task) throw new NotFoundException('后台重投任务不存在'); + const [itemGroups, recentItems] = await Promise.all([ + this.prisma.downstreamRequeueTaskItem.groupBy({ by: ['status'], where: { taskId: id }, _count: { _all: true } }), + this.prisma.downstreamRequeueTaskItem.findMany({ where: { taskId: id }, include: { delivery: { select: { messageId: true, deliveryType: true, status: true, lastError: true } } }, orderBy: { updatedAt: 'desc' }, take: 50 }), + ]); + return { ...task, itemCounts: Object.fromEntries(itemGroups.map((item) => [item.status, item._count._all])), recentItems }; + } + + async changeStatus(id: string, action: 'pause' | 'resume' | 'terminate', operatorId?: string) { + if (!['pause', 'resume', 'terminate'].includes(action)) throw new BadRequestException('不支持的任务操作'); + const task = await this.prisma.downstreamRequeueTask.findUnique({ where: { id } }); + if (!task) throw new NotFoundException('后台重投任务不存在'); + const allowed = action === 'pause' ? ['queued', 'running'] : action === 'resume' ? ['paused'] : ['queued', 'running', 'paused']; + if (!allowed.includes(task.status)) throw new BadRequestException('当前任务状态不允许此操作'); + const status = action === 'pause' ? 'paused' : action === 'resume' ? 'queued' : 'terminated'; + const updated = await this.prisma.downstreamRequeueTask.update({ where: { id }, data: { status, pausedAt: status === 'paused' ? new Date() : null, finishedAt: status === 'terminated' ? new Date() : undefined } }); + if (status === 'terminated') await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { taskId: id, status: 'queued' }, data: { status: 'unprocessed', skipReason: '任务已终止', completedAt: new Date() } }); + await this.prisma.operationLog.create({ data: { userId: operatorId, action: `gateway.downstream_requeue_task_${action}`, resource: 'downstream_requeue_task', resourceId: id, detail: { taskNo: task.taskNo, previousStatus: task.status, status } } }); + return updated; + } + + async runScan() { + const tasks = await this.prisma.downstreamRequeueTask.findMany({ where: { status: { in: ['queued', 'running'] } }, orderBy: { createdAt: 'asc' }, take: 3 }); + for (const task of tasks) await this.processTask(task.id); + } + + private async processTask(taskId: string) { + const task = await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId } }); + if (!task || !['queued', 'running'].includes(task.status)) return; + await this.reconcileWaiting(taskId); + await this.prisma.downstreamRequeueTask.update({ where: { id: taskId }, data: { status: 'running', startedAt: task.startedAt ?? new Date() } }); + const items = await this.prisma.downstreamRequeueTaskItem.findMany({ where: { taskId, status: 'queued' }, orderBy: { createdAt: 'asc' }, take: Math.min(10, task.ratePerSecond), select: { id: true, deliveryId: true } }); + let consecutiveFailures = task.consecutiveFailures; + for (const item of items) { + const latestTask = await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId }, select: { status: true } }); + if (latestTask?.status === 'paused' || latestTask?.status === 'terminated') break; + const claimed = await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { id: item.id, status: 'queued' }, data: { status: 'processing', claimedAt: new Date() } }); + if (!claimed.count) continue; + try { + const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ + where: { id: item.deliveryId }, + include: { application: { select: { status: true, interfaceEnabled: true } } }, + }); + if (!delivery) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'skipped', skipReason: '投递记录已不存在', completedAt: new Date() } }); + continue; + } + if (!REPLAYABLE_STATUSES.includes(delivery.status)) { + if (delivery.status === 'awaiting_ack') { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'waiting_external_ack', skipReason: null } }); + } else { + const skipReason = delivery.status === 'delivered' ? '已被客户确认' : '执行前状态已变化'; + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'skipped', skipReason, completedAt: new Date() } }); + } + continue; + } + if (delivery.application.status !== 'active' || !delivery.application.interfaceEnabled) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'skipped', skipReason: '应用或投递能力已停用', completedAt: new Date() } }); + continue; + } + if (!delivery.payload || !['receipt', 'uplink'].includes(delivery.deliveryType)) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'skipped', skipReason: '投递数据不完整', completedAt: new Date() } }); + continue; + } + const activeOther = await this.prisma.downstreamRequeueTaskItem.findFirst({ where: { deliveryId: item.deliveryId, id: { not: item.id }, status: { in: ['processing', 'waiting_ack', 'success'] } }, select: { id: true } }); + if (activeOther) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'skipped', skipReason: '已被其他任务处理', completedAt: new Date() } }); + continue; + } + const result = await this.facade.requeueDownstreamDelivery(item.deliveryId) as { status?: string; lastError?: string | null }; + if (result?.status === 'awaiting_ack' || result?.status === 'delivered') { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: result.status === 'delivered' ? 'success' : 'waiting_ack', completedAt: result.status === 'delivered' ? new Date() : null } }); + consecutiveFailures = 0; + } else { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'failed', errorMessage: result?.lastError ?? 'Gateway未进入等待ACK状态', completedAt: new Date() } }); + consecutiveFailures += 1; + } + } catch (error) { + const message = error instanceof Error ? error.message : '后台重投失败'; + const skipReason = /已被其他操作处理|状态|等待客户端确认/.test(message) ? '执行前状态已变化' + : /payload|投递类型/.test(message) ? '投递数据不完整' + : /Submit|Msg_Id|Sequence/.test(message) ? '缺少原Submit映射，无法安全重投' + : null; + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: skipReason ? 'skipped' : 'failed', skipReason, errorMessage: skipReason ? null : message, completedAt: new Date() } }); + if (!skipReason) consecutiveFailures += 1; + } + await this.prisma.downstreamRequeueTask.update({ where: { id: taskId }, data: { consecutiveFailures } }); + if (consecutiveFailures >= task.consecutiveFailureLimit) { + // Stop before claiming another delivery: a customer or Gateway outage must not become a retry flood. + const pausedAt = new Date(); + await this.prisma.downstreamRequeueTask.update({ where: { id: taskId }, data: { status: 'paused', pausedAt, lastError: `连续失败达到安全阈值 ${task.consecutiveFailureLimit} 条，任务已自动暂停` } }); + await this.prisma.operationLog.create({ data: { action: 'gateway.downstream_requeue_task_auto_paused', resource: 'downstream_requeue_task', resourceId: taskId, detail: { taskNo: task.taskNo, consecutiveFailures, failureLimit: task.consecutiveFailureLimit } } }); + break; + } + } + await this.reconcileWaiting(taskId); + await this.refreshTask(taskId); + } + + private async reconcileWaiting(taskId: string) { + const items = await this.prisma.downstreamRequeueTaskItem.findMany({ where: { taskId, status: { in: ['waiting_ack', 'waiting_external_ack'] } }, include: { delivery: { select: { status: true, ackResult: true, ackDeadlineAt: true, lastError: true } } }, take: 100 }); + const now = new Date(); + for (const item of items) { + if (item.delivery.status === 'delivered' && item.delivery.ackResult === 0) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: item.status === 'waiting_external_ack' ? { status: 'skipped', skipReason: '已由其他投递链路完成', completedAt: now } : { status: 'success', completedAt: now } }); + } else if (['failed', 'rejected', 'unconfirmed'].includes(item.delivery.status) || (item.delivery.ackDeadlineAt && item.delivery.ackDeadlineAt <= now)) { + await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: item.status === 'waiting_external_ack' ? { status: 'queued', skipReason: null, errorMessage: null, claimedAt: null } : { status: 'failed', errorMessage: item.delivery.lastError ?? '客户端ACK失败或超时', completedAt: now } }); + } + } + } + + private async refreshTask(taskId: string) { + const groups = await this.prisma.downstreamRequeueTaskItem.groupBy({ by: ['status'], where: { taskId }, _count: { _all: true } }); + const counts = new Map(groups.map((item) => [item.status, item._count._all])); + const queued = counts.get('queued') ?? 0; + const active = (counts.get('processing') ?? 0) + (counts.get('waiting_ack') ?? 0) + (counts.get('waiting_external_ack') ?? 0); + const failed = counts.get('failed') ?? 0; + const current = await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId }, select: { status: true } }); + const status = current?.status === 'paused' || current?.status === 'terminated' ? current.status : queued + active === 0 ? (failed > 0 ? 'partial_completed' : 'completed') : 'running'; + await this.prisma.downstreamRequeueTask.update({ where: { id: taskId }, data: { status, successCount: counts.get('success') ?? 0, failedCount: failed, skippedCount: counts.get('skipped') ?? 0, waitingCount: active, ...(status === 'completed' || status === 'partial_completed' ? { finishedAt: new Date() } : {}) } }); + } +} diff --git a/api/src/send-chain/send-retry.service.ts b/api/src/send-chain/send-retry.service.ts index c213a2e..76bdd62 100644 --- a/api/src/send-chain/send-retry.service.ts +++ b/api/src/send-chain/send-retry.service.ts @@ -166,6 +166,47 @@ export class SendRetryService { return updated; } + async resolveGatewaySubmitDeadLetter(id: string, operatorId?: string) { + const deadLetter = await this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } }); + if (!deadLetter) { + throw new NotFoundException('Gateway提交异常记录不存在'); + } + if (deadLetter.status === 'resolved') { + return deadLetter; + } + if (deadLetter.status !== 'pending') { + throw new BadRequestException('只有待处理的提交异常可以标记为已处理'); + } + const resolvedAt = new Date(); + const resolved = await this.prisma.gatewaySubmitDeadLetter.updateMany({ + where: { id, status: 'pending' }, + data: { + status: 'resolved', + resolvedAt, + resolvedStatus: 'manually_resolved', + }, + }); + if (resolved.count !== 1) { + throw new BadRequestException('该提交异常已被其他操作处理，请刷新后重试'); + } + await this.prisma.operationLog.create({ + data: { + tenantId: deadLetter.tenantId ?? undefined, + userId: operatorId, + action: 'gateway.submit_dead_letter_resolved', + resource: 'gateway_submit_dead_letter', + resourceId: deadLetter.id, + detail: { + previousStatus: deadLetter.status, + resolvedStatus: 'manually_resolved', + messageId: deadLetter.messageId, + submitId: deadLetter.submitId, + }, + }, + }); + return this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } }); + } + async recoverStaleGatewaySubmitRequeues(now = new Date()) { const staleCutoff = new Date(now.getTime() - positiveInteger( process.env.GATEWAY_SUBMIT_REQUEUE_STALE_MS, diff --git a/api/src/signature-retirement/signature-retirement.service.spec.ts b/api/src/signature-retirement/signature-retirement.service.spec.ts index 36f1caf..7737317 100644 --- a/api/src/signature-retirement/signature-retirement.service.spec.ts +++ b/api/src/signature-retirement/signature-retirement.service.spec.ts @@ -112,6 +112,13 @@ describe('SignatureRetirementService dimensions', () => { page: 2, pageSize: 10, }); + const query = prisma.$queryRaw.mock.calls[0]?.[0] as { strings?: readonly string[] }; + const sql = query.strings?.join('?') ?? ''; + expect(sql).toContain("SUBSTRING(message.content FROM '^【[^【】]+】')"); + expect(sql).toContain('message."signatureId" IS NULL'); + expect(sql).toContain('FROM "SmsSignature" signature'); + expect(sql).toContain('signature."applicationId" = extracted.application_id'); + expect(sql).not.toContain('FROM "ChannelSignatureReportTask" report'); }); it('returns a filtered historical message page with application metadata', async () => { diff --git a/api/src/signature-retirement/signature-retirement.service.ts b/api/src/signature-retirement/signature-retirement.service.ts index fee79ed..e216eee 100644 --- a/api/src/signature-retirement/signature-retirement.service.ts +++ b/api/src/signature-retirement/signature-retirement.service.ts @@ -270,49 +270,44 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy messageCount: number; rowCount: number; }>>(Prisma.sql` - WITH unreported AS ( + WITH extracted AS ( SELECT - signature.id AS signature_id, - signature.name AS signature_name, + message."tenantId" AS tenant_id, + message."applicationId" AS application_id, + SUBSTRING(message.content FROM '^【[^【】]+】') AS signature_name + FROM "SmsMessageRecord" message + WHERE message."queuedAt" >= ${shanghaiStart(date)} + AND message."queuedAt" < ${shanghaiStart(addDays(date, 1))} + AND message."signatureId" IS NULL + ), unreported AS ( + SELECT + CONCAT('unregistered:', MD5(extracted.tenant_id || ':' || extracted.application_id || ':' || extracted.signature_name)) AS signature_id, + extracted.signature_name, tenant.id AS tenant_id, tenant.name AS tenant_name, application.id AS application_id, application.name AS application_name, COUNT(*)::integer AS message_count - FROM "SmsMessageRecord" message - JOIN "SmsSignature" signature ON signature.id = message."signatureId" - JOIN "Tenant" tenant ON tenant.id = signature."tenantId" - LEFT JOIN "SmsApplication" application ON application.id = message."applicationId" - WHERE message."queuedAt" >= ${shanghaiStart(date)} - AND message."queuedAt" < ${shanghaiStart(addDays(date, 1))} + FROM extracted + JOIN "Tenant" tenant ON tenant.id = extracted.tenant_id + JOIN "SmsApplication" application ON application.id = extracted.application_id + WHERE extracted.signature_name IS NOT NULL + -- 未报备签名指系统签名库中不存在，而不是已有签名缺少某个通道的运营商报备。 AND NOT EXISTS ( SELECT 1 - FROM "ChannelSignatureReportTask" report - JOIN "SmsChannel" channel ON channel.id = report."channelId" - WHERE report."signatureId" = message."signatureId" - AND report."reportType" = 'signature' - AND report.status = 'approved' - AND channel.status <> 'deleted' - AND ( - report."approvalScope" = 'legacy_channel' - OR ( - report."approvalScope" = 'carrier_specific' - AND report.carrier = CASE - WHEN LOWER(COALESCE(message.carrier, '')) IN ('mobile', 'cmcc', '移动', '中国移动') THEN 'mobile' - WHEN LOWER(COALESCE(message.carrier, '')) IN ('unicom', 'cucc', '联通', '中国联通') THEN 'unicom' - WHEN LOWER(COALESCE(message.carrier, '')) IN ('telecom', 'ctcc', '电信', '中国电信') THEN 'telecom' - ELSE '__unknown__' - END - ) - ) + FROM "SmsSignature" signature + WHERE signature."tenantId" = extracted.tenant_id + AND signature."applicationId" = extracted.application_id + AND signature.name = extracted.signature_name + AND signature."auditStatus" <> 'deleted' ) AND ( ${keyword}::text IS NULL - OR signature.name ILIKE ${keywordPattern} + OR extracted.signature_name ILIKE ${keywordPattern} OR tenant.name ILIKE ${keywordPattern} OR application.name ILIKE ${keywordPattern} ) - GROUP BY signature.id, signature.name, tenant.id, tenant.name, application.id, application.name + GROUP BY extracted.signature_name, tenant.id, tenant.name, application.id, application.name ) SELECT signature_id AS "signatureId", diff --git a/docs/contracts/send-chain-r10-completion.json b/docs/contracts/send-chain-r10-completion.json index e45a4f8..84911f7 100644 --- a/docs/contracts/send-chain-r10-completion.json +++ b/docs/contracts/send-chain-r10-completion.json @@ -1,7 +1,7 @@ { "version": "R10", "generatedAt": "2026-08-03", - "source": "api/src/send-chain/send-chain.service.ts at R9 local baseline", + "source": "api/src/send-chain/send-chain.service.ts at R9 local baseline; facade extended by 5 downstream requeue task methods on 2026-08-12", "facade": "api/src/send-chain/send-completion.service.ts", "methods": [ { diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index b0bf0b3..50eb32b 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -1917,6 +1917,7 @@ - 已按整条级成功形成`delivered`终态后，如果同一提交尝试又收到明确失败回执，平台保留原始回执和分片审计，但不得自动把已送达终态改成失败、重复退款或向客户推送互相矛盾的失败结果；系统按稳定异常键写入`SmsReceiptAnomaly`，重复矛盾回执累加发生次数。 - 运营菜单“Gateway提交异常”更名为“网关异常”，原路由保持兼容。页面使用“提交异常”和“回执异常”两个Tab，均查询真实后端和PostgreSQL；每个Tab必须在标题区说明其数据来源、业务含义、不能代表的结论及人工处理注意事项，避免运营人员间隔较久后误判。 - “提交异常”展示Gateway消费提交命令连续失败且没有明确供应商提交结果的死信，可在严格确认未被供应商接收后重新入队；“回执异常”展示供应商回执与平台既有终态冲突的结构化异常判定。回执异常详情必须指明真实回执保存在回执记录、原始CMPP报文在通讯交互日志，不得用异常摘要替代原始证据。 +- “提交异常”的待处理记录允许运营人员标记为“已处理”。操作只将异常状态原子更新为`resolved`并记录处理人、处理时间和操作日志，不删除异常证据、不重新入队、也不发送短信；已进入重新入队流程的记录不得并发标记。 ## 运营端休眠唤醒与会话锁定恢复（2026-08-09） @@ -2015,7 +2016,7 @@ - 页面模块顺序固定为“签名通道发送质量”在最上方，其后依次为企业、通道热力图。两张热力图按维度行各自独立分页，每页10行；翻动其中一张不得改变另一张页码，30日日期列继续在各自表格内横向滚动。 - 两张热力图的日期列从左到右按日期由大到小展示，即从`T-1`依次到`T-30`。行首主信息只展示签名名称；通道热力图保留识别维度所必需的通道名称和运营商标签，企业名称与企业应用名称不在行内常驻，鼠标悬停签名时再展示。每张热力图内部提供独立搜索框，可按企业名称、企业应用名称或签名名称筛选，并在筛选后回到第一页，不影响另一张热力图。 - 热力图有真实检测快照的发送量格子悬停文案必须明确区分“提交条数”和“发送成功条数”，同时可补充上游接受条数、成功率和阈值；不得把上游接受或`SubmitResp status=0`写成发送成功。报备前仍显示“不适用”，没有检测快照仍显示当日无快照。 -- 页面最下方增加“未报备签名”模块，按所选北京时间自然日统计已提交到平台且有签名的真实`SmsMessageRecord`。当短信号码运营商没有匹配到该签名当前可用的运营商级报备成功事实，且也没有仍处于通过状态的历史通道级兼容报备事实时，计入未报备短信；运营商级事实可位于任一未删除通道，历史兼容事实仅用于避免把旧系统真实通过误报为未报备。结果按“签名 × 实际企业应用”聚合业务短信条数，展示签名、企业、企业应用，支持按三者搜索和每页10行独立分页；无签名的异常消息不在该模块伪造成签名。 +- 页面最下方增加“未报备签名”模块，按所选北京时间自然日统计真实`SmsMessageRecord`中“短信正文以规范`【签名】`开头，但该企业应用的有效签名库中没有同名记录”的业务短信。判定不再依赖通道或运营商报备任务：已有系统签名、仅缺少通道/运营商报备成功事实的短信不进入本模块；无法从正文开头提取规范签名的异常消息也不得伪造成签名。结果按“正文签名 × 实际企业应用”聚合号码级业务短信条数，展示签名、企业、企业应用，支持按三者搜索和每页10行独立分页。 - 数据统计菜单改名为“签名质量检测”，并删除企业应用发送排行、通道占比、当天发送量和当天成功率模块。本项删除已确认，不作为后续可选项保留。 ## 通道组按通道筛选（2026-08-09） @@ -2023,3 +2024,12 @@ - 运营端“短信通道组管理”在通道组名称条件之外增加“通道”筛选。选定一个通道后，只展示成员配置中真实包含该`channelId`的未删除通道组；与通道组名称同时输入时按两个条件取交集。 - 通道选项必须来自真实通道API，不使用静态列表、Mock或localStorage。页面首次加载时通道组与通道两个独立请求并行执行；选项同时展示通道名称和编码，已删除通道明确标记“已删除”。未加入任何通道组的真实通道仍可选择，选中后结果为零而不得隐藏该选项。 - 通道选择控件必须为通用下拉可搜索控件，支持按通道名称或编码搜索；“全部通道”表示不按通道限制，点击“重置”必须同时清空通道组名称和通道条件并回到第一页。 +# 下游投递后台重投任务（2026-08-12） + +1. 运营端“下游投递记录”必须同时保留单条重投、当前页勾选批量重投，并新增“按筛选条件重投”；分页支持每页 `10/25/50` 条，切换后回到第一页并重新查询真实后端。 +2. 后台任务使用当前企业、应用、投递类型、状态、创建日期和关键词的后端筛选快照，分页不属于任务范围；任务创建时固定 `snapshotAt`，之后产生的记录不得被卷入。 +3. 第一版只允许 `pending/failed/unconfirmed/rejected`，不支持批量重投客户端已确认的 `delivered`，`awaiting_ack` 不得并发重投。创建前必须真实预检命中、可重投、跳过和状态分布，原因必填。 +4. 任务按应用分批执行，默认每秒 10 条；单条失败不阻断整批，连续失败达到 10 条或 ACK 超时/拒绝达到安全阈值时自动暂停。客户离线、等待 ACK 属于等待状态，不得误记为跳过。 +5. 跳过只表示未调用 Gateway，第一版原因包括：状态已变化、已被客户确认、已被其他任务处理、本任务已成功处理、不属于任务快照、应用或投递能力已停用、投递数据不完整、缺少原 Submit 映射。 +6. 任务必须支持列表、详情、暂停、继续和终止；终止只影响尚未发送的记录。任务项以 `taskId + deliveryId` 幂等，执行前原子认领并复核状态，API 重启后可继续，成功 ACK 的项目不得再次发送。 +7. 所有创建、暂停、继续、终止和自动暂停均写操作日志；任务使用真实 PostgreSQL、Gateway 和客户 ACK，不得使用 mock、静态数据或 localStorage。 diff --git a/docs/signature-retirement-alert-design.md b/docs/signature-retirement-alert-design.md index 864f20c..8c2aee5 100644 --- a/docs/signature-retirement-alert-design.md +++ b/docs/signature-retirement-alert-design.md @@ -171,7 +171,7 @@ 热力图日期列按`T-1`至`T-30`由新到旧排列；行首只常驻签名以及通道维度必要的通道/运营商标识，企业和企业应用通过签名悬停提示查看。企业、通道模块分别在前端对后端返回的真实维度按企业、企业应用、签名过滤并独立分页。每行增加30日上游受理业务短信合计并按合计降序排列。格子悬停明确展示提交条数和最终发送成功条数，不混淆上游接受。报备后的完整观察窗口只控制清退预警资格，观察期仍按日生成`observing`快照并展示真实发送量，不创建预警周期、站内消息或Webhook。 -页面底部增加独立的未报备签名查询。后端按所选北京时间自然日，以`SmsMessageRecord`为业务短信事实，按短信运营商检查该签名是否存在任一未删除通道上的当前运营商级`approved`任务；仍处于`approved`的历史通道级兼容任务视为已有真实旧报备，避免迁移期误报。其余按签名和消息实际企业应用聚合，后端完成关键字过滤、总数和分页，前端不使用静态数据或全量拉取伪分页。 +页面底部增加独立的未报备签名查询。后端按所选北京时间自然日，以`SmsMessageRecord`为业务短信事实，从正文开头提取规范`【签名】`；仅当消息没有关联签名且当前企业应用的有效签名库不存在同名记录时计入。该模块用于发现类似`【湘银物业】`的系统外签名，不再检查通道或运营商报备任务。结果按正文签名和消息实际企业应用聚合，后端完成关键字过滤、总数和分页，前端不使用静态数据或全量拉取伪分页。 ### 第16步：完整验证与分阶段预生产发布 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 4d71f97..f4f281e 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3627,6 +3627,7 @@ npm run verify:phase8 | --- | --- | --- | | TC-GW-SUBMIT-EXCEPTION-001 | 制造一条超过 Gateway 最大处理次数的真实 `SubmitCommand`，打开运营端“网关异常”的“提交异常”Tab，按状态、应用、通道和关键字筛选并查看详情。 | 记录写入 PostgreSQL，页面汇总、分页和详情来自 NestJS API；手机号脱敏，命令中的密码、密钥和原始 payload 不返回浏览器，页面不使用“死信”作为业务名称。 | | TC-GW-SUBMIT-EXCEPTION-002 | 对短信仍处于 pending/failed、通道 active 且 connected 的异常记录，输入 5～500 字原因，勾选“已确认上游未受理”并重新入队。 | 近期认证通过后服务端原子抢占记录、真实写入 Redis Stream；记录变为 requeued，人工次数、操作人、原因、Stream ID 和时间完整留痕，收到 SubmitResult 后变为 resolved。 | +| TC-GW-SUBMIT-EXCEPTION-004 | 对一条`pending`提交异常点击“已处理”，在确认弹窗中取消后再次确认。 | 取消不调用接口；确认后仅将记录原子更新为`resolved/manually_resolved`，保留原异常和命令证据并写操作日志，不写Redis Stream、不触发短信提交；非`pending`记录不展示按钮且后端拒绝并发变更。 | | TC-GW-SUBMIT-EXCEPTION-003 | 不勾选确认、原因过短、重复点击同一记录，或分别把短信置为 accepted/submitted/delivered/unknown、把通道置为停用/断开、人工重试达到 3 次后尝试重新入队。 | API 拒绝危险或重复操作，不产生额外 Stream 命令；页面显示可读原因，操作日志不伪造成功。 | | TC-GW-RATE-001 | 给通道 A 配置 10 TPS，连续投递 20 条；通道 B 同时配置 20 TPS 并投递，另让提交命令携带高于通道配置的数值。 | Gateway A 实际提交节奏不超过 10 TPS，B 独立按自身额度执行；消息值不能放大 A 的权威上限，同一通道跨通道组共享额度。 | | TC-GW-RATE-002 | 超过通道 TPS 后观察 Redis Stream consumer group，并在存在等待消息时重启 Gateway。 | 超流速消息保留在 Stream pending，不直接失败；重启后通过 PEL/XAUTOCLAIM 恢复并继续按通道 TPS 排队提交，不丢失、不重复 ACK。 | @@ -4556,8 +4557,9 @@ npm run verify:phase8 | TC-SIGNATURE-RETIREMENT-019 | 检查两张热力图日期、行首和悬停信息 | 日期从左到右为`T-1`至`T-30`；行首不常驻企业和企业应用，悬停签名可看到企业、企业应用；通道维度仍能识别通道和运营商 | | TC-SIGNATURE-RETIREMENT-020 | 分别在企业、通道热力图搜索企业、企业应用、签名并清空 | 每张热力图只过滤自身真实维度并回到第一页；三类关键字均可命中，清空恢复，另一张热力图的关键字和页码不变 | | TC-SIGNATURE-RETIREMENT-021 | 悬停报备前、无快照、零量和非零量格子 | 报备前显示不适用；无快照说明当日无检测；真实快照明确显示提交条数、上游接受条数、发送成功条数和成功率，发送成功等于真实最终送达而非受理成功 | -| TC-SIGNATURE-RETIREMENT-022 | 所选日期构造有签名短信：运营商级已通过、历史通道级已通过、无通过任务、仅其他运营商通过、无签名 | 前两类不进入未报备模块；无通过任务和仅其他运营商通过按签名×实际应用计入；无签名记录不伪造成签名行；总条数与真实`SmsMessageRecord`一致 | +| TC-SIGNATURE-RETIREMENT-022 | 所选日期构造正文以规范`【签名】`开头的短信：当前企业应用签名库存在同名记录、签名库不存在、仅其他企业应用存在同名记录、正文没有规范开头签名 | 只有当前企业应用签名库不存在的规范正文签名进入未报备模块；已有同名系统签名不受通道或运营商报备状态影响，其他应用同名签名不能替代当前应用记录，无规范签名正文不伪造成签名行 | | TC-SIGNATURE-RETIREMENT-023 | 在未报备签名模块按企业、企业应用、签名搜索并翻页 | 后端搜索、总数、每页10行和分页结果一致；列表展示签名、企业、实际企业应用和未报备短信条数，修改主统计日期后按新的北京时间自然日重新查询 | +| TC-SIGNATURE-RETIREMENT-028 | 同一企业应用在所选北京时间自然日提交正文以`【湘银物业】`开头的短信，消息未关联`signatureId`且有效签名库无同名记录；另准备已有签名但缺少通道报备、其他应用同名签名、正文无规范开头签名三组对照数据 | 仅正文签名在当前企业应用签名库不存在的消息进入“未报备签名”，并按正文签名和实际企业应用聚合；已有系统签名但缺通道/运营商报备、其他应用的记录和无规范签名正文不得误判 | | TC-SIGNATURE-RETIREMENT-024 | 首次打开预警页面，随后选择历史日期区间 | 页签和区块标题均为“预警消息”；默认开始、结束均为今日且只返回今日消息，历史区间返回对应历史消息，每页10条并显示真实总数 | | TC-SIGNATURE-RETIREMENT-025 | 分别或组合选择企业、企业应用、签名关键字、通道及日期区间并翻页 | 后端同时应用全部条件，列表、总数和页码一致；条件变化查询后回到第1页，企业应用选项受企业筛选约束 | | TC-SIGNATURE-RETIREMENT-026 | 点击消息“抑制”，分别选择临时截止日期和永久抑制并填写原因 | 只出现平台自研弹窗；临时模式要求未来截止日期，永久模式不显示日期，两种模式原因必填，保存调用真实抑制接口且刷新当前筛选页 | @@ -4580,3 +4582,16 @@ npm run verify:phase8 | TC-CHANNEL-GROUP-FILTER-003 | 选择某通道 | 只展示`items.channelId`包含该通道的通道组，不展示仅运营商相同但未配置该通道的组；总数和分页与筛选结果一致 | | TC-CHANNEL-GROUP-FILTER-004 | 同时输入通道组名称并选择通道 | 按名称包含与成员通道两个条件取交集，条件变更后回到第一页 | | TC-CHANNEL-GROUP-FILTER-005 | 点击“重置” | 通道组名称和通道条件同时清空，恢复全部未删除通道组并回到第一页 | +# 下游投递后台重投任务专项用例（2026-08-12） + +| 编号 | 场景 | 预期 | +|---|---|---| +| TC-DOWNSTREAM-REQUEUE-TASK-001 | 当前筛选条件预检 | 后端按企业、应用、类型、状态、日期、关键词和 `snapshotAt` 返回真实命中、可重投、跳过及状态分布；分页不影响数量。 | +| TC-DOWNSTREAM-REQUEUE-TASK-002 | 创建任务 | 原因少于 5 字拒绝；仅物化 `pending/failed/unconfirmed/rejected`；`delivered/awaiting_ack` 不进入执行。 | +| TC-DOWNSTREAM-REQUEUE-TASK-003 | 快照边界 | 创建任务后新增或筛选条件外记录不进入任务。 | +| TC-DOWNSTREAM-REQUEUE-TASK-004 | 并发与幂等 | `taskId+deliveryId` 唯一；重复扫描、API 重启及并发任务不会重复调用 Gateway。 | +| TC-DOWNSTREAM-REQUEUE-TASK-005 | ACK 闭环 | Gateway 写出后项目进入等待 ACK；`Result=0` 成功，拒绝/超时计失败并按阈值自动暂停。 | +| TC-DOWNSTREAM-REQUEUE-TASK-006 | 状态变化跳过 | 执行前状态变化、已确认或被其他操作认领时不调用 Gateway，记录明确跳过原因。 | +| TC-DOWNSTREAM-REQUEUE-TASK-007 | 任务控制 | 待执行/执行中任务可暂停、继续、终止；终止不撤回已写出消息。 | +| TC-DOWNSTREAM-REQUEUE-TASK-008 | 审计 | 创建、暂停、继续、终止和自动暂停记录操作人、筛选快照、原因和结果。 | +| TC-DOWNSTREAM-PAGE-SIZE-001 | 分页数量 | 可选 10/25/50；切换回第一页，后端返回对应条数，总数和筛选条件保持一致。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index f405a21..47ba090 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3448,6 +3448,13 @@ git diff --check - 真实PostgreSQL聚合返回“完全未报备”6条、“仅移动已报备但提交电信”4条；浏览器按企业B搜索后只显示后者4条。企业热力图按企业B搜索只保留3个相关维度，通道热力图仍保留全部7个维度；签名悬停属性显示真实企业和应用，格子悬停属性显示五项明确口径，日期首列为08-09、末列为07-11，控制台error/warn为0。 - 清退专项7/7、API全量35个suite/446项通过，API与前端TypeScript、API正式构建、Vite 8.1.5生产构建通过（2538 modules，仅既有大chunk提示）。API全量用例本身12.573秒完成，但既有异步句柄使Jest不自行退出，本次使用`--forceExit`收尾并保留该提示；两次外层超时遗留的本轮Jest进程已按精确命令行确认后停止，未影响API、前端、PostgreSQL或Redis。 +## 2026-08-12 未报备签名判定修正（未提交、未发布） + +- 生产只读核查确认【彩生活物业】2026-08-12北京时间自然日有4个号码级`SmsMessageRecord`，4条均为2个计费分片；2026-08-11另有2个号码级消息。因此页面当日显示4条正确，用户所见6条为相邻两日累计，不修改签名质量统计代码，也不增加额外列表说明。 +- “未报备签名”按确认口径改为系统签名库缺失：从`signatureId IS NULL`消息正文开头提取规范`【签名】`，仅在同企业应用不存在未删除同名`SmsSignature`时计入；不再用通道、运营商或报备任务通过状态判定。结果仍按正文签名和实际企业应用聚合、搜索和分页。 +- 使用生产数据只读回放新聚合SQL，【湘银物业】在2026-08-12正确返回266条，企业为“王斯评与聆界中转企业”、应用为“王斯评平台To百信互动物业”；全过程未写生产数据库、未发送或重投短信。 +- 签名清退专项9/9、API与前端TypeScript、API正式构建、Vite 8.1.5生产构建及`git diff --check`通过；Vite仅保留既有约2.08MB单chunk提示。代码按要求未提交、未推送、未部署。 + ## 2026-08-10 预警消息检索分页、抑制弹窗与备注列宽（未提交、未发布） - “今日预警”已调整为“预警消息”，后端按预警日期、企业、企业应用、签名和通道执行真实PostgreSQL筛选及分页；页面默认选中北京时间今日，仅查询今日，支持历史日期区间并固定每页10条。本地回放最近5个检测日后，今日共11条：浏览器验收第1页10条、第2页1条；选择近7天并按“跨通道”签名查询返回15条、2页，可见`2026/8/9 08:00:00`历史消息及真实企业应用名称。 @@ -3500,3 +3507,15 @@ git diff --check - 新增统一`MoneyText`只读金额组件，运营端和客户端现有余额、授信、单价、消费、返还、充值、收入、成本、利润及短信计费等金额，小数点和小数部分使用统一次级文字色；输入框、CSV、复制文本和底层金额值不拆分、不改变。 - 本地正式`SignatureRetirementService`在真实PostgreSQL执行2026-08-12检测，生成11条alert、2条healthy、6条observing快照；执行前后站内消息均55条、Webhook投递均0条，证明检测阶段不外发。浏览器真实API验收热力图首列为08-11、显示30日合计且合计491/134/65/0按降序，6个观察期格子可见；企业与应用字重为400；金额小数色为`rgb(107, 114, 128)`；客户端登录Canvas为2560×1440并正常绘制，页面控制台error/warn为0。 - API全量35个suite/451项、签名清退与利润专项18/18项、前后端TypeScript、API正式构建、Vite 8.1.5生产构建、4份Gateway队列契约、Gateway `go test ./...`和`go vet ./...`通过；Vite仅保留既有约2.06MB单chunk提示，`git diff --check`仅有既有LF/CRLF提示。 +# 2026-08-12 下游投递后台重投任务与分页数量（已完成，待发布） + +- 已确认第一版设计：按当前真实筛选条件和创建时快照建立后台任务，只允许 `pending/failed/unconfirmed/rejected`，不批量重投 `delivered`，等待连接/ACK 与真正跳过严格分开。 +- 已增加任务/任务项真实 PostgreSQL 模型、预检、创建、分批原子认领、ACK 闭环、暂停/继续/终止、操作审计，以及运营端任务列表和详情；下游投递列表同步增加每页 `10/25/50` 条选择。 +- 本地PostgreSQL已真实应用`20260812153000_add_downstream_requeue_tasks`，当前共86条migration；Prisma validate、专项4项、API全量36个suite/458项、API/前端TypeScript、API/Vite生产构建、4份Gateway队列契约、R10结构契约和`git diff --check`通过。R10稳定门面方法数同步为104。 +- 全量Jest的458项断言均通过；仓库仍有既有异步句柄导致不自行退出，使用`--forceExit`取得退出码0。新增后台任务扫描器已在相关启动定时器用例中显式关闭，复跑时不再产生缺少测试Prisma delegate的循环错误日志。 + +# 2026-08-12 网关提交异常人工标记已处理（已完成，待发布） + +- “网关异常 / 提交异常”对`pending`记录增加“已处理”按钮和自研确认弹窗；确认后真实调用后端，将记录原子更新为`resolved`、写`resolvedAt`和`manually_resolved`，保留原始异常证据，不重新入队、不发送短信。 +- 后端记录操作人和`gateway.submit_dead_letter_resolved`审计日志；非`pending`状态拒绝并发标记，重复读取已处理记录保持幂等。 +- 功能纳入API全量36个suite/458项验证；前端TypeScript、API正式TypeScript构建、Vite生产构建和`git diff --check`通过，Vite仅有既有大chunk提示。 diff --git a/src/api/admin/operations.api.ts b/src/api/admin/operations.api.ts index e4326d1..a082d3e 100644 --- a/src/api/admin/operations.api.ts +++ b/src/api/admin/operations.api.ts @@ -1,5 +1,5 @@ import { request, requestBlob, requestForm, withQuery } from '../core/httpClient'; -import type { BatchRequeueResponse, BatchTaskMessagePage, DailyProfitReport, DailyQualityReport, DailyReconciliationReport, DashboardResponse, DownstreamDeliveryDashboard, DownstreamDeliveryRecord, DownstreamRecoveryStatusExportQuery, DownstreamRecoveryStatusResponse, GatewayDownstreamRecoveryStatus, GatewaySubmitException, GatewaySubmitExceptionResponse, OperationLogResponse, PagedResponse, PagedResult, PendingAuditCounts, ProfitReportSummary, ProtocolInteractionLogResponse, QualityReportSummary, ReceiptAnomalyResponse, ReconciliationReportSummary, SendQualityResponse, SignatureChannelQualityResponse, SmsBatchTask, SmsMessageRecord, SmsMessageSegmentAudit, SmsUplinkMessage, SystemLogExportResult } from '../types'; +import type { BatchRequeueResponse, BatchTaskMessagePage, DailyProfitReport, DailyQualityReport, DailyReconciliationReport, DashboardResponse, DownstreamDeliveryDashboard, DownstreamDeliveryRecord, DownstreamRecoveryStatusExportQuery, DownstreamRecoveryStatusResponse, DownstreamRequeueFilter, DownstreamRequeuePreview, DownstreamRequeueTask, GatewayDownstreamRecoveryStatus, GatewaySubmitException, GatewaySubmitExceptionResponse, OperationLogResponse, PagedResponse, PagedResult, PendingAuditCounts, ProfitReportSummary, ProtocolInteractionLogResponse, QualityReportSummary, ReceiptAnomalyResponse, ReconciliationReportSummary, SendQualityResponse, SignatureChannelQualityResponse, SmsBatchTask, SmsMessageRecord, SmsMessageSegmentAudit, SmsUplinkMessage, SystemLogExportResult } from '../types'; // Read-heavy operations endpoints are isolated from configuration mutations. export const adminOperationsApi = { @@ -53,6 +53,8 @@ export const adminOperationsApi = { request(withQuery('/admin/operations/gateway-submit-dead-letters', query)), requeueGatewaySubmitException: (id: string, body: { confirmedNotSubmitted: boolean; reason: string }) => request(`/admin/operations/gateway-submit-dead-letters/${id}/requeue`, { method: 'POST', body: JSON.stringify(body) }), + resolveGatewaySubmitException: (id: string) => + request(`/admin/operations/gateway-submit-dead-letters/${id}/resolve`, { method: 'POST', body: JSON.stringify({}) }), listReceiptAnomalies: (query: { tenantId?: string; applicationId?: string; channelId?: string; anomalyType?: string; status?: string; keyword?: string; page?: number; pageSize?: number } = {}) => request(withQuery('/admin/operations/receipt-anomalies', query)), listStatistics: (query: { tenantId?: string; groupBy?: string } = {}) => request>>(withQuery('/admin/operations/statistics', query)), @@ -70,4 +72,13 @@ export const adminOperationsApi = { request(`/admin/operations/downstream-deliveries/${id}/requeue`, { method: 'POST', body: JSON.stringify({}) }), batchRequeueDownstreamDeliveries: (ids: string[]) => request('/admin/operations/downstream-deliveries/requeue', { method: 'POST', body: JSON.stringify({ ids }) }), + previewDownstreamRequeueTask: (filter: DownstreamRequeueFilter) => + request('/admin/operations/downstream-requeue-tasks/preview', { method: 'POST', body: JSON.stringify({ filter }) }), + createDownstreamRequeueTask: (body: { filter: DownstreamRequeueFilter; snapshotAt: string; reason: string; ratePerSecond: number; consecutiveFailureLimit: number }) => + request('/admin/operations/downstream-requeue-tasks', { method: 'POST', body: JSON.stringify(body) }), + listDownstreamRequeueTasks: (query: { status?: string; page?: number; pageSize?: number } = {}) => + request>(withQuery('/admin/operations/downstream-requeue-tasks', query)), + getDownstreamRequeueTask: (id: string) => request(`/admin/operations/downstream-requeue-tasks/${id}`), + changeDownstreamRequeueTaskStatus: (id: string, action: 'pause' | 'resume' | 'terminate') => + request(`/admin/operations/downstream-requeue-tasks/${id}/${action}`, { method: 'POST', body: JSON.stringify({}) }), }; diff --git a/src/api/types/operations.ts b/src/api/types/operations.ts index 7fc1401..d313fa6 100644 --- a/src/api/types/operations.ts +++ b/src/api/types/operations.ts @@ -483,6 +483,51 @@ export type BatchRequeueResponse = { results: Array<{ id: string; status: 'success' | 'failed'; errorMessage?: string }>; }; +export type DownstreamRequeueFilter = { + tenantId?: string; + applicationId?: string; + deliveryType?: string; + status?: string; + keyword?: string; + createdAtFrom?: string; + createdAtTo?: string; +}; + +export type DownstreamRequeuePreview = { + snapshotAt: string; + matchedCount: number; + replayableCount: number; + skippedCount: number; + applicationCount: number; + oldestCreatedAt?: string | null; + statusCounts: Record; + filter: DownstreamRequeueFilter; +}; + +export type DownstreamRequeueTask = { + id: string; + taskNo: string; + status: string; + filterSnapshot: DownstreamRequeueFilter; + snapshotAt: string; + reason: string; + ratePerSecond: number; + totalCount: number; + successCount: number; + failedCount: number; + skippedCount: number; + waitingCount: number; + lastError?: string | null; + createdAt: string; + startedAt?: string | null; + finishedAt?: string | null; + tenant?: TenantOption | null; + application?: EnterpriseApplication | null; + createdBy?: { id: string; displayName: string; username: string } | null; + itemCounts?: Record; + recentItems?: Array<{ id: string; status: string; skipReason?: string | null; errorMessage?: string | null; delivery: { messageId?: string | null; deliveryType: string; status: string; lastError?: string | null } }>; +}; + export type DownstreamDeliveryDashboard = { summary: { total: number; diff --git a/src/apps/admin/AdminAnalyticsPage.tsx b/src/apps/admin/AdminAnalyticsPage.tsx index 4af1664..9dc28b6 100644 --- a/src/apps/admin/AdminAnalyticsPage.tsx +++ b/src/apps/admin/AdminAnalyticsPage.tsx @@ -407,7 +407,7 @@ function UnreportedSignaturesCard({

未报备签名

待处理
-

{data?.date ?? '所选日期'} 已进入平台、但短信运营商没有匹配报备成功事实的业务短信。

+

{data?.date ?? '所选日期'} 已进入平台、但系统签名库中没有对应记录的业务短信。

} onClick={onSearch} variant="secondary">查询
-
统计说明：每条业务短信只计一次；运营商级或仍有效的历史兼容报备已通过时不计入。
+
统计说明：从短信正文开头识别签名；当前企业应用存在同名签名记录时不计入。
(null); const requeueInFlightRef = useRef(false); + const [taskPreview, setTaskPreview] = useState(null); + const [taskPreviewBusy, setTaskPreviewBusy] = useState(false); + const [taskReason, setTaskReason] = useState(''); + const [taskRate, setTaskRate] = useState(10); + const [taskCreateBusy, setTaskCreateBusy] = useState(false); + const [requeueTasks, setRequeueTasks] = useState([]); + const [selectedTask, setSelectedTask] = useState(null); + + const currentTaskFilter = useCallback(() => ({ + keyword: keyword || undefined, + status, + deliveryType, + applicationId, + createdAtFrom: dateRange.start, + createdAtTo: dateRange.end, + }), [applicationId, dateRange.end, dateRange.start, deliveryType, keyword, status]); + + const loadRequeueTasks = useCallback(() => { + adminApi.listDownstreamRequeueTasks({ page: 1, pageSize: 10 }) + .then((response) => setRequeueTasks(response.items)) + .catch(() => undefined); + }, []); const loadData = useCallback(() => { setLoading(true); @@ -202,6 +224,39 @@ export function AdminDownstreamDeliveriesPage() { loadData(); }, [loadData]); + useEffect(() => { + loadRequeueTasks(); + const timer = window.setInterval(loadRequeueTasks, 3000); + return () => window.clearInterval(timer); + }, [loadRequeueTasks]); + + const openTaskPreview = async () => { + setTaskPreviewBusy(true); + setError(''); + try { + setTaskPreview(await adminApi.previewDownstreamRequeueTask(currentTaskFilter())); + setTaskReason(''); + } catch (failure) { + setError(failure instanceof Error ? failure.message : '后台重投范围预检失败'); + } finally { + setTaskPreviewBusy(false); + } + }; + + const createRequeueTask = async () => { + if (!taskPreview || taskReason.trim().length < 5) return; + setTaskCreateBusy(true); + try { + await adminApi.createDownstreamRequeueTask({ filter: taskPreview.filter, snapshotAt: taskPreview.snapshotAt, reason: taskReason.trim(), ratePerSecond: taskRate, consecutiveFailureLimit: 10 }); + setTaskPreview(null); + loadRequeueTasks(); + } catch (failure) { + setError(failure instanceof Error ? failure.message : '后台重投任务创建失败'); + } finally { + setTaskCreateBusy(false); + } + }; + const replayableStatuses = useMemo(() => new Set(['pending', 'delivered', 'failed', 'unconfirmed', 'rejected']), []); const selectableIds = useMemo( () => records.filter((item) => replayableStatuses.has(item.status)).map((item) => item.id), @@ -455,6 +510,17 @@ export function AdminDownstreamDeliveriesPage() { 已选择 {selectedIds.length} 条；客户端已确认的记录也允许人工重投，请注意重复处理风险。

+ +
+
+
+

后台重投任务

大批量事故恢复按筛选快照分批执行，不受投递记录分页影响。

+ +
+
+ {requeueTasks.map((task) => ( +
+
{task.taskNo}{formatDateTime(task.createdAt)}
+
{task.application?.name ?? '多个应用'}{task.reason}
+
{task.status}{task.successCount + task.failedCount + task.skippedCount}/{task.totalCount}
+
+ + {task.status === 'running' || task.status === 'queued' ? : null} + {task.status === 'paused' ? : null} + {['queued', 'running', 'paused'].includes(task.status) ? : null} +
+
+ ))} + {requeueTasks.length === 0 ?

暂无后台重投任务

: null} +
+
+ {detail ? setDetail(null)} /> : null} + {taskPreview ? ( + !taskCreateBusy && setTaskPreview(null)} title="创建下游后台重投任务" size="xl" footer={<>}> +
+
+
筛选命中{taskPreview.matchedCount}
+
可重投{taskPreview.replayableCount}
+
规则跳过{taskPreview.skippedCount}
+
涉及应用{taskPreview.applicationCount}
+
最早记录{taskPreview.oldestCreatedAt ? formatDateTime(taskPreview.oldestCreatedAt) : '-'}
+
快照时间{formatDateTime(taskPreview.snapshotAt)}
+
+

状态分布：{Object.entries(taskPreview.statusCounts).map(([key, value]) => `${statusLabel[key] ?? key} ${value}`).join('；') || '无'}

+