import { mkdtempSync, rmSync } from "node:fs";
import { writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import type { MediaOperationRegistration } from "@rakazo/adapters";
import { ComposioEmulator } from "@rakazo/adapters";
import type { MediaJobBlock, MediaSettingsView, MessageBlock } from "@rakazo/contracts";
import { TranscribeInputSchema } from "@rakazo/contracts";
import { createThreadMessage, ensureMainConversation, MAIN_BOT_SPAWN_KEY } from "@rakazo/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import type { createApp } from "../../../apps/api/src/app.ts";
import { sessionCookieHeader } from "./index.js";

type App = { request: (input: string, init?: RequestInit) => Response | Promise<Response> };
type AppHandles = Awaited<ReturnType<typeof createApp>>;

process.env.WAKEUP_DRIVER = "memory";
process.env.SANDBOX_PROVIDER = "fake";
process.env.AGENT_RUNTIME = "scripted";

const hasDb = process.env.VERIFY_DATABASE === "1" && Boolean(process.env.DATABASE_URL);
const describeWithDatabase = hasDb ? describe.sequential : describe.skip;

const PNG_BASE64 = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1, 2, 3]).toString(
  "base64",
);

// testkit has no zod of its own; the contracts' schemas carry it.
const SourceInputSchema = TranscribeInputSchema.pick({ artifact_id: true });

/** Fake card operations: an export that copies its source, and a transcription that waits. */
function fakeOperations(): MediaOperationRegistration[] {
  const exportVideo: MediaOperationRegistration<{ artifact_id: string }> = {
    operation: "export_video",
    inputSchema: SourceInputSchema,
    title: () => "Export",
    available: async () => true,
    artifactIds: (input) => [input.artifact_id],
    handler: {
      operation: "export_video",
      run: async (context) => {
        const { artifact_id } = context.input as { artifact_id: string };
        const source = await context.fetchArtifact(artifact_id);
        const out = path.join(context.tempDir, "export.png");
        await writeFile(out, Buffer.from(PNG_BASE64, "base64"));
        await context.addOutput({
          path: out,
          name: "export.png",
          mimeType: "image/png",
          parentArtifactId: source.artifactId,
          preset: "square",
        });
        return { summary: "Exported." };
      },
    },
  };
  const transcribe: MediaOperationRegistration<{ artifact_id: string }> = {
    operation: "transcribe",
    inputSchema: SourceInputSchema,
    title: () => "Transcript",
    available: async () => true,
    artifactIds: (input) => [input.artifact_id],
    handler: {
      operation: "transcribe",
      run: async (context) => {
        await new Promise<void>((resolve) => {
          context.signal.addEventListener("abort", () => resolve(), { once: true });
        });
        context.signal.throwIfAborted();
        return { summary: "" };
      },
    },
  };
  return [exportVideo, transcribe] as MediaOperationRegistration[];
}

describeWithDatabase("media RPCs (PostgreSQL)", () => {
  let handles: AppHandles;
  let app: App;
  const stamp = `${Date.now()}-${Math.random().toString(36).slice(2)}`;
  const dataDir = mkdtempSync(path.join(tmpdir(), "rakazo-media-"));

  beforeAll(async () => {
    const { createApp } = await import("../../../apps/api/src/app.ts");
    handles = await createApp({
      databaseUrl: process.env.DATABASE_URL!,
      dataDir,
      sandboxProvider: "fake",
      agentRuntime: "scripted",
      wakeupDriver: "memory",
      signupsEnabled: "true",
      composio: new ComposioEmulator(),
      mediaOperations: fakeOperations(),
    });
    app = handles.app;
  });

  afterAll(async () => {
    await handles?.prisma.integrationProviderConfig.deleteMany({ where: { id: "media" } });
    await handles?.stop();
    rmSync(dataDir, { recursive: true, force: true });
  });

  async function user(label: string) {
    const cookie = await signup(app, `${label}-${stamp}@rakazo.test`, label);
    const me = await rpc<{ userId: string; spaceId: string }>(app, cookie, "me");
    const bot = await rpc<{ id: string }>(app, cookie, "bots/create", {
      name: `${label} bot`,
      title: "",
      description: "",
      instructions: "",
      notifyOnFinish: false,
    });
    const thread = await handles.prisma.thread.findUniqueOrThrow({ where: { botId: bot.id } });
    const card = await createThreadMessage(handles.prisma, {
      threadId: thread.id,
      role: "bot",
      botId: bot.id,
      blocks: [{ kind: "text", text: "Here is your clip." }],
    });
    const artifact = await rpc<{ id: string }>(app, cookie, "artifacts/create", {
      botId: bot.id,
      name: "clip.png",
      mimeType: "image/png",
      contentBase64: PNG_BASE64,
    });
    return { cookie, me, bot, thread, card, artifact };
  }

  async function waitForStatus(jobId: string, status: string) {
    const start = Date.now();
    while (Date.now() - start < 20_000) {
      const job = await handles.prisma.mediaJob.findUnique({ where: { id: jobId } });
      if (job?.status === status) return job;
      await new Promise((resolve) => setTimeout(resolve, 50));
    }
    throw new Error(`media job ${jobId} never reached ${status}`);
  }

  it("lets only the deployment owner read and change media settings, and never returns keys", async () => {
    const owner = await user("media-owner");
    const other = await user("media-other");
    await handles.prisma.deploymentSettings.upsert({
      where: { id: "default" },
      create: { id: "default", ownerUserId: owner.me.userId },
      update: { ownerUserId: owner.me.userId },
    });

    const update = {
      comfyui: {
        baseUrl: "http://127.0.0.1:8188/",
        apiKey: "comfy-test-key",
        imageModel: "sd_xl_base_1.0.safetensors",
        imageWorkflow: '{"6":{"class_type":"CLIPTextEncode","inputs":{"text":"{{prompt}}"}}}',
        videoWorkflow: "",
      },
      transcription: { baseUrl: "http://127.0.0.1:8080", model: "whisper-1", apiKey: "" },
    };
    expect((await raw(app, other.cookie, "media/settings/get", {})).status).toBe(403);
    expect((await raw(app, other.cookie, "media/settings/update", update)).status).toBe(403);
    expect(await handles.prisma.integrationProviderConfig.count({ where: { id: "media" } })).toBe(
      0,
    );

    const saved = await rpc<MediaSettingsView>(app, owner.cookie, "media/settings/update", update);
    expect(saved).toEqual({
      comfyui: {
        baseUrl: "http://127.0.0.1:8188",
        hasApiKey: true,
        imageModel: "sd_xl_base_1.0.safetensors",
        imageWorkflow: update.comfyui.imageWorkflow,
      },
      transcription: { baseUrl: "http://127.0.0.1:8080/v1", model: "whisper-1", hasApiKey: false },
    });
    expect(JSON.stringify(saved)).not.toContain("comfy-test-key");
    expect(await rpc(app, owner.cookie, "media/settings/get")).toEqual(saved);
    const row = await handles.prisma.integrationProviderConfig.findUniqueOrThrow({
      where: { id: "media" },
    });
    expect(row.ciphertext).not.toContain("comfy-test-key");
    expect(row.ciphertext).not.toContain("127.0.0.1");

    // An omitted key stays with the same server, never follows the settings elsewhere.
    const { apiKey: _key, ...withoutKey } = update.comfyui;
    const kept = await rpc<MediaSettingsView>(app, owner.cookie, "media/settings/update", {
      ...update,
      comfyui: withoutKey,
    });
    expect(kept.comfyui?.hasApiKey).toBe(true);
    const moved = await rpc<MediaSettingsView>(app, owner.cookie, "media/settings/update", {
      ...update,
      comfyui: { ...withoutKey, baseUrl: "http://127.0.0.1:8189" },
    });
    expect(moved.comfyui?.hasApiKey).toBe(false);

    for (const baseUrl of ["http://169.254.169.254", "http://user:pass@127.0.0.1:8188"]) {
      const refused = await raw(app, owner.cookie, "media/settings/update", {
        ...update,
        comfyui: { ...update.comfyui, baseUrl },
      });
      expect(refused.status, baseUrl).toBe(400);
    }
    const badWorkflow = await raw(app, owner.cookie, "media/settings/update", {
      ...update,
      comfyui: { ...update.comfyui, imageWorkflow: "[1, 2" },
    });
    expect(badWorkflow.status).toBe(400);

    const previous = process.env.RAKAZO_OPENAI_COMPAT_ALLOW_PUBLIC;
    try {
      delete process.env.RAKAZO_OPENAI_COMPAT_ALLOW_PUBLIC;
      expect(
        (
          await raw(app, owner.cookie, "media/settings/update", {
            comfyui: null,
            transcription: { baseUrl: "http://transcribe.example.com" },
          })
        ).status,
      ).toBe(400);
      process.env.RAKAZO_OPENAI_COMPAT_ALLOW_PUBLIC = "1";
      // A key is never sent in cleartext to a public host.
      expect(
        (
          await raw(app, owner.cookie, "media/settings/update", {
            comfyui: null,
            transcription: { baseUrl: "http://transcribe.example.com", apiKey: "secret" },
          })
        ).status,
      ).toBe(400);
      const https = await rpc<MediaSettingsView>(app, owner.cookie, "media/settings/update", {
        comfyui: null,
        transcription: { baseUrl: "https://transcribe.example.com", apiKey: "secret" },
      });
      expect(https.transcription).toMatchObject({
        baseUrl: "https://transcribe.example.com/v1",
        hasApiKey: true,
      });
    } finally {
      if (previous === undefined) delete process.env.RAKAZO_OPENAI_COMPAT_ALLOW_PUBLIC;
      else process.env.RAKAZO_OPENAI_COMPAT_ALLOW_PUBLIC = previous;
    }

    expect(
      await rpc(app, owner.cookie, "media/settings/update", { comfyui: null, transcription: null }),
    ).toEqual({ comfyui: null, transcription: null });
    expect(await handles.prisma.integrationProviderConfig.count({ where: { id: "media" } })).toBe(
      0,
    );
  });

  it("starts, reads and cancels card jobs only for the actor's own threads and files", async () => {
    const alice = await user("media-alice");
    const mallory = await user("media-mallory");

    const started = await rpc<{ jobId: string; messageId: string }>(
      app,
      alice.cookie,
      "media/jobs/start",
      {
        botId: alice.bot.id,
        messageId: alice.card.id,
        operation: "export_video",
        input: { artifact_id: alice.artifact.id },
      },
    );
    const job = await waitForStatus(started.jobId, "done");
    expect(job).toMatchObject({
      spaceId: alice.me.spaceId,
      userId: alice.me.userId,
      botId: alice.bot.id,
      threadId: alice.thread.id,
      runId: null,
      continueAfter: false,
    });
    const snapshot = await rpc<{ status: string; outputs: MediaJobBlock["outputs"] }>(
      app,
      alice.cookie,
      "media/jobs/get",
      { jobId: started.jobId },
    );
    expect(snapshot.status).toBe("done");
    expect(snapshot.outputs).toEqual([
      expect.objectContaining({
        name: "export.png",
        parentArtifactId: alice.artifact.id,
        preset: "square",
      }),
    ]);
    const card = await handles.prisma.message.findUniqueOrThrow({
      where: { id: started.messageId },
    });
    expect(card).toMatchObject({ threadId: alice.thread.id, botId: alice.bot.id, role: "bot" });
    expect((card.blocks as MessageBlock[])[0]).toMatchObject({ kind: "media", status: "done" });

    // Another user can neither read nor cancel the job, nor start one in Alice's conversation
    // or on her files.
    expect(
      (await raw(app, mallory.cookie, "media/jobs/get", { jobId: started.jobId })).status,
    ).toBe(404);
    expect(
      (await raw(app, mallory.cookie, "media/jobs/cancel", { jobId: started.jobId })).status,
    ).toBe(404);
    for (const body of [
      {
        botId: alice.bot.id,
        messageId: alice.card.id,
        operation: "export_video",
        input: { artifact_id: mallory.artifact.id },
      },
      {
        botId: mallory.bot.id,
        messageId: alice.card.id,
        operation: "export_video",
        input: { artifact_id: mallory.artifact.id },
      },
      {
        botId: mallory.bot.id,
        messageId: mallory.card.id,
        operation: "export_video",
        input: { artifact_id: alice.artifact.id },
      },
      {
        botId: mallory.bot.id,
        messageId: mallory.card.id,
        operation: "generate_image",
        input: { prompt: "anything" },
      },
    ]) {
      const response = await raw(app, mallory.cookie, "media/jobs/start", body);
      expect(response.status, JSON.stringify(body)).toBeGreaterThanOrEqual(400);
    }
    expect(await handles.prisma.mediaJob.count({ where: { userId: mallory.me.userId } })).toBe(0);

    const slow = await rpc<{ jobId: string }>(app, alice.cookie, "media/jobs/start", {
      botId: alice.bot.id,
      messageId: alice.card.id,
      operation: "transcribe",
      input: { artifact_id: alice.artifact.id },
    });
    await waitForStatus(slow.jobId, "running");
    const cancelled = await rpc<{ status: string }>(app, alice.cookie, "media/jobs/cancel", {
      jobId: slow.jobId,
    });
    expect(["running", "cancelled"]).toContain(cancelled.status);
    await waitForStatus(slow.jobId, "cancelled");
  });

  it("acts on a card where it lives, whichever of the user's spaces is selected", async () => {
    const cookie = await signup(app, `media-main-${stamp}@rakazo.test`, "media-main");
    const me = await rpc<{ userId: string; spaceId: string }>(app, cookie, "me");
    await rpc(app, cookie, "bots/create", {
      name: "Chief",
      title: "",
      description: "",
      instructions: "",
      notifyOnFinish: false,
      spawnKey: MAIN_BOT_SPAWN_KEY,
    });
    const main = await ensureMainConversation(handles.prisma, me.userId);
    expect(main.spaceId).toBe(me.spaceId);
    const card = await createThreadMessage(handles.prisma, {
      threadId: main.threadId,
      role: "bot",
      botId: main.leadBotId,
      blocks: [{ kind: "text", text: "Here is your clip." }],
    });
    const clip = await rpc<{ id: string }>(app, cookie, "artifacts/create", {
      botId: main.leadBotId,
      name: "clip.png",
      mimeType: "image/png",
      contentBase64: PNG_BASE64,
    });
    // The user has moved on to another space; the main conversation stays in the home space.
    const elsewhere = await rpc<{ id: string }>(app, cookie, "spaces/create", {
      name: "Elsewhere",
    });

    const started = await rpc<{ jobId: string; messageId: string }>(
      app,
      cookie,
      "media/jobs/start",
      {
        groupId: main.groupId,
        messageId: card.id,
        operation: "export_video",
        input: { artifact_id: clip.id },
      },
      elsewhere.id,
    );
    expect(await waitForStatus(started.jobId, "done")).toMatchObject({
      spaceId: main.spaceId,
      threadId: main.threadId,
      groupId: main.groupId,
      botId: main.leadBotId,
    });
    expect(
      await rpc<{ status: string }>(
        app,
        cookie,
        "media/jobs/get",
        { jobId: started.jobId },
        elsewhere.id,
      ),
    ).toMatchObject({ status: "done" });
    expect(
      await rpc(app, cookie, "media/capabilities", { jobId: started.jobId }, elsewhere.id),
    ).toMatchObject({ export_video: true, transcribe: true, generate_image: false });
    const slow = await rpc<{ jobId: string }>(
      app,
      cookie,
      "media/jobs/start",
      {
        groupId: main.groupId,
        messageId: card.id,
        operation: "transcribe",
        input: { artifact_id: clip.id },
      },
      elsewhere.id,
    );
    await waitForStatus(slow.jobId, "running");
    await rpc(app, cookie, "media/jobs/cancel", { jobId: slow.jobId }, elsewhere.id);
    await waitForStatus(slow.jobId, "cancelled");

    // Another user gets nothing, from any space of theirs.
    const mallory = await user("media-main-mallory");
    for (const [proc, body] of [
      ["media/jobs/get", { jobId: started.jobId }],
      ["media/jobs/cancel", { jobId: started.jobId }],
      ["media/capabilities", { jobId: started.jobId }],
      [
        "media/jobs/start",
        {
          groupId: main.groupId,
          messageId: card.id,
          operation: "export_video",
          input: { artifact_id: clip.id },
        },
      ],
      [
        "media/jobs/start",
        {
          botId: mallory.bot.id,
          messageId: card.id,
          operation: "export_video",
          input: { artifact_id: mallory.artifact.id },
        },
      ],
    ] as const) {
      expect((await raw(app, mallory.cookie, proc, body)).status, proc).toBe(404);
    }

    // A card in a space the user has left reads as not found too.
    const bot = await rpc<{ id: string }>(
      app,
      cookie,
      "bots/create",
      { name: "Away", title: "", description: "", instructions: "", notifyOnFinish: false },
      elsewhere.id,
    );
    const awayThread = await handles.prisma.thread.findUniqueOrThrow({ where: { botId: bot.id } });
    const awayCard = await createThreadMessage(handles.prisma, {
      threadId: awayThread.id,
      role: "bot",
      botId: bot.id,
      blocks: [{ kind: "text", text: "Here is another clip." }],
    });
    const awayClip = await rpc<{ id: string }>(
      app,
      cookie,
      "artifacts/create",
      { botId: bot.id, name: "away.png", mimeType: "image/png", contentBase64: PNG_BASE64 },
      elsewhere.id,
    );
    const away = await rpc<{ jobId: string }>(app, cookie, "media/jobs/start", {
      botId: bot.id,
      messageId: awayCard.id,
      operation: "export_video",
      input: { artifact_id: awayClip.id },
    });
    await waitForStatus(away.jobId, "done");
    await handles.prisma.spaceMember.deleteMany({
      where: { spaceId: elsewhere.id, userId: me.userId },
    });
    for (const [proc, body] of [
      ["media/jobs/get", { jobId: away.jobId }],
      ["media/jobs/cancel", { jobId: away.jobId }],
      ["media/capabilities", { jobId: away.jobId }],
      [
        "media/jobs/start",
        {
          botId: bot.id,
          messageId: awayCard.id,
          operation: "export_video",
          input: { artifact_id: awayClip.id },
        },
      ],
    ] as const) {
      expect((await raw(app, cookie, proc, body)).status, proc).toBe(404);
    }
  });
});

async function signup(app: App, email: string, name: string) {
  const res = await app.request("/api/auth/sign-up/email", {
    method: "POST",
    headers: { "content-type": "application/json", origin: "http://127.0.0.1:5173" },
    body: JSON.stringify({ email, password: "password12", name }),
  });
  if (res.status >= 400) throw new Error(`signup failed ${res.status}: ${await res.text()}`);
  return sessionCookieHeader(res);
}

function raw(app: App, cookie: string, proc: string, body: unknown = {}, spaceId?: string) {
  return app.request(`/rpc/${proc}`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      cookie,
      origin: "http://127.0.0.1:5173",
      ...(spaceId ? { "x-rakazo-space-id": spaceId } : {}),
    },
    body: JSON.stringify({ json: body }),
  });
}

async function rpc<T>(
  app: App,
  cookie: string,
  proc: string,
  body: unknown = {},
  spaceId?: string,
): Promise<T> {
  const res = await raw(app, cookie, proc, body, spaceId);
  const text = await res.text();
  const parsed = JSON.parse(text) as { json?: T; error?: { message?: string } };
  if (res.status >= 400 || parsed.error) {
    throw new Error(`${proc} ${res.status}: ${parsed.error?.message ?? text}`);
  }
  return parsed.json as T;
}