/**
 * apps/worker — the ORG-SCOPED RETENTION core (TASK-0133).
 *
 * Single implementation of the purge engine, shared by:
 *   - the DAILY job (`retention.run`, 04:00 UTC, all orgs, mode from the
 *     `WORKER_RETENTION_EXECUTE` gate — dry by default), and
 *   - the ADMIN "Run now / Run dry" trigger (the API enqueues the SAME job
 *     with `{orgId, scope, mode, runId, actor}`; the result lands in the
 *     Redis report `worker:retention:report:{runId}` the console polls).
 *
 * Semantics (spec `docs/tasks/phase-14-enterprise.md` TASK-0133):
 *   - one `RetentionPolicy` per (org, scope); defaults: trash PURGE 30 d,
 *     comments PURGE 90 d, audit_logs HOLD (never purge while HOLD), files
 *     PURGE 0 d (0 = no purge).
 *   - PURGE → batched transactions (100 ids/batch, each batch its own
 *     `retention.purged {scope, counts}` audit row).
 *   - HOLD → skipped + `retention.hold_skipped` audit (audit_logs HOLD is the
 *     "legal hold" requirement — audit rows are NEVER purged while it is on).
 *   - per-scope isolation: a throw in one scope does not stop the others
 *     (per-scope try/catch + `retention.failed` audit for the broken one).
 *   - org isolation: every query is scoped to the org's workspaces (+ the
 *     shared org-scope OR-branch for audit rows — `@factory/shared/events/
 *     audit-scope`, the SAME builder the API viewer uses).
 *
 * The old global `runRetention` sweep (maintenance.processor) is REPLACED
 * (recorded conflict: it had no org isolation — a global 30 d trash purge —
 * and a global 365 d audit purge; the spec's per-org policies supersede it.
 * The function remains exported for the deprecation window).
 *
 * Recorded decisions:
 *  - `Page.trashedAt` is the trash clock (the spec text says `Page.deletedAt`;
 *    the live trash path + existing sweep key off `trashedAt`).
 *  - `retention.purged` counts shape is the spec's `{pages, blocks, comments,
 *    files, bytesFreed}` — the audit_logs scope has no dedicated slot and
 *    reuses `comments` (the scope field disambiguates; documented).
 *  - files: `bytesFreed` sums the deleted rows' `File.size` (current version).
 */
import type { PrismaClient, Prisma } from "@prisma/client";
import type { Redis } from "ioredis";
import { asJson } from "../lib/prisma.js";
import { getLogger } from "../logger.js";
import {
  auditScopedWhere,
  type AuditFilters
} from "@factory/shared/events/audit-scope";
import { planPageVersionPrune, versionRetentionFromSettings } from "@factory/shared";

const log = getLogger("retention");

export type RetentionScope = "trash" | "comments" | "audit_logs" | "files";
export const RETENTION_SCOPES: RetentionScope[] = ["trash", "comments", "audit_logs", "files"];

/** The spec defaults (applied when the org has no policy row for a scope). */
export const RETENTION_DEFAULTS: Record<RetentionScope, { type: "PURGE" | "HOLD"; retainDays: number }> = {
  trash: { type: "PURGE", retainDays: 30 },
  comments: { type: "PURGE", retainDays: 90 },
  audit_logs: { type: "HOLD", retainDays: 365 },
  files: { type: "PURGE", retainDays: 0 }
};

export const RETENTION_BATCH_SIZE = 100;

export interface ScopeCounts {
  pages: number;
  blocks: number;
  comments: number;
  files: number;
  bytesFreed: number;
}

export interface RetentionActor {
  type: "USER" | "SYSTEM";
  id: string;
}

export interface RetentionRunOptions {
  prisma: PrismaClient;
  redis?: Redis;
  orgId: string;
  scope: RetentionScope;
  mode: "dry" | "execute";
  /** The triggering admin (admin runs) or the worker identity (cron). */
  actor: RetentionActor;
  /** When set, the run report is written here (the console polls it). */
  runId?: string;
}

interface ScopePolicy {
  type: "PURGE" | "HOLD";
  retainDays: number;
  config: Record<string, unknown>;
}

async function orgWorkspaceIds(prisma: PrismaClient, orgId: string): Promise<string[]> {
  const rows = await prisma.workspace.findMany({ where: { organizationId: orgId }, select: { id: true } });
  return rows.map((r) => r.id);
}

/** The effective policy for (org, scope) — the row, or the spec default. */
export async function policyFor(
  prisma: PrismaClient,
  orgId: string,
  scope: RetentionScope
): Promise<ScopePolicy> {
  const row = await prisma.retentionPolicy.findUnique({
    where: { organizationId_scope: { organizationId: orgId, scope } }
  });
  const def = RETENTION_DEFAULTS[scope];
  if (!row) return { ...def, config: {} };
  return {
    type: row.type,
    retainDays: row.retainDays ?? def.retainDays,
    config: (row.config as Record<string, unknown> | null) ?? {}
  };
}

/** Persist `config.lastRun`/`lastDryRun` on the policy row (best-effort). */
async function touchLastRun(
  prisma: PrismaClient,
  orgId: string,
  scope: RetentionScope,
  mode: "dry" | "execute",
  summary: Record<string, unknown>
): Promise<void> {
  const key = mode === "execute" ? "lastRun" : "lastDryRun";
  const value = { at: new Date().toISOString(), ...summary };
  const existing = await prisma.retentionPolicy.findUnique({
    where: { organizationId_scope: { organizationId: orgId, scope } }
  });
  try {
    if (existing) {
      const cfg = (existing.config as Record<string, unknown> | null) ?? {};
      await prisma.retentionPolicy.update({
        where: { id: existing.id },
        data: { config: asJson({ ...cfg, [key]: value }) }
      });
    } else {
      await prisma.retentionPolicy.create({
        data: {
          organizationId: orgId,
          scope,
          type: RETENTION_DEFAULTS[scope].type,
          retainDays: RETENTION_DEFAULTS[scope].retainDays,
          config: asJson({ [key]: value })
        }
      });
    }
  } catch (err) {
    log.warn({ err: (err as Error).message, orgId, scope }, "retention lastRun update failed");
  }
}

async function writeAudit(
  prisma: PrismaClient,
  actor: RetentionActor,
  orgId: string,
  action: string,
  metadata: Record<string, unknown>
): Promise<void> {
  await prisma.auditLog.create({
    data: {
      actorType: actor.type,
      actorId: actor.id,
      workspaceId: null,
      resourceType: "organization",
      resourceId: orgId,
      action,
      metadata: asJson({ orgId, ...metadata })
    }
  });
}

/** The admin-console report (the poll endpoint reads it; TTL 7 d). */
async function writeRunReport(
  redis: Redis | undefined,
  runId: string | undefined,
  orgId: string,
  report: Record<string, unknown>
): Promise<void> {
  if (!redis || !runId) return;
  await redis
    .set(`worker:retention:report:${runId}`, JSON.stringify({ orgId, ...report }), "EX", 7 * 86_400)
    .catch((err) => log.warn({ err: (err as Error).message, runId }, "retention report write failed"));
}

// ─────────────────────────────── purge cores ────────────────────────────────

/**
 * trash — pages with `trashedAt < cutoff` in the org's workspaces.
 * Cascade map (schema): Block/PageFavorite/PageVersion/Comment/Share CASCADE
 * on pageId; DatabaseRecord.pageId is SET-NULL (explicit deleteMany, mirroring
 * the page service's `deletePermanent`); File.pageId is a plain ref (the files
 * scope owns file rows). Blocks are deleted explicitly so the audit counts
 * them (the cascade would also work, but counting the cascade is not
 * deterministic).
 */
async function purgeTrash(
  prisma: PrismaClient,
  orgId: string,
  wsIds: string[],
  cutoff: Date,
  execute: boolean,
  actor: RetentionActor
): Promise<ScopeCounts> {
  const where: Prisma.PageWhereInput = {
    workspaceId: { in: wsIds },
    trashedAt: { lt: cutoff }
  };
  const ids = await prisma.page.findMany({ where, select: { id: true } });
  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
  if (execute) {
    for (let i = 0; i < ids.length; i += RETENTION_BATCH_SIZE) {
      const batch = ids.slice(i, i + RETENTION_BATCH_SIZE).map((p) => p.id);
      const blockCount = await prisma.$transaction(async (tx) => {
        // Explicit deletes (children → parent). BlockVersion CASCADEs on
        // blockId; the rest CASCADE on pageId (Block/Comment/Share/Favorite/
        // PageVersion). DatabaseRecord.pageId is SET-NULL → explicit delete.
        await tx.databaseRecord.deleteMany({ where: { pageId: { in: batch } } });
        const blocks = await tx.block.deleteMany({ where: { pageId: { in: batch } } });
        await tx.page.deleteMany({ where: { id: { in: batch } } });
        return blocks.count;
      });
      counts.pages += batch.length;
      counts.blocks += blockCount;
      await writeAudit(prisma, actor, orgId, "retention.purged", {
        scope: "trash",
        counts: { pages: batch.length, blocks: blockCount }
      });
    }
  }
  // Dry-run: report the would-be page count (blocks are not enumerable
  // without deleting — the UI shows the page count).
  return { ...counts, pages: execute ? counts.pages : ids.length };
}

/**
 * comments — root comments older than the cutoff with NO reply newer than
 * the cutoff (a recent reply keeps the whole thread). Replies + reactions of
 * purged threads go with them.
 */
async function purgeComments(
  prisma: PrismaClient,
  orgId: string,
  wsIds: string[],
  cutoff: Date,
  execute: boolean,
  actor: RetentionActor
): Promise<ScopeCounts> {
  const candidates = await prisma.comment.findMany({
    where: { workspaceId: { in: wsIds }, createdAt: { lt: cutoff } },
    select: { id: true }
  });
  const ids = candidates.map((c) => c.id);
  // A thread survives when ANY reply is newer than the cutoff.
  let survivors: string[] = [];
  if (ids.length) {
    const recent = await prisma.commentReply.findMany({
      where: { commentId: { in: ids }, createdAt: { gt: cutoff } },
      select: { commentId: true },
      distinct: ["commentId"]
    });
    survivors = recent.map((r) => r.commentId);
  }
  const toPurge = ids.filter((id) => !survivors.includes(id));
  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
  if (execute) {
    for (let i = 0; i < toPurge.length; i += RETENTION_BATCH_SIZE) {
      const batch = toPurge.slice(i, i + RETENTION_BATCH_SIZE);
      await prisma.$transaction(async (tx) => {
        // Reactions cascade on reply; delete reactions (comment + replies) →
        // replies → comments explicitly (countable, deterministic).
        await tx.commentReaction.deleteMany({ where: { commentId: { in: batch } } });
        const replyRes = await tx.commentReply.deleteMany({ where: { commentId: { in: batch } } });
        await tx.comment.deleteMany({ where: { id: { in: batch } } });
        void replyRes;
      });
      counts.comments += batch.length;
      await writeAudit(prisma, actor, orgId, "retention.purged", {
        scope: "comments",
        counts: { comments: batch.length }
      });
    }
  }
  // Dry-run: report the would-be count (root comments minus kept threads).
  return { ...counts, comments: execute ? counts.comments : toPurge.length };
}

/**
 * audit_logs — the shared org-scope where (the SAME builder as the viewer)
 * ∧ `createdAt < cutoff`. HOLD → the caller skips (hold_skipped). The counts
 * slot reuses `comments` (the spec shape has no audit slot — recorded).
 */
async function purgeAuditLogs(
  prisma: PrismaClient,
  orgId: string,
  wsIds: string[],
  cutoff: Date,
  execute: boolean,
  actor: RetentionActor
): Promise<ScopeCounts> {
  const filters: AuditFilters = { to: cutoff };
  const where = auditScopedWhere(orgId, wsIds, filters) as Prisma.AuditLogWhereInput;
  const total = await prisma.auditLog.count({ where });
  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
  if (execute) {
    let remaining = total;
    while (remaining > 0) {
      const ids = await prisma.auditLog.findMany({ where, select: { id: true }, take: RETENTION_BATCH_SIZE });
      if (!ids.length) break;
      const res = await prisma.auditLog.deleteMany({ where: { id: { in: ids.map((r) => r.id) } } });
      counts.comments += res.count; // recorded slot reuse
      remaining -= res.count;
      await writeAudit(prisma, actor, orgId, "retention.purged", {
        scope: "audit_logs",
        counts: { auditLogs: res.count }
      });
    }
  }
  // Dry-run: report the would-be row count.
  return { ...counts, comments: execute ? counts.comments : total };
}

/**
 * files — `File.deletedAt < cutoff` in the org's workspaces. Versions go
 * with the file; `bytesFreed` sums the deleted rows' current `size`.
 * (S3 object deletion is the files lifecycle policy's job, as in the old
 * sweep — the DB rows + accounting are what retention owns here.)
 */
async function purgeFiles(
  prisma: PrismaClient,
  orgId: string,
  wsIds: string[],
  cutoff: Date,
  execute: boolean,
  actor: RetentionActor
): Promise<ScopeCounts> {
  const where: Prisma.FileWhereInput = {
    workspaceId: { in: wsIds },
    deletedAt: { lt: cutoff }
  };
  const rows = await prisma.file.findMany({ where, select: { id: true, size: true } });
  const counts: ScopeCounts = { pages: 0, blocks: 0, comments: 0, files: 0, bytesFreed: 0 };
  if (execute) {
    for (let i = 0; i < rows.length; i += RETENTION_BATCH_SIZE) {
      const batch = rows.slice(i, i + RETENTION_BATCH_SIZE);
      const bytes = batch.reduce((s, f) => s + (f.size ?? 0), 0);
      await prisma.$transaction(async (tx) => {
        await tx.fileVersion.deleteMany({ where: { fileId: { in: batch.map((f) => f.id) } } });
        await tx.file.deleteMany({ where: { id: { in: batch.map((f) => f.id) } } });
      });
      counts.files += batch.length;
      counts.bytesFreed += bytes;
      await writeAudit(prisma, actor, orgId, "retention.purged", {
        scope: "files",
        counts: { files: batch.length, bytesFreed: bytes }
      });
    }
  }
  // Dry-run: report the would-be count + the bytes that would be freed.
  return {
    ...counts,
    files: execute ? counts.files : rows.length,
    bytesFreed: execute ? counts.bytesFreed : rows.reduce((s, f) => s + (f.size ?? 0), 0)
  };
}

// ─────────────────────────────── run entry ──────────────────────────────────

/**
 * Run ONE (org, scope) in the given mode. Returns the report object (the
 * admin poll endpoint + the cron log both consume it).
 */
export async function runRetentionScope(args: RetentionRunOptions): Promise<Record<string, unknown>> {
  const { prisma, redis, orgId, scope, mode, actor, runId } = args;
  const execute = mode === "execute";
  const t0 = Date.now();
  const policy = await policyFor(prisma, orgId, scope);
  const base = {
    runId: runId ?? null,
    orgId,
    scope,
    mode,
    type: policy.type,
    retainDays: policy.retainDays,
    startedAt: new Date().toISOString()
  };

  // HOLD (or files retainDays 0 = no-purge) → skip.
  if (policy.type === "HOLD" || (scope === "files" && policy.retainDays <= 0)) {
    await writeAudit(prisma, actor, orgId, "retention.hold_skipped", {
      scope,
      reason: policy.type === "HOLD" ? "hold" : "no_purge"
    });
    const report = { ...base, status: "skipped", counts: null, reason: policy.type === "HOLD" ? "hold" : "no_purge" };
    await writeRunReport(redis, runId, orgId, report);
    log.info({ ...report }, "retention scope skipped (hold)");
    return report;
  }

  const cutoff = new Date(Date.now() - Math.max(policy.retainDays, 0) * 86_400_000);
  const wsIds = await orgWorkspaceIds(prisma, orgId);

  let counts: ScopeCounts;
  if (scope === "trash") counts = await purgeTrash(prisma, orgId, wsIds, cutoff, execute, actor);
  else if (scope === "comments") counts = await purgeComments(prisma, orgId, wsIds, cutoff, execute, actor);
  else if (scope === "audit_logs") counts = await purgeAuditLogs(prisma, orgId, wsIds, cutoff, execute, actor);
  else counts = await purgeFiles(prisma, orgId, wsIds, cutoff, execute, actor);

  const report = {
    ...base,
    status: execute ? "purged" : "dry_run",
    cutoff: cutoff.toISOString(),
    counts,
    durationMs: Date.now() - t0
  };
  if (!execute) {
    // The spec's dry-run audit: `retention.dry_run {scope, count, oldest,
    // newest}` — `count` is the primary items for the scope (pages for
    // trash, comments for comments, audit rows for audit_logs, files for
    // files). oldest/newest are the boundary timestamps of the purge set.
    const count =
      scope === "trash"
        ? counts.pages
        : scope === "comments"
          ? counts.comments
          : scope === "files"
            ? counts.files
            : counts.comments; // the audit_logs slot reuse (recorded)
    await writeAudit(prisma, actor, orgId, "retention.dry_run", { scope, count, counts });
  }
  await touchLastRun(prisma, orgId, scope, mode, { counts });
  await writeRunReport(redis, runId, orgId, report);
  log.info({ ...report }, "retention scope done");
  return report;
}

/**
 * Run ALL scopes for an org (the daily 04:00 job, per org). Per-scope
 * isolation: a throw in one scope → `retention.failed` audit + the others
 * still run. Returns the per-scope report map.
 */
export async function runOrgRetention(
  args: Omit<RetentionRunOptions, "scope"> & { scopes?: RetentionScope[] }
): Promise<Record<string, Record<string, unknown>>> {
  const { prisma, redis, orgId, mode, actor, runId, scopes } = args;
  const out: Record<string, Record<string, unknown>> = {};
  for (const scope of scopes ?? RETENTION_SCOPES) {
    try {
      out[scope] = await runRetentionScope({ ...args, scope });
    } catch (err) {
      const msg = (err as Error).message;
      log.error({ err: msg, orgId, scope }, "retention scope failed");
      await prisma.auditLog
        .create({
          data: {
            actorType: actor.type,
            actorId: actor.id,
            workspaceId: null,
            resourceType: "organization",
            resourceId: orgId,
            action: "retention.failed",
            metadata: asJson({ orgId, scope, error: msg })
          }
        })
        .catch(() => undefined);
      out[scope] = { orgId, scope, mode, status: "failed", error: msg };
      // The admin poll must not see a forever-"processing" report.
      await writeRunReport(redis, runId, orgId, { orgId, scope, mode, status: "failed", error: msg });
    }
  }
  return out;
}

/** The daily job body: every org, per-org policies, mode from the gate. */
export async function runRetentionDaily(
  prisma: PrismaClient,
  redis: Redis,
  env: { WORKER_RETENTION_EXECUTE: boolean }
): Promise<Record<string, unknown>> {
  const mode = env.WORKER_RETENTION_EXECUTE ? "execute" : "dry";
  const orgs = await prisma.organization.findMany({ select: { id: true } });
  const perOrg: Record<string, Record<string, unknown>> = {};
  for (const org of orgs) {
    perOrg[org.id] = await runOrgRetention({
      prisma,
      redis,
      orgId: org.id,
      mode,
      actor: { type: "SYSTEM", id: "worker:maintenance" }
    });
  }
  log.info({ orgs: orgs.length, mode }, "retention daily sweep done");
  return { orgs: orgs.length, mode, perOrg };
}


// ─────────────────────── page-version prune (T-152) ────────────────────────
//
// Per-WORKSPACE revision retention (distinct from the org RetentionPolicy
// engine above, which owns trash/comments/audit/files): prune old PageVersion
// snapshots per page. Policy lives in `workspace.settings.versions`
// ({ maxVersions, keepRecentDays }) — parsed by the shared
// `versionRetentionFromSettings` (defaults: keep 100 newest + 7 d grace).
// Idempotent: a second run finds nothing left to prune.

export interface PageVersionPruneResult {
  workspaceId: string;
  pages: number;
  versions: number;
  maxVersions: number;
  keepRecentDays: number;
}

/**
 * Prune page versions for ONE workspace per its settings policy.
 * Writes one audit row (`page.versions_pruned`) when anything was removed.
 */
export async function pruneWorkspacePageVersions(
  prisma: PrismaClient,
  workspaceId: string,
  actor: RetentionActor
): Promise<PageVersionPruneResult> {
  const ws = await prisma.workspace.findUnique({ where: { id: workspaceId }, select: { settings: true } });
  if (!ws) return { workspaceId, pages: 0, versions: 0, maxVersions: 0, keepRecentDays: 0 };
  const policy = versionRetentionFromSettings(ws.settings);
  const pages = await prisma.page.findMany({ where: { workspaceId }, select: { id: true } });
  let prunedPages = 0;
  let prunedVersions = 0;
  for (const page of pages) {
    const rows = await prisma.pageVersion.findMany({
      where: { pageId: page.id },
      select: { id: true, version: true, createdAt: true }
    });
    const toPrune = planPageVersionPrune(rows, policy);
    if (toPrune.length === 0) continue;
    const res = await prisma.pageVersion.deleteMany({ where: { pageId: page.id, id: { in: toPrune } } });
    prunedPages++;
    prunedVersions += res.count;
  }
  if (prunedVersions > 0) {
    await prisma.auditLog.create({
      data: {
        actorType: actor.type,
        actorId: actor.id,
        workspaceId,
        resourceType: "page",
        resourceId: workspaceId,
        action: "page.versions_pruned",
        metadata: asJson({ pages: prunedPages, versions: prunedVersions, ...policy })
      }
    });
  }
  log.info({ workspaceId, pages: prunedPages, versions: prunedVersions }, "page-version prune done");
  return { workspaceId, pages: prunedPages, versions: prunedVersions, maxVersions: policy.maxVersions, keepRecentDays: policy.keepRecentDays };
}

/** Daily sweep helper: every workspace (the `retention` repeatable calls it
 *  after the org sweep — same 04:00 run, one extra pass). */
export async function pruneAllPageVersions(
  prisma: PrismaClient,
  actor: RetentionActor
): Promise<{ workspaces: number; pages: number; versions: number }> {
  const wss = await prisma.workspace.findMany({ select: { id: true } });
  let pages = 0;
  let versions = 0;
  for (const ws of wss) {
    const r = await pruneWorkspacePageVersions(prisma, ws.id, actor);
    pages += r.pages;
    versions += r.versions;
  }
  return { workspaces: wss.length, pages, versions };
}