chore(dataStore): Update datastore API docs and fix race conditions in tests (#1596)

* chore: add delay to observe and observeQuery tests

* chore: update api docs for datastore

Author: Jordan-Nelson

packages/amplify/amplify_flutter/lib/src/categories/amplify_datastore_category.dart

@@ -15,12 +15,15 @@
 
 part of amplify_interface;
 
-/// Interface for DataStore category. This expose all the APIs that
-/// are supported by this category's plugins. This class will accept plugins to
-/// be registered and configured and then subsequent API calls will be forwarded
-/// to those plugins.
+/// {@template amplify_flutter.amplify_datastore_category}
+/// Interface for the DataStore category.
+///
+/// This exposes all the APIs that are supported by this category's plugins.
+/// This class will accept plugins to be registered and configured and then
+/// subsequent API calls will be forwarded to those plugins.
+/// {@endtemplate}
 class DataStoreCategory {
-  /// Default constant constructor
+  /// {@macro amplify_flutter.amplify_datastore_category}
   const DataStoreCategory();
 
   /// Added DataStore plugins.
@@ -66,7 +69,9 @@ class DataStoreCategory {
   }
 
   /// Query the DataStore to find all items of the specified [modelType] that satisfy the specified
-  /// query predicate [where]. Returned items are paginated by [pagination] and sorted by [sortBy].
+  /// query predicate [where].
+  ///
+  /// Returned items are paginated by [pagination] and sorted by [sortBy].
   Future<List<T>> query<T extends Model>(ModelType<T> modelType,
       {QueryPredicate? where,
       QueryPagination? pagination,
@@ -99,17 +104,25 @@ class DataStoreCategory {
         : throw _pluginNotAddedException('DataStore');
   }
 
-  /// Stops the underlying DataStore, resetting the plugin to the initialized state, and deletes all data
-  /// from the local device. Remotely synced data can be re-synced back when starting DataStore using
-  /// [start]. local-only data will be lost permanently.
+  /// Stops the underlying DataStore's synchronization with a remote system, if DataStore is configured to
+  /// support remote synchronization, resetting the plugin to the initialized state, and deletes all data
+  /// from the local device.
+  ///
+  /// Any items pending synchronization in the outbound queue will be lost. Remotely synced data can be re-synced
+  /// back when starting DataStore using [start]. **Local-only data will be lost permanently.**
+  ///
+  /// Synchronization processes will be restarted on the next interaction with the DataStore, or can be
+  /// restarted manually by calling [start].
   Future<void> clear() {
     return plugins.length == 1
         ? plugins[0].clear()
         : throw _pluginNotAddedException('DataStore');
   }
 
   /// Starts the DataStore's synchronization with a remote system, if DataStore is configured to support
-  /// remote synchronization. This only needs to be called if you wish to start the synchronization eagerly.
+  /// remote synchronization.
+  ///
+  /// This only needs to be called if you wish to start the synchronization eagerly.
   /// If you don't call start(), the synchronization will start automatically, prior to executing any other
   /// operations (query, save, delete, update).
   Future<void> start() {
@@ -120,16 +133,19 @@ class DataStoreCategory {
 
   /// Stops the underlying DataStore's synchronization with a remote system, if DataStore is configured to
   /// support remote synchronization.
+  ///
+  /// Synchronization processes will be restarted on the next interaction with the DataStore, or can be
+  /// restarted manually by calling [start].
   Future<void> stop() {
     return plugins.length == 1
         ? plugins[0].stop()
         : throw _pluginNotAddedException('DataStore');
   }
 
-  /// Observe the result set of a given Query
+  /// Observe the result set of a given Query.
   ///
   /// Emits an initial [QuerySnapshot] with data from the local store, as well as
-  /// subsequent events with data synced over the network
+  /// subsequent events with data synced over the network.
   Stream<QuerySnapshot<T>> observeQuery<T extends Model>(
     ModelType<T> modelType, {
     QueryPredicate? where,

packages/amplify_datastore/example/integration_test/observe_query_test.dart

@@ -27,6 +27,7 @@ void main() {
     setUp(() async {
       await configureDataStore();
       await clearDataStore();
+      await waitForObserve();
     });
 
     testWidgets(
@@ -59,14 +60,18 @@ void main() {
         Blog.classType,
       ).map((event) => event.items);
 
+      // should emit an initial snapshot with the data in the store
+      final firstSnapshot = await observeQueryItemStream.first;
+      expect(firstSnapshot, orderedEquals(blogs));
+
       Blog newBlog1 = Blog(name: 'new blog 1');
       Blog newBlog2 = Blog(name: 'new blog 2');
       Blog newBlog1Copy = newBlog1.copyWith(name: 'new name');
 
+      // should emit a snapshot for each save/update
       expectLater(
         observeQueryItemStream,
         emitsInOrder([
-          orderedEquals([...blogs]),
           orderedEquals([...blogs, newBlog1]),
           orderedEquals([...blogs, newBlog1, newBlog2]),
           orderedEquals([...blogs, newBlog1Copy, newBlog2]),
@@ -92,15 +97,19 @@ void main() {
         where: Blog.NAME.contains('blog'),
       ).map((event) => event.items);
 
+      // should emit an initial snapshot with the data in the store
+      final firstSnapshot = await observeQueryItemStream.first;
+      expect(firstSnapshot, orderedEquals(blogs));
+
       Blog newBlog1 = Blog(name: 'new blog 1');
       Blog newBlog2 = Blog(name: 'new blog 2');
       Blog newBlog3 = Blog(name: 'new 3');
       Blog newBlog1Copy = newBlog1.copyWith(name: 'new name');
 
+      // should emit a snapshot for each save/update
       expectLater(
         observeQueryItemStream,
         emitsInOrder([
-          orderedEquals([...blogs]),
           orderedEquals([...blogs, newBlog1]),
           orderedEquals([...blogs, newBlog1, newBlog2]),
           orderedEquals([...blogs, newBlog2]),
@@ -125,14 +134,18 @@ void main() {
         sortBy: [Blog.NAME.ascending()],
       ).map((event) => event.items);
 
+      // should emit an initial snapshot with the data in the store
+      final firstSnapshot = await observeQueryItemStream.first;
+      expect(firstSnapshot, orderedEquals(blogs));
+
       Blog newBlog1 = Blog(name: 'aaa blog');
       Blog newBlog2 = Blog(name: 'ccc blog');
       Blog newBlog2Copy = newBlog2.copyWith(name: 'azz blog');
 
+      // should emit a snapshot for each save/update
       expectLater(
         observeQueryItemStream,
         emitsInOrder([
-          orderedEquals([...blogs]),
           orderedEquals([newBlog1, ...blogs]),
           orderedEquals([newBlog1, ...blogs, newBlog2]),
           orderedEquals([newBlog1, newBlog2Copy, ...blogs]),

packages/amplify_datastore/example/integration_test/observe_test.dart

@@ -28,6 +28,22 @@ void main() {
     setUp(() async {
       await configureDataStore();
       await clearDataStore();
+      await waitForObserve();
+    });
+
+    testWidgets('should emit an event for each item saved',
+        (WidgetTester tester) async {
+      var itemStream = Amplify.DataStore.observe(Blog.classType)
+          .map((event) => event.item.name);
+
+      expectLater(
+        itemStream,
+        emitsInOrder(['blog 1', 'blog 2', 'blog 3']),
+      );
+
+      await Amplify.DataStore.save(Blog(name: 'blog 1'));
+      await Amplify.DataStore.save(Blog(name: 'blog 2'));
+      await Amplify.DataStore.save(Blog(name: 'blog 3'));
     });
 
     testWidgets('should broadcast events for create, update, and delete',

packages/amplify_datastore/example/integration_test/utils/setup_utils.dart

@@ -26,6 +26,7 @@ const ENABLE_CLOUD_SYNC =
 const DATASTORE_READY_EVENT_TIMEOUT = const Duration(minutes: 2);
 const DELAY_TO_START_DATASTORE = const Duration(milliseconds: 500);
 const DELAY_TO_CLEAR_DATASTORE = const Duration(seconds: 2);
+const DELAY_FOR_OBSERVE = const Duration(milliseconds: 100);
 
 /// Configure [AmplifyDataStore] plugin with given [modelProvider].
 /// When [ENABLE_CLOUD_SYNC] environment variable is set to true, it also
@@ -62,6 +63,34 @@ Future<void> clearDataStore() async {
   await Amplify.DataStore.clear();
 }
 
+/// Waits for observe to be set up properly.
+///
+/// There is a bug in observe that causes events to be missed when save/delete is called
+/// immediately after. This work around allows the datastore tests that use observe to still run.
+///
+/// See: https://github.com/aws-amplify/amplify-flutter/issues/1590
+Future<void> waitForObserve() async {
+  bool isObserveSetUp = false;
+  final blog = Blog(name: 'TEST BLOG - Used to wait for observe to start');
+
+  // set up observe subscription and set isObserveSetUp to true when the first update event comes in
+  Amplify.DataStore.observe(Blog.classType)
+      .where((event) => event.item.id == blog.id)
+      .first
+      .then((event) {
+    isObserveSetUp = true;
+  });
+
+  // update in a loop until observe is set up
+  while (!isObserveSetUp) {
+    await Future.delayed(DELAY_FOR_OBSERVE);
+    await Amplify.DataStore.save(blog);
+  }
+
+  // clean up blog
+  await Amplify.DataStore.delete(blog);
+}
+
 /// Am async operator that starts DataStore API sync.
 /// It returns a Future which complets when [Amplify.Hub] receives the ready
 /// event on [HubChannel.DataStore], or completes with an timeout error if