import { createHash, randomUUID } from "node:crypto";
import { lstat, mkdir, open, readFile, readdir, rename, stat } from "node:fs/promises";
import { join, resolve } from "node:path";
import type { AttachmentStore } from "../attachments/attachment-store.js";
import { HarnessError } from "../errors/harness-error.js";
import { ImageVideoDraftSchema, recordSchema, taskIdSchema, validationSchema, type ImageVideoDraft, type ImageVideoPublicTask, type ImageVideoRecord, type ImageVideoRemote, type ImageVideoValidation } from "./contracts.js";

const locks = new Map<string, Promise<unknown>>();
const submissionReceiptSchema = recordSchema.pick({ version: true, id: true, revision: true, requestHash: true, jobId: true }).required().strict();
export const sha256 = (bytes: Buffer | string) => createHash("sha256").update(bytes).digest("hex");
const fail = (code: string, message: string): never => { throw new HarnessError(code, message); };
function safeRemoteCode(error: unknown): string {
  const codes = ["I2V_REMOTE_REJECTED", "I2V_RESPONSE_UNSUPPORTED", "I2V_REMOTE_FAILED", "I2V_SCHEMA_CHANGED", "I2V_TRANSFER_TRANSIENT"];
  return error instanceof HarnessError && codes.includes(error.code) ? error.code : "I2V_OPERATION_FAILED";
}
export type ImageVideoOptions = { root: string; attachments: Pick<AttachmentStore, "get" | "openNormalized">; remote: ImageVideoRemote; collect: (url: string, directory: string, duration: number) => Promise<ImageVideoValidation> };

/** One process owns this private store. Per-task locks are shared by service instances. */
export class ImageVideoService {
  readonly root: string;
  constructor(private readonly options: ImageVideoOptions) { this.root = resolve(options.root); }

  async create(sessionId: string, input: ImageVideoDraft): Promise<ImageVideoPublicTask> {
    const draft = ImageVideoDraftSchema.parse(input);
    const attachment = this.options.attachments.get(sessionId, draft.attachmentId);
    if (!attachment || attachment.sessionId !== sessionId) return fail("I2V_TASK_NOT_FOUND", "Reference image not found in this Session");
    const bytes = await this.options.attachments.openNormalized(sessionId, draft.attachmentId);
    const now = new Date().toISOString(); const id = `i2v_${randomUUID()}`;
    const record = recordSchema.parse({ version: 1, id, ownerSessionId: sessionId, revision: 0, draft, referenceHash: sha256(bytes), requestHash: sha256(JSON.stringify(draft)), referenceType: attachment.mediaType, state: "draft", createdAt: now, updatedAt: now });
    await mkdir(join(this.root, id), { recursive: true, mode: 0o700 });
    await this.atomic(join(this.root, id, "reference"), bytes);
    await this.atomic(join(this.root, id, "record.json"), JSON.stringify(record));
    return this.public(record);
  }

  async list(): Promise<ImageVideoPublicTask[]> {
    const entries = await readdir(this.root).catch(() => []);
    const tasks = await Promise.all(entries.filter(id => taskIdSchema.safeParse(id).success).map(id => this.view(id).catch(() => undefined)));
    return tasks.filter((task): task is ImageVideoPublicTask => task !== undefined).sort((a, b) => b.createdAt.localeCompare(a.createdAt));
  }
  async view(id: string): Promise<ImageVideoPublicTask> { return this.lock(id, async () => this.public(await this.read(id))); }
  async owned(sessionId: string, id: string): Promise<ImageVideoPublicTask> { return this.lock(id, async () => { const r = await this.read(id); this.owner(r, sessionId); return this.public(r); }); }

  /** Local-page confirmation only. Not exposed to the model as a tool. */
  async continueTask(sessionId: string, id: string, expectedRevision: number, ownerIsActive: (sessionId: string) => boolean, requireTargetSession: () => void): Promise<ImageVideoPublicTask> {
    return this.lock(id, async () => {
      const r = await this.read(id);
      requireTargetSession();
      if (r.revision !== expectedRevision) return fail("I2V_STALE_REVISION", "任务已更新，请重新核对后接续。");
      if (!["draft", "awaiting_submission"].includes(r.state) || r.jobId || (r.state === "awaiting_submission" && !r.assetId)) return fail("I2V_CONTINUE_BLOCKED", "当前状态不允许接续；不得重置未知结果或重新生成已有任务。");
      if (r.ownerSessionId === sessionId) return this.public(r);
      if (ownerIsActive(r.ownerSessionId)) return fail("I2V_OWNER_ACTIVE", "原会话仍有效，请回原页面继续，不能由其他会话接管。");
      if ((r.continuations?.length ?? 0) >= 100) return fail("I2V_CONTINUE_BLOCKED", "接续记录已达上限，请人工核实。");
      await this.reference(r);
      requireTargetSession();
      r.continuations = [...(r.continuations ?? []), { fromSessionId: r.ownerSessionId, toSessionId: sessionId, revision: r.revision, at: new Date().toISOString() }];
      r.ownerSessionId = sessionId;
      await this.save(r);
      return this.public(r);
    });
  }

  async upload(sessionId: string, id: string): Promise<ImageVideoPublicTask> {
    return this.lock(id, async () => {
      const r = await this.read(id); this.owner(r, sessionId);
      if (r.state === "unknown_upload") return fail("I2V_UNKNOWN_UPLOAD", "Upload outcome unknown; inspect the saved receipt before any new import");
      if (r.state !== "draft") return this.public(r);
      // Read-only discovery and identity validation happen before any remote mutation intent.
      r.identity = await this.options.remote.preflight();
      const bytes = await this.reference(r);
      r.state = "initializing_upload"; await this.save(r);
      try {
        const initialized = await this.options.remote.initialize(r, bytes);
        if ("reused" in initialized) {
          r.assetId = initialized.assetId; r.state = "awaiting_submission"; delete r.error; await this.save(r);
          return this.public(r);
        }
        r.upload = initialized; r.state = "uploading"; await this.save(r);
        await this.options.remote.put(r, bytes); r.state = "finalizing"; await this.save(r);
        await this.options.remote.finalize(r, bytes); r.state = "importing"; await this.save(r);
      } catch (error) {
        const stage = r.state;
        const code = safeRemoteCode(error);
        r.state = "unknown_upload"; r.error = `参考图上传或入库未完成（${stage}/${code}）；禁止重复导入，请核实本任务收据。`; await this.save(r);
      }
      return this.public(r);
    });
  }

  async submit(sessionId: string, id: string): Promise<ImageVideoPublicTask> {
    return this.lock(id, async () => {
      const r = await this.read(id); this.owner(r, sessionId);
      if (r.state === "unknown_submission") return fail("I2V_UNKNOWN_SUBMISSION", "Submission outcome unknown; no safe retry or new billing request is allowed");
      if (r.jobId) return this.public(r);
      if (r.state !== "awaiting_submission" || !r.assetId) return fail("I2V_NOT_READY", "Reference import must be ready before generation approval");
      await this.options.remote.preflight(); await this.reference(r);
      r.state = "submitting"; await this.save(r);
      try {
        // Deliberately independent of chat cancellation: retain any returned receipt.
        const receipt = await this.options.remote.submit(r);
        if (!receipt.jobId) throw new Error("Missing job ID");
        await this.atomic(join(this.root, id, "submission-receipt.json"), JSON.stringify({ version: 1, id, revision: r.revision, requestHash: r.requestHash, jobId: receipt.jobId }));
        r.jobId = receipt.jobId; r.state = "queued"; delete r.lastPollAt; await this.save(r);
      } catch (error) { r.state = "unknown_submission"; r.error = `生成提交结果未知（submitting/${safeRemoteCode(error)}）；禁止重投。请核实私有回执，避免重复计费。`; await this.save(r); }
      return this.public(r);
    });
  }

  async refresh(id: string): Promise<ImageVideoPublicTask> {
    return this.lock(id, async () => {
      const r = await this.read(id);
      if (r.state === "unknown_submission" || r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r);
      if (!r.jobId && !(r.upload && ["importing", "unknown_upload"].includes(r.state))) return this.public(r);
      r.lastPollAt = Date.now(); await this.save(r);
      let collecting = false;
      try {
        if (!r.jobId) {
          const result = await this.options.remote.importStatus(r);
          if (result.state === "ready" && result.assetId) { r.assetId = result.assetId; r.state = "awaiting_submission"; delete r.error; }
          else if (result.state === "failed") { r.state = "failed"; r.error = "参考图入库失败。"; }
        } else {
          const result = await this.options.remote.status(r);
          if (result.state === "succeeded") {
            if (!result.artifactUrl) throw new Error("No unambiguous MP4 artifact");
            r.artifactUrl = result.artifactUrl; r.state = "validating"; await this.save(r);
            collecting = true;
            r.validation = validationSchema.parse(await this.options.collect(result.artifactUrl, join(this.root, id), r.draft.durationSeconds));
            if (Math.abs(r.validation.videoSeconds - r.draft.durationSeconds) > 0.002 || Math.abs(r.validation.audioSeconds - r.draft.durationSeconds) > 0.05) throw new Error("Duration mismatch");
            r.state = "ready"; delete r.error;
          } else { r.state = result.state; if (result.state === "failed") r.error = "H3 生成失败；未自动重投。"; }
        }
      } catch (error) {
        const retryCollection = collecting && error instanceof HarnessError && error.code === "I2V_TRANSFER_TRANSIENT";
        if (collecting && !retryCollection) r.state = "failed";
        r.error = retryCollection ? "成片收取暂时失败；30 秒后查询原任务并重试收取，不会重新生成。" : r.state === "failed" ? "成片技术验证失败：请核实 MP4、精确时长、尺寸、帧率和音轨；未自动重投。" : "状态响应暂不可用或契约不明确；保留原任务，下次只读查询。";
      }
      await this.save(r); return this.public(r);
    });
  }

  async media(id: string, kind: "reference" | "video") {
    return this.lock(id, async () => {
      const r = await this.read(id);
      if (kind === "video" && (r.state !== "ready" || !r.validation)) return fail("I2V_MEDIA_NOT_FOUND", "Validated video not ready");
      const path = join(this.root, id, kind === "reference" ? "reference" : "video.mp4");
      await this.regular(path); const bytes = await readFile(path);
      if (sha256(bytes) !== (kind === "reference" ? r.referenceHash : r.validation?.sha256)) return fail("I2V_MEDIA_INVALID", "Registered media hash mismatch");
      return { path, size: (await stat(path)).size, contentType: kind === "reference" ? r.referenceType : "video/mp4" };
    });
  }
  private owner(r: ImageVideoRecord, sessionId: string) { if (r.ownerSessionId !== sessionId) fail("I2V_TASK_NOT_FOUND", "Task not found in this Session"); }
  private public(r: ImageVideoRecord): ImageVideoPublicTask {
    return { id: r.id, revision: r.revision, title: r.draft.title, prompt: r.draft.prompt, durationSeconds: r.draft.durationSeconds, state: r.state, createdAt: r.createdAt, updatedAt: r.updatedAt, referenceUrl: `/api/image-video/tasks/${r.id}/reference`, ...(r.jobId ? { jobId: r.jobId } : {}), ...(r.error ? { error: r.error } : {}), ...(r.validation ? { validation: r.validation } : {}), ...(r.state === "ready" ? { videoUrl: `/api/image-video/tasks/${r.id}/video` } : {}) };
  }
  private async reference(r: ImageVideoRecord) { await this.regular(join(this.root, r.id, "reference")); const bytes = await readFile(join(this.root, r.id, "reference")); if (sha256(bytes) !== r.referenceHash) fail("I2V_MEDIA_INVALID", "Reference changed after draft"); return bytes; }
  private async regular(path: string) { const info = await lstat(path); if (!info.isFile() || info.isSymbolicLink()) fail("I2V_RECORD_INVALID", "Unsafe registered file"); }
  private async matchingSubmissionReceipt(r: ImageVideoRecord): Promise<string | undefined> {
    const path = join(this.root, r.id, "submission-receipt.json");
    try {
      await this.regular(path);
      const receipt = submissionReceiptSchema.parse(JSON.parse(await readFile(path, "utf8")));
      if (receipt.id !== r.id || receipt.requestHash !== r.requestHash || receipt.revision !== r.revision || !receipt.jobId.trim() || (r.jobId !== undefined && r.jobId !== receipt.jobId)) return undefined;
      return receipt.jobId;
    } catch { return undefined; } // Preserve absent, malformed and conflicting evidence for manual reconciliation.
  }
  private async read(id: string): Promise<ImageVideoRecord> {
    if (!taskIdSchema.safeParse(id).success) return fail("I2V_TASK_NOT_FOUND", "Task not found");
    const path = join(this.root, id, "record.json");
    const dir = await lstat(join(this.root, id)).catch(() => undefined);
    if (!dir || !dir.isDirectory() || dir.isSymbolicLink()) return fail("I2V_TASK_NOT_FOUND", "Task not found");
    try {
      await this.regular(path); const r = recordSchema.parse(JSON.parse(await readFile(path, "utf8")));
      if (r.id !== id || r.requestHash !== sha256(JSON.stringify(r.draft))) throw new Error("Binding mismatch");
      if (r.state === "submitting") {
        const jobId = await this.matchingSubmissionReceipt(r);
        if (jobId !== undefined) { r.jobId = jobId; r.state = "queued"; delete r.lastPollAt; delete r.error; await this.save(r); }
        else { r.state = "unknown_submission"; await this.save(r); }
      }
      if (["initializing_upload", "uploading", "finalizing"].includes(r.state)) { r.state = "unknown_upload"; await this.save(r); }
      return r;
    } catch { return fail("I2V_RECORD_INVALID", "Task record is invalid; manual recovery required"); }
  }
  private async save(r: ImageVideoRecord) {
    const path = join(this.root, r.id, "record.json");
    const previous = recordSchema.parse(JSON.parse(await readFile(path, "utf8")));
    if (previous.revision !== r.revision) return fail("I2V_STALE_REVISION", "Stale task result rejected");
    r.revision++; r.updatedAt = new Date().toISOString();
    await this.atomic(path, JSON.stringify(recordSchema.parse(r)));
  }
  private async atomic(path: string, content: Buffer | string) { const temp = `${path}.${randomUUID()}.tmp`; const file = await open(temp, "wx", 0o600); try { await file.writeFile(content); await file.sync(); } finally { await file.close(); } await rename(temp, path); }
  private async lock<T>(id: string, operation: () => Promise<T>): Promise<T> {
    const key = `${this.root}/${id}`; const previous = locks.get(key) ?? Promise.resolve();
    const work = previous.catch(() => undefined).then(operation); locks.set(key, work);
    try { return await work; } finally { if (locks.get(key) === work) locks.delete(key); }
  }
}