wulf-pulse/lib/services/pax8-sync-service.ts

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;
}