import "./stt-test-env";
import assert from "node:assert/strict";
import { execFileSync } from "node:child_process";
import { readFileSync } from "node:fs";
import { after, before, beforeEach, describe, test } from "node:test";
import type { NextRequest } from "next/server";
import { GET, POST } from "../app/api/stt/route";
import { prisma } from "../lib/prisma";
import { AsrError } from "../lib/stt/dashscope";
import {
  claimTask,
  completeTask,
  enqueueMissingTranscriptions,
  enqueueTranscription,
  readTranscription,
} from "../lib/stt/tasks";
import { processTask } from "../lib/stt/worker";

const database = process.env.TEST_DATABASE_URL;
const migration = readFileSync(
  "prisma/migrations/20261003000000_plain_transcription/migration.sql",
  "utf8",
);
const sql = (source: string) =>
  execFileSync("psql", [database ?? "", "-X", "-v", "ON_ERROR_STOP=1", "-q"], {
    input: source,
    stdio: ["pipe", "pipe", "pipe"],
  });
const fakeServices = {
  prepareAudio: async () => "https://storage.test/audio.mp3" as string | null,
  submitTranscription: async () => "provider-1",
  queryTranscription: async () => ({
    status: "SUCCEEDED" as const,
    resultURL: "https://result.test/file",
  }),
  downloadTranscript: async () => ["new searchable words"],
};
async function due(videoId: string) {
  await prisma.transcriptionTask.update({
    where: { videoId },
    data: { nextPollAt: new Date(0) },
  });
}
async function claim() {
  const task = await claimTask();
  assert.ok(task);
  return task;
}
function request(url: string, body?: unknown): NextRequest {
  return Object.assign(
    new Request(
      url,
      body === undefined
        ? {}
        : {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify(body),
          },
    ),
    { nextUrl: new URL(url) },
  ) as NextRequest;
}
describe("PostgreSQL transcription lifecycle", { skip: !database }, () => {
  before(async () => {
    sql("DROP SCHEMA public CASCADE; CREATE SCHEMA public;");
    sql(
      readFileSync("prisma/migrations/20251025_baseline/migration.sql", "utf8"),
    );
    await prisma.author.create({
      data: {
        sec_uid: "test-author",
        uid: "uid",
        nickname: "测试作者",
        signature: "",
        avatar_url: "",
        follower_count: 0,
        total_favorited: 0,
        short_id: "test",
      },
    });
    for (const id of ["historical", "empty-history", "v1", "v2", "v3", "v4"]) {
      await prisma.video.create({
        data: {
          aweme_id: id,
          desc: "测试视频",
          duration_ms: 1000,
          created_at: new Date(),
          share_url: "",
          digg_count: 0,
          comment_count: 0,
          share_count: 0,
          collect_count: 0,
          authorId: "test-author",
          tags: [],
          video_url: "video.mp4",
        },
      });
    }
    sql(
      `INSERT INTO "VideoTranscript" (id,"videoId",speech_detected,transcript,"updatedAt") VALUES ('old','historical',true,ARRAY['original searchable words'],NOW()), ('empty','empty-history',false,ARRAY[]::text[],NOW());`,
    );
    sql(`ALTER TABLE "VideoTranscript" ADD COLUMN content_tags TEXT[] DEFAULT ARRAY[]::TEXT[];
      CREATE TEXT SEARCH CONFIGURATION public.zhcfg (COPY = pg_catalog.simple);
      CREATE FUNCTION public.vt_update_transcript_zh_tsv() RETURNS trigger AS $$ BEGIN RETURN NEW; END; $$ LANGUAGE plpgsql;
      CREATE TRIGGER trg_vt_transcript_zh_tsv BEFORE INSERT OR UPDATE OF transcript ON "VideoTranscript" FOR EACH ROW EXECUTE FUNCTION public.vt_update_transcript_zh_tsv();`);
    sql(migration);
    sql(migration); // migration DDL is repeatable, historical text is unchanged
  });
  beforeEach(async () => {
    process.env.DASHSCOPE_API_KEY = "test-key";
    delete process.env.DASHSCOPE_BASE_URL;
    delete process.env.DASHSCOPE_ASR_MODEL;
    await prisma.transcriptionTask.deleteMany();
    await prisma.videoTranscript.deleteMany({
      where: { videoId: { in: ["v1", "v2", "v3", "v4"] } },
    });
  });
  after(async () => {
    await prisma.$disconnect();
  });
  test("migration preserves historical and empty text and rebuilds search", async () => {
    assert.equal(
      (await readTranscription("empty-history"))?.status,
      "SUCCEEDED",
    );
    assert.deepEqual((await readTranscription("historical"))?.transcript, [
      "original searchable words",
    ]);
    const rows = await prisma.$queryRaw<
      { count: bigint }[]
    >`SELECT COUNT(*) FROM "VideoTranscript" WHERE transcript_zh_tsv @@ websearch_to_tsquery('zhcfg', 'original')`;
    assert.equal(Number(rows[0].count), 1);
  });
  test("concurrent enqueue is idempotent; automatic enqueue never retranscribes history", async () => {
    await Promise.all(
      Array.from({ length: 10 }, () => enqueueTranscription("v1", true)),
    );
    assert.equal(await prisma.transcriptionTask.count(), 1);
    assert.equal(
      (await enqueueTranscription("historical"))?.status,
      "SUCCEEDED",
    );
    assert.equal(await prisma.transcriptionTask.count(), 1);
  });
  test("batch skips history and active tasks and retries failed missing transcripts", async () => {
    await enqueueTranscription("v1");
    await prisma.transcriptionTask.update({
      where: { videoId: "v1" },
      data: { status: "FAILED" },
    });
    await enqueueTranscription("v2");
    assert.equal(await enqueueMissingTranscriptions(), 3);
    assert.equal(await enqueueMissingTranscriptions(), 0);
    assert.equal(await prisma.transcriptionTask.count(), 4);
  });
  test("admission never exceeds two active provider jobs across workers", async () => {
    await enqueueMissingTranscriptions();
    await prisma.transcriptionTask.updateMany({
      data: { nextPollAt: new Date(0) },
    });
    const claims = await Promise.all([
      claimTask(),
      claimTask(),
      claimTask(),
      claimTask(),
    ]);
    assert.equal(claims.filter(Boolean).length, 2);
    assert.equal(
      await prisma.transcriptionTask.count({ where: { status: "PROCESSING" } }),
      2,
    );
  });
  test("provider ID survives restart; completion saves text and search atomically", async () => {
    await enqueueTranscription("v1");
    let submits = 0;
    const services = {
      ...fakeServices,
      submitTranscription: async () => {
        submits++;
        return "restart-task";
      },
    };
    await processTask(await claim(), services);
    assert.equal(
      (
        await prisma.transcriptionTask.findUniqueOrThrow({
          where: { videoId: "v1" },
        })
      ).providerTaskId,
      "restart-task",
    );
    await due("v1");
    await processTask(await claim(), services);
    assert.equal(submits, 1);
    assert.deepEqual(await readTranscription("v1"), {
      status: "SUCCEEDED",
      error: null,
      transcript: ["new searchable words"],
    });
    const hits = await prisma.$queryRaw<
      { count: bigint }[]
    >`SELECT COUNT(*) FROM "VideoTranscript" WHERE "videoId" = 'v1' AND transcript_zh_tsv @@ websearch_to_tsquery('zhcfg', 'new')`;
    assert.equal(Number(hits[0].count), 1);
  });
  test("failed retranscription retains old text; success replaces text and clears stale search", async () => {
    await prisma.videoTranscript.create({
      data: { videoId: "v1", transcript: ["oldword"] },
    });
    await enqueueTranscription("v1", true);
    await processTask(await claim(), fakeServices);
    await due("v1");
    await processTask(await claim(), {
      ...fakeServices,
      queryTranscription: async () => {
        throw new AsrError("FILE_403_FORBIDDEN");
      },
    });
    assert.equal((await readTranscription("v1"))?.status, "FAILED");
    assert.deepEqual((await readTranscription("v1"))?.transcript, ["oldword"]);
    await enqueueTranscription("v1", true);
    await processTask(await claim(), fakeServices);
    await due("v1");
    await processTask(await claim(), {
      ...fakeServices,
      downloadTranscript: async () => [],
    });
    assert.deepEqual((await readTranscription("v1"))?.transcript, []);
    const hits = await prisma.$queryRaw<
      { count: bigint }[]
    >`SELECT COUNT(*) FROM "VideoTranscript" WHERE "videoId" = 'v1' AND transcript_zh_tsv @@ websearch_to_tsquery('zhcfg', 'oldword')`;
    assert.equal(Number(hits[0].count), 0);
  });
  test("crashed/ambiguous submission is never automatically resubmitted", async () => {
    await enqueueTranscription("v1");
    const task = await claim();
    await prisma.transcriptionTask.update({
      where: { videoId: "v1" },
      data: { submittingAt: new Date(), leaseUntil: new Date(0) },
    });
    let submits = 0;
    await processTask(await claim(), {
      ...fakeServices,
      submitTranscription: async () => {
        submits++;
        return "bad";
      },
    });
    assert.equal(submits, 0);
    assert.equal((await readTranscription("v1"))?.status, "FAILED");
    await assert.rejects(completeTask(task, ["stale result"]), /租约/);
    assert.equal((await readTranscription("v1"))?.transcript, null);
  });
  test("temporary query errors back off and keep provider ID", async () => {
    await enqueueTranscription("v1");
    await processTask(await claim(), fakeServices);
    await due("v1");
    await processTask(await claim(), {
      ...fakeServices,
      queryTranscription: async () => {
        throw new AsrError("网络超时", true);
      },
    });
    const task = await prisma.transcriptionTask.findUniqueOrThrow({
      where: { videoId: "v1" },
    });
    assert.equal(task.status, "PROCESSING");
    assert.equal(task.providerTaskId, "provider-1");
    assert.ok(task.nextPollAt.getTime() > Date.now());
    assert.equal(task.attempts, 1);
  });
  test("video with no audio track completes with no ASR submission", async () => {
    await enqueueTranscription("v1");
    await processTask(await claim(), {
      ...fakeServices,
      prepareAudio: async () => null,
      submitTranscription: async () => {
        throw new Error("must not submit");
      },
    });
    assert.deepEqual(await readTranscription("v1"), {
      status: "SUCCEEDED",
      error: null,
      transcript: [],
    });
  });
  test("missing key is visible; GET has no side effect; POST validates input", async () => {
    delete process.env.DASHSCOPE_API_KEY;
    const state = await enqueueTranscription("v1");
    assert.equal(state?.status, "FAILED");
    assert.match(state?.error ?? "", /DASHSCOPE_API_KEY/);
    assert.equal(
      (await GET(request("http://local/api/stt?awemeId=v2"))).status,
      200,
    );
    assert.equal(await prisma.transcriptionTask.count(), 1);
    assert.equal(
      (await GET(request("http://local/api/stt?awemeId=missing"))).status,
      404,
    );
    assert.equal((await POST(request("http://local/api/stt", {}))).status, 400);
    process.env.DASHSCOPE_API_KEY = "test-key";
    assert.equal(
      (await POST(request("http://local/api/stt", { awemeId: "v2" }))).status,
      202,
    );
  });
});