import { createClient } from "@supabase/supabase-js";
import { existsSync } from "node:fs";
import { randomUUID, createHash } from "node:crypto";
import { mkdir, writeFile, readFile, readdir, rm } from "node:fs/promises";
import { gunzipSync } from "node:zlib";
import sharp from "sharp";
import { datasetSchema, faces, type TerrainSnapshot } from "../shared/protocol";
import { renderSnapshot } from "../renderer/render.mjs";
import type { Database } from "../shared/database";
if (existsSync(".env.local")) process.loadEnvFile(".env.local");
const client = createClient<Database>(
  process.env.NEXT_PUBLIC_SUPABASE_URL!,
  process.env.SUPABASE_SECRET_KEY!,
  { auth: { persistSession: false, autoRefreshToken: false } },
);
const worker = "worker-" + randomUUID();
let stopping = false;
process.on("SIGINT", () => (stopping = true));
process.on("SIGTERM", () => (stopping = true));
function check<T>(r: {
  data: T;
  error: { message: string } | null;
}): NonNullable<T> {
  if (r.error) throw new Error(r.error.message);
  return r.data!;
}
async function processJob(job: {
  id: string;
  dataset_id: string;
  attempts: number;
}) {
  const abort = new AbortController(),
    directory = `.cache/jobs/${job.id}`;
  await mkdir(directory, { recursive: true });
  let lastProgressWrite = 0;
  let lastLoggedAt = 0;
  let lastLoggedPercent = -1;
  async function progress(percent: number, message: string) {
    if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
    const value = Math.max(0, Math.min(99, Math.floor(percent)));
    const now = Date.now();
    if (now - lastLoggedAt >= 30000 || value >= lastLoggedPercent + 5) {
      console.log(`Job ${job.id}: ${value}% — ${message}`);
      lastLoggedAt = now;
      lastLoggedPercent = value;
    }
    if (now - lastProgressWrite < 2000 && value !== 99) return;
    lastProgressWrite = now;
    const rows = check(
      await client
        .from("upload_jobs")
        .update({
          progress: value,
          error: null,
          updated_at: new Date().toISOString(),
        })
        .eq("id", job.id)
        .eq("lease_owner", worker)
        .eq("state", "running")
        .select("id"),
    );
    if (!rows.length) {
      abort.abort();
      throw new Error("Lease lost or cancelled");
    }
  }
  const heartbeat = setInterval(async () => {
    try {
      const rows = check(
        await client
          .from("upload_jobs")
          .update({
            lease_until: new Date(Date.now() + 120000).toISOString(),
            updated_at: new Date().toISOString(),
          })
          .eq("id", job.id)
          .eq("lease_owner", worker)
          .eq("state", "running")
          .select("id"),
      );
      if (!rows.length) abort.abort();
    } catch {
      abort.abort();
    }
  }, 20000);
  try {
    const dataset = check(
      await client
        .from("datasets")
        .select("*")
        .eq("id", job.dataset_id)
        .single(),
    );
    const input = check(
      await client
        .from("dataset_inputs")
        .select("manifest")
        .eq("dataset_id", job.dataset_id)
        .single(),
    );
    const manifest = datasetSchema.parse(input.manifest);
    await progress(0, `Downloading ${manifest.assets.length} export files`);
    let downloaded = 0;
    for (const asset of manifest.assets) {
      if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
      const blob = check(
        await client.storage
          .from("minegeo-raw")
          .download(`${dataset.owner_id}/${job.dataset_id}/${asset.path}`),
      );
      const bytes = Buffer.from(await blob.arrayBuffer());
      if (
        bytes.length !== asset.bytes ||
        createHash("sha256").update(bytes).digest("hex") !== asset.sha256
      )
        throw new Error(`Asset checksum/size mismatch: ${asset.path}`);
      const path = `${directory}/raw/${asset.path}`;
      await mkdir(path.slice(0, path.lastIndexOf("/")), { recursive: true });
      await writeFile(path, bytes);
      await progress(
        (20 * ++downloaded) / manifest.assets.length,
        `Downloaded ${downloaded}/${manifest.assets.length} export files`,
      );
    }
    const output = `${directory}/media`;
    await mkdir(output, { recursive: true });
    if (manifest.source === "plugin") {
      await progress(20, "Reading the terrain snapshot");
      const compressed = await readFile(`${directory}/raw/terrain.json.gz`);
      const terrain = JSON.parse(
        gunzipSync(compressed, {
          maxOutputLength: 256 * 1024 * 1024,
        }).toString(),
      ) as TerrainSnapshot;
      if (
        !Array.isArray(terrain.blocks) ||
        terrain.blocks.length > 5000000 ||
        !Array.isArray(terrain.coverage)
      )
        throw new Error("Terrain snapshot exceeds supported render budget");
      if (
        ["minX", "minZ", "maxX", "maxZ"].some(
          (k) =>
            terrain.bounds[k as keyof typeof terrain.bounds] !==
            manifest.bounds[k as keyof typeof manifest.bounds],
        )
      )
        throw new Error("Snapshot bounds disagree with manifest");
      for (const b of terrain.blocks) {
        if (
          !Array.isArray(b) ||
          b.length < 4 ||
          !b.slice(0, 3).every((v) => Number.isInteger(v)) ||
          typeof b[3] !== "string" ||
          !/^minecraft:[a-z0-9_]+(\[[a-z0-9_=,]+\])?$/.test(b[3])
        )
          throw new Error("Invalid block record");
      }
      console.log(
        `Job ${job.id}: ${terrain.blocks.length.toLocaleString()} blocks, ${manifest.viewpoints.length} panoramas; software rendering may take time.`,
      );
      const report = await renderSnapshot(
        terrain,
        manifest.viewpoints,
        output,
        {
          size: Number(process.env.MINEGEO_FACE_SIZE || 1024),
          signal: abort.signal,
          onProgress: ({ percent, message }) => progress(percent, message),
        },
      );
      console.log(`Rendered ${job.dataset_id}: ${JSON.stringify(report)}`);
    } else {
      await progress(25, "Compressing the captured map");
      await sharp(await readFile(`${directory}/raw/map.png`), {
        limitInputPixels: 16777216,
      })
        .webp({ quality: 90 })
        .toFile(`${output}/map.webp`);
      let processedFaces = 0;
      for (const point of manifest.viewpoints) {
        await mkdir(`${output}/${point.id}`, { recursive: true });
        for (const face of faces) {
          const bytes = await readFile(
            `${directory}/raw/${point.id}/${face}.png`,
          );
          const meta = await sharp(bytes, {
            limitInputPixels: 16777216,
          }).metadata();
          if (
            !meta.width ||
            meta.width !== meta.height ||
            meta.width < 256 ||
            meta.width > 2048
          )
            throw new Error("Panorama faces must be square, 256–2048 pixels");
          await sharp(bytes, { limitInputPixels: 16777216 })
            .webp({ quality: 85 })
            .toFile(`${output}/${point.id}/${face}.webp`);
          if (face === "nz")
            await sharp(bytes)
              .resize(512, 512)
              .webp({ quality: 76 })
              .toFile(`${output}/${point.id}/preview.webp`);
          await progress(
            35 +
              (55 * ++processedFaces) /
                (manifest.viewpoints.length * faces.length),
            `Compressing panorama faces ${processedFaces}/${manifest.viewpoints.length * faces.length}`,
          );
        }
      }
    }
    const mapMeta = await sharp(`${output}/map.webp`).metadata();
    if (!mapMeta.width) throw new Error("Missing map output");
    await progress(90, "Saving processed images");
    const images = (await readdir(output, { recursive: true })).filter(
      (entry) => entry.endsWith(".webp"),
    );
    let uploaded = 0;
    for (const entry of images) {
      if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
      const normalized = entry.replaceAll("\\", "/");
      check(
        await client.storage
          .from("minegeo-media")
          .upload(
            `${job.dataset_id}/${normalized}`,
            await readFile(`${output}/${entry}`),
            {
              contentType: "image/webp",
              upsert: true,
              cacheControl: "31536000",
            },
          ),
      );
      await progress(
        90 + (9 * ++uploaded) / images.length,
        `Saved ${uploaded}/${images.length} processed images`,
      );
    }
    if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
    check(
      await client.rpc("complete_job", {
        p_job: job.id,
        p_worker: worker,
        p_count: manifest.viewpoints.length,
        p_cover: manifest.viewpoints[0].id,
      }),
    );
    console.log(`Job ${job.id}: 100% — private draft ready for preview.`);
  } catch (error) {
    const message =
      error instanceof Error ? error.message : "Processing failed";
    console.error(`Job ${job.id}: ${message}`);
    const state = job.attempts >= 3 ? "failed" : "queued";
    const rows = check(
      await client
        .from("upload_jobs")
        .update({ state, error: message, lease_until: null, progress: 0 })
        .eq("id", job.id)
        .eq("lease_owner", worker)
        .eq("state", "running")
        .select("id"),
    );
    if (state === "failed" && rows.length)
      await client
        .from("datasets")
        .update({ state: "failed", rejection_reason: message })
        .eq("id", job.dataset_id)
        .eq("state", "processing");
  } finally {
    clearInterval(heartbeat);
    await rm(directory, { recursive: true, force: true });
  }
}
let housekeeping = 0;
console.log(`MineGeo ${worker} ready; PostgreSQL leases through Supabase API.`);
while (!stopping) {
  try {
    if (Date.now() - housekeeping > 300000) {
      housekeeping = Date.now();
      const dead = check(
        await client
          .from("upload_jobs")
          .update({
            state: "failed",
            error: "Worker lease expired after three attempts",
          })
          .eq("state", "running")
          .lt("lease_until", new Date().toISOString())
          .gte("attempts", 3)
          .select("dataset_id"),
      );
      for (const j of dead)
        await client
          .from("datasets")
          .update({
            state: "failed",
            rejection_reason: "Processing exhausted retry limit",
          })
          .eq("id", j.dataset_id)
          .eq("state", "processing");
      await client
        .from("pairing_codes")
        .delete()
        .lt("expires_at", new Date().toISOString());
      await client
        .from("rate_limits")
        .delete()
        .lt("expires_at", new Date().toISOString());
      await client
        .from("game_sessions")
        .delete()
        .is("owner_id", null)
        .lt("created_at", new Date(Date.now() - 30 * 86400000).toISOString());
      const abandoned = check(
        await client
          .from("datasets")
          .select("id,owner_id")
          .eq("state", "uploading")
          .lt("created_at", new Date(Date.now() - 86400000).toISOString()),
      );
      for (const d of abandoned) {
        const input = check(
          await client
            .from("dataset_inputs")
            .select("manifest")
            .eq("dataset_id", d.id)
            .single(),
        );
        await client.storage
          .from("minegeo-raw")
          .remove(
            datasetSchema
              .parse(input.manifest)
              .assets.map(
                (a: { path: string }) => `${d.owner_id}/${d.id}/${a.path}`,
              ),
          );
        await client
          .from("datasets")
          .delete()
          .eq("id", d.id)
          .eq("state", "uploading");
      }
    }
    const jobs = check(await client.rpc("claim_job", { p_worker: worker }));
    if (jobs?.length) await processJob(jobs[0]);
    else await new Promise((r) => setTimeout(r, 2500));
  } catch (error) {
    console.error("Worker:", error instanceof Error ? error.message : error);
    await new Promise((r) => setTimeout(r, 5000));
  }
}