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() } : {}) } });
  }
}