diff --git a/apps/importer/Scripts/enqueue-shopee.ts b/apps/importer/Scripts/enqueue-shopee.ts
new file mode 100644
index 00000000..fd938005
--- /dev/null
+++ b/apps/importer/Scripts/enqueue-shopee.ts
@@ -0,0 +1,78 @@
+#!/usr/bin/env node
+/**
+ * Enqueue Shopee product URLs as rate.crawl_jobs rows (via enqueue_crawl_job RPC).
+ *
+ * Usage:
+ *   tsx Scripts/enqueue-shopee.ts <url1> [url2 ...]
+ *   tsx Scripts/enqueue-shopee.ts --file targets.txt
+ */
+import fs from 'node:fs';
+import dotenv from 'dotenv';
+import { createClient } from '@supabase/supabase-js';
+
+dotenv.config();
+
+const admin = createClient(process.env.SUPABASE_URL!, process.env.SUPABASE_SERVICE_ROLE_KEY!, {
+  auth: { autoRefreshToken: false, persistSession: false },
+});
+const db = (admin as any).schema('rate');
+
+function parseItem(url: string): { itemId: string; shopId: string } | null {
+  const m = url.match(/(?:a-)?i\.(\d+)\.(\d+)/);
+  return m ? { shopId: m[1], itemId: m[2] } : null;
+}
+
+async function main() {
+  const args = process.argv.slice(2);
+  let urls: string[] = [];
+  if (args[0] === '--file') {
+    urls = fs.readFileSync(args[1], 'utf-8').split(/\r?\n/).map((l) => l.trim()).filter((l) => l && !l.startsWith('#'));
+  } else {
+    urls = args.filter((a) => a.startsWith('http'));
+  }
+  if (urls.length === 0) {
+    console.error('Usage: tsx Scripts/enqueue-shopee.ts <url1> [url2...] | --file targets.txt');
+    process.exit(1);
+  }
+
+  const { data: runId, error: runErr } = await db.rpc('start_crawl_run', { p_source: 'enqueue-cli', p_params: { count: urls.length } });
+  if (runErr || !runId) {
+    console.error(`[Enqueue] start_crawl_run failed: ${runErr?.message}`);
+    process.exit(1);
+  }
+  console.log(`[Enqueue] run=${runId}`);
+
+  let failures = 0;
+  for (const url of urls) {
+    const ids = parseItem(url);
+    if (!ids) {
+      console.error(`[Enqueue] SKIP unparseable: ${url}`);
+      failures += 1;
+      continue;
+    }
+    const { data, error } = await db.rpc('enqueue_crawl_job', {
+      p_type: 'crawl_listing',
+      p_platform: 'shopee',
+      p_payload: { schema_version: 1, url },
+      p_target_url: url,
+      p_platform_item_id: ids.itemId,
+      p_platform_shop_id: ids.shopId,
+      p_priority: 0,
+      p_run_id: runId,
+    });
+    if (error) {
+      console.error(`[Enqueue] FAIL ${ids.itemId}: ${error.message}`);
+      failures += 1;
+    } else {
+      console.log(`[Enqueue] job=${data[0]?.job_id ?? data?.job_id} state=${data[0]?.state ?? data?.state} item=${ids.itemId}`);
+    }
+  }
+  // Non-zero exit when anything failed so automation cannot mistake a
+  // partially-failed or fully-skipped enqueue for success (Codex SHOULD).
+  process.exit(failures > 0 ? 1 : 0);
+}
+
+main().catch((e) => {
+  console.error(e.message);
+  process.exit(1);
+});
diff --git a/apps/importer/Scripts/run-queue-crawl.ts b/apps/importer/Scripts/run-queue-crawl.ts
new file mode 100644
index 00000000..fc9dc556
--- /dev/null
+++ b/apps/importer/Scripts/run-queue-crawl.ts
@@ -0,0 +1,168 @@
+#!/usr/bin/env node
+/**
+ * Queue-driven Shopee crawl driver (plan v7 §4.1 recipe):
+ *   1. Launch cloned-Chrome-Stable via CDP + optional 4G proxy (home IP is the
+ *      proven default for Shopee - pass USE_4G=1 to route through mproxy).
+ *   2. Warm up the store page (cold PDP jumps are what Shopee risk control flags).
+ *   3. Run the queue worker until the queue drains, then finish the run and
+ *      tear the browser down.
+ *
+ * Usage: tsx Scripts/run-queue-crawl.ts [--max-jobs N]
+ * Enqueue jobs first with Scripts/enqueue-shopee.ts.
+ */
+import fs from 'node:fs';
+import { execSync } from 'node:child_process';
+import dotenv from 'dotenv';
+import { chromium } from 'playwright';
+import { createClient } from '@supabase/supabase-js';
+import { ProxyManager } from '../src/shopee/proxy-manager.js';
+import { ShopeeScraper } from '../src/shopee/shopee-scraper.js';
+import { ShopeeIngestionPipeline } from '../src/shopee/shopee-to-supabase.js';
+import { QueueDrivenShopeeCrawler } from '../src/shopee/queue-worker.js';
+
+dotenv.config();
+
+const CDP = 'http://127.0.0.1:9222';
+const WORKER_ID = `worker-${process.pid}-${new Date().toISOString().slice(0, 16)}`;
+// Proven-safe cloned profile (Chrome refuses CDP on the default User Data dir).
+const PROFILE_DIR = process.env.CRAWL_PROFILE_DIR || 'C:\\chrome_cdp_profiles\\stable';
+
+// Single-driver assumption: this kills ANY chrome carrying CDP port 9222 that
+// uses one of our crawl profile dirs (never the user's own Chrome profiles).
+// Two drivers on one machine would fight over the port - run one at a time.
+function killCdpBrowser(): void {
+  try {
+    execSync(
+      `powershell -NoProfile -Command "Get-CimInstance Win32_Process -Filter \\"Name='chrome.exe'\\" | Where-Object { $_.CommandLine -like '*remote-debugging-port=9222*' -and $_.CommandLine -notlike '*--type=*' -and ($_.CommandLine -like '*chrome_cdp_profiles*' -or $_.CommandLine -like '*chrome_bot*') } | ForEach-Object { taskkill /PID $_.ProcessId /T /F 2>&1 | Out-Null }"`,
+      { stdio: 'ignore' }
+    );
+  } catch {}
+}
+
+async function main() {
+  const maxJobsArg = process.argv.indexOf('--max-jobs');
+  const parsed = maxJobsArg > -1 ? parseInt(process.argv[maxJobsArg + 1], 10) : NaN;
+  // NaN/absent -> unlimited (Codex SHOULD: NaN previously meant "process 0
+  // jobs and report success").
+  const maxJobs = Number.isFinite(parsed) && parsed > 0 ? parsed : undefined;
+  if (maxJobsArg > -1 && maxJobs === undefined) {
+    console.error('[Run] --max-jobs needs a positive integer; running until the queue drains.');
+  }
+  const use4g = process.env.USE_4G === '1';
+
+  const admin = createClient(process.env.SUPABASE_URL!, process.env.SUPABASE_SERVICE_ROLE_KEY!, {
+    auth: { autoRefreshToken: false, persistSession: false },
+  });
+  const db = (admin as any).schema('rate');
+
+  const pm = new ProxyManager();
+  if (use4g) {
+    const ip = await pm.getCurrentIp();
+    console.log(`[Run] 4G proxy IP: ${ip}`);
+  } else {
+    console.log('[Run] Direct home IP (proven recipe for Shopee).');
+  }
+
+  killCdpBrowser();
+
+  let chromeUp = false;
+  try {
+    if (use4g) {
+      chromeUp = await pm.launchChromeWithProxy(9222, PROFILE_DIR);
+    } else {
+      const { spawn } = await import('node:child_process');
+      const http = await import('node:http');
+      const proc = spawn(
+        'C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe',
+        ['--remote-debugging-port=9222', `--user-data-dir=${PROFILE_DIR}`, '--no-first-run', '--no-default-browser-check'],
+        { detached: true, stdio: 'ignore' }
+      );
+      proc.unref();
+      for (let i = 0; i < 20; i++) {
+        await new Promise((r) => setTimeout(r, 500));
+        chromeUp = await new Promise<boolean>((resolve) => {
+          http.get(`${CDP}/json/version`, (res) => resolve(res.statusCode === 200)).on('error', () => resolve(false));
+        });
+        if (chromeUp) break;
+      }
+    }
+  } catch (e: any) {
+    console.error(`[Run] Chrome launch failed: ${e.message}`);
+  }
+  if (!chromeUp) {
+    console.error('[Run] CDP Chrome did not come up; aborting.');
+    await pm.stopLocalForwarder().catch(() => {});
+    process.exit(1);
+  }
+  console.log('[Run] CDP Chrome is up.');
+
+  const { data: runId, error: runErr } = await db.rpc('start_crawl_run', {
+    p_source: 'queue-worker',
+    p_params: { worker: WORKER_ID, use4g, maxJobs: maxJobs ?? null },
+  });
+  if (runErr) {
+    console.error(`[Run] start_crawl_run failed: ${runErr.message}`);
+    killCdpBrowser();
+    await pm.stopLocalForwarder().catch(() => {});
+    process.exit(1);
+  }
+  console.log(`[Run] crawl run=${runId}`);
+
+  // Everything after run creation lives in try/finally: a setup crash must
+  // still finish the run and tear the browser down (Codex SHOULD).
+  let processed = 0;
+  try {
+    const b = await chromium.connectOverCDP(CDP);
+    const ctx = b.contexts()[0];
+    const scraper = new ShopeeScraper({ cdpEndpoint: CDP, maxReviews: 10, timeoutMs: 90000, proxy: use4g, autoRotateOnBlock: true });
+    const pipeline = new ShopeeIngestionPipeline();
+    await pipeline.init();
+
+    const worker = new QueueDrivenShopeeCrawler({
+      supabase: db,
+      pipeline,
+      workerId: WORKER_ID,
+      platform: 'shopee',
+      getRunId: () => runId,
+      maxJobs,
+      idlePollMs: 30000,
+      maxIdlePolls: 4,
+      scrape: async (job) => {
+        const url = job.target_url;
+        if (!url) throw new Error('job has no target_url');
+        // Human-like warm-up FIRST (cold PDP jump is what risk control flags).
+        const probe = await ctx.newPage();
+        try {
+          await probe.goto('https://shopee.vn/samsung_official_store', { waitUntil: 'domcontentloaded', timeout: 60000 });
+          await probe.waitForTimeout(3000);
+          await probe.evaluate(() => window.scrollBy({ top: 800, behavior: 'smooth' }));
+          await probe.waitForTimeout(1000);
+        } catch (e: any) {
+          console.warn(`[Run] store warm-up skipped: ${e.message.split('\n')[0]}`);
+        } finally {
+          await probe.close().catch(() => {});
+        }
+        return scraper.scrapeProduct(url);
+      },
+    });
+
+    processed = await worker.run();
+    console.log(`[Run] worker finished, processed=${processed}`);
+  } finally {
+    try {
+      const { error: finErr } = await db.rpc('finish_crawl_run', { p_run_id: runId, p_stats: { processed } });
+      if (finErr) console.warn(`[Run] finish_crawl_run failed: ${finErr.message}`);
+    } catch (e: any) {
+      console.warn(`[Run] finish_crawl_run threw: ${e.message}`);
+    }
+    killCdpBrowser(); // only CDP-carrying chrome (remote-debugging-port=9222)
+    await pm.stopLocalForwarder().catch(() => {});
+    console.log('[Run] Chrome closed, forwarder stopped.');
+  }
+  process.exit(0);
+}
+
+main().catch((e) => {
+  console.error('[Run] FATAL:', e.message);
+  process.exit(1);
+});
diff --git a/apps/importer/src/shopee/queue-worker.ts b/apps/importer/src/shopee/queue-worker.ts
new file mode 100644
index 00000000..d48d294c
--- /dev/null
+++ b/apps/importer/src/shopee/queue-worker.ts
@@ -0,0 +1,232 @@
+import type { ShopeeProduct } from './shopee-scraper.js';
+import { ShopeeIngestionPipeline } from './shopee-to-supabase.js';
+
+export interface QueueJobRow {
+  id: string;
+  type: string;
+  platform: string;
+  status: string;
+  payload: Record<string, unknown>;
+  target_url: string | null;
+  platform_item_id: string | null;
+  platform_shop_id: string | null;
+  attempts: number;
+  max_attempts: number;
+  claim_token: string;
+  worker_id: string | null;
+  progress: Record<string, unknown>;
+}
+
+export interface QueueWorkerOptions {
+  /** Supabase client already scoped to schema 'rate' (getRateClient-style). */
+  supabase: any;
+  /** The ingest pipeline (built once, reused across jobs). */
+  pipeline: ShopeeIngestionPipeline;
+  /** How a job's target URL is scraped. Returns the ShopeeProduct. */
+  scrape: (job: QueueJobRow) => Promise<ShopeeProduct>;
+  workerId: string;
+  platform: string;
+  /** Lease duration per claim (heartbeat extends it for long crawls). */
+  leaseMinutes?: number;
+  /** Heartbeat interval while scraping (must be < lease). */
+  heartbeatSeconds?: number;
+  /** Exit after this many processed jobs (default: run until queue empty). */
+  maxJobs?: number;
+  /** Sleep between polls when the queue is empty (ms). */
+  idlePollMs?: number;
+  /** Give up idle-waiting after this many consecutive empty polls. Jobs in
+   * exponential backoff may outlive this wait - a later run picks them up. */
+  maxIdlePolls?: number;
+  /** Default run id for ack'd jobs (start_crawl_run once per worker session). */
+  getRunId?: () => string | null;
+}
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+/**
+ * Queue-driven crawler worker (plan v7 §4.1): claims rate.crawl_jobs through
+ * the SECURITY DEFINER RPCs, heartbeats long scrapes, checkpoints section
+ * completion, and acks/fails with the per-claim token fence. Direct DML on
+ * crawl_jobs is revoked server-side, so these RPCs are the only interaction.
+ */
+export class QueueDrivenShopeeCrawler {
+  constructor(private opts: QueueWorkerOptions) {}
+
+  private db() {
+    return this.opts.supabase;
+  }
+
+  /** Claim exactly one job. Returns null when the queue is empty. */
+  public async dequeue(): Promise<QueueJobRow | null> {
+    const { data, error } = await this.db().rpc('dequeue_crawl_job', {
+      p_worker_id: this.opts.workerId,
+      p_platform: this.opts.platform,
+      p_lease: `${this.opts.leaseMinutes ?? 10} minutes`,
+    });
+    if (error) {
+      throw new Error(`dequeue_crawl_job failed: ${error.message}`);
+    }
+    const row = Array.isArray(data) ? data[0] : data;
+    return (row as QueueJobRow) ?? null;
+  }
+
+  public async heartbeat(job: QueueJobRow, extend = `${this.opts.leaseMinutes ?? 10} minutes`): Promise<boolean> {
+    const { data, error } = await this.db().rpc('heartbeat_crawl_job', {
+      p_id: job.id,
+      p_claim_token: job.claim_token,
+      p_extend: extend,
+    });
+    if (error) return false;
+    return data === true;
+  }
+
+  /** Raised when the lease was lost (heartbeat rejected or ack refused):
+   * this worker no longer owns the job and must stop touching it. */
+  static LeaseLost = 'LEASE_LOST';
+
+  public async checkpoint(job: QueueJobRow, progress: Record<string, unknown>): Promise<void> {
+    const { error } = await this.db().rpc('checkpoint_crawl_job', {
+      p_id: job.id,
+      p_claim_token: job.claim_token,
+      p_progress: { ...job.progress, ...progress },
+    });
+    if (error) {
+      // A checkpoint failure must not kill the job - it only degrades resume
+      // granularity. Surface loudly, keep going.
+      console.warn(`[QueueWorker] checkpoint failed for job ${job.id}: ${error.message}`);
+    }
+  }
+
+  public async ack(job: QueueJobRow, stats: Record<string, unknown>): Promise<boolean> {
+    const { data, error } = await this.db().rpc('ack_crawl_job', {
+      p_id: job.id,
+      p_claim_token: job.claim_token,
+      p_stats: stats,
+    });
+    if (error) throw new Error(`ack_crawl_job failed: ${error.message}`);
+    return data === true;
+  }
+
+  public async fail(job: QueueJobRow, errorMessage: string): Promise<boolean> {
+    const { data, error } = await this.db().rpc('fail_crawl_job', {
+      p_id: job.id,
+      p_claim_token: job.claim_token,
+      p_error: errorMessage.slice(0, 2000),
+    });
+    if (error) throw new Error(`fail_crawl_job failed: ${error.message}`);
+    return data === true;
+  }
+
+  /** Process one claimed job end-to-end: scrape -> ingest (checkpointed) -> ack.
+   * Lease enforcement (Codex blocker): a rejected heartbeat or a refused ack
+   * aborts the job with LEASE_LOST - a worker that lost its claim must never
+   * touch the listing again (the true owner is scraping it right now). */
+  public async processJob(job: QueueJobRow): Promise<void> {
+    const sections: Record<string, unknown> = { started_at: new Date().toISOString() };
+    await this.checkpoint(job, { started: true });
+
+    // Heartbeat while scraping (long): independent interval; the in-flight
+    // promise is tracked so teardown awaits it instead of cutting it mid-call.
+    const hbMs = (this.opts.heartbeatSeconds ?? 120) * 1000;
+    let leaseLost = false;
+    let hbInFlight: Promise<void> | null = null;
+    const hb = setInterval(() => {
+      // Skip this tick if the previous heartbeat is still in flight (Codex
+      // round-3: chaining queued an unbounded backlog when RPCs were slow,
+      // and teardown would then await all of it). Heartbeats are periodic;
+      // skipping a tick only shortens the effective lease, and the lease is
+      // comfortably longer than the interval. Once leaseLost fires the
+      // interval keeps ticking but does nothing useful - processJob's checks
+      // abandon the job at the next section boundary either way.
+      if (hbInFlight) return;
+      hbInFlight = this.heartbeat(job).then(
+        (ok) => {
+          if (!ok) {
+            leaseLost = true;
+            console.error(`[QueueWorker] lease LOST for job ${job.id}; abandoning work on next check`);
+          }
+        },
+        (e) => console.warn(`[QueueWorker] heartbeat error: ${e.message}`)
+      ).finally(() => {
+        hbInFlight = null;
+      });
+    }, hbMs);
+
+    try {
+      const product = await this.opts.scrape(job);
+      if (leaseLost) throw new Error(QueueDrivenShopeeCrawler.LeaseLost);
+      sections.title = product.title;
+
+      const result = await this.opts.pipeline.ingestProduct(product, async (section) => {
+        if (leaseLost) throw new Error(QueueDrivenShopeeCrawler.LeaseLost);
+        sections[section] = new Date().toISOString();
+        await this.checkpoint(job, { [`${section}_at`]: sections[section] });
+        // Re-check AFTER the awaited checkpoint: the lease can die while it
+        // is in flight, and the pipeline must not start the next section's
+        // writes once it has (Codex round-2).
+        if (leaseLost) throw new Error(QueueDrivenShopeeCrawler.LeaseLost);
+      });
+      if (leaseLost) throw new Error(QueueDrivenShopeeCrawler.LeaseLost);
+
+      sections.productId = result.productId;
+      sections.masterId = result.masterId;
+      const acked = await this.ack(job, sections);
+      if (!acked) throw new Error(QueueDrivenShopeeCrawler.LeaseLost);
+      console.log(`[QueueWorker] job ${job.id} ACKED (product ${result.productId})`);
+    } catch (err: any) {
+      const message = err?.message || String(err);
+      // A lost lease must NEVER reach fail_crawl_job even when the work
+      // itself threw a normal error afterwards (Codex round-2): the current
+      // owner owns the outcome.
+      if (message === QueueDrivenShopeeCrawler.LeaseLost || leaseLost) {
+        console.error(`[QueueWorker] job ${job.id} abandoned (lease lost mid-work).`);
+        return;
+      }
+      console.error(`[QueueWorker] job ${job.id} FAILED: ${message}`);
+      await this.fail(job, message);
+    } finally {
+      clearInterval(hb);
+      await Promise.resolve(hbInFlight).catch(() => {});
+    }
+  }
+
+  /**
+   * Main loop: dequeue -> process, until maxJobs reached or the queue stays
+   * empty for maxIdlePolls consecutive polls. Returns processed count.
+   */
+  public async run(): Promise<number> {
+    let processed = 0;
+    let idlePolls = 0;
+    const maxJobs = this.opts.maxJobs ?? Number.POSITIVE_INFINITY;
+    const idlePollMs = this.opts.idlePollMs ?? 30000;
+    const maxIdlePolls = this.opts.maxIdlePolls ?? 4;
+
+    while (processed < maxJobs) {
+      let job: QueueJobRow | null = null;
+      try {
+        job = await this.dequeue();
+      } catch (e: any) {
+        console.error(`[QueueWorker] dequeue error: ${e.message}; retrying in ${idlePollMs}ms`);
+        await sleep(idlePollMs);
+        continue;
+      }
+
+      if (!job) {
+        idlePolls += 1;
+        if (idlePolls >= maxIdlePolls) {
+          console.log('[QueueWorker] queue empty, exiting.');
+          break;
+        }
+        console.log(`[QueueWorker] queue empty (${idlePolls}/${maxIdlePolls}); note: jobs in backoff may not be claimable yet, waiting...`);
+        await sleep(idlePollMs);
+        continue;
+      }
+
+      idlePolls = 0;
+      console.log(`[QueueWorker] claimed job ${job.id} (type=${job.type}, attempt=${job.attempts}, url=${job.target_url?.slice(0, 60)})`);
+      await this.processJob(job);
+      processed += 1;
+    }
+    return processed;
+  }
+}
diff --git a/apps/importer/src/shopee/shopee-to-supabase.ts b/apps/importer/src/shopee/shopee-to-supabase.ts
index 8e8e5d08..2c72dc2b 100644
--- a/apps/importer/src/shopee/shopee-to-supabase.ts
+++ b/apps/importer/src/shopee/shopee-to-supabase.ts
@@ -56,9 +56,16 @@ export class ShopeeIngestionPipeline {
   }
 
   /**
-   * Ingest a full Shopee product into rate schema with Master Product deduplication
+   * Ingest a full Shopee product into rate schema with Master Product deduplication.
+   * `report` (optional) is awaited after each section completes so a queue
+   * worker can checkpoint progress. Ingest sections are idempotent upserts,
+   * so a retried job simply re-runs them; the checkpoint trail is for
+   * observability (which section a crash died in), not selective resume.
    */
-  public async ingestProduct(product: ShopeeProduct): Promise<{
+  public async ingestProduct(
+    product: ShopeeProduct,
+    report?: (section: 'merchant' | 'master' | 'listing' | 'price' | 'reviews') => Promise<void>
+  ): Promise<{
     productId: string;
     merchantId: string;
     masterId: string;
@@ -98,6 +105,7 @@ export class ShopeeIngestionPipeline {
 
     const merchantId = merchant.id;
     console.log(`[ShopeeIngest] Merchant resolved: ${product.shop.name} -> ID: ${merchantId}`);
+    await report?.('merchant');
 
     // 2. Normalize Specs & Match / Seed Master Product
     const candidate = this.normalizer.normalize(product);
@@ -108,6 +116,7 @@ export class ShopeeIngestionPipeline {
       this.smartphoneCategoryId || undefined
     );
     console.log(`[ShopeeIngest] Master Product ${isNew ? 'CREATED' : 'LINKED'}: ID = ${masterId}`);
+    await report?.('master');
 
     // 3. Upsert product listing into rate.products (linked via cluster_id)
     const { data: savedProduct, error: productErr } = await this.supabase
@@ -148,6 +157,7 @@ export class ShopeeIngestionPipeline {
 
     const productId = savedProduct.id;
     console.log(`[ShopeeIngest] Listing upserted -> ID: ${productId}`);
+    await report?.('listing');
 
     // 4. Price history entry with a stable observation id (plan v7 §4.1,
     // Codex #8): re-crawls of the same price on the same day upsert instead of
@@ -181,11 +191,19 @@ export class ShopeeIngestionPipeline {
         source: 'shopee_crawler',
       });
       if (priceErr) {
-        throw new Error(`Failed to record price history (observation ${observationId}): ${priceErr.message}`);
+        const pCode = (priceErr as { code?: string }).code || '';
+        if (pCode === '23505' || /duplicate key/i.test(priceErr.message)) {
+          // Concurrent worker recorded the same observation first - benign.
+          console.warn('[ShopeeIngest] Concurrent price observation insert lost the race; treated as dedupe.');
+        } else {
+          throw new Error(`Failed to record price history (observation ${observationId}): ${priceErr.message}`);
+        }
       }
     }
+    await report?.('price');
 
-    // 5. Batch insert reviews if available
+    // 5. Batch insert reviews if available (the section checkpoint fires in
+    // both branches - a zero-review ingest is still a completed section).
     if (product.reviews && product.reviews.length > 0) {
       const reviewRows = product.reviews.map((r) => ({
         product_id: productId,
@@ -236,14 +254,45 @@ export class ShopeeIngestionPipeline {
       if (newRows.length === 0) {
         console.log(`[ShopeeIngest] No new reviews to ingest (${seen.size} duplicates skipped).`);
       } else {
-        const { error: reviewErr } = await this.supabase.from('reviews').insert(newRows);
-        if (reviewErr) {
-          // Fail the whole ingestion (plan v7 §4.1, Codex blocker): a review
-          // write failure must NOT let the caller ack the job as done.
-          throw new Error(`Failed to insert reviews: ${reviewErr.message}`);
+        // Batch inserts are atomic: one raced row rolls back the whole batch
+        // (Codex P1 - swallowing 23505 would drop genuinely-new reviews that
+        // share the batch). On 23505, re-read which ids now exist, drop them,
+        // and retry the remainder once.
+        let pending = newRows;
+        for (let pass = 0; pass < 2; pass++) {
+          const { error: reviewErr } = await this.supabase.from('reviews').insert(pending);
+          if (!reviewErr) {
+            console.log(`[ShopeeIngest] Successfully ingested ${pending.length} reviews (batch pass ${pass + 1}).`);
+            pending = [];
+            break;
+          }
+          const code = (reviewErr as { code?: string }).code || '';
+          const isRace = code === '23505' || /duplicate key/i.test(reviewErr.message);
+          if (!isRace || pass > 0) {
+            throw new Error(`Failed to insert reviews: ${reviewErr.message}`);
+          }
+          // Unique-index race: drop the ids that already exist now and retry
+          // the rest. Anything that vanishes here was inserted by the winner.
+          console.warn('[ShopeeIngest] Review batch hit a unique-index race; retrying the remainder.');
+          const { data: existingNow, error: refetchErr } = await this.supabase
+            .from('reviews')
+            .select('platform_review_id')
+            .eq('product_id', productId)
+            .not('platform_review_id', 'is', null);
+          if (refetchErr) {
+            throw new Error(`Failed to refetch reviews after race: ${refetchErr.message}`);
+          }
+          const nowSeen = new Set<string>((existingNow || []).map((r: { platform_review_id: string }) => r.platform_review_id));
+          pending = pending.filter((r) => !nowSeen.has(r.platform_review_id as string));
+          if (pending.length === 0) {
+            console.log('[ShopeeIngest] All raced reviews already stored by the winner.');
+          }
+        }
+        if (pending.length > 0) {
+          throw new Error(`Failed to insert reviews: unexplained retry state (${pending.length} rows left)`);
         }
-        console.log(`[ShopeeIngest] Successfully ingested ${newRows.length} reviews (${reviewRows.length - newRows.length} duplicates skipped).`);
       }
+      await report?.('reviews');
     }
 
     console.log(`[ShopeeIngest] Product ingestion complete! (Master: ${masterId}, Product: ${productId})`);