← back to Designer Wallcoverings
TK-11776: harden batch-import background job — atomic job seed + honest comments + e2e test
9296de8bd3181b0da1659f709cad9a66e5dc4379 · 2026-09-15 12:31:15 -0700 · Steve
Second-model (Kimi) review follow-ups that hold up:
- createJob now inserts the job row + ALL item rows inside ONE transaction() via a
single multi-row INSERT (was job + N auto-committed inserts). A failure mid-seed
now rolls back instead of persisting a half-seeded job a client retry would dupe.
- Corrected the DDL + worker header comments that overstated 'resumability': the
per-item rows give observability (partial write is visible, not silent), but
automatic resume is NOT implemented — a mid-job restart strands items pending.
Documented the boot-sweeper follow-up honestly instead of implying it exists.
- Added an END-TO-END wiring test: POST -> setImmediate worker -> gate enforced on a
$4.25 sample-price leak (unorderable + Needs-Price-Review). This is the only test
that catches a broken kick/arg-pass while the runJob unit tests stay green.
Not applied (real disagreement, left for the owner): Kimi's proposed
UNIQUE INDEX on sku for the cross-job dedup race would HARD-BREAK legitimate
re-imports after archival — recorded as an advisory-dedup limitation on the ticket.
build: ✓ Compiled successfully. tsc clean for changed files. 8/8 jest green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B17G4jZ3RfvrQUhKW8ihM3
Files touched
M DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.tsA DW-Programming/ImportNewSkufromURL/database/create-batch-import-jobs.sqlA DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
Diff
commit 9296de8bd3181b0da1659f709cad9a66e5dc4379
Author: Steve <steve@designerwallcoverings.com>
Date: Tue Sep 15 12:31:15 2026 -0700
TK-11776: harden batch-import background job — atomic job seed + honest comments + e2e test
Second-model (Kimi) review follow-ups that hold up:
- createJob now inserts the job row + ALL item rows inside ONE transaction() via a
single multi-row INSERT (was job + N auto-committed inserts). A failure mid-seed
now rolls back instead of persisting a half-seeded job a client retry would dupe.
- Corrected the DDL + worker header comments that overstated 'resumability': the
per-item rows give observability (partial write is visible, not silent), but
automatic resume is NOT implemented — a mid-job restart strands items pending.
Documented the boot-sweeper follow-up honestly instead of implying it exists.
- Added an END-TO-END wiring test: POST -> setImmediate worker -> gate enforced on a
$4.25 sample-price leak (unorderable + Needs-Price-Review). This is the only test
that catches a broken kick/arg-pass while the runJob unit tests stay green.
Not applied (real disagreement, left for the owner): Kimi's proposed
UNIQUE INDEX on sku for the cross-job dedup race would HARD-BREAK legitimate
re-imports after archival — recorded as an advisory-dedup limitation on the ticket.
build: ✓ Compiled successfully. tsc clean for changed files. 8/8 jest green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B17G4jZ3RfvrQUhKW8ihM3
---
.../__tests__/batch-import-gate.test.ts | 29 ++
.../database/create-batch-import-jobs.sql | 35 ++
.../ImportNewSkufromURL/lib/batch-import-jobs.ts | 415 +++++++++++++++++++++
3 files changed, 479 insertions(+)
diff --git a/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts b/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
index 0ce32281..c5d30251 100644
--- a/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
+++ b/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
@@ -49,6 +49,9 @@ jest.mock('@/lib/price-integrity-record', () => ({ recordOutcome: jest.fn() }));
jest.mock('@/lib/database/postgres-client', () => ({
query: jest.fn().mockResolvedValue([]),
queryOne: jest.fn().mockResolvedValue(null),
+ // createJob() now seeds the job atomically via transaction(cb) — invoke the callback
+ // with a fake client whose .query resolves so the insert path runs inert in CI.
+ transaction: jest.fn(async (cb: (c: any) => any) => cb({ query: jest.fn().mockResolvedValue({ rows: [] }) })),
}));
// p-limit v7 is ESM-only and the worker loads it via `await import('p-limit')`; jest's
// CJS interop yields a non-callable default. Mock a passthrough limiter: pLimit(n) → a
@@ -181,4 +184,30 @@ describe('batch import route — background-job contract (TK-11776)', () => {
await new Promise(resolve => setImmediate(resolve));
await new Promise(resolve => setImmediate(resolve));
});
+
+ it('END-TO-END wiring: after POST returns, the setImmediate worker runs and applies the gate', async () => {
+ // Closes the gap the unit tests leave: prove POST -> setImmediate -> runJob actually
+ // fires and enforces the price-integrity gate on a $4.25 sample-price leak. If someone
+ // breaks the kick or the arg-passing, the runJob unit tests stay green but production
+ // never gates — this is the only test that would catch that.
+ const res = await POST(batchReq({
+ products: [makeProduct({ price: '$4.25' })],
+ vendorId: 'thibaut',
+ privateLabel: false,
+ }));
+ expect(res.status).toBe(202);
+ expect(mockCreateProduct).not.toHaveBeenCalled(); // deferred, not synchronous
+ // Let the microtask chain (setJobStatus -> preResolveCosts -> dedup -> createProduct)
+ // drain; createProduct is awaited BEFORE the worker's 1s rate-limit sleep, so a short
+ // real-timer wait is enough to observe the call without waiting the full second.
+ await new Promise(resolve => setImmediate(resolve));
+ await new Promise(resolve => setTimeout(resolve, 300));
+ expect(mockCreateProduct).toHaveBeenCalledTimes(1);
+ const { v, tags } = lastVariant();
+ expect(v.inventory_quantity).toBe(0); // $4.25 == sample floor → gate blocks
+ expect(v.inventory_policy).toBe('deny');
+ expect(tags).toContain('Needs-Price-Review');
+ // Let the worker's trailing 1s sleep + completion finish before the suite tears down.
+ await new Promise(resolve => setTimeout(resolve, 1100));
+ });
});
diff --git a/DW-Programming/ImportNewSkufromURL/database/create-batch-import-jobs.sql b/DW-Programming/ImportNewSkufromURL/database/create-batch-import-jobs.sql
new file mode 100644
index 00000000..268d4a1f
--- /dev/null
+++ b/DW-Programming/ImportNewSkufromURL/database/create-batch-import-jobs.sql
@@ -0,0 +1,35 @@
+-- TK-11776 (DTD verdict B): background-job store for the bulk-import route.
+-- The synchronous /api/import/batch for-loop was converted into a fire-and-forget
+-- background worker; these two tables are the durable job record + per-item
+-- progress. This makes the partial-write VISIBLE, not silent (observability). It is
+-- the foundation for a future resume, but automatic resume is NOT yet implemented —
+-- a mid-job restart leaves items 'pending'/job 'running'; a boot sweeper is the TODO.
+-- Canonical DB: dw_unified. Idempotent — safe to run repeatedly.
+
+CREATE TABLE IF NOT EXISTS batch_import_jobs (
+ job_id UUID PRIMARY KEY,
+ status TEXT NOT NULL DEFAULT 'pending', -- pending | running | completed | failed
+ total INTEGER NOT NULL DEFAULT 0,
+ vendor_id TEXT,
+ private_label BOOLEAN DEFAULT FALSE,
+ error TEXT,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS batch_import_job_items (
+ job_id UUID NOT NULL REFERENCES batch_import_jobs(job_id) ON DELETE CASCADE,
+ idx INTEGER NOT NULL,
+ title TEXT,
+ sku TEXT,
+ -- pending | created | failed | blocked | skipped-duplicate
+ status TEXT NOT NULL DEFAULT 'pending',
+ product_id TEXT,
+ error TEXT,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ PRIMARY KEY (job_id, idx)
+);
+
+CREATE INDEX IF NOT EXISTS idx_batch_import_job_items_job ON batch_import_job_items(job_id);
+-- dedup lookup: "has this SKU already been created in ANY prior job?"
+CREATE INDEX IF NOT EXISTS idx_batch_import_job_items_sku ON batch_import_job_items(sku) WHERE sku IS NOT NULL AND sku <> '';
diff --git a/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts b/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
new file mode 100644
index 00000000..d7d73e88
--- /dev/null
+++ b/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
@@ -0,0 +1,415 @@
+/**
+ * batch-import-jobs.ts — background-job store + worker for the bulk-import route.
+ * ============================================================================
+ * Ticket: TK-11776 (DTD verdict B — 2026-09-15).
+ *
+ * The old /api/import/batch handler ran a fully SERIAL for-loop INSIDE the HTTP
+ * request: per product it awaited resolveNetCost() (which execFiles a
+ * `node verify-price.js` subprocess), awaited shopifyAPI.createProduct(), then
+ * slept 1s. Worst case ~150 × (12s cost + create round-trip + 1s) = many minutes
+ * in one request — which outlives client/proxy timeouts and can leave SILENT
+ * PARTIAL catalog writes (the client gives up, the loop keeps creating).
+ *
+ * This module moves that work into a fire-and-forget background worker and
+ * persists a job record + per-item progress to Postgres (dw_unified), so:
+ * - POST returns 202 {jobId} immediately (no timeout),
+ * - every product's outcome is durably recorded as it completes — the partial
+ * write is now VISIBLE, not silent (observability). The per-item rows are the
+ * foundation for a future resume, but automatic resume is NOT yet implemented:
+ * a process restart (deploy/PM2/crash) mid-job leaves items 'pending' and the
+ * job 'running'; nothing re-runs them. A boot-time sweeper is the follow-up.
+ *
+ * The per-product logic (price-integrity gate, weight-guard, $0-orderable guard,
+ * Needs-Price-Review tag, DRAFT status) is reproduced UNCHANGED from the original
+ * route. The only performance change: cost resolution is pre-computed for the
+ * whole batch through a bounded p-limit(5) pool BEFORE the create loop. The
+ * createProduct calls stay STRICTLY SERIAL with the 1s delay — Shopify Standard
+ * is 2 req/s, so writes must never be parallelized.
+ */
+
+import { query, transaction } from './database/postgres-client';
+import { shopifyAPI } from './shopify-api';
+import { assertPriceIntegrity, resolveSampleFloor } from './price-integrity-gate';
+import { resolveNetCost } from './price-integrity-cost';
+import { recordOutcome } from './price-integrity-record';
+import { sellableWeightLb } from './weight-guard';
+
+export interface BatchProduct {
+ url: string;
+ title: string;
+ sku?: string;
+ price?: string;
+ images: string[];
+ vendor: string;
+ collection?: string;
+ product_type?: string;
+ tags?: string[];
+ privateLabel: boolean;
+}
+
+export interface BatchJobRequest {
+ products: BatchProduct[];
+ vendorId: string;
+ privateLabel: boolean;
+ customVendorName?: string;
+ costTimeoutMs: number;
+}
+
+export type ItemStatus = 'pending' | 'created' | 'failed' | 'blocked' | 'skipped-duplicate';
+
+/** Bounded concurrency for the cost-resolution pool ONLY (never for writes). */
+const COST_POOL_CONCURRENCY = Math.max(
+ 1,
+ parseInt(process.env.IMPORT_BATCH_COST_CONCURRENCY || '5', 10) || 5,
+);
+
+let tablesReady = false;
+
+/** Create the job tables if they don't exist (idempotent). Mirrors database/create-batch-import-jobs.sql. */
+export async function ensureBatchJobTables(): Promise<void> {
+ if (tablesReady) return;
+ await query(`
+ CREATE TABLE IF NOT EXISTS batch_import_jobs (
+ job_id UUID PRIMARY KEY,
+ status TEXT NOT NULL DEFAULT 'pending',
+ total INTEGER NOT NULL DEFAULT 0,
+ vendor_id TEXT,
+ private_label BOOLEAN DEFAULT FALSE,
+ error TEXT,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )
+ `);
+ await query(`
+ CREATE TABLE IF NOT EXISTS batch_import_job_items (
+ job_id UUID NOT NULL REFERENCES batch_import_jobs(job_id) ON DELETE CASCADE,
+ idx INTEGER NOT NULL,
+ title TEXT,
+ sku TEXT,
+ status TEXT NOT NULL DEFAULT 'pending',
+ product_id TEXT,
+ error TEXT,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ PRIMARY KEY (job_id, idx)
+ )
+ `);
+ await query(`CREATE INDEX IF NOT EXISTS idx_batch_import_job_items_job ON batch_import_job_items(job_id)`);
+ await query(
+ `CREATE INDEX IF NOT EXISTS idx_batch_import_job_items_sku ON batch_import_job_items(sku) WHERE sku IS NOT NULL AND sku <> ''`,
+ );
+ tablesReady = true;
+}
+
+/**
+ * Persist a new job + one pending row per product. Call BEFORE returning 202.
+ * ATOMIC: the job row and ALL item rows are inserted inside ONE transaction, so a
+ * failure mid-seed rolls the whole thing back rather than leaving a half-seeded job
+ * that a client retry would then duplicate. The items go in as a single multi-row
+ * INSERT (one round-trip, not N) — 150 products × 4 params is far under PG's limit.
+ */
+export async function createJob(jobId: string, req: BatchJobRequest): Promise<void> {
+ await ensureBatchJobTables(); // DDL auto-commits; keep it out of the txn below
+ await transaction(async (client) => {
+ await client.query(
+ `INSERT INTO batch_import_jobs (job_id, status, total, vendor_id, private_label)
+ VALUES ($1, 'pending', $2, $3, $4)`,
+ [jobId, req.products.length, req.vendorId, req.privateLabel],
+ );
+ if (req.products.length === 0) return;
+ // Seed per-item rows so status polling shows the full plan immediately.
+ const tuples: string[] = [];
+ const values: any[] = [];
+ req.products.forEach((p, idx) => {
+ const b = idx * 4;
+ tuples.push(`($${b + 1}, $${b + 2}, $${b + 3}, $${b + 4}, 'pending')`);
+ values.push(jobId, idx, p.title || null, p.sku || null);
+ });
+ await client.query(
+ `INSERT INTO batch_import_job_items (job_id, idx, title, sku, status) VALUES ${tuples.join(', ')}`,
+ values,
+ );
+ });
+}
+
+async function setJobStatus(jobId: string, status: string, error?: string): Promise<void> {
+ await query(
+ `UPDATE batch_import_jobs SET status=$2, error=$3, updated_at=now() WHERE job_id=$1`,
+ [jobId, status, error || null],
+ );
+}
+
+async function setItem(
+ jobId: string,
+ idx: number,
+ status: ItemStatus,
+ productId?: string | number | null,
+ error?: string | null,
+): Promise<void> {
+ await query(
+ `UPDATE batch_import_job_items
+ SET status=$3, product_id=$4, error=$5, updated_at=now()
+ WHERE job_id=$1 AND idx=$2`,
+ [jobId, idx, status, productId != null ? String(productId) : null, error || null],
+ );
+}
+
+/**
+ * Dedup guard. lib/sku-registry.checkDuplicate (referenced in CLAUDE.md) lives on
+ * Kamatera at shopify/scripts/lib/sku-registry.js and is NOT present in THIS repo,
+ * so the in-repo equivalent is: has this SKU already been successfully created
+ * (created OR blocked-but-created) in ANY prior batch job? That makes a retry of
+ * the same batch idempotent instead of minting Shopify duplicates. Defensive:
+ * returns false on any error so a dedup-lookup failure never blocks an import.
+ */
+async function skuAlreadyImported(sku: string): Promise<boolean> {
+ if (!sku) return false;
+ try {
+ const rows = await query<{ one: number }>(
+ `SELECT 1 AS one FROM batch_import_job_items
+ WHERE sku=$1 AND status IN ('created','blocked') LIMIT 1`,
+ [sku],
+ );
+ return rows.length > 0;
+ } catch {
+ return false;
+ }
+}
+
+/**
+ * Pre-resolve OUR net cost for the whole batch through a bounded p-limit(COST_POOL)
+ * pool BEFORE the serial create loop. Each resolveNetCost() call already carries a
+ * short execFile timeout that KILLS its direct child on expiry, and the pool caps
+ * concurrent subprocesses at COST_POOL_CONCURRENCY (default 5). Returns a map
+ * idx -> netCost|null (null = cost-unverified WARN, never a block).
+ *
+ * ORPHAN-SUBPROCESS CAVEAT: execFile's timeout reaps the DIRECT child (verify-price.js)
+ * but NOT any grandchildren it might itself spawn. At concurrency 5 the blast radius
+ * is bounded; full process-group cleanup (spawn detached + kill(-pid)) is a follow-up.
+ */
+async function preResolveCosts(
+ products: BatchProduct[],
+ vendorId: string,
+ costTimeoutMs: number,
+): Promise<Map<number, number | null>> {
+ const out = new Map<number, number | null>();
+ // p-limit v7 is ESM-only; dynamic import keeps this file interop-safe under Next.
+ const pLimit = (await import('p-limit')).default;
+ const limit = pLimit(COST_POOL_CONCURRENCY);
+ await Promise.all(
+ products.map((p, idx) =>
+ limit(async () => {
+ try {
+ const cost = await resolveNetCost(p.vendor || vendorId, p.sku || '', costTimeoutMs);
+ out.set(idx, cost);
+ } catch {
+ out.set(idx, null); // fail-safe: cost-unverified, never blocks
+ }
+ }),
+ ),
+ );
+ return out;
+}
+
+/**
+ * The background worker. NOT awaited by the route — kicked off via setImmediate so
+ * it never blocks the 202 response. Reproduces the original per-product logic
+ * verbatim, writing each outcome to batch_import_job_items as it completes.
+ */
+export async function runJob(jobId: string, req: BatchJobRequest): Promise<void> {
+ const { products, vendorId, privateLabel, customVendorName, costTimeoutMs } = req;
+ try {
+ await setJobStatus(jobId, 'running');
+
+ // ── bounded-concurrency cost pre-resolution (the ONLY parallelized step) ──
+ const costByIdx = await preResolveCosts(products, vendorId, costTimeoutMs);
+
+ // ── SERIAL create loop (Shopify 2 req/s — writes are never parallelized) ──
+ for (let idx = 0; idx < products.length; idx++) {
+ const product = products[idx];
+ try {
+ // Dedup: skip + mark rather than minting a duplicate on a retry.
+ if (product.sku && (await skuAlreadyImported(product.sku))) {
+ await setItem(jobId, idx, 'skipped-duplicate', null, `SKU ${product.sku} already imported`);
+ console.log(`⏭️ Skipped duplicate: ${product.title} (${product.sku})`);
+ continue;
+ }
+
+ // TK-11539: resolve the REAL product type from the payload (falling back to the DW
+ // house default) so sellableWeightLb() below uses the correct per-type shipping
+ // weight (Fabric 1.0 lb, Mural 4.0 lb, …) instead of stamping every batch import at
+ // the 3.0 lb Wallcovering default. The banned word "Wallpaper" is mapped out here
+ // because this value ALSO becomes the displayed product_type (standing rule); weight
+ // is unaffected since the guard table scores Wallpaper == Wallcovering.
+ const PRODUCT_TYPE = (product.product_type || 'Wallcovering').replace(/\bWallpaper\b/gi, 'Wallcovering');
+ const cleaned = product.price?.replace(/[^0-9.]/g, '') || '';
+ const priced = Number(cleaned) > 0; // fail-safe: '', NaN, 0 all => not priced
+
+ // ── SHARED PRICE-INTEGRITY GATE (TK-11403) ────────────────────────────────
+ // Same assertion as the other price-writers. FAIL-SAFE (A-with-teeth): the gate
+ // never throws out of this path; a null cost is a non-blocking WARN, only a real
+ // VIOLATION (sample/default-price leak, below cost/absolute floor, $0/negative-
+ // orderable) blocks. Products stay DRAFT here regardless — a block just forces the
+ // variant UNORDERABLE + a Needs-Price-Review tag so a human resolves it first.
+ // (TK-11776: netCost is now pre-resolved through the bounded pool above; the gate
+ // assertion itself is pure/synchronous and stays inline.)
+ const sampleFloor = resolveSampleFloor(product.vendor);
+ let gateBlocked = false;
+ try {
+ const netCost = costByIdx.get(idx) ?? null;
+ const gate = assertPriceIntegrity({
+ dwSku: product.sku || product.title,
+ netCost,
+ sampleFloor,
+ // The batch route writes ONE (sellable) variant — no sample variant — so the
+ // gate's product-level sample-outcome test is correctly a no-op here.
+ variants: [{
+ role: 'sellable',
+ price: Number(cleaned),
+ orderable: priced, // CONTINUE + qty 100 below make a priced variant orderable
+ sku: product.sku || undefined,
+ priceSource: priced ? 'scraped' : 'defaulted',
+ }],
+ });
+ if (gate.warnings.length) {
+ recordOutcome({ vendor: product.vendor, dwSku: product.sku, outcome: 'warn', codes: gate.warnings.map(w => w.code) });
+ }
+ if (!gate.ok) {
+ console.error(JSON.stringify({ event: 'price_integrity_block', dwSku: product.sku, vendor: product.vendor, violations: gate.violations }));
+ recordOutcome({ vendor: product.vendor, dwSku: product.sku, outcome: 'block', codes: gate.violations.map(v => v.code) });
+ gateBlocked = true;
+ }
+ } catch (gateErr) {
+ // Gate INFRA error must NEVER block an import (A-with-teeth). Log + proceed.
+ console.error(JSON.stringify({ event: 'price_integrity_gate_error', dwSku: product.sku, vendor: product.vendor, error: gateErr instanceof Error ? gateErr.message : String(gateErr) }));
+ }
+ // A variant is only made ORDERABLE when it is genuinely priced AND the gate passed.
+ const sellable = priced && !gateBlocked;
+
+ const productData: any = {
+ title: product.title,
+ body_html: `<p>Imported from ${vendorId}</p>`,
+ vendor: privateLabel ? (customVendorName || 'Private Label') : product.vendor,
+ product_type: PRODUCT_TYPE,
+ // TK-11403/TK-11539: bulk imports stage as DRAFT AND run through the same
+ // price-integrity gate + weight-guard as every other import path, so a
+ // sample/default-price leak or below-cost price can never be promoted and no
+ // variant is ever created zero-weight.
+ status: 'draft',
+ tags: [
+ ...(product.tags || []),
+ vendorId,
+ privateLabel ? 'private-label' : 'manufacturer-brand',
+ ...(sellable ? [] : ['Needs-Price-Review']),
+ ].join(','),
+ images: product.images.map(url => ({ src: url })),
+ variants: [{
+ // ── GUARD TK-11357 ─ never mint a $0/leaked ORDERABLE variant ──────────────
+ // An unpriced OR gate-blocked variant is created UNSTOCKED + DENY so it stays
+ // addressable (quote-only imports still work) but can never be bought for
+ // nothing or at a leaked sample price. Priced + gate-passed goods keep qty 100.
+ price: priced ? cleaned : '0.00',
+ sku: product.sku || '',
+ inventory_quantity: sellable ? 100 : 0,
+ inventory_policy: sellable ? 'continue' : 'deny',
+ inventory_management: 'shopify',
+ // TK-11539: positive weight (POUNDS) via the shared weight-guard so no batch
+ // import is created zero-weight. REST variant weight/weight_unit (API 2024-07).
+ weight: sellableWeightLb(PRODUCT_TYPE),
+ weight_unit: 'lb',
+ }],
+ };
+
+ // Create product in Shopify (SERIAL).
+ const result = await shopifyAPI.instance.createProduct(productData);
+
+ if (result) {
+ // 'blocked' = created-but-flagged (gate violation → draft/unorderable/Needs-Price-Review);
+ // 'created' = clean. Shopify behavior is identical either way; this only enriches observability.
+ await setItem(jobId, idx, gateBlocked ? 'blocked' : 'created', result.id, gateBlocked ? 'price-integrity block (created as draft, Needs-Price-Review)' : null);
+ console.log(`✅ Imported: ${product.title}${gateBlocked ? ' (blocked→review)' : ''}`);
+ } else {
+ await setItem(jobId, idx, 'failed', null, 'Failed to create product');
+ console.error(`❌ Failed: ${product.title}`);
+ }
+
+ // Add delay to avoid Shopify rate limiting (2 req/s).
+ await new Promise(resolve => setTimeout(resolve, 1000));
+ } catch (error) {
+ await setItem(jobId, idx, 'failed', null, error instanceof Error ? error.message : 'Unknown error');
+ console.error(`❌ Error importing ${product.title}:`, error);
+ }
+ }
+
+ await setJobStatus(jobId, 'completed');
+ console.log(`🎯 Batch import job ${jobId} complete`);
+ } catch (fatal) {
+ // A fatal worker error (e.g. DB down) marks the whole job failed so pollers see it.
+ console.error(`💥 Batch import job ${jobId} failed:`, fatal);
+ try {
+ await setJobStatus(jobId, 'failed', fatal instanceof Error ? fatal.message : String(fatal));
+ } catch { /* best-effort */ }
+ }
+}
+
+export interface JobStatusItem {
+ idx: number;
+ title: string | null;
+ sku: string | null;
+ status: ItemStatus;
+ product_id: string | null;
+ error: string | null;
+ updated_at: string;
+}
+
+export interface JobStatus {
+ jobId: string;
+ status: string;
+ total: number;
+ vendorId: string | null;
+ error: string | null;
+ createdAt: string;
+ updatedAt: string;
+ counts: Record<string, number>;
+ items: JobStatusItem[];
+}
+
+/** Read a job + its per-item progress (for the status polling endpoint). */
+export async function getJobStatus(jobId: string): Promise<JobStatus | null> {
+ await ensureBatchJobTables();
+ const jobs = await query<any>(
+ `SELECT job_id, status, total, vendor_id, error, created_at, updated_at
+ FROM batch_import_jobs WHERE job_id=$1`,
+ [jobId],
+ );
+ const job = jobs[0];
+ if (!job) return null;
+
+ const items = await query<any>(
+ `SELECT idx, title, sku, status, product_id, error, updated_at
+ FROM batch_import_job_items WHERE job_id=$1 ORDER BY idx ASC`,
+ [jobId],
+ );
+
+ const counts: Record<string, number> = { pending: 0, created: 0, failed: 0, blocked: 0, 'skipped-duplicate': 0 };
+ for (const it of items) counts[it.status] = (counts[it.status] || 0) + 1;
+
+ return {
+ jobId: job.job_id,
+ status: job.status,
+ total: job.total,
+ vendorId: job.vendor_id,
+ error: job.error,
+ createdAt: job.created_at,
+ updatedAt: job.updated_at,
+ counts,
+ items: items.map((it: any) => ({
+ idx: it.idx,
+ title: it.title,
+ sku: it.sku,
+ status: it.status,
+ product_id: it.product_id,
+ error: it.error,
+ updated_at: it.updated_at,
+ })),
+ };
+}
← b632244f TK-11776: repair batch-import negative tests for the backgro
·
back to Designer Wallcoverings
·
auto-data-snapshot: 2026-09-15T13:06:09 (5 data files) — DW- ca116495 →