applyChangeset method

Implementation

Future<int> applyChangeset(
  Uint8List bytes, {
  ChangesetConflictHandler? onConflict,
}) async {
  final changes = Session.decode(bytes);
  final handler = onConflict ?? (_, __) => ConflictResolution.skip;
  var applied = 0;
  return _lock.write(() async {
    for (final c in changes) {
      final t = _tables[c.table] ??
          _tables[c.table.toLowerCase()] ??
          _tables[c.table.toUpperCase()];
      if (t == null) {
        final res = handler(c, ConflictKind.notFound);
        if (res == ConflictResolution.abort) return applied;
        continue;
      }
      // Map recorded column order -> current column order.
      int? colIdxIn(List<String> cols, String name) {
        for (var i = 0; i < cols.length; i++) {
          if (cols[i].toLowerCase() == name.toLowerCase()) return i;
        }
        return null;
      }

      List<Object?> projectToCurrent(List<Object?> recorded) {
        final out = List<Object?>.filled(t.columns.length, null);
        for (var i = 0; i < t.columns.length; i++) {
          final src = colIdxIn(c.columns, t.columns[i].name);
          if (src != null && src < recorded.length) out[i] = recorded[src];
        }
        return out;
      }

      int? findRow(List<Object?> recorded) {
        if (c.pkColumns.isEmpty) {
          // Fall back to whole-row equality on the recorded columns.
          for (var ri = 0; ri < t.rows.length; ri++) {
            var match = true;
            for (var k = 0; k < c.columns.length; k++) {
              final ti = colIdxIn([
                for (final col in t.columns) col.name,
              ], c.columns[k]);
              if (ti == null) {
                match = false;
                break;
              }
              if (t.rows[ri][ti] != recorded[k]) {
                match = false;
                break;
              }
            }
            if (match) return ri;
          }
          return null;
        }
        // Locate by primary-key columns.
        final pkIdxRecorded = [
          for (final p in c.pkColumns) colIdxIn(c.columns, p),
        ];
        if (pkIdxRecorded.contains(null)) return null;
        final pkIdxTable = [for (final p in c.pkColumns) t.columnIndex(p)];
        for (var ri = 0; ri < t.rows.length; ri++) {
          var match = true;
          for (var k = 0; k < c.pkColumns.length; k++) {
            if (t.rows[ri][pkIdxTable[k]] != recorded[pkIdxRecorded[k]!]) {
              match = false;
              break;
            }
          }
          if (match) return ri;
        }
        return null;
      }

      switch (c.op) {
        case 'INSERT':
          final candidate = projectToCurrent(c.newValues!);
          if (findRow(c.newValues!) != null) {
            final res = handler(c, ConflictKind.notUnique);
            if (res == ConflictResolution.abort) return applied;
            if (res == ConflictResolution.replace) {
              final ri = findRow(c.newValues!)!;
              t.rows[ri] = candidate;
              _rebuildIndexes(t);
              applied++;
            }
            continue;
          }
          t.insertRow(candidate);
          _rebuildIndexes(t);
          applied++;
          break;
        case 'DELETE':
          final ri = findRow(c.oldValues!);
          if (ri == null) {
            final res = handler(c, ConflictKind.notFound);
            if (res == ConflictResolution.abort) return applied;
            continue;
          }
          t.rows.removeAt(ri);
          _rebuildIndexes(t);
          applied++;
          break;
        case 'UPDATE':
          final ri = findRow(c.oldValues!);
          if (ri == null) {
            final res = handler(c, ConflictKind.notFound);
            if (res == ConflictResolution.abort) return applied;
            continue;
          }
          t.rows[ri] = projectToCurrent(c.newValues!);
          _rebuildIndexes(t);
          applied++;
          break;
        default:
          // unknown op — treat as conflict
          final res = handler(c, ConflictKind.data);
          if (res == ConflictResolution.abort) return applied;
      }
    }
    await _persist();
    return applied;
  });
}