| 26 |
26 |
// Atomically claim a batch (status pending → processing) so several resolver workers can run |
| 27 |
27 |
// concurrently without processing the same rows; rows are released back to pending on crash. |
| 28 |
28 |
const connectorFilter = opts.connectorId ? sql` and connector_id = ${opts.connectorId}` : sql``; |
| 29 |
|
− const rows = (await db().execute(sql` |
| 30 |
|
− with claimed as ( |
| 31 |
|
− select id from normalized_records |
| 32 |
|
− where status = 'pending'${connectorFilter} |
| 33 |
|
− order by case kind when 'catalog_item' then 0 when 'population_report' then 1 when 'sale' then 2 when 'price_observation' then 3 when 'listing' then 4 else 5 end, created_at |
| 34 |
|
− limit ${limit} |
| 35 |
|
− for update skip locked |
| 36 |
|
− ) |
| 37 |
|
− update normalized_records n set status = 'processing' |
| 38 |
|
− from claimed where n.id = claimed.id |
| 39 |
|
− returning n.id, n.raw_record_id as "rawRecordId", n.connector_id as "connectorId", n.source_id as "sourceId", n.kind, n.payload, n.seq, n.asset_id as "assetId", n.variant_id as "variantId", n.match_method as "matchMethod", n.match_confidence as "matchConfidence", n.status, n.reject_reason as "rejectReason", n.target_id as "targetId", n.created_at as "createdAt", n.processed_at as "processedAt" |
| 40 |
|
− `)) as unknown as Array<typeof normalizedRecords.$inferSelect>; |
|
29 |
+ // Kind priority (catalog first so identifiers exist before sales/listings resolve against them). |
|
30 |
+ // Claim per kind so the query uses the (status, kind, created_at) index instead of sorting every pending row. |
|
31 |
+ const KIND_ORDER = ['catalog_item', 'population_report', 'sale', 'price_observation', 'listing', 'auction_lot', 'news_item']; |
|
32 |
+ let rows: Array<typeof normalizedRecords.$inferSelect> = []; |
|
33 |
+ for (const kind of KIND_ORDER) { |
|
34 |
+ rows = (await db().execute(sql` |
|
35 |
+ with claimed as ( |
|
36 |
+ select id from normalized_records |
|
37 |
+ where status = 'pending' and kind = ${kind}${connectorFilter} |
|
38 |
+ order by created_at |
|
39 |
+ limit ${limit} |
|
40 |
+ for update skip locked |
|
41 |
+ ) |
|
42 |
+ update normalized_records n set status = 'processing' |
|
43 |
+ from claimed where n.id = claimed.id |
|
44 |
+ returning n.id, n.raw_record_id as "rawRecordId", n.connector_id as "connectorId", n.source_id as "sourceId", n.kind, n.payload, n.seq, n.asset_id as "assetId", n.variant_id as "variantId", n.match_method as "matchMethod", n.match_confidence as "matchConfidence", n.status, n.reject_reason as "rejectReason", n.target_id as "targetId", n.created_at as "createdAt", n.processed_at as "processedAt" |
|
45 |
+ `)) as unknown as Array<typeof normalizedRecords.$inferSelect>; |
|
46 |
+ if (rows.length) break; |
|
47 |
+ } |
| 41 |
48 |
const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] }; |
| 42 |
49 |
if (rows.length === 0) return res; |
| 43 |
50 |
const touched = new Set<string>(); |
| 44 |
51 |
|