import { internalAction, internalMutation, internalQuery, mutation, query } from "./_generated/server";
import { v } from "convex/values";
import { nowIso } from "./_utils/time";
import { api, internal } from "./_generated/api";
import { lightragIngestText } from "./_utils/lightrag";
import { extractTextFromDocumentContent, extractTextFromMindmapData } from "./_utils/text";
import { enqueueIngestDocumentJob, enqueueIngestMediaAssetJob, enqueueIngestMindmapJob } from "./_utils/ingestJobs";
import { extractTextFromAttachment } from "./_utils/attachmentExtract";

type KernelAwareRefreshTarget = {
  documentIds: string[];
  mindmapRefs: Array<{ docId: string; mindmapId: string }>;
  assetIds: string[];
};

function uniqueNonEmptyStrings(values: Iterable<string | null | undefined>, limit: number): string[] {
  const seen = new Set<string>();
  const output: string[] = [];
  for (const value of values) {
    const normalized = String(value ?? "").trim();
    if (!normalized || seen.has(normalized)) {
      continue;
    }
    seen.add(normalized);
    output.push(normalized);
    if (output.length >= limit) {
      break;
    }
  }
  return output;
}

function truncateKernelText(value: string, limit = 800): string {
  const normalized = String(value ?? "").replace(/\s+/g, " ").trim();
  if (!normalized) {
    return "";
  }
  return normalized.length > limit ? `${normalized.slice(0, Math.max(0, limit - 1))}…` : normalized;
}

function makeKernelAwareRefreshText(input: {
  workspaceId: string;
  documentRows: Array<{ id: string; title?: string | null; raw_text?: string | null; updated_at?: string | null }>;
  mindmapRows: Array<{ document_id?: string | null; mindmap_id?: string | null; data?: unknown }>;
  assetRows: Array<{ id: string; document_id?: string | null; file_name?: string | null; ocr_text?: string | null }>;
}): string {
  const lines: string[] = [];
  lines.push(`# kernel-aware refresh workspace ${input.workspaceId}`);
  lines.push("");

  for (const doc of input.documentRows) {
    const title = String(doc.title ?? "").trim() || "无标题";
    const updatedAt = String(doc.updated_at ?? "").trim() || "unknown";
    const rawText = truncateKernelText(String(doc.raw_text ?? ""), 600);
    lines.push(`## node:${doc.id}`);
    lines.push(`title=${title}`);
    lines.push(`subtreeRoot=${doc.id}`);
    lines.push(`updatedAt=${updatedAt}`);
    if (rawText) {
      lines.push(`evidence=${rawText}`);
    }
    lines.push("");
  }

  for (const mindmap of input.mindmapRows) {
    const docId = String(mindmap.document_id ?? "").trim();
    const mindmapId = String(mindmap.mindmap_id ?? "").trim();
    if (!docId || !mindmapId) {
      continue;
    }
    const text = truncateKernelText(extractTextFromMindmapData(mindmap.data ?? null), 400);
    if (!text) {
      continue;
    }
    lines.push(`## subtree:${docId}:${mindmapId}`);
    lines.push(`node=${docId}`);
    lines.push(`subtreeRoot=${docId}`);
    lines.push(`evidence=${text}`);
    lines.push("");
  }

  for (const asset of input.assetRows) {
    const assetId = String(asset.id ?? "").trim();
    const docId = String(asset.document_id ?? "").trim();
    if (!assetId || !docId) {
      continue;
    }
    const title = String(asset.file_name ?? "").trim() || assetId;
    const ocrText = truncateKernelText(String(asset.ocr_text ?? ""), 400);
    if (!ocrText) {
      continue;
    }
    lines.push(`## evidence:${assetId}`);
    lines.push(`node=${docId}`);
    lines.push(`subtreeRoot=${docId}`);
    lines.push(`title=${title}`);
    lines.push(`evidence=${ocrText}`);
    lines.push("");
  }

  return lines.join("\n").trim();
}

async function selectKernelAwareRefreshTargets(ctx: any, args: { workspaceId: string; userId: string }): Promise<KernelAwareRefreshTarget> {
  const [documents, mindmaps, assets] = await Promise.all([
    ctx.db
      .query("documents")
      .withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
      .collect(),
    ctx.db
      .query("mindmaps")
      .withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
      .collect(),
    ctx.db
      .query("media_assets")
      .withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
      .collect(),
  ]);

  const aliveDocuments = documents
    .filter((row: any) => row.deleted_at == null)
    .filter((row: any) => String(row.user_id ?? "") === args.userId)
    .sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));
  const ownedMindmaps = mindmaps
    .filter((row: any) => row.deleted_at == null)
    .filter((row: any) => String(row.user_id ?? "") === args.userId)
    .sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));
  const aliveAssets = assets
    .filter((row: any) => row.deleted_at == null && row.purged_at == null)
    .sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));

  const documentIds = uniqueNonEmptyStrings(aliveDocuments.map((row: any) => row.id), 12);
  const selectedDocumentIds = new Set(documentIds);
  const mindmapRefs = ownedMindmaps
    .filter((row: any) => selectedDocumentIds.has(String(row.document_id ?? "").trim()))
    .slice(0, 8)
    .map((row: any) => ({
      docId: String(row.document_id ?? "").trim(),
      mindmapId: String(row.mindmap_id ?? "").trim(),
    }))
    .filter((row: { docId: string; mindmapId: string }) => row.docId && row.mindmapId);
  const assetIds = uniqueNonEmptyStrings(
    aliveAssets
      .filter((row: any) => selectedDocumentIds.has(String(row.document_id ?? "").trim()))
      .map((row: any) => row.id),
    10,
  );

  return { documentIds, mindmapRefs, assetIds };
}

async function buildKernelAwareRefreshPayload(ctx: any, args: {
  workspaceId: string;
  userId: string;
  target: KernelAwareRefreshTarget;
}) {
  const documentRows = await Promise.all(
    args.target.documentIds.map(async (documentId) => {
      const [meta, contentRes] = await Promise.all([
        ctx.runQuery(internal.documents.getMetaForIngest, { userId: args.userId, id: documentId }),
        ctx.runQuery(internal.documents.getContentForIngest, { userId: args.userId, id: documentId }),
      ]);
      if (!meta) {
        return null;
      }
      return {
        id: documentId,
        title: meta.title ?? "无标题",
        raw_text: extractTextFromDocumentContent(contentRes?.content ?? null),
        updated_at: meta.updated_at ?? null,
      };
    }),
  );

  const mindmapRows = await Promise.all(
    args.target.mindmapRefs.map(async (ref) => {
      const result = await ctx.runQuery(internal.mindmaps.getForIngest, {
        userId: args.userId,
        docId: ref.docId,
        mindmapId: ref.mindmapId,
      });
      if (!result?.ok || !result.meta?.exists) {
        return null;
      }
      return {
        document_id: ref.docId,
        mindmap_id: ref.mindmapId,
        data: result.data ?? null,
      };
    }),
  );

  const assetRows = await Promise.all(
    args.target.assetIds.map(async (assetId) => {
      const asset = await ctx.runQuery(api.mediaAssets.getById, { userId: args.userId, id: assetId });
      if (!asset) {
        return null;
      }
      return {
        id: assetId,
        document_id: asset.document_id ?? null,
        file_name: asset.file_name ?? null,
        ocr_text: asset.ocr_text ?? null,
      };
    }),
  );

  return {
    documentRows: documentRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
    mindmapRows: mindmapRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
    assetRows: assetRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
  };
}

export const get = query({
  args: { userId: v.string(), id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return null;
    if (job.user_id !== args.userId) return null;
    return {
      id: job.id,
      type: job.type,
      status: job.status,
      payload: job.payload,
      result: job.result,
      error: job.error,
      created_at: job.created_at,
      updated_at: job.updated_at,
      started_at: job.started_at,
      finished_at: job.finished_at,
    };
  },
});

export const enqueueDemo = mutation({
  args: { userId: v.string(), id: v.string(), ms: v.optional(v.number()) },
  handler: async (ctx, args) => {
    const ts = nowIso();
    const payload = { ms: args.ms ?? 800 };
    await ctx.db.insert("jobs", {
      id: args.id,
      user_id: args.userId,
      type: "demo.sleep",
      status: "queued",
      payload,
      result: null,
      error: null,
      created_at: ts,
      updated_at: ts,
      started_at: null,
      finished_at: null,
    });

    // 说明：阶段 5 骨架——用 scheduler 触发内部 mutation，再由内部 action 执行耗时逻辑。
    await ctx.scheduler.runAfter(0, internal.jobs.start, { id: args.id });
    return { ok: true, id: args.id };
  },
});

export const enqueueRagIndexDocument = mutation({
  args: { userId: v.string(), documentId: v.string() },
  handler: async (ctx, args) => {
    const { id } = await enqueueIngestDocumentJob(ctx, { userId: args.userId, documentId: args.documentId, debounceMs: 0 });
    return { ok: true, id };
  },
});

export const enqueueRagIndexMindmap = mutation({
  args: { userId: v.string(), docId: v.string(), mindmapId: v.string() },
  handler: async (ctx, args) => {
    const { id } = await enqueueIngestMindmapJob(ctx, {
      userId: args.userId,
      docId: args.docId,
      mindmapId: args.mindmapId,
      debounceMs: 0,
    });
    return { ok: true, id };
  },
});

export const enqueueRagIndexMediaAsset = mutation({
  args: { userId: v.string(), assetId: v.string() },
  handler: async (ctx, args) => {
    const { id } = await enqueueIngestMediaAssetJob(ctx, { userId: args.userId, assetId: args.assetId, debounceMs: 0 });
    return { ok: true, id };
  },
});

export const enqueueKernelAwareRefresh = mutation({
  args: { userId: v.string(), workspaceId: v.string() },
  handler: async (ctx, args) => {
    const workspaceId = String(args.workspaceId ?? "").trim();
    if (!workspaceId) {
      throw new Error("缺少 workspaceId");
    }
    const membership = await ctx.db
      .query("workspace_members")
      .withIndex("by_workspace_user", (q) => q.eq("workspace_id", workspaceId).eq("user_id", args.userId))
      .first();
    if (!membership) {
      throw new Error("无权访问该工作空间");
    }
    const ts = nowIso();
    const id = `refresh:kernel-aware:${workspaceId}`;
    const payload = {
      workspaceId,
      enqueuedAt: ts,
      trigger: "manual",
    };
    const existing = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", id))
      .first();
    if (existing) {
      await ctx.db.patch(existing._id, {
        user_id: args.userId,
        type: "refresh.kernel_aware_transition",
        status: "queued",
        payload,
        result: null,
        error: null,
        updated_at: ts,
        started_at: null,
        finished_at: null,
      });
    } else {
      await ctx.db.insert("jobs", {
        id,
        user_id: args.userId,
        type: "refresh.kernel_aware_transition",
        status: "queued",
        payload,
        result: null,
        error: null,
        created_at: ts,
        updated_at: ts,
        started_at: null,
        finished_at: null,
      });
    }
    await ctx.scheduler.runAfter(0, internal.jobs.start, { id });
    return { ok: true, id };
  },
});

export const enqueueKernelAwareRefreshSweep = internalMutation({
  args: {},
  handler: async (ctx) => {
    const memberships = await ctx.db.query("workspace_members").collect();
    const ownersByWorkspace = new Map<string, string>();
    for (const membership of memberships) {
      if (membership.role !== "owner") {
        continue;
      }
      if (!ownersByWorkspace.has(membership.workspace_id)) {
        ownersByWorkspace.set(membership.workspace_id, membership.user_id);
      }
    }

    const scheduled: string[] = [];
    for (const [workspaceId, userId] of ownersByWorkspace.entries()) {
      const id = `refresh:kernel-aware:${workspaceId}`;
      const ts = nowIso();
      const payload = {
        workspaceId,
        enqueuedAt: ts,
        trigger: "cron",
      };
      const existing = await ctx.db
        .query("jobs")
        .withIndex("by_job_id", (q) => q.eq("id", id))
        .first();
      if (existing && (existing.status === "queued" || existing.status === "running")) {
        scheduled.push(id);
        continue;
      }
      if (existing) {
        await ctx.db.patch(existing._id, {
          user_id: userId,
          type: "refresh.kernel_aware_transition",
          status: "queued",
          payload,
          result: null,
          error: null,
          updated_at: ts,
          started_at: null,
          finished_at: null,
        });
      } else {
        await ctx.db.insert("jobs", {
          id,
          user_id: userId,
          type: "refresh.kernel_aware_transition",
          status: "queued",
          payload,
          result: null,
          error: null,
          created_at: ts,
          updated_at: ts,
          started_at: null,
          finished_at: null,
        });
      }
      await ctx.scheduler.runAfter(0, internal.jobs.start, { id });
      scheduled.push(id);
    }

    return {
      ok: true,
      scheduledCount: scheduled.length,
      jobIds: scheduled,
    };
  },
});

export const start = internalMutation({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    if (job.status !== "queued") return;
    const ts = nowIso();
    await ctx.db.patch(job._id, { status: "running", started_at: ts, updated_at: ts });
    await ctx.scheduler.runAfter(0, internal.jobs.run, { id: args.id });
  },
});

export const run = internalAction({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.runQuery(internal.jobs._getInternal, { id: args.id });
    if (!job) return;
    if (job.status !== "running") return;

    try {
      if (job.type === "demo.sleep") {
        const ms = typeof job.payload?.ms === "number" ? job.payload.ms : 800;
        await new Promise((r) => setTimeout(r, ms));
        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: { ok: true, sleptMs: ms },
        });
        return;
      }

      if (job.type === "ingest.rag_index_document") {
        const documentId = String(job.payload?.documentId ?? "").trim();
        if (!documentId) throw new Error("缺少 documentId");

        const [meta, contentRes] = await Promise.all([
          ctx.runQuery(internal.documents.getMetaForIngest, { userId: job.user_id, id: documentId }),
          ctx.runQuery(internal.documents.getContentForIngest, { userId: job.user_id, id: documentId }),
        ]);

        if (!meta) throw new Error("页面不存在或无权限");

        const title = meta.title ?? "无标题";
        const text = extractTextFromDocumentContent(contentRes?.content ?? null);
        const fileSource = `document:${documentId}`;
        const ingest = await lightragIngestText({ fileSource, text: `# ${title}\n\n${text}` });
        if (ingest.skipped) {
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "document", documentId, skipped: true, reason: ingest.reason ?? "skipped" },
          });
          return;
        }

        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: { ok: true, kind: "document", documentId, trackId: ingest.trackId },
        });
        return;
      }

      if (job.type === "ingest.rag_index_mindmap") {
        const docId = String(job.payload?.docId ?? "").trim();
        const mindmapId = String(job.payload?.mindmapId ?? "").trim();
        if (!docId) throw new Error("缺少 docId");
        if (!mindmapId) throw new Error("缺少 mindmapId");

        const [docMeta, mindmapRes] = await Promise.all([
          ctx.runQuery(internal.documents.getMetaForIngest, { userId: job.user_id, id: docId }),
          ctx.runQuery(internal.mindmaps.getForIngest, { userId: job.user_id, docId, mindmapId }),
        ]);

        if (!docMeta) throw new Error("页面不存在或无权限");
        if (!mindmapRes?.ok) throw new Error("导图读取失败");
        if (!mindmapRes.meta?.exists) {
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "mindmap", docId, mindmapId, skipped: true, reason: "mindmap_not_exists" },
          });
          return;
        }

        const title = docMeta.title ?? "无标题";
        const text = extractTextFromMindmapData(mindmapRes.data ?? null);
        const fileSource = `mindmap:${docId}:${mindmapId}`;
        const ingest = await lightragIngestText({ fileSource, text: `# ${title}\n\n${text}` });
        if (ingest.skipped) {
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "mindmap", docId, mindmapId, skipped: true, reason: ingest.reason ?? "skipped" },
          });
          return;
        }

        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: { ok: true, kind: "mindmap", docId, mindmapId, trackId: ingest.trackId },
        });
        return;
      }

      if (job.type === "ingest.rag_index_media_asset") {
        const assetId = String(job.payload?.assetId ?? "").trim();
        if (!assetId) throw new Error("缺少 assetId");

        const asset = await ctx.runQuery(api.mediaAssets.getById, { userId: job.user_id, id: assetId });
        if (!asset) throw new Error("资源不存在或无权限");

        const title = asset.file_name ?? asset.id;
        const text = String(asset.ocr_text ?? "").trim();
        if (!text) {
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "media_asset", assetId, skipped: true, reason: "empty_ocr_text" },
          });
          return;
        }
        const fileSource = `media:${assetId}`;
        const ingest = await lightragIngestText({ fileSource, text: `# ${title}\n\n${text}` });
        if (ingest.skipped) {
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "media_asset", assetId, skipped: true, reason: ingest.reason ?? "skipped" },
          });
          return;
        }

        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: { ok: true, kind: "media_asset", assetId, trackId: ingest.trackId },
        });
        return;
      }

      if (job.type === "extract.media_asset_text") {
        const assetId = String(job.payload?.assetId ?? "").trim();
        if (!assetId) throw new Error("缺少 assetId");

        const asset = await ctx.runQuery(api.mediaAssets.getById, { userId: job.user_id, id: assetId });
        if (!asset) throw new Error("资源不存在或无权限");

        // 说明：只处理 file 类型的常见附件（pdf/docx/pptx/xlsx）。
        if (String(asset.asset_type ?? "") !== "file") {
          await ctx.runMutation(api.mediaAssets.patchById, {
            userId: job.user_id,
            id: assetId,
            patch: {
              ocr_status: "skipped",
              ocr_payload: { reason: "non_file_asset" },
              ocr_strategy: "attachment_extract",
            },
          });
          await ctx.runMutation(internal.jobs.finishSuccess, {
            id: args.id,
            result: { ok: true, kind: "extract_media_asset_text", assetId, skipped: true, reason: "non_file_asset" },
          });
          return;
        }

        const fileSize = typeof asset.file_size === "number" ? asset.file_size : null;
        const maxBytes = 25 * 1024 * 1024;
        if (typeof fileSize === "number" && fileSize > maxBytes) {
          await ctx.runMutation(api.mediaAssets.patchById, {
            userId: job.user_id,
            id: assetId,
            patch: {
              ocr_status: "failed",
              ocr_payload: { error: `文件过大（${fileSize} bytes），暂不解析`, maxBytes },
              ocr_strategy: "attachment_extract",
            },
          });
          await ctx.runMutation(internal.jobs.finishFailure, {
            id: args.id,
            error: "文件过大，暂不解析",
          });
          return;
        }

        // 说明：Convex Files 的 getUrl 可能过期，先刷新并获取当前可用链接。
        const refreshed = await ctx.runMutation(api.mediaAssets.refreshUrl, { userId: job.user_id, id: assetId });
        const url = String((refreshed as any)?.signedUrl ?? asset.file_url ?? "").trim();
        if (!url) throw new Error("缺少可用文件链接");

        const res = await fetch(url);
        if (!res.ok) {
          throw new Error(`下载附件失败：${res.status}`);
        }
        const bytes = await res.arrayBuffer();

        const extracted = await extractTextFromAttachment({
          mimeType: (asset as any).mime_type ?? null,
          fileName: (asset as any).file_name ?? null,
          bytes,
        });

        if (!extracted.ok) {
          await ctx.runMutation(api.mediaAssets.patchById, {
            userId: job.user_id,
            id: assetId,
            patch: {
              ocr_status: "failed",
              ocr_payload: { error: extracted.reason, meta: extracted.meta ?? null },
              ocr_strategy: extracted.strategy,
            },
          });
          await ctx.runMutation(internal.jobs.finishFailure, { id: args.id, error: extracted.reason });
          return;
        }

        await ctx.runMutation(api.mediaAssets.patchById, {
          userId: job.user_id,
          id: assetId,
          patch: {
            ocr_text: extracted.text,
            ocr_status: "completed",
            ocr_payload: extracted.meta ?? null,
            ocr_strategy: extracted.strategy,
          },
        });

        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: {
            ok: true,
            kind: "extract_media_asset_text",
            assetId,
            strategy: extracted.strategy,
            chars: extracted.text.length,
          },
        });
        return;
      }

      if (job.type === "refresh.kernel_aware_transition") {
        const workspaceId = String(job.payload?.workspaceId ?? "").trim();
        if (!workspaceId) throw new Error("缺少 workspaceId");

        const membership = await ctx.runQuery((internal as any).jobs._getWorkspaceMembership, {
          workspaceId,
          userId: job.user_id,
        });
        if (!membership) {
          throw new Error("工作空间不存在或无权限");
        }

        const target = await ctx.runQuery((internal as any).jobs._selectKernelAwareRefreshTargets, {
          workspaceId,
          userId: job.user_id,
        });
        const payload = await buildKernelAwareRefreshPayload(ctx, {
          workspaceId,
          userId: job.user_id,
          target,
        });
        const text = makeKernelAwareRefreshText({
          workspaceId,
          documentRows: payload.documentRows,
          mindmapRows: payload.mindmapRows,
          assetRows: payload.assetRows,
        });
        const ingest = await lightragIngestText({
          fileSource: `kernel-aware-refresh:${workspaceId}`,
          text,
        });

        await Promise.all([
          ...target.documentIds.slice(0, 6).map((documentId: string) =>
            ctx.runMutation(api.jobs.enqueueRagIndexDocument, { userId: job.user_id, documentId }).catch(() => null),
          ),
          ...target.mindmapRefs.slice(0, 4).map((item: { docId: string; mindmapId: string }) =>
            ctx.runMutation(api.jobs.enqueueRagIndexMindmap, {
              userId: job.user_id,
              docId: item.docId,
              mindmapId: item.mindmapId,
            }).catch(() => null),
          ),
          ...target.assetIds.slice(0, 4).map((assetId: string) =>
            ctx.runMutation(api.jobs.enqueueRagIndexMediaAsset, { userId: job.user_id, assetId }).catch(() => null),
          ),
        ]);

        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: {
            ok: true,
            kind: "kernel_aware_refresh",
            workspaceId,
            refreshMode: "kernel_aware_transition",
            bridge: {
              backend: "lightrag",
              fileSource: `kernel-aware-refresh:${workspaceId}`,
              skipped: Boolean(ingest.skipped),
              reason: ingest.reason ?? null,
              trackId: ingest.trackId,
            },
            refreshedDocuments: target.documentIds.length,
            refreshedMindmaps: target.mindmapRefs.length,
            refreshedAssets: target.assetIds.length,
            kernelPreview: {
              nodeIds: payload.documentRows.map((item) => item.id).slice(0, 8),
              subtreeRootIds: payload.documentRows.map((item) => item.id).slice(0, 8),
              evidenceAssetIds: payload.assetRows.map((item) => item.id).slice(0, 8),
            },
          },
        });
        return;
      }

      throw new Error(`未知任务类型：${job.type}`);
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      await ctx.runMutation(internal.jobs.finishFailure, { id: args.id, error: message });
    }
  },
});

export const _getInternal = internalQuery({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return null;
    return {
      id: job.id,
      user_id: job.user_id,
      type: job.type,
      status: job.status,
      payload: job.payload,
    };
  },
});

export const _getWorkspaceMembership = internalQuery({
  args: { workspaceId: v.string(), userId: v.string() },
  handler: async (ctx, args) => {
    return await ctx.db
      .query("workspace_members")
      .withIndex("by_workspace_user", (q) => q.eq("workspace_id", args.workspaceId).eq("user_id", args.userId))
      .first();
  },
});

export const _selectKernelAwareRefreshTargets = internalQuery({
  args: { workspaceId: v.string(), userId: v.string() },
  handler: async (ctx, args) => {
    return await selectKernelAwareRefreshTargets(ctx, args);
  },
});

export const finishSuccess = internalMutation({
  args: { id: v.string(), result: v.any() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    const ts = nowIso();
    await ctx.db.patch(job._id, {
      status: "succeeded",
      result: args.result,
      error: null,
      finished_at: ts,
      updated_at: ts,
    });
  },
});

export const finishFailure = internalMutation({
  args: { id: v.string(), error: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    const ts = nowIso();
    await ctx.db.patch(job._id, {
      status: "failed",
      result: null,
      error: args.error,
      finished_at: ts,
      updated_at: ts,
    });
  },
});