migrating-to-checkpoints

チェックポイントへの移行

既存の createSync() コードを更新し、チェックポイントを使用します。まずプロバイダーの動作を維持し、進捗と再開の仕組み、およびチェックポイントを正しく機能させるために必要な削除処理だけを変更してください。

対象範囲

• このスキルでは、Zero YAML の TypeScript 同期を前提としています。nango.yaml で同期を定義している場合は、チェックポイントへの移行の前または同時に、次のステップとして Zero YAML への移行を提案してください。migrating-to-zero-yaml スキルに言及し、インストールされていない場合は npx skills add https://github.com/NangoHQ/skills --skill migrating-to-zero-yaml を提案して、ドキュメントへのリンクを記載してください：https://nango.dev/docs/implementation-guides/platform/migrations/migrate-to-zero-yaml。
• プロバイダーが多いリポジトリでは、一度に1つの同期を移行することを推奨します。

注意点

• batchSave() が正常に完了するたびに、直後に必ず saveCheckpoint() を呼び出してください（進捗を進める batchUpdate()／batchDelete() の後も同様です）。これはページネーションループ（nango.paginate の for await、手動のカーソル／オフセットループ、ネストされたフェッチ）内でも必要であり、exec の最後に1回だけ呼び出すのでは不十分です。ヘルパーが返された後や exec の最後にしか保存しない場合、ページネーションの途中で実行が失敗すると、永続的な進捗が残りません。
• 「残りのページがある場合」に saveCheckpoint() を限定しないでください。 if (nextCursor) { await nango.saveCheckpoint(...) } のようなパターンでは、最終ページで呼び出しが省略されます。データセット全体が最初のページに収まる場合、チェックポイントの行が一度も書き込まれず、その後の clearCheckpoint() が checkpoint_conflict（「チェックポイントは最後に読み取られてから更新されています」）で失敗します。これは実際の同時書き込みが発生した場合と同じエラーです。clearCheckpoint() は楽観ロックを使用しており、削除対象が0行の場合も同じように拒否されるためです。クリアする前に、最終ページを含むすべてのページで必ずチェックポイントを保存してください。
• 保存するだけのチェックポイントを追加しないでください。次回の実行では、プロバイダーへのリクエストパラメーターまたはページネーション／再開状態でチェックポイントを使用する必要があります。
• チェックポイントのオブジェクトはフラットな形式にしてください。値として使用できるのは文字列、数値、ブール値のみです。日付は ISO 文字列として保存し、Date オブジェクト、配列、ネストされたオブジェクトは保存しないでください。
• デプロイ後の初回実行では、まだチェックポイントがないため、初回同期として動作し、後続の増分実行より時間がかかることがあります。該当する場合は顧客に伝えてください。
• 変更分のみを取得する増分同期では、trackDeletesStart()／trackDeletesEnd() を使用しないでください。これらのエンドポイントは変更のない行を返さないため、Nango が変更のないレコードを削除済みとして扱ってしまいます。

棚卸し

既存のパターンや不完全なチェックポイントパターンを検索します。

rg -n "lastSyncDate|syncType:\s*['\"]incremental|deleteRecordsFromPreviousExecutions|trackDeletes:|track_deletes|getCheckpoint|saveCheckpoint|clearCheckpoint" .

同期ごとに、以下を確認してください。

• 既存の下限または再開マーカー：updated_at、modified_since、変更レコード用エンドポイントのカーソル、ページトークン、オフセット／ページ番号、since_id など。
• プロバイダーのリクエストが変更された行のみを返すのか、データセット全体を取得するのか。
• ページネーションの形式と、各ページの後に進捗を保存できるか。
• batchSave、batchUpdate、batchDelete に渡されるモデル名。
• 削除の方式：削除済みレコード用の明示的なエンドポイント／トゥームストーン／Webhook、全件更新時の差分検出、または安全な削除シグナルなし。

チェックポイントを選択する

• タイムスタンプの下限：z.object({ updated_after: z.string() })。
• カーソルまたは変更フィード：z.object({ cursor: z.string() })。
• ページ番号／オフセットからの再開：z.object({ page: z.number() }) または z.object({ offset: z.number() })。
• 複合的な再開状態：フラットなプリミティブ値を組み合わせます。例：z.object({ updated_after: z.string(), page_token: z.string().optional() })。
• 全件更新の耐障害性：実行中にカーソル／ページのチェックポイントを保存し、データセット全体の取得と保存が正常に完了した後にのみ clearCheckpoint() を呼び出します。

プロバイダーが変更を絞り込めず、ページネーションも再開できない場合は、無理にチェックポイントを導入しないでください。プロバイダー側の制約を明記し、明確な理由がある全件更新を維持するか、全件更新に切り替えてください。

移行手順

1. 同期の近くに名前付きチェックポイントスキーマを追加し、createSync() に渡します。

const CheckpointSchema = z.object({
  updated_after: z.string(),
});

export default createSync({
  description: "Sync contacts",
  frequency: "every hour",
  checkpoint: CheckpointSchema,
  models: { Contact },
  exec: async (nango) => {
    // ...
  },
});

2. 増分移行から syncType: 'incremental' を削除します。checkpoint フィールドがこの従来のシグナルに置き換わります。

3. nango.lastSyncDate の読み取りを await nango.getCheckpoint() に置き換え、プロバイダーへのリクエストでチェックポイントを使用します。

const checkpoint = await nango.getCheckpoint();

const response = await nango.get({
  endpoint: "/contacts",
  params: {
    ...(checkpoint?.updated_after && {
      updated_after: checkpoint.updated_after,
    }),
  },
  retries: 3,
});

4. すべての batchSave() の後、同じループの反復内で saveCheckpoint() を呼び出します。nango.paginate 内や、抽出したページネーションヘルパー内も含みます。両者を対にし、チェックポイントの書き込みを exec の最後まで遅らせないでください。

const contacts = response.data.items.map(mapContact);
await nango.batchSave(contacts, "Contact");

if (contacts.length > 0) {
  const lastContact = contacts[contacts.length - 1]!;
  await nango.saveCheckpoint({ updated_after: lastContact.updated_at });
}

ページネーションを伴う同期：

for await (const page of nango.paginate<Item>(config)) {
  const records = page.map(mapRecord);
  await nango.batchSave(records, "Contact");
  await nango.saveCheckpoint({ updated_after: latestTimestamp(records) });
}

無効な例：ループ内で batchSave だけを呼び出す await getAndSaveUsers(nango) を実行し、そのヘルパーの戻り後、exec 内で saveCheckpoint を一度だけ呼び出すこと。

5. タイムスタンプチェックポイントには、プロバイダーのレコードを最終更新値でソートしたものを優先して使用します。レコードのタイムスタンプが同じ場合がある、または API が安定したソートに対応していない場合は、プロバイダーのカーソル／ページトークンや last_id のようなタイブレーカー項目を含む複合チェックポイントを使用します。

6. 最後にチェックポイントを一度だけ保存するために、すべてのレコードをメモリに蓄積しないでください。ページごとに処理し、batchSave と saveCheckpoint を実行します。

チェックポイントを使用するフルリフレッシュ

API が変更された行を返せないものの、ページネーションを再開できる場合にのみ使用します。チェックポイントは障害復旧用であり、増分フィルタリング用ではありません。また、任意ではなく必須です。Nango の同期は時間制限のある実行ウィンドウ内で動作します。チェックポイントがないと、そのウィンドウ内に完了しなかったフルリフレッシュは次回の実行で1ページ目から再開され、毎回同じ序盤のページを再取得することになり、データセットの残りに到達できません。

• 取得を始める前に、保存済みのカーソル／ページを読み取ります。
• リフレッシュの実行ごとに trackDeletesStart() を呼び出します。繰り返し呼び出しても安全です。以前の実行で同じ論理リフレッシュの削除追跡ウィンドウがすでに開始されている場合、その開始時点が上書きされることはありません。
• チェックポイントが存在する場合は、保存済みのカーソル／ページからページネーションを開始します。
• batchSave() が成功するたびに、次のカーソル／ページを指定して saveCheckpoint() を呼び出します（同じループ本体内で呼び出し、exec の最後だけで呼び出すことはしないでください）。次のページが残っている場合だけでなく、最後のページでも呼び出します。
• 最後のページを保存した後にのみ clearCheckpoint() を呼び出します。処理済みの各ページ（最後のページを含む）を保存すれば、通常のページ処理経路でクリア対象のチェックポイント行が必ず存在します。別の経路でチェックポイントが一度も作成されない場合（たとえば、ページを一つも処理しない場合）、その経路では clearCheckpoint() を呼び出さないでください。実行時に checkpoint_conflict が発生します。これは最後のページを保存する代わりにはなりません。
• trackDeletesEnd() はその clearCheckpoint() の後にのみ呼び出し、データセット全体の処理が実際に完了した実行で一度だけ実行されるようにします。
• 次回のスケジュール実行では、チェックポイントがクリアされているため、同期は再び最初から開始されます。

削除の処理

• プロバイダーが削除済みレコード、トゥームストーン、または削除 webhook を提供する場合は、同じチェックポイントウィンドウを使って、それらの ID に対して batchDelete() を呼び出します。
• データセット全体との比較が削除検出に必要な場合は、フルリフレッシュを使用します。取得／保存の前に trackDeletesStart('Model') を呼び出し、データセット全体の保存とチェックポイントのクリアが完了した後にのみ trackDeletesEnd('Model') を呼び出します。
• 複数の実行にまたがるチェックポイント付きフルリフレッシュでは、各実行の開始時に trackDeletesStart() を呼び出せます。これは安全で冪等であり、すでに開いているウィンドウの開始時点を上書きしません。trackDeletesEnd() は、フルリフレッシュを完了してチェックポイントをクリアする実行でのみ呼び出します。
• deleteRecordsFromPreviousExecutions() を削除します。これはチェックポイント付き同期と互換性がありません。

テスト

Nango プロジェクトのルートから、ローカルの検証ループを実行します。

nango compile
nango dryrun <sync-name> <connection-id> --validate -e dev --no-interactive --auto-confirm
nango dryrun <sync-name> <connection-id> --validate -e dev --no-interactive --auto-confirm --checkpoint '{"updated_after":"2024-06-01T00:00:00Z"}'

同期でメタデータが必要な場合は --metadata を使用し、スキーマに合わせて --checkpoint のペイロードを調整します。リポジトリに既存のテストスイートがある場合は、それも実行します。

最終チェックリスト

• 移行した同期に nango.lastSyncDate の参照が残っていない
• チェックポイントを使用する増分同期に syncType: 'incremental' が残っていない
• checkpoint スキーマが存在し、フラットなプリミティブ値のみを使用している
• プロバイダーへのリクエスト前に getCheckpoint() が呼び出されている
• 保存したチェックポイントによって、次のプロバイダーリクエストまたは再開時の状態が変わる
• 進捗を進めるすべての batchSave()／batchUpdate()／batchDelete() の直後に、同じループ内で saveCheckpoint() が呼び出されている（nango.paginate やページネーションヘルパー内も含みます。exec の戻り後に一度だけ呼び出すのではありません）
• 全件更新同期にチェックポイントスキーマがあり、そこからページネーションを再開し、各ページの後（最後のページも含む）にチェックポイントを保存し、最後のページが正常に完了した後にのみ clearCheckpoint() を呼び出している
• 全件更新の saveCheckpoint() 呼び出しが、次のページがあるかどうかによって条件付けられていない
• clearCheckpoint() を呼び出すすべての経路でチェックポイントを作成または再開している。チェックポイントを作成しない経路では呼び出していない
• 全件更新の trackDeletesEnd() は clearCheckpoint() の後にのみ実行され、最後のページを完了する実行内で呼び出されている
• 削除処理が同期の種類と一致している。プロバイダーが明示的に削除を通知する場合は batchDelete() を使用し、trackDeletesStart/End は全件更新の場合にのみ使用している
• 現実的な --checkpoint を指定してドライランをテストした

参考ドキュメント

• チェックポイント：https://nango.dev/docs/implementation-guides/use-cases/syncs/checkpoints
• 移行ガイド：https://nango.dev/docs/implementation-guides/platform/migrations/migrate-to-checkpoints
• 削除の検出：https://nango.dev/docs/implementation-guides/use-cases/syncs/deletion-detection
• Functions SDK リファレンス：https://nango.dev/docs/reference/functions#checkpoints