feat(datastore): Dispatch outboxMutationEnqueued, outboxMutationProcessed events (#759)

* initial outboxMutation PR

* renamed OutboxMutationEventMetadata as OutboxMutationEventElement

* fixed one unit test per my PR

* Added one integration test

* Updated some comments

* added one comment

* outboxMutationProcessed is emitted only when there is a successful response

* updated the logic

* did some clean up

* about to merge

Author: 618coffee

Amplify/Categories/DataStore/DataStoreCategory+HubPayloadEventName.swift

@@ -37,4 +37,12 @@ public extension HubPayload.EventName.DataStore {
     /// Dispatched when DataStore is about to start sync queries
     /// HubPayload `syncQueriesStartedEvent` contains an array of each model's `name`
     static let syncQueriesStarted = "DataStore.syncQueriesStarted"
+
+    /// Dispatched when a local mutation is enqueued into the outgoing mutation queue `outbox`
+    /// HubPayload `outboxMutationEvent` contains the name and instance of the model
+    static let outboxMutationEnqueued = "DataStore.outboxMutationEnqueued"
+
+    /// Dispatched when a mutation from outgoing mutation queue `outbox` is sent to backend and updated locally.
+    /// HubPayload `outboxMutationEvent` contains the name and instance of the model
+    static let outboxMutationProcessed = "DataStore.outboxMutationProcessed"
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/Events/OutboxMutationEvent.swift

@@ -0,0 +1,42 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Amplify
+import AWSPluginsCore
+
+/// Used as HubPayload for `OutboxMutationEnqueued` and `OutboxMutationProcessed`
+public struct OutboxMutationEvent {
+    public let modelName: String
+    public let element: OutboxMutationEventElement
+
+    public init(modelName: String, element: OutboxMutationEventElement) {
+        self.modelName = modelName
+        self.element = element
+    }
+    public static func fromModelWithMetadata(modelName: String,
+                                             model: Model,
+                                             mutationSync: MutationSync<AnyModel>) -> OutboxMutationEvent {
+        let element = OutboxMutationEventElement(model: model,
+                                                 version: mutationSync.syncMetadata.version,
+                                                 lastChangedAt: mutationSync.syncMetadata.lastChangedAt,
+                                                 deleted: mutationSync.syncMetadata.deleted)
+        return OutboxMutationEvent(modelName: modelName, element: element)
+    }
+
+    public static func fromModelWithoutMetadata(modelName: String,
+                                                model: Model) -> OutboxMutationEvent {
+        let element = OutboxMutationEventElement(model: model)
+        return OutboxMutationEvent(modelName: modelName, element: element)
+    }
+
+    public struct OutboxMutationEventElement {
+        public let model: Model
+        public var version: Int?
+        public var lastChangedAt: Int?
+        public var deleted: Bool?
+    }
+}

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift

@@ -190,6 +190,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                 self.processSyncMutationToCloudResult(result, mutationEvent: mutationEvent, api: api)
         }
 
+        dispatchOutboxMutationEnqueuedEvent(mutationEvent: mutationEvent)
         dispatchOutboxStatusEvent(isEmpty: false)
         operationQueue.addOperation(syncMutationToCloudOperation)
         stateMachine.notify(action: .enqueuedEvent)
@@ -198,18 +199,21 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
     private func processSyncMutationToCloudResult(_ result: GraphQLOperation<MutationSync<AnyModel>>.OperationResult,
                                                   mutationEvent: MutationEvent,
                                                   api: APICategoryGraphQLBehavior) {
-        if case let .success(graphQLResponse) = result, case let .failure(graphQLResponseError) = graphQLResponse {
-            processMutationErrorFromCloud(mutationEvent: mutationEvent,
-                                          api: api,
-                                          apiError: nil,
-                                          graphQLResponseError: graphQLResponseError)
+        if case let .success(graphQLResponse) = result {
+            if case let .success(graphQLResult) = graphQLResponse {
+                completeProcessingEvent(mutationEvent,
+                                        mutationSyncMetadata: graphQLResult)
+            } else if case let .failure(graphQLResponseError) = graphQLResponse {
+                processMutationErrorFromCloud(mutationEvent: mutationEvent,
+                                              api: api,
+                                              apiError: nil,
+                                              graphQLResponseError: graphQLResponseError)
+            }
         } else if case let .failure(apiError) = result {
             processMutationErrorFromCloud(mutationEvent: mutationEvent,
                                           api: api,
                                           apiError: apiError,
                                           graphQLResponseError: nil)
-        } else {
-            completeProcessingEvent(mutationEvent)
         }
     }
 
@@ -229,12 +233,14 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                     let outgoingMutationEvent = mutationEventOptional {
                     self.outgoingMutationQueueSubject.send(outgoingMutationEvent)
                 }
-                self.completeProcessingEvent(mutationEvent)
+                self.completeProcessingEvent(mutationEvent,
+                                             mutationSyncMetadata: nil)
         }
         operationQueue.addOperation(processMutationErrorFromCloudOperation)
     }
 
-    private func completeProcessingEvent(_ mutationEvent: MutationEvent) {
+    private func completeProcessingEvent(_ mutationEvent: MutationEvent,
+                                         mutationSyncMetadata: MutationSync<AnyModel>?) {
         // This doesn't belong here--need to add a `delete` API to the MutationEventSource and pass a
         // reference into the mutation queue.
         Amplify.DataStore.delete(mutationEvent) { result in
@@ -245,6 +251,10 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                 self.log.verbose("mutationEvent deleted successfully")
             }
 
+            if let mutationSyncMetadata = mutationSyncMetadata {
+                self.dispatchOutboxMutationProcessedEvent(mutationEvent: mutationEvent,
+                                                          mutationSync: mutationSyncMetadata)
+            }
             self.queryMutationEventsFromStorage {
                 self.stateMachine.notify(action: .processedEvent)
             }
@@ -269,6 +279,38 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
         }
     }
 
+    private func dispatchOutboxMutationProcessedEvent(mutationEvent: MutationEvent,
+                                                      mutationSync: MutationSync<AnyModel>) {
+        do {
+            let localModel = try mutationEvent.decodeModel()
+            let outboxMutationProcessedEvent = OutboxMutationEvent
+                .fromModelWithMetadata(modelName: mutationEvent.modelName,
+                                       model: localModel,
+                                       mutationSync: mutationSync)
+            let payload = HubPayload(eventName: HubPayload.EventName.DataStore.outboxMutationProcessed,
+                                     data: outboxMutationProcessedEvent)
+            Amplify.Hub.dispatch(to: .dataStore, payload: payload)
+        } catch {
+            log.error("\(#function) Couldn't decode local model as \(mutationEvent.modelName)")
+            return
+        }
+    }
+
+    private func dispatchOutboxMutationEnqueuedEvent(mutationEvent: MutationEvent) {
+        do {
+            let localModel = try mutationEvent.decodeModel()
+            let outboxMutationEnqueuedEvent = OutboxMutationEvent
+                .fromModelWithoutMetadata(modelName: mutationEvent.modelName,
+                                          model: localModel)
+            let payload = HubPayload(eventName: HubPayload.EventName.DataStore.outboxMutationEnqueued,
+                                     data: outboxMutationEnqueuedEvent)
+            Amplify.Hub.dispatch(to: .dataStore, payload: payload)
+        } catch {
+            log.error("\(#function) Couldn't decode local model as \(mutationEvent.modelName)")
+            return
+        }
+    }
+
     private func dispatchOutboxStatusEvent(isEmpty: Bool) {
         let outboxStatusEvent = OutboxStatusEvent(isEmpty: isEmpty)
         let outboxStatusEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.outboxStatus,

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/DataStoreHubEventsTests.swift

@@ -63,5 +63,62 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
         }
 
         waitForExpectations(timeout: networkTimeout, handler: nil)
+        Amplify.Hub.removeListener(hubListener)
+    }
+
+    /// - Given:
+    ///    - registered two models from `TestModelRegistration`
+    ///    - no pending MutationEvents in MutationEvent database
+    /// - When:
+    ///    - Call `Amplify.DataStore.save()` to save a Post model
+    /// - Then:
+    ///    - outboxMutationEnqueued received, payload should be:
+    ///      {modelName: "Post", element: {id: #, content: "some content"}}
+    ///    - outboxMutationProcessed received, payload should be:
+    ///      {modelName: "Post", element: {model: {id: #, content: "some content"}, version: 1, deleted: false, lastChangedAt: "some time"}}
+    func testOutboxMutationEvents() throws {
+
+        let post = Post(title: "title", content: "content", createdAt: .now())
+
+        let outboxMutationEnqueuedReceived = expectation(description: "outboxMutationEnqueued received")
+        let outboxMutationProcessedReceived = expectation(description: "outboxMutationProcessed received")
+
+        let hubListener = Amplify.Hub.listen(to: .dataStore) { payload in
+            if payload.eventName == HubPayload.EventName.DataStore.outboxMutationEnqueued {
+                guard let outboxMutationEnqueuedEvent = payload.data as? OutboxMutationEvent else {
+                    XCTFail("Failed to cast payload data as OutboxMutationEvent")
+                    return
+                }
+                XCTAssertEqual(outboxMutationEnqueuedEvent.modelName, "Post")
+                XCTAssertEqual(outboxMutationEnqueuedEvent.element.model.modelName, "Post")
+                XCTAssertNil(outboxMutationEnqueuedEvent.element.version)
+                XCTAssertNil(outboxMutationEnqueuedEvent.element.lastChangedAt)
+                XCTAssertNil(outboxMutationEnqueuedEvent.element.deleted)
+                outboxMutationEnqueuedReceived.fulfill()
+            }
+
+            if payload.eventName == HubPayload.EventName.DataStore.outboxMutationProcessed {
+                guard let outboxMutationProcessedEvent = payload.data as? OutboxMutationEvent else {
+                    XCTFail("Failed to cast payload data as OutboxMutationEvent")
+                    return
+                }
+                XCTAssertEqual(outboxMutationProcessedEvent.modelName, "Post")
+                XCTAssertEqual(outboxMutationProcessedEvent.element.model.modelName, "Post")
+                XCTAssertEqual(outboxMutationProcessedEvent.element.version, 1)
+                XCTAssertNotNil(outboxMutationProcessedEvent.element.lastChangedAt)
+                XCTAssertEqual(outboxMutationProcessedEvent.element.deleted, false)
+                outboxMutationProcessedReceived.fulfill()
+            }
+        }
+
+        guard try HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
+            XCTFail("Listener not registered for hub")
+            return
+        }
+
+        Amplify.DataStore.save(post) { _ in }
+
+        waitForExpectations(timeout: networkTimeout, handler: nil)
+        Amplify.Hub.removeListener(hubListener)
     }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/README.md

@@ -27,6 +27,6 @@ The following steps demonstrate how to set up DataStore with a conflict resoluti
 
 4. `amplify push`
 
-5. Copy `awsconfiguration.json` over to the Config folder
+5. Copy `amplifyconfiguration.json` over to the Config folder
 
 You should now be able to run all of the tests 

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/MutationQueue/OutgoingMutationQueueTests.swift

@@ -33,21 +33,35 @@ class OutgoingMutationQueueTests: SyncEngineTestBase {
         var outboxStatusReceivedCurrentCount = 0
         let outboxStatusOnStart = expectation(description: "On DataStore start, outboxStatus received")
         let outboxStatusOnMutationEnqueued = expectation(description: "Mutation enqueued, outboxStatus received")
+        let outboxMutationEnqueued = expectation(description: "Mutation enqueued, outboxMutationEnqueued received")
+
+        let outboxStatusFilter = HubFilters.forEventName(HubPayload.EventName.DataStore.outboxStatus)
+        let outboxMutationEnqueuedFilter = HubFilters.forEventName(HubPayload.EventName.DataStore.outboxMutationEnqueued)
+        let filters = HubFilters.any(filters: outboxStatusFilter, outboxMutationEnqueuedFilter)
+        let hubListener = Amplify.Hub.listen(to: .dataStore, isIncluded: filters) { payload in
+            if payload.eventName == HubPayload.EventName.DataStore.outboxStatus {
+                outboxStatusReceivedCurrentCount += 1
+                guard let outboxStatusEvent = payload.data as? OutboxStatusEvent else {
+                    XCTFail("Failed to cast payload data as OutboxStatusEvent")
+                    return
+                }
 
-        let filter = HubFilters.forEventName(HubPayload.EventName.DataStore.outboxStatus)
-        let hubListener = Amplify.Hub.listen(to: .dataStore, isIncluded: filter) { payload in
-            outboxStatusReceivedCurrentCount += 1
-            guard let outboxStatusEvent = payload.data as? OutboxStatusEvent else {
-                XCTFail("Failed to cast payload data as OutboxStatusEvent")
-                return
+                if outboxStatusReceivedCurrentCount == 1 {
+                    XCTAssertTrue(outboxStatusEvent.isEmpty)
+                    outboxStatusOnStart.fulfill()
+                } else {
+                    XCTAssertFalse(outboxStatusEvent.isEmpty)
+                    outboxStatusOnMutationEnqueued.fulfill()
+                }
             }
 
-            if outboxStatusReceivedCurrentCount == 1 {
-                XCTAssertTrue(outboxStatusEvent.isEmpty)
-                outboxStatusOnStart.fulfill()
-            } else {
-                XCTAssertFalse(outboxStatusEvent.isEmpty)
-                outboxStatusOnMutationEnqueued.fulfill()
+            if payload.eventName == HubPayload.EventName.DataStore.outboxMutationEnqueued {
+                guard let outboxStatusEvent = payload.data as? OutboxMutationEvent else {
+                    XCTFail("Failed to cast payload data as OutboxMutationEvent")
+                    return
+                }
+                XCTAssertEqual(outboxStatusEvent.modelName, "Post")
+                outboxMutationEnqueued.fulfill()
             }
         }
 

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/MutationQueue/OutgoingMutationQueueTestsWithMockStateMachine.swift

@@ -65,33 +65,58 @@ class OutgoingMutationQueueMockStateTest: XCTestCase {
         waitForExpectations(timeout: 1)
     }
 
-    func testRequestingEvent_subscriptionSetup() {
+    func testRequestingEvent_subscriptionSetup() throws {
         let semaphore = DispatchSemaphore(value: 0)
         stateMachine.pushExpectActionCriteria { action in
             XCTAssertEqual(action, OutgoingMutationQueue.Action.receivedSubscription)
             semaphore.signal()
         }
         stateMachine.state = .starting(apiBehavior, publisher)
         semaphore.wait()
+
+        let json = "{\"id\":\"1234\",\"title\":\"t\",\"content\":\"c\",\"createdAt\":\"2020-09-03T22:55:13.424Z\"}"
         let futureResult = MutationEvent(modelId: "1",
                                          modelName: "Post",
-                                         json: "{}",
+                                         json: json,
                                          mutationType: MutationEvent.MutationType.create)
         eventSource.pushMutationEvent(futureResult: .success(futureResult))
 
         let enqueueEvent = expectation(description: "state requestingEvent, enqueueEvent")
         let processEvent = expectation(description: "state requestingEvent, processedEvent")
+
         stateMachine.pushExpectActionCriteria { action in
             XCTAssertEqual(action, OutgoingMutationQueue.Action.enqueuedEvent)
             enqueueEvent.fulfill()
         }
+
+        let mutateAPICallExpecation = expectation(description: "Call to api category for mutate")
+        var listenerFromRequest: GraphQLOperation<MutationSync<AnyModel>>.ResultListener!
+        let responder = MutateRequestListenerResponder<MutationSync<AnyModel>> { _, eventListener in
+            mutateAPICallExpecation.fulfill()
+            listenerFromRequest = eventListener
+            return nil
+        }
+        apiBehavior.responders[.mutateRequestListener] = responder
+
+        stateMachine.state = .requestingEvent
+
+        wait(for: [enqueueEvent, mutateAPICallExpecation], timeout: 1)
+
         stateMachine.pushExpectActionCriteria { action in
             XCTAssertEqual(action, OutgoingMutationQueue.Action.processedEvent)
             processEvent.fulfill()
         }
-        stateMachine.state = .requestingEvent
 
-        waitForExpectations(timeout: 1)
+        let model = MockSynced(id: "id-1")
+        let anyModel = try model.eraseToAnyModel()
+        let remoteSyncMetadata = MutationSyncMetadata(id: model.id,
+                                                      deleted: false,
+                                                      lastChangedAt: Date().unixSeconds,
+                                                      version: 2)
+        let remoteMutationSync = MutationSync(model: anyModel, syncMetadata: remoteSyncMetadata)
+        listenerFromRequest(.success(.success(remoteMutationSync)))
+
+        wait(for: [processEvent], timeout: 1)
     }
 
     func testRequestingEvent_nosubscription() {