diff --git a/lib/services/pax8-sync-service.ts b/lib/services/pax8-sync-service.ts new file mode 100644 index 0000000..fe990e0 --- /dev/null +++ b/lib/services/pax8-sync-service.ts @@ -0,0 +1,368 @@ +/** + * 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 type { + Pax8Company, + Pax8Subscription, + Pax8Product, + Pax8EntitySyncResult, + Pax8SyncResult, +} from '@/lib/types/pax8'; + +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 { + 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 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 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` + ); + + 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 { + const start = Date.now(); + try { + const companies = await this.client.listAllCompanies(); + console.log(`[Pax8Sync] Fetched ${companies.length} companies`); + + let upserted = 0; + const seen: string[] = []; + for (const c of companies) { + if (!c.id) continue; + 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++; + } + + 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'}`); + + return { entity: 'companies', success: true, upserted, tombstoned, durationMs: Date.now() - start }; + } 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 }> { + const start = Date.now(); + const referencedProductIds = new Set(); + try { + const subscriptions = await this.client.listAllSubscriptions(); + console.log(`[Pax8Sync] Fetched ${subscriptions.length} subscriptions`); + + let upserted = 0; + const seen: string[] = []; + for (const s of subscriptions) { + if (!s.id) continue; + 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++; + } + + 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)`); + + return { + result: { entity: 'subscriptions', success: true, upserted, tombstoned, durationMs: Date.now() - start }, + 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): Promise { + 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(); + for (const p of products) { + if (p.id) byId.set(p.id, p); + } + + let upserted = 0; + 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; + } + 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++; + } + + 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)`); + + return { entity: 'products', success: true, upserted, tombstoned, durationMs: Date.now() - start }; + } 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 }; + } + } + + // ─── Helpers ──────────────────────────────────────────────────────────── + + private async insertHistoryStarted(triggeredBy: string, startedAt: Date): Promise { + 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 { + 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; +}