pig/apps/api/src/integrations/hubspot/sync.ts

import { createHash } from 'node:crypto';
import type { HubSpotObjectType, HubSpotRecord } from './contracts';

export interface HubSpotSyncCursor {
  after: string | null;
  phase: 'initial' | 'reconcile';
}

export interface HubSpotSyncStore {
  getCursor(connectionId: string, objectType: HubSpotObjectType): Promise<HubSpotSyncCursor>;
  /** Records and the next cursor must commit in the same transaction. */
  commitPage(input: {
    connectionId: string;
    objectType: HubSpotObjectType;
    phase: 'initial' | 'reconcile';
    expectedAfter: string | null;
    nextAfter: string | null;
    records: readonly HubSpotSyncRecord[];
    completedAt: Date;
  }): Promise<void>;
}

export interface HubSpotSyncTokenProvider {
  getAccessToken(connectionId: string, signal?: AbortSignal): Promise<string>;
}

export interface HubSpotSyncCrmClient {
  listObjects(
    accessToken: string,
    objectType: HubSpotObjectType,
    options?: { after?: string | null; signal?: AbortSignal },
  ): Promise<{ results: HubSpotRecord[]; nextAfter: string | null }>;
}

export interface HubSpotSyncRecord extends HubSpotRecord {
  contentHash: string;
  fetchedAt: Date;
}

export interface HubSpotSyncPageResult {
  objectType: HubSpotObjectType;
  records: number;
  nextAfter: string | null;
  complete: boolean;
}

export class HubSpotSyncService {
  constructor(
    private readonly store: HubSpotSyncStore,
    private readonly tokens: HubSpotSyncTokenProvider,
    private readonly crm: HubSpotSyncCrmClient,
    private readonly now: () => Date = () => new Date(),
  ) {}

  async syncNextPage(
    connectionId: string,
    objectType: HubSpotObjectType,
    signal?: AbortSignal,
  ): Promise<HubSpotSyncPageResult> {
    const cursor = await this.store.getCursor(connectionId, objectType);
    const accessToken = await this.tokens.getAccessToken(connectionId, signal);
    const page = await this.crm.listObjects(accessToken, objectType, {
      after: cursor.after,
      signal,
    });
    const fetchedAt = this.now();
    const records = page.results.map((record) => ({
      ...record,
      fetchedAt,
      contentHash: hashHubSpotRecord(record),
    }));
    await this.store.commitPage({
      connectionId,
      objectType,
      phase: cursor.phase,
      expectedAfter: cursor.after,
      nextAfter: page.nextAfter,
      records,
      completedAt: fetchedAt,
    });
    return {
      objectType,
      records: records.length,
      nextAfter: page.nextAfter,
      complete: page.nextAfter === null,
    };
  }
}

export function hashHubSpotRecord(record: HubSpotRecord): string {
  const properties = Object.fromEntries(
    Object.entries(record.properties).sort(([left], [right]) => left.localeCompare(right)),
  );
  return createHash('sha256').update(JSON.stringify({
    id: record.id,
    properties,
    createdAt: record.createdAt,
    updatedAt: record.updatedAt,
    archived: record.archived,
  })).digest('hex');
}