Function handleExecuteJob

integrations/smart-ingest/index.ts:895–988

(req: Request)

// ── Execute a dry-run job ───────────────────────────────────────────────────

async function handleExecuteJob(req: Request): Promise<Response> {
  let body: Record<string, unknown>;
  try { body = await req.json(); } catch { return json({ error: "Invalid JSON body" }, 400); }

  const jobId = typeof body.job_id === "number" ? body.job_id : 0;
  if (!jobId) return json({ error: "job_id is required" }, 400);

  const { data: job, error: jobErr } = await supabase
    .from("ingestion_jobs").select("*").eq("id", jobId).single();
  if (jobErr || !job) return json({ error: `Job #${jobId} not found` }, 404);
  if (job.status === "complete") return json({ ...job, message: "Job already complete" }, 200);
  if (job.status !== "dry_run_complete") {
    return json({ error: `Job status is '${job.status}', expected 'dry_run_complete'` }, 400);
  }

  const { data: itemRows } = await supabase
    .from("ingestion_items").select("*").eq("job_id", jobId).order("id");
  const items = itemRows ?? [];

  // CAS: only transition dry_run_complete -> executing; concurrent requests get 409
  const { data: casRow, error: casErr } = await supabase
    .from("ingestion_jobs")
    .update({ status: "executing" })
    .eq("id", jobId)
    .eq("status", "dry_run_complete")
    .select("id, status")
    .maybeSingle();
  if (casErr || !casRow || casRow.status !== "executing") {
    return json({ error: "Job execution conflict — another request may have claimed this job" }, 409);
  }

  let addedCount = 0, skippedCount = 0, appendedCount = 0, revisedCount = 0;
  const sourceLabel = job.source_label ?? null;
  const jobMeta = (job.metadata ?? {}) as Record<string, unknown>;
  const sourceType = jobMeta.source_type as string ?? "smart_ingest";
  const skipClassification = body.skip_classification === true || jobMeta.skip_classification === true;
  const jobSourceMetadata = (jobMeta.source_client || jobMeta.capture_mode)
    ? jobMeta as Record<string, unknown>
    : null;

  for (const item of items) {
    if (item.action === "skip") { skippedCount++; continue; }
    try {
      const fakeItem: IngestionItem = {
        content: item.extracted_content,
        type: sanitizeType((item.metadata as Record<string, unknown>)?.type),
        importance: sanitizeImportance((item.metadata as Record<string, unknown>)?.importance),
        tags: sanitizeTags((item.metadata as Record<string, unknown>)?.tags),
        source_snippet: sanitizeSourceSnippet((item.metadata as Record<string, unknown>)?.source_snippet),
        content_fingerprint: "",
        action: item.action as ReconcileAction,
        reason: item.reason ?? "",
        matched_thought_id: item.matched_thought_id,
        similarity_score: item.similarity_score,
        status: "pending",
        error_message: null,
      };
      let embedding: number[] = [];