feat(DataStore): Multiple models ReconcileAndLocalSave transaction (#1237)

* feat(DataStore): Multiple models ReconcileAndLocalSave transaction

* feat(DataStore): address PR comments

* fix(DataStore): address PR comments 2

Author: lawmicha

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Storage/SQLite/StorageEngineAdapter+SQLite.swift

@@ -18,6 +18,15 @@ final class SQLiteStorageEngineAdapter: StorageEngineAdapter {
     private var dbFilePath: URL?
     static let dbVersionKey = "com.amazonaws.DataStore.dbVersion"
 
+    // TODO benchmark whether a SELECT FROM FOO WHERE ID IN (1, 2, 3...) performs measurably
+    // better than SELECT FROM FOO WHERE ID = 1 OR ID=2 OR ID=3
+    //
+    // SQLite supports up to 1000 expressions per SQLStatement. We have chosen to use 50 expressions
+    // less (equaling 950) than the maximum because it is possible that our SQLStatement already has
+    // some expressions.  If we encounter performance problems in the future, we will want to profile
+    // our system and find an optimal value.
+    static var maxNumberOfPredicates: Int = 950
+
     convenience init(version: String,
                      databaseName: String = "database",
                      userDefaults: UserDefaults = UserDefaults.standard) throws {
@@ -289,12 +298,28 @@ final class SQLiteStorageEngineAdapter: StorageEngineAdapter {
     }
 
     func queryMutationSyncMetadata(for modelId: Model.Identifier) throws -> MutationSyncMetadata? {
+        let results = try queryMutationSyncMetadata(for: [modelId])
+        return try results.unique()
+    }
+
+    func queryMutationSyncMetadata(for modelIds: [Model.Identifier]) throws -> [MutationSyncMetadata] {
         let modelType = MutationSyncMetadata.self
-        let statement = SelectStatement(from: modelType.schema, predicate: field("id").eq(modelId))
-        let rows = try connection.prepare(statement.stringValue).run(statement.variables)
-        let result = try rows.convert(to: modelType,
-                                      using: statement)
-        return try result.unique()
+        let fields = MutationSyncMetadata.keys
+        var results = [MutationSyncMetadata]()
+        let chunkedModelIdsArr = modelIds.chunked(into: SQLiteStorageEngineAdapter.maxNumberOfPredicates)
+        for chunkedModelIds in chunkedModelIdsArr {
+            var queryPredicates: [QueryPredicateOperation] = []
+            for id in chunkedModelIds {
+                queryPredicates.append(QueryPredicateOperation(field: fields.id.stringValue, operator: .equals(id)))
+            }
+            let groupedQueryPredicates = QueryPredicateGroup(type: .or, predicates: queryPredicates)
+            let statement = SelectStatement(from: modelType.schema, predicate: groupedQueryPredicates)
+            let rows = try connection.prepare(statement.stringValue).run(statement.variables)
+            let result = try rows.convert(to: modelType,
+                                          using: statement)
+            results.append(contentsOf: result)
+        }
+        return results
     }
 
     func queryModelSyncMetadata(for modelSchema: ModelSchema) throws -> ModelSyncMetadata? {

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Storage/StorageEngine+DeleteTransaction.swift

@@ -122,13 +122,8 @@ extension StorageEngine {
             return []
         }
 
-        // SQLite supports up to 1000 expressions per SQLStatement. We have chosen to use 50 expressions
-        // less (equaling 950) than the maximum because it is possible that our SQLStatement already has
-        // some expressions.  If we encounter performance problems in the future, we will want to profile
-        // our system and find an optimal value.
-        let maxNumberOfPredicates = 950
         var queriedModels: [Model] = []
-        let chunkedArrays = ids.chunked(into: maxNumberOfPredicates)
+        let chunkedArrays = ids.chunked(into: SQLiteStorageEngineAdapter.maxNumberOfPredicates)
         for chunkedArray in chunkedArrays {
             // TODO: Add conveinence to queryPredicate where we have a list of items, to be all or'ed
             var queryPredicates: [QueryPredicateOperation] = []

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Storage/StorageEngineAdapter.swift

@@ -11,6 +11,8 @@ import AWSPluginsCore
 
 protocol StorageEngineAdapter: class, ModelStorageBehavior {
 
+    static var maxNumberOfPredicates: Int { get }
+
     // MARK: - Async APIs
     func save(untypedModel: Model, completion: @escaping DataStoreCallback<Model>)
 
@@ -47,6 +49,8 @@ protocol StorageEngineAdapter: class, ModelStorageBehavior {
 
     func queryMutationSyncMetadata(for modelId: Model.Identifier) throws -> MutationSyncMetadata?
 
+    func queryMutationSyncMetadata(for modelIds: [Model.Identifier]) throws -> [MutationSyncMetadata]
+
     func queryModelSyncMetadata(for modelSchema: ModelSchema) throws -> ModelSyncMetadata?
 
     func transaction(_ basicClosure: BasicThrowableClosure) throws

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/InitialSyncOperation.swift

@@ -168,8 +168,8 @@ final class InitialSyncOperation: AsynchronousOperation {
         let items = syncQueryResult.items
         recordsReceived += UInt(items.count)
 
+        reconciliationQueue.offer(items, modelSchema: modelSchema)
         for item in items {
-            reconciliationQueue.offer(item, modelSchema: modelSchema)
             initialSyncOperationTopic.send(.enqueued(item))
         }
 

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/MutationSync/AWSMutationDatabaseAdapter/AWSMutationDatabaseAdapter+MutationEventIngester.swift

@@ -49,7 +49,7 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
         }
 
         MutationEvent.pendingMutationEvents(
-            forModelId: mutationEvent.modelId,
+            for: mutationEvent.modelId,
             storageAdapter: storageAdapter) { result in
                 switch result {
                 case .failure(let dataStoreError):

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/AWSIncomingEventReconciliationQueue.swift

@@ -102,13 +102,13 @@ final class AWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueu
         eventReconciliationQueueTopic.send(.paused)
     }
 
-    func offer(_ remoteModel: MutationSync<AnyModel>, modelSchema: ModelSchema) {
+    func offer(_ remoteModels: [MutationSync<AnyModel>], modelSchema: ModelSchema) {
         guard let queue = reconciliationQueues[modelSchema.name] else {
             // TODO: Error handling
             return
         }
 
-        queue.enqueue(remoteModel)
+        queue.enqueue(remoteModels)
     }
 
     private func onReceiveCompletion(completed: Subscribers.Completion<DataStoreError>) {

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/IncomingEventReconciliationQueue.swift

@@ -25,6 +25,6 @@ enum IncomingEventReconciliationQueueEvent {
 protocol IncomingEventReconciliationQueue: class, AmplifyCancellable {
     func start()
     func pause()
-    func offer(_ remoteModel: MutationSync<AnyModel>, modelSchema: ModelSchema)
+    func offer(_ remoteModels: [MutationSync<AnyModel>], modelSchema: ModelSchema)
     var publisher: AnyPublisher<IncomingEventReconciliationQueueEvent, DataStoreError> { get }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/AWSModelReconciliationQueue.swift

@@ -135,9 +135,14 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
         incomingSubscriptionEventQueue.cancelAllOperations()
     }
 
-    func enqueue(_ remoteModel: MutationSync<AnyModel>) {
+    func enqueue(_ remoteModels: [MutationSync<AnyModel>]) {
+        guard let remoteModelName = remoteModels.first?.model.modelName else {
+            log.debug("\(#function) skipping reconciliation, no models to enqueue.")
+            return
+        }
+
         let reconcileOp = ReconcileAndLocalSaveOperation(modelSchema: modelSchema,
-                                                         remoteModel: remoteModel,
+                                                         remoteModels: remoteModels,
                                                          storageAdapter: storageAdapter)
         var reconcileAndLocalSaveOperationSink: AnyCancellable?
         reconcileAndLocalSaveOperationSink = reconcileOp.publisher.sink(receiveCompletion: { completion in
@@ -154,7 +159,7 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
             }
         })
         reconcileAndLocalSaveOperationSinks.with { $0.insert(reconcileAndLocalSaveOperationSink) }
-        reconcileAndSaveQueue.addOperation(reconcileOp, modelName: remoteModel.model.modelName)
+        reconcileAndSaveQueue.addOperation(reconcileOp, modelName: remoteModelName)
     }
 
     private func receive(_ receive: IncomingSubscriptionEventPublisherEvent) {
@@ -166,7 +171,7 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
                 }
             }
             incomingSubscriptionEventQueue.addOperation(CancelAwareBlockOperation {
-                self.enqueue(remoteModel)
+                self.enqueue([remoteModel])
             })
         case .connectionConnected:
             modelReconciliationQueueSubject.send(.connected(modelName: modelSchema.name))

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/ModelReconciliationQueue.swift

@@ -28,6 +28,6 @@ protocol ModelReconciliationQueue {
     func start()
     func pause()
     func cancel()
-    func enqueue(_ remoteModel: MutationSync<AnyModel>)
+    func enqueue(_ remoteModels: [MutationSync<AnyModel>])
     var publisher: AnyPublisher<ModelReconciliationQueueEvent, DataStoreError> { get }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/ReconcileAndLocalSaveOperation+Action.swift

@@ -13,23 +13,10 @@ extension ReconcileAndLocalSaveOperation {
     /// Actions are declarative, they say what I just did
     enum Action {
         /// Operation has been started by the queue
-        case started(RemoteModel)
+        case started([RemoteModel])
 
-        /// Operation has retrieved RemoteModel's corresponding sync metadata from local database
-        case queried(RemoteModel, LocalMetadata?)
-
-        /// Operation has reconciled the incoming remote model with local model and sync metadata
-        case reconciled(RemoteSyncReconciler.Disposition)
-
-        /// Operation has applied the incoming RemoteModel to the local database per the reconciled disposition. This
-        /// could result in either a save to the local database, or a delete from the local database.
-        case applied(AppliedModel, mutationType: MutationEvent.MutationType)
-
-        /// Operation dropped the remote model per the reconciled disposition.
-        case dropped(modelName: String)
-
-        /// Operation notified listeners and callbacks of completion
-        case notified
+        /// Operation completed reconcilliation
+        case reconciled
 
         /// Operation has been cancelled by the queue
         case cancelled