Function applyChanges

packages/core/src/sync/simple-sync.ts:278–403

(
  payload: DeviceSyncPayload,
  options: { forceApply?: boolean } = {},
)

// ---------------------------------------------------------------------------

export async function applyChanges(
  payload: DeviceSyncPayload,
  options: { forceApply?: boolean } = {},
): Promise<{ applied: number; skipped: number }> {
  return runSerializedDbTask(() =>
    withDatabaseLockRetry(async () => {
      await ensureNoTransaction();
      const db = await getDB();
      if (shouldRunSyncCleanup()) {
        await cleanupOrphanedSyncRows(db);
      }
      let applied = 0;
      let skipped = 0;

      // Keep this transaction-free. On some adapters, explicit BEGIN/COMMIT can
      // lose state across awaited calls and end with "cannot commit - no transaction is active".
      for (const tableInfo of SYNC_TABLES) {
        const tableName = tableInfo.name;
        const tableData = payload.tables[tableName];
        if (!tableData) continue;

        const { pk, timestampCol } = tableInfo;
        const exclude = tableInfo.excludeColumns ?? [];
        console.log(
          `[SimpleSync] Applying table ${tableName}: ${tableData.records.length} record(s), ${tableData.deletedIds.length} deletion(s)`,
        );
        const allIdsToCheck = [
          ...tableData.records.map((record) => record[pk]).filter((value) => value !== undefined),
          ...tableData.deletedIds,
        ];
        const existingRecords = await loadExistingRecordStates(
          db,
          tableName,
          pk,
          timestampCol,
          allIdsToCheck,
        );
        const remoteRecordIds = new Set(
          tableData.records
            .map((record) => record[pk])
            .filter((value) => value !== undefined)
            .map(String),
        );
        let processedRecords = 0;

        for (const record of tableData.records) {
          const pkValue = record[pk];
          const remoteTs = record[timestampCol] as number;

          const safeRecord =
            exclude.length > 0
              ? Object.fromEntries(Object.entries(record).filter(([k]) => !exclude.includes(k)))
              : record;

          const localState = existingRecords.get(String(pkValue));
          if (!options.forceApply && !shouldApplyRemoteRecord(record, timestampCol, localState)) {
            skipped++;
          } else {