/**
 * apps/worker — S3 orphan-object sweeper (TASK-0104 §1/§3, spec file
 * `apps/worker/src/processors/files/cleanup.ts`).
 *
 * A file object can outlive its database rows without a purge (a crashed
 * finalize, a partial transaction, an externally-deleted row). The sweeper
 * reconciles the `files/` prefix of the FILES bucket against the
 * `File` + `FileVersion` ledger:
 *
 *   - every object key under `files/{ws}/{fileId}/{v}/{name}` must be a
 *     live object key: a non-deleted File row's current `File.key`, OR a
 *     FileVersion key whose File is non-deleted (retained old versions —
 *     reclaimed by "delete old versions"/purge);
 *   - any other `files/` object is an ORPHAN;
 *   - an orphan is deleted only after a GRACE PERIOD (default 7 days — the
 *     spec's "7-day grace" — the window for a crashed pipeline to finish and
 *     register its FileVersion row);
 *   - every deleted object is audited `file.orphan_purged {key, reason}`
 *     (actor SYSTEM/worker).
 *
 * Two entry points share this core:
 *   - the root-level `scripts/storage-orphans.ts` (`pnpm storage:orphans`)
 *     — DRY-RUN by default (reports orphans without deleting; `--purge`
 *     makes it act);
 *   - the daily `maintenance:orphans` repeatable (grace-aware, deletes).
 *
 * Dependency note: this module imports ONLY `@prisma/client` (types) +
 * `@factory/storage` (S3Client) so the ROOT `scripts/storage-orphans.ts`
 * can import it (the root has no `pino` hoist — logging is injected).
 *
 * Scope note: only the `files/` prefix is swept. Thumbs are derived (a
 * thumb without a version row is re-generated on next view; a thumb for a
 * purged file is deleted by the purge itself), so sweeping them would only
 * add noise.
 */
import type { PrismaClient } from "@prisma/client";
import { S3Client } from "@factory/storage";

/** Minimal structural logger (pino's Logger satisfies it; the root CLI
 *  passes a console-backed one). */
export interface OrphanLogger {
  info(obj: Record<string, unknown>, msg?: string): void;
  warn(obj: Record<string, unknown>, msg?: string): void;
}

const consoleLogger: OrphanLogger = {
  info: (obj, msg) => console.log(msg ? `${msg} ${JSON.stringify(obj)}` : JSON.stringify(obj)),
  warn: (obj, msg) => console.warn(msg ? `${msg} ${JSON.stringify(obj)}` : JSON.stringify(obj)),
};

/** Spec's "7-day grace" (docs/tasks/phase-09-files.md TASK-0104 §1). */
export const ORPHAN_GRACE_DAYS_DEFAULT = 7;

export interface OrphanSweeperOptions {
  s3: S3Client;
  prisma: PrismaClient;
  /** Grace before an orphan object may be deleted. 0 = delete immediately
   *  (used by tests). Default 7 days. */
  graceDays?: number;
  /** Report only (never delete). Default false (the maintenance run);
   *  true for the `storage:orphans` CLI. */
  dryRun?: boolean;
  /** Bound the object count per run (the run is re-entrant; a bounded batch
   *  keeps the daily cron short). Default 10_000. */
  maxObjects?: number;
  /** Restrict the sweep to one file's objects (key path segment 3 = this
   *  fileId). Used by the `storage:orphans --file <id>` acceptance check so
   *  a purge can be verified in isolation from unrelated pre-existing
   *  detritus in the bucket. */
  fileId?: string;
  /** Logger. Defaults to a console logger (the worker passes pino). */
  log?: OrphanLogger;
}

export interface OrphanReport {
  scannedKeys: number;
  /** Set when the sweep was scoped to one file (`fileId`). */
  fileId?: string;
  orphansFound: number;
  deleted: number;
  /** Orphans kept because they are inside the grace period (or because the
   *  run was a dry-run / the object vanished in the meantime). */
  retained: number;
  dryRun: boolean;
  keys: string[];
}

export interface ParsedOrphanKey {
  workspaceId: string;
  fileId: string;
  version: number | null;
  name: string;
}

/**
 * Parse a files-bucket object key into `{workspaceId, fileId, version, name}`.
 * The pipeline key is `files/{ws}/{fileId}/{version}/{name}` (buildFileKey),
 * so a key that does not match (including a missing version segment) is
 * malformed → orphan.
 */
export function parseOrphanKey(key: string): ParsedOrphanKey | null {
  const m = /^files\/([^/]+)\/([^/]+)\/(\d+)\/(.+)$/.exec(key);
  if (!m) return null;
  const version = Number(m[3]);
  return {
    workspaceId: m[1]!,
    fileId: m[2]!,
    version: Number.isSafeInteger(version) && version >= 1 ? version : null,
    name: m[4]!,
  };
}

/**
 * Live ledger: the exact set of object keys that are NOT orphans —
 *   - each non-deleted File row's CURRENT key (`File.key`), plus
 *   - each FileVersion key whose File is non-deleted (the retained old
 *     versions that a replace created and that "delete old versions"/purge
 *     will eventually reclaim).
 * (A FileVersion's key always names the same (fileId, version) as the key
 * parse yields, so exact-key membership is the precise check.)
 */
async function loadLiveLedger(prisma: PrismaClient): Promise<Set<string>> {
  const [files, versions] = await Promise.all([
    prisma.file.findMany({
      where: { deletedAt: null },
      select: { key: true },
    }),
    prisma.fileVersion.findMany({
      where: { file: { deletedAt: null } },
      select: { key: true },
    }),
  ]);
  const live = new Set<string>();
  for (const f of files) live.add(f.key);
  for (const v of versions) live.add(v.key);
  return live;
}

export async function runOrphanSweep(opts: OrphanSweeperOptions): Promise<OrphanReport> {
  const { s3, prisma } = opts;
  const log = opts.log ?? consoleLogger;
  const graceDays = opts.graceDays ?? ORPHAN_GRACE_DAYS_DEFAULT;
  const dryRun = opts.dryRun ?? false;
  const maxObjects = opts.maxObjects ?? 10_000;
  const t0 = Date.now();

  // 1. List ALL objects under `files/` (paginated past the 1000-key cap).
  const keys: string[] = [];
  let token: string | undefined;
  do {
    const page = await s3.listObjects("files/", 1000, token);
    keys.push(...page.keys);
    token = page.truncated ? page.nextToken : undefined;
    if (keys.length >= maxObjects) break;
  } while (token);

  // 2. Load the live ledger.
  const live = await loadLiveLedger(prisma);

  // 3. Classify each key; delete orphans past grace.
  const now = Date.now();
  const graceMs = graceDays * 24 * 3600 * 1000;
  const report: OrphanReport = {
    scannedKeys: keys.length,
    orphansFound: 0,
    deleted: 0,
    retained: 0,
    dryRun,
    keys: [],
  };
  if (opts.fileId !== undefined) report.fileId = opts.fileId;

  for (const key of keys.slice(0, maxObjects)) {
    const parsed = parseOrphanKey(key);
    // Scoped sweep (--file): ignore every object that is not this file's.
    if (opts.fileId && parsed?.fileId !== opts.fileId) continue;
    let reason: string;
    if (!parsed) {
      reason = "malformed_key";
    } else if (live.has(key)) {
      continue; // the key maps to a live file/version → NOT an orphan
    } else {
      reason = "no_ledger_row";
    }

    report.orphansFound += 1;
    report.keys.push(key);

    if (dryRun) {
      report.retained += 1;
      continue;
    }

    // Grace anchor = the object's Last-Modified (a fresh orphan — a crashed
    // finalize about to land its FileVersion row — is NOT deleted).
    let lastModified = 0;
    try {
      const head = await s3.headObject(key);
      lastModified = head?.lastModified ? new Date(head.lastModified).getTime() : 0;
    } catch {
      // Vanished in the meantime — nothing to do.
      report.retained += 1;
      continue;
    }
    if (graceMs > 0 && now - lastModified < graceMs) {
      report.retained += 1;
      continue;
    }

    await s3.deleteObject(key);
    report.deleted += 1;
    const ws = parsed?.workspaceId;
    const fileId = parsed?.fileId;
    log.warn(
      { key, reason, ws, fileId, ageDays: lastModified ? Math.floor((now - lastModified) / 86_400_000) : 0 },
      "orphan object purged"
    );
    // Audit (best-effort — the object delete is already done; a failed
    // audit must not re-raise and lose the run's other deletes).
    try {
      await prisma.auditLog.create({
        data: {
          actorType: "SYSTEM",
          actorId: "worker",
          workspaceId: ws ?? null,
          resourceType: "file",
          resourceId: fileId ?? "unknown",
          action: "file.orphan_purged",
          metadata: { key, reason },
        },
      });
    } catch (err) {
      log.warn({ err: (err as Error).message, key }, "orphan purge audit failed (non-fatal)");
    }
  }

  log.info({ ...report, ms: Date.now() - t0 }, "orphan sweep done");
  return report;
}