name: crm-contact-sync
description: Hourly sync of CRM contacts to the contact store

trigger:
  type: schedule
  cron: "0 * * * *"  # every hour

steps:
  - id: fetch_contacts
    name: Fetch CRM Contacts
    description: Fetches all contacts from the CRM API with pagination
    file: fetch_contacts.ts
    credentials:
      - CRM_API_KEY
    on_success: upsert_contacts
    on_failure: handle_error

  - id: upsert_contacts
    name: Upsert Contacts
    description: Upserts contacts into the local data store
    file: upsert_contacts.ts
    input:
      contacts: "${{ steps.fetch_contacts.output.contacts }}"

  - id: handle_error
    name: Handle Error
    description: Logs sync failure
    file: handle_error.ts
    input:
      error_message: "${{ error.message }}"

export async function run(input: Record<string, any>, ctx: StepContext) {
  ctx.log("fetch_contacts: start", {});

  let allContacts: any[] = [];
  let page = 1;

  while (true) {
    const response = await ctx.http.get(`https://api.crm.example.com/contacts?page=${page}&limit=100`, {
      headers: { Authorization: `Bearer ${ctx.credentials.CRM_API_KEY}` },
      timeout: 25000,
    });
    if (!response.ok) {
      throw new Error(`CRM API returned ${response.status}`);
    }
    const data = await response.json();
    const batch = data.contacts || [];
    if (batch.length === 0) break;

    allContacts = allContacts.concat(batch);
    page++;
  }

  ctx.log("fetch_contacts: done", { count: allContacts.length });
  return { contacts: allContacts };
}

export async function run(input: Record<string, any>, ctx: StepContext) {
  const contacts: any[] = input.contacts;
  ctx.log("upsert_contacts: start", { count: contacts.length });

  let inserted = 0;
  let updated = 0;

  for (const contact of contacts) {
    const existing = await ctx.stores.crm_contacts.find_one({ crm_id: String(contact.id) });
    if (existing) {
      await ctx.stores.crm_contacts.update(
        { crm_id: String(contact.id) },
        { name: contact.name, email: contact.email }
      );
      updated++;
    } else {
      await ctx.stores.crm_contacts.insert({
        crm_id: String(contact.id),
        name: contact.name,
        email: contact.email,
      });
      inserted++;
    }
  }

  ctx.log("upsert_contacts: done", { inserted, updated });
  return { inserted, updated };
}