694 lines
28 KiB
TypeScript
694 lines
28 KiB
TypeScript
/**
|
|
* PAX8 Sync Service
|
|
*
|
|
* Read-only current-state sync: companies, subscriptions (with dual cost),
|
|
* and a referenced-only product catalog. There is no incrementalSync — PAX8
|
|
* exposes no modified-since/delta filter for any of these entity types, so
|
|
* every run is a full sync with a full tombstone/reconciliation pass (D-07,
|
|
* D-08). This service calls only Pax8Client's read methods; it never issues
|
|
* an HTTP request itself and never calls a mutating PAX8 verb (PAX8-08).
|
|
*/
|
|
|
|
import postgresClient from './postgres-client';
|
|
import { Pax8Client } from './pax8-client';
|
|
import { getPax8Client } from './pax8-factory';
|
|
import { matchPax8Companies } from './pax8-company-matcher';
|
|
import type {
|
|
Pax8Company,
|
|
Pax8Subscription,
|
|
Pax8Product,
|
|
Pax8Invoice,
|
|
Pax8InvoiceItem,
|
|
Pax8EntitySyncResult,
|
|
Pax8SyncResult,
|
|
} from '@/lib/types/pax8';
|
|
|
|
/**
|
|
* Resolves the four dual-cost columns for a single invoice item, branching on
|
|
* `item.type` per 12-02-SUMMARY.md's live spot-check verdicts. All three
|
|
* observed types (subscription, prorate, one-time) were confirmed CONFIRM
|
|
* against the default mapping — no divergence was found — so this switch
|
|
* currently collapses to a single default branch. It's kept as a named,
|
|
* reviewable seam (not inline ternaries in the upsert call) so any future
|
|
* divergence discovered for a new item type has an obvious place to branch.
|
|
*/
|
|
function resolveCostColumns(item: Pax8InvoiceItem): {
|
|
unitPrice: number | null;
|
|
lineTotal: number | null;
|
|
partnerCost: number | null;
|
|
partnerCostTotal: number | null;
|
|
} {
|
|
switch (item.type) {
|
|
// 12-02-SUMMARY.md: subscription, prorate, and one-time all verified
|
|
// CONFIRM against the default mapping below — no per-type divergence to
|
|
// encode yet.
|
|
default:
|
|
return {
|
|
unitPrice: item.price ?? null,
|
|
lineTotal: item.amountDue ?? null,
|
|
partnerCost: item.cost ?? null,
|
|
partnerCostTotal: item.costTotal ?? null,
|
|
};
|
|
}
|
|
}
|
|
|
|
export class Pax8SyncService {
|
|
private client: Pax8Client;
|
|
private syncing = false;
|
|
|
|
constructor(client?: Pax8Client) {
|
|
this.client = client ?? getPax8Client();
|
|
}
|
|
|
|
isSyncInProgress(): boolean {
|
|
return this.syncing;
|
|
}
|
|
|
|
/**
|
|
* Full sync — companies, then subscriptions, then the referenced-only
|
|
* product catalog. No incrementalSync exists: PAX8 has no modified-since
|
|
* filter for companies/subscriptions/products, so full-sync-only covers
|
|
* every entity type (D-07), and every run performs the complete tombstone
|
|
* reconciliation pass unconditionally (D-08).
|
|
*/
|
|
async fullSync(triggeredBy = 'manual'): Promise<Pax8SyncResult> {
|
|
if (this.syncing) {
|
|
throw new Error('PAX8 sync already in progress');
|
|
}
|
|
this.syncing = true;
|
|
|
|
const syncId = `pax8_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
|
|
const startedAt = new Date();
|
|
const entities: Pax8EntitySyncResult[] = [];
|
|
|
|
console.log(`[Pax8Sync] Starting full sync (${syncId}) — triggered by ${triggeredBy}`);
|
|
|
|
await this.insertHistoryStarted(triggeredBy, startedAt);
|
|
|
|
try {
|
|
const companiesResult = await this.syncCompanies();
|
|
entities.push(companiesResult);
|
|
|
|
const { result: subscriptionsResult, referencedProductIds } = await this.syncSubscriptions();
|
|
entities.push(subscriptionsResult);
|
|
|
|
const productsResult = await this.syncProducts(referencedProductIds);
|
|
entities.push(productsResult);
|
|
|
|
const ordersResult = await this.syncOrders();
|
|
entities.push(ordersResult);
|
|
|
|
// Matching runs after companies + orders so pax8_companies is fully
|
|
// populated first (PAX8-10, PAX8-11).
|
|
const matchResult = await this.syncCompanyMatches();
|
|
entities.push(matchResult);
|
|
|
|
const completedAt = new Date();
|
|
const success = entities.every(e => e.success);
|
|
const status: 'completed' | 'failed' = success ? 'completed' : 'failed';
|
|
const totalUpserted = entities.reduce((sum, e) => sum + e.upserted, 0);
|
|
const totalTombstoned = entities.reduce((sum, e) => sum + e.tombstoned, 0);
|
|
const totalFlaggedForReview = entities.reduce((sum, e) => sum + (e.flaggedForReview ?? 0), 0);
|
|
const errors = entities.filter(e => e.error).map(e => `${e.entity}: ${e.error}`);
|
|
|
|
await this.updateHistory(startedAt, status, totalUpserted, totalTombstoned, errors.join('; ') || null);
|
|
|
|
console.log(
|
|
`[Pax8Sync] Full sync ${status} (${syncId}) — ${totalUpserted} upserted, ${totalTombstoned} tombstoned, ${totalFlaggedForReview} flagged for review`
|
|
);
|
|
|
|
return {
|
|
syncId,
|
|
status,
|
|
startedAt,
|
|
completedAt,
|
|
durationMs: completedAt.getTime() - startedAt.getTime(),
|
|
entities,
|
|
errors,
|
|
};
|
|
} catch (err) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
const completedAt = new Date();
|
|
console.error(`[Pax8Sync] Full sync failed catastrophically:`, msg);
|
|
|
|
const totalUpserted = entities.reduce((sum, e) => sum + e.upserted, 0);
|
|
const totalTombstoned = entities.reduce((sum, e) => sum + e.tombstoned, 0);
|
|
await this.updateHistory(startedAt, 'failed', totalUpserted, totalTombstoned, msg);
|
|
|
|
return {
|
|
syncId,
|
|
status: 'failed',
|
|
startedAt,
|
|
completedAt,
|
|
durationMs: completedAt.getTime() - startedAt.getTime(),
|
|
entities,
|
|
errors: [msg],
|
|
};
|
|
} finally {
|
|
this.syncing = false;
|
|
}
|
|
}
|
|
|
|
// ─── Entity syncs ───────────────────────────────────────────────────────
|
|
|
|
private async syncCompanies(): Promise<Pax8EntitySyncResult> {
|
|
const start = Date.now();
|
|
try {
|
|
const companies = await this.client.listAllCompanies();
|
|
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);
|
|
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
|
|
? 0
|
|
: (await postgresClient.query(
|
|
`UPDATE pax8_companies SET is_deleted = true, deleted_at = NOW()
|
|
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
|
[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: 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);
|
|
return { entity: 'companies', success: false, upserted: 0, tombstoned: 0, durationMs: Date.now() - start, error: msg };
|
|
}
|
|
}
|
|
|
|
private async syncSubscriptions(): Promise<{ result: Pax8EntitySyncResult; referencedProductIds: Set<string> }> {
|
|
const start = Date.now();
|
|
const referencedProductIds = new Set<string>();
|
|
try {
|
|
const subscriptions = await this.client.listAllSubscriptions();
|
|
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);
|
|
|
|
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
|
|
? 0
|
|
: (await postgresClient.query(
|
|
`UPDATE pax8_subscriptions SET is_deleted = true, deleted_at = NOW()
|
|
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
|
[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: 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) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
console.error(`[Pax8Sync] Subscription sync failed:`, msg);
|
|
return {
|
|
result: {
|
|
entity: 'subscriptions',
|
|
success: false,
|
|
upserted: 0,
|
|
tombstoned: 0,
|
|
durationMs: Date.now() - start,
|
|
error: msg,
|
|
},
|
|
referencedProductIds,
|
|
};
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Products are upserted ONLY for ids referenced by a subscription (D-01
|
|
* referenced-only catalog). PAX8 exposes no per-SKU product-detail
|
|
* endpoint, so the full catalog is fetched once via listAllProducts and
|
|
* filtered in memory to referencedProductIds — a single-pass mechanism,
|
|
* not a separate two-pass batch step (D-02).
|
|
*/
|
|
private async syncProducts(referencedProductIds: Set<string>): Promise<Pax8EntitySyncResult> {
|
|
const start = Date.now();
|
|
try {
|
|
const products = await this.client.listAllProducts();
|
|
console.log(`[Pax8Sync] Fetched ${products.length} products (catalog); ${referencedProductIds.size} referenced`);
|
|
|
|
const byId = new Map<string, Pax8Product>();
|
|
for (const p of products) {
|
|
if (p.id) byId.set(p.id, p);
|
|
}
|
|
|
|
let upserted = 0;
|
|
let failed = 0;
|
|
let firstError: string | null = null;
|
|
const seen: string[] = [];
|
|
for (const productId of referencedProductIds) {
|
|
const p = byId.get(productId);
|
|
if (!p) {
|
|
// Referenced-but-unknown (e.g. discontinued) — log and continue.
|
|
// Do not fabricate a row; the subscription's raw_payload retains productName.
|
|
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);
|
|
|
|
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
|
|
? 0
|
|
: (await postgresClient.query(
|
|
`UPDATE pax8_products SET is_deleted = true, deleted_at = NOW()
|
|
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
|
[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: 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);
|
|
return { entity: 'products', success: false, upserted: 0, tombstoned: 0, durationMs: Date.now() - start, error: msg };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Historical invoice/line-item sync (PAX8-06). Nested per-parent fetch
|
|
* (12-RESEARCH.md Pattern 1): page all invoice headers once, then for each
|
|
* header page its own items — there is no flat `/invoice-items` endpoint.
|
|
* Per-company cost lives ONLY on the item; `companyId` is always NULL on
|
|
* the header for this single-tenant reseller account (12-RESEARCH.md
|
|
* Pitfall 1), so `pax8_orders.pax8_company_id` intentionally stays NULL
|
|
* and all per-company cost joins go through `pax8_order_items` instead.
|
|
*/
|
|
private async syncOrders(): Promise<Pax8EntitySyncResult> {
|
|
const start = Date.now();
|
|
try {
|
|
const invoices = await this.client.listAllInvoices();
|
|
console.log(`[Pax8Sync] Fetched ${invoices.length} invoices`);
|
|
|
|
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);
|
|
|
|
try {
|
|
// 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 {
|
|
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(
|
|
`UPDATE pax8_order_items SET is_deleted = true, deleted_at = NOW()
|
|
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
|
[seenItemIds]
|
|
)).rowCount ?? 0;
|
|
|
|
const ordersTombstoned = seenOrderIds.length === 0
|
|
? 0
|
|
: (await postgresClient.query(
|
|
`UPDATE pax8_orders SET is_deleted = true, deleted_at = NOW()
|
|
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
|
[seenOrderIds]
|
|
)).rowCount ?? 0;
|
|
|
|
const tombstoned = itemsTombstoned + ordersTombstoned;
|
|
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: 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);
|
|
console.error(`[Pax8Sync] Order sync failed:`, msg);
|
|
return { entity: 'orders', success: false, upserted: 0, tombstoned: 0, durationMs: Date.now() - start, error: msg };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* PAX8 <-> Autotask company fuzzy-name matching (PAX8-10, PAX8-11).
|
|
* Delegates entirely to matchPax8Companies() (Plan 03) — this wrapper just
|
|
* shapes the result into the standard Pax8EntitySyncResult so it rolls up
|
|
* into fullSync()'s totals and sync_history the same way every other
|
|
* entity step does.
|
|
*/
|
|
private async syncCompanyMatches(): Promise<Pax8EntitySyncResult> {
|
|
const start = Date.now();
|
|
try {
|
|
const result = await matchPax8Companies();
|
|
const flaggedForReview = result.flaggedAmbiguous + result.flaggedNoCandidate;
|
|
if (flaggedForReview > 0) {
|
|
console.log(`[Pax8Sync] Company match: ${flaggedForReview} compan${flaggedForReview === 1 ? 'y' : 'ies'} flagged for manual review (${result.flaggedAmbiguous} ambiguous, ${result.flaggedNoCandidate} no-candidate)`);
|
|
}
|
|
return {
|
|
entity: 'company_matches',
|
|
success: true,
|
|
upserted: result.autoLinked,
|
|
// No soft-deletes are ever performed by matching — real deletion
|
|
// counts belong in `tombstoned`/sync_history.records_deleted, and
|
|
// conflating the two fabricates "deleted" counts in the admin Sync
|
|
// History UI (CR-02). Review-flag counts are reported separately via
|
|
// `flaggedForReview`.
|
|
tombstoned: 0,
|
|
flaggedForReview,
|
|
durationMs: result.durationMs,
|
|
};
|
|
} catch (err) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
console.error(`[Pax8Sync] Company match failed:`, msg);
|
|
return {
|
|
entity: 'company_matches',
|
|
success: false,
|
|
upserted: 0,
|
|
tombstoned: 0,
|
|
durationMs: Date.now() - start,
|
|
error: msg,
|
|
};
|
|
}
|
|
}
|
|
|
|
// ─── Helpers ────────────────────────────────────────────────────────────
|
|
|
|
private async insertHistoryStarted(triggeredBy: string, startedAt: Date): Promise<void> {
|
|
try {
|
|
await postgresClient.query(
|
|
`INSERT INTO sync_history (entity_type, sync_type, status, started_at, triggered_by)
|
|
VALUES ('pax8', 'full', 'started', $1, $2)`,
|
|
[startedAt, triggeredBy]
|
|
);
|
|
} catch (err) {
|
|
console.warn('[Pax8Sync] Could not create sync_history row:', err);
|
|
}
|
|
}
|
|
|
|
private async updateHistory(
|
|
startedAt: Date,
|
|
status: 'completed' | 'failed',
|
|
recordsAdded: number,
|
|
recordsDeleted: number,
|
|
errorMessage: string | null
|
|
): Promise<void> {
|
|
try {
|
|
await postgresClient.query(
|
|
`UPDATE sync_history
|
|
SET status = $1, completed_at = NOW(), records_added = $2, records_deleted = $3, error_message = $4
|
|
WHERE entity_type = 'pax8' AND sync_type = 'full' AND started_at = $5`,
|
|
[status, recordsAdded, recordsDeleted, errorMessage, startedAt]
|
|
);
|
|
} catch (err) {
|
|
console.warn('[Pax8Sync] Could not update sync_history row:', err);
|
|
}
|
|
}
|
|
}
|
|
|
|
let _instance: Pax8SyncService | null = null;
|
|
|
|
export function getPax8SyncService(): Pax8SyncService {
|
|
if (!_instance) {
|
|
_instance = new Pax8SyncService();
|
|
}
|
|
return _instance;
|
|
}
|