Fix retry local after a deletion

If a conflict occurs during a deletion and a retry local strategy is used in the conflict handler then we want to invoke a delete operation. This was previously always doing an update operation, which would result in the deletion being lost.

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/ConflictResolver.java

@@ -99,7 +99,12 @@ <T extends Model> Single<ModelWithMetadata<T>> resolve(
                 LOG.debug(String.format("Conflict handler decision: %s", decision));
                 @SuppressWarnings("unchecked")
                 ConflictResolutionDecision<T> typedDecision = (ConflictResolutionDecision<T>) decision;
-                return resolveModelAndMetadata(conflictData, metadata, typedDecision);
+                return resolveModelAndMetadata(
+                    conflictData,
+                    metadata,
+                    typedDecision,
+                    pendingMutation.getMutationType()
+                );
             });
     }
 
@@ -156,30 +161,42 @@ private <T extends Model> T getServerModel(@NonNull ModelWithMetadata<T> serverD
     private <T extends Model> Single<ModelWithMetadata<T>> resolveModelAndMetadata(
             @NonNull ConflictData<T> conflictData,
             @NonNull ModelMetadata metadata,
-            @NonNull ConflictResolutionDecision<T> decision) {
+            @NonNull ConflictResolutionDecision<T> decision,
+            @NonNull PendingMutation.Type mutationType) {
 
         switch (decision.getResolutionStrategy()) {
             case RETRY_LOCAL:
-                return publish(conflictData.getLocal(), metadata.getVersion());
+                // When retrying the local mutation we pass the mutation type so that we can correctly retry a deletion
+                // instead of an update
+                return publish(conflictData.getLocal(), metadata.getVersion(), mutationType);
             case APPLY_REMOTE:
                 // No network operations to do. The resolution is just to return
                 // the resolved data, so it can be applied locally.
                 return Single.just(new ModelWithMetadata<>(conflictData.getRemote(), metadata));
             case RETRY:
-                return publish(decision.getCustomModel(), metadata.getVersion());
+                // When retrying with a custom model the mutation is always an update
+                return publish(decision.getCustomModel(), metadata.getVersion(), PendingMutation.Type.UPDATE);
             default:
                 throw new IllegalStateException("Unknown resolution strategy = " + decision.getResolutionStrategy());
         }
     }
 
     @NonNull
-    private <T extends Model> Single<ModelWithMetadata<T>> publish(@NonNull T model, int version) {
+    private <T extends Model> Single<ModelWithMetadata<T>> publish(
+        @NonNull T model,
+        int version,
+        @NonNull PendingMutation.Type mutationType
+    ) {
         return Single
             .<GraphQLResponse<ModelWithMetadata<T>>>create(emitter -> {
                 //SchemaRegistry.instance().getModelSchemaForModelClass method supports schema generation for flutter
                 //models.
                 final ModelSchema schema = SchemaRegistry.instance().getModelSchemaForModelClass(model.getModelName());
-                appSync.update(model, schema, version, emitter::onSuccess, emitter::onError);
+                if (mutationType == PendingMutation.Type.DELETE) {
+                    appSync.delete(model, schema, version, emitter::onSuccess, emitter::onError);
+                } else {
+                    appSync.update(model, schema, version, emitter::onSuccess, emitter::onError);
+                }
             })
             .flatMap(response -> {
                 if (response.hasErrors() || !response.hasData()) {

aws-datastore/src/test/java/com/amplifyframework/datastore/syncengine/ConflictResolverTest.java

@@ -171,17 +171,67 @@ public void conflictIsResolvedByRetryingLocalData() throws DataStoreException {
             .awaitDone(TIMEOUT_SECONDS, TimeUnit.SECONDS)
             .assertValue(versionFromAppSyncResponse);
 
-        // The handler should have called up to AppSync to update hte model
+        // The handler should have called up to AppSync to update the model
         verify(appSync)
             .update(eq(localModel), any(), eq(metadata.getVersion()), any(), any());
     }
 
     /**
-     * When the user elects to retry the mutation using the local copy of the data,
-     * the following is expected:
-     * 1. The AppSync API is invoked, with the local mutation data
-     * 2. We assume that the AppSync API will respond differently
-     *    upon retry
+     * When the user elects to retry the mutation using the local copy of the data, and the mutation is a delete, the
+     * following is expected:
+     * 1. The AppSync delete API is invoked
+     * 2. We assume that the AppSync API will respond differently upon retry
+     * @throws DataStoreException On failure to arrange metadata into storage
+     */
+    @Test
+    public void conflictIsResolvedByRetryingLocalDeletion() throws DataStoreException {
+        // Arrange for the user-provided conflict handler to always request local retry.
+        when(configurationProvider.getConfiguration())
+            .thenReturn(DataStoreConfiguration.builder()
+                            .conflictHandler(DataStoreConflictHandler.alwaysRetryLocal())
+                            .build()
+            );
+
+        // Arrange a pending mutation that includes the local data
+        BlogOwner localModel = BlogOwner.builder()
+                                   .name("Local Blogger")
+                                   .build();
+        PendingMutation<BlogOwner> mutation = PendingMutation.deletion(localModel, schema);
+
+        // Arrange server state for the model, in conflict to local data
+        BlogOwner serverModel = localModel.copyOfBuilder()
+                                    .name("Server Blogger")
+                                    .build();
+        Temporal.Timestamp now = Temporal.Timestamp.now();
+        ModelMetadata metadata = new ModelMetadata(serverModel.getId(), false, 4, now);
+        ModelWithMetadata<BlogOwner> serverData = new ModelWithMetadata<>(serverModel, metadata);
+
+        // Arrange a hypothetical conflict error from AppSync
+        AppSyncConflictUnhandledError<BlogOwner> unhandledConflictError =
+            AppSyncConflictUnhandledErrorFactory.createUnhandledConflictError(serverData);
+
+        // Assume that the AppSync call succeeds this time.
+        ModelWithMetadata<BlogOwner> versionFromAppSyncResponse =
+            new ModelWithMetadata<>(localModel, metadata);
+        AppSyncMocking.delete(appSync)
+            .mockSuccessResponse(localModel, metadata.getVersion(), versionFromAppSyncResponse);
+
+        // Act: when the resolver is invoked, we expect the resolved version
+        // to include the server's metadata, but with the local data.
+        resolver.resolve(mutation, unhandledConflictError)
+            .test()
+            .awaitDone(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+            .assertValue(versionFromAppSyncResponse);
+
+        // The handler should have called up to AppSync to delete the model
+        verify(appSync)
+            .delete(eq(localModel), any(), eq(metadata.getVersion()), any(), any());
+    }
+
+    /**
+     * When the user elects to retry the mutation using the local copy of the data, the following is expected: 1. The
+     * AppSync API is invoked, with the local mutation data 2. We assume that the AppSync API will respond differently
+     * upon retry
      * @throws AmplifyException On failure to arrange metadata into storage
      */
     @Test