/**
 * apps/worker — maintenance queue processor (scheduled housekeeping).
 *
 * Repeatable jobs (registered in index.ts via BullMQ repeat):
 *   retention  — daily 04:00  (id: maintenance:retention) → per-org policies
 *   reconcile  — weekly 02:30 (id: maintenance:reconcile)  → search drift check
 *   dlq-report — hourly      (id: maintenance:webhook-dlq) → stale failed deliveries
 *   cleanup    — hourly      (id: maintenance:cleanup)     → expired sessions/tokens/invites/keys
 *   resync     — ad-hoc      ({type:"workspace-resync", workspaceId}) from permission events
 * Ad-hoc jobs:
 *   retention.run — {orgId, scope, mode, runId, actor} (the org-admin console
 *                   "Run now / Run dry" — the SAME core as the daily job)
 *   audit.export  — {orgId, exportId} (the >10k-row async CSV export)
 *
 * All jobs are idempotent (each op is a count/compare or a scoped delete)
 * and log their timing + counts.
 */
import { Worker, type Job } from "bullmq";
import { withJobOtel } from "../lib/job-otel.js";

import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient } from "@prisma/client";
import { QUEUES, enqueue } from "../queues.js";
import { getLogger } from "../logger.js";
import type { WorkerEnv } from "../config.js";
import { notifyWorkspaceAdmins } from "../lib/notify.js";
import { runDigestRun } from "./email-decision.processor.js";
import { S3Client } from "@factory/storage";
import { runOrphanSweep } from "./files/cleanup.js";
import { runAuditExport } from "./audit-export.js";
import { runRetentionScope, runRetentionDaily, pruneAllPageVersions, pruneWorkspacePageVersions } from "./retention.js";
import { runTaskDueSweep, runReviewDueSweep } from "./tasks.js";

const log = getLogger("maintenance");

const INVITE_EXPIRY_DAYS = 7;

export const maintenanceJobSchema = z.union([
  z.object({ type: z.literal("retention") }),
  // TASK-0133: the org-admin "Run now / Run dry" (one scope, the triggering
  // admin carried in the job — audited as the admin, not the worker).
  z.object({
    type: z.literal("retention.run"),
    orgId: z.string().min(1),
    scope: z.enum(["trash", "comments", "audit_logs", "files"]),
    mode: z.enum(["dry", "execute"]),
    runId: z.string().min(1),
    actor: z.object({ type: z.enum(["USER", "SYSTEM"]), id: z.string().min(1) })
  }),
  // TASK-0133: the >10k-row async CSV export (enqueued by the API).
  z.object({
    type: z.literal("audit.export"),
    orgId: z.string().min(1),
    exportId: z.string().min(1)
  }),
  z.object({ type: z.literal("reconcile") }),
  // TASK-0096 §7: 15-minute contentHash reconcile sweep (enqueues the same
  // search-index job `reconcile` — one sweep implementation, single writer).
  z.object({ type: z.literal("search-reconcile") }),
  z.object({ type: z.literal("dlq-report") }),
  z.object({ type: z.literal("cleanup") }),
  z.object({ type: z.literal("workspace-resync"), workspaceId: z.string().min(1) }),
  // TASK-0094 R6: the hourly notification digest.
  z.object({ type: z.literal("notification-digest") }),
  // TASK-0104 §1: the daily S3 orphan-object sweep (grace-aware delete).
  z.object({ type: z.literal("orphans") }),
  // T-152: per-workspace page-version retention (ad-hoc, admin-triggered).
  z.object({
    type: z.literal("versions.prune"),
    workspaceId: z.string().min(1),
    actor: z.object({ type: z.enum(["USER", "SYSTEM"]), id: z.string().min(1) })
  }),
  // T-160: the hourly task due-reminder sweep (repeatable; ad-hoc for tests).
  z.object({ type: z.literal("task-due-sweep") }),
  // T-162/T-163: the hourly review-deadline reminder sweep.
  z.object({ type: z.literal("review-due-sweep") })
]);
export type MaintenanceJobData = z.infer<typeof maintenanceJobSchema>;

export const MAINTENANCE_REPEATABLES = [
  // TASK-0133: the spec's 04:00 UTC (was 03:00 — the per-org policy engine
  // replaces the old global sweep; the orphans sweep stays offset at 03:30).
  { id: "maintenance:retention", pattern: "0 4 * * *", data: { type: "retention" } },
  { id: "maintenance:reconcile", pattern: "30 2 * * 0", data: { type: "reconcile" } },
  // TASK-0096 §7: search drift repair every 15 minutes (the backstop for
  // missed/late events; idempotent contentHash sweep).
  { id: "maintenance:search-reconcile", pattern: "*/15 * * * *", data: { type: "search-reconcile" } },
  { id: "maintenance:webhook-dlq", pattern: "0 * * * *", data: { type: "dlq-report" } },
  { id: "maintenance:cleanup", pattern: "15 * * * *", data: { type: "cleanup" } },
  // The notification digest (TASK-0094 R6): hourly, offset from the other
  // hourly jobs to spread the load.
  { id: "maintenance:notification-digest", pattern: "30 * * * *", data: { type: "notification-digest" } },
  // TASK-0104 §1: the daily S3 orphan sweep (offset from the 03:00 retention
  // run so the two heavy maintenance jobs don't overlap).
  { id: "maintenance:orphans", pattern: "30 3 * * *", data: { type: "orphans" } },
  // T-160: task due reminders — hourly (offset from the :15 cleanup and the
  // :30 digest). The 1h look-back in the core covers the sweep timing.
  { id: "maintenance:task-due", pattern: "45 * * * *", data: { type: "task-due-sweep" } },
  { id: "maintenance:review-due", pattern: "46 * * * *", data: { type: "review-due-sweep" } }
] as const;

/**
 * Retention (TASK-0133) — the DAILY per-org policy engine.
 *
 * REPLACES the old global sweep (recorded conflict: it purged a global 30 d
 * trash + 365 d audit with NO org isolation; the spec's per-org policies
 * supersede it). Delegates to `runRetentionDaily` (retention.ts) — one
 * implementation for the cron AND the admin "Run now / Run dry" trigger.
 * Dry-run by default; WORKER_RETENTION_EXECUTE=true → execute.
 *
 * The legacy single-key report (`worker:retention:report`) is still written
 * for backward compatibility (per-run reports live at
 * `worker:retention:report:{runId}` — the org console polls those).
 */
export async function runRetention(
  prisma: PrismaClient,
  redis: Redis,
  env: WorkerEnv
): Promise<Record<string, unknown>> {
  const report = await runRetentionDaily(prisma, redis, env);
  // T-152: per-workspace page-version retention rides the same 04:00 run.
  report.versions = await pruneAllPageVersions(prisma, { type: "SYSTEM", id: "worker:maintenance" });
  await redis
    .set("worker:retention:report", JSON.stringify({ ...report, at: new Date().toISOString() }), "EX", 7 * 86_400)
    .catch(() => undefined);
  return report;
}

/**
 * Search reconciliation (TASK-0096): enqueue a contentHash sweep on the
 * `search-index` queue — the single search writer owns the sweep
 * (runReconcile in search-index/processor.ts). The weekly `reconcile` job and
 * the 15-min `search-reconcile` repeatable both funnel through this.
 */
export async function runReconcile(connection: Redis): Promise<{ enqueued: "search-index:reconcile" }> {
  await enqueue(connection, QUEUES.searchIndex, { type: "reconcile" });
  log.info({}, "search reconcile enqueued (search-index queue)");
  return { enqueued: "search-index:reconcile" };
}

/**
 * Hourly webhook DLQ report: deliveries that are FAILED and older than 24h
 * and not yet reported → admin notification (deduped by a redis flag per
 * delivery so each is reported once).
 */
export async function runWebhookDlqReport(
  prisma: PrismaClient,
  connection: Redis,
  env: WorkerEnv
): Promise<{ stale: number; notified: number }> {
  const t0 = Date.now();
  const cutoff = new Date(Date.now() - 24 * 3_600_000);
  const stale = await prisma.webhookDelivery.findMany({
    where: { status: "FAILED", createdAt: { lt: cutoff } },
    select: { id: true, webhookId: true, event: true, error: true },
    take: 100
  });
  let notified = 0;
  const byWorkspace = new Map<string, typeof stale>();
  for (const d of stale) {
    const webhook = await prisma.webhook.findUnique({
      where: { id: d.webhookId },
      select: { workspaceId: true }
    });
    if (!webhook) continue;
    const flag = `worker:dlq-reported:${d.id}`;
    const already = await connection.exists(flag).catch(() => 0);
    if (already) continue;
    await connection.set(flag, "1", "EX", 7 * 86_400).catch(() => undefined);
    const list = byWorkspace.get(webhook.workspaceId) ?? [];
    list.push(d);
    byWorkspace.set(webhook.workspaceId, list);
    notified++;
  }
  for (const [workspaceId, items] of byWorkspace) {
    await notifyWorkspaceAdmins(
      prisma,
      connection,
      env.SMTP_HOST,
      workspaceId,
      `${items.length} webhook delivery(s) dead-lettered`,
      items
        .slice(0, 5)
        .map((d) => `• ${d.event}: ${d.error ?? "unknown error"}`)
        .join("\n"),
      undefined,
      "INTEGRATION"
    );
  }
  log.info({ stale: stale.length, notified, durationMs: Date.now() - t0 }, "webhook DLQ report done");
  return { stale: stale.length, notified };
}

/**
 * Hourly cleanup: expired refresh tokens, expired invitations (7d), expired
 * API keys (mark revoked). Sessions with expired tokens are revoked too.
 */
export async function runCleanup(prisma: PrismaClient): Promise<Record<string, number>> {
  const t0 = Date.now();
  const now = new Date();
  const inviteCutoff = new Date(Date.now() - INVITE_EXPIRY_DAYS * 86_400_000);

  const refreshTokenRes = await prisma.refreshToken.deleteMany({
    where: { expiresAt: { lt: now }, revokedAt: null }
  });
  const sessionRes = await prisma.session.updateMany({
    where: { expiresAt: { lt: now }, revokedAt: null },
    data: { revokedAt: now }
  });
  const invitationRes = await prisma.invitation.updateMany({
    where: {
      status: "PENDING",
      expiresAt: { lt: now },
      revokedAt: null,
      createdAt: { lt: inviteCutoff }
    },
    data: { status: "EXPIRED" }
  });
  const apiKeyRes = await prisma.apiKey.updateMany({
    where: { expiresAt: { lt: now }, revokedAt: null },
    data: { revokedAt: now }
  });

  const counts = {
    expiredRefreshTokens: refreshTokenRes.count,
    revokedSessions: sessionRes.count,
    expiredInvitations: invitationRes.count,
    revokedApiKeys: apiKeyRes.count
  };
  log.info({ ...counts, durationMs: Date.now() - t0 }, "cleanup done");
  return counts;
}

/**
 * TASK-0104 §1: the daily S3 orphan sweep. Runs the grace-aware sweep over the
 * `files/` prefix (default 7-day grace — a crashed finalize is given a week
 * to land its FileVersion row before its object is reclaimed). Re-entrant +
 * bounded (10k objects/run) so a huge tenant can't starve the cron.
 *
 * Builds the canonical `@factory/storage` S3Client (it has `listObjects`; the
 * worker's `lib/s3.ts` client — the pipeline's — does not, and I won't add a
 * third listing path to it) from the env, scoped to the files bucket.
 */
export async function runOrphans(
  prisma: PrismaClient,
  env: WorkerEnv
): Promise<Record<string, unknown>> {
  const s3 = new S3Client({
    config: {
      // Mirrors apps/api/src/modules/files/s3.ts makeFilesClient (the
      // canonical files-bucket client) — the sweep must address the SAME
      // bucket the pipeline writes to.
      endpoint: env.S3_ENDPOINT.replace(/\/+$/, ""),
      region: env.S3_REGION,
      bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
      accessKeyId: env.S3_ACCESS_KEY,
      secretAccessKey: env.S3_SECRET_KEY,
      forcePathStyle: true
    }
  });
  const report = await runOrphanSweep({ s3, prisma, log });
  return { ...report };
}

// ---------------------------------------------------------------------------
// Worker
// ---------------------------------------------------------------------------

export function createMaintenanceWorker(opts: {
  connection: Redis;
  prisma: PrismaClient;
  env: WorkerEnv;
  concurrency?: number;
}): Worker {
  const { connection, prisma, env } = opts;
  return new Worker(
    QUEUES.maintenance,
    withJobOtel(QUEUES.maintenance, async (job: Job) => {
      const started = Date.now();
      const data = maintenanceJobSchema.parse(job.data);
      log.info({ jobId: job.id, type: data.type }, "maintenance job");
      switch (data.type) {
        case "retention":
          return runRetention(prisma, connection, env);
        case "retention.run": {
          // TASK-0133: the org-admin "Run now / Run dry" (ONE scope). The
          // report lands at `worker:retention:report:{runId}` (the console
          // polls it); the purge is audited as the triggering admin.
          const report = await runRetentionScope({
            prisma,
            redis: connection,
            orgId: data.orgId,
            scope: data.scope,
            mode: data.mode,
            actor: data.actor,
            runId: data.runId
          });
          log.info({ jobId: job.id, orgId: data.orgId, scope: data.scope, ...report }, "retention.run done");
          return report;
        }
        case "audit.export": {
          // TASK-0133: the >10k-row async CSV export (cursor → CSV → S3).
          const res = await runAuditExport(prisma, connection, data.orgId, data.exportId);
          log.info({ jobId: job.id, ...res }, "audit export done");
          return res;
        }
        case "reconcile":
        case "search-reconcile":
          return runReconcile(connection);
        case "dlq-report":
          return runWebhookDlqReport(prisma, connection, env);
        case "cleanup":
          return runCleanup(prisma);
        case "notification-digest": {
          // TASK-0094 R6: the hourly digest (one email per (user, workspace)).
          const res = await runDigestRun(prisma, connection, env);
          log.info({ jobId: job.id, enqueued: res.enqueued }, "notification digest done");
          return res;
        }
        case "orphans": {
          // TASK-0104 §1: the daily S3 orphan sweep (builds its own client).
          const res = await runOrphans(prisma, env);
          log.info({ jobId: job.id, ...res }, "orphan sweep done");
          return res;
        }
        case "versions.prune": {
          const res = await pruneWorkspacePageVersions(prisma, data.workspaceId, data.actor);
          log.info({ jobId: job.id, ...res }, "versions.prune done");
          return res;
        }
        case "task-due-sweep": {
          // T-160: notify assignees of tasks due within the 24h window.
          const res = await runTaskDueSweep(prisma, connection);
          log.info({ jobId: job.id, ...res }, "task-due-sweep done");
          return res;
        }
        case "review-due-sweep": {
          // T-162/T-163: remind reviewers of record of review deadlines.
          const res = await runReviewDueSweep(prisma, connection);
          log.info({ jobId: job.id, ...res }, "review-due-sweep done");
          return res;
        }
        case "workspace-resync": {
          // Enqueued by the search-index consumer on workspace-scoped
          // permission changes.
          await enqueue(connection, QUEUES.searchIndex, {
            type: "reindex",
            workspaceId: data.workspaceId
          });
          log.info({ jobId: job.id, workspaceId: data.workspaceId }, "workspace resync enqueued");
          return { resynced: data.workspaceId };
        }
      }
      log.info({ jobId: job.id, ms: Date.now() - started }, "done");
    }),
    {
      connection,
      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
    }
  );
}