Adopt in example

Author: simolus3

demos/supabase-todolist/lib/widgets/todo_list_page.dart

@@ -1,4 +1,5 @@
 import 'package:flutter/material.dart';
+import 'package:powersync/powersync.dart';
 
 import '../powersync.dart';
 import './status_app_bar.dart';
@@ -38,30 +39,82 @@ class TodoListPage extends StatelessWidget {
   }
 }
 
-class TodoListWidget extends StatelessWidget {
+class TodoListWidget extends StatefulWidget {
   final TodoList list;
 
   const TodoListWidget({super.key, required this.list});
 
+  @override
+  State<TodoListWidget> createState() => _TodoListWidgetState();
+}
+
+class _TodoListWidgetState extends State<TodoListWidget> {
+  SyncStreamSubscription? _listSubscription;
+
+  void _subscribe(String listId) {
+    db
+        .syncStream('todos', {'list': listId})
+        .subscribe(ttl: const Duration(hours: 1))
+        .then((sub) {
+          if (mounted && widget.list.id == listId) {
+            setState(() {
+              _listSubscription = sub;
+            });
+          } else {
+            sub.unsubscribe();
+          }
+        });
+  }
+
+  @override
+  void initState() {
+    super.initState();
+    _subscribe(widget.list.id);
+  }
+
+  @override
+  void didUpdateWidget(covariant TodoListWidget oldWidget) {
+    super.didUpdateWidget(oldWidget);
+    _subscribe(widget.list.id);
+  }
+
+  @override
+  void dispose() {
+    super.dispose();
+    _listSubscription?.unsubscribe();
+  }
+
   @override
   Widget build(BuildContext context) {
     return StreamBuilder(
-      stream: TodoList.watchSyncStatus().map((e) => e.hasSynced),
-      initialData: db.currentStatus.hasSynced,
+      stream: db.statusStream,
+      initialData: db.currentStatus,
       builder: (context, snapshot) {
-        return StreamBuilder(
-          stream: list.watchItems(),
-          builder: (context, snapshot) {
-            final items = snapshot.data ?? const [];
-
-            return ListView(
-              padding: const EdgeInsets.symmetric(vertical: 8.0),
-              children: items.map((todo) {
-                return TodoItemWidget(todo: todo);
-              }).toList(),
-            );
-          },
-        );
+        final hasSynced = switch (_listSubscription) {
+              null => null,
+              final sub => snapshot.requireData.statusFor(sub),
+            }
+                ?.subscription
+                .hasSynced ??
+            false;
+
+        if (!hasSynced) {
+          return const CircularProgressIndicator();
+        } else {
+          return StreamBuilder(
+            stream: widget.list.watchItems(),
+            builder: (context, snapshot) {
+              final items = snapshot.data ?? const [];
+
+              return ListView(
+                padding: const EdgeInsets.symmetric(vertical: 8.0),
+                children: items.map((todo) {
+                  return TodoItemWidget(todo: todo);
+                }).toList(),
+              );
+            },
+          );
+        }
       },
     );
   }

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -376,7 +376,7 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
         await shutdown();
       } else if (action == 'changed_subscriptions') {
         openedStreamingSync
-            ?.updateSubscriptions(action[1] as List<SubscribedStream>);
+            ?.updateSubscriptions(message[1] as List<SubscribedStream>);
       }
     }
   });

packages/powersync_core/lib/src/sync/options.dart

@@ -107,13 +107,15 @@ extension type ResolvedSyncOptions(SyncOptions source) {
       crudThrottleTime: other.crudThrottleTime ?? crudThrottleTime,
       retryDelay: other.retryDelay ?? retryDelay,
       params: other.params ?? params,
+      syncImplementation: other.syncImplementation,
       includeDefaultStreams:
           other.includeDefaultStreams ?? includeDefaultStreams,
     );
 
     final didChange = !_mapEquality.equals(newOptions.params, params) ||
         newOptions.crudThrottleTime != crudThrottleTime ||
         newOptions.retryDelay != retryDelay ||
+        newOptions.syncImplementation != source.syncImplementation ||
         newOptions.includeDefaultStreams != includeDefaultStreams;
     return (ResolvedSyncOptions(newOptions), didChange);
   }

packages/powersync_core/lib/src/sync/stream.dart

@@ -144,4 +144,24 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
       },
     );
   }
+
+  Map<String, Object?> toJson() {
+    return {
+      'name': name,
+      'parameters': parameters,
+      'priority': priority.priorityNumber,
+      'associated_buckets': associatedBuckets,
+      'active': active,
+      'is_default': isDefault,
+      'has_explicit_subscription': hasExplicitSubscription,
+      'expires_at': switch (expiresAt) {
+        null => null,
+        final expiresAt => expiresAt.millisecondsSinceEpoch / 1000,
+      },
+      'last_synced_at': switch (lastSyncedAt) {
+        null => null,
+        final lastSyncedAt => lastSyncedAt.millisecondsSinceEpoch / 1000,
+      }
+    };
+  }
 }

packages/powersync_core/lib/src/sync/sync_status.dart

@@ -58,6 +58,7 @@ final class SyncStatus {
 
   final List<CoreActiveStreamSubscription>? _internalSubscriptions;
 
+  @internal
   const SyncStatus({
     this.connected = false,
     this.connecting = false,

packages/powersync_core/lib/src/web/sync_worker.dart

@@ -109,6 +109,8 @@ class _ConnectedClient {
             _runner = null;
             return (JSObject(), null);
           case SyncWorkerMessageType.updateSubscriptions:
+            _runner?.updateClientSubscriptions(
+                this, (payload as UpdateSubscriptions).toDart);
             return (JSObject(), null);
           default:
             throw StateError('Unexpected message type $type');

packages/powersync_core/lib/src/web/sync_worker_protocol.dart

@@ -5,6 +5,7 @@ import 'dart:js_interop';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/sync/options.dart';
+import 'package:powersync_core/src/sync/stream.dart';
 import 'package:web/web.dart';
 
 import '../connector.dart';
@@ -247,6 +248,7 @@ extension type SerializedSyncStatus._(JSObject _) implements JSObject {
     required String? downloadError,
     required JSArray? priorityStatusEntries,
     required JSArray<SerializedBucketProgress>? syncProgress,
+    required JSString streamSubscriptions,
   });
 
   factory SerializedSyncStatus.from(SyncStatus status) {
@@ -272,6 +274,7 @@ extension type SerializedSyncStatus._(JSObject _) implements JSObject {
         var other => SerializedBucketProgress.serialize(
             InternalSyncDownloadProgress.ofPublic(other).buckets),
       },
+      streamSubscriptions: json.encode(status.internalSubscriptions).toJS,
     );
   }
 
@@ -285,8 +288,11 @@ extension type SerializedSyncStatus._(JSObject _) implements JSObject {
   external String? downloadError;
   external JSArray? priorityStatusEntries;
   external JSArray<SerializedBucketProgress>? syncProgress;
+  external JSString? streamSubscriptions;
 
   SyncStatus asSyncStatus() {
+    final streamSubscriptions = this.streamSubscriptions?.toDart;
+
     return SyncStatus(
       connected: connected,
       connecting: connecting,
@@ -306,7 +312,7 @@ extension type SerializedSyncStatus._(JSObject _) implements JSObject {
             final syncedMillis = (rawSynced as JSNumber?)?.toDartInt;
 
             return (
-              priority: BucketPriority((rawPriority as JSNumber).toDartInt),
+              priority: StreamPriority((rawPriority as JSNumber).toDartInt),
               lastSyncedAt: syncedMillis != null
                   ? DateTime.fromMicrosecondsSinceEpoch(syncedMillis)
                   : null,
@@ -320,6 +326,13 @@ extension type SerializedSyncStatus._(JSObject _) implements JSObject {
                 SerializedBucketProgress.deserialize(serializedProgress))
             .asSyncDownloadProgress,
       },
+      streamSubscriptions: switch (streamSubscriptions) {
+        null => null,
+        final serialized => (json.decode(serialized) as List)
+            .map((e) => CoreActiveStreamSubscription.fromJson(
+                e as Map<String, Object?>))
+            .toList(),
+      },
     );
   }
 }