Синхронизация данных по расписанию с инкрементальным курсором

{
  "schemaVersion": 1,
  "nodes": [
    { "id": "cron", "kind": "cronTrigger", "name": "Hourly", "position": { "x": 0, "y": 0 }, "config": { "cron": "0 * * * *", "timezone": "UTC" } },
    { "id": "sync", "kind": "lambda", "name": "Sync CRM contacts", "position": { "x": 240, "y": 0 }, "config": { "functionId": "sync-crm-contacts", "onError": "failPipeline" } }
  ],
  "edges": [
    { "id": "e1", "sourceNodeId": "cron", "targetNodeId": "sync", "sourceHandle": "default" }
  ]
}

export async function handler(event) {
  // Read watermark — default to 24h ago on first run
  const cursor = await db.syncCursors.get('crm-contacts') ?? new Date(Date.now() - 86_400_000).toISOString();
  let page = 0, synced = 0, newCursor = cursor;
  do {
    const { contacts, nextPage } = await crm.fetchContacts({ updatedAfter: cursor, page });
    if (contacts.length === 0) break;
    await db.contacts.upsertBatch(contacts); // upsert by contacts[].externalId
    synced += contacts.length;
    newCursor = contacts.at(-1)?.updatedAt ?? newCursor;
    page = nextPage;
  } while (page);
  await db.syncCursors.set('crm-contacts', newCursor);
  return { synced, cursor, newCursor };
}