fix(datastore): emit observeQuery snapshot when model create mutation results in an updated model

packages/amplify_core/lib/src/types/datastore/models/query_snapshot.dart

@@ -71,37 +71,60 @@ class QuerySnapshot<T extends Model> {
   QuerySnapshot<T> withSubscriptionEvent({
     required SubscriptionEvent<T> event,
   }) {
-    final sortedListCopy = SortedList<T>.from(_sortedList);
     SortedList<T>? updatedSortedList;
 
     final newItem = event.item;
-    final newItemMatchesPredicate = where == null || where!.evaluate(newItem);
-    final currentItemIndex =
+    final matchesPredicate = where == null || where!.evaluate(newItem);
+    final currentIndex =
         // TODO(HuiSF): remove the ignore when merging CPK feature commits
         // ignore: deprecated_member_use_from_same_package
-        sortedListCopy.indexWhere((item) => item.getId() == newItem.getId());
-    final currentItem =
-        currentItemIndex == -1 ? null : sortedListCopy[currentItemIndex];
-    final currentItemMatchesPredicate =
-        currentItem != null && (where == null || where!.evaluate(currentItem));
+        _sortedList.indexWhere((item) => item.getId() == newItem.getId());
+    final currentItem = currentIndex == -1 ? null : _sortedList[currentIndex];
 
-    if (event.eventType == EventType.create &&
-        newItemMatchesPredicate &&
-        currentItem == null) {
-      updatedSortedList = sortedListCopy..addSorted(newItem);
-    } else if (event.eventType == EventType.delete && currentItem != null) {
-      updatedSortedList = sortedListCopy..removeAt(currentItemIndex);
-    } else if (event.eventType == EventType.update) {
-      if (currentItemMatchesPredicate &&
-          newItemMatchesPredicate &&
-          currentItem != newItem) {
-        updatedSortedList = sortedListCopy
-          ..updateAtSorted(currentItemIndex, newItem);
-      } else if (currentItemMatchesPredicate && !newItemMatchesPredicate) {
-        updatedSortedList = sortedListCopy..removeAt(currentItemIndex);
-      } else if (currentItem == null && newItemMatchesPredicate) {
-        updatedSortedList = sortedListCopy..addSorted(newItem);
-      }
+    switch (event.eventType) {
+      case EventType.create:
+        // Skip any new item that doesn't match the predicate.
+        if (!matchesPredicate) break;
+        if (currentItem == null) {
+          // Add the item to the list. This is a new item and matches the
+          // predicate, it should be added.
+          updatedSortedList = _sortedList.copy()..addSorted(newItem);
+        } else if (currentItem != newItem) {
+          // Update the item in the list. This is a "new" item, but it already
+          // exists in the list with a different value. This is the result of
+          // the item being created on this device and App Sync returning an
+          // updated item during the create mutation. This can happen when using
+          // custom resolvers.
+          updatedSortedList = _sortedList.copy()
+            ..updateAtSorted(
+              currentIndex,
+              newItem,
+            );
+        }
+      case EventType.update:
+        if (currentItem == null && matchesPredicate) {
+          // Add the item to the list. This is an existing item that matches the
+          // predicate but is not yet in the list.
+          updatedSortedList = _sortedList.copy()..addSorted(newItem);
+        } else if (currentItem != newItem && matchesPredicate) {
+          // Update the item in the list. This item exists in the list but the
+          // value of the item has changed.
+          updatedSortedList = _sortedList.copy()
+            ..updateAtSorted(
+              currentIndex,
+              newItem,
+            );
+        } else if (currentItem != null && !matchesPredicate) {
+          // Remove the item from the list. The item exist in the list but no
+          // longer matches the predicate.
+          updatedSortedList = _sortedList.copy()..removeAt(currentIndex);
+        }
+      case EventType.delete:
+        if (currentItem != null) {
+          // Remove the item from the list. The item exists in the list but was
+          // just deleted.
+          updatedSortedList = _sortedList.copy()..removeAt(currentIndex);
+        }
     }
     if (updatedSortedList != null) {
       return QuerySnapshot._(

packages/amplify_core/lib/src/types/datastore/models/sorted_list.dart

@@ -25,10 +25,6 @@ class SortedList<E> with ListMixin<E> {
   })  : _items = items,
         _compare = compare;
 
-  SortedList.from(SortedList<E> list)
-      : _items = List.from(list._items),
-        _compare = list._compare;
-
   // Required for ListMixin
   @override
   set length(int newLength) {
@@ -79,6 +75,14 @@ class SortedList<E> with ListMixin<E> {
     }
   }
 
+  /// Returns a copy of the object
+  SortedList<E> copy() {
+    return SortedList.fromPresortedList(
+      items: List.from(_items),
+      compare: _compare,
+    );
+  }
+
   /// Finds the index to insert the [item] at
   ///
   /// O(log(n)) time complexity

packages/amplify_datastore_plugin_interface/test/query_snapshot_test.dart

@@ -39,6 +39,39 @@ void main() {
             expect(updatedSnapshot.items.length, snapshot.items.length + 1);
             expect(updatedSnapshot.items.last, newBlog);
           });
+
+          test(
+              'returns a QuerySnapshot with an updated item if the item already exists',
+              () async {
+            final blog = Blog(name: 'new blog');
+
+            final subscriptionEvent = SubscriptionEvent(
+              item: blog,
+              modelType: Blog.classType,
+              eventType: EventType.create,
+            );
+
+            final updatedSnapshot = snapshot.withSubscriptionEvent(
+              event: subscriptionEvent,
+            );
+
+            expect(updatedSnapshot.items.contains(blog), isTrue);
+
+            final updatedBlog = blog.copyWith(name: 'updated name');
+
+            final subscriptionEvent2 = SubscriptionEvent(
+              item: updatedBlog,
+              modelType: Blog.classType,
+              eventType: EventType.create,
+            );
+
+            final updatedSnapshot2 = updatedSnapshot.withSubscriptionEvent(
+              event: subscriptionEvent2,
+            );
+
+            expect(updatedSnapshot2.items.contains(updatedBlog), isTrue);
+            expect(updatedSnapshot2.items.contains(blog), isFalse);
+          });
         });
 
         group('update event', () {

packages/amplify_datastore_plugin_interface/test/sorted_list_test.dart

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import 'package:amplify_datastore_plugin_interface/amplify_datastore_plugin_interface.dart';
+import 'package:collection/collection.dart';
 import 'package:flutter_test/flutter_test.dart';
 
 void main() {
@@ -61,5 +62,20 @@ void main() {
         expect(sortedList.toList(), orderedEquals(expectedItems));
       });
     });
+
+    group('copy()', () {
+      test('returns a copy of the current list', () {
+        final items = [1, 2, 3, 7, 10];
+        final sortedList = SortedList.fromPresortedList(items: items);
+        final copy = sortedList.copy();
+
+        expect(sortedList.equals(copy), isTrue);
+
+        copy.add(20);
+        expect(sortedList.equals(copy), isFalse);
+        expect(sortedList.length, items.length);
+        expect(copy.length, items.length + 1);
+      });
+    });
   });
 }