/** * 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( '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(); 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;