feat: Sync skip deletes <do not merge>

packages/at_persistence_secondary_server/lib/src/log/commitlog/at_commit_log.dart

@@ -127,12 +127,13 @@ class AtCommitLog extends BaseAtCommitLog {
   /// Returns the Iterator of [_commitLogCacheMap] from the commitId specified.
   @server
   Iterator<MapEntry<String, CommitEntry>> getEntries(int commitId,
-      {String? regex, int limit = 25}) {
+      {String? regex, int limit = 25, bool skipDeletes = false}) {
     // If regex is null or isEmpty set regex to match all keys
     if (regex == null || regex.isEmpty) {
       regex = '.*';
     }
-    return _commitLogKeyStore.getEntries(commitId, regex: regex, limit: limit);
+    return _commitLogKeyStore.getEntries(commitId,
+        regex: regex, limit: limit, skipDeletes: skipDeletes);
   }
 
   Future<void> _publishChangeEvent(CommitEntry commitEntry) async {

packages/at_persistence_secondary_server/lib/src/log/commitlog/commit_log_keystore.dart

@@ -1,6 +1,9 @@
 import 'package:at_commons/at_commons.dart';
 import 'package:at_persistence_secondary_server/at_persistence_secondary_server.dart';
 import 'package:at_persistence_secondary_server/src/keystore/hive_base.dart';
+import 'package:at_persistence_secondary_server/src/log/commitlog/sync/fetch_all_keys_strategy.dart';
+import 'package:at_persistence_secondary_server/src/log/commitlog/sync/skip_deletes_strategy.dart';
+import 'package:at_persistence_secondary_server/src/log/commitlog/sync/sync_keys_fetch_strategy.dart';
 import 'package:at_utils/at_utils.dart';
 import 'package:hive/hive.dart';
 import 'package:meta/meta.dart';
@@ -10,10 +13,13 @@ class CommitLogKeyStore extends BaseCommitLogKeyStore {
   final _logger = AtSignLogger('CommitLogKeyStore');
   late CommitLogCache commitLogCache;
 
+  late SyncKeysFetchStrategy _syncKeysFetchStrategy;
+
   int get latestCommitId => commitLogCache.latestCommitId;
 
   CommitLogKeyStore(String currentAtSign) : super(currentAtSign) {
     commitLogCache = CommitLogCache(this);
+    _syncKeysFetchStrategy = FetchAllKeysStrategy();
   }
 
   @override
@@ -184,51 +190,8 @@ class CommitLogKeyStore extends BaseCommitLogKeyStore {
   /// if regex is passed, key has to match the regex or it has to be a special key.
   bool _shouldIncludeKeyInSyncResponse(String atKey, String regex,
       {List<String>? enrolledNamespace}) {
-    return _isNamespaceAuthorised(atKey, enrolledNamespace) &&
-        (_keyMatchesRegex(atKey, regex) || _alwaysIncludeInSync(atKey));
-  }
-
-  bool _isNamespaceAuthorised(
-      String atKeyAsString, List<String>? enrolledNamespace) {
-    // This is work-around for : https://github.com/atsign-foundation/at_server/issues/1570
-    if (atKeyAsString.toLowerCase() == 'configkey') {
-      return true;
-    }
-    late AtKey atKey;
-    try {
-      atKey = AtKey.fromString(atKeyAsString);
-    } on InvalidSyntaxException catch (_) {
-      _logger.warning(
-          '_isNamespaceAuthorized found an invalid key "$atKeyAsString" in the commit log. Returning false');
-      return false;
-    }
-    String? keyNamespace = atKey.namespace;
-    // If enrolledNamespace is null or keyNamespace is null, fallback to
-    // existing behaviour - the key is authorized for the client to receive. So return true.
-    if (enrolledNamespace == null ||
-        enrolledNamespace.isEmpty ||
-        (keyNamespace == null || keyNamespace.isEmpty)) {
-      return true;
-    }
-    if (enrolledNamespace.contains('*') ||
-        enrolledNamespace.contains(keyNamespace)) {
-      return true;
-    }
-    return false;
-  }
-
-  bool _keyMatchesRegex(String atKey, String regex) {
-    return RegExp(regex).hasMatch(atKey);
-  }
-
-  /// match keys which have to included in sync irrespective of whether regex matches
-  /// e.g @bob:shared_key@alice, shared_key.bob@alice, public:publickey@alice,
-  /// public:phone@alice (public key without namespace)
-  bool _alwaysIncludeInSync(String atKey) {
-    return (atKey.contains(AtConstants.atEncryptionSharedKey) &&
-            RegexUtil.keyType(atKey, false) == KeyType.reservedKey) ||
-        atKey.startsWith(AtConstants.atEncryptionPublicKey) ||
-        (atKey.startsWith('public:') && !atKey.contains('.'));
+    return _syncKeysFetchStrategy.shouldIncludeKeyInSyncResponse(atKey, regex,
+        enrolledNamespace: enrolledNamespace);
   }
 
   /// Returns the latest commitEntry of the key.
@@ -238,13 +201,15 @@ class CommitLogKeyStore extends BaseCommitLogKeyStore {
 
   /// Returns the Iterator of entries as Key value pairs after the given the [commitId] for the keys that matches the [regex]
   Iterator<MapEntry<String, CommitEntry>> getEntries(int commitId,
-      {String regex = '.*', int limit = 25}) {
+      {String regex = '.*', int limit = 25, bool skipDeletes = false}) {
+    SyncKeysFetchStrategy syncKeysFetchStrategy =
+        skipDeletes ? SkipDeleteStrategy() : _syncKeysFetchStrategy;
     Iterable<MapEntry<String, CommitEntry>> commitEntriesIterable =
         commitLogCache
             .entriesList()
             .where((element) =>
-                element.value.commitId! >= commitId &&
-                _shouldIncludeKeyInSyncResponse(element.value.atKey!, regex))
+                syncKeysFetchStrategy.shouldIncludeEntryInSyncResponse(
+                    element.value, commitId, regex))
             .take(limit);
     return commitEntriesIterable.iterator;
   }

packages/at_persistence_secondary_server/lib/src/log/commitlog/sync/fetch_all_keys_strategy.dart

@@ -0,0 +1,13 @@
+import 'package:at_persistence_secondary_server/src/log/commitlog/commit_entry.dart';
+import 'package:at_persistence_secondary_server/src/log/commitlog/sync/sync_keys_fetch_strategy.dart';
+
+class FetchAllKeysStrategy extends SyncKeysFetchStrategy {
+  @override
+  bool shouldIncludeEntryInSyncResponse(
+      CommitEntry commitEntry, int commitId, String regex,
+      {List<String>? enrolledNamespace}) {
+    return commitEntry.commitId! >= commitId &&
+        super.shouldIncludeKeyInSyncResponse(commitEntry.atKey!, regex,
+            enrolledNamespace: enrolledNamespace);
+  }
+}

packages/at_persistence_secondary_server/lib/src/log/commitlog/sync/skip_deletes_strategy.dart

@@ -0,0 +1,14 @@
+import 'package:at_persistence_secondary_server/src/log/commitlog/commit_entry.dart';
+import 'package:at_persistence_secondary_server/src/log/commitlog/sync/sync_keys_fetch_strategy.dart';
+
+class SkipDeleteStrategy extends SyncKeysFetchStrategy {
+  @override
+  bool shouldIncludeEntryInSyncResponse(
+      CommitEntry commitEntry, int commitId, String regex,
+      {List<String>? enrolledNamespace}) {
+    return commitEntry.commitId! >= commitId &&
+        super.shouldIncludeKeyInSyncResponse(commitEntry.atKey!, regex,
+            enrolledNamespace: enrolledNamespace) &&
+        commitEntry.operation != CommitOp.DELETE;
+  }
+}

packages/at_persistence_secondary_server/lib/src/log/commitlog/sync/sync_keys_fetch_strategy.dart

@@ -0,0 +1,59 @@
+import 'package:at_commons/at_commons.dart';
+import 'package:at_persistence_secondary_server/at_persistence_secondary_server.dart';
+import 'package:at_utils/at_utils.dart';
+
+abstract class SyncKeysFetchStrategy {
+  final _logger = AtSignLogger('SyncKeysFetchStrategy');
+  bool shouldIncludeEntryInSyncResponse(
+      CommitEntry commitEntry, int commitId, String regex,
+      {List<String>? enrolledNamespace});
+
+  bool shouldIncludeKeyInSyncResponse(String atKey, String regex,
+      {List<String>? enrolledNamespace}) {
+    return isNamespaceAuthorised(atKey, enrolledNamespace) &&
+        (keyMatchesRegex(atKey, regex) || alwaysIncludeInSync(atKey));
+  }
+
+  bool isNamespaceAuthorised(
+      String atKeyAsString, List<String>? enrolledNamespace) {
+    // This is work-around for : https://github.com/atsign-foundation/at_server/issues/1570
+    if (atKeyAsString.toLowerCase() == 'configkey') {
+      return true;
+    }
+    late AtKey atKey;
+    try {
+      atKey = AtKey.fromString(atKeyAsString);
+    } on InvalidSyntaxException catch (_) {
+      _logger.warning(
+          '_isNamespaceAuthorized found an invalid key "$atKeyAsString" in the commit log. Returning false');
+      return false;
+    }
+    String? keyNamespace = atKey.namespace;
+    // If enrolledNamespace is null or keyNamespace is null, fallback to
+    // existing behaviour - the key is authorized for the client to receive. So return true.
+    if (enrolledNamespace == null ||
+        enrolledNamespace.isEmpty ||
+        (keyNamespace == null || keyNamespace.isEmpty)) {
+      return true;
+    }
+    if (enrolledNamespace.contains('*') ||
+        enrolledNamespace.contains(keyNamespace)) {
+      return true;
+    }
+    return false;
+  }
+
+  bool keyMatchesRegex(String atKey, String regex) {
+    return RegExp(regex).hasMatch(atKey);
+  }
+
+  /// match keys which have to included in sync irrespective of whether regex matches
+  /// e.g @bob:shared_key@alice, shared_key.bob@alice, public:publickey@alice,
+  /// public:phone@alice (public key without namespace)
+  bool alwaysIncludeInSync(String atKey) {
+    return (atKey.contains(AtConstants.atEncryptionSharedKey) &&
+            RegexUtil.keyType(atKey, false) == KeyType.reservedKey) ||
+        atKey.startsWith(AtConstants.atEncryptionPublicKey) ||
+        (atKey.startsWith('public:') && !atKey.contains('.'));
+  }
+}

packages/at_persistence_secondary_server/test/commit_log_test.dart

@@ -840,6 +840,33 @@ void main() async {
         expect(commitEntriesMap.containsKey('public:phone.wavi@alice'), false);
         expect(commitEntriesMap.containsKey('public:location@alice'), true);
       });
+      test(
+          'A test to verify delete commit entries are NOT returned when skipDeletes is true',
+          () async {
+        var commitLogInstance =
+            await (AtCommitLogManagerImpl.getInstance().getCommitLog('@alice'));
+        var commitLogKeystore = commitLogInstance!.commitLogKeyStore;
+        await commitLogKeystore.add(CommitEntry(
+            'test_key_true_1@alice', CommitOp.UPDATE, DateTime.now()));
+        await commitLogKeystore.add(CommitEntry(
+            'test_key_true_2@alice', CommitOp.DELETE, DateTime.now()));
+        await commitLogKeystore.add(CommitEntry(
+            'test_key_true_3@alice', CommitOp.DELETE, DateTime.now()));
+        await commitLogKeystore.add(CommitEntry(
+            'test_key_true_4@alice', CommitOp.UPDATE, DateTime.now()));
+        Iterator<MapEntry<String, CommitEntry>>? changes = commitLogInstance
+            .commitLogKeyStore
+            .getEntries(-1, skipDeletes: true);
+        Map<String?, CommitEntry> commitEntriesMap = {};
+        while (changes.moveNext()) {
+          var commitEntry = changes.current.value;
+          commitEntriesMap[commitEntry.atKey] = commitEntry;
+        }
+        expect(commitEntriesMap.containsKey('test_key_true_1@alice'), true);
+        expect(commitEntriesMap.containsKey('test_key_true_2@alice'), false);
+        expect(commitEntriesMap.containsKey('test_key_true_3@alice'), false);
+        expect(commitEntriesMap.containsKey('test_key_true_4@alice'), true);
+      });
     });
     tearDown(() async => await tearDownFunc());
   });

packages/at_secondary_server/lib/src/verb/handler/sync_progressive_verb_handler.dart

@@ -43,7 +43,8 @@ class SyncProgressiveVerbHandler extends AbstractVerbHandler {
     // Get entries to sync
     var commitEntryIterator = atCommitLog!.getEntries(
         int.parse(verbParams[AtConstants.fromCommitSequence]!) + 1,
-        regex: verbParams['regex']);
+        regex: verbParams['regex'],
+        skipDeletes: verbParams['skipDeletes'] == 'true');
 
     List<KeyStoreEntry> syncResponse = [];
     await prepareResponse(capacity, syncResponse, commitEntryIterator,

packages/at_secondary_server/pubspec.yaml

@@ -34,6 +34,15 @@ dependencies:
   yaml: 3.1.2
   logging: 1.2.0
 
+dependency_overrides:
+  at_commons:
+    git:
+      url: https://github.com/atsign-foundation/at_libraries.git
+      path: packages/at_commons
+      ref: sync_skip_delete
+  at_persistence_secondary_server:
+    path: ../at_persistence_secondary_server
+
 dev_dependencies:
   build_runner: ^2.3.3
   test: ^1.24.4

packages/at_secondary_server/test/sync_unit_test.dart

@@ -469,6 +469,39 @@ void main() {
         expect(syncResponse[3]['operation'], '*');
       });
 
+      test(
+          'test to verify delete commit entries are not sent when skipDeletes is true',
+          () async {
+        await secondaryPersistenceStore!
+            .getSecondaryKeyStore()
+            ?.put('test_key_1@alice', AtData()..data = 'alice');
+        await secondaryPersistenceStore!
+            .getSecondaryKeyStore()
+            ?.remove('test_key_1@alice');
+        await secondaryPersistenceStore!
+            .getSecondaryKeyStore()
+            ?.put('test_key_2@alice', AtData()..data = 'alice');
+        await secondaryPersistenceStore!
+            .getSecondaryKeyStore()
+            ?.remove('test_key_2@alice');
+        var syncProgressiveVerbHandler = SyncProgressiveVerbHandler(
+            secondaryPersistenceStore!.getSecondaryKeyStore()!);
+        var response = Response();
+        var inBoundSessionId = '_6665436c-29ff-481b-8dc6-129e89199718';
+        var atConnection = InboundConnectionImpl(mockSocket, inBoundSessionId);
+        atConnection.metaData.isAuthenticated = true;
+        var syncVerbParams = HashMap<String, String>();
+        syncVerbParams.putIfAbsent(AtConstants.fromCommitSequence, () => '-1');
+        syncVerbParams.putIfAbsent('skipDeletes', () => 'true');
+        await syncProgressiveVerbHandler.processVerb(
+            response, syncVerbParams, atConnection);
+        List syncResponse = jsonDecode(response.data!);
+        for (var entry in syncResponse) {
+          expect(entry['atKey'] != 'test_key_1@alice', true);
+          expect(entry['atKey'] != 'test_key_2@alice', true);
+        }
+      });
+
       test(
           'test to verify only entries matching the regex are added to sync response',
           () async {