batchCooking/apps/api/src/modules/internal/tech-step-worker.service.ts

import { HttpError } from "@batch-cooking/error-tools";
import {
  ErrorCode,
  type PendingTechStepCorrectionView,
  type SubmitTrainingSuggestionsInput,
  type TechStepAuditClauseView,
  type WorkerHeartbeatInput,
} from "@batch-cooking/shared";
import type { Prisma } from "@prisma/client";
import { prisma } from "../../db/prisma.js";
import {
  CONFIDENCE_THRESHOLD,
  techStepClassifier,
} from "../../lib/recipe-matching/tech-step-matcher.js";

/**
 * Read/write surface `services/tech-step-llm-worker` calls through
 * `/internal/tech-steps/*` (`tech-step-worker.routes.ts`, guarded by
 * `requireInternalWorker`) — the worker has no Prisma client or database
 * credentials of its own (see that service's own README), so every
 * corrections/audit-sample read and every suggestion write goes through
 * here rather than the worker touching this schema directly. Keeps
 * `apps/api` the single owner of the schema/migrations, and keeps the
 * worker a pure "read some text, run inference, post a suggestion" process
 * with nothing to keep in sync if the schema changes shape.
 */

/**
 * How many of the most recently created `Step`s {@link getAuditBatch} scans
 * per call before filtering down to low-confidence clauses — a fixed
 * recency-biased sample, not every `Step` in the database, to keep this
 * endpoint's cost bounded regardless of how large the recipe catalog gets.
 * Recently-added steps are also the steps most likely to still use
 * vocabulary the training corpus hasn't caught up with yet, which is
 * exactly what this audit is for. A smarter sampling strategy (e.g.
 * weighted by how often a recipe is actually viewed/planned) is future
 * work, not needed for this feature's first version.
 */
const AUDIT_SAMPLE_SIZE = 200;

/**
 * Every low-confidence clause found across a recency-biased sample of
 * existing `Step`s (see {@link AUDIT_SAMPLE_SIZE}), for
 * `services/tech-step-llm-worker`'s `audit-low-confidence` job to get a
 * second opinion on. "Low-confidence" mirrors exactly what
 * `TechStepClassifierService._classifyClause` itself distrusts (a clause
 * with an NER anchor but a classifier score under
 * {@link CONFIDENCE_THRESHOLD}) — the same clauses that pipeline already
 * has to fall back to keyword-anchor guessing for, not an arbitrary
 * separate cutoff.
 */
export async function getAuditBatch(
  locale: string,
  limit: number,
): Promise<TechStepAuditClauseView[]> {
  try {
    const steps = await prisma.step.findMany({
      orderBy: { id: "desc" },
      take: AUDIT_SAMPLE_SIZE,
      select: { id: true, recipeId: true, description: true },
    });

    const results: TechStepAuditClauseView[] = [];
    for (const step of steps) {
      if (results.length >= limit) break;
      const clauses = await techStepClassifier.classifyClauses(step.description, locale);
      for (const clause of clauses) {
        if (results.length >= limit) break;
        const isLowConfidence = clause.anchorUid !== null && clause.score < CONFIDENCE_THRESHOLD;
        if (!isLowConfidence) continue;
        results.push({
          stepId: step.id,
          recipeId: step.recipeId,
          clauseText: clause.clauseText,
          anchorKey: clause.anchorUid,
          intentKey: clause.intentUid,
          score: clause.score,
          locale,
        });
      }
    }
    return results;
  } catch (err) {
    throw err; // see recipe.service.ts's equivalent catch comment
  }
}

/**
 * Every `StepTechStepCorrection` not yet turned into a
 * `TechStepTrainingSuggestion` (`consumedAt IS NULL`), oldest first — a
 * FIFO queue the worker's `transform-corrections` job drains, `limit` at a
 * time.
 *
 * `correctedTechStepId IS NOT NULL` on top of `consumedAt IS NULL`: a
 * correction that *removes* a match ("no technique belongs here",
 * `correctedTechStepId: null` — see `StepTechStepCorrection`'s schema doc
 * comment) has no technique to propose new positive training data *for*.
 * Surfacing it here would leave it permanently unconsumable (the worker
 * has nothing to submit a suggestion for, so it would never stamp
 * `consumedAt`, and it would keep re-appearing in every future batch
 * forever) — excluded at the source instead, not filtered/skipped
 * downstream by the worker.
 */
export async function getPendingCorrections(
  limit: number,
): Promise<PendingTechStepCorrectionView[]> {
  try {
    const corrections = await prisma.stepTechStepCorrection.findMany({
      where: { consumedAt: null, correctedTechStepId: { not: null } },
      orderBy: { createdAt: "asc" },
      take: limit,
      include: {
        step: { select: { id: true, recipeId: true, description: true } },
        previousTechStep: { select: { key: true } },
        correctedTechStep: { select: { key: true } },
      },
    });

    return corrections.map((correction) => ({
      id: correction.id,
      stepId: correction.step.id,
      recipeId: correction.step.recipeId,
      clauseText: correction.step.description.slice(correction.start, correction.end),
      start: correction.start,
      end: correction.end,
      previousTechStepKey: correction.previousTechStep?.key ?? null,
      correctedTechStepKey: correction.correctedTechStep?.key ?? null,
    }));
  } catch (err) {
    throw err; // see recipe.service.ts's equivalent catch comment
  }
}

/**
 * Persists a batch of `TechStepTrainingSuggestion`s and, for every
 * suggestion sourced from a correction, stamps that correction's
 * `consumedAt` in the same transaction — so a worker run that crashes
 * partway through never leaves a correction consumed with no matching
 * suggestion, or a suggestion created against a correction still (wrongly)
 * eligible to be picked up again by the next run.
 *
 * @throws {HttpError} `404 TECH_STEP_NOT_FOUND` if any `techStepKey` in the
 *   batch doesn't match a reference `TechStep` — rejects the *whole* batch
 *   rather than skipping the bad entries, on the theory that a worker
 *   sending an unknown key is more likely a version-skew bug (its own
 *   taxonomy copy, `services/tech-step-llm-worker/src/tech-step-taxonomy.ts`,
 *   drifting from this API's `TechStep` catalog) than a one-off it should
 *   silently tolerate.
 */
export async function submitTrainingSuggestions(
  input: SubmitTrainingSuggestionsInput,
): Promise<{ created: number }> {
  try {
    const techStepKeys = [
      ...new Set(input.suggestions.map((suggestion) => suggestion.techStepKey)),
    ];
    const techSteps = await prisma.techStep.findMany({
      where: { key: { in: techStepKeys } },
      select: { id: true, key: true },
    });
    const techStepIdByKey = new Map(techSteps.map((techStep) => [techStep.key, techStep.id]));
    const missingKeys = techStepKeys.filter((key) => !techStepIdByKey.has(key));
    if (missingKeys.length > 0) {
      throw new HttpError(
        404,
        ErrorCode.TECH_STEP_NOT_FOUND,
        `Unknown techStepKey(s): ${missingKeys.join(", ")}`,
      );
    }

    await prisma.$transaction(async (tx) => {
      for (const suggestion of input.suggestions) {
        // Non-null by construction — every key in `input.suggestions` was
        // just confirmed present in `techStepIdByKey` above (the `missingKeys`
        // check would have thrown otherwise).
        const techStepId = techStepIdByKey.get(suggestion.techStepKey);
        if (techStepId === undefined) continue;

        await tx.techStepTrainingSuggestion.create({
          data: {
            techStepId,
            locale: suggestion.locale,
            suggestedSynonyms: suggestion.suggestedSynonyms,
            suggestedUtterances: suggestion.suggestedUtterances,
            sourceType: suggestion.sourceType,
            sourceCorrectionId: suggestion.sourceCorrectionId ?? null,
          },
        });

        if (suggestion.sourceCorrectionId !== null && suggestion.sourceCorrectionId !== undefined) {
          await tx.stepTechStepCorrection.update({
            where: { id: suggestion.sourceCorrectionId },
            data: { consumedAt: new Date() },
          });
        }
      }
    });

    return { created: input.suggestions.length };
  } catch (err) {
    throw err; // see recipe.service.ts's equivalent catch comment
  }
}

/**
 * The one worker with a `WorkerHeartbeat` row today — a fixed key, not
 * something the caller supplies (only one worker exists, and letting it
 * name itself would just be a spoofing surface behind the same shared
 * secret).
 */
const WORKER_KEY = "tech-step-llm-worker";

/**
 * Upserts `services/tech-step-llm-worker`'s heartbeat row (see
 * `POST /internal/tech-steps/heartbeat`). Every ping bumps `lastSeenAt`; a
 * `"job"` ping also records `lastRunAt` + a small `lastResult` summary so
 * the admin monitoring board can show what the worker last did and whether
 * it worked.
 */
export async function recordWorkerHeartbeat(input: WorkerHeartbeatInput): Promise<void> {
  try {
    const now = new Date();
    const jobResult: Prisma.InputJsonValue | undefined =
      input.event === "job"
        ? { job: input.job ?? null, ok: input.ok ?? null, counts: input.counts ?? {} }
        : undefined;

    await prisma.workerHeartbeat.upsert({
      where: { workerKey: WORKER_KEY },
      create: {
        workerKey: WORKER_KEY,
        lastSeenAt: now,
        lastRunAt: input.event === "job" ? now : null,
        lastResult: jobResult,
      },
      update: {
        lastSeenAt: now,
        ...(input.event === "job" ? { lastRunAt: now, lastResult: jobResult } : {}),
      },
    });
  } catch (err) {
    throw err; // see recipe.service.ts's equivalent catch comment
  }
}