From cb8ae85737718cea974f750ee510d07f7cfbe85b Mon Sep 17 00:00:00 2001 From: lorentz Date: Sat, 11 Jul 2026 07:20:54 -0400 Subject: [PATCH] fix(12): WR-01 isolate per-row failures in pax8 sync loops so one bad record can't abort the batch --- lib/services/pax8-sync-service.ts | 436 ++++++++++++++++++------------ 1 file changed, 267 insertions(+), 169 deletions(-) diff --git a/lib/services/pax8-sync-service.ts b/lib/services/pax8-sync-service.ts index d322744..4112e78 100644 --- a/lib/services/pax8-sync-service.ts +++ b/lib/services/pax8-sync-service.ts @@ -158,42 +158,56 @@ export class Pax8SyncService { console.log(`[Pax8Sync] Fetched ${companies.length} companies`); let upserted = 0; + let failed = 0; + let firstError: string | null = null; const seen: string[] = []; for (const c of companies) { if (!c.id) continue; + // Recorded as seen regardless of upsert outcome below (WR-01): a + // per-row failure here must not make the tombstone step below + // mistake "failed to write this run" for "gone from PAX8". seen.push(c.id); - await postgresClient.query( - `INSERT INTO pax8_companies - (id, name, external_id, website, status, city, state_or_province, postal_code, country, - raw_payload, synced_at, is_deleted, deleted_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, NOW(), false, NULL) - ON CONFLICT (id) DO UPDATE SET - name = EXCLUDED.name, - external_id = EXCLUDED.external_id, - website = EXCLUDED.website, - status = EXCLUDED.status, - city = EXCLUDED.city, - state_or_province = EXCLUDED.state_or_province, - postal_code = EXCLUDED.postal_code, - country = EXCLUDED.country, - raw_payload = EXCLUDED.raw_payload, - synced_at = NOW(), - is_deleted = false, - deleted_at = NULL`, - [ - c.id, - c.name, - c.externalId ?? null, - c.website ?? null, - c.status ?? null, - c.city ?? null, - c.stateOrProvince ?? null, - c.postalCode ?? null, - c.country ?? null, - JSON.stringify(c), - ] - ); - upserted++; + try { + await postgresClient.query( + `INSERT INTO pax8_companies + (id, name, external_id, website, status, city, state_or_province, postal_code, country, + raw_payload, synced_at, is_deleted, deleted_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, NOW(), false, NULL) + ON CONFLICT (id) DO UPDATE SET + name = EXCLUDED.name, + external_id = EXCLUDED.external_id, + website = EXCLUDED.website, + status = EXCLUDED.status, + city = EXCLUDED.city, + state_or_province = EXCLUDED.state_or_province, + postal_code = EXCLUDED.postal_code, + country = EXCLUDED.country, + raw_payload = EXCLUDED.raw_payload, + synced_at = NOW(), + is_deleted = false, + deleted_at = NULL`, + [ + c.id, + c.name, + c.externalId ?? null, + c.website ?? null, + c.status ?? null, + c.city ?? null, + c.stateOrProvince ?? null, + c.postalCode ?? null, + c.country ?? null, + JSON.stringify(c), + ] + ); + upserted++; + } catch (rowErr) { + // WR-01: isolate a single bad row so it doesn't abort the rest of + // the batch or skip tombstoning below. + failed++; + const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr); + firstError ??= rowMsg; + console.error(`[Pax8Sync] Company ${c.id} upsert failed, continuing:`, rowMsg); + } } const tombstoned = seen.length === 0 @@ -204,8 +218,16 @@ export class Pax8SyncService { [seen] )).rowCount ?? 0; if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} compan${tombstoned === 1 ? 'y' : 'ies'}`); + if (failed > 0) console.error(`[Pax8Sync] Company sync completed with ${failed} row failure(s)`); - return { entity: 'companies', success: true, upserted, tombstoned, durationMs: Date.now() - start }; + return { + entity: 'companies', + success: failed === 0, + upserted, + tombstoned, + durationMs: Date.now() - start, + error: failed > 0 ? `${failed} of ${companies.length} companies failed to upsert; first error: ${firstError}` : undefined, + }; } catch (err) { const msg = err instanceof Error ? err.message : String(err); console.error(`[Pax8Sync] Company sync failed:`, msg); @@ -221,46 +243,59 @@ export class Pax8SyncService { console.log(`[Pax8Sync] Fetched ${subscriptions.length} subscriptions`); let upserted = 0; + let failed = 0; + let firstError: string | null = null; const seen: string[] = []; for (const s of subscriptions) { if (!s.id) continue; + // Recorded as seen regardless of upsert outcome below (WR-01) — see + // syncCompanies for rationale. seen.push(s.id); if (s.productId) referencedProductIds.add(s.productId); - await postgresClient.query( - `INSERT INTO pax8_subscriptions - (id, pax8_company_id, product_id, quantity, billing_term, status, start_date, - price, partner_cost, currency, raw_payload, synced_at, is_deleted, deleted_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11, NOW(), false, NULL) - ON CONFLICT (id) DO UPDATE SET - pax8_company_id = EXCLUDED.pax8_company_id, - product_id = EXCLUDED.product_id, - quantity = EXCLUDED.quantity, - billing_term = EXCLUDED.billing_term, - status = EXCLUDED.status, - start_date = EXCLUDED.start_date, - price = EXCLUDED.price, - partner_cost = EXCLUDED.partner_cost, - currency = EXCLUDED.currency, - raw_payload = EXCLUDED.raw_payload, - synced_at = NOW(), - is_deleted = false, - deleted_at = NULL`, - [ - s.id, - s.companyId ?? null, - s.productId ?? null, - s.quantity ?? null, - s.billingTerm ?? null, - s.status ?? null, - s.startDate ?? null, - s.price ?? null, - s.partnerCost ?? null, - s.currencyCode ?? 'USD', - JSON.stringify(s), - ] - ); - upserted++; + try { + await postgresClient.query( + `INSERT INTO pax8_subscriptions + (id, pax8_company_id, product_id, quantity, billing_term, status, start_date, + price, partner_cost, currency, raw_payload, synced_at, is_deleted, deleted_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11, NOW(), false, NULL) + ON CONFLICT (id) DO UPDATE SET + pax8_company_id = EXCLUDED.pax8_company_id, + product_id = EXCLUDED.product_id, + quantity = EXCLUDED.quantity, + billing_term = EXCLUDED.billing_term, + status = EXCLUDED.status, + start_date = EXCLUDED.start_date, + price = EXCLUDED.price, + partner_cost = EXCLUDED.partner_cost, + currency = EXCLUDED.currency, + raw_payload = EXCLUDED.raw_payload, + synced_at = NOW(), + is_deleted = false, + deleted_at = NULL`, + [ + s.id, + s.companyId ?? null, + s.productId ?? null, + s.quantity ?? null, + s.billingTerm ?? null, + s.status ?? null, + s.startDate ?? null, + s.price ?? null, + s.partnerCost ?? null, + s.currencyCode ?? 'USD', + JSON.stringify(s), + ] + ); + upserted++; + } catch (rowErr) { + // WR-01: isolate a single bad row so it doesn't abort the rest of + // the batch or skip tombstoning below. + failed++; + const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr); + firstError ??= rowMsg; + console.error(`[Pax8Sync] Subscription ${s.id} upsert failed, continuing:`, rowMsg); + } } const tombstoned = seen.length === 0 @@ -271,9 +306,17 @@ export class Pax8SyncService { [seen] )).rowCount ?? 0; if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} subscription(s)`); + if (failed > 0) console.error(`[Pax8Sync] Subscription sync completed with ${failed} row failure(s)`); return { - result: { entity: 'subscriptions', success: true, upserted, tombstoned, durationMs: Date.now() - start }, + result: { + entity: 'subscriptions', + success: failed === 0, + upserted, + tombstoned, + durationMs: Date.now() - start, + error: failed > 0 ? `${failed} of ${subscriptions.length} subscriptions failed to upsert; first error: ${firstError}` : undefined, + }, referencedProductIds, }; } catch (err) { @@ -312,6 +355,8 @@ export class Pax8SyncService { } let upserted = 0; + let failed = 0; + let firstError: string | null = null; const seen: string[] = []; for (const productId of referencedProductIds) { const p = byId.get(productId); @@ -321,31 +366,42 @@ export class Pax8SyncService { console.warn(`[Pax8Sync] Referenced product ${productId} not found in catalog — skipping`); continue; } + // Recorded as seen regardless of upsert outcome below (WR-01) — see + // syncCompanies for rationale. seen.push(p.id); - await postgresClient.query( - `INSERT INTO pax8_products - (id, sku, vendor_sku, name, category, raw_payload, synced_at, is_deleted, deleted_at) - VALUES ($1,$2,$3,$4,$5,$6, NOW(), false, NULL) - ON CONFLICT (id) DO UPDATE SET - sku = EXCLUDED.sku, - vendor_sku = EXCLUDED.vendor_sku, - name = EXCLUDED.name, - category = EXCLUDED.category, - raw_payload = EXCLUDED.raw_payload, - synced_at = NOW(), - is_deleted = false, - deleted_at = NULL`, - [ - p.id, - p.sku ?? null, - p.vendorSku ?? null, - p.name ?? null, - (p.category as string | null) ?? p.vendorName ?? null, - JSON.stringify(p), - ] - ); - upserted++; + try { + await postgresClient.query( + `INSERT INTO pax8_products + (id, sku, vendor_sku, name, category, raw_payload, synced_at, is_deleted, deleted_at) + VALUES ($1,$2,$3,$4,$5,$6, NOW(), false, NULL) + ON CONFLICT (id) DO UPDATE SET + sku = EXCLUDED.sku, + vendor_sku = EXCLUDED.vendor_sku, + name = EXCLUDED.name, + category = EXCLUDED.category, + raw_payload = EXCLUDED.raw_payload, + synced_at = NOW(), + is_deleted = false, + deleted_at = NULL`, + [ + p.id, + p.sku ?? null, + p.vendorSku ?? null, + p.name ?? null, + (p.category as string | null) ?? p.vendorName ?? null, + JSON.stringify(p), + ] + ); + upserted++; + } catch (rowErr) { + // WR-01: isolate a single bad row so it doesn't abort the rest of + // the batch or skip tombstoning below. + failed++; + const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr); + firstError ??= rowMsg; + console.error(`[Pax8Sync] Product ${p.id} upsert failed, continuing:`, rowMsg); + } } const tombstoned = seen.length === 0 @@ -356,8 +412,16 @@ export class Pax8SyncService { [seen] )).rowCount ?? 0; if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} product(s)`); + if (failed > 0) console.error(`[Pax8Sync] Product sync completed with ${failed} row failure(s)`); - return { entity: 'products', success: true, upserted, tombstoned, durationMs: Date.now() - start }; + return { + entity: 'products', + success: failed === 0, + upserted, + tombstoned, + durationMs: Date.now() - start, + error: failed > 0 ? `${failed} products failed to upsert; first error: ${firstError}` : undefined, + }; } catch (err) { const msg = err instanceof Error ? err.message : String(err); console.error(`[Pax8Sync] Product sync failed:`, msg); @@ -382,99 +446,128 @@ export class Pax8SyncService { let ordersUpserted = 0; let itemsUpserted = 0; + let failedInvoices = 0; + let failedItems = 0; + let firstError: string | null = null; const seenOrderIds: string[] = []; const seenItemIds: string[] = []; for (const invoice of invoices) { if (!invoice.id) continue; + // Recorded as seen regardless of upsert outcome below (WR-01) — see + // syncCompanies for rationale. Isolated per-invoice so one bad + // invoice (or a transient failure fetching its items) can't abort + // the remaining invoices/items in this run or skip tombstoning. seenOrderIds.push(invoice.id); - // pax8_company_id stays NULL — see method doc / Pitfall 1. - await postgresClient.query( - `INSERT INTO pax8_orders - (id, pax8_company_id, order_date, total, status, currency, - raw_payload, synced_at, is_deleted, deleted_at) - VALUES ($1, NULL, $2, $3, $4, $5, $6, NOW(), false, NULL) - ON CONFLICT (id) DO UPDATE SET - order_date = EXCLUDED.order_date, - total = EXCLUDED.total, - status = EXCLUDED.status, - currency = EXCLUDED.currency, - raw_payload = EXCLUDED.raw_payload, - synced_at = NOW(), - is_deleted = false, - deleted_at = NULL`, - [ - invoice.id, - invoice.invoiceDate ?? null, - invoice.total ?? null, - invoice.status ?? null, - invoice.currencyCode ?? 'USD', - JSON.stringify(invoice), - ] - ); - ordersUpserted++; - - const items = await this.client.listAllInvoiceItems(invoice.id); - for (const item of items) { - if (!item.id) continue; - seenItemIds.push(item.id); - - const { unitPrice, lineTotal, partnerCost, partnerCostTotal } = resolveCostColumns(item); - + try { + // pax8_company_id stays NULL — see method doc / Pitfall 1. await postgresClient.query( - `INSERT INTO pax8_order_items - (id, order_id, product_id, quantity, unit_price, line_total, currency, - raw_payload, synced_at, is_deleted, deleted_at, - pax8_company_id, subscription_id, item_type, sku, description, - start_period, end_period, partner_cost, partner_cost_total) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8, NOW(), false, NULL, - $9,$10,$11,$12,$13,$14,$15,$16,$17) + `INSERT INTO pax8_orders + (id, pax8_company_id, order_date, total, status, currency, + raw_payload, synced_at, is_deleted, deleted_at) + VALUES ($1, NULL, $2, $3, $4, $5, $6, NOW(), false, NULL) ON CONFLICT (id) DO UPDATE SET - order_id = EXCLUDED.order_id, - product_id = EXCLUDED.product_id, - quantity = EXCLUDED.quantity, - unit_price = EXCLUDED.unit_price, - line_total = EXCLUDED.line_total, - currency = EXCLUDED.currency, - raw_payload = EXCLUDED.raw_payload, - synced_at = NOW(), - is_deleted = false, - deleted_at = NULL, - pax8_company_id = EXCLUDED.pax8_company_id, - subscription_id = EXCLUDED.subscription_id, - item_type = EXCLUDED.item_type, - sku = EXCLUDED.sku, - description = EXCLUDED.description, - start_period = EXCLUDED.start_period, - end_period = EXCLUDED.end_period, - partner_cost = EXCLUDED.partner_cost, - partner_cost_total = EXCLUDED.partner_cost_total`, + order_date = EXCLUDED.order_date, + total = EXCLUDED.total, + status = EXCLUDED.status, + currency = EXCLUDED.currency, + raw_payload = EXCLUDED.raw_payload, + synced_at = NOW(), + is_deleted = false, + deleted_at = NULL`, [ - item.id, invoice.id, - item.productId ?? null, - item.quantity ?? null, - unitPrice, - lineTotal, - item.currencyCode ?? 'USD', - JSON.stringify(item), - item.companyId ?? null, - item.subscriptionId ?? null, - item.type ?? null, - item.sku ?? null, - item.description ?? null, - item.startPeriod ?? null, - item.endPeriod ?? null, - partnerCost, - partnerCostTotal, + invoice.invoiceDate ?? null, + invoice.total ?? null, + invoice.status ?? null, + invoice.currencyCode ?? 'USD', + JSON.stringify(invoice), ] ); - itemsUpserted++; + ordersUpserted++; + + const items = await this.client.listAllInvoiceItems(invoice.id); + for (const item of items) { + if (!item.id) continue; + seenItemIds.push(item.id); + + const { unitPrice, lineTotal, partnerCost, partnerCostTotal } = resolveCostColumns(item); + + try { + await postgresClient.query( + `INSERT INTO pax8_order_items + (id, order_id, product_id, quantity, unit_price, line_total, currency, + raw_payload, synced_at, is_deleted, deleted_at, + pax8_company_id, subscription_id, item_type, sku, description, + start_period, end_period, partner_cost, partner_cost_total) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8, NOW(), false, NULL, + $9,$10,$11,$12,$13,$14,$15,$16,$17) + ON CONFLICT (id) DO UPDATE SET + order_id = EXCLUDED.order_id, + product_id = EXCLUDED.product_id, + quantity = EXCLUDED.quantity, + unit_price = EXCLUDED.unit_price, + line_total = EXCLUDED.line_total, + currency = EXCLUDED.currency, + raw_payload = EXCLUDED.raw_payload, + synced_at = NOW(), + is_deleted = false, + deleted_at = NULL, + pax8_company_id = EXCLUDED.pax8_company_id, + subscription_id = EXCLUDED.subscription_id, + item_type = EXCLUDED.item_type, + sku = EXCLUDED.sku, + description = EXCLUDED.description, + start_period = EXCLUDED.start_period, + end_period = EXCLUDED.end_period, + partner_cost = EXCLUDED.partner_cost, + partner_cost_total = EXCLUDED.partner_cost_total`, + [ + item.id, + invoice.id, + item.productId ?? null, + item.quantity ?? null, + unitPrice, + lineTotal, + item.currencyCode ?? 'USD', + JSON.stringify(item), + item.companyId ?? null, + item.subscriptionId ?? null, + item.type ?? null, + item.sku ?? null, + item.description ?? null, + item.startPeriod ?? null, + item.endPeriod ?? null, + partnerCost, + partnerCostTotal, + ] + ); + itemsUpserted++; + } catch (itemErr) { + // WR-01: isolate a single bad item so it doesn't abort the + // rest of this invoice's items or any remaining invoice. + failedItems++; + const itemMsg = itemErr instanceof Error ? itemErr.message : String(itemErr); + firstError ??= itemMsg; + console.error(`[Pax8Sync] Order item ${item.id} (invoice ${invoice.id}) upsert failed, continuing:`, itemMsg); + } + } + } catch (invoiceErr) { + // WR-01: isolate a single bad invoice (header upsert or item-fetch + // failure) so it doesn't abort any remaining invoice or skip + // tombstoning below. + failedInvoices++; + const invoiceMsg = invoiceErr instanceof Error ? invoiceErr.message : String(invoiceErr); + firstError ??= invoiceMsg; + console.error(`[Pax8Sync] Invoice ${invoice.id} sync failed, continuing:`, invoiceMsg); } } // Child-then-parent tombstoning to respect the order_items -> orders FK. + // Runs unconditionally over what was actually seen this run (WR-01) — + // a mid-loop row failure above must not suppress reconciliation of the + // records that did succeed. const itemsTombstoned = seenItemIds.length === 0 ? 0 : (await postgresClient.query( @@ -495,13 +588,18 @@ export class Pax8SyncService { if (tombstoned > 0) { console.log(`[Pax8Sync] Tombstoned ${ordersTombstoned} order(s), ${itemsTombstoned} order item(s)`); } + const failed = failedInvoices + failedItems; + if (failed > 0) { + console.error(`[Pax8Sync] Order sync completed with ${failedInvoices} invoice failure(s), ${failedItems} item failure(s)`); + } return { entity: 'orders', - success: true, + success: failed === 0, upserted: ordersUpserted + itemsUpserted, tombstoned, durationMs: Date.now() - start, + error: failed > 0 ? `${failedInvoices} invoice(s) and ${failedItems} item(s) failed to upsert; first error: ${firstError}` : undefined, }; } catch (err) { const msg = err instanceof Error ? err.message : String(err);