|
| 1 | +// |
| 2 | +// Copyright Amazon.com Inc. or its affiliates. |
| 3 | +// All Rights Reserved. |
| 4 | +// |
| 5 | +// SPDX-License-Identifier: Apache-2.0 |
| 6 | +// |
| 7 | + |
| 8 | +import Amplify |
| 9 | +import Dispatch |
| 10 | +import AWSPluginsCore |
| 11 | + |
| 12 | +extension MutationEvent { |
| 13 | + // Consecutive operations that modify a model results in a sequence of pending mutation events that |
| 14 | + // have the current version of the model. The first mutation event has the correct version of the model, |
| 15 | + // while the subsequent events will have lower versions if the first mutation event is successfully synced |
| 16 | + // to the cloud. By reconciling the pending mutation events after syncing the first mutation event, |
| 17 | + // we attempt to update the pending version to the latest version from the response. |
| 18 | + // The before and after conditions for consecutive update scenarios are as below: |
| 19 | + // - Save, then immediately update |
| 20 | + // Queue Before - [(version: nil, inprocess: true, type: .create), |
| 21 | + // (version: nil, inprocess: false, type: .update)] |
| 22 | + // Response - [version: 1, type: .create] |
| 23 | + // Queue After - [(version: 1, inprocess: false, type: .update)] |
| 24 | + // - Save, then immediately delete |
| 25 | + // Queue Before - [(version: nil, inprocess: true, type: .create), |
| 26 | + // (version: nil, inprocess: false, type: .delete)] |
| 27 | + // Response - [version: 1, type: .create] |
| 28 | + // Queue After - [(version: 1, inprocess: false, type: .delete)] |
| 29 | + // - Save, sync, then immediately update and delete |
| 30 | + // Queue Before (After save, sync) |
| 31 | + // - [(version: 1, inprocess: true, type: .update), (version: 1, inprocess: false, type: .delete)] |
| 32 | + // Response - [version: 2, type: .update] |
| 33 | + // Queue After - [(version: 2, inprocess: false, type: .delete)] |
| 34 | + // |
| 35 | + // For a given model `id`, checks the version of the head of pending mutation event queue |
| 36 | + // against the API response version in `mutationSync` and saves it in the mutation event table if |
| 37 | + // the response version is a newer one |
| 38 | + static func reconcilePendingMutationEventsVersion(sent mutationEvent: MutationEvent, |
| 39 | + received mutationSync: MutationSync<AnyModel>, |
| 40 | + storageAdapter: StorageEngineAdapter, |
| 41 | + completion: @escaping DataStoreCallback<Void>) { |
| 42 | + MutationEvent.pendingMutationEvents( |
| 43 | + forMutationEvent: mutationEvent, |
| 44 | + storageAdapter: storageAdapter) { queryResult in |
| 45 | + switch queryResult { |
| 46 | + case .failure(let dataStoreError): |
| 47 | + completion(.failure(dataStoreError)) |
| 48 | + case .success(let localMutationEvents): |
| 49 | + guard let existingEvent = localMutationEvents.first else { |
| 50 | + completion(.success(())) |
| 51 | + return |
| 52 | + } |
| 53 | + |
| 54 | + guard let reconciledEvent = reconcile(pendingMutationEvent: existingEvent, |
| 55 | + with: mutationEvent, |
| 56 | + responseMutationSync: mutationSync) else { |
| 57 | + completion(.success(())) |
| 58 | + return |
| 59 | + } |
| 60 | + |
| 61 | + storageAdapter.save(reconciledEvent, condition: nil) { result in |
| 62 | + switch result { |
| 63 | + case .failure(let dataStoreError): |
| 64 | + completion(.failure(dataStoreError)) |
| 65 | + case .success: |
| 66 | + completion(.success(())) |
| 67 | + } |
| 68 | + } |
| 69 | + } |
| 70 | + } |
| 71 | + } |
| 72 | + |
| 73 | + static func reconcile(pendingMutationEvent: MutationEvent, |
| 74 | + with requestMutationEvent: MutationEvent, |
| 75 | + responseMutationSync: MutationSync<AnyModel>) -> MutationEvent? { |
| 76 | + // return if version of the pending mutation event is not nil and |
| 77 | + // is >= version contained in the response |
| 78 | + if pendingMutationEvent.version != nil && |
| 79 | + pendingMutationEvent.version! >= responseMutationSync.syncMetadata.version { |
| 80 | + return nil |
| 81 | + } |
| 82 | + |
| 83 | + do { |
| 84 | + let responseModel = responseMutationSync.model.instance |
| 85 | + let requestModel = try requestMutationEvent.decodeModel() |
| 86 | + |
| 87 | + // check if the data sent in the request is the same as the response |
| 88 | + // if it is, update the pending mutation event version to the response version |
| 89 | + guard let modelSchema = ModelRegistry.modelSchema(from: requestMutationEvent.modelName), |
| 90 | + modelSchema.compare(responseModel, requestModel) else { |
| 91 | + return nil |
| 92 | + } |
| 93 | + |
| 94 | + var pendingMutationEvent = pendingMutationEvent |
| 95 | + pendingMutationEvent.version = responseMutationSync.syncMetadata.version |
| 96 | + return pendingMutationEvent |
| 97 | + } catch { |
| 98 | + Amplify.log.verbose("Error decoding models: \(error)") |
| 99 | + return nil |
| 100 | + } |
| 101 | + } |
| 102 | + |
| 103 | +} |
0 commit comments