← back to Designer Wallcoverings
batch-import: close 3 review findings on the TK-11776 background-job worker
e6659530ae517308f65e04879423c7140bdb807e · 2026-09-16 09:24:03 -0700 · Steve
Landing the review fixes for the background bulk-import worker (with the
in-flight TK-11776/TK-11786 background-job rearchitecture they depend on):
#1 Catalog-wide duplicate pre-check. The per-job claim ledger only knew this
route's own creations, so re-importing a SKU already created via the
single-import path / cadence / pre-ledger would mint a DUPLICATE. Add a
read-only pre-check against the three authoritative sources
(dw_sku_registry / shopify_products / scraped_products), mirroring
check-duplicates-batch; matches are skipped before any claim or create.
Fail-open on DB error (never blocks an import).
#2 Process-wide Shopify write lock + in-loop 429 retry. POST returns 202 and
fires a setImmediate worker, so overlapping batches could run concurrent
create loops and blow past Shopify's 2 req/s. Serialize create loops across
jobs with a module-level lock; retry a 429 in-loop with backoff instead of
dropping the item on the first throttle (a 429 is pre-persist, so retry
never risks a duplicate).
#3 Chunk the job-item seed INSERT (500 rows/chunk) so it can't overflow
Postgres's 65,535 bind-param cap when IMPORT_BATCH_MAX is raised; stays
atomic inside the seed transaction.
#4 (auth) verify-only per direction — no code change (app has no route auth
today; adding it only here would be inconsistent).
Tests: +4 negative/behavior tests (catalog-dup skip, 429 retry, non-429
fail-fast) + a test-only write-lock reset for isolation. Full suite 44/44.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
M DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.tsM DW-Programming/ImportNewSkufromURL/app/api/import/batch/route.tsA DW-Programming/ImportNewSkufromURL/app/api/import/batch/status/[jobId]/route.tsM DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
Diff
commit e6659530ae517308f65e04879423c7140bdb807e
Author: Steve <steve@designerwallcoverings.com>
Date: Wed Sep 16 09:24:03 2026 -0700
batch-import: close 3 review findings on the TK-11776 background-job worker
Landing the review fixes for the background bulk-import worker (with the
in-flight TK-11776/TK-11786 background-job rearchitecture they depend on):
#1 Catalog-wide duplicate pre-check. The per-job claim ledger only knew this
route's own creations, so re-importing a SKU already created via the
single-import path / cadence / pre-ledger would mint a DUPLICATE. Add a
read-only pre-check against the three authoritative sources
(dw_sku_registry / shopify_products / scraped_products), mirroring
check-duplicates-batch; matches are skipped before any claim or create.
Fail-open on DB error (never blocks an import).
#2 Process-wide Shopify write lock + in-loop 429 retry. POST returns 202 and
fires a setImmediate worker, so overlapping batches could run concurrent
create loops and blow past Shopify's 2 req/s. Serialize create loops across
jobs with a module-level lock; retry a 429 in-loop with backoff instead of
dropping the item on the first throttle (a 429 is pre-persist, so retry
never risks a duplicate).
#3 Chunk the job-item seed INSERT (500 rows/chunk) so it can't overflow
Postgres's 65,535 bind-param cap when IMPORT_BATCH_MAX is raised; stays
atomic inside the seed transaction.
#4 (auth) verify-only per direction — no code change (app has no route auth
today; adding it only here would be inconsistent).
Tests: +4 negative/behavior tests (catalog-dup skip, 429 retry, non-429
fail-fast) + a test-only write-lock reset for isolation. Full suite 44/44.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
.../__tests__/batch-import-gate.test.ts | 362 ++++++++++++-
.../app/api/import/batch/route.ts | 211 ++------
.../app/api/import/batch/status/[jobId]/route.ts | 31 ++
.../ImportNewSkufromURL/lib/batch-import-jobs.ts | 591 +++++++++++++++++++--
4 files changed, 980 insertions(+), 215 deletions(-)
diff --git a/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts b/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
index c5d30251..8cde5d9e 100644
--- a/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
+++ b/DW-Programming/ImportNewSkufromURL/__tests__/batch-import-gate.test.ts
@@ -30,7 +30,14 @@
* jest's CJS interop cannot execute for the real ESM-only p-limit v7).
*/
import { POST } from '@/app/api/import/batch/route';
-import { runJob, BatchJobRequest } from '@/lib/batch-import-jobs';
+import {
+ runJob,
+ BatchJobRequest,
+ deriveDedupKey,
+ resolveJobOwner,
+ reclaimOrphanedJobs,
+ __resetShopifyWriteLockForTests,
+} from '@/lib/batch-import-jobs';
import { NextRequest } from 'next/server';
const mockCreateProduct = jest.fn();
@@ -44,10 +51,17 @@ jest.mock('@/lib/price-integrity-cost', () => ({
}));
jest.mock('@/lib/price-integrity-record', () => ({ recordOutcome: jest.fn() }));
// In-memory job store: the worker + route call query() for createJob/setItem/status/
-// dedup. Return [] for every call → createJob/setItem are inert and the dedup lookup
-// ('has this SKU already been imported?') returns false, so nothing is skipped.
+// dedup. Return [] for every call so createJob/setItem are inert — EXCEPT the atomic
+// dedup CLAIM (`INSERT INTO batch_import_created_skus ... RETURNING sku`), which must
+// return a row to signal the SKU was WON (not already claimed); a bare [] there would
+// read as "lost the claim" and skip every product. Modelling "no duplicates in the
+// test" = every claim wins, so nothing is skipped.
jest.mock('@/lib/database/postgres-client', () => ({
- query: jest.fn().mockResolvedValue([]),
+ query: jest.fn((sql: string) =>
+ /insert\s+into\s+batch_import_created_skus/i.test(String(sql))
+ ? Promise.resolve([{ sku: 'won' }]) // claim won → worker proceeds to createProduct
+ : Promise.resolve([]),
+ ),
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.
@@ -101,6 +115,9 @@ async function runOne(over: Partial<any> = {}) {
beforeEach(() => {
mockCreateProduct.mockReset();
mockCreateProduct.mockResolvedValue({ id: 999 });
+ // Reset the process-wide Shopify write lock so a prior test's un-drained fire-and-forget
+ // worker (setImmediate) can't hold it into the next case (TK-11786 review #2, write lock).
+ __resetShopifyWriteLockForTests();
});
describe('batch import worker — price-integrity + weight wiring (TK-11403/TK-11539)', () => {
@@ -210,4 +227,341 @@ describe('batch import route — background-job contract (TK-11776)', () => {
// Let the worker's trailing 1s sleep + completion finish before the suite tears down.
await new Promise(resolve => setTimeout(resolve, 1100));
});
+
+ it('DUPLICATE SKU (claim LOST) → skipped, NO Shopify create (TK-11776 F2 dedup)', async () => {
+ // The atomic claim is what prevents the concurrent/overlapping-job duplicate mint.
+ // Simulate the SKU already claimed by a prior/concurrent job: the claim
+ // `INSERT ... ON CONFLICT DO NOTHING RETURNING sku` yields NO row → claimSku()
+ // returns false → the worker must skip the product and NEVER call createProduct.
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const claimAwareDefault = query.getMockImplementation()!;
+ query.mockImplementation((sql: string) =>
+ /insert\s+into\s+batch_import_created_skus/i.test(String(sql))
+ ? Promise.resolve([]) // conflict → claim LOST (SKU already taken)
+ : Promise.resolve([]),
+ );
+ try {
+ await runJob('job-dup', {
+ products: [makeProduct({ sku: 'DWTS-DUP1', price: '$195.00' })],
+ vendorId: 'thibaut',
+ privateLabel: false,
+ costTimeoutMs: 12000,
+ });
+ expect(mockCreateProduct).not.toHaveBeenCalled(); // a duplicate is never minted
+ } finally {
+ query.mockImplementation(claimAwareDefault); // restore the claim-wins default
+ }
+ });
+});
+
+// ═══════════════════════════════════════════════════════════════════════════════════════
+// TK-11786 — the three review holes closed. Each block is a NEGATIVE test (CLAUDE.md rule 3):
+// it goes RED if the corresponding fix is reverted.
+// ═══════════════════════════════════════════════════════════════════════════════════════
+
+describe('TK-11786 H1 — SKU-less composite dedup (deriveDedupKey + claimKey)', () => {
+ it('a real SKU is the dedup key (unchanged behavior)', () => {
+ expect(deriveDedupKey({ sku: 'DWTS-9', vendor: 'Thibaut', title: 'Palm', collection: 'Summer' })).toBe('DWTS-9');
+ });
+
+ it('an SKU-less product derives a normalized composite vendor+title+collection key', () => {
+ expect(deriveDedupKey({ vendor: 'Thibaut', title: 'Palm Beach', collection: 'Summer' }))
+ .toBe('composite:thibaut|palm beach|summer');
+ // whitespace-trim/collapse + lowercase makes the key stable across submits
+ expect(deriveDedupKey({ vendor: ' Thibaut ', title: 'Palm Beach', collection: '' }))
+ .toBe('composite:thibaut|palm beach|');
+ });
+
+ it('SKU-less AND title-less → empty key → proceeds unguarded (no coarse false-dedup)', () => {
+ expect(deriveDedupKey({ vendor: 'Thibaut', collection: 'Summer' })).toBe('');
+ expect(deriveDedupKey({})).toBe('');
+ });
+
+ it('WIRING: an SKU-less product now CLAIMS the composite key before creating (the H1 fix)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ query.mockClear(); // inspect only THIS test's calls
+ await runOne({ sku: undefined, price: '$195.00' }); // claim-wins default → should create
+ const claimCall = query.mock.calls.find(([sql]: any[]) =>
+ /insert\s+into\s+batch_import_created_skus/i.test(String(sql)));
+ // If H1 is reverted (SKU-less skips the claim entirely) this is undefined → test FAILS.
+ expect(claimCall).toBeDefined();
+ expect(String(claimCall![1][0])).toMatch(/^composite:/); // claimed on the composite key, not a SKU
+ expect(mockCreateProduct).toHaveBeenCalledTimes(1); // claim won → product created
+ });
+
+ it('WIRING: an SKU-less DUPLICATE (composite claim LOST) → skipped, NO Shopify create', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ query.mockImplementation((sql: string) =>
+ /insert\s+into\s+batch_import_created_skus/i.test(String(sql))
+ ? Promise.resolve([]) // ON CONFLICT DO NOTHING → no row → claim LOST
+ : Promise.resolve([]));
+ try {
+ await runJob('job-composite-dup', {
+ products: [makeProduct({ sku: undefined, price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ expect(mockCreateProduct).not.toHaveBeenCalled(); // a SKU-less duplicate is never re-minted
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+});
+
+describe('TK-11786 H2/H3 — boot-time orphan reclaim (crash-safe + owner-scoped)', () => {
+ it('H3: reclaim is scoped to THIS owner (dev boot can never fail prod jobs)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ const calls: Array<{ sql: string; params: any[] }> = [];
+ query.mockImplementation((sql: string, params?: any[]) => {
+ calls.push({ sql: String(sql), params: params || [] });
+ if (/update\s+batch_import_jobs/i.test(String(sql)) && /returning\s+job_id/i.test(String(sql))) {
+ return Promise.resolve([{ job_id: 'j-orphan' }]); // one orphan so downstream steps run
+ }
+ return Promise.resolve([]);
+ });
+ try {
+ await reclaimOrphanedJobs('kamatera-prod');
+ const jobsUpd = calls.find(c => /update\s+batch_import_jobs/i.test(c.sql) && /returning\s+job_id/i.test(c.sql))!;
+ // If the owner scope is reverted, this match fails → test RED (the dev/prod stomp returns).
+ expect(jobsUpd.sql).toMatch(/owner\s*=\s*\$1/i);
+ expect(jobsUpd.params[0]).toBe('kamatera-prod');
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+
+ it('H2: not-started orphans RELEASE their claim (false-fail fix); in-flight ones are flagged + retained', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ const calls: Array<{ sql: string; params: any[] }> = [];
+ query.mockImplementation((sql: string, params?: any[]) => {
+ calls.push({ sql: String(sql), params: params || [] });
+ if (/update\s+batch_import_jobs/i.test(String(sql)) && /returning\s+job_id/i.test(String(sql))) {
+ return Promise.resolve([{ job_id: 'j-orphan' }]);
+ }
+ return Promise.resolve([]);
+ });
+ try {
+ await reclaimOrphanedJobs('kamatera-prod');
+
+ // in-flight (create_started_at IS NOT NULL) items are flagged, claim KEPT (never auto-mint a dup)
+ expect(calls.some(c =>
+ /update\s+batch_import_job_items/i.test(c.sql) && /create_started_at\s+is\s+not\s+null/i.test(c.sql),
+ )).toBe(true);
+
+ // not-started items RELEASE the claim so a re-submit re-imports them — the crash-ordering
+ // false-fail fix. If this DELETE is removed, a never-created SKU stays claimed forever.
+ const release = calls.find(c => /delete\s+from\s+batch_import_created_skus/i.test(c.sql));
+ expect(release).toBeDefined();
+ expect(release!.sql).toMatch(/create_started_at\s+is\s+null/i);
+ expect(release!.sql).toMatch(/status\s+not\s+in\s*\(\s*'created'\s*,\s*'blocked'\s*\)/i); // never drop a created SKU's claim
+ // (Kimi review d) the release is CLAIM-OWNER-scoped: it joins the claim row to the
+ // item that CREATED it (c.job_id=i.job_id AND c.idx=i.idx), so a loser (skipped-
+ // duplicate) item can NEVER trigger deletion of the winning job's live claim.
+ expect(release!.sql).toMatch(/c\.job_id\s*=\s*i\.job_id/i);
+ expect(release!.sql).toMatch(/c\.idx\s*=\s*i\.idx/i);
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+
+ it('H2 (Kimi b): an AMBIGUOUS create failure (network/timeout, no HTTP status) KEEPS the claim (no re-mint)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ query.mockClear();
+ mockCreateProduct.mockReset();
+ mockCreateProduct.mockRejectedValueOnce(new Error('fetch failed: ECONNRESET')); // no HTTP status → ambiguous
+ await runJob('job-ambiguous', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ // releaseKey issues `DELETE FROM batch_import_created_skus WHERE sku=$1`; the sweeper's
+ // delete uses `... c USING ...`. An ambiguous failure must NOT release the claim.
+ const releasedClaim = query.mock.calls.some(([sql]: any[]) =>
+ /delete\s+from\s+batch_import_created_skus\s+where\s+sku/i.test(String(sql)));
+ expect(releasedClaim).toBe(false); // claim retained → a re-submit can't re-mint the maybe-created product
+ });
+
+ it('FINDING A: a createProduct throw AFTER the marker was set RETAINS the claim (release NOT called)', async () => {
+ // Proves the catch is marker-aware: once markCreateStarted has stamped create_started_at
+ // (create was attempted), an ambiguous createProduct throw is an UNKNOWN Shopify outcome —
+ // the claim MUST be kept so a re-submit can never auto-mint a duplicate, mirroring the boot
+ // sweeper's in-flight branch. Goes RED if the fix is reverted to "release on any throw"
+ // (releaseKey fires → releasedClaim true) OR if the marker-set retention is removed.
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ query.mockClear();
+ mockCreateProduct.mockReset();
+ mockCreateProduct.mockRejectedValueOnce(new Error('socket hang up')); // no HTTP status → ambiguous
+ await runJob('job-marker-set-throw', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ // (1) the create-started marker WAS stamped before the throw (proves we're on the marker-set path)
+ const markerStamped = query.mock.calls.some(([sql]: any[]) =>
+ /update\s+batch_import_job_items[\s\S]*create_started_at\s*=\s*now\(\)/i.test(String(sql)));
+ expect(markerStamped).toBe(true);
+ // (2) the claim was NOT released — releaseKey issues `DELETE ... WHERE sku=$1`
+ const releasedClaim = query.mock.calls.some(([sql]: any[]) =>
+ /delete\s+from\s+batch_import_created_skus\s+where\s+sku/i.test(String(sql)));
+ expect(releasedClaim).toBe(false); // marker set + ambiguous → claim retained (no re-mint)
+ });
+
+ it('FINDING A (complement): a PRE-marker throw (before createProduct is attempted) RELEASES the claim — the real "claim leaks forever" fix', async () => {
+ // The OTHER half of the marker-aware catch, and the branch every other negative test
+ // leaves uncovered: every existing mockRejectedValue rejects createProduct itself, which
+ // is AFTER markCreateStarted stamps create_started_at. This proves the `!markerSet` path —
+ // a throw BETWEEN winning the dedup claim and stamping the marker (price / product-type /
+ // payload resolution, i.e. BEFORE any Shopify create is attempted). There the product
+ // provably does NOT exist, so OUR claim MUST be released or a re-submit would FALSELY skip
+ // a never-created product forever. Mirrors the boot sweeper's create_started_at IS NULL
+ // branch. Goes RED if the `!markerSet` release branch is reverted (claim leaks) OR if the
+ // marker were stamped before the pre-create work.
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ query.mockClear();
+ await runJob('job-precreate-throw', {
+ // product-type resolution runs AFTER claimKey wins the claim (holdsClaim=true) but BEFORE
+ // markCreateStarted, and OUTSIDE the gate's swallowing inner try — so a non-string
+ // product_type makes `.replace(...)` throw there: a deterministic PRE-CREATE fault.
+ products: [makeProduct({ price: '$195.00', product_type: 123 as any })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ // (0) the throw was genuinely pre-create — Shopify createProduct was never reached
+ expect(mockCreateProduct).not.toHaveBeenCalled();
+ // (1) create_started_at was NEVER stamped → we're provably on the !markerSet path
+ const markerStamped = query.mock.calls.some(([sql]: any[]) =>
+ /update\s+batch_import_job_items[\s\S]*create_started_at\s*=\s*now\(\)/i.test(String(sql)));
+ expect(markerStamped).toBe(false);
+ // (2) the dedup claim WAS released — releaseKey issues `DELETE ... WHERE sku=$1`
+ const releasedClaim = query.mock.calls.some(([sql]: any[]) =>
+ /delete\s+from\s+batch_import_created_skus\s+where\s+sku/i.test(String(sql)));
+ expect(releasedClaim).toBe(true); // pre-create throw → claim freed so a re-submit re-imports
+ });
+
+ it('H2 (Kimi b): a DEFINITIVE 4xx create failure RELEASES the claim (safe retry, product not created)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ query.mockClear();
+ mockCreateProduct.mockReset();
+ mockCreateProduct.mockRejectedValueOnce(new Error('Shopify API Error (422): title invalid')); // 4xx → not created
+ await runJob('job-4xx', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ const releasedClaim = query.mock.calls.some(([sql]: any[]) =>
+ /delete\s+from\s+batch_import_created_skus\s+where\s+sku/i.test(String(sql)));
+ expect(releasedClaim).toBe(true); // definitively not created → claim freed so a re-submit retries
+ });
+
+ it('H2 invariant: if the create-started marker write FAILS, the Shopify create is SKIPPED (no untracked dup)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ query.mockImplementation((sql: string) => {
+ const s = String(sql);
+ if (/insert\s+into\s+batch_import_created_skus/i.test(s)) return Promise.resolve([{ sku: 'won' }]); // claim wins
+ if (/update\s+batch_import_job_items[\s\S]*create_started_at\s*=\s*now\(\)/i.test(s)) {
+ return Promise.reject(new Error('simulated marker-write failure')); // markCreateStarted fails
+ }
+ return Promise.resolve([]);
+ });
+ try {
+ await runJob('job-marker-fail', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ // If the invariant is reverted (create runs with the marker unset), this fails.
+ expect(mockCreateProduct).not.toHaveBeenCalled();
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+
+ it('reclaim with NO orphans for this owner does nothing further (no claim deletes)', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ const seen: string[] = [];
+ query.mockImplementation((sql: string) => { seen.push(String(sql)); return Promise.resolve([]); });
+ try {
+ await reclaimOrphanedJobs('nobody'); // jobs UPDATE returns [] → early return
+ expect(seen.some(s => /delete\s+from\s+batch_import_created_skus/i.test(s))).toBe(false);
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+
+ it('H3: resolveJobOwner honors IMPORT_JOB_OWNER, else falls back to hostname', () => {
+ const prev = process.env.IMPORT_JOB_OWNER;
+ process.env.IMPORT_JOB_OWNER = 'prod-box';
+ try {
+ expect(resolveJobOwner()).toBe('prod-box');
+ } finally {
+ if (prev === undefined) delete process.env.IMPORT_JOB_OWNER;
+ else process.env.IMPORT_JOB_OWNER = prev;
+ }
+ // with no override it still returns a non-empty stable token (the hostname)
+ delete process.env.IMPORT_JOB_OWNER;
+ expect(resolveJobOwner().length).toBeGreaterThan(0);
+ if (prev !== undefined) process.env.IMPORT_JOB_OWNER = prev;
+ });
+});
+
+// ═══════════════════════════════════════════════════════════════════════════════════════
+// Review fixes (2026-09-16) — each is a NEGATIVE test: it goes RED if the fix is reverted.
+// ═══════════════════════════════════════════════════════════════════════════════════════
+
+describe('review #1 — catalog-wide duplicate pre-check (no cross-path re-mint)', () => {
+ it('a real SKU already in the catalog is SKIPPED before any claim or create', async () => {
+ const { query } = require('@/lib/database/postgres-client') as { query: jest.Mock };
+ const def = query.getMockImplementation()!;
+ query.mockImplementation((sql: string) => {
+ // the catalog pre-check UNION selects FROM dw_sku_registry → report a hit (lowercased)
+ if (/from\s+dw_sku_registry/i.test(String(sql))) return Promise.resolve([{ k: 'dwts-dup1' }]);
+ if (/insert\s+into\s+batch_import_created_skus/i.test(String(sql))) return Promise.resolve([{ sku: 'won' }]);
+ return Promise.resolve([]);
+ });
+ query.mockClear(); // inspect only THIS test's query calls (calls accumulate across the run)
+ try {
+ await runJob('job-catalog-dup', {
+ products: [makeProduct({ sku: 'DWTS-DUP1', price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ expect(mockCreateProduct).not.toHaveBeenCalled(); // catalog dup → never re-minted
+ // The skip happens BEFORE the claim, so no claim INSERT was issued for this item.
+ const claimed = query.mock.calls.some(([sql]: any[]) =>
+ /insert\s+into\s+batch_import_created_skus/i.test(String(sql)));
+ expect(claimed).toBe(false);
+ } finally {
+ query.mockImplementation(def);
+ }
+ });
+
+ it('a real SKU NOT in the catalog proceeds to create (pre-check is not over-broad)', async () => {
+ // default mock: the dw_sku_registry SELECT returns [] → not a catalog dup → creates.
+ const { v } = await runOne({ sku: 'DWTS-FRESH', price: '$195.00' });
+ expect(mockCreateProduct).toHaveBeenCalledTimes(1);
+ expect(v.price).toBe('195.00');
+ });
+});
+
+describe('review #2 — Shopify 429 is retried in-loop, other 4xx fail fast', () => {
+ it('a 429 is retried (not dropped) and the product is then created', async () => {
+ mockCreateProduct.mockReset();
+ mockCreateProduct
+ .mockRejectedValueOnce(new Error('Shopify API Error (429): Too Many Requests'))
+ .mockResolvedValueOnce({ id: 777 });
+ await runJob('job-429', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ // If the 429-retry is reverted (first-429-drops), this is 1 → test RED.
+ expect(mockCreateProduct).toHaveBeenCalledTimes(2);
+ }, 15000);
+
+ it('a non-429 4xx is NOT retried (fails fast, single attempt)', async () => {
+ mockCreateProduct.mockReset();
+ mockCreateProduct.mockRejectedValue(new Error('Shopify API Error (422): Unprocessable'));
+ await runJob('job-422', {
+ products: [makeProduct({ price: '$195.00' })],
+ vendorId: 'thibaut', privateLabel: false, costTimeoutMs: 12000,
+ });
+ expect(mockCreateProduct).toHaveBeenCalledTimes(1);
+ });
});
diff --git a/DW-Programming/ImportNewSkufromURL/app/api/import/batch/route.ts b/DW-Programming/ImportNewSkufromURL/app/api/import/batch/route.ts
index 88240933..fd92891c 100644
--- a/DW-Programming/ImportNewSkufromURL/app/api/import/batch/route.ts
+++ b/DW-Programming/ImportNewSkufromURL/app/api/import/batch/route.ts
@@ -1,13 +1,14 @@
import { NextRequest, NextResponse } from 'next/server';
-import { shopifyAPI } from '@/lib/shopify-api';
-// TK-11403/TK-11539 (2026-09-14): wire the batch route into the SAME shared
-// price-integrity gate + weight-guard as createOrGetProduct / schumacher /
-// import-queue-runner, so this path is no longer the one exception on the weaker
-// inline price>0 check.
-import { assertPriceIntegrity, resolveSampleFloor } from '@/lib/price-integrity-gate';
-import { resolveNetCost } from '@/lib/price-integrity-cost';
-import { recordOutcome } from '@/lib/price-integrity-record';
-import { sellableWeightLb } from '@/lib/weight-guard';
+import { randomUUID } from 'node:crypto';
+// TK-11776 (DTD verdict B, 2026-09-15): the batch import is now a BACKGROUND JOB.
+// The old handler ran a fully serial per-product loop (resolveNetCost subprocess +
+// createProduct + 1s delay) INSIDE this HTTP request — worst case many minutes,
+// which outlives client/proxy timeouts and leaves SILENT partial catalog writes.
+// POST now validates + caps, persists a job to dw_unified, kicks a fire-and-forget
+// worker, and returns 202 {jobId} immediately. All per-product logic (price-integrity
+// gate, weight-guard, MAX_BATCH cap, Needs-Price-Review tagging, $0-orderable guard)
+// is PRESERVED unchanged inside the worker (lib/batch-import-jobs.ts).
+import { createJob, runJob, BatchJobRequest } from '@/lib/batch-import-jobs';
interface BatchImportRequest {
products: Array<{
@@ -34,7 +35,7 @@ export async function POST(request: NextRequest) {
try {
const body: BatchImportRequest = await request.json();
const { products, vendorId, privateLabel, customVendorName } = body;
-
+
if (!products || products.length === 0) {
return NextResponse.json(
{ success: false, error: 'No products to import' },
@@ -42,12 +43,10 @@ export async function POST(request: NextRequest) {
);
}
- // (review 2026-09-14) Bound the request. This handler runs strictly serially —
- // per product it spawns a `node verify-price.js` subprocess for cost resolution
- // AND waits a 1s Shopify-rate-limit delay — so an unbounded `products.length`
- // makes the HTTP request run for minutes-to-hours and the client/proxy times out.
- // Cap the batch (env-overridable) and reject oversized ones with 400 so callers
- // page their imports instead of firing one giant request.
+ // (review 2026-09-14, preserved TK-11776) Bound the request. The worker still
+ // runs the create loop strictly serially with a 1s Shopify-rate-limit delay,
+ // so cap the batch (env-overridable) and reject oversized ones with 400 so
+ // callers page their imports instead of firing one giant request.
const MAX_BATCH = Math.max(1, parseInt(process.env.IMPORT_BATCH_MAX || '150', 10) || 150);
if (products.length > MAX_BATCH) {
return NextResponse.json(
@@ -58,162 +57,58 @@ export async function POST(request: NextRequest) {
{ status: 400 }
);
}
+
// Per-product cost-lookup timeout for the batch path — far shorter than the
// single-import default (60s) so one slow/hung price-finder can't stall the
- // whole batch request. A timeout is a fail-safe null cost (cost-unverified WARN),
- // not a block, so a short timeout never wrongly rejects an import.
+ // whole batch. A timeout is a fail-safe null cost (cost-unverified WARN), not
+ // a block, so a short timeout never wrongly rejects an import.
const BATCH_COST_TIMEOUT_MS = Math.max(
1000,
parseInt(process.env.IMPORT_BATCH_COST_TIMEOUT_MS || '12000', 10) || 12000
);
- console.log(`📦 Starting batch import of ${products.length} products`);
-
- const results = [];
- let successCount = 0;
- let failCount = 0;
-
- for (const product of products) {
- try {
- // Prepare product data (match ShopifyProduct interface)
- // 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
+ const jobId = randomUUID();
+ const jobReq: BatchJobRequest = {
+ products,
+ vendorId,
+ privateLabel,
+ customVendorName,
+ costTimeoutMs: BATCH_COST_TIMEOUT_MS,
+ };
- // ── SHARED PRICE-INTEGRITY GATE (TK-11403) ────────────────────────────────
- // Run the batch route through the SAME assertion as the other price-writers
- // instead of the old inline price>0-only check. FAIL-SAFE (A-with-teeth):
- // resolveNetCost + the gate never throw 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 before anything can promote it to ACTIVE.
- const sampleFloor = resolveSampleFloor(product.vendor);
- let gateBlocked = false;
- try {
- const netCost = await resolveNetCost(product.vendor || vendorId, product.sku || '', BATCH_COST_TIMEOUT_MS);
- 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;
+ // Persist the job + one pending row per product BEFORE responding, so a poll of
+ // /status/{jobId} immediately after the 202 always finds the record.
+ await createJob(jobId, jobReq);
- 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 (review 2026-09-14): bulk imports stage as DRAFT AND now 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
- const result = await shopifyAPI.instance.createProduct(productData);
-
- if (result) {
- successCount++;
- results.push({
- success: true,
- title: product.title,
- productId: result.id
- });
- console.log(`✅ Imported: ${product.title}`);
- } else {
- failCount++;
- results.push({
- success: false,
- title: product.title,
- error: 'Failed to create product'
- });
- console.error(`❌ Failed: ${product.title}`);
- }
-
- // Add delay to avoid rate limiting
- await new Promise(resolve => setTimeout(resolve, 1000));
-
- } catch (error) {
- failCount++;
- results.push({
- success: false,
- title: product.title,
- error: error instanceof Error ? error.message : 'Unknown error'
- });
- console.error(`❌ Error importing ${product.title}:`, error);
- }
- }
-
- console.log(`🎯 Batch import complete: ${successCount} success, ${failCount} failed`);
-
- return NextResponse.json({
- success: true,
- imported: successCount,
- failed: failCount,
- total: products.length,
- results
+ console.log(`📦 Queued batch import job ${jobId}: ${products.length} products`);
+
+ // Kick the worker AFTER responding — setImmediate defers it past the response so
+ // the HTTP request returns in milliseconds. (This app runs under a long-lived
+ // `next start` / PM2 process, so the floating promise continues to completion.)
+ setImmediate(() => {
+ runJob(jobId, jobReq).catch(err =>
+ console.error(`💥 Unhandled batch worker error for job ${jobId}:`, err)
+ );
});
-
+
+ return NextResponse.json(
+ {
+ success: true,
+ jobId,
+ total: products.length,
+ status: 'pending',
+ statusUrl: `/api/import/batch/status/${jobId}`,
+ },
+ { status: 202 }
+ );
} catch (error) {
console.error('Batch import error:', error);
return NextResponse.json(
- {
- success: false,
- error: error instanceof Error ? error.message : 'Failed to import products'
+ {
+ success: false,
+ error: error instanceof Error ? error.message : 'Failed to queue batch import',
},
{ status: 500 }
);
}
-}
\ No newline at end of file
+}
diff --git a/DW-Programming/ImportNewSkufromURL/app/api/import/batch/status/[jobId]/route.ts b/DW-Programming/ImportNewSkufromURL/app/api/import/batch/status/[jobId]/route.ts
new file mode 100644
index 00000000..839a4b52
--- /dev/null
+++ b/DW-Programming/ImportNewSkufromURL/app/api/import/batch/status/[jobId]/route.ts
@@ -0,0 +1,31 @@
+import { NextRequest, NextResponse } from 'next/server';
+// TK-11776 (DTD verdict B): polling endpoint for the background bulk-import job.
+// Returns the job record + per-item progress (counts + list) so a UI can poll
+// until status === 'completed' | 'failed'. This is the observability contract that
+// closes the old silent-partial-write hole.
+import { getJobStatus } from '@/lib/batch-import-jobs';
+
+export async function GET(
+ request: NextRequest,
+ { params }: { params: { jobId: string } }
+) {
+ try {
+ const { jobId } = params;
+ if (!jobId) {
+ return NextResponse.json({ success: false, error: 'jobId required' }, { status: 400 });
+ }
+
+ const status = await getJobStatus(jobId);
+ if (!status) {
+ return NextResponse.json({ success: false, error: 'Job not found' }, { status: 404 });
+ }
+
+ return NextResponse.json({ success: true, ...status });
+ } catch (error) {
+ console.error('Batch status error:', error);
+ return NextResponse.json(
+ { success: false, error: error instanceof Error ? error.message : 'Failed to read job status' },
+ { status: 500 }
+ );
+ }
+}
diff --git a/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts b/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
index d7d73e88..3a69b75f 100644
--- a/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
+++ b/DW-Programming/ImportNewSkufromURL/lib/batch-import-jobs.ts
@@ -14,10 +14,14 @@
* 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.
+ * write is now VISIBLE, not silent (observability). A process restart
+ * (deploy/PM2/crash) mid-job is caught by a boot-time orphan sweeper
+ * (reclaimOrphanedJobs, run once per process inside ensureBatchJobTables): any
+ * job left 'pending'/'running' by a dead process is marked 'failed' with a
+ * re-submit prompt, so a /status poller gets a TERMINAL state instead of polling
+ * a forever-'running' job. We deliberately do NOT auto-recreate the missing
+ * products (that would be an unattended customer-facing Shopify write) — a human
+ * re-submits the batch and already-created SKUs are skipped as duplicates.
*
* The per-product logic (price-integrity gate, weight-guard, $0-orderable guard,
* Needs-Price-Review tag, DRAFT status) is reproduced UNCHANGED from the original
@@ -25,8 +29,30 @@
* 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.
+ *
+ * TK-11786 (review of TK-11776 — code-reviewer + contrarian, 2026-09-15) closed
+ * three holes, Steve-approved fixes, DEPLOY HELD:
+ * H1 — SKU-LESS DEDUP: the F2 claim only guarded products that HAVE a SKU; an
+ * SKU-less quote item imported unguarded, so a re-submit re-minted it. Fixed
+ * with a COMPOSITE dedup key (vendor+title+collection) for SKU-less products
+ * (deriveDedupKey) — claimed atomically in the same ledger table.
+ * H2 — CRASH-ORDERING FALSE-FAIL: the claim was INSERTed before createProduct with
+ * no durable marker between them, so after a restart a claimed-but-never-created
+ * SKU was indistinguishable from a claimed-and-created one; the sweeper kept the
+ * claim, and a re-submit then FALSELY skipped a product that was never created —
+ * forever. Fixed by stamping create_started_at right before the create: the boot
+ * sweeper RELEASES claims of orphaned items whose create never started (safe to
+ * re-import) and KEEPS+FLAGS the one item that was mid-create at the crash (its
+ * Shopify state is unknown — retain the claim to never auto-mint a duplicate, and
+ * surface it for a human to verify).
+ * H3 — DEV/PROD SWEEP STOMP: reclaimOrphanedJobs marked ALL 'pending'/'running' jobs
+ * 'failed' at boot; dev and prod share dw_unified, so a dev boot stomped prod's
+ * genuinely-running jobs. Fixed by stamping each job with an OWNER
+ * (IMPORT_JOB_OWNER || hostname) and scoping the reclaim to owner = THIS instance,
+ * so one environment can never reclaim another's live jobs.
*/
+import os from 'node:os';
import { query, transaction } from './database/postgres-client';
import { shopifyAPI } from './shopify-api';
import { assertPriceIntegrity, resolveSampleFloor } from './price-integrity-gate';
@@ -63,11 +89,122 @@ const COST_POOL_CONCURRENCY = Math.max(
parseInt(process.env.IMPORT_BATCH_COST_CONCURRENCY || '5', 10) || 5,
);
-let tablesReady = false;
+/**
+ * #2 (review 2026-09-16): process-wide Shopify WRITE lock. The 1s in-loop delay keeps
+ * writes serial WITHIN a job, but POST returns 202 + fire-and-forget (setImmediate), so
+ * two overlapping batch submits would run two create loops AT ONCE and jointly exceed
+ * Shopify Standard's 2 req/s ceiling. This module-level promise chains every job's create
+ * loop so only ONE runs at a time in this process — writes stay serial ACROSS jobs, not
+ * just within one. A second job simply awaits its turn (fine: it is a background job with
+ * status polling). Acquire = reassign the chain to a fresh promise and await the previous;
+ * release = resolve that promise for the next waiter (done in a finally).
+ */
+let shopifyWriteLock: Promise<void> = Promise.resolve();
-/** Create the job tables if they don't exist (idempotent). Mirrors database/create-batch-import-jobs.sql. */
+/**
+ * TEST-ONLY: reset the process-wide write lock between jest cases. Fire-and-forget
+ * workers (setImmediate) can still hold the lock when a test that didn't fully drain them
+ * ends; without this, that stale hold would block the next test's worker. Never called in
+ * production. Detaching is safe: the still-running worker resolves its OWN captured promise,
+ * which nothing is awaiting once the chain head is reset.
+ */
+export function __resetShopifyWriteLockForTests(): void {
+ shopifyWriteLock = Promise.resolve();
+}
+
+/** How many times to retry a Shopify 429 (rate-limit) before giving up. Env-overridable. */
+const RATE_LIMIT_RETRIES = Math.max(0, parseInt(process.env.IMPORT_BATCH_429_RETRIES || '3', 10) || 3);
+
+/**
+ * #2: create a Shopify product with in-loop retry on HTTP 429 (rate limit). shopify-api's
+ * request() throws `Shopify API Error (429): …` on a throttle. A 429 is returned BEFORE the
+ * product is persisted, so retrying is always safe (it can never mint a duplicate). Non-429
+ * errors rethrow IMMEDIATELY and flow through the caller's existing created/ambiguous/
+ * definitive classification unchanged. If 429s persist past the retry budget the final throw
+ * is still a 4xx → createDefinitivelyFailed() → the claim is released + the item marked
+ * failed (safe: 429 = not persisted → retryable on re-submit). This replaces the old
+ * behavior where the FIRST 429 immediately dropped the item.
+ */
+async function createWithRateLimitRetry(productData: any): Promise<any> {
+ for (let attempt = 0; ; attempt++) {
+ try {
+ return await shopifyAPI.instance.createProduct(productData);
+ } catch (err) {
+ const msg = err instanceof Error ? err.message : String(err);
+ const is429 = /Shopify API Error \(429\)/.test(msg);
+ if (!is429 || attempt >= RATE_LIMIT_RETRIES) throw err;
+ const backoffMs = 1000 * Math.pow(2, attempt); // 1s, 2s, 4s …
+ console.warn(`⏳ Shopify 429 (attempt ${attempt + 1}/${RATE_LIMIT_RETRIES}) — backing off ${backoffMs}ms`);
+ await new Promise((r) => setTimeout(r, backoffMs));
+ }
+ }
+}
+
+/**
+ * H3 (TK-11786): the owner token that scopes the boot-time orphan reclaim to THIS
+ * environment. dev and prod share dw_unified, so an unscoped boot sweep on a dev
+ * process would mark prod's genuinely-running jobs 'failed'. The token must be
+ * STABLE across restarts of the same environment but DIFFERENT between dev and prod:
+ * an explicit IMPORT_JOB_OWNER wins; otherwise the machine hostname separates the
+ * Kamatera prod box from a Mac dev box automatically (no config needed). Stamped on
+ * every job at createJob and matched exactly by reclaimOrphanedJobs.
+ *
+ * DEPLOY NOTE: the hostname fallback assumes a STABLE hostname across restarts (true for
+ * the PM2-on-fixed-host Kamatera target). In an ephemeral-hostname environment (a
+ * container that gets a fresh random hostname each boot) the fallback would change every
+ * restart, so a crashed job's owner would never match the new boot and never be reclaimed
+ * — set IMPORT_JOB_OWNER to a stable per-deployment value there.
+ */
+export function resolveJobOwner(): string {
+ return String(process.env.IMPORT_JOB_OWNER || os.hostname() || 'unknown').slice(0, 200);
+}
+
+/**
+ * H1 (TK-11786): the dedup key a product is CLAIMED under. A real SKU is the key
+ * (unchanged behavior). An SKU-LESS product (quote-only line, no vendor SKU) now
+ * derives a COMPOSITE key from vendor+title+collection so a re-submit of the same
+ * batch can't re-mint it — the hole the F2 SKU-only claim left wide open.
+ *
+ * Normalization (trim + lowercase + whitespace-collapse) makes the key stable across
+ * incidental formatting differences between submits. TITLE is the discriminating
+ * field: with no SKU AND no title there is no reliable key, so we return '' and the
+ * import proceeds UNGUARDED (matching the old SKU-less behavior) rather than risk a
+ * too-coarse key that would falsely skip distinct products. CLAUDE.md forbids blank/
+ * "Unknown" titles, so a real import always has a title to key on.
+ */
+export function deriveDedupKey(p: {
+ sku?: string;
+ vendor?: string;
+ title?: string;
+ collection?: string;
+}): string {
+ const sku = (p.sku || '').trim();
+ if (sku) return sku; // real SKU — unchanged behavior
+ const norm = (s?: string) => (s || '').trim().toLowerCase().replace(/\s+/g, ' ');
+ const title = norm(p.title);
+ if (!title) return ''; // no reliable key → proceed unguarded (old SKU-less behavior)
+ return `composite:${norm(p.vendor)}|${title}|${norm(p.collection)}`;
+}
+
+// Single-flight init: the DDL + one-time boot reclaim run EXACTLY once per process,
+// and every caller (createJob / getJobStatus) awaits the SAME promise — so a fresh
+// submit can never race the reclaim (its job row is inserted only after init resolves,
+// and the reclaim runs strictly inside init, never again).
+let initPromise: Promise<void> | null = null;
+
+/** Ensure the job tables exist AND reclaim orphaned jobs — once per process. */
export async function ensureBatchJobTables(): Promise<void> {
- if (tablesReady) return;
+ if (!initPromise) {
+ initPromise = initBatchJobStore().catch((err) => {
+ initPromise = null; // failed init: let the next call retry rather than wedge forever
+ throw err;
+ });
+ }
+ return initPromise;
+}
+
+/** Create the job tables if they don't exist (idempotent). Mirrors database/create-batch-import-jobs.sql. */
+async function initBatchJobStore(): Promise<void> {
await query(`
CREATE TABLE IF NOT EXISTS batch_import_jobs (
job_id UUID PRIMARY KEY,
@@ -75,29 +212,146 @@ export async function ensureBatchJobTables(): Promise<void> {
total INTEGER NOT NULL DEFAULT 0,
vendor_id TEXT,
private_label BOOLEAN DEFAULT FALSE,
+ owner TEXT,
error TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
`);
+ // H3 (TK-11786): upgrade a pre-existing table (created before the owner column) —
+ // idempotent, no-op once present. Owner scopes the boot reclaim to one environment.
+ await query(`ALTER TABLE batch_import_jobs ADD COLUMN IF NOT EXISTS owner TEXT`);
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(),
+ 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,
+ create_started_at TIMESTAMPTZ,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (job_id, idx)
)
`);
+ // H2 (TK-11786): upgrade a pre-existing table — create_started_at is the durable
+ // "a Shopify create was attempted" marker that lets the sweeper tell a claimed-but-
+ // -never-created item (safe to release + re-import) from one mid-create at the crash
+ // (unknown Shopify state — keep the claim, flag for a human).
+ await query(`ALTER TABLE batch_import_job_items ADD COLUMN IF NOT EXISTS create_started_at TIMESTAMPTZ`);
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;
+ // TK-11776 F2 (+ TK-11786 H1): atomic cross-job dedup ledger. PRIMARY KEY(sku) is the
+ // concurrency arbiter — claimKey() INSERTs here ON CONFLICT DO NOTHING so exactly one
+ // racing job wins the key and the rest skip the Shopify create (no duplicate mint). The
+ // `sku` column holds the dedup KEY (real SKU, or a composite vendor+title+collection key).
+ await query(`
+ CREATE TABLE IF NOT EXISTS batch_import_created_skus (
+ sku TEXT PRIMARY KEY,
+ job_id UUID,
+ idx INTEGER,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )
+ `);
+ await reclaimOrphanedJobs(resolveJobOwner());
+}
+
+/**
+ * Boot-time orphan reclaim (TK-11776 follow-up; hardened under TK-11786).
+ * Runs EXACTLY ONCE per process, inside the single-flight init above, BEFORE this process
+ * creates any job. Marks every non-terminal ('pending'/'running') job left by a
+ * crashed/restarted process 'failed' with an actionable error, so a /status poller gets a
+ * TERMINAL state instead of polling a forever-'running' job, and the interrupted partial
+ * import becomes visible + fixable. We deliberately DO NOT auto-recreate the missing
+ * products — that would fire customer-facing Shopify writes with no human in the loop.
+ *
+ * H3 — OWNER-SCOPED (TK-11786): the reclaim is filtered to `owner = this instance` so a
+ * dev boot can NEVER mark a prod (or another env's) live job 'failed'. dev and prod share
+ * dw_unified; without this scope a `npm run dev` boot stomped prod's in-flight batch. The
+ * owner token is stable per environment (IMPORT_JOB_OWNER || hostname), so a genuine
+ * prod restart still reclaims prod's OWN orphans. (Two processes sharing one owner token —
+ * e.g. two PM2 procs on one host — remains the single-instance assumption; that is a
+ * per-worker boot-id follow-up, out of scope here.)
+ *
+ * H2 — CRASH-SAFE CLAIM HANDLING (TK-11786): a claim is INSERTed before the Shopify create,
+ * so after a crash a claimed SKU may or may not actually exist in Shopify. We split the
+ * orphaned items by their durable create_started_at marker:
+ * • create_started_at IS NULL → the create never began; the product provably does NOT
+ * exist, so we RELEASE its dedup claim so a re-submit re-imports it (fixes the
+ * crash-ordering false-fail where a never-created SKU was skipped forever).
+ * • create_started_at IS NOT NULL → a create was in flight at the crash; Shopify's state
+ * is UNKNOWN, so we KEEP the claim (never auto-mint a duplicate — DW's cardinal rule)
+ * and flag the item so a human verifies and clears the one claim row if needed.
+ *
+ * Best-effort: a reclaim failure is logged, not thrown — it must never block table init or
+ * new submits (the stale rows simply persist until the next boot).
+ */
+export async function reclaimOrphanedJobs(owner: string): Promise<void> {
+ try {
+ // MIGRATION (owner-NULL legacy rows): the H3 owner scope (`owner = $1`) never matched
+ // rows that PREDATE the owner column — those were left NULL by the pre-upgrade code, so a
+ // job stuck 'pending'/'running' from before the upgrade was NEVER reclaimed and polled a
+ // /status endpoint as forever-'running'. Reclaiming `owner IS NULL` from ANY instance is
+ // safe because it CANNOT be another environment's LIVE job: every job created after the
+ // owner column shipped is stamped a non-null owner at createJob, so a NULL owner is by
+ // construction a stale pre-upgrade orphan whose creating process is long gone. The UPDATE
+ // is atomic, so if dev+prod boot at once the first flips the rows out of ('pending','running')
+ // and the second matches nothing — no double-processing.
+ const orphaned = await query<{ job_id: string }>(
+ `UPDATE batch_import_jobs
+ SET status = 'failed',
+ error = 'Worker interrupted by a process restart before completion; some products may not have been created. Re-submit the batch to import the remainder (already-created SKUs are skipped as duplicates).',
+ updated_at = now()
+ WHERE status IN ('pending', 'running') AND (owner = $1 OR owner IS NULL)
+ RETURNING job_id`,
+ [owner],
+ );
+ if (orphaned.length === 0) return;
+ const ids = orphaned.map((r) => r.job_id);
+
+ // (H2 step 1) IN-FLIGHT items (create was attempted, outcome unrecorded): KEEP the
+ // dedup claim and flag them — Shopify may or may not have the product; a human verifies.
+ await query(
+ `UPDATE batch_import_job_items
+ SET status = 'failed',
+ error = 'Interrupted mid-create by a process restart — this product MAY already exist in Shopify. Its dedup claim is RETAINED to prevent a duplicate. Verify in Shopify; if it was NOT created, delete its row from batch_import_created_skus and re-submit.',
+ updated_at = now()
+ WHERE job_id = ANY($1::uuid[]) AND status = 'pending' AND create_started_at IS NOT NULL`,
+ [ids],
+ );
+
+ // (H2 step 2) NOT-STARTED items (claimed but the create never began): RELEASE their
+ // claims so a re-submit actually re-imports them. Scoped to this owner's orphaned jobs
+ // and to items that are neither confirmed-created nor confirmed-blocked, so a genuinely
+ // created SKU's claim is never dropped.
+ await query(
+ `DELETE FROM batch_import_created_skus c
+ USING batch_import_job_items i
+ WHERE c.job_id = i.job_id
+ AND c.idx = i.idx
+ AND i.job_id = ANY($1::uuid[])
+ AND i.status NOT IN ('created', 'blocked')
+ AND i.create_started_at IS NULL`,
+ [ids],
+ );
+
+ // (H2 step 3) Flag the remaining still-'pending' (not-started) items so counts stay
+ // coherent. 'created'/'blocked' items are left intact — they stay dedup-protected.
+ await query(
+ `UPDATE batch_import_job_items
+ SET status = 'failed',
+ error = COALESCE(error, 'orphaned by process restart (create not started; dedup claim released for re-import)'),
+ updated_at = now()
+ WHERE status = 'pending' AND job_id = ANY($1::uuid[])`,
+ [ids],
+ );
+ console.warn(`♻️ [batch-import] Reclaimed ${orphaned.length} orphaned job(s) for owner "${owner}" (marked failed; re-submit to finish).`);
+ } catch (err) {
+ console.error('⚠️ [batch-import] orphan reclaim failed (non-fatal):', err);
+ }
}
/**
@@ -111,23 +365,33 @@ export async function createJob(jobId: string, req: BatchJobRequest): Promise<vo
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],
+ `INSERT INTO batch_import_jobs (job_id, status, total, vendor_id, private_label, owner)
+ VALUES ($1, 'pending', $2, $3, $4, $5)`,
+ [jobId, req.products.length, req.vendorId, req.privateLabel, resolveJobOwner()],
);
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,
- );
+ // #3 (review 2026-09-16): CHUNK the multi-row INSERT. At 4 bind params/item a single
+ // INSERT overflows Postgres's hard 65,535-parameter cap once IMPORT_BATCH_MAX (the
+ // env-configurable route cap, default 150) is raised past ~16k, throwing the whole
+ // job-seed transaction (→ 500, no job). 500 rows/chunk = 2,000 params — safely
+ // independent of the batch-size cap. All chunks run inside the SAME transaction, so
+ // the seed stays atomic (any failure rolls back the job + every item).
+ const ITEM_INSERT_CHUNK = 500;
+ for (let start = 0; start < req.products.length; start += ITEM_INSERT_CHUNK) {
+ const slice = req.products.slice(start, start + ITEM_INSERT_CHUNK);
+ const tuples: string[] = [];
+ const values: any[] = [];
+ slice.forEach((p, j) => {
+ const b = j * 4;
+ tuples.push(`($${b + 1}, $${b + 2}, $${b + 3}, $${b + 4}, 'pending')`);
+ values.push(jobId, start + j, 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,
+ );
+ }
});
}
@@ -154,27 +418,112 @@ async function setItem(
}
/**
- * 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.
+ * Atomic cross-job dedup (TK-11776 F2; SKU-less coverage added TK-11786 H1). The in-repo
+ * stand-in for the Kamatera lib/sku-registry.checkDuplicate (referenced in CLAUDE.md, NOT
+ * present here).
+ *
+ * The prior guard was a plain `SELECT ... WHERE status IN ('created','blocked')`
+ * followed by a create — a TOCTOU: two concurrent/overlapping jobs (or a user
+ * re-submitting a batch that looks stuck while the first is still 'running') both
+ * read "not yet created" and both mint the product, since nothing serializes the gap
+ * and there is no uniqueness constraint. Instead we CLAIM a dedup key in a dedicated
+ * table whose PRIMARY KEY makes the claim atomic: `INSERT ... ON CONFLICT DO
+ * NOTHING RETURNING sku` hands a row to the single winner only; every other caller
+ * is told the key is taken and skips the Shopify create entirely. A kept claim IS
+ * the durable dedup ledger for a successfully-created product.
+ *
+ * The key is deriveDedupKey(product): a real SKU, or a composite vendor+title+collection
+ * key for SKU-less products (H1). The `sku` COLUMN stores whichever key applies — it is
+ * the dedup KEY, not necessarily a vendor SKU. Empty key => nothing reliable to dedup on:
+ * returns true so the import proceeds unguarded (matching prior SKU-less behavior).
+ * Fail-OPEN on an infra error (returns true) so a claim-table blip never blocks an import.
*/
-async function skuAlreadyImported(sku: string): Promise<boolean> {
- if (!sku) return false;
+async function claimKey(key: string, jobId: string, idx: number): Promise<boolean> {
+ if (!key) return true; // nothing to dedup on — proceed
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],
+ const rows = await query<{ sku: string }>(
+ `INSERT INTO batch_import_created_skus (sku, job_id, idx)
+ VALUES ($1, $2, $3)
+ ON CONFLICT (sku) DO NOTHING
+ RETURNING sku`,
+ [key, jobId, idx],
);
- return rows.length > 0;
- } catch {
+ return rows.length > 0; // won the claim iff a row was actually inserted
+ } catch (err) {
+ console.error(JSON.stringify({ event: 'batch_claim_error', key, error: err instanceof Error ? err.message : String(err) }));
+ return true; // fail-open — never block an import on a dedup-infra error
+ }
+}
+
+/**
+ * Release a claim so a RETRYABLE failure (Shopify returned null / threw before the
+ * product existed) can be re-attempted on the next submit. Deliberately NOT called
+ * after the create SUCCEEDED — a claim kept then safely BLOCKS a re-mint (DW's cardinal
+ * rule: never auto-mint a duplicate). On a hard process crash the claim is released ONLY
+ * by the boot sweeper, and only for items whose create never started (create_started_at
+ * IS NULL); an in-flight item's claim is retained. Best-effort: a release failure is
+ * logged, never thrown.
+ */
+async function releaseKey(key: string): Promise<void> {
+ if (!key) return;
+ try {
+ await query(`DELETE FROM batch_import_created_skus WHERE sku=$1`, [key]);
+ } catch (err) {
+ console.error(JSON.stringify({ event: 'batch_claim_release_error', key, error: err instanceof Error ? err.message : String(err) }));
+ }
+}
+
+/**
+ * H2 (TK-11786): durably stamp the "a Shopify create was attempted for this item" marker
+ * IMMEDIATELY before calling createProduct — and return whether that write SUCCEEDED. The
+ * sweeper's crash-safety rests on a hard invariant: `create_started_at is durably set
+ * BEFORE createProduct is called, or createProduct is NOT called`. If the marker write
+ * fails we return false and the caller SKIPS the create (fails the item, releases the
+ * claim, retry on re-submit) — because a create that ran with the marker still NULL could
+ * be crash-released by the sweeper and then re-minted as a duplicate. Failing closed here
+ * (skip one item, retry later) is strictly safer than risking a duplicate.
+ */
+async function markCreateStarted(jobId: string, idx: number): Promise<boolean> {
+ try {
+ // RETURN the row so we can prove EXACTLY the target item transitioned. A bare
+ // UPDATE returning true on no-throw silently satisfied the "marker set before
+ // createProduct" invariant even on a 0-row match (missing item row, idx skew, or
+ // create_started_at already non-NULL) — in which case createProduct would run with
+ // the marker still NULL, and a crash would let the sweeper (create_started_at IS
+ // NULL branch) RELEASE the claim of a product that WAS created → duplicate on
+ // re-submit. Requiring rowCount === 1 closes that: on 0 rows we return false and the
+ // caller skips the create + releases the claim (retry on re-submit), never risking a dup.
+ const rows = await query<{ idx: number }>(
+ `UPDATE batch_import_job_items SET create_started_at = now(), updated_at = now()
+ WHERE job_id=$1 AND idx=$2 AND create_started_at IS NULL
+ RETURNING idx`,
+ [jobId, idx],
+ );
+ return rows.length === 1;
+ } catch (err) {
+ console.error(JSON.stringify({ event: 'batch_mark_create_started_error', jobId, idx, error: err instanceof Error ? err.message : String(err) }));
return false;
}
}
+/**
+ * H2 (TK-11786, folded in from a Kimi second-model review): classify a createProduct
+ * failure as DEFINITELY-not-created vs AMBIGUOUS. shopify-api.request() throws on any
+ * non-2xx with the status embedded ("Shopify API Error (NNN): …") and throws a bare
+ * network/timeout error otherwise. A 4xx (validation/auth/not-found/429-rate-limit) means
+ * Shopify REJECTED the request before persisting a product — safe to release the claim and
+ * retry. A 5xx, or a network error / timeout with NO HTTP status, is AMBIGUOUS: the write
+ * may have reached Shopify and created the product before the response was lost, so the
+ * claim MUST be retained (never auto-mint a duplicate) and the item flagged for a human.
+ */
+function createDefinitivelyFailed(err: unknown): boolean {
+ const msg = err instanceof Error ? err.message : String(err);
+ const m = msg.match(/Shopify API Error \((\d{3})\)/);
+ if (!m) return false; // no HTTP status → network/timeout → AMBIGUOUS → keep the claim
+ const code = parseInt(m[1], 10);
+ return code >= 400 && code < 500; // any 4xx = request rejected, product not created → release
+}
+
/**
* 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
@@ -210,6 +559,45 @@ async function preResolveCosts(
return out;
}
+/**
+ * #1 (review 2026-09-16): CATALOG-WIDE duplicate pre-check — the hole the per-job claim
+ * ledger (batch_import_created_skus) leaves open. That ledger only records THIS route's own
+ * creations, so a batch re-import of an mfr SKU already created via the single-import path,
+ * via cadence, or before the ledger existed would sail past the claim and MINT A DUPLICATE
+ * (the single-import path avoids this by going through assignDwSku/createOrGetProduct, which
+ * the batch worker does not). This mirrors app/api/shopify/check-duplicates-batch: test every
+ * real mfr SKU against the THREE authoritative sources in ONE query, case-insensitive, and
+ * return the set of lowercased SKUs that already exist. Fail-OPEN (empty set) on any DB error
+ * so a hiccup never blocks an import — same posture as check-duplicates-batch and the rest of
+ * this file. SKU-less products cannot be checked here (the catalog is keyed on mfr SKU); they
+ * stay covered by the composite-key claim ledger (H1). scraped_products is joined only where
+ * shopify_id IS NOT NULL, so a mid-flight scrape artifact never masks a real product.
+ */
+async function preResolveCatalogDuplicates(products: BatchProduct[]): Promise<Set<string>> {
+ const keys = Array.from(
+ new Set(products.map((p) => (p.sku || '').trim().toLowerCase()).filter(Boolean)),
+ );
+ if (keys.length === 0) return new Set();
+ try {
+ const rows = await query<{ k: string }>(
+ `SELECT lower(mfr_sku) AS k FROM dw_sku_registry WHERE lower(mfr_sku) = ANY($1)
+ UNION
+ SELECT lower(mfr_sku) AS k FROM shopify_products WHERE lower(mfr_sku) = ANY($1)
+ UNION
+ SELECT lower(sku) AS k FROM shopify_products WHERE lower(sku) = ANY($1)
+ UNION
+ SELECT lower(variant_sku) AS k FROM shopify_products WHERE lower(variant_sku) = ANY($1)
+ UNION
+ SELECT lower(sku) AS k FROM scraped_products WHERE lower(sku) = ANY($1) AND shopify_id IS NOT NULL`,
+ [keys],
+ );
+ return new Set(rows.map((r) => r.k).filter(Boolean));
+ } catch (err) {
+ console.error(JSON.stringify({ event: 'batch_catalog_dup_precheck_error', error: err instanceof Error ? err.message : String(err) }));
+ return new Set(); // fail-open — never block an import on a dedup-infra error
+ }
+}
+
/**
* 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
@@ -222,18 +610,61 @@ export async function runJob(jobId: string, req: BatchJobRequest): Promise<void>
// ── bounded-concurrency cost pre-resolution (the ONLY parallelized step) ──
const costByIdx = await preResolveCosts(products, vendorId, costTimeoutMs);
+ // #1 (review 2026-09-16): one read-only round-trip resolving which real mfr SKUs already
+ // exist ANYWHERE in the catalog, so the loop can skip them instead of minting duplicates.
+ // Runs BEFORE the write lock (it is a read; no reason to serialize it across jobs).
+ const catalogDupes = await preResolveCatalogDuplicates(products);
// ── SERIAL create loop (Shopify 2 req/s — writes are never parallelized) ──
+ // #2 (review 2026-09-16): acquire the process-wide Shopify WRITE lock so two overlapping
+ // jobs' create loops can't run concurrently and jointly exceed 2 req/s. The for-loop body
+ // below is intentionally left at its original indentation to keep this a minimal diff; the
+ // try/finally guarantees the lock is released even if the loop throws unexpectedly.
+ const prevWriteLock = shopifyWriteLock;
+ let releaseWriteLock!: () => void;
+ shopifyWriteLock = new Promise<void>((r) => { releaseWriteLock = r; });
+ await prevWriteLock;
+ try {
for (let idx = 0; idx < products.length; idx++) {
const product = products[idx];
+ let created = false; // flips true once Shopify HAS the product — gates claim release on error
+ let holdsClaim = false; // true once WE win our OWN dedup claim — gates releaseKey so a
+ // loser (skipped-duplicate) path can never release the WINNER's claim
+ let markerSet = false; // true once create_started_at is durably stamped (create was attempted);
+ // mirrors the boot sweeper's create_started_at axis (H2, TK-11786)
+ // H1 (TK-11786): the dedup key is the SKU when present, else a composite
+ // vendor+title+collection key so SKU-less quote items are deduped too (no re-mint
+ // on re-submit). '' => no reliable key → proceed unguarded (old SKU-less behavior).
+ const dedupKey = deriveDedupKey(product);
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})`);
+ // #1 (review 2026-09-16): CATALOG-WIDE dedup — skip a real mfr SKU that already exists
+ // anywhere in the catalog (dw_sku_registry / shopify_products / scraped_products), which
+ // the per-job claim ledger below cannot see. Do this BEFORE claiming so we neither claim
+ // nor create a duplicate. SKU-less products fall through to the composite-key claim.
+ const catalogKey = (product.sku || '').trim().toLowerCase();
+ if (catalogKey && catalogDupes.has(catalogKey)) {
+ await setItem(jobId, idx, 'skipped-duplicate', null,
+ `SKU ${product.sku} already exists in the catalog (dw_sku_registry/shopify_products/scraped_products) — not re-minted`);
+ console.log(`⏭️ Skipped catalog duplicate: ${product.title} (${product.sku})`);
continue;
}
+ // Dedup (TK-11776 F2 + TK-11786 H1): atomically CLAIM the dedup key before doing any
+ // work. The claim's PRIMARY KEY serializes concurrent/overlapping jobs — the loser is
+ // told the key is taken and skips the Shopify create, so a retry (or a second batch
+ // racing the same product) can never mint a duplicate. Empty key => no claim.
+ if (dedupKey) {
+ if (!(await claimKey(dedupKey, jobId, idx))) {
+ await setItem(jobId, idx, 'skipped-duplicate', null,
+ product.sku
+ ? `SKU ${product.sku} already imported (claimed by a prior/concurrent job)`
+ : `Already imported (matched on vendor+title+collection by a prior/concurrent job)`);
+ console.log(`⏭️ Skipped duplicate: ${product.title}${product.sku ? ` (${product.sku})` : ' (SKU-less; composite match)'}`);
+ continue;
+ }
+ holdsClaim = true; // WE won this key — safe to release it if the create later provably fails
+ }
+
// 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
@@ -319,26 +750,80 @@ export async function runJob(jobId: string, req: BatchJobRequest): Promise<void>
}],
};
- // Create product in Shopify (SERIAL).
- const result = await shopifyAPI.instance.createProduct(productData);
+ // H2 (TK-11786): durably stamp the create-attempted marker IMMEDIATELY before the
+ // Shopify write. If the process dies during createProduct, the boot sweeper sees
+ // create_started_at set and RETAINS the claim (Shopify state unknown — never
+ // auto-mint a dup); if it dies before this line, the claim is released for re-import.
+ // HARD INVARIANT: never call createProduct with the marker unset — if the marker
+ // write fails, skip the create (a create that ran while create_started_at was still
+ // NULL could be crash-released by the sweeper and re-minted as a duplicate).
+ if (!(await markCreateStarted(jobId, idx))) {
+ await releaseKey(dedupKey); // never created — free the claim so a re-submit retries
+ await setItem(jobId, idx, 'failed', null, 'Could not persist the create-started marker; skipped the Shopify create to avoid an untracked duplicate after a restart. Re-submit to retry.');
+ console.error(`❌ Marker write failed, skipped create: ${product.title}`);
+ continue;
+ }
+ markerSet = true; // create_started_at is durably stamped — from here, a throw is an UNKNOWN
+ // Shopify outcome (matching the sweeper's in-flight branch), NOT "never created".
+
+ // Create product in Shopify (SERIAL, under the process-wide write lock). #2: retries
+ // in-loop on a 429 (rate limit) with backoff instead of dropping the item on the first
+ // throttle; a 429 is returned before persistence so the retry can never mint a duplicate.
+ const result = await createWithRateLimitRetry(productData);
if (result) {
+ created = true; // Shopify HAS it — keep the claim as the permanent dedup record
// '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}`);
+ // Falsy result without a throw is an unexpected "2xx but no product" — treat as
+ // AMBIGUOUS (a product may exist): KEEP the claim + flag, never risk a re-mint.
+ await setItem(jobId, idx, 'failed', null, 'Create returned no product (ambiguous outcome). The dedup claim is RETAINED to prevent a duplicate. Verify in Shopify; if it was NOT created, delete its row from batch_import_created_skus and re-submit.');
+ console.error(`❌ Failed (ambiguous, no product returned): ${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');
+ // FINDING A (TK-11786 hardening): the dedup claim is released ONLY when the product
+ // PROVABLY does not exist in Shopify. The decisive axis is the durable create_started_at
+ // marker, so the in-process failure path is CONSISTENT with the boot sweeper (which keys
+ // purely off that marker):
+ // • created === true → createProduct returned; a LATER step (setItem/…) threw.
+ // The product EXISTS → KEEP the claim (dedup-protected).
+ // • !markerSet → the throw happened BEFORE create was attempted (marker
+ // never stamped), so the product provably does NOT exist. Release OUR claim (guarded by
+ // holdsClaim so a skipped-duplicate path can't drop the WINNER's claim) → re-submit retries.
+ // Mirrors the sweeper's create_started_at IS NULL branch.
+ // • markerSet && 4xx → createProduct was attempted but Shopify REJECTED it
+ // (4xx = not persisted) → provably not created → release + retry.
+ // • markerSet && (5xx/network/timeout/unknown) → createProduct was attempted and its
+ // outcome is UNKNOWN (the response may have been lost AFTER Shopify persisted). KEEP the
+ // claim (never auto-mint a duplicate — DW's cardinal rule) and flag for a human. Mirrors
+ // the sweeper's create_started_at IS NOT NULL (in-flight) branch. This is the case
+ // Finding A protects: a lost/timed-out response after a successful persist must NOT release.
+ const msg = error instanceof Error ? error.message : 'Unknown error';
+ if (created) {
+ // Created in Shopify but a post-create step threw — keep the claim (dedup-protected).
+ await setItem(jobId, idx, 'failed', null, `Created in Shopify but a post-create step failed (${msg}). Dedup claim retained; the product exists.`);
+ } else if (!markerSet) {
+ if (holdsClaim) await releaseKey(dedupKey); // create never attempted → provably not created → retryable
+ await setItem(jobId, idx, 'failed', null, msg);
+ } else if (createDefinitivelyFailed(error)) {
+ await releaseKey(dedupKey); // marker set but Shopify returned a 4xx → not persisted → retryable
+ await setItem(jobId, idx, 'failed', null, msg);
+ } else {
+ // Marker set + AMBIGUOUS failure — the write may have landed. RETAIN the claim.
+ await setItem(jobId, idx, 'failed', null, `Create failed AMBIGUOUSLY (${msg}). The product MAY exist in Shopify; the dedup claim is RETAINED to prevent a duplicate. Verify in Shopify; if it was NOT created, delete its row from batch_import_created_skus and re-submit.`);
+ }
console.error(`❌ Error importing ${product.title}:`, error);
}
}
+ } finally {
+ releaseWriteLock(); // #2: hand the Shopify write lock to the next waiting job
+ }
await setJobStatus(jobId, 'completed');
console.log(`🎯 Batch import job ${jobId} complete`);
← de6d2ec5 auto-data-snapshot: 2026-09-16T08:57:47 (3 data files) — sho
·
back to Designer Wallcoverings
·
auto-data-snapshot: 2026-09-16T10:07:35 (3 data files) — sho c28d35df →