import { mkdtemp, mkdir, readFile, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { Readable } from "node:stream";
import sharp from "sharp";
import { describe, expect, it, vi } from "vitest";
import * as harness from "../src/index.js";
import type { ImageVideoRemote, ImageVideoRecord } from "../src/image-video/contracts.js";

async function fixture() {
  expect(harness).toHaveProperty("ImageVideoService");
  const root = await mkdtemp(join(tmpdir(), "i2v-test-"));
  await mkdir(join(root, "ses_owner"));
  const attachments = new harness.FileAttachmentStore({ sessionsRoot: root });
  const attachment = await attachments.add("ses_owner", { name: "test.png", mediaType: "image/png", stream: Readable.from(await sharp({ create: { width: 8, height: 8, channels: 3, background: "red" } }).png().toBuffer()) });
  const calls: string[] = [];
  const remote: ImageVideoRemote = {
    preflight() { calls.push("identity"); return Promise.resolve({ tenantId: "tenant", platform: "tiktok" }); },
    initialize() { calls.push("initialize"); return Promise.resolve({ uploadId: "upload_1", objectKey: "obj", uploadUrl: "https://public.example/put?token=private" }); },
    put() { calls.push("put"); return Promise.resolve(); }, finalize() { calls.push("finalize"); return Promise.resolve(); },
    importStatus() { calls.push("import"); return Promise.resolve({ state: "ready", assetId: "asset_1" }); },
    submit() { calls.push("submit"); return Promise.resolve({ jobId: "remote_job_1" }); },
    status() { calls.push("status"); return Promise.resolve({ state: "queued" }); },
  };
  const options = { root: join(root, "tasks"), attachments, remote, collect: () => Promise.reject(new Error("No fake video")) };
  const service = new harness.ImageVideoService(options);
  const draft = { attachmentId: attachment.attachmentId, title: "Motion", prompt: "Natural movement with ambient sound", durationSeconds: 10 };
  return { service, options, draft, calls, remote, root };
}

describe("durable image video tasks", () => {
  it("copies a local reference without transmitting it and hides private bindings", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    expect(f.calls).toEqual([]); expect(task.state).toBe("draft");
    await f.options.attachments.removeAll("ses_owner");
    expect((await f.service.media(task.id, "reference")).size).toBeGreaterThan(0);
    expect(JSON.stringify(await f.service.list())).not.toMatch(/ownerSessionId|uploadUrl|objectKey|referenceHash|\/private\//);
  });
  it("rejects foreign owners before any external operation", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await expect(f.service.upload("ses_other", task.id)).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
    await expect(f.service.submit("ses_other", task.id)).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
    expect(f.calls).toEqual([]);
  });
  it("requires upload first and coalesces concurrent submission, restoring only the known job", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await expect(f.service.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_NOT_READY" });
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
    const [a, b] = await Promise.all([f.service.submit("ses_owner", task.id), f.service.submit("ses_owner", task.id)]);
    expect(a.jobId).toBe("remote_job_1"); expect(b.jobId).toBe(a.jobId);
    const restored = new harness.ImageVideoService(f.options);
    expect((await restored.view(task.id)).jobId).toBe(a.jobId);
    await restored.submit("ses_owner", task.id);
    expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
    expect(JSON.stringify(await restored.view(task.id))).not.toContain("private");
  });
  it("isolates unknown submissions after timeout and after process interruption", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
    f.remote.submit = () => { f.calls.push("submit"); return Promise.reject(new Error("token=private")); };
    expect((await f.service.submit("ses_owner", task.id)).state).toBe("unknown_submission");
    await expect(new harness.ImageVideoService(f.options).submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" });
    expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
    const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
    record.state = "submitting"; await writeFile(file, JSON.stringify(record));
    expect((await new harness.ImageVideoService(f.options).view(task.id)).state).toBe("unknown_submission");
  });
  it("never reinitializes uncertain uploads and blocks stale persisted revisions", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    f.remote.initialize = () => { f.calls.push("initialize"); return Promise.reject(new Error("timeout")); };
    expect((await f.service.upload("ses_owner", task.id)).state).toBe("unknown_upload");
    await expect(f.service.upload("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_UPLOAD" });
    expect(f.calls.filter(call => call === "initialize")).toHaveLength(1);
    const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
    record.requestHash = "invalid"; await writeFile(file, JSON.stringify(record));
    await expect(f.service.view(task.id)).rejects.toMatchObject({ code: "I2V_RECORD_INVALID" });
  });
  it("throttles read-only polling even across instances and accepts no arbitrary identifiers", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
    await Promise.all([f.service.refresh(task.id), new harness.ImageVideoService(f.options).refresh(task.id)]);
    expect(f.calls.filter(call => call === "status")).toHaveLength(1);
    await expect(f.service.view("../../etc/passwd")).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
  });
  it("rejects a late submit result from overwriting a newer revision, while retaining its known receipt", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
    f.remote.submit = async () => {
      const file = join(f.options.root, task.id, "record.json"); const newer = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
      newer.revision += 1; newer.state = "unknown_submission"; await writeFile(file, JSON.stringify(newer));
      return { jobId: "late_job" };
    };
    await expect(f.service.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_STALE_REVISION" });
    expect((await f.service.view(task.id)).state).toBe("unknown_submission");
    const receipt = JSON.parse(await readFile(join(f.options.root, task.id, "submission-receipt.json"), "utf8")) as { jobId: string };
    expect(receipt.jobId).toBe("late_job");
  });
  it("recovers a matching persisted submit receipt after restart and polls that job without resubmitting", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
    const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
    record.state = "submitting"; record.revision++;
    await writeFile(file, JSON.stringify(record));
    await writeFile(join(f.options.root, task.id, "submission-receipt.json"), JSON.stringify({ version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "accepted_before_crash" }));
    f.calls.length = 0;
    f.remote.status = (r) => { f.calls.push(`status:${r.jobId ?? "missing"}`); return Promise.resolve({ state: "running" }); };
    const restored = new harness.ImageVideoService(f.options);
    expect(await restored.view(task.id)).toMatchObject({ state: "queued", jobId: "accepted_before_crash" });
    expect(JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord).toMatchObject({ state: "queued", jobId: "accepted_before_crash", revision: record.revision + 1 });
    expect(await restored.refresh(task.id)).toMatchObject({ state: "running", jobId: "accepted_before_crash" });
    await new harness.ImageVideoService(f.options).submit("ses_owner", task.id);
    expect(f.calls).toEqual(["status:accepted_before_crash"]);
  });
  it("retries a transient download after restart using the original job's current artifact, at the polling cadence", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
    const clock = vi.spyOn(Date, "now").mockReturnValue(2_000_000_000_000);
    const downloaded: string[] = []; let queries = 0;
    f.remote.status = r => { f.calls.push(`status:${r.jobId ?? "missing"}`); queries++; return Promise.resolve({ state: "succeeded", artifactUrl: `https://public.example/${String(queries)}.mp4` }); };
    const collect = (url: string) => {
      downloaded.push(url);
      if (downloaded.length === 1) return Promise.reject(new harness.HarnessError("I2V_TRANSFER_TRANSIENT", "Temporary download timeout"));
      return Promise.resolve({ videoSeconds: 10, audioSeconds: 10, width: 720, height: 1280, fps: 24, decoded: true as const, sha256: "a".repeat(64), visualReview: "pending" as const, audioReview: "pending" as const });
    };
    try {
      const initial = new harness.ImageVideoService({ ...f.options, collect });
      expect(await initial.refresh(task.id)).toMatchObject({ state: "validating", jobId: "remote_job_1" });
      const restored = new harness.ImageVideoService({ ...f.options, collect });
      await restored.refresh(task.id); expect(downloaded).toHaveLength(1);
      clock.mockReturnValue(2_000_000_029_999); await restored.refresh(task.id); expect(downloaded).toHaveLength(1);
      clock.mockReturnValue(2_000_000_030_000);
      expect(await restored.refresh(task.id)).toMatchObject({ state: "ready", jobId: "remote_job_1" });
      await restored.submit("ses_owner", task.id);
      expect(downloaded).toEqual(["https://public.example/1.mp4", "https://public.example/2.mp4"]);
      expect(f.calls.filter(call => call.startsWith("status:"))).toEqual(["status:remote_job_1", "status:remote_job_1"]);
      expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
    } finally { clock.mockRestore(); }
  });
  it("keeps deterministic invalid media terminal across restart with no false ready or second generation", async () => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
    f.remote.status = () => Promise.resolve({ state: "succeeded", artifactUrl: "https://public.example/invalid.mp4" });
    let collections = 0;
    const collect = () => { collections++; return Promise.reject(new harness.HarnessError("I2V_INVALID_MP4", "Invalid MP4")); };
    expect((await new harness.ImageVideoService({ ...f.options, collect }).refresh(task.id)).state).toBe("failed");
    const restored = new harness.ImageVideoService({ ...f.options, collect });
    expect(await restored.refresh(task.id)).toMatchObject({ state: "failed", jobId: "remote_job_1" });
    await restored.submit("ses_owner", task.id);
    expect(collections).toBe(1); expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
    await expect(restored.media(task.id, "video")).rejects.toMatchObject({ code: "I2V_MEDIA_NOT_FOUND" });
  });
  it.each(["task", "hash", "stale", "future", "malformed", "empty_job", "version", "extra_field", "conflicting_job"] as const)("quarantines a %s receipt without polling it or submitting again", async (kind) => {
    const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
    await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
    const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
    record.state = "submitting"; record.revision++;
    delete record.lastPollAt;
    if (kind === "conflicting_job") record.jobId = "different_record_job";
    const receipt: Record<string, unknown> = { version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "untrusted_receipt_job" };
    if (kind === "task") receipt.id = "i2v_11111111-1111-1111-1111-111111111111";
    if (kind === "hash") receipt.requestHash = "a".repeat(64);
    if (kind === "stale") receipt.revision = record.revision - 1;
    if (kind === "future") receipt.revision = record.revision + 1;
    if (kind === "empty_job") receipt.jobId = "";
    if (kind === "version") receipt.version = 2;
    if (kind === "extra_field") receipt.unexpected = "conflicting format";
    const serialized = kind === "malformed" ? "{broken-json" : JSON.stringify(receipt);
    await writeFile(file, JSON.stringify(record));
    const receiptPath = join(f.options.root, task.id, "submission-receipt.json"); await writeFile(receiptPath, serialized);
    f.calls.length = 0;
    const restored = new harness.ImageVideoService(f.options);
    expect((await restored.view(task.id)).state).toBe("unknown_submission");
    expect((await restored.refresh(task.id)).state).toBe("unknown_submission");
    await expect(restored.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" });
    expect(f.calls).toEqual([]);
    expect(await readFile(receiptPath, "utf8")).toBe(serialized);
  });
});