import http from "node:http";
import { createReadStream, createWriteStream } from "node:fs";
import {
  mkdir,
  readFile,
  rename,
  readdir,
  rm,
  stat,
  statfs,
  open,
} from "node:fs/promises";
import path from "node:path";
import { createHash, timingSafeEqual } from "node:crypto";
import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import { z } from "zod";
import {
  audioJobInputSchema,
  type AudioJobInput,
} from "../../lib/video-canvas/smart-edit/audio-contract";
import { audioJob, audioHealth } from "./job";

const root = path.resolve(
  process.env.AUDIO_DATA_DIR || "/var/lib/audio-processing"
);
const token = process.env.AUDIO_API_TOKEN || "";
if (token.length < 32)
  throw Error("AUDIO_API_TOKEN must have at least 32 characters");
const jobs = new Map<string, Job>();
type Job = {
  id: string;
  digest: string;
  tenant: string;
  plan: AudioJobInput;
  status: "awaiting-input" | "queued" | "running" | "succeeded" | "failed";
  stage: string;
  createdAt: string;
  updatedAt: string;
  attempts: number;
  error?: string;
  result?: Record<string, any>;
};
const inputSchema = z.object({
  id: z.string().regex(/^[a-f0-9]{64}$/),
  tenant: z.string().regex(/^[a-f0-9]{64}$/),
  plan: audioJobInputSchema,
});
let busy = false,
  stopping = false,
  accepting = false;
const uploading = new Set<string>();
let capabilities = { asr: false, separation: false };
const file = (id: string) => path.join(root, id + ".json");
async function persist(job: Job) {
  job.updatedAt = new Date().toISOString();
  const temp = file(job.id) + ".tmp";
  const handle = await open(temp, "w", 0o600);
  try {
    await handle.writeFile(JSON.stringify(job));
    await handle.sync();
  } finally {
    await handle.close();
  }
  await rename(temp, file(job.id));
  const directory = await open(root, "r");
  try {
    await directory.sync();
  } finally {
    await directory.close();
  }
  jobs.set(job.id, job);
}
function publicJob(job: Job) {
  return {
    id: job.id,
    status: job.status,
    stage: job.stage,
    attempts: job.attempts,
    createdAt: job.createdAt,
    updatedAt: job.updatedAt,
    error: job.error,
    result: job.result,
  };
}
async function prune() {
  for (const job of jobs.values())
    if (
      ["awaiting-input", "succeeded", "failed"].includes(job.status) &&
      !uploading.has(job.id) &&
      Date.now() - Date.parse(job.updatedAt) > 7 * 86400000
    ) {
      await rm(path.join(root, job.id), { recursive: true, force: true });
      await rm(file(job.id), { force: true });
      jobs.delete(job.id);
    }
}
async function work() {
  if (busy || stopping) return;
  const job = [...jobs.values()].find((j) => j.status === "queued");
  if (!job) return;
  busy = true;
  const dir = path.join(root, job.id);
  try {
    await prune();
    const disk = await statfs(root);
    if (disk.bavail * disk.bsize < 10 * 1024 ** 3)
      throw Error("音频磁盘剩余空间不足 10 GiB");
    job.status = "running";
    job.attempts++;
    job.stage = "准备素材";
    await persist(job);
    await mkdir(dir, { recursive: true, mode: 0o700 });
    const result = await audioJob(job.plan, job.tenant, dir, async (stage) => {
      job.stage = stage;
      await persist(job);
    });
    job.result = result;
    job.status = "succeeded";
    job.stage = "音频完成";
    await persist(job);
    // Keep only the deliverables. Intermediate files must not accumulate per job.
    for (const name of await readdir(dir))
      if (!["vocals.wav", "background.wav"].includes(name))
        await rm(path.join(dir, name), { recursive: true, force: true }).catch(
          () => {
            console.warn(
              JSON.stringify({ event: "audio-cleanup-deferred", id: job.id })
            );
          }
        );
  } catch (error) {
    if (stopping) return; // Preserve running journal; restart will resume this same job.
    job.status = "failed";
    job.stage = "音频失败";
    // Do not expose ffmpeg arguments, private cache paths or signed source URLs.
    job.error =
      error instanceof Error &&
      !/Command failed|https?:|\/var\/|\/tmp\//.test(error.message)
        ? error.message.slice(0, 300)
        : "素材处理失败，请检查格式、参数或服务日志";
    await rm(dir, { recursive: true, force: true }).catch(() => {});
    await persist(job);
    console.error(
      JSON.stringify({ event: "audio-failed", id: job.id, error: job.error })
    );
  } finally {
    busy = false;
  }
  setImmediate(
    () =>
      void work().catch((e) => {
        console.error("Audio queue persistence failure");
        process.exit(1);
      })
  );
}
function json(res: http.ServerResponse, status: number, value: unknown) {
  res.writeHead(status, {
    "content-type": "application/json; charset=utf-8",
    "cache-control": "no-store",
  });
  res.end(JSON.stringify(value));
}
async function body(req: http.IncomingMessage) {
  let size = 0;
  const chunks: Buffer[] = [];
  for await (const chunk of req) {
    size += chunk.length;
    if (size > 1024 * 1024) throw Error("请求超过 1 MiB");
    chunks.push(chunk);
  }
  return JSON.parse(Buffer.concat(chunks).toString("utf8"));
}
function authenticated(req: http.IncomingMessage) {
  const expected = Buffer.from("Bearer " + token),
    actual = Buffer.from(req.headers.authorization || "");
  return expected.length === actual.length && timingSafeEqual(expected, actual);
}
async function main() {
  capabilities = await audioHealth();
  await mkdir(root, { recursive: true, mode: 0o700 });
  // This service runs as a single systemd process. A restart requeues local work at most twice.
  for (const name of await readdir(root))
    if (/^[a-f0-9]{64}\.json$/.test(name)) {
      const job: Job = JSON.parse(
        await readFile(path.join(root, name), "utf8")
      );
      if (job.id + ".json" !== name) throw Error("Invalid audio record");
      if (job.status === "running") {
        job.status = job.attempts < 3 ? "queued" : "failed";
        job.stage =
          job.status === "queued"
            ? "服务重启，等待恢复音频"
            : "多次中断，已停止";
        if (job.status === "failed") job.error = job.stage;
        await persist(job);
      }
      jobs.set(job.id, job);
    }
  await prune();
  const server = http.createServer(async (req, res) => {
    if (!authenticated(req)) return json(res, 401, { error: "Unauthorized" });
    const url = new URL(req.url || "/", "http://localhost");
    try {
      if (req.method === "GET" && url.pathname === "/health")
        return json(res, 200, {
          status: "ok",
          version: 1,
          active: busy ? 1 : 0,
          queued: [...jobs.values()].filter((j) => j.status === "queued")
            .length,
          capabilities,
          engine: "FunASR-Paraformer / HTDemucs",
          device: "cpu",
        });
      if (req.method === "POST" && url.pathname === "/v1/audio-jobs") {
        const input = inputSchema.parse(await body(req));
        // Serialize admission across the async filesystem write to preserve idempotency.
        if (accepting) return json(res, 503, { error: "请重试提交" });
        accepting = true;
        try {
          const digest = createHash("sha256")
            .update(JSON.stringify({ tenant: input.tenant, plan: input.plan }))
            .digest("hex");
          const existing = jobs.get(input.id);
          if (existing)
            return json(
              res,
              existing.digest === digest ? 200 : 409,
              existing.digest === digest
                ? publicJob(existing)
                : { error: "任务 ID 已用于不同参数" }
            );
          if (
            jobs.size >= 200 ||
            [...jobs.values()].filter((j) =>
              ["awaiting-input", "queued", "running"].includes(j.status)
            ).length >= 30
          )
            return json(res, 429, { error: "音频队列已满，请稍后重试" });
          const job: Job = {
            ...input,
            digest,
            status: "awaiting-input",
            stage: "排队中",
            attempts: 0,
            createdAt: new Date().toISOString(),
            updatedAt: new Date().toISOString(),
          };
          await persist(job);
          json(res, 202, publicJob(job));
          void work().catch(() => {
            console.error("Audio queue failed");
            process.exit(1);
          });
          return;
        } finally {
          accepting = false;
        }
      }
      const match = url.pathname.match(
        /^\/v1\/audio-jobs\/([a-f0-9]{64})(?:\/(input|vocals|background))?$/
      );
      if (req.method === "PUT" && match?.[2] === "input") {
        const job = jobs.get(match[1]);
        if (!job) return json(res, 404, { error: "任务不存在" });
        if (job.status !== "awaiting-input")
          return json(res, 200, publicJob(job));
        if (uploading.has(job.id))
          return json(res, 409, { error: "同一音频正在上传" });
        uploading.add(job.id);
        const dir = path.join(root, job.id),
          temp = path.join(dir, "input.part");
        try {
          await mkdir(dir, { recursive: true, mode: 0o700 });
          let size = 0;
          const hash = createHash("sha256");
          const limit = new Transform({
            transform(chunk, _encoding, cb) {
              size += chunk.length;
              hash.update(chunk);
              cb(
                size > job.plan.bytes ? Error("音频超过声明大小") : null,
                chunk
              );
            },
          });
          await pipeline(req, limit, createWriteStream(temp, { mode: 0o600 }));
          if (size !== job.plan.bytes || hash.digest("hex") !== job.plan.sha256)
            throw Error("音频大小或校验和不匹配");
          await rename(temp, path.join(dir, "input.wav"));
          job.status = "queued";
          job.stage = "等待音频处理";
          await persist(job);
          json(res, 202, publicJob(job));
          void work().catch(() => process.exit(1));
        } finally {
          uploading.delete(job.id);
          await rm(temp, { force: true });
        }
        return;
      }
      if (req.method === "GET" && match) {
        const job = jobs.get(match[1]);
        if (!job) return json(res, 404, { error: "任务不存在或已过期" });
        if (!match[2]) return json(res, 200, publicJob(job));
        if (job.status !== "succeeded")
          return json(res, 409, { error: "音轨尚未就绪" });
        if (
          !["vocals", "background"].includes(match[2]) ||
          job.plan.operation !== "separate"
        )
          return json(res, 404, { error: "结果不存在" });
        const output = path.join(root, job.id, match[2] + ".wav");
        const info = await stat(output);
        res.writeHead(200, {
          "content-type": "audio/wav",
          "content-length": info.size,
          "cache-control": "no-store",
        });
        await pipeline(createReadStream(output), res);
        return;
      }
      json(res, 404, { error: "Not found" });
    } catch (error) {
      if (res.headersSent) {
        res.destroy();
        return;
      }
      json(res, 400, {
        error:
          error instanceof z.ZodError
            ? "音频参数无效"
            : error instanceof Error && !/https?:|\/var\//.test(error.message)
              ? error.message.slice(0, 300)
              : "请求处理失败",
      });
    }
  });
  server.requestTimeout = 180000;
  server.headersTimeout = 10000;
  server.listen(
    Number(process.env.AUDIO_PORT || 18342),
    process.env.AUDIO_HOST || "127.0.0.1"
  );
  const fatal = () => {
    console.error("Audio queue persistence failure");
    process.exit(1);
  };
  const timer = setInterval(() => {
    void work().catch(fatal);
  }, 3000);
  const cleanupTimer = setInterval(() => {
    if (!busy && !accepting) void prune().catch(fatal);
  }, 3600000);
  cleanupTimer.unref();
  for (const signal of ["SIGTERM", "SIGINT"])
    process.on(signal, () => {
      stopping = true;
      clearInterval(timer);
      clearInterval(cleanupTimer);
      server.close();
      // systemd KillMode=mixed also terminates FFmpeg; persisted running jobs resume on restart.
      setTimeout(() => process.exit(0), 1000).unref();
    });
  console.log("Audio processing service ready");
}
void main().catch(() => {
  console.error("Audio processing service startup failed");
  process.exitCode = 1;
});