feat: run background queue tasks

Sources/MessagingPush/MessagingPushImplementation.swift

@@ -78,10 +78,6 @@ internal class MessagingPushImplementation: MessagingPushInstance {
      Delete the currently registered device token
      */
     public func deleteDeviceToken(onComplete: @escaping (Result<Void, CustomerIOError>) -> Void) {
-        guard let bodyData = jsonAdapter.toJson(DeleteDeviceRequest()) else {
-            return onComplete(.failure(.http(.noRequestMade(nil))))
-        }
-
         guard let deviceToken = self.deviceToken else {
             // no device token, delete has already happened or is not needed
             return onComplete(Result.success(()))
@@ -96,7 +92,7 @@ internal class MessagingPushImplementation: MessagingPushInstance {
         let httpRequestParameters =
             HttpRequestParams(endpoint: .deleteDevice(identifier: identifier,
                                                       deviceToken: deviceToken),
-                              headers: nil, body: bodyData)
+                              headers: nil, body: nil)
 
         httpClient
             .request(httpRequestParameters) { [weak self] result in

Sources/Tracking/Background Queue/Queue.swift

@@ -1,20 +1,132 @@
 import Foundation
 
-public protocol Queue {
-    func addTask(type: QueueTaskType, data: Data) -> (success: Bool, queueStatus: QueueStatus)
+/**
+ A background queue to perform actions in the background (probably network requests).
+ A queue exists to make our public facing SDK functions synchronous:
+ `CustomerIO.shared.trackEvent("foo")` where we perform the API call sometime in the future
+ and handle errors/retry so customers don't have to.
+
+ The queue is designed with the main purpose of performing API calls. Let's show an example of
+ how it's recommended you use the queue and how *not* to use the queue.
+
+ Let's say you're identifying a profile.
+ ```
+ // recommended use of the background queue
+ func identifyProfile(identifier: String, data: Encodable) {
+    // perform all operations here, first before touching the queue
+    keyValueStorage.save(identifier)
+
+    if deviceTokenSavedToDifferentProfile {
+       // it's OK to add tasks to the queue here before we identify the new profile
+       queue.addTask(.deleteDeviceToken, identifier: oldProfileIdentifier)
+       keyValueStorage.delete(deviceToken)
+    }
+
+    // then, add a background queue task
+    queue.addTask(.identifyProfile, identifier: identifier)
+ }
+
+ // then later on in the code, the background queue task runs
+ // *all* of our logic for identifying a new profile.
+ func runQueueTask() {
+     httpClient.identifyProfile(newProfileIdentifier
+ }
+ ```
+
+ Not recommended way of using the background queue:
+ ```
+ func identifyProfile(identifier: String, data: Encodable) {
+    queue.addTask(.identifyProfile, identifier: identifier, oldProfileIdentifier)
+ }
+
+ // then later on in the code, the background queue task runs
+ // *all* of our logic for identifying a new profile.
+ func runQueueTask() {
+     let newProfileIdentifier = ...
+
+     keyValueStorage.save(identifier)
+
+     if deviceTokenSavedToDifferentProfile {
+        httpClient.delete(deviceTokenFromOldProfile)
+
+        keyValueStorage.delete(deviceToken)
+     }
+
+     httpClient.identifyProfile(newProfileIdentifier
+ }
+ ```
+ */
+public protocol Queue: AutoMockable {
+    /**
+     Add a task to the queue to be performed sometime in the future.
+
+     `data` - Probably a struct that contains "a snapshot" of the data needed to perform the
+     background task (probably a network request).
+     */
+    func addTask<TaskData: Codable>(
+        type: QueueTaskType,
+        // sourcery:Type=AnyEncodable
+        // sourcery:TypeCast="AnyEncodable(data)"
+        data: TaskData
+    ) -> (success: Bool, queueStatus: QueueStatus)
+    func run(onComplete: @escaping () -> Void)
 }
 
 // sourcery: InjectRegister = "Queue"
 public class CioQueue: Queue {
     private let storage: QueueStorage
     private let siteId: SiteId
+    private let runRequest: QueueRunRequest
+    private let jsonAdapter: JsonAdapter
+    private let logger: Logger
+    private let sdkConfigStore: SdkConfigStore
 
-    init(siteId: SiteId, storage: QueueStorage) {
+    init(
+        siteId: SiteId,
+        storage: QueueStorage,
+        runRequest: QueueRunRequest,
+        jsonAdapter: JsonAdapter,
+        logger: Logger,
+        sdkConfigStore: SdkConfigStore
+    ) {
         self.storage = storage
         self.siteId = siteId
+        self.runRequest = runRequest
+        self.jsonAdapter = jsonAdapter
+        self.logger = logger
+        self.sdkConfigStore = sdkConfigStore
+    }
+
+    public func addTask<T: Codable>(type: QueueTaskType, data: T) -> (success: Bool, queueStatus: QueueStatus) {
+        guard let data = jsonAdapter.toJson(data, encoder: nil) else {
+            return (success: false,
+                    queueStatus: QueueStatus(queueId: siteId, numTasksInQueue: storage.getInventory().count))
+        }
+
+        let addTaskResult = storage.create(type: type, data: data)
+        processQueueStatus(addTaskResult.queueStatus)
+
+        return addTaskResult
     }
 
-    public func addTask(type: QueueTaskType, data: Data) -> (success: Bool, queueStatus: QueueStatus) {
-        return storage.create(type: type, data: data)
+    public func run(onComplete: @escaping () -> Void) {
+        logger.verbose("Manually running background queue")
+
+        runRequest.start(onComplete: onComplete)
+    }
+
+    private func processQueueStatus(_ status: QueueStatus) {
+        logger.verbose("Processing queue status \(status).")
+        let isManyTasksInQueue = status.numTasksInQueue >= sdkConfigStore.config.backgroundQueueMinNumberOfTasks
+
+        let runQueue = isManyTasksInQueue
+
+        if runQueue {
+            logger.verbose("Automatically running background queue")
+
+            runRequest.start { [weak self] in
+                self?.logger.verbose("Automatic running background queue completed")
+            }
+        }
     }
 }

Sources/Tracking/Background Queue/QueueRequestManager.swift

@@ -0,0 +1,49 @@
+import Foundation
+
+/**
+ Simple singleton that helps assert that the background queue is only running
+ one run request at one time (1 run request per site id since each site id can have it's own background queue).
+
+ When the background queue wants to run it's tasks, it's important that the queue only have 1 concurrent runner
+ running at one time to prevent race conditions and tasks running multiple times.
+ */
+public protocol QueueRequestManager: AutoMockable {
+    /// call when a runner run request is complete
+    func requestComplete()
+    /// call when a new run request is requested. adds callback to list of callbacks
+    /// to call when run request is done running.
+    /// returns is an existing run request is currently running or not.
+    func startRequest(onComplete: @escaping () -> Void) -> Bool
+}
+
+// sourcery: InjectRegister = "QueueRequestManager"
+// sourcery: InjectSingleton
+public class CioQueueRequestManager: QueueRequestManager {
+    @Atomic internal var isRunningRequest = false
+    @Atomic internal var callbacks: [() -> Void] = []
+
+    public func requestComplete() {
+        let existingCallbacks = callbacks
+
+        callbacks = []
+        isRunningRequest = false
+
+        existingCallbacks.forEach { callback in
+            callback()
+        }
+    }
+
+    public func startRequest(onComplete: @escaping () -> Void) -> Bool {
+        let isQueueRunningARequest = isRunningRequest
+
+        callbacks.append(onComplete)
+
+        if !isQueueRunningARequest {
+            isRunningRequest = true
+        }
+
+        // return the isRunningRequest value before modification or we will
+        // *always* return true (since we modify to true or ignore)
+        return isQueueRunningARequest
+    }
+}

Sources/Tracking/Background Queue/QueueRunRequest.swift

@@ -0,0 +1,93 @@
+import Foundation
+
+public protocol QueueRunRequest: AutoMockable {
+    func start(onComplete: @escaping () -> Void)
+}
+
+// sourcery: InjectRegister = "QueueRunRequest"
+public class CioQueueRunRequest: QueueRunRequest {
+    private let runner: QueueRunner
+    private let storage: QueueStorage
+    private let requestManger: QueueRequestManager
+    private let logger: Logger
+
+    init(runner: QueueRunner, storage: QueueStorage, requestManger: QueueRequestManager, logger: Logger) {
+        self.runner = runner
+        self.storage = storage
+        self.requestManger = requestManger
+        self.logger = logger
+    }
+
+    public func start(onComplete: @escaping () -> Void) {
+        let isRequestCurrentlyRunning = requestManger.startRequest(onComplete: onComplete)
+
+        if !isRequestCurrentlyRunning {
+            startNewRequestRun()
+        }
+    }
+
+    private func startNewRequestRun() {
+        let inventory = storage.getInventory()
+
+        runTasks(query: inventory)
+    }
+
+    private func runTasks(query: [QueueTaskMetadata]) {
+        let goToNextTask: () -> Void = {
+            var newQuery = query
+            newQuery.removeFirst()
+            self.runTasks(query: newQuery)
+        }
+
+        if query.isEmpty { // we hit the end of the current inventory. Done!
+            logger.verbose("background queue out of tasks.")
+
+            requestManger.requestComplete()
+
+            return
+        }
+
+        let nextTaskToRunInventoryItem = query[0]
+        let nextTaskStorageId = nextTaskToRunInventoryItem.taskPersistedId
+        guard let nextTaskToRun = storage.get(storageId: nextTaskStorageId) else {
+            // delete task from inventory since we can't find it in storage so we don't want to run it.
+            // ignore result because if it's successful or not, all we can do is try and delete and move on.
+            _ = storage.delete(storageId: nextTaskStorageId)
+
+            // log error. this scenario shouldn't happen where task can't be found.
+            logger.error("Tried to get queue task with storage id: \(nextTaskStorageId), but storage couldn't find it.")
+
+            return goToNextTask()
+        }
+
+        logger
+            .verbose("background queue next task \(nextTaskStorageId). query tasks remaining: \(query.count)")
+        logger.debug("next background queue task to run: \(nextTaskToRunInventoryItem) => \(nextTaskToRun)")
+
+        // we are not using [weak self] because if the task is currently running,
+        // we dont want the result handler to get garbage collected which could
+        // make the task run again when it shouldn't.
+        //
+        // if we wanted to use [weak self] then we should allow running a task to cancel
+        // while executing which would then allow this to use [weak self].
+        runner.runTask(nextTaskToRun) { result in
+            switch result {
+            case .success:
+                self.logger.verbose("background queue task \(nextTaskStorageId) success")
+
+                _ = self.storage.delete(storageId: nextTaskToRunInventoryItem.taskPersistedId)
+            case .failure(let error):
+                self.logger.verbose("background queue task \(nextTaskStorageId) fail - \(error.localizedDescription)")
+
+                let executedTaskPreviousRunResults = nextTaskToRun.runResults
+                let newRunResults = executedTaskPreviousRunResults
+                    .totalRunsSet(executedTaskPreviousRunResults.totalRuns + 1)
+
+                _ = self.storage.update(storageId: nextTaskToRunInventoryItem.taskPersistedId,
+                                        runResults: newRunResults)
+            }
+
+            return goToNextTask()
+        }
+    }
+}

Sources/Tracking/Background Queue/QueueRunner.swift

@@ -0,0 +1,77 @@
+import Foundation
+
+/**
+ Where queue tasks are executed asynchronously.
+
+ To keep this class testable, try to keep it small. So, the class's job is to take a
+ task type and generic `Data` for task data and call some other code to perform the
+ actual work on executing the task.
+ */
+public protocol QueueRunner: AutoMockable {
+    func runTask(_ task: QueueTask, onComplete: @escaping (Result<Void, CustomerIOError>) -> Void)
+}
+
+// sourcery: InjectRegister = "QueueRunner"
+public class CioQueueRunner: QueueRunner {
+    private let jsonAdapter: JsonAdapter
+    private let siteId: SiteId
+    private let logger: Logger
+    private let httpClient: HttpClient
+
+    init(siteId: SiteId, jsonAdapter: JsonAdapter, logger: Logger, httpClient: HttpClient) {
+        self.siteId = siteId
+        self.jsonAdapter = jsonAdapter
+        self.logger = logger
+        self.httpClient = httpClient
+    }
+
+    public func runTask(_ task: QueueTask, onComplete: @escaping (Result<Void, CustomerIOError>) -> Void) {
+        let failureIfDontDecodeTaskData: Result<Void, CustomerIOError> = .failure(.http(.noRequestMade(nil)))
+
+        switch task.type {
+        case .identifyProfile:
+            guard let taskData = getTaskData(task, type: IdentifyProfileQueueTaskData.self) else {
+                return onComplete(failureIfDontDecodeTaskData)
+            }
+
+            let httpParams = HttpRequestParams(endpoint: .identifyCustomer(identifier: taskData.identifier),
+                                               headers: nil, body: taskData.attributesJsonString?.data)
+
+            performHttpRequest(params: httpParams, onComplete: onComplete)
+        case .trackEvent:
+            guard let taskData = getTaskData(task, type: TrackEventQueueTaskData.self) else {
+                return onComplete(failureIfDontDecodeTaskData)
+            }
+
+            let httpParams = HttpRequestParams(endpoint: .trackCustomerEvent(identifier: taskData.identifier),
+                                               headers: nil, body: taskData.attributesJsonString.data)
+
+            performHttpRequest(params: httpParams, onComplete: onComplete)
+        }
+    }
+
+    /// (1) less code for `runTask` function to decode JSON and (2) one place to do error logging if decoding wrong.
+    private func getTaskData<T: Decodable>(_ task: QueueTask, type: T.Type) -> T? {
+        let taskData: T? = jsonAdapter.fromJson(task.data, decoder: nil)
+
+        if taskData == nil {
+            /// log as error because it's a developer error since SDK is who encoded the TaskData in the first place
+            /// we should always be able to decode it without problem.
+            logger.error("Failure decoding: \(task.data.string ?? "()") to \(type)")
+        }
+
+        return taskData
+    }
+
+    private func performHttpRequest(
+        params: HttpRequestParams,
+        onComplete: @escaping (Result<Void, CustomerIOError>) -> Void
+    ) {
+        httpClient.request(params) { result in
+            switch result {
+            case .success: onComplete(.success(()))
+            case .failure(let httpError): onComplete(.failure(.http(httpError)))
+            }
+        }
+    }
+}