feat: add public question bank sync worker

apps/worker/package.json

@@ -11,7 +11,8 @@
     "crm:once": "tsx src/index.ts --once --job crm",
     "commerce:once": "tsx src/index.ts --once --job commerce",
     "assets:once": "tsx src/index.ts --once --job assets",
-    "imports:once": "tsx src/index.ts --once --job imports"
+    "imports:once": "tsx src/index.ts --once --job imports",
+    "public-banks:once": "tsx src/index.ts --once --job public-banks"
   },
   "dependencies": {
     "@supabase/storage-js": "^2.108.2",

apps/worker/src/config.ts

@@ -20,6 +20,10 @@ export interface WorkerConfig {
   importBatchSize: number;
   importWorkerId: string;
   importBackoffSeconds: number[];
+  publicBankSyncBatchSize: number;
+  publicBankSyncCopyLimit: number;
+  publicBankSyncWorkerId: string;
+  publicBankSyncClaimStaleSeconds: number;
   storageMaxUploadBytes: number;
   storageAllowedMimePrefixes: string[];
   storageAllowedMimeTypes: string[];
@@ -61,6 +65,10 @@ export const config: WorkerConfig = {
   importBackoffSeconds: envList('WORKER_IMPORT_BACKOFF_SECONDS', '30,120,600,1800')
     .map((value: string) => Number(value))
     .filter((value: number) => Number.isFinite(value) && value > 0),
+  publicBankSyncBatchSize: envNumber('WORKER_PUBLIC_BANK_SYNC_BATCH_SIZE', 5),
+  publicBankSyncCopyLimit: envNumber('WORKER_PUBLIC_BANK_SYNC_COPY_LIMIT', 1000),
+  publicBankSyncWorkerId: envString('WORKER_PUBLIC_BANK_SYNC_ID', `public-banks-${process.pid}`),
+  publicBankSyncClaimStaleSeconds: envNumber('WORKER_PUBLIC_BANK_SYNC_CLAIM_STALE_SECONDS', 15 * 60),
   storageMaxUploadBytes: envNumber('STORAGE_MAX_UPLOAD_BYTES', 1024 * 1024 * 500),
   storageAllowedMimePrefixes: envList('STORAGE_ALLOWED_MIME_PREFIXES', 'image/,video/,audio/'),
   storageAllowedMimeTypes: envList(

apps/worker/src/index.ts

@@ -50,6 +50,17 @@ async function runOnce() {
     );
     return;
   }
+  if (job === 'public-banks') {
+    const { closePublicBankSyncExecutorPool, processPublicBankSyncBatch } = await import('./jobs/public-banks.js');
+    extraClosers.add(closePublicBankSyncExecutorPool);
+    const result = await processPublicBankSyncBatch();
+    console.log(
+      `[worker] public-banks batch processed=${result.processed}`
+        + ` synced=${result.synced} conflicts=${result.conflicts}`
+        + ` failed=${result.failed} skipped=${result.skipped}`,
+    );
+    return;
+  }
   throw new Error(`Unsupported worker job: ${job}`);
 }
 

apps/worker/src/jobs/public-banks.ts

@@ -0,0 +1,197 @@
+import crypto from 'node:crypto';
+import { pool } from '../db.js';
+import { config } from '../config.js';
+import {
+  executePublicQuestionBankSync,
+} from '../../../api/src/features/tenant-content/public-banks.js';
+import { closePool as closeApiPublicBankPool } from '../../../api/src/core/db.js';
+
+interface PublicBankSyncCandidate {
+  id: string;
+  tenantId: string;
+  createdBy: string | null;
+  updatedBy: string | null;
+}
+
+interface PublicBankSyncWorkerResult {
+  processed: number;
+  synced: number;
+  conflicts: number;
+  failed: number;
+  skipped: number;
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function errorMessage(error: unknown) {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function errorCode(error: unknown) {
+  return typeof error === 'object' && error !== null && 'code' in error
+    ? String((error as { code?: unknown }).code || 'PUBLIC_BANK_SYNC_WORKER_ERROR')
+    : 'PUBLIC_BANK_SYNC_WORKER_ERROR';
+}
+
+function truncate(value: unknown, max = 1900) {
+  return String(value ?? '').slice(0, max);
+}
+
+async function claimPublicBankSyncCandidates(limit: number, claimId: string) {
+  const client = await pool.connect();
+  try {
+    await client.query('begin');
+    const result = await client.query<PublicBankSyncCandidate>(
+      `
+      with candidates as (
+        select a.id
+        from public.tenant_question_bank_adoptions a
+        join public.question_banks qb on qb.id = a.source_question_bank_id
+        left join public.question_bank_grants g on g.id = a.grant_id
+        where a.status in ('active', 'sync_pending')
+          and (a.sync_status <> 'failed' or a.status = 'sync_pending')
+          and a.grant_id is not null
+          and a.target_question_bank_id is not null
+          and a.target_entry_id is not null
+          and a.target_collection_id is not null
+          and qb.source_scope = 'platform'
+          and qb.status = 'active'
+          and (
+            a.sync_status = 'pending'
+            or a.status = 'sync_pending'
+            or a.last_synced_at is null
+            or qb.updated_at > coalesce(a.last_synced_at, '1970-01-01'::timestamptz)
+            or coalesce(g.updated_at, '1970-01-01'::timestamptz) > coalesce(a.last_synced_at, '1970-01-01'::timestamptz)
+            or exists (
+              select 1
+              from public.questions q
+              where q.tenant_id = qb.tenant_id
+                and q.question_bank_id = qb.id
+                and q.status = 'published'
+                and q.updated_at > coalesce(a.last_synced_at, '1970-01-01'::timestamptz)
+            )
+          )
+          and (
+            a.status <> 'sync_pending'
+            or coalesce(nullif(a.metadata #>> '{publicBankSyncWorker,claimedAt}', '')::timestamptz, '1970-01-01'::timestamptz)
+              <= now() - ($2::integer * interval '1 second')
+          )
+        order by coalesce(a.last_synced_at, '1970-01-01'::timestamptz) asc, a.updated_at asc
+        limit $1
+        for update of a skip locked
+      )
+      update public.tenant_question_bank_adoptions a
+      set status = 'sync_pending',
+          sync_status = 'pending',
+          metadata = jsonb_set(
+            coalesce(a.metadata, '{}'::jsonb),
+            '{publicBankSyncWorker}',
+            coalesce(a.metadata->'publicBankSyncWorker', '{}'::jsonb) || $3::jsonb,
+            true
+          ),
+          updated_at = now()
+      from candidates
+      where a.id = candidates.id
+      returning a.id,
+                a.tenant_id as "tenantId",
+                a.created_by as "createdBy",
+                a.updated_by as "updatedBy"
+      `,
+      [
+        limit,
+        Math.max(60, config.publicBankSyncClaimStaleSeconds),
+        JSON.stringify({
+          claimId,
+          workerId: config.publicBankSyncWorkerId,
+          claimedAt: nowIso(),
+        }),
+      ],
+    );
+    await client.query('commit');
+    return result.rows;
+  } catch (error) {
+    await client.query('rollback');
+    throw error;
+  } finally {
+    client.release();
+  }
+}
+
+async function markPublicBankSyncFailed(candidate: PublicBankSyncCandidate, error: unknown) {
+  const details = {
+    code: errorCode(error),
+    message: truncate(errorMessage(error)),
+    workerId: config.publicBankSyncWorkerId,
+    failedAt: nowIso(),
+  };
+  await pool.query(
+    `
+    update public.tenant_question_bank_adoptions
+    set status = 'active',
+        sync_status = 'failed',
+        metadata = jsonb_set(
+          coalesce(metadata, '{}'::jsonb),
+          '{publicBankSyncWorker}',
+          coalesce(metadata->'publicBankSyncWorker', '{}'::jsonb) || $3::jsonb,
+          true
+        ),
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [
+      candidate.tenantId,
+      candidate.id,
+      JSON.stringify({
+        lastStatus: 'failed',
+        lastError: details,
+        lastWorkerId: config.publicBankSyncWorkerId,
+        lastFinishedAt: nowIso(),
+      }),
+    ],
+  );
+  await pool.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, null, 'content.public_question_bank.sync_worker_failed', 'tenant_question_bank_adoption', $2, $3::jsonb)
+    `,
+    [candidate.tenantId, candidate.id, JSON.stringify(details)],
+  );
+}
+
+export async function processPublicBankSyncBatch(limit = config.publicBankSyncBatchSize): Promise<PublicBankSyncWorkerResult> {
+  const candidates = await claimPublicBankSyncCandidates(limit, crypto.randomUUID());
+  const result: PublicBankSyncWorkerResult = {
+    processed: candidates.length,
+    synced: 0,
+    conflicts: 0,
+    failed: 0,
+    skipped: 0,
+  };
+
+  for (const candidate of candidates) {
+    try {
+      const sync = await executePublicQuestionBankSync({
+        tenantId: candidate.tenantId,
+        adoptionId: candidate.id,
+        actorUserId: candidate.updatedBy || candidate.createdBy || null,
+        copyLimit: config.publicBankSyncCopyLimit,
+        triggeredBy: 'worker',
+        workerId: config.publicBankSyncWorkerId,
+      });
+      if (sync.sync.status === 'conflict') result.conflicts += 1;
+      else if (sync.sync.counts.inserted || sync.sync.counts.updated || sync.sync.counts.skipped) result.synced += 1;
+      else result.skipped += 1;
+    } catch (error) {
+      await markPublicBankSyncFailed(candidate, error);
+      result.failed += 1;
+    }
+  }
+
+  return result;
+}
+
+export async function closePublicBankSyncExecutorPool() {
+  await closeApiPublicBankPool();
+}