Curio/jobs/regenerate-blueprint-job.ts

/**
 * BullMQ worker: regenerate all lessons for a stale blueprint at a new contentVersion.
 *
 * Idempotent: idempotency key encodes `blueprintId:targetContentVersion` so
 * retries never double-bill. Serves old content until regeneration completes.
 */
import { Worker } from 'bullmq';
import { eq, and } from 'drizzle-orm';
import { db } from '../src/lib/db';
import { jobs, blueprints, concepts, lessons } from '../src/lib/db/schema';
import { generateLesson } from '../src/lib/generation/generate-lesson';
import { verifyLesson } from '../src/lib/verification/verify-content';
import { clearBlueprintStale } from '../src/lib/generation/staleness';
import { getLessonResponse } from '../src/lib/db/queries';
import { setCachedLesson } from '../src/lib/cache/lesson';
import { enqueuePromoteBlueprint } from '../src/lib/jobs/queue';
import type { RegenerateBlueprintJobData } from '../src/lib/jobs/queue';

const connection = {
  url: process.env.REDIS_URL ?? 'redis://localhost:6379',
};

const worker = new Worker<RegenerateBlueprintJobData>(
  'regenerate-blueprint',
  async (job) => {
    const { blueprintId, intentKey, targetContentVersion, idempotencyKey } = job.data;

    // 1. Idempotency check
    const [jobRow] = await db
      .select({ id: jobs.id, status: jobs.status, attempts: jobs.attempts })
      .from(jobs)
      .where(eq(jobs.id, idempotencyKey))
      .limit(1);

    if (jobRow?.status === 'done') return;

    await db
      .update(jobs)
      .set({ status: 'running', attempts: (jobRow?.attempts ?? 0) + 1 })
      .where(eq(jobs.id, idempotencyKey));

    // 2. Verify the blueprint still needs this version (may have been superseded)
    const [bp] = await db
      .select({ id: blueprints.id, contentVersion: blueprints.contentVersion, title: blueprints.title })
      .from(blueprints)
      .where(eq(blueprints.id, blueprintId))
      .limit(1);

    if (!bp) throw new Error(`Blueprint not found: ${blueprintId}`);

    // If the blueprint has already been rebuilt at or past targetContentVersion, skip.
    if (bp.contentVersion > targetContentVersion) {
      await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));
      return;
    }

    // 3. Load concepts
    const conceptRows = await db
      .select({ id: concepts.id, name: concepts.name, ord: concepts.ord, retrievalCtxRef: concepts.retrievalCtxRef })
      .from(concepts)
      .where(eq(concepts.blueprintId, blueprintId));

    if (conceptRows.length === 0) throw new Error(`No concepts for blueprint: ${blueprintId}`);

    // 4. Find all unique (locale, depth, ageGroup) variants already generated for this blueprint
    //    so we regenerate the same set.
    const existingVariants = await db
      .select({ locale: lessons.locale, depth: lessons.depth, ageGroup: lessons.ageGroup })
      .from(lessons)
      .where(eq(lessons.blueprintId, blueprintId));

    // Dedupe variants; always include default.
    const variantSet = new Map<string, { locale: string; depth: string; ageGroup: string }>();
    variantSet.set('en:standard:adult', { locale: 'en', depth: 'standard', ageGroup: 'adult' });
    for (const v of existingVariants) {
      variantSet.set(`${v.locale}:${v.depth}:${v.ageGroup}`, v);
    }

    // 5. Regenerate each variant
    for (const variant of variantSet.values()) {
      const { locale, depth, ageGroup } = variant;
      const { lessonId } = await generateLesson({
        blueprintId,
        lessonOrd: 0,
        topicTitle: bp.title,
        conceptsToGenerate: conceptRows.map((c) => ({
          id: c.id,
          name: c.name,
          retrievalCtxRef: c.retrievalCtxRef,
        })),
        locale,
        depth: depth as import('../src/lib/generation/depth').LessonDepth,
        ageGroup: ageGroup as import('../src/schemas/preferences').AgeGroup,
      });

      // T1 verify (T2 handled by promote job)
      await verifyLesson({ lessonId, runT2: false });

      // Update cache
      const lesson = await getLessonResponse(
        intentKey, undefined, locale,
        depth as import('../src/lib/generation/depth').LessonDepth,
        ageGroup as import('../src/schemas/preferences').AgeGroup,
      );
      if (lesson?.state === 'ready') {
        await setCachedLesson(
          intentKey, locale, lesson,
          depth as import('../src/lib/generation/depth').LessonDepth,
          ageGroup as import('../src/schemas/preferences').AgeGroup,
        );
      }
    }

    // 6. Clear stale flag + stamp current signature
    await clearBlueprintStale(blueprintId);

    // 7. Bump contentVersion to targetContentVersion (done after generation succeeds)
    await db
      .update(blueprints)
      .set({ contentVersion: targetContentVersion })
      .where(and(eq(blueprints.id, blueprintId)));

    // 8. Mark job done
    await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));

    // 9. Enqueue promotion (T2 + misconceptions)
    const promoteKey = `promote-blueprint:${blueprintId}:v${targetContentVersion}`;
    const [existingPromote] = await db
      .select({ id: jobs.id })
      .from(jobs)
      .where(eq(jobs.id, promoteKey))
      .limit(1);

    if (!existingPromote) {
      const [promoteJobRow] = await db
        .insert(jobs)
        .values({
          type: 'promote-blueprint',
          idempotencyKey: promoteKey,
          status: 'pending',
          payloadJson: { blueprintId, intentKey },
        })
        .returning({ id: jobs.id });

      await enqueuePromoteBlueprint({
        blueprintId,
        intentKey,
        idempotencyKey: promoteJobRow.id,
      });
    }
  },
  { connection, concurrency: 1 },
);

worker.on('failed', async (job, err) => {
  if (job?.data.idempotencyKey) {
    await db
      .update(jobs)
      .set({ status: 'failed' })
      .where(eq(jobs.id, job.data.idempotencyKey));
  }
  console.error('[regenerate-blueprint] job failed:', err);
});

export default worker;