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

/**
 * 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 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<Pax8EntitySyncResult> {
    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<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;
      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<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;
      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 };
    }
  }

  /**
   * 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;
      const seenOrderIds: string[] = [];
      const seenItemIds: string[] = [];

      for (const invoice of invoices) {
        if (!invoice.id) continue;
        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);

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

      // Child-then-parent tombstoning to respect the order_items -> orders FK.
      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)`);
      }

      return {
        entity: 'orders',
        success: true,
        upserted: ordersUpserted + itemsUpserted,
        tombstoned,
        durationMs: Date.now() - start,
      };
    } 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();
      return {
        entity: 'company_matches',
        success: true,
        upserted: result.autoLinked,
        // Repurposed: not a soft-delete tombstone count — the number of
        // pax8_companies rows flagged for manual review this run (ambiguous
        // + no-candidate), surfaced through the same rollup field.
        tombstoned: result.flaggedAmbiguous + result.flaggedNoCandidate,
        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;
}