feat: add spreadsheet async imports

apps/worker/package.json

@@ -10,7 +10,8 @@
     "check": "tsc -p tsconfig.json --noEmit",
     "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"
+    "assets:once": "tsx src/index.ts --once --job assets",
+    "imports:once": "tsx src/index.ts --once --job imports"
   },
   "dependencies": {
     "@supabase/storage-js": "^2.108.2",

apps/worker/src/config.ts

@@ -17,6 +17,9 @@ export interface WorkerConfig {
   assetMinAgeSeconds: number;
   assetRecheckIntervalSeconds: number;
   assetRequestTimeoutMs: number;
+  importBatchSize: number;
+  importWorkerId: string;
+  importBackoffSeconds: number[];
   storageMaxUploadBytes: number;
   storageAllowedMimePrefixes: string[];
   storageAllowedMimeTypes: string[];
@@ -53,6 +56,11 @@ export const config: WorkerConfig = {
   assetMinAgeSeconds: envNumber('WORKER_ASSET_MIN_AGE_SECONDS', 300),
   assetRecheckIntervalSeconds: envNumber('WORKER_ASSET_RECHECK_INTERVAL_SECONDS', 60 * 60 * 24),
   assetRequestTimeoutMs: envNumber('WORKER_ASSET_REQUEST_TIMEOUT_MS', 10_000),
+  importBatchSize: envNumber('WORKER_IMPORT_BATCH_SIZE', 5),
+  importWorkerId: envString('WORKER_IMPORT_ID', `imports-${process.pid}`),
+  importBackoffSeconds: envList('WORKER_IMPORT_BACKOFF_SECONDS', '30,120,600,1800')
+    .map((value: string) => Number(value))
+    .filter((value: number) => Number.isFinite(value) && value > 0),
   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

@@ -4,6 +4,8 @@ import { processCrmBatch } from './jobs/crm.js';
 import { processCommerceBatch } from './jobs/commerce.js';
 import { processAssetBatch } from './jobs/assets.js';
 
+const extraClosers = new Set<() => Promise<void>>();
+
 function hasArg(name: string) {
   return process.argv.includes(name);
 }
@@ -37,6 +39,17 @@ async function runOnce() {
     );
     return;
   }
+  if (job === 'imports') {
+    const { closeImportExecutorPool, processImportBatch } = await import('./jobs/imports.js');
+    extraClosers.add(closeImportExecutorPool);
+    const result = await processImportBatch();
+    console.log(
+      `[worker] imports batch processed=${result.processed}`
+        + ` completed=${result.completed} completedWithErrors=${result.completedWithErrors}`
+        + ` failed=${result.failed} retrying=${result.retrying} skipped=${result.skipped}`,
+    );
+    return;
+  }
   throw new Error(`Unsupported worker job: ${job}`);
 }
 
@@ -66,5 +79,8 @@ try {
     await runOnce();
   }
 } finally {
+  for (const closeExtra of extraClosers) {
+    await closeExtra();
+  }
   await closePool();
 }

apps/worker/src/jobs/imports.ts

@@ -0,0 +1,221 @@
+import { pool } from '../db.js';
+import { config } from '../config.js';
+import { executeContentImportJob, type ExecutableContentImportType } from '../../../api/src/features/tenant-content/imports.js';
+import { closePool as closeApiImportPool } from '../../../api/src/core/db.js';
+
+interface ImportJobRow {
+  id: string;
+  tenantId: string;
+  createdBy: string | null;
+  importType: ExecutableContentImportType;
+  status: string;
+  attemptCount: number;
+  maxAttempts: number;
+  summary: Record<string, unknown>;
+}
+
+interface ImportWorkerResult {
+  processed: number;
+  completed: number;
+  completedWithErrors: number;
+  failed: number;
+  retrying: number;
+  skipped: number;
+}
+
+function objectValue(value: unknown): Record<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
+}
+
+function boolValue(value: unknown, fallback: boolean) {
+  return typeof value === 'boolean' ? value : fallback;
+}
+
+function numberValue(value: unknown, fallback: number) {
+  const parsed = Number(value ?? fallback);
+  return Number.isFinite(parsed) ? parsed : fallback;
+}
+
+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 || 'IMPORT_WORKER_ERROR')
+    : 'IMPORT_WORKER_ERROR';
+}
+
+function truncate(value: unknown, max = 1900) {
+  return String(value ?? '').slice(0, max);
+}
+
+function backoffSeconds(attemptCount: number) {
+  const backoffs = config.importBackoffSeconds.length ? config.importBackoffSeconds : [30, 120, 600, 1800];
+  return backoffs[Math.min(Math.max(0, attemptCount - 1), backoffs.length - 1)];
+}
+
+function importOptions(summary: Record<string, unknown>) {
+  const options = objectValue(summary.importOptions);
+  return {
+    allowPartial: boolValue(options.allowPartial, false),
+  };
+}
+
+async function claimImportJobs() {
+  const client = await pool.connect();
+  try {
+    await client.query('begin');
+    const result = await client.query<ImportJobRow>(
+      `
+      select id,
+             tenant_id as "tenantId",
+             created_by as "createdBy",
+             import_type as "importType",
+             status,
+             attempt_count as "attemptCount",
+             max_attempts as "maxAttempts",
+             summary
+      from public.content_import_jobs
+      where execution_mode = 'async'
+        and status = 'pending'
+        and attempt_count < max_attempts
+        and (next_attempt_at is null or next_attempt_at <= now())
+      order by created_at asc
+      limit $1
+      for update skip locked
+      `,
+      [config.importBatchSize],
+    );
+
+    const ids = result.rows.map(row => row.id);
+    if (ids.length > 0) {
+      await client.query(
+        `
+        update public.content_import_jobs
+        set locked_at = now(),
+            locked_by = $2,
+            attempt_count = attempt_count + 1,
+            updated_at = now()
+        where id = any($1::uuid[])
+        `,
+        [ids, config.importWorkerId],
+      );
+    }
+    await client.query('commit');
+    return result.rows;
+  } catch (error) {
+    await client.query('rollback');
+    throw error;
+  } finally {
+    client.release();
+  }
+}
+
+async function markImportFailed(job: ImportJobRow, error: unknown) {
+  const nextAttempt = job.attemptCount + 1;
+  const willRetry = nextAttempt < job.maxAttempts;
+  const status = willRetry ? 'pending' : 'failed';
+  await pool.query(
+    `
+    update public.content_import_jobs
+    set status = $3,
+        error_message = $4,
+        summary = coalesce(summary, '{}'::jsonb) || $5::jsonb,
+        next_attempt_at = case when $6::boolean then now() + make_interval(secs => $7::integer) else null end,
+        locked_at = null,
+        locked_by = null,
+        finished_at = case when $3 = 'failed' then now() else finished_at end,
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [
+      job.tenantId,
+      job.id,
+      status,
+      truncate(errorMessage(error)),
+      JSON.stringify({
+        lastWorkerError: {
+          code: errorCode(error),
+          message: truncate(errorMessage(error)),
+          workerId: config.importWorkerId,
+          failedAt: new Date().toISOString(),
+          nextAttempt,
+          maxAttempts: job.maxAttempts,
+          willRetry,
+        },
+      }),
+      willRetry,
+      backoffSeconds(nextAttempt),
+    ],
+  );
+
+  await pool.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, $2, $3, 'content_import_job', $4, $5::jsonb)
+    `,
+    [
+      job.tenantId,
+      job.createdBy,
+      willRetry ? `content.import.${job.importType}.retry_scheduled` : `content.import.${job.importType}.failed`,
+      job.id,
+      JSON.stringify({
+        code: errorCode(error),
+        message: truncate(errorMessage(error)),
+        workerId: config.importWorkerId,
+        nextAttempt,
+        maxAttempts: job.maxAttempts,
+      }),
+    ],
+  );
+
+  return willRetry ? 'retrying' : 'failed';
+}
+
+export async function processImportBatch(): Promise<ImportWorkerResult> {
+  const jobs = await claimImportJobs();
+  const result: ImportWorkerResult = {
+    processed: jobs.length,
+    completed: 0,
+    completedWithErrors: 0,
+    failed: 0,
+    retrying: 0,
+    skipped: 0,
+  };
+
+  for (const job of jobs) {
+    try {
+      const execution = await executeContentImportJob(
+        {
+          tenantId: job.tenantId,
+          userId: job.createdBy || job.tenantId,
+          role: 'system_worker',
+          permissions: { 'content:*': true },
+          templatePermissions: {},
+        },
+        {
+          jobId: job.id,
+          importType: job.importType,
+          allowPartial: importOptions(job.summary).allowPartial,
+          allowQueuedJob: true,
+        },
+      );
+
+      if (execution.idempotent) result.skipped += 1;
+      else if (execution.status === 'completed_with_errors') result.completedWithErrors += 1;
+      else if (execution.status === 'completed') result.completed += 1;
+      else result.skipped += 1;
+    } catch (error) {
+      const state = await markImportFailed(job, error);
+      if (state === 'retrying') result.retrying += 1;
+      else result.failed += 1;
+    }
+  }
+
+  return result;
+}
+
+export async function closeImportExecutorPool() {
+  await closeApiImportPool();
+}