/**
 * apps/worker — the async AUDIT CSV EXPORT job (TASK-0133).
 *
 * `POST /orgs/:orgId/admin/audit/export` with >10k matching rows returns
 * `202 {exportId}` and enqueues THIS job (`maintenance` queue, `audit.export`).
 * It streams the org-scoped rows (cursor, 5k/batch, the SAME org scope the
 * viewer uses — `@factory/shared/events/audit-scope`), renders the stable
 * 13-column CSV (`@factory/shared/events/audit-csv`), caps at 500k rows
 * (`truncated: true`), uploads to the files bucket under
 * `exports/{orgId}/{exportId}.csv` (7-day retention — the object lifecycle
 * policy), and flips the `AuditExport` row to `ready` (or `failed`).
 *
 * The API's `GET .../audit/exports/:id` then mints a SINGLE-USE, 15-minute
 * presigned GET (the worker stores no URLs — only the object key, which is a
 * deterministic function of the ids).
 *
 * Memory note (documented, mirrors the files pipeline): the rendered CSV is
 * accumulated in memory before the one-shot `putObject` (the worker's SigV4
 * client hashes the body, so it cannot stream an upload). This is bounded by
 * the 500k-row cap; production orgs well under the cap keep the buffer small.
 *
 * Name resolution: incremental — each 5k batch bulk-fetches only the
 * user/workspace names it has not seen yet, so a 100k-row export issues a
 * handful of lookups (not 100k, and not one per row).
 */
import type { PrismaClient, Prisma } from "@prisma/client";
import type { Redis } from "ioredis";
import { S3Client } from "../lib/s3.js";
import { asJson } from "../lib/prisma.js";
import { getLogger } from "../logger.js";
import { loadEnv } from "../config.js";
import { auditScopedWhere, type AuditFilters } from "@factory/shared/events/audit-scope";
import { AUDIT_CSV_HEADER, auditCsvLine, type AuditCsvRow } from "@factory/shared/events/audit-csv";

const log = getLogger("audit-export");

const BATCH = 5_000;
export const ASYNC_ROW_CAP = 500_000;

/** Rebuild the viewer's filter set from the stored JSON (Dates restored). */
function parseStoredFilters(raw: unknown): AuditFilters {
  const f = (raw as Record<string, unknown> | null) ?? {};
  const date = (v: unknown): Date | undefined =>
    typeof v === "string" && v ? new Date(v) : undefined;
  return {
    actorId: typeof f.actorId === "string" ? f.actorId : undefined,
    actorType: typeof f.actorType === "string" ? f.actorType : undefined,
    action: typeof f.action === "string" ? f.action : undefined,
    actionPrefix: typeof f.actionPrefix === "string" ? f.actionPrefix : undefined,
    resourceType: typeof f.resourceType === "string" ? f.resourceType : undefined,
    workspaceId: typeof f.workspaceId === "string" ? f.workspaceId : undefined,
    from: date(f.from),
    to: date(f.to)
  };
}

/**
 * @param opts.rowCap — the truncation cap (default `ASYNC_ROW_CAP`);
 * injectable ONLY so the unit test can exercise the truncation path with a
 * small number (seeding 500k rows is not a realistic unit fixture).
 */
export async function runAuditExport(
  prisma: PrismaClient,
  redis: Redis,
  orgId: string,
  exportId: string,
  opts: { rowCap?: number } = {}
): Promise<Record<string, unknown>> {
  const t0 = Date.now();
  const rowCap = opts.rowCap ?? ASYNC_ROW_CAP;
  const key = `exports/${orgId}/${exportId}.csv`;
  const mark = (patch: Record<string, unknown>) =>
    prisma.auditExport.update({ where: { id: exportId }, data: { ...patch } as never });

  try {
    const row = await prisma.auditExport.findUnique({ where: { id: exportId } });
    if (!row || row.organizationId !== orgId) {
      throw new Error("export row not found (or cross-org)");
    }
    const filters = parseStoredFilters(row.filters);
    const wsRows = await prisma.workspace.findMany({
      where: { organizationId: orgId },
      select: { id: true }
    });
    const wsIds = wsRows.map((w) => w.id);
    const baseWhere = auditScopedWhere(orgId, wsIds, filters) as Prisma.AuditLogWhereInput;

    const userNameCache = new Map<string, string>();
    const wsNameCache = new Map<string, string>();
    const lines: string[] = [AUDIT_CSV_HEADER];
    let rows = 0;
    let truncated = false;

    // Cursor over (createdAt DESC, id DESC): after each full batch, the next
    // query is anchored strictly after the batch's last row (keyset — stable
    // even under concurrent inserts; a 90-day window + org scope keeps it
    // index-driven).
    let cursor: Prisma.AuditLogWhereInput = {};
    for (;;) {
      const batch = await prisma.auditLog.findMany({
        where: { AND: [baseWhere, cursor] },
        orderBy: [{ createdAt: "desc" }, { id: "desc" }],
        take: BATCH
      });
      if (!batch.length) break;

      // Bulk-fetch this batch's unseen names (incremental caches).
      const batchUsers = [...new Set(batch.filter((r) => r.actorType === "USER").map((r) => r.actorId))];
      const batchWs = [...new Set(batch.map((r) => r.workspaceId).filter((w): w is string => Boolean(w)))];
      if (batchUsers.length) {
        const users = await prisma.user.findMany({
          where: { id: { in: batchUsers } },
          select: { id: true, name: true }
        });
        for (const u of users) userNameCache.set(u.id, u.name);
      }
      if (batchWs.length) {
        const wss = await prisma.workspace.findMany({
          where: { id: { in: batchWs } },
          select: { id: true, name: true }
        });
        for (const w of wss) wsNameCache.set(w.id, w.name);
      }

      for (const r of batch) {
        if (rows >= rowCap) {
          truncated = true;
          break;
        }
        rows++;
        const cr: AuditCsvRow = {
          id: r.id,
          createdAt: r.createdAt,
          actorType: r.actorType,
          actorId: r.actorId,
          workspaceId: r.workspaceId,
          resourceType: r.resourceType,
          resourceId: r.resourceId,
          action: r.action,
          ipAddress: r.ipAddress,
          userAgent: r.userAgent,
          metadata: r.metadata
        };
        lines.push(auditCsvLine(cr, cr.actorType === "USER" ? (userNameCache.get(cr.actorId) ?? "-") : "-", wsNameCache));
      }
      if (truncated) break;
      if (batch.length < BATCH) break; // last (short) batch
      const last = batch[batch.length - 1]!;
      cursor = {
        OR: [
          { createdAt: { lt: last.createdAt } },
          { createdAt: last.createdAt, id: { lt: last.id } }
        ]
      };
    }

    const csv = lines.join("\n") + "\n";
    const env = loadEnv();
    const s3 = new S3Client({
      endpoint: env.S3_ENDPOINT.replace(/\/+$/, ""),
      region: env.S3_REGION,
      bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
      accessKey: env.S3_ACCESS_KEY,
      secretKey: env.S3_SECRET_KEY
    });
    await s3.putObject(key, Buffer.from(csv, "utf8"), "text/csv; charset=utf-8");

    await mark({ status: "ready", rows, truncated, completedAt: new Date(), error: null });
    await prisma.auditLog.create({
      data: {
        actorType: "SYSTEM",
        actorId: "worker:maintenance",
        workspaceId: null,
        resourceType: "organization",
        resourceId: orgId,
        action: "audit.export.completed",
        metadata: asJson({ orgId, exportId, rows, truncated, key, durationMs: Date.now() - t0 })
      }
    });
    log.info({ orgId, exportId, rows, truncated, durationMs: Date.now() - t0 }, "audit export ready");
    return { exportId, status: "ready", rows, truncated, key };
  } catch (err) {
    const msg = (err as Error).message;
    log.error({ orgId, exportId, err: msg }, "audit export failed");
    await mark({ status: "failed", error: msg, completedAt: new Date() }).catch(() => undefined);
    await prisma.auditLog
      .create({
        data: {
          actorType: "SYSTEM",
          actorId: "worker:maintenance",
          workspaceId: null,
          resourceType: "organization",
          resourceId: orgId,
          action: "audit.export.failed",
          metadata: asJson({ orgId, exportId, error: msg })
        }
      })
      .catch(() => undefined);
    throw err; // let BullMQ record the job failure (retries per the worker config)
  }
  void redis;
}