Use sealed classes for sync lines

Author: simolus3

packages/powersync_core/lib/src/bucket_storage.dart

@@ -347,105 +347,6 @@ class SyncDataBatch {
   SyncDataBatch(this.buckets);
 }
 
-class SyncBucketData {
-  final String bucket;
-  final List<OplogEntry> data;
-  final bool hasMore;
-  final String? after;
-  final String? nextAfter;
-
-  const SyncBucketData(
-      {required this.bucket,
-      required this.data,
-      this.hasMore = false,
-      this.after,
-      this.nextAfter});
-
-  SyncBucketData.fromJson(Map<String, dynamic> json)
-      : bucket = json['bucket'],
-        hasMore = json['has_more'] ?? false,
-        after = json['after'],
-        nextAfter = json['next_after'],
-        data =
-            (json['data'] as List).map((e) => OplogEntry.fromJson(e)).toList();
-
-  Map<String, dynamic> toJson() {
-    return {
-      'bucket': bucket,
-      'has_more': hasMore,
-      'after': after,
-      'next_after': nextAfter,
-      'data': data
-    };
-  }
-}
-
-class OplogEntry {
-  final String opId;
-
-  final OpType? op;
-
-  /// rowType + rowId uniquely identifies an entry in the local database.
-  final String? rowType;
-  final String? rowId;
-
-  /// Together with rowType and rowId, this uniquely identifies a source entry
-  /// per bucket in the oplog. There may be multiple source entries for a single
-  /// "rowType + rowId" combination.
-  final String? subkey;
-
-  final String? data;
-  final int checksum;
-
-  const OplogEntry(
-      {required this.opId,
-      required this.op,
-      this.subkey,
-      this.rowType,
-      this.rowId,
-      this.data,
-      required this.checksum});
-
-  OplogEntry.fromJson(Map<String, dynamic> json)
-      : opId = json['op_id'],
-        op = OpType.fromJson(json['op']),
-        rowType = json['object_type'],
-        rowId = json['object_id'],
-        checksum = json['checksum'],
-        data = json['data'] is String ? json['data'] : jsonEncode(json['data']),
-        subkey = json['subkey'] is String ? json['subkey'] : null;
-
-  Map<String, dynamic>? get parsedData {
-    return data == null ? null : jsonDecode(data!);
-  }
-
-  /// Key to uniquely represent a source entry in a bucket.
-  /// This is used to supersede old entries.
-  /// Relevant for put and remove ops.
-  String get key {
-    return "$rowType/$rowId/$subkey";
-  }
-
-  Map<String, dynamic> toJson() {
-    return {
-      'op_id': opId,
-      'op': op?.toJson(),
-      'object_type': rowType,
-      'object_id': rowId,
-      'checksum': checksum,
-      'subkey': subkey,
-      'data': data
-    };
-  }
-}
-
-class SqliteOp {
-  String sql;
-  List<dynamic> args;
-
-  SqliteOp(this.sql, this.args);
-}
-
 class SyncLocalDatabaseResult {
   final bool ready;
   final bool checkpointValid;
@@ -483,42 +384,3 @@ class ChecksumCache {
 
   ChecksumCache(this.lastOpId, this.checksums);
 }
-
-enum OpType {
-  clear(1),
-  move(2),
-  put(3),
-  remove(4);
-
-  final int value;
-
-  const OpType(this.value);
-
-  static OpType? fromJson(String json) {
-    switch (json) {
-      case 'CLEAR':
-        return clear;
-      case 'MOVE':
-        return move;
-      case 'PUT':
-        return put;
-      case 'REMOVE':
-        return remove;
-      default:
-        return null;
-    }
-  }
-
-  String toJson() {
-    switch (this) {
-      case clear:
-        return 'CLEAR';
-      case move:
-        return 'MOVE';
-      case put:
-        return 'PUT';
-      case remove:
-        return 'REMOVE';
-    }
-  }
-}

packages/powersync_core/lib/src/streaming_sync.dart

@@ -53,7 +53,7 @@ class StreamingSyncImplementation implements StreamingSync {
 
   late final http.Client _client;
 
-  final StreamController<String?> _localPingController =
+  final StreamController<Null> _localPingController =
       StreamController.broadcast();
 
   final Duration retryDelay;
@@ -340,96 +340,19 @@ class StreamingSyncImplementation implements StreamingSync {
       }
 
       _updateStatus(connected: true, connecting: false);
-      if (line is Checkpoint) {
-        targetCheckpoint = line;
-        final Set<String> bucketsToDelete = {...bucketSet};
-        final Set<String> newBuckets = {};
-        for (final checksum in line.checksums) {
-          newBuckets.add(checksum.bucket);
-          bucketsToDelete.remove(checksum.bucket);
-        }
-        bucketSet = newBuckets;
-        await adapter.removeBuckets([...bucketsToDelete]);
-        _updateStatus(downloading: true);
-      } else if (line is StreamingSyncCheckpointComplete) {
-        final result = await adapter.syncLocalDatabase(targetCheckpoint!);
-        if (!result.checkpointValid) {
-          // This means checksums failed. Start again with a new checkpoint.
-          // TODO: better back-off
-          // await new Promise((resolve) => setTimeout(resolve, 50));
-          return false;
-        } else if (!result.ready) {
-          // Checksums valid, but need more data for a consistent checkpoint.
-          // Continue waiting.
-        } else {
-          appliedCheckpoint = targetCheckpoint;
-
-          _updateStatus(
-              downloading: false,
-              downloadError: _noError,
-              lastSyncedAt: DateTime.now());
-        }
-
-        validatedCheckpoint = targetCheckpoint;
-      } else if (line is StreamingSyncCheckpointDiff) {
-        // TODO: It may be faster to just keep track of the diff, instead of the entire checkpoint
-        if (targetCheckpoint == null) {
-          throw PowerSyncProtocolException(
-              'Checkpoint diff without previous checkpoint');
-        }
-        _updateStatus(downloading: true);
-        final diff = line;
-        final Map<String, BucketChecksum> newBuckets = {};
-        for (var checksum in targetCheckpoint.checksums) {
-          newBuckets[checksum.bucket] = checksum;
-        }
-        for (var checksum in diff.updatedBuckets) {
-          newBuckets[checksum.bucket] = checksum;
-        }
-        for (var bucket in diff.removedBuckets) {
-          newBuckets.remove(bucket);
-        }
-
-        final newCheckpoint = Checkpoint(
-            lastOpId: diff.lastOpId,
-            checksums: [...newBuckets.values],
-            writeCheckpoint: diff.writeCheckpoint);
-        targetCheckpoint = newCheckpoint;
-
-        bucketSet = Set.from(newBuckets.keys);
-        await adapter.removeBuckets(diff.removedBuckets);
-        adapter.setTargetCheckpoint(targetCheckpoint);
-      } else if (line is SyncBucketData) {
-        _updateStatus(downloading: true);
-        await adapter.saveSyncData(SyncDataBatch([line]));
-      } else if (line is StreamingSyncKeepalive) {
-        if (line.tokenExpiresIn == 0) {
-          // Token expired already - stop the connection immediately
-          invalidCredentialsCallback?.call().ignore();
-          break;
-        } else if (line.tokenExpiresIn <= 30) {
-          // Token expires soon - refresh it in the background
-          if (credentialsInvalidation == null &&
-              invalidCredentialsCallback != null) {
-            credentialsInvalidation = invalidCredentialsCallback!().then((_) {
-              // Token has been refreshed - we should restart the connection.
-              haveInvalidated = true;
-              // trigger next loop iteration ASAP, don't wait for another
-              // message from the server.
-              _localPingController.add(null);
-            }, onError: (_) {
-              // Token refresh failed - retry on next keepalive.
-              credentialsInvalidation = null;
-            });
+      switch (line) {
+        case Checkpoint():
+          targetCheckpoint = line;
+          final Set<String> bucketsToDelete = {...bucketSet};
+          final Set<String> newBuckets = {};
+          for (final checksum in line.checksums) {
+            newBuckets.add(checksum.bucket);
+            bucketsToDelete.remove(checksum.bucket);
           }
-        }
-      } else {
-        if (targetCheckpoint == appliedCheckpoint) {
-          _updateStatus(
-              downloading: false,
-              downloadError: _noError,
-              lastSyncedAt: DateTime.now());
-        } else if (validatedCheckpoint == targetCheckpoint) {
+          bucketSet = newBuckets;
+          await adapter.removeBuckets([...bucketsToDelete]);
+          _updateStatus(downloading: true);
+        case StreamingSyncCheckpointComplete():
           final result = await adapter.syncLocalDatabase(targetCheckpoint!);
           if (!result.checkpointValid) {
             // This means checksums failed. Start again with a new checkpoint.
@@ -447,7 +370,85 @@ class StreamingSyncImplementation implements StreamingSync {
                 downloadError: _noError,
                 lastSyncedAt: DateTime.now());
           }
-        }
+
+          validatedCheckpoint = targetCheckpoint;
+        case StreamingSyncCheckpointDiff():
+          // TODO: It may be faster to just keep track of the diff, instead of the entire checkpoint
+          if (targetCheckpoint == null) {
+            throw PowerSyncProtocolException(
+                'Checkpoint diff without previous checkpoint');
+          }
+          _updateStatus(downloading: true);
+          final diff = line;
+          final Map<String, BucketChecksum> newBuckets = {};
+          for (var checksum in targetCheckpoint.checksums) {
+            newBuckets[checksum.bucket] = checksum;
+          }
+          for (var checksum in diff.updatedBuckets) {
+            newBuckets[checksum.bucket] = checksum;
+          }
+          for (var bucket in diff.removedBuckets) {
+            newBuckets.remove(bucket);
+          }
+
+          final newCheckpoint = Checkpoint(
+              lastOpId: diff.lastOpId,
+              checksums: [...newBuckets.values],
+              writeCheckpoint: diff.writeCheckpoint);
+          targetCheckpoint = newCheckpoint;
+
+          bucketSet = Set.from(newBuckets.keys);
+          await adapter.removeBuckets(diff.removedBuckets);
+          adapter.setTargetCheckpoint(targetCheckpoint);
+        case SyncBucketData():
+          _updateStatus(downloading: true);
+          await adapter.saveSyncData(SyncDataBatch([line]));
+        case StreamingSyncKeepalive():
+          if (line.tokenExpiresIn == 0) {
+            // Token expired already - stop the connection immediately
+            invalidCredentialsCallback?.call().ignore();
+            break;
+          } else if (line.tokenExpiresIn <= 30) {
+            // Token expires soon - refresh it in the background
+            if (credentialsInvalidation == null &&
+                invalidCredentialsCallback != null) {
+              credentialsInvalidation = invalidCredentialsCallback!().then((_) {
+                // Token has been refreshed - we should restart the connection.
+                haveInvalidated = true;
+                // trigger next loop iteration ASAP, don't wait for another
+                // message from the server.
+                _localPingController.add(null);
+              }, onError: (_) {
+                // Token refresh failed - retry on next keepalive.
+                credentialsInvalidation = null;
+              });
+            }
+          }
+        case null: // Local ping
+          if (targetCheckpoint == appliedCheckpoint) {
+            _updateStatus(
+                downloading: false,
+                downloadError: _noError,
+                lastSyncedAt: DateTime.now());
+          } else if (validatedCheckpoint == targetCheckpoint) {
+            final result = await adapter.syncLocalDatabase(targetCheckpoint!);
+            if (!result.checkpointValid) {
+              // This means checksums failed. Start again with a new checkpoint.
+              // TODO: better back-off
+              // await new Promise((resolve) => setTimeout(resolve, 50));
+              return false;
+            } else if (!result.ready) {
+              // Checksums valid, but need more data for a consistent checkpoint.
+              // Continue waiting.
+            } else {
+              appliedCheckpoint = targetCheckpoint;
+
+              _updateStatus(
+                  downloading: false,
+                  downloadError: _noError,
+                  lastSyncedAt: DateTime.now());
+            }
+          }
       }
 
       if (haveInvalidated) {
@@ -458,7 +459,8 @@ class StreamingSyncImplementation implements StreamingSync {
     return true;
   }
 
-  Stream<Object?> streamingSyncRequest(StreamingSyncRequest data) async* {
+  Stream<StreamingSyncLine?> streamingSyncRequest(
+      StreamingSyncRequest data) async* {
     final credentials = await credentialsCallback();
     if (credentials == null) {
       throw CredentialsException('Not logged in');
@@ -498,7 +500,7 @@ class StreamingSyncImplementation implements StreamingSync {
       if (aborted) {
         break;
       }
-      yield parseStreamingSyncLine(line as Map<String, dynamic>);
+      yield StreamingSyncLine.fromJson(line as Map<String, dynamic>);
     }
   }