wulf-pulse/lib/services/ticket-reconciliation-service.ts

/**
 * Ticket Reconciliation Service (STOPGAP)
 *
 * Nightly backstop for drift between the postgres `tickets` mirror and Autotask.
 * Webhooks can drop events and the incremental sync's high-water-mark filter
 * can miss tickets whose lastActivityDate never advances after a status change.
 * This job bounds local staleness at 7 days by re-fetching each stale-open
 * ticket from Autotask and either re-upserting or soft-deleting.
 *
 * NOTE: This is a stopgap. The underlying incremental-filter issue is tracked
 * separately and should fix the root cause.
 */

import { postgresClient } from './postgres-client';
import { AutotaskClient } from './autotask-client';
import { autotaskRateLimiter } from './rate-limiter';
import { mapAutotaskToDatabase } from '../utils/entity-mapper';
import { getTableName } from '../utils/sync-helpers';
import { EntityType } from '../types/sync';
import { createSyncLogger } from '../utils/sync-logger';

export interface ReconcileResult {
  scanned: number;
  updated: number;
  statusFlippedToComplete: number;
  softDeleted: number;
  errors: number;
}

const STALE_DAYS = 7;
const SCAN_LIMIT = 500;
const COMPLETE_STATUS = 5; // Autotask status id for Complete

let _client: AutotaskClient | null = null;
function getClient(): AutotaskClient {
  if (!_client) {
    _client = new AutotaskClient({
      apiUrl: process.env.AUTOTASK_API_URL || '',
      username: process.env.AUTOTASK_USERNAME || '',
      password: process.env.AUTOTASK_SECRET || '',
      apiIntegrationCode: process.env.AUTOTASK_API_INTEGRATION_CODE || '',
    });
  }
  return _client;
}

/**
 * Re-fetch up to SCAN_LIMIT stale-open tickets from Autotask and reconcile
 * the local mirror. Idempotent and safe to run alongside the regular sync.
 */
export async function reconcileStaleTickets(): Promise<ReconcileResult> {
  const logger = createSyncLogger({ component: 'TicketReconciliation' });
  const startedAt = Date.now();
  const result: ReconcileResult = {
    scanned: 0,
    updated: 0,
    statusFlippedToComplete: 0,
    softDeleted: 0,
    errors: 0,
  };

  // 1. Find stale-open tickets in the postgres mirror.
  const staleQuery = `
    SELECT id, status
    FROM tickets
    WHERE is_deleted = false
      AND status <> $1
      AND synced_at < NOW() - INTERVAL '${STALE_DAYS} days'
    ORDER BY synced_at ASC
    LIMIT $2
  `;
  const stale = await postgresClient.query<{ id: string; status: number | null }>(
    staleQuery,
    [COMPLETE_STATUS, SCAN_LIMIT]
  );
  result.scanned = stale.rows.length;

  logger.info(`Scanning ${result.scanned} stale-open tickets (>${STALE_DAYS}d, limit=${SCAN_LIMIT})`);

  if (result.scanned === 0) {
    logger.info('No stale tickets found', { duration: Date.now() - startedAt });
    return result;
  }

  const client = getClient();
  const tableName = getTableName(EntityType.TICKETS); // 'tickets'

  // 2. Walk each id. Use the singleton rate limiter as an extra guard on top of
  //    AutotaskClient's internal limiter (defense in depth for parallel callers).
  for (const row of stale.rows) {
    const ticketId = Number(row.id);
    const priorStatus = row.status;
    try {
      const entity = await autotaskRateLimiter.throttle(() =>
        client.getEntityById<Record<string, any>>('Tickets', ticketId)
      );

      if (!entity) {
        // Not in Autotask anymore -> soft-delete locally.
        await postgresClient.query(
          `UPDATE ${tableName}
             SET is_deleted = true, deleted_at = NOW(), updated_at = NOW()
           WHERE id = $1`,
          [ticketId]
        );
        result.softDeleted += 1;
        continue;
      }

      // 3. Upsert using the same SQL shape the webhook handler uses.
      const mapped = mapAutotaskToDatabase(EntityType.TICKETS, entity);
      if (!mapped) {
        throw new Error(`mapAutotaskToDatabase returned null for ticket ${ticketId}`);
      }
      const keys = Object.keys(mapped);
      const values = Object.values(mapped);
      const placeholders = keys.map((_, i) => `$${i + 1}`).join(', ');
      const updateClause = keys
        .filter(k => k !== 'id')
        .map(k => `${k} = EXCLUDED.${k}`)
        .join(', ');
      const upsertSql = `
        INSERT INTO ${tableName} (${keys.join(', ')})
        VALUES (${placeholders})
        ON CONFLICT (id)
        DO UPDATE SET ${updateClause}, updated_at = NOW()
      `;
      await postgresClient.query(upsertSql, values);
      result.updated += 1;

      const newStatus = (mapped.status ?? null) as number | null;
      if (priorStatus !== COMPLETE_STATUS && newStatus === COMPLETE_STATUS) {
        result.statusFlippedToComplete += 1;
      }
    } catch (err) {
      result.errors += 1;
      logger.warn(
        `Reconcile failed for ticket ${ticketId}`,
        { ticketId },
        err instanceof Error ? err : new Error(String(err))
      );
    }
  }

  logger.info(
    `Reconciliation complete: scanned=${result.scanned} updated=${result.updated} flippedComplete=${result.statusFlippedToComplete} softDeleted=${result.softDeleted} errors=${result.errors}`,
    { duration: Date.now() - startedAt }
  );

  return result;
}