// Live end-to-end: upload a real Semrush fixture to R2 → enqueue an analytics_jobs
// row → the running worker drains it (R2 read_csv → DuckDB → upsert → R2 delete) →
// assert keywords upserted in Postgres + job `done` + object gone. Cleans up after.
// Run (with the service running): cd services/analytics &&
//   bun --env-file=../../.env.local run scripts/e2e-r2.ts
import { randomUUID } from "node:crypto";
import { S3Client } from "bun";
import { and, eq, inArray, notInArray } from "drizzle-orm";
import { drizzle } from "drizzle-orm/postgres-js";
import postgres from "postgres";
import * as schema from "@/lib/db/domain-schema";

const fail = (m: string): never => {
  console.error(`E2E_FAIL: ${m}`);
  process.exit(1);
};

const DATABASE_URL = process.env.DATABASE_URL ?? fail("DATABASE_URL missing");
const bucket = process.env.R2_BUCKET ?? fail("R2_BUCKET missing");
const accountId = process.env.R2_ACCOUNT_ID ?? fail("R2_ACCOUNT_ID missing");

const sql = postgres(DATABASE_URL as string, { max: 2 });
const db = drizzle(sql, { schema });
const s3 = new S3Client({
  accessKeyId: process.env.R2_ACCESS_KEY_ID,
  secretAccessKey: process.env.R2_SECRET_ACCESS_KEY,
  bucket: bucket as string,
  endpoint: `https://${accountId}.r2.cloudflarestorage.com`,
  region: "auto",
});

const EXPECTED_SLUGS = ["seo-audit", "link-building"];

try {
  // 1. Pick a real site.
  const [site] = await db.select().from(schema.sites).limit(1);
  if (!site) fail("no sites in DB — create one first");
  console.log(`[1] site: ${site.id} (${site.name ?? site.domain ?? "?"})`);

  // 2. Snapshot pre-existing keyword ids for this site (so cleanup only removes ours).
  const before = new Set(
    (
      await db
        .select({ id: schema.keywords.id })
        .from(schema.keywords)
        .where(eq(schema.keywords.siteId, site.id))
    ).map((r) => r.id),
  );

  // 3. Upload the Semrush fixture to a server-style key.
  const objectKey = `exports/${site.id}/${randomUUID()}.csv`;
  const csv = await Bun.file("test/fixtures/semrush.csv").text();
  await s3.file(objectKey).write(csv, { type: "text/csv" });
  console.log(`[2] uploaded → r2://${bucket}/${objectKey}`);

  // 4. Enqueue (the running worker will claim it via SKIP LOCKED).
  const jobId = randomUUID();
  await db.insert(schema.analyticsJobs).values({
    id: jobId,
    siteId: site.id,
    source: "semrush",
    objectKey,
    status: "queued",
  });
  console.log(`[3] enqueued job ${jobId}`);

  // 5. Poll until terminal (≤30s).
  let job: typeof schema.analyticsJobs.$inferSelect | undefined;
  for (let i = 0; i < 60; i++) {
    await Bun.sleep(500);
    [job] = await db
      .select()
      .from(schema.analyticsJobs)
      .where(eq(schema.analyticsJobs.id, jobId));
    if (job && (job.status === "done" || job.status === "failed")) break;
  }
  if (!job) fail("job row vanished");
  console.log(
    `[4] job status: ${job.status}  totals: ${JSON.stringify(job.totals)}`,
  );
  if (job.status !== "done")
    fail(`job not done (status=${job.status}, error=${job.error})`);

  // 6. Assert keywords upserted.
  const upserted = await db
    .select()
    .from(schema.keywords)
    .where(
      and(
        eq(schema.keywords.siteId, site.id),
        inArray(
          schema.keywords.id,
          EXPECTED_SLUGS.map((s) => `${site.id}-${s}`),
        ),
      ),
    );
  console.log(
    `[5] keywords upserted: ${upserted.map((k) => `${k.term}[${k.primarySource},vol=${k.metrics.semrush?.volume}]`).join(", ")}`,
  );
  if (upserted.length !== EXPECTED_SLUGS.length)
    fail(`expected ${EXPECTED_SLUGS.length} keywords, got ${upserted.length}`);
  const seoAudit = upserted.find((k) => k.id === `${site.id}-seo-audit`);
  if (seoAudit?.metrics.semrush?.volume !== 1000)
    fail(`seo-audit volume wrong: ${seoAudit?.metrics.semrush?.volume}`);
  if (job.totals.read !== 3)
    fail(`totals.read expected 3, got ${job.totals.read}`);

  // 7. Object should be deleted by the worker on success.
  const stillThere = await s3.file(objectKey).exists();
  console.log(
    `[6] R2 object after import: ${stillThere ? "PRESENT" : "deleted"}`,
  );
  if (stillThere) fail("worker did not delete the object after success");

  // 8. Cleanup — only the rows WE created.
  const created = upserted.map((k) => k.id).filter((id) => !before.has(id));
  if (created.length) {
    await db.delete(schema.keywords).where(
      and(
        eq(schema.keywords.siteId, site.id),
        inArray(schema.keywords.id, created),
        // belt-and-suspenders: never touch pre-existing ids
        notInArray(
          schema.keywords.id,
          [...before].length ? [...before] : ["__none__"],
        ),
      ),
    );
  }
  // dummy use of `like` import removed; delete the job row.
  await db
    .delete(schema.analyticsJobs)
    .where(eq(schema.analyticsJobs.id, jobId));
  console.log(`[7] cleaned up ${created.length} keyword(s) + job row`);

  console.log(
    "\nE2E_OK — full live pipeline (R2 → DuckDB → Postgres upsert → R2 delete) passed.",
  );
  await sql.end();
  process.exit(0);
} catch (e) {
  console.error(e);
  await sql.end();
  fail(String(e));
}