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

/**
 * QuickBooks Online Sync Service
 * Fetches invoices, payments, deposits, transactions, and financial reports
 * from QBO and upserts them into PostgreSQL.
 */

import postgresClient from './postgres-client';
import { QboClient, getQboClient } from './qbo-client';
import {
  QboInvoice,
  QboPayment,
  QboDeposit,
  QboPurchase,
  QboJournalEntry,
  QboReport,
  QboSyncResult,
  QboEntitySyncResult,
} from '@/lib/types/qbo';

export class QboSyncService {
  private client: QboClient;
  private isSyncing = false;

  constructor(client?: QboClient) {
    this.client = client || getQboClient();
  }

  isSyncInProgress(): boolean {
    return this.isSyncing;
  }

  async fullSync(triggeredBy = 'system'): Promise<QboSyncResult> {
    return this.executeSync('full', triggeredBy);
  }

  async incrementalSync(triggeredBy = 'system'): Promise<QboSyncResult> {
    return this.executeSync('incremental', triggeredBy);
  }

  private async executeSync(
    syncType: 'full' | 'incremental',
    triggeredBy: string
  ): Promise<QboSyncResult> {
    if (this.isSyncing) {
      throw new Error('QBO sync already in progress');
    }

    this.isSyncing = true;
    const syncId = `qbo_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
    const startedAt = new Date();
    const entities: QboEntitySyncResult[] = [];
    const errors: string[] = [];

    const realmId = process.env.QBO_REALM_ID || '';

    console.log(`[QboSync] Starting ${syncType} sync (${syncId}) triggered by ${triggeredBy}`);

    try {
      // Determine updatedSince for incremental
      let updatedSince: Date | undefined;
      if (syncType === 'incremental') {
        updatedSince = await this.getLastSyncTime();
      }

      // Sync invoices.  Pass syncType so a full sync can tombstone any
      // rows QBO no longer returns (voided / deleted there).
      entities.push(await this.syncInvoices(realmId, updatedSince, syncType));

      // Sync payments
      entities.push(await this.syncPayments(realmId, updatedSince));

      // Sync deposits
      entities.push(await this.syncDeposits(realmId, updatedSince));

      // Sync purchases (expenses/credit card charges)
      entities.push(await this.syncPurchases(realmId, updatedSince));

      // Sync journal entries
      entities.push(await this.syncJournalEntries(realmId, updatedSince));

      // Sync reports — always fetch current + last 12 months on full, current month on incremental
      if (syncType === 'full') {
        entities.push(await this.syncReports(realmId, 12));
      } else {
        entities.push(await this.syncReports(realmId, 1));
      }

      // Record sync history
      await this.recordSyncHistory(syncId, syncType, triggeredBy, startedAt, 'completed', entities);

    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      errors.push(msg);
      console.error(`[QboSync] Sync failed: ${msg}`);
      await this.recordSyncHistory(syncId, syncType, triggeredBy, startedAt, 'failed', entities, msg);
    } finally {
      this.isSyncing = false;
    }

    const completedAt = new Date();
    return {
      syncId,
      realmId,
      status: errors.length === 0 ? 'completed' : 'failed',
      startedAt,
      completedAt,
      duration: completedAt.getTime() - startedAt.getTime(),
      entities,
      errors,
    };
  }

  // ─── Entity Sync Methods ───────────────────────────────────────────────────

  private async syncInvoices(
    realmId: string,
    updatedSince?: Date,
    syncType: 'full' | 'incremental' = 'incremental',
  ): Promise<QboEntitySyncResult> {
    const start = Date.now();
    try {
      const invoices = await this.client.getInvoices(updatedSince);
      console.log(`[QboSync] Fetched ${invoices.length} invoices`);

      let upserted = 0;
      const seenIds: string[] = [];
      for (const inv of invoices) {
        if (inv.Id) seenIds.push(inv.Id);
        const status = this.deriveInvoiceStatus(inv);
        await postgresClient.query(
          `INSERT INTO qbo_invoices
             (id, realm_id, doc_number, txn_date, due_date, customer_ref_id, customer_ref_name,
              email_address, total_amt, balance, status, currency_code, line_items, linked_txns,
              sync_token, qbo_created_at, qbo_updated_at, synced_at)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,NOW())
           ON CONFLICT (id) DO UPDATE SET
             doc_number = EXCLUDED.doc_number,
             txn_date = EXCLUDED.txn_date,
             due_date = EXCLUDED.due_date,
             customer_ref_id = EXCLUDED.customer_ref_id,
             customer_ref_name = EXCLUDED.customer_ref_name,
             email_address = EXCLUDED.email_address,
             total_amt = EXCLUDED.total_amt,
             balance = EXCLUDED.balance,
             status = EXCLUDED.status,
             currency_code = EXCLUDED.currency_code,
             line_items = EXCLUDED.line_items,
             linked_txns = EXCLUDED.linked_txns,
             sync_token = EXCLUDED.sync_token,
             qbo_updated_at = EXCLUDED.qbo_updated_at,
             synced_at = NOW(),
             is_deleted = false,
             deleted_at = NULL`,
          [
            inv.Id, realmId, inv.DocNumber ?? null,
            inv.TxnDate ?? null, inv.DueDate ?? null,
            inv.CustomerRef?.value ?? null, inv.CustomerRef?.name ?? null,
            inv.BillEmail?.Address ?? null,
            inv.TotalAmt ?? null, inv.Balance ?? null,
            status, inv.CurrencyRef?.value ?? 'USD',
            JSON.stringify(inv.Line ?? []),
            JSON.stringify(inv.LinkedTxn ?? []),
            inv.SyncToken,
            inv.MetaData?.CreateTime ?? null,
            inv.MetaData?.LastUpdatedTime ?? null,
          ]
        );
        upserted++;
      }

      // Tombstone pass — full syncs only.
      //
      // QBO's Invoice query never reports deletions; voided/deleted invoices
      // just stop appearing.  A full sync just fetched every live invoice
      // for this realm, so any Pulse row not in `seenIds` is an orphan and
      // should be soft-deleted so it stops counting toward A/R.
      //
      // Incremental syncs are skipped here — they only see recently-changed
      // rows, so applying the same logic would tombstone the entire ledger.
      let tombstoned = 0;
      if (syncType === 'full' && seenIds.length > 0) {
        const tombstoneRes = await postgresClient.query<{ id: string }>(
          `UPDATE qbo_invoices
              SET is_deleted = true,
                  deleted_at = NOW()
            WHERE realm_id = $1
              AND is_deleted = false
              AND id <> ALL($2::text[])
            RETURNING id`,
          [realmId, seenIds],
        );
        tombstoned = tombstoneRes.rowCount ?? tombstoneRes.rows.length;
        if (tombstoned > 0) {
          console.log(`[QboSync] Tombstoned ${tombstoned} invoice(s) absent from full-sync response`);
        }
      }

      return {
        entity: 'invoices',
        success: true,
        recordsUpserted: upserted,
        recordsDeleted: tombstoned,
        duration: Date.now() - start,
      };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Invoice sync failed: ${msg}`);
      return { entity: 'invoices', success: false, recordsUpserted: 0, duration: Date.now() - start, error: msg };
    }
  }

  private async syncPayments(realmId: string, updatedSince?: Date): Promise<QboEntitySyncResult> {
    const start = Date.now();
    try {
      const payments = await this.client.getPayments(updatedSince);
      console.log(`[QboSync] Fetched ${payments.length} payments`);

      let upserted = 0;
      for (const pay of payments) {
        await postgresClient.query(
          `INSERT INTO qbo_payments
             (id, realm_id, txn_date, customer_ref_id, customer_ref_name, total_amt, unapplied_amt,
              currency_code, payment_method_ref, deposit_account_ref, linked_txns,
              sync_token, qbo_created_at, qbo_updated_at, synced_at)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,NOW())
           ON CONFLICT (id) DO UPDATE SET
             txn_date = EXCLUDED.txn_date,
             customer_ref_id = EXCLUDED.customer_ref_id,
             customer_ref_name = EXCLUDED.customer_ref_name,
             total_amt = EXCLUDED.total_amt,
             unapplied_amt = EXCLUDED.unapplied_amt,
             currency_code = EXCLUDED.currency_code,
             payment_method_ref = EXCLUDED.payment_method_ref,
             deposit_account_ref = EXCLUDED.deposit_account_ref,
             linked_txns = EXCLUDED.linked_txns,
             sync_token = EXCLUDED.sync_token,
             qbo_updated_at = EXCLUDED.qbo_updated_at,
             synced_at = NOW()`,
          [
            pay.Id, realmId, pay.TxnDate ?? null,
            pay.CustomerRef?.value ?? null, pay.CustomerRef?.name ?? null,
            pay.TotalAmt ?? null, pay.UnappliedAmt ?? null,
            pay.CurrencyRef?.value ?? 'USD',
            pay.PaymentMethodRef?.name ?? null,
            pay.DepositToAccountRef?.name ?? null,
            JSON.stringify(pay.Line ?? []),
            pay.SyncToken,
            pay.MetaData?.CreateTime ?? null,
            pay.MetaData?.LastUpdatedTime ?? null,
          ]
        );
        upserted++;
      }

      return { entity: 'payments', success: true, recordsUpserted: upserted, duration: Date.now() - start };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Payment sync failed: ${msg}`);
      return { entity: 'payments', success: false, recordsUpserted: 0, duration: Date.now() - start, error: msg };
    }
  }

  private async syncDeposits(realmId: string, updatedSince?: Date): Promise<QboEntitySyncResult> {
    const start = Date.now();
    try {
      const deposits = await this.client.getDeposits(updatedSince);
      console.log(`[QboSync] Fetched ${deposits.length} deposits`);

      let upserted = 0;
      for (const dep of deposits) {
        await postgresClient.query(
          `INSERT INTO qbo_deposits
             (id, realm_id, txn_date, deposit_to_account_ref_id, deposit_to_account_ref_name,
              total_amt, line_items, sync_token, qbo_created_at, qbo_updated_at, synced_at)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,NOW())
           ON CONFLICT (id) DO UPDATE SET
             txn_date = EXCLUDED.txn_date,
             deposit_to_account_ref_id = EXCLUDED.deposit_to_account_ref_id,
             deposit_to_account_ref_name = EXCLUDED.deposit_to_account_ref_name,
             total_amt = EXCLUDED.total_amt,
             line_items = EXCLUDED.line_items,
             sync_token = EXCLUDED.sync_token,
             qbo_updated_at = EXCLUDED.qbo_updated_at,
             synced_at = NOW()`,
          [
            dep.Id, realmId, dep.TxnDate ?? null,
            dep.DepositToAccountRef?.value ?? null,
            dep.DepositToAccountRef?.name ?? null,
            dep.TotalAmt ?? null,
            JSON.stringify(dep.Line ?? []),
            dep.SyncToken,
            dep.MetaData?.CreateTime ?? null,
            dep.MetaData?.LastUpdatedTime ?? null,
          ]
        );
        upserted++;
      }

      return { entity: 'deposits', success: true, recordsUpserted: upserted, duration: Date.now() - start };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Deposit sync failed: ${msg}`);
      return { entity: 'deposits', success: false, recordsUpserted: 0, duration: Date.now() - start, error: msg };
    }
  }

  private async syncPurchases(realmId: string, updatedSince?: Date): Promise<QboEntitySyncResult> {
    const start = Date.now();
    try {
      const purchases = await this.client.getPurchases(updatedSince);
      console.log(`[QboSync] Fetched ${purchases.length} purchases`);

      let upserted = 0;
      for (const p of purchases) {
        await postgresClient.query(
          `INSERT INTO qbo_transactions
             (id, txn_type, realm_id, txn_date, doc_number, entity_ref_id, entity_ref_name,
              entity_type, account_ref_id, account_ref_name, total_amt, currency_code,
              private_note, line_items, sync_token, qbo_created_at, qbo_updated_at, synced_at)
           VALUES ($1,'Purchase',$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,NOW())
           ON CONFLICT (id, txn_type) DO UPDATE SET
             txn_date = EXCLUDED.txn_date,
             doc_number = EXCLUDED.doc_number,
             entity_ref_id = EXCLUDED.entity_ref_id,
             entity_ref_name = EXCLUDED.entity_ref_name,
             entity_type = EXCLUDED.entity_type,
             account_ref_id = EXCLUDED.account_ref_id,
             account_ref_name = EXCLUDED.account_ref_name,
             total_amt = EXCLUDED.total_amt,
             currency_code = EXCLUDED.currency_code,
             private_note = EXCLUDED.private_note,
             line_items = EXCLUDED.line_items,
             sync_token = EXCLUDED.sync_token,
             qbo_updated_at = EXCLUDED.qbo_updated_at,
             synced_at = NOW()`,
          [
            p.Id, realmId, p.TxnDate ?? null, p.DocNumber ?? null,
            p.EntityRef?.value ?? null, p.EntityRef?.name ?? null,
            p.EntityRef?.type ?? null,
            p.AccountRef?.value ?? null, p.AccountRef?.name ?? null,
            p.TotalAmt ?? null, p.CurrencyRef?.value ?? 'USD',
            p.PrivateNote ?? null,
            JSON.stringify(p.Line ?? []),
            p.SyncToken,
            p.MetaData?.CreateTime ?? null,
            p.MetaData?.LastUpdatedTime ?? null,
          ]
        );
        upserted++;
      }

      return { entity: 'purchases', success: true, recordsUpserted: upserted, duration: Date.now() - start };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Purchase sync failed: ${msg}`);
      return { entity: 'purchases', success: false, recordsUpserted: 0, duration: Date.now() - start, error: msg };
    }
  }

  private async syncJournalEntries(realmId: string, updatedSince?: Date): Promise<QboEntitySyncResult> {
    const start = Date.now();
    try {
      const entries = await this.client.getJournalEntries(updatedSince);
      console.log(`[QboSync] Fetched ${entries.length} journal entries`);

      let upserted = 0;
      for (const je of entries) {
        await postgresClient.query(
          `INSERT INTO qbo_transactions
             (id, txn_type, realm_id, txn_date, doc_number, total_amt, currency_code,
              private_note, line_items, sync_token, qbo_created_at, qbo_updated_at, synced_at)
           VALUES ($1,'JournalEntry',$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,NOW())
           ON CONFLICT (id, txn_type) DO UPDATE SET
             txn_date = EXCLUDED.txn_date,
             doc_number = EXCLUDED.doc_number,
             total_amt = EXCLUDED.total_amt,
             currency_code = EXCLUDED.currency_code,
             private_note = EXCLUDED.private_note,
             line_items = EXCLUDED.line_items,
             sync_token = EXCLUDED.sync_token,
             qbo_updated_at = EXCLUDED.qbo_updated_at,
             synced_at = NOW()`,
          [
            je.Id, realmId, je.TxnDate ?? null, je.DocNumber ?? null,
            je.TotalAmt ?? null, je.CurrencyRef?.value ?? 'USD',
            je.PrivateNote ?? null,
            JSON.stringify(je.Line ?? []),
            je.SyncToken,
            je.MetaData?.CreateTime ?? null,
            je.MetaData?.LastUpdatedTime ?? null,
          ]
        );
        upserted++;
      }

      return { entity: 'journal_entries', success: true, recordsUpserted: upserted, duration: Date.now() - start };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Journal entry sync failed: ${msg}`);
      return { entity: 'journal_entries', success: false, recordsUpserted: 0, duration: Date.now() - start, error: msg };
    }
  }

  private async syncReports(realmId: string, monthsBack: number): Promise<QboEntitySyncResult> {
    const start = Date.now();
    let upserted = 0;
    let skipped = 0;
    const errors: string[] = [];

    try {
      // Fetch company accounting preference once
      const { reportBasis } = await this.client.getPreferences();
      console.log(`[QboSync] Report sync using accounting method: ${reportBasis}`);

      const now = new Date();

      for (let i = 0; i < monthsBack; i++) {
        const periodEnd = new Date(now.getFullYear(), now.getMonth() - i + 1, 0);
        const periodStart = new Date(now.getFullYear(), now.getMonth() - i, 1);

        const startStr = this.formatDate(periodStart);
        const endStr = this.formatDate(periodEnd);

        // P&L
        try {
          const pl = await this.client.getProfitAndLoss(startStr, endStr, reportBasis);
          if (pl?.Header?.NoReportData === 'true') {
            console.log(`[QboSync] P&L ${startStr}: no data, skipping`);
            skipped++;
          } else {
            await this.upsertReport(realmId, 'ProfitAndLoss', periodStart, periodEnd, pl);
            console.log(`[QboSync] P&L ${startStr}: upserted`);
            upserted++;
          }
        } catch (err) {
          const msg = err instanceof Error ? err.message : String(err);
          errors.push(`P&L ${startStr}: ${msg}`);
        }

        // Balance Sheet
        try {
          const bs = await this.client.getBalanceSheet(startStr, endStr, reportBasis);
          if (bs?.Header?.NoReportData === 'true') {
            console.log(`[QboSync] BalanceSheet ${startStr}: no data, skipping`);
            skipped++;
          } else {
            await this.upsertReport(realmId, 'BalanceSheet', periodStart, periodEnd, bs);
            console.log(`[QboSync] BalanceSheet ${startStr}: upserted`);
            upserted++;
          }
        } catch (err) {
          const msg = err instanceof Error ? err.message : String(err);
          errors.push(`BalanceSheet ${startStr}: ${msg}`);
        }

        // Cash Flow
        try {
          const cf = await this.client.getCashFlow(startStr, endStr);
          if (cf?.Header?.NoReportData === 'true') {
            console.log(`[QboSync] CashFlow ${startStr}: no data, skipping`);
            skipped++;
          } else {
            await this.upsertReport(realmId, 'CashFlow', periodStart, periodEnd, cf);
            console.log(`[QboSync] CashFlow ${startStr}: upserted`);
            upserted++;
          }
        } catch (err) {
          const msg = err instanceof Error ? err.message : String(err);
          errors.push(`CashFlow ${startStr}: ${msg}`);
        }
      }

      console.log(`[QboSync] Reports: ${upserted} upserted, ${skipped} skipped, ${errors.length} errors`);
      if (errors.length > 0) {
        console.warn(`[QboSync] Report errors: ${errors.join('; ')}`);
      }

      return {
        entity: 'reports',
        success: errors.length === 0,
        recordsUpserted: upserted,
        duration: Date.now() - start,
        error: errors.length > 0 ? errors.join('; ') : undefined,
      };
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      console.error(`[QboSync] Report sync failed: ${msg}`);
      return { entity: 'reports', success: false, recordsUpserted: upserted, duration: Date.now() - start, error: msg };
    }
  }

  private async upsertReport(
    realmId: string,
    reportType: string,
    periodStart: Date,
    periodEnd: Date,
    data: QboReport
  ): Promise<void> {
    await postgresClient.query(
      `INSERT INTO qbo_reports (realm_id, report_type, period_start, period_end, report_data, synced_at)
       VALUES ($1, $2, $3, $4, $5, NOW())
       ON CONFLICT (realm_id, report_type, period_start, period_end) DO UPDATE SET
         report_data = EXCLUDED.report_data,
         synced_at = NOW()`,
      [realmId, reportType, periodStart, periodEnd, JSON.stringify(data)]
    );
  }

  // ─── Helpers ────────────────────────────────────────────────────────────────

  private deriveInvoiceStatus(inv: QboInvoice): string {
    if ((inv.Balance ?? 0) === 0 && (inv.TotalAmt ?? 0) > 0) return 'Paid';
    if ((inv.Balance ?? 0) > 0 && inv.DueDate && new Date(inv.DueDate) < new Date()) return 'Overdue';
    if ((inv.Balance ?? 0) > 0) return 'Open';
    return 'Unknown';
  }

  private formatDate(d: Date): string {
    return d.toISOString().split('T')[0];
  }

  private async getLastSyncTime(): Promise<Date | undefined> {
    try {
      const result = await postgresClient.query<{ synced_at: Date }>(
        `SELECT MAX(synced_at) as synced_at FROM qbo_invoices`
      );
      return result.rows[0]?.synced_at ?? undefined;
    } catch {
      return undefined;
    }
  }

  private async recordSyncHistory(
    _syncId: string,
    syncType: string,
    triggeredBy: string,
    startedAt: Date,
    status: string,
    entities: QboEntitySyncResult[],
    errorMessage?: string
  ): Promise<void> {
    try {
      const totalRecords = entities.reduce((sum, e) => sum + e.recordsUpserted, 0);
      const normalizedType = syncType === 'incremental' ? 'incremental' : 'full';
      await postgresClient.query(
        `INSERT INTO sync_history
           (entity_type, sync_type, status, records_added, started_at, completed_at, triggered_by, error_message)
         VALUES ('qbo', $1, $2, $3, $4, NOW(), $5, $6)`,
        [normalizedType, status, totalRecords, startedAt, triggeredBy, errorMessage ?? null]
      );
    } catch (err) {
      console.warn('[QboSync] Failed to record sync history:', err);
    }
  }
}

let _instance: QboSyncService | null = null;

export function getQboSyncService(): QboSyncService {
  if (!_instance) {
    _instance = new QboSyncService();
  }
  return _instance;
}