/**
 * TASK-0133 — the async (>10k) audit CSV EXPORT, the two halves of the
 * `> SYNC_ROW_LIMIT (10_000)` path:
 *
 *   PART 1 (always runs, no S3 network call):
 *     - the API switches on the matching row count: >10k → `202 {exportId,
 *       cap, total}` + an enqueued `maintenance`/`audit.export` job. The 202
 *       IS the enqueue assertion (the controller `await`s `enqueueExportJob`
 *       before responding).
 *     - the worker flips the `AuditExport` row to `ready` (simulated here by
 *       updating the row — the real S3 write is Part 2).
 *     - `GET .../audit/exports/:id` mints a SINGLE-USE 15-minute presigned
 *       GET. This is PURE SIGNING (no S3 round trip), so it runs without
 *       MinIO: the first call returns the URL + flips `downloadUsed`; every
 *       subsequent call (with or without `?redownload=1`) returns no URL.
 *
 *   PART 2 (`describe.skipIf(!minioReady)` — the real S3 leg):
 *     - calls the worker's `runAuditExport` directly (the same function the
 *       `maintenance` processor invokes) against real Postgres + a real
 *       S3/MinIO: it streams the org-scoped rows, writes the 13-column CSV to
 *       `exports/{orgId}/{exportId}.csv`, flips the row to `ready`, and
 *       emits `audit.export.completed`. We then read the object back from S3
 *       (content-type + header + row count) and assert the audit row.
 *
 * Truncation at the 500k cap is UNIT-tested (the cap is a non-injectable
 * constant; seeding 500k rows is not a realistic integration fixture).
 *
 * Org scoping: `AuditLog` has NO `organizationId` column — the org scope is
 * `workspaceId IN (org's workspaces)` OR (org-level rows). We therefore create
 * the workspace with `organizationId = orgId` and seed rows against it
 * (`@factory/shared/events/audit-scope` is the single source of truth).
 *
 * NOTE (integration fetch bodies): `app.request` returns a `Response` whose
 * body is a read-once stream. Every call is funneled through `read()`, which
 * consumes the body ONCE (raw text + a JSON parse) so an assertion message can
 * quote the raw body without a second read.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { PrismaClient } from "@prisma/client";
import { Redis } from "ioredis";
import { createTestApp, type TestAppHandle } from "./helpers/api";
import { authHeaders, createUser } from "./helpers/seed";
import { getRedisUrl } from "./helpers/containers";
import { runAuditExport, ASYNC_ROW_CAP } from "../../apps/worker/src/processors/audit-export.js";
import { resetEnvCache } from "../../apps/worker/src/config.js";
import { S3Client } from "../../apps/worker/src/lib/s3.js";
import { AUDIT_CSV_HEADER } from "@factory/shared";

// ── MinIO gate (Part 2) ─────────────────────────────────────────────────
const S3_ENDPOINT = (process.env.S3_ENDPOINT ?? "http://localhost:9000").replace(/\/+$/, "");
const S3_REGION = process.env.S3_REGION ?? "us-east-1";
const S3_BUCKET = process.env.S3_BUCKET_FILES ?? process.env.S3_BUCKET ?? "factory-files";
const S3_ACCESS = process.env.S3_ACCESS_KEY ?? "minioadmin";
const S3_SECRET = process.env.S3_SECRET_KEY ?? "minioadmin";

async function minioUp(url: string): Promise<boolean> {
  try {
    const res = await fetch(`${url}/minio/health/live`, { signal: AbortSignal.timeout(4_000) });
    return res.ok;
  } catch {
    return false;
  }
}
const minioReady = await minioUp(S3_ENDPOINT);

/** Read a response body ONCE (raw text + a JSON parse for the convenience). */
async function read(res: Response | Promise<Response>): Promise<{ status: number; raw: string; data: Record<string, unknown> }> {
  const r = await res;
  const raw = await r.text();
  let data: Record<string, unknown> = {};
  try {
    data = JSON.parse(raw) as Record<string, unknown>;
  } catch {
    /* not JSON (e.g. text/csv) */
  }
  return { status: r.status, raw, data };
}

let app: TestAppHandle;
let redis: Redis;
let prisma: PrismaClient;

/** Create an org through the API; the caller becomes the OWNER member. */
async function makeOrg(accessToken: string): Promise<string> {
  const out = await read(
    app.request("POST", "/api/v1/orgs", {
      headers: authHeaders(accessToken),
      body: { name: `AsyncExport Org ${Date.now()}-${Math.random().toString(36).slice(2, 7)}` }
    })
  );
  const orgId = out.data.id as string | undefined;
  if (!orgId) throw new Error(`makeOrg failed: ${out.status} ${out.raw}`);
  return orgId;
}

/** Seed N audit rows into the org (via an org-owned workspace) + return the ws. */
async function seedRows(orgId: string, ownerId: string, n: number): Promise<string> {
  const ws = await prisma.workspace.create({
    data: { name: "ws", slug: `ae-ws-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`, organizationId: orgId, ownerId }
  });
  const rows = Array.from({ length: n }, (_, i) => ({
    actorType: "USER" as const,
    actorId: ownerId,
    workspaceId: ws.id,
    resourceType: "page",
    resourceId: `page_${i}`,
    action: i % 2 === 0 ? "page.created" : "page.updated",
    metadata: {}
  }));
  // Batches keep each single insert statement small (createMany is one insert).
  for (let i = 0; i < rows.length; i += 2_000) {
    await prisma.auditLog.createMany({ data: rows.slice(i, i + 2_000) as never });
  }
  return ws.id;
}

/** Insert a processing AuditExport row (the API's async branch shape). */
async function seedExport(orgId: string, total: number): Promise<string> {
  const exportId = `ae-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
  await prisma.auditExport.create({
    data: { id: exportId, organizationId: orgId, status: "processing", totalRows: total, capRows: ASYNC_ROW_CAP, filters: JSON.stringify({}) as never }
  });
  return exportId;
}

describe("TASK-0133 async audit export (the >10k path)", () => {
  beforeAll(async () => {
    app = await createTestApp();
    prisma = app.prisma;
    redis = new Redis(await getRedisUrl(), { maxRetriesPerRequest: null });
  }, 180_000);
  afterAll(async () => {
    await redis?.quit().catch(() => {});
    await app?.close();
  });

  it("PART 1: API >10k → 202 {exportId,cap,total} (the enqueue), then a SINGLE-USE presigned URL (pure signing, no S3)", async () => {
    const owner = await createUser(app.request);
    const orgId = await makeOrg(owner.accessToken!); // also emits an org.created row
    // 10,001 workspace rows — just over SYNC_ROW_LIMIT (10_000) → the async
    // branch. (+ the org.created org-level row; the exact total is >10_000.)
    await seedRows(orgId, owner.userId, 10_001);

    const h = authHeaders(owner.accessToken!);
    const startRes = await read(app.request("POST", `/api/v1/orgs/${orgId}/admin/audit/export`, { headers: h, body: {} }));
    expect(startRes.status, startRes.raw).toBe(202);
    const start = startRes.data as { exportId: string; cap: number; total: number };
    expect(start.total).toBeGreaterThan(10_000);
    expect(start.cap).toBe(ASYNC_ROW_CAP);
    expect(start.exportId).toMatch(/^[A-Za-z0-9-]+$/);
    // The 202 only happens AFTER `await enqueueExportJob(...)` — so a 202 IS
    // the proof the `maintenance`/`audit.export` job was enqueued.

    // The row starts `processing`; no URL until it is ready.
    const pending = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
      status: string;
      downloadUrl?: string
    };
    expect(pending.status).toBe("processing");
    expect(pending.downloadUrl).toBeUndefined();

    // Simulate the worker finishing (the real S3 write is PART 2): flip the
    // row to `ready` with the row count.
    await prisma.auditExport.update({
      where: { id: start.exportId },
      data: { status: "ready", rows: 10_001, truncated: false, completedAt: new Date() }
    });

    // FIRST status call → mints the single-use URL (pure SigV4 signing).
    const first = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
      status: string;
      rows: number;
      truncated: boolean;
      downloadUrl?: string;
      urlTtlSec?: number
    };
    expect(first.status).toBe("ready");
    expect(first.rows).toBe(10_001);
    expect(first.truncated).toBe(false);
    expect(first.downloadUrl).toBeDefined();
    expect(first.urlTtlSec).toBe(900);
    // The signed URL targets the deterministic object key + carries SigV4 sig.
    expect(first.downloadUrl).toContain(`exports/${orgId}/${start.exportId}.csv`);
    expect(first.downloadUrl).toMatch(/X-Amz-Signature=/);

    // SINGLE-USE: the second call (and even `?redownload=1`) returns NO url —
    // `downloadUsed` was flipped on the first mint.
    const second = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
      downloadUrl?: string
    };
    expect(second.downloadUrl).toBeUndefined();
    const third = (await read(
      app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}?redownload=1`, { headers: h })
    )).data as { downloadUrl?: string };
    expect(third.downloadUrl).toBeUndefined();
  });

  describe.skipIf(!minioReady)("PART 2: worker runAuditExport writes the CSV to S3 (MinIO)", () => {
    it("streams the org-scoped rows → putObject → headObject/getObjectStream round trip + audit.export.completed", async () => {
      // Reset the worker's cached env so it parses the CURRENT process.env
      // (the test app set DATABASE_URL/REDIS_URL; the S3_* are at their
      // defaults = this MinIO endpoint).
      resetEnvCache();

      const owner = await createUser(app.request);
      const orgId = await makeOrg(owner.accessToken!); // emits the org.created org-level row
      const N = 120;
      await seedRows(orgId, owner.userId, N);
      const exportId = await seedExport(orgId, N);

      const res = await runAuditExport(prisma, redis, orgId, exportId);
      expect(res.status).toBe("ready");
      // The org scope = the N workspace rows + the org.created org-level row
      // (the OR-branch), so at least N (and in practice N+1) rows flow through.
      expect(res.rows).toBeGreaterThanOrEqual(N);
      expect(res.truncated).toBe(false);
      expect(res.key).toBe(`exports/${orgId}/${exportId}.csv`);

      // The DB row is ready with the same count.
      const row = await prisma.auditExport.findUnique({ where: { id: exportId } });
      expect(row?.status).toBe("ready");
      expect(row?.rows).toBe(res.rows);

      // The object is in S3 with the right content type + 13-column header, and
      // the CSV invariant: exactly one line per counted row (+ header).
      const s3 = new S3Client({ endpoint: S3_ENDPOINT, region: S3_REGION, bucket: S3_BUCKET, accessKey: S3_ACCESS, secretKey: S3_SECRET });
      const head = await s3.headObject(res.key as string);
      expect(head).not.toBeNull();
      expect(head?.contentType).toContain("text/csv");
      const buf = await s3.getObjectStream(res.key as string);
      const csv = buf.toString("utf8");
      const lines = csv.trim().split("\n");
      expect(lines[0]).toBe(AUDIT_CSV_HEADER);
      expect(lines.length).toBe(res.rows + 1); // header + res.rows rows

      // The completion audit row (actor SYSTEM, action audit.export.completed).
      const audit = await prisma.auditLog.findFirst({
        where: { action: "audit.export.completed", resourceId: orgId, metadata: { path: ["exportId"], equals: exportId } }
      });
      expect(audit).not.toBeNull();
    });
  });
});