From 10a2a790050ece3d73e5049c7a27698776a46056 Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 15:58:09 +0700 Subject: [PATCH 1/6] feat(importer): wire crawler to the rate.crawl_jobs queue - QueueDrivenShopeeCrawler: claims jobs via dequeue_crawl_job, heartbeats long scrapes, checkpoints per-section ingest progress, acks/fails with the per-claim token fence - all through the SECURITY DEFINER RPCs (direct DML on crawl_jobs stays revoked server-side) - ShopeeIngestionPipeline.ingestProduct accepts an optional section report callback (merchant/master/listing/price/reviews) so the worker checkpoints after each idempotent section (plan v7 section 4.1) - Scripts/enqueue-shopee.ts: CLI to enqueue product URLs (parses shop/item ids, opens a crawl run) - Scripts/run-queue-crawl.ts: full driver - cloned-Chrome CDP launch (home IP default, USE_4G=1 for mproxy), human-like store warm-up before each PDP, worker loop until queue drains, finish_crawl_run + browser teardown in finally Verified against the live prod schema (migrations propagated): enqueue -> dequeue(attempt 1, token) -> checkpoint ok -> ack ok; re-enqueue after done correctly creates a fresh pending job (terminal states never block recrawl). tsc clean. --- apps/importer/Scripts/enqueue-shopee.ts | 69 +++++++ apps/importer/Scripts/run-queue-crawl.ts | 152 ++++++++++++++ apps/importer/src/shopee/queue-worker.ts | 192 ++++++++++++++++++ .../importer/src/shopee/shopee-to-supabase.ts | 15 +- 4 files changed, 426 insertions(+), 2 deletions(-) create mode 100644 apps/importer/Scripts/enqueue-shopee.ts create mode 100644 apps/importer/Scripts/run-queue-crawl.ts create mode 100644 apps/importer/src/shopee/queue-worker.ts diff --git a/apps/importer/Scripts/enqueue-shopee.ts b/apps/importer/Scripts/enqueue-shopee.ts new file mode 100644 index 00000000..e5c42270 --- /dev/null +++ b/apps/importer/Scripts/enqueue-shopee.ts @@ -0,0 +1,69 @@ +#!/usr/bin/env node +/** + * Enqueue Shopee product URLs as rate.crawl_jobs rows (via enqueue_crawl_job RPC). + * + * Usage: + * tsx Scripts/enqueue-shopee.ts [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 [url2...] | --file targets.txt'); + process.exit(1); + } + + const runId = await db.rpc('start_crawl_run', { p_source: 'enqueue-cli', p_params: { count: urls.length } }); + console.log(`[Enqueue] run=${runId.data}`); + + for (const url of urls) { + const ids = parseItem(url); + if (!ids) { + console.error(`[Enqueue] SKIP unparseable: ${url}`); + 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.data, + }); + if (error) { + console.error(`[Enqueue] FAIL ${ids.itemId}: ${error.message}`); + } else { + console.log(`[Enqueue] job=${data[0]?.job_id ?? data?.job_id} state=${data[0]?.state ?? data?.state} item=${ids.itemId}`); + } + } + process.exit(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..69a403f8 --- /dev/null +++ b/apps/importer/Scripts/run-queue-crawl.ts @@ -0,0 +1,152 @@ +#!/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'; + +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=*' } | 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 maxJobs = maxJobsArg > -1 ? parseInt(process.argv[maxJobsArg + 1], 10) : undefined; + 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((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}`); + + 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: 15000, + maxIdlePolls: 2, + 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); + }, + }); + + let processed = 0; + try { + processed = await worker.run(); + console.log(`[Run] worker finished, processed=${processed}`); + } finally { + await db.rpc('finish_crawl_run', { p_run_id: runId, p_stats: { processed } }).catch(() => {}); + killCdpBrowser(); // only our CDP instance + 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..8c963178 --- /dev/null +++ b/apps/importer/src/shopee/queue-worker.ts @@ -0,0 +1,192 @@ +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; + 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; +} + +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; + 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. */ + 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 { + 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 { + 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; + } + + public async checkpoint(job: QueueJobRow, progress: Record): Promise { + 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): Promise { + 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 { + 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. */ + public async processJob(job: QueueJobRow): Promise { + const sections: Record = { started_at: new Date().toISOString() }; + await this.checkpoint(job, { started: true }); + + // Heartbeat while scraping (long): independent interval, cleared on exit. + const hbMs = (this.opts.heartbeatSeconds ?? 120) * 1000; + const hb = setInterval(() => { + this.heartbeat(job).then( + (ok) => { + if (!ok) console.warn(`[QueueWorker] heartbeat rejected for job ${job.id} (lease lost?)`); + }, + (e) => console.warn(`[QueueWorker] heartbeat error: ${e.message}`) + ); + }, hbMs); + + try { + const product = await this.opts.scrape(job); + sections.title = product.title; + + const result = await this.opts.pipeline.ingestProduct(product, async (section) => { + sections[section] = new Date().toISOString(); + await this.checkpoint(job, { [`${section}_at`]: sections[section] }); + }); + + sections.productId = result.productId; + sections.masterId = result.masterId; + await this.ack(job, sections); + console.log(`[QueueWorker] job ${job.id} ACKED (product ${result.productId})`); + } catch (err: any) { + const message = err?.message || String(err); + console.error(`[QueueWorker] job ${job.id} FAILED: ${message}`); + await this.fail(job, message); + } finally { + clearInterval(hb); + } + } + + /** + * Main loop: dequeue -> process, until maxJobs reached or the queue stays + * empty for maxIdlePolls consecutive polls. Returns processed count. + */ + public async run(): Promise { + let processed = 0; + let idlePolls = 0; + const maxJobs = this.opts.maxJobs ?? Number.POSITIVE_INFINITY; + const idlePollMs = this.opts.idlePollMs ?? 15000; + const maxIdlePolls = this.opts.maxIdlePolls ?? 2; + + 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}), 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..de77d135 100644 --- a/apps/importer/src/shopee/shopee-to-supabase.ts +++ b/apps/importer/src/shopee/shopee-to-supabase.ts @@ -56,9 +56,15 @@ 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 (plan v7 §4.1: a retry resumes from the + * first incomplete section instead of redoing everything). */ - public async ingestProduct(product: ShopeeProduct): Promise<{ + public async ingestProduct( + product: ShopeeProduct, + report?: (section: 'merchant' | 'master' | 'listing' | 'price' | 'reviews') => Promise + ): Promise<{ productId: string; merchantId: string; masterId: string; @@ -98,6 +104,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 +115,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 +156,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 @@ -184,6 +193,7 @@ export class ShopeeIngestionPipeline { throw new Error(`Failed to record price history (observation ${observationId}): ${priceErr.message}`); } } + await report?.('price'); // 5. Batch insert reviews if available if (product.reviews && product.reviews.length > 0) { @@ -244,6 +254,7 @@ export class ShopeeIngestionPipeline { } 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})`); From e440e94a32669afcadd838ba7da815abe4684bb5 Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 16:09:48 +0700 Subject: [PATCH 2/6] fix(importer): enforce lease loss, honest lifecycle, CLI exit codes Codex round-1 review of the queue wiring: - BLOCKER: a rejected heartbeat or refused ack now aborts the job with LEASE_LOST - the worker stops touching the listing and never logs a false ACKED (the true owner owns the outcome); the in-flight heartbeat promise is awaited on teardown - BLOCKER: post-run cleanup awaits finish_crawl_run in try/catch (the PostgREST builder has no .catch); stopLocalForwarder was missing on the branch base because dev had unpushed Shopee commits - dev is now synced (c941957) and the rebased branch has it - SHOULDs: setup after start_crawl_run is inside try/finally; kill is scoped to crawl profile dirs (single-driver documented); idle semantics document backoff-aware waiting (30s x 4); enqueue CLI exits non-zero on run failure, enqueue failure or skipped URLs; --max-jobs NaN guard; reviews-section checkpoint fires on empty review lists; checkpoint trail documented as observability (retries re-run the idempotent pipeline) --- apps/importer/Scripts/enqueue-shopee.ts | 17 +++- apps/importer/Scripts/run-queue-crawl.ts | 92 +++++++++++-------- apps/importer/src/shopee/queue-worker.ts | 44 +++++++-- .../importer/src/shopee/shopee-to-supabase.ts | 10 +- 4 files changed, 109 insertions(+), 54 deletions(-) diff --git a/apps/importer/Scripts/enqueue-shopee.ts b/apps/importer/Scripts/enqueue-shopee.ts index e5c42270..fd938005 100644 --- a/apps/importer/Scripts/enqueue-shopee.ts +++ b/apps/importer/Scripts/enqueue-shopee.ts @@ -35,13 +35,19 @@ async function main() { process.exit(1); } - const runId = await db.rpc('start_crawl_run', { p_source: 'enqueue-cli', p_params: { count: urls.length } }); - console.log(`[Enqueue] run=${runId.data}`); + 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', { @@ -52,15 +58,18 @@ async function main() { p_platform_item_id: ids.itemId, p_platform_shop_id: ids.shopId, p_priority: 0, - p_run_id: runId.data, + 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}`); } } - process.exit(0); + // 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) => { diff --git a/apps/importer/Scripts/run-queue-crawl.ts b/apps/importer/Scripts/run-queue-crawl.ts index 69a403f8..fc9dc556 100644 --- a/apps/importer/Scripts/run-queue-crawl.ts +++ b/apps/importer/Scripts/run-queue-crawl.ts @@ -27,10 +27,13 @@ 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=*' } | ForEach-Object { taskkill /PID $_.ProcessId /T /F 2>&1 | Out-Null }"`, + `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 {} @@ -38,7 +41,13 @@ function killCdpBrowser(): void { async function main() { const maxJobsArg = process.argv.indexOf('--max-jobs'); - const maxJobs = maxJobsArg > -1 ? parseInt(process.argv[maxJobsArg + 1], 10) : undefined; + 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!, { @@ -99,47 +108,54 @@ async function main() { } console.log(`[Run] crawl run=${runId}`); - 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: 15000, - maxIdlePolls: 2, - 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); - }, - }); - + // 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 { - await db.rpc('finish_crawl_run', { p_run_id: runId, p_stats: { processed } }).catch(() => {}); - killCdpBrowser(); // only our CDP instance + 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.'); } diff --git a/apps/importer/src/shopee/queue-worker.ts b/apps/importer/src/shopee/queue-worker.ts index 8c963178..a270b4b9 100644 --- a/apps/importer/src/shopee/queue-worker.ts +++ b/apps/importer/src/shopee/queue-worker.ts @@ -34,7 +34,8 @@ export interface QueueWorkerOptions { maxJobs?: number; /** Sleep between polls when the queue is empty (ms). */ idlePollMs?: number; - /** Give up idle-waiting after this many consecutive empty polls. */ + /** 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; @@ -79,6 +80,10 @@ export class QueueDrivenShopeeCrawler { 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): Promise { const { error } = await this.db().rpc('checkpoint_crawl_job', { p_id: job.id, @@ -112,17 +117,26 @@ export class QueueDrivenShopeeCrawler { return data === true; } - /** Process one claimed job end-to-end: scrape -> ingest (checkpointed) -> ack. */ + /** 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 { const sections: Record = { started_at: new Date().toISOString() }; await this.checkpoint(job, { started: true }); - // Heartbeat while scraping (long): independent interval, cleared on exit. + // 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 | null = null; const hb = setInterval(() => { - this.heartbeat(job).then( + hbInFlight = this.heartbeat(job).then( (ok) => { - if (!ok) console.warn(`[QueueWorker] heartbeat rejected for job ${job.id} (lease lost?)`); + 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}`) ); @@ -130,23 +144,35 @@ export class QueueDrivenShopeeCrawler { 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] }); }); + if (leaseLost) throw new Error(QueueDrivenShopeeCrawler.LeaseLost); sections.productId = result.productId; sections.masterId = result.masterId; - await this.ack(job, sections); + 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); + if (message === QueueDrivenShopeeCrawler.LeaseLost) { + // Not ours anymore: the current owner owns the outcome. Do NOT ack/fail. + 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); + // hbInFlight is assigned from an interval callback, so TS sees it as + // always-null here; the cast documents the runtime reality. + await (hbInFlight as unknown as Promise | undefined)?.catch(() => {}); } } @@ -158,8 +184,8 @@ export class QueueDrivenShopeeCrawler { let processed = 0; let idlePolls = 0; const maxJobs = this.opts.maxJobs ?? Number.POSITIVE_INFINITY; - const idlePollMs = this.opts.idlePollMs ?? 15000; - const maxIdlePolls = this.opts.maxIdlePolls ?? 2; + const idlePollMs = this.opts.idlePollMs ?? 30000; + const maxIdlePolls = this.opts.maxIdlePolls ?? 4; while (processed < maxJobs) { let job: QueueJobRow | null = null; @@ -177,7 +203,7 @@ export class QueueDrivenShopeeCrawler { console.log('[QueueWorker] queue empty, exiting.'); break; } - console.log(`[QueueWorker] queue empty (${idlePolls}/${maxIdlePolls}), waiting...`); + console.log(`[QueueWorker] queue empty (${idlePolls}/${maxIdlePolls}); note: jobs in backoff may not be claimable yet, waiting...`); await sleep(idlePollMs); continue; } diff --git a/apps/importer/src/shopee/shopee-to-supabase.ts b/apps/importer/src/shopee/shopee-to-supabase.ts index de77d135..a1461300 100644 --- a/apps/importer/src/shopee/shopee-to-supabase.ts +++ b/apps/importer/src/shopee/shopee-to-supabase.ts @@ -58,8 +58,9 @@ export class ShopeeIngestionPipeline { /** * 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 (plan v7 §4.1: a retry resumes from the - * first incomplete section instead of redoing everything). + * 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, @@ -195,7 +196,8 @@ export class ShopeeIngestionPipeline { } 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, @@ -255,6 +257,8 @@ export class ShopeeIngestionPipeline { console.log(`[ShopeeIngest] Successfully ingested ${newRows.length} reviews (${reviewRows.length - newRows.length} duplicates skipped).`); } await report?.('reviews'); + } else { + await report?.('reviews'); } console.log(`[ShopeeIngest] Product ingestion complete! (Master: ${masterId}, Product: ${productId})`); From c67e96a32665cd3aefb70156240e55eb15f3ae1a Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 16:17:17 +0700 Subject: [PATCH 3/6] fix(importer): close round-2 lease holes - post-await checks, flag-aware catch, chained heartbeats - re-check leaseLost after every awaited checkpoint inside the ingest report callback: the pipeline must not start the next section's writes once the lease died mid-checkpoint - catch treats leaseLost=true as abandonment even when the work threw a normal error afterwards - fail_crawl_job must never run for a job this worker no longer owns - heartbeats chain onto the in-flight promise so overlapping interval ticks cannot overwrite the tracked heartbeat (teardown awaits them all) --- apps/importer/src/shopee/queue-worker.ts | 33 ++++++++++++++++-------- 1 file changed, 22 insertions(+), 11 deletions(-) diff --git a/apps/importer/src/shopee/queue-worker.ts b/apps/importer/src/shopee/queue-worker.ts index a270b4b9..2ba74273 100644 --- a/apps/importer/src/shopee/queue-worker.ts +++ b/apps/importer/src/shopee/queue-worker.ts @@ -131,15 +131,20 @@ export class QueueDrivenShopeeCrawler { let leaseLost = false; let hbInFlight: Promise | null = null; const hb = setInterval(() => { - 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}`) - ); + // Chain onto any in-flight heartbeat: overlapping calls must not + // overwrite the tracked promise (Codex round-2) or teardown would + // await only the newest while an older one is still in the air. + hbInFlight = (hbInFlight as unknown as Promise | undefined ?? Promise.resolve()) + .then(() => 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}`) + ); }, hbMs); try { @@ -151,6 +156,10 @@ export class QueueDrivenShopeeCrawler { 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); @@ -161,8 +170,10 @@ export class QueueDrivenShopeeCrawler { console.log(`[QueueWorker] job ${job.id} ACKED (product ${result.productId})`); } catch (err: any) { const message = err?.message || String(err); - if (message === QueueDrivenShopeeCrawler.LeaseLost) { - // Not ours anymore: the current owner owns the outcome. Do NOT ack/fail. + // 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; } From 3ce6f2785e432de0219806198f999475b58c26e8 Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 16:20:03 +0700 Subject: [PATCH 4/6] fix(importer): skip overlapping heartbeat ticks instead of chaining Codex round-3: chaining queued an unbounded heartbeat backlog when RPCs were slow and teardown awaited all of it. A tick now no-ops while the previous heartbeat is in flight (lease >> interval, so a skipped tick only shortens the effective lease marginally); teardown awaits just the one running RPC; hbInFlight resets to null on completion so the cast workaround is gone. --- apps/importer/src/shopee/queue-worker.ts | 37 +++++++++++++----------- 1 file changed, 20 insertions(+), 17 deletions(-) diff --git a/apps/importer/src/shopee/queue-worker.ts b/apps/importer/src/shopee/queue-worker.ts index 2ba74273..d48d294c 100644 --- a/apps/importer/src/shopee/queue-worker.ts +++ b/apps/importer/src/shopee/queue-worker.ts @@ -131,20 +131,25 @@ export class QueueDrivenShopeeCrawler { let leaseLost = false; let hbInFlight: Promise | null = null; const hb = setInterval(() => { - // Chain onto any in-flight heartbeat: overlapping calls must not - // overwrite the tracked promise (Codex round-2) or teardown would - // await only the newest while an older one is still in the air. - hbInFlight = (hbInFlight as unknown as Promise | undefined ?? Promise.resolve()) - .then(() => 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}`) - ); + // 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 { @@ -181,9 +186,7 @@ export class QueueDrivenShopeeCrawler { await this.fail(job, message); } finally { clearInterval(hb); - // hbInFlight is assigned from an interval callback, so TS sees it as - // always-null here; the cast documents the runtime reality. - await (hbInFlight as unknown as Promise | undefined)?.catch(() => {}); + await Promise.resolve(hbInFlight).catch(() => {}); } } From cd57a58c15c1ea26150d0ef584943905e4303cfb Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 16:26:46 +0700 Subject: [PATCH 5/6] fix(importer): treat 23505 insert races as benign dedupe Qwen review: the check-then-insert pattern is not atomic across concurrent workers; the DB unique index arbitrates, so a 23505 (or 'duplicate key') on the reviews/price inserts is a lost dedupe race, not a failure - warn and continue. All other errors still fail the ingestion. --- .../importer/src/shopee/shopee-to-supabase.ts | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/apps/importer/src/shopee/shopee-to-supabase.ts b/apps/importer/src/shopee/shopee-to-supabase.ts index a1461300..3bb80b5d 100644 --- a/apps/importer/src/shopee/shopee-to-supabase.ts +++ b/apps/importer/src/shopee/shopee-to-supabase.ts @@ -191,7 +191,13 @@ 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'); @@ -250,9 +256,16 @@ export class ShopeeIngestionPipeline { } 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}`); + // A concurrent worker inserting the same review loses to the unique + // index (23505) - that race is benign dedupe, not a failure (Qwen + // review: check-then-insert is not atomic; the DB index arbitrates). + // Any other error fails the whole ingestion (plan v7 §4.1): the + // caller must not ack the job as done. + const code = (reviewErr as { code?: string }).code || ''; + if (code !== '23505' && !/duplicate key/i.test(reviewErr.message)) { + throw new Error(`Failed to insert reviews: ${reviewErr.message}`); + } + console.warn('[ShopeeIngest] Concurrent review insert lost the unique-index race; treated as dedupe.'); } console.log(`[ShopeeIngest] Successfully ingested ${newRows.length} reviews (${reviewRows.length - newRows.length} duplicates skipped).`); } From a10d72205f96600421b0a1df7a0f21840f446f6d Mon Sep 17 00:00:00 2001 From: JOY Date: Wed, 7 Oct 2026 16:29:34 +0700 Subject: [PATCH 6/6] fix(importer): on 23505 retry the un-raced review remainder instead of swallowing the batch Codex P1: batch inserts are atomic - swallowing a 23505 dropped the genuinely-new rows sharing the batch while the job still acked. On a race the worker now refetches which ids exist, retries the remainder once, and throws if rows remain unexplained. --- .../importer/src/shopee/shopee-to-supabase.ts | 45 ++++++++++++++----- 1 file changed, 33 insertions(+), 12 deletions(-) diff --git a/apps/importer/src/shopee/shopee-to-supabase.ts b/apps/importer/src/shopee/shopee-to-supabase.ts index 3bb80b5d..2c72dc2b 100644 --- a/apps/importer/src/shopee/shopee-to-supabase.ts +++ b/apps/importer/src/shopee/shopee-to-supabase.ts @@ -254,24 +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) { - // A concurrent worker inserting the same review loses to the unique - // index (23505) - that race is benign dedupe, not a failure (Qwen - // review: check-then-insert is not atomic; the DB index arbitrates). - // Any other error fails the whole ingestion (plan v7 §4.1): the - // caller must not ack the job as done. + // 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 || ''; - if (code !== '23505' && !/duplicate key/i.test(reviewErr.message)) { + const isRace = code === '23505' || /duplicate key/i.test(reviewErr.message); + if (!isRace || pass > 0) { throw new Error(`Failed to insert reviews: ${reviewErr.message}`); } - console.warn('[ShopeeIngest] Concurrent review insert lost the unique-index race; treated as dedupe.'); + // 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((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'); - } else { - await report?.('reviews'); } console.log(`[ShopeeIngest] Product ingestion complete! (Master: ${masterId}, Product: ${productId})`);