354 lines
26 KiB
TypeScript
354 lines
26 KiB
TypeScript
import { BadRequestException, NotFoundException } from '@nestjs/common';
|
|
import { Prisma } from '@prisma/client';
|
|
import { createHash, createHmac, randomUUID, timingSafeEqual } from 'node:crypto';
|
|
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'];
|
|
const ACTIVE_TASK_STATUSES = ['queued', 'running', 'paused'];
|
|
const PROCESSING_LEASE_MS = 2 * 60_000;
|
|
const SCAN_LEASE_MS = 15_000;
|
|
const PREVIEW_TOKEN_TTL_MS = 15 * 60_000;
|
|
|
|
function normalizedFilter(filter: DownstreamRequeueFilter): DownstreamRequeueFilter {
|
|
return {
|
|
tenantId: filter.tenantId || 'all',
|
|
applicationId: filter.applicationId || 'all',
|
|
deliveryType: filter.deliveryType || 'all',
|
|
status: filter.status || 'all',
|
|
keyword: filter.keyword?.trim() || undefined,
|
|
createdAtFrom: filter.createdAtFrom || undefined,
|
|
createdAtTo: filter.createdAtTo || undefined,
|
|
};
|
|
}
|
|
|
|
function taskWhere(filter: DownstreamRequeueFilter, snapshotAt: Date, replayableByDefault = true): Prisma.CmppDownstreamDeliveryWhereInput {
|
|
const normalized = normalizedFilter(filter);
|
|
const from = parseDateBoundary(normalized.createdAtFrom, false);
|
|
const to = parseDateBoundary(normalized.createdAtTo, true);
|
|
return {
|
|
tenantId: normalized.tenantId !== 'all' ? normalized.tenantId : undefined,
|
|
applicationId: normalized.applicationId !== 'all' ? normalized.applicationId : undefined,
|
|
deliveryType: normalized.deliveryType !== 'all' ? normalized.deliveryType : undefined,
|
|
status: normalized.status !== 'all' ? normalized.status : replayableByDefault ? { in: REPLAYABLE_STATUSES } : undefined,
|
|
createdAt: { ...(from ? { gte: from } : {}), lte: to && to < snapshotAt ? to : snapshotAt },
|
|
OR: normalized.keyword ? [
|
|
{ messageId: { contains: normalized.keyword } },
|
|
{ payload: { path: ['account'], string_contains: normalized.keyword } },
|
|
{ payload: { path: ['phoneNumber'], string_contains: normalized.keyword } },
|
|
{ lastError: { contains: normalized.keyword } },
|
|
{ tenant: { name: { contains: normalized.keyword } } },
|
|
{ application: { name: { contains: normalized.keyword } } },
|
|
] : undefined,
|
|
};
|
|
}
|
|
|
|
function previewSecret() {
|
|
const source = process.env.DOWNSTREAM_REQUEUE_PREVIEW_SECRET || process.env.DATABASE_URL;
|
|
if (!source) throw new BadRequestException('后台重投预检签名密钥未配置');
|
|
return createHash('sha256').update(`cmpp-downstream-requeue-preview\0${source}`).digest();
|
|
}
|
|
|
|
function signPreview(payload: Record<string, unknown>) {
|
|
const encoded = Buffer.from(JSON.stringify(payload)).toString('base64url');
|
|
const signature = createHmac('sha256', previewSecret()).update(encoded).digest('base64url');
|
|
return `${encoded}.${signature}`;
|
|
}
|
|
|
|
function verifyPreview(token: string, operatorId?: string) {
|
|
const [encoded, supplied] = String(token || '').split('.');
|
|
if (!encoded || !supplied) throw new BadRequestException('预检凭证无效，请重新预检');
|
|
const expected = createHmac('sha256', previewSecret()).update(encoded).digest();
|
|
let actual: Buffer;
|
|
try { actual = Buffer.from(supplied, 'base64url'); } catch { throw new BadRequestException('预检凭证无效，请重新预检'); }
|
|
if (expected.length !== actual.length || !timingSafeEqual(expected, actual)) throw new BadRequestException('预检凭证无效，请重新预检');
|
|
const payload = JSON.parse(Buffer.from(encoded, 'base64url').toString('utf8')) as { filter: DownstreamRequeueFilter; snapshotAt: string; operatorId?: string; expiresAt: number };
|
|
if (payload.expiresAt < Date.now()) throw new BadRequestException('预检凭证已过期，请重新预检');
|
|
if ((payload.operatorId || '') !== (operatorId || '')) throw new BadRequestException('预检凭证与当前操作人不一致');
|
|
return payload;
|
|
}
|
|
|
|
function jsonFailures(value: unknown): Record<string, number> {
|
|
if (!value || typeof value !== 'object' || Array.isArray(value)) return {};
|
|
return Object.fromEntries(Object.entries(value).map(([key, count]) => [key, Math.max(0, Number(count) || 0)]));
|
|
}
|
|
|
|
export class SendDownstreamRequeueTaskService {
|
|
constructor(private readonly prisma: PrismaService, private readonly facade: RequeueFacade) {}
|
|
|
|
async preview(filter: DownstreamRequeueFilter, operatorId?: string) {
|
|
const snapshotAt = new Date();
|
|
const normalized = normalizedFilter(filter);
|
|
const base = taskWhere(normalized, snapshotAt, false);
|
|
const replayableWhere = { AND: [base, { status: { in: REPLAYABLE_STATUSES } }] } as Prisma.CmppDownstreamDeliveryWhereInput;
|
|
const [matchedCount, replayableCount, statusGroups, appGroups, oldest] = await Promise.all([
|
|
this.prisma.cmppDownstreamDelivery.count({ where: base }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: replayableWhere }),
|
|
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 } }),
|
|
]);
|
|
const tokenPayload = { filter: normalized, snapshotAt: snapshotAt.toISOString(), operatorId: operatorId || '', expiresAt: Date.now() + PREVIEW_TOKEN_TTL_MS };
|
|
return {
|
|
snapshotAt,
|
|
previewToken: signPreview(tokenPayload),
|
|
matchedCount,
|
|
replayableCount,
|
|
skippedCount: matchedCount - replayableCount,
|
|
applicationCount: appGroups.length,
|
|
oldestCreatedAt: oldest?.createdAt ?? null,
|
|
statusCounts: Object.fromEntries(statusGroups.map((item) => [item.status, item._count._all])),
|
|
filter: normalized,
|
|
};
|
|
}
|
|
|
|
async create(data: { previewToken: string; reason: string; ratePerSecond?: number; consecutiveFailureLimit?: number }, createdById?: string) {
|
|
const reason = data.reason?.trim();
|
|
if (!reason || reason.length < 5) throw new BadRequestException('任务原因至少填写5个字');
|
|
const preview = verifyPreview(data.previewToken, createdById);
|
|
const filter = normalizedFilter(preview.filter);
|
|
const snapshotAt = new Date(preview.snapshotAt);
|
|
if (filter.status === 'delivered' || filter.status === 'awaiting_ack') throw new BadRequestException('第一版后台任务不支持已确认或等待ACK记录');
|
|
const activeTask = await this.prisma.downstreamRequeueTask.findFirst({ where: {
|
|
status: { in: ACTIVE_TASK_STATUSES },
|
|
...(filter.applicationId !== 'all' ? { OR: [{ applicationId: filter.applicationId }, { applicationId: null }] } : {}),
|
|
}, select: { taskNo: true } });
|
|
if (activeTask) throw new BadRequestException(`当前应用范围已有未结束任务 ${activeTask.taskNo}`);
|
|
const where = { AND: [taskWhere(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, 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: filter.tenantId !== 'all' ? filter.tenantId : null,
|
|
applicationId: filter.applicationId !== 'all' ? filter.applicationId : null,
|
|
filterSnapshot: 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, ratePerSecond, consecutiveFailureLimit: failureLimit } } });
|
|
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 = await this.prisma.downstreamRequeueTaskItem.groupBy({ by: ['status'], where: { taskId: id }, _count: { _all: true } });
|
|
return { ...task, itemCounts: Object.fromEntries(itemGroups.map((item) => [item.status, item._count._all])) };
|
|
}
|
|
|
|
async listItems(id: string, query: { status?: string; keyword?: string; page?: number; pageSize?: number }) {
|
|
const task = await this.prisma.downstreamRequeueTask.findUnique({ where: { id }, select: { id: true } });
|
|
if (!task) throw new NotFoundException('后台重投任务不存在');
|
|
const page = Math.max(1, Number(query.page ?? 1));
|
|
const pageSize = Math.min(100, Math.max(1, Number(query.pageSize ?? 20)));
|
|
const keyword = query.keyword?.trim();
|
|
const where: Prisma.DownstreamRequeueTaskItemWhereInput = {
|
|
taskId: id,
|
|
status: query.status && query.status !== 'all' ? query.status : undefined,
|
|
OR: keyword ? [
|
|
{ delivery: { messageId: { contains: keyword } } },
|
|
{ skipReason: { contains: keyword } },
|
|
{ errorMessage: { contains: keyword } },
|
|
] : undefined,
|
|
};
|
|
const [items, total] = await Promise.all([
|
|
this.prisma.downstreamRequeueTaskItem.findMany({ where, include: { delivery: { select: { messageId: true, deliveryType: true, status: true, lastError: true } } }, orderBy: [{ updatedAt: 'desc' }, { id: 'desc' }], skip: (page - 1) * pageSize, take: pageSize }),
|
|
this.prisma.downstreamRequeueTaskItem.count({ where }),
|
|
]);
|
|
return { items, total, page, pageSize };
|
|
}
|
|
|
|
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, scanLeaseOwner: null, scanLeaseUntil: null } });
|
|
if (status === 'terminated') await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { taskId: id, status: { in: ['queued', 'waiting_connection'] } }, 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() {
|
|
await this.prisma.downstreamRequeueRateWindow.deleteMany({ where: { windowStartedAt: { lt: new Date(Date.now() - 5 * 60_000) } } });
|
|
const tasks = await this.prisma.downstreamRequeueTask.findMany({ where: { status: { in: ['queued', 'running'] } }, orderBy: { createdAt: 'asc' }, take: 3, select: { id: true } });
|
|
for (const task of tasks) await this.processTask(task.id);
|
|
}
|
|
|
|
private async processTask(taskId: string) {
|
|
const leaseOwner = randomUUID();
|
|
const now = new Date();
|
|
const lease = await this.prisma.downstreamRequeueTask.updateMany({
|
|
where: { id: taskId, status: { in: ['queued', 'running'] }, OR: [{ scanLeaseUntil: null }, { scanLeaseUntil: { lt: now } }] },
|
|
data: { scanLeaseOwner: leaseOwner, scanLeaseUntil: new Date(now.getTime() + SCAN_LEASE_MS) },
|
|
});
|
|
if (!lease.count) return;
|
|
try {
|
|
const task = await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId } });
|
|
if (!task || !['queued', 'running'].includes(task.status)) return;
|
|
// A process may die after the database claim but before the Gateway call. The lease makes that
|
|
// ambiguous window visible and recoverable; every recovered item is revalidated before replay.
|
|
await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { taskId, status: 'processing', claimedAt: { lt: new Date(Date.now() - PROCESSING_LEASE_MS) } }, data: { status: 'queued', claimedAt: null, errorMessage: '执行进程中断，已回收并等待重新复核' } });
|
|
let failures = await this.reconcileWaiting(taskId, jsonFailures(task.applicationFailures));
|
|
const existingFailureEntry = Object.entries(failures).find(([, count]) => count >= task.consecutiveFailureLimit);
|
|
if (existingFailureEntry) {
|
|
await this.autoPause(task, existingFailureEntry[0], existingFailureEntry[1]);
|
|
await this.refreshTask(taskId);
|
|
return;
|
|
}
|
|
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(200, task.ratePerSecond * 3), select: { id: true, deliveryId: true, applicationId: true } });
|
|
for (const item of items) {
|
|
const latestTask = await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId }, select: { status: true } });
|
|
if (!latestTask || !['queued', 'running'].includes(latestTask.status)) break;
|
|
if (!(await this.consumeRate(item.applicationId, task.ratePerSecond))) continue;
|
|
const claimed = await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { id: item.id, status: 'queued' }, data: { status: 'processing', claimedAt: new Date() } });
|
|
if (!claimed.count) continue;
|
|
const outcome = await this.processItem(item.id, item.deliveryId);
|
|
if (outcome === 'success') failures[item.applicationId] = 0;
|
|
if (outcome === 'failed') failures[item.applicationId] = (failures[item.applicationId] ?? 0) + 1;
|
|
const maxFailures = Math.max(0, ...Object.values(failures));
|
|
await this.prisma.downstreamRequeueTask.update({ where: { id: taskId }, data: { applicationFailures: failures as Prisma.InputJsonValue, consecutiveFailures: maxFailures } });
|
|
if ((failures[item.applicationId] ?? 0) >= task.consecutiveFailureLimit) {
|
|
await this.autoPause(task, item.applicationId, failures[item.applicationId]);
|
|
break;
|
|
}
|
|
}
|
|
failures = await this.reconcileWaiting(taskId, failures);
|
|
const maxFailures = Math.max(0, ...Object.values(failures));
|
|
await this.prisma.downstreamRequeueTask.updateMany({ where: { id: taskId, status: { in: ['queued', 'running'] } }, data: { applicationFailures: failures as Prisma.InputJsonValue, consecutiveFailures: maxFailures } });
|
|
const ackFailureEntry = Object.entries(failures).find(([, count]) => count >= task.consecutiveFailureLimit);
|
|
if (ackFailureEntry) await this.autoPause(task, ackFailureEntry[0], ackFailureEntry[1]);
|
|
await this.refreshTask(taskId);
|
|
} finally {
|
|
await this.prisma.downstreamRequeueTask.updateMany({ where: { id: taskId, scanLeaseOwner: leaseOwner }, data: { scanLeaseOwner: null, scanLeaseUntil: null } });
|
|
}
|
|
}
|
|
|
|
private async processItem(itemId: string, deliveryId: string): Promise<'success' | 'failed' | 'waiting' | 'skipped'> {
|
|
try {
|
|
const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({
|
|
where: { id: deliveryId },
|
|
include: { application: { select: { status: true, interfaceEnabled: true } } },
|
|
});
|
|
if (!delivery) return this.finishItem(itemId, 'skipped', '投递记录已不存在');
|
|
if (!REPLAYABLE_STATUSES.includes(delivery.status)) {
|
|
if (delivery.status === 'awaiting_ack') { await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status: 'waiting_external_ack', skipReason: null } }); return 'waiting'; }
|
|
return this.finishItem(itemId, 'skipped', delivery.status === 'delivered' ? '已被客户确认' : '执行前状态已变化');
|
|
}
|
|
if (delivery.application.status !== 'active' || !delivery.application.interfaceEnabled) return this.finishItem(itemId, 'skipped', '应用或投递能力已停用');
|
|
if (!delivery.payload || !['receipt', 'uplink'].includes(delivery.deliveryType)) return this.finishItem(itemId, 'skipped', '投递数据不完整');
|
|
const activeOther = await this.prisma.downstreamRequeueTaskItem.findFirst({ where: { deliveryId, id: { not: itemId }, status: { in: ['processing', 'waiting_ack', 'success'] } }, select: { id: true } });
|
|
if (activeOther) return this.finishItem(itemId, 'skipped', '已被其他任务处理');
|
|
const connected = await this.prisma.cmppDownstreamConnection.count({ where: { applicationId: delivery.applicationId, status: 'connected' } });
|
|
if (connected === 0) { await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status: 'waiting_connection', claimedAt: null, errorMessage: '客户当前离线，等待连接恢复' } }); return 'waiting'; }
|
|
const result = await this.facade.requeueDownstreamDelivery(deliveryId) as { status?: string; lastError?: string | null };
|
|
if (result?.status === 'delivered') { await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status: 'success', completedAt: new Date() } }); return 'success'; }
|
|
if (result?.status === 'awaiting_ack') { await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status: 'waiting_ack', completedAt: null } }); return 'waiting'; }
|
|
await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status: 'failed', errorMessage: result?.lastError ?? 'Gateway未进入等待ACK状态', completedAt: new Date() } });
|
|
return 'failed';
|
|
} 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: itemId }, data: { status: skipReason ? 'skipped' : 'failed', skipReason, errorMessage: skipReason ? null : message, completedAt: new Date() } });
|
|
return skipReason ? 'skipped' : 'failed';
|
|
}
|
|
}
|
|
|
|
private async finishItem(itemId: string, status: 'skipped', reason: string): Promise<'skipped'> {
|
|
await this.prisma.downstreamRequeueTaskItem.update({ where: { id: itemId }, data: { status, skipReason: reason, completedAt: new Date() } });
|
|
return 'skipped';
|
|
}
|
|
|
|
private async consumeRate(applicationId: string, limit: number) {
|
|
const windowStartedAt = new Date(Math.floor(Date.now() / 1000) * 1000);
|
|
const rows = await this.prisma.$queryRaw<Array<{ consumed: number }>>(Prisma.sql`
|
|
INSERT INTO "DownstreamRequeueRateWindow" ("id", "applicationId", "windowStartedAt", "consumed", "updatedAt")
|
|
VALUES (${randomUUID()}, ${applicationId}, ${windowStartedAt}, 1, NOW())
|
|
ON CONFLICT ("applicationId", "windowStartedAt") DO UPDATE
|
|
SET "consumed" = "DownstreamRequeueRateWindow"."consumed" + 1, "updatedAt" = NOW()
|
|
WHERE "DownstreamRequeueRateWindow"."consumed" < ${limit}
|
|
RETURNING "consumed"
|
|
`);
|
|
return rows.length === 1;
|
|
}
|
|
|
|
private async reconcileWaiting(taskId: string, currentFailures?: Record<string, number>) {
|
|
const task = currentFailures ? null : await this.prisma.downstreamRequeueTask.findUnique({ where: { id: taskId }, select: { applicationFailures: true } });
|
|
const failures = currentFailures ?? jsonFailures(task?.applicationFailures);
|
|
const connectionItems = await this.prisma.downstreamRequeueTaskItem.findMany({ where: { taskId, status: 'waiting_connection' }, select: { id: true, applicationId: true } });
|
|
for (const item of connectionItems) {
|
|
const connected = await this.prisma.cmppDownstreamConnection.count({ where: { applicationId: item.applicationId, status: 'connected' } });
|
|
if (connected > 0) await this.prisma.downstreamRequeueTaskItem.updateMany({ where: { id: item.id, status: 'waiting_connection' }, data: { status: 'queued', errorMessage: null, claimedAt: null } });
|
|
}
|
|
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: 500 });
|
|
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 } });
|
|
if (item.status === 'waiting_ack') failures[item.applicationId] = 0;
|
|
} else if (['failed', 'rejected', 'unconfirmed'].includes(item.delivery.status) || (item.delivery.ackDeadlineAt && item.delivery.ackDeadlineAt <= now)) {
|
|
if (item.status === 'waiting_external_ack') {
|
|
await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'queued', skipReason: null, errorMessage: null, claimedAt: null } });
|
|
} else {
|
|
await this.prisma.downstreamRequeueTaskItem.update({ where: { id: item.id }, data: { status: 'failed', errorMessage: item.delivery.lastError ?? '客户端ACK失败或超时', completedAt: now } });
|
|
failures[item.applicationId] = (failures[item.applicationId] ?? 0) + 1;
|
|
}
|
|
}
|
|
}
|
|
return failures;
|
|
}
|
|
|
|
private async autoPause(task: { id: string; taskNo: string; consecutiveFailureLimit: number }, applicationId: string, count: number) {
|
|
const pausedAt = new Date();
|
|
const message = `应用 ${applicationId} 连续失败达到安全阈值 ${task.consecutiveFailureLimit} 条，任务已自动暂停`;
|
|
const updated = await this.prisma.downstreamRequeueTask.updateMany({ where: { id: task.id, status: { in: ['queued', 'running'] } }, data: { status: 'paused', pausedAt, lastError: message } });
|
|
if (updated.count) await this.prisma.operationLog.create({ data: { action: 'gateway.downstream_requeue_task_auto_paused', resource: 'downstream_requeue_task', resourceId: task.id, detail: { taskNo: task.taskNo, applicationId, consecutiveFailures: count, failureLimit: task.consecutiveFailureLimit, pausedAt } } });
|
|
}
|
|
|
|
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_connection') ?? 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() } : {}) } });
|
|
}
|
|
}
|