applyServerDelta method

Set applyServerDelta(
  1. BsatnRowList deletes,
  2. BsatnRowList inserts,
  3. EventContext context, {
  4. required int querySetId,
  5. Set? protectedKeys,
  6. bool reconcile = false,
  7. bool trackImbalance = true,
  8. void onProtectedKeyCommitted(
    1. dynamic primaryKey,
    2. Map<String, dynamic>? committedRowJson
    )?,
})

Implementation

Set<dynamic> applyServerDelta(
  BsatnRowList deletes,
  BsatnRowList inserts,
  EventContext context, {
  required int querySetId,
  Set<dynamic>? protectedKeys,
  bool reconcile = false,
  bool trackImbalance = true,
  void Function(dynamic primaryKey, Map<String, dynamic>? committedRowJson)?
  onProtectedKeyCommitted,
}) {
  final hadOwnersBefore = <dynamic>{
    for (final pk in inserts.getRows().map(
      (bytes) => decoder.getPrimaryKey(decoder.decode(BsatnDecoder(bytes))),
    ))
      if (pk != null && _rowOwners[pk]?.isNotEmpty == true) pk,
  };

  final changes = _applyChanges(
    deletes,
    inserts,
    protectedKeys: protectedKeys,
    reconcile: reconcile,
    onProtectedKeyCommitted: onProtectedKeyCommitted,
  );

  for (final pk in changes.serverPresentKeys) {
    _rowOwners.putIfAbsent(pk, () => <int>{}).add(querySetId);
  }

  final suppressInsertEvents = <dynamic>{};
  for (final pk in changes.serverPresentKeys) {
    if (hadOwnersBefore.contains(pk)) suppressInsertEvents.add(pk);
  }
  if (suppressInsertEvents.isNotEmpty) {
    changes.inserted.removeWhere((row) {
      final pk = decoder.getPrimaryKey(row);
      return pk != null && suppressInsertEvents.contains(pk);
    });
  }

  final deletedRowValues = <dynamic, T>{
    for (final row in changes.deleted)
      if (decoder.getPrimaryKey(row) != null) decoder.getPrimaryKey(row): row,
  };

  final toEvict = <dynamic>[];
  final surviving = <dynamic>[];
  for (final pk in changes.serverAbsentKeys) {
    final owners = _rowOwners[pk];
    if (owners == null) {
      if (trackImbalance && deletedRowValues.containsKey(pk)) {
        _ownershipImbalanceCount++;
        SdkLogger.w(
          'OWNERSHIP_IMBALANCE[$tableName]: querySetId=$querySetId '
          'reported pk=$pk absent but it has no owner entry',
        );
      }
      continue;
    }
    if (!owners.remove(querySetId) && trackImbalance) {
      _ownershipImbalanceCount++;
      SdkLogger.w(
        'OWNERSHIP_IMBALANCE[$tableName]: querySetId=$querySetId '
        'reported pk=$pk absent but was not among its owners $owners',
      );
    }
    if (owners.isEmpty) {
      _rowOwners.remove(pk);
      toEvict.add(pk);
    } else {
      surviving.add(pk);
    }
  }

  for (final pk in toEvict) {
    if (protectedKeys != null && protectedKeys.contains(pk)) continue;
    _evictRow(pk);
  }

  final suppressDeleteEvents = <dynamic>{};
  for (final pk in surviving) {
    final row = deletedRowValues[pk];
    if (row != null) {
      _rowsByPrimaryKey[pk] = row;
      suppressDeleteEvents.add(pk);
    }
  }
  if (suppressDeleteEvents.isNotEmpty) {
    changes.deleted.removeWhere((row) {
      final pk = decoder.getPrimaryKey(row);
      return pk != null && suppressDeleteEvents.contains(pk);
    });
  }

  _emitChanges(changes, context);

  if (isEvent) {
    _rowsByPrimaryKey.clear();
    _rows.clear();
  }

  return changes.touchedKeys;
}