feat: 增强下游重投与签名质量检测
This commit is contained in:
hectorzhao
@@ -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;
|
||||
+84
-22
@@ -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
|
||||
|
||||
@@ -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<string, string | undefined> }) {
|
||||
return this.sendChain.previewDownstreamRequeueTask(body.filter ?? {});
|
||||
}
|
||||
|
||||
@Post('downstream-requeue-tasks')
|
||||
createDownstreamRequeueTask(
|
||||
@Body() body: { filter?: Record<string, string | undefined>; 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')
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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<typeof setInterval>;
|
||||
private upstreamReceiptInboxInitialTimer?: ReturnType<typeof setTimeout>;
|
||||
private upstreamReceiptInboxIntervalTimer?: ReturnType<typeof setInterval>;
|
||||
private downstreamRequeueTaskIntervalTimer?: ReturnType<typeof setInterval>;
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
import { BadRequestException } from '@nestjs/common';
|
||||
import { SendDownstreamRequeueTaskService } from './send-downstream-requeue-task.service';
|
||||
|
||||
function prismaMock(): Record<string, any> {
|
||||
const result: Record<string, any> = {
|
||||
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<string, any>;
|
||||
|
||||
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: '已被客户确认' }) }));
|
||||
});
|
||||
});
|
||||
@@ -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<unknown> };
|
||||
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() } : {}) } });
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user