You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
LeStorage/LeStorage/ApiCallCollection.swift

478 lines
17 KiB

//
// ApiCallCollection.swift
// LeStorage
//
// Created by Laurent Morvillier on 17/06/2024.
//
import Foundation
protocol SomeCallCollection {
func findCallById(_ id: String) async -> (any SomeCall)?
func deleteById(_ id: String) async
func hasPendingCalls() async -> Bool
func contentOfFile() async -> String?
func reset() async
func resumeApiCalls() async
func type() async -> any Storable.Type
func resourceName() async -> String
}
enum ApiCallError: Error, LocalizedError {
case encodingError(id: String, type: String)
var errorDescription: String? {
switch self {
case .encodingError(let id, let type):
return "Can't encode instance \(type) with id: \(id)"
}
}
}
/// ApiCallCollection is an object communicating with a server to synchronize data managed locally
/// The Api calls are serialized and stored in a JSON file
/// Failing Api calls are stored forever and will be executed again later
actor ApiCallCollection<T: SyncedStorable>: SomeCallCollection {
fileprivate var storeCenter: StoreCenter
/// The list of api calls
fileprivate(set) var items: [ApiCall<T>] = []
/// The number of time an execution loop has been called
fileprivate var _attemptLoops: Int = 0
/// Indicates if the collection is currently retrying ApiCalls
fileprivate var _isExecutingCalls: Bool = false
fileprivate var _schedulingTask: Task<(), Never>? = nil
/// Indicates whether the collection content has changed
/// Initiates a write when true
fileprivate var _hasChanged: Bool = false {
didSet {
if self._hasChanged {
self._write()
self._hasChanged = false
}
}
}
init(storeCenter: StoreCenter) {
self.storeCenter = storeCenter
}
/// Starts the JSON file decoding synchronously or asynchronously
/// Reschedule Api calls if not empty
func loadFromFile() throws {
try self._decodeJSONFile()
}
/// Returns the file URL of the collection
fileprivate func _urlForJSONFile() throws -> URL {
return try self.storeCenter.jsonFileURL(for: ApiCall<T>.self)
}
/// Decodes the json file into the items array
fileprivate func _decodeJSONFile() throws {
let fileURL = try self._urlForJSONFile()
if FileManager.default.fileExists(atPath: fileURL.path()) {
let jsonString: String = try FileUtils.readFile(fileURL: fileURL)
do {
let decoded: [ApiCall<T>] = try jsonString.decodeArray() ?? []
self.items = decoded
} catch {
let decoded: [OldApiCall<T>] = try jsonString.decodeArray() ?? []
self.items = decoded.compactMap { $0.toNewApiCall() }
}
}
}
/// Writes the content of the data
fileprivate func _write() {
let fileName = ApiCall<T>.fileName()
DispatchQueue(label: "lestorage.queue.write", qos: .utility).asyncAndWait {
do {
let jsonString: String = try self.items.jsonString()
try self.storeCenter.write(content: jsonString, fileName: fileName)
} catch {
Logger.error(error)
}
}
}
/// Adds or update an API call instance
func addOrUpdate(_ instance: ApiCall<T>) {
if let index = self.items.firstIndex(where: { $0.id == instance.id }) {
self.items[index] = instance
} else {
self.items.append(instance)
}
self._hasChanged = true
}
/// Deletes an API call by [id]
func deleteById(_ id: String) {
self.items.removeAll(where: { $0.id == id })
// Logger.log("\(T.resourceName()) > Delete by id, count after deletion = \(self.items.count)")
self._hasChanged = true
}
/// Deletes a call by a data id
func deleteByDataId(_ dataId: String) {
if let apiCallIndex = self.items.firstIndex(where: { $0.data?.stringId == dataId }) {
self.items.remove(at: apiCallIndex)
self._hasChanged = true
}
}
/// Returns the Api call associated with the provided id
func findById(_ id: String) -> ApiCall<T>? {
return self.items.first(where: { $0.id == id })
}
/// Returns the Api call associated with the provided id
func findCallById(_ id: String) async -> (any SomeCall)? {
return self.findById(id)
}
/// Removes all objects in memory and deletes the JSON file
func reset() {
self._isExecutingCalls = false
self._schedulingTask?.cancel()
self.items.removeAll()
self._hasChanged = true
do {
let url: URL = try self._urlForJSONFile()
if FileManager.default.fileExists(atPath: url.path()) {
try FileManager.default.removeItem(at: url)
}
} catch {
Logger.error(error)
}
}
func resumeApiCalls() {
self._attemptLoops = -1
self.rescheduleApiCallsIfNecessary()
if self._schedulingTask != nil && self._attemptLoops > 2 {
self._schedulingTask?.cancel()
self._attemptLoops = -1
self.rescheduleApiCallsIfNecessary()
}
}
/// Reschedule API calls without waiting
func rescheduleImmediately() {
self._attemptLoops = -1
self.rescheduleApiCallsIfNecessary()
}
/// Reschedule API calls if necessary
func rescheduleApiCallsIfNecessary() {
if self.items.isNotEmpty && !self._isExecutingCalls {
self._schedulingTask = Task {
await self._waitAndExecuteApiCalls()
}
}
}
/// Reschedule the execution of API calls
fileprivate func _waitAndExecuteApiCalls() async {
// Logger.log("\(T.resourceName()) > RESCHED")
guard !self._isExecutingCalls, self.storeCenter.forceNoSynchronization == false else { return }
guard self.items.isNotEmpty else { return }
self._isExecutingCalls = true
self._attemptLoops += 1
await self._wait()
await self._batchExecution()
// Logger.log("\(T.resourceName()) > EXECUTE CALLS: \(self.items.count)")
// let batches = Dictionary(grouping: self.items, by: { $0.transactionId })
//
// for batch in batches.values {
// do {
// if batch.count == 1, let apiCall = batch.first, apiCall.method == .get {
// try await self._executeGetCall(apiCall: apiCall)
// } else {
// let results = try await self._executeApiCalls(batch)
// if T.copyServerResponse {
// let instances = results.compactMap { $0.data }
// StoreCenter.main.updateLocalInstances(instances)
// }
// }
// } catch {
// Logger.error(error)
// }
// }
// Logger.log("\(T.resourceName()) > EXECUTE CALLS ENDED !")
self._isExecutingCalls = false
if self.items.isNotEmpty {
await self._waitAndExecuteApiCalls()
}
// Logger.log("\(T.resourceName()) > isRescheduling = \(self._isRescheduling)")
}
fileprivate func _batchExecution() async {
let batches = Dictionary(grouping: self.items, by: { $0.transactionId })
for batch in batches.values {
do {
if batch.count == 1, let apiCall = batch.first, apiCall.method == .get {
try await self._executeGetCall(apiCall: apiCall)
} else {
let results: [OperationResult<T>] = try await self._executeApiCalls(batch)
if T.copyServerResponse {
let instances: [T] = results.compactMap { $0.data }
self.storeCenter.updateLocalInstances(instances)
}
}
} catch {
Logger.error(error)
}
}
}
@discardableResult func _executeGetCall(apiCall: ApiCall<T>) async throws -> Data {
let data = try await self.storeCenter.executeGet(apiCall: apiCall)
Logger.log("GET received = \(T.resourceName())")
if T.self == GetSyncData.self {
let syncData = try SyncData(data: data, storeCenter: self.storeCenter)
await self.storeCenter.synchronizeContent(syncData)
} else {
let results: [T] = try self._decode(data: data)
await self.storeCenter.itemsRetrieved(results, storeId: apiCall.storeId, clear: apiCall.option != .additive)
}
return data
}
fileprivate func _decode<V: Decodable>(data: Data) throws -> V {
if !(V.self is Empty?.Type || V.self is Empty.Type) {
return try JSON.decoder.decode(V.self, from: data)
} else {
return try JSON.decoder.decode(V.self, from: "{}".data(using: .utf8)!)
}
}
/// Wait for an exponentionnaly long time depending on the number of attemps
fileprivate func _wait() async {
guard self._attemptLoops > 0 else { return }
var seconds = self._attemptLoops
if self._attemptLoops > 5 {
let delay = pow(2, self._attemptLoops - 2) // starts at 16s
seconds = NSDecimalNumber(decimal: delay).intValue
}
Logger.log("\(T.resourceName()): wait for \(seconds) sec")
do {
try await Task.sleep(until: .now + .seconds(seconds))
} catch {
Logger.w("*** WAITING CRASHED !!!")
Logger.error(error)
}
}
// MARK: - Synchronization
/// Returns an APICall instance for the Storable [instance] and an HTTP [method]
/// The method makes some clean up when necessary:
/// - When deleting, we delete other calls as they are unecessary
/// - When updating, we delete other PUT as we don't want them to be executed in random orders
fileprivate func _prepareCall(instance: T, method: HTTPMethod, transactionId: String? = nil) {
// cleanup if necessary
switch method {
case .delete: // we don't want anything else than a DELETE in the queue
let existingCalls = self.items.filter { $0.data?.stringId == instance.stringId }
self._deleteCalls(existingCalls)
case .put: // we don't want mixed PUT calls so we delete the others
let existingPuts = self.items.filter { $0.data?.stringId == instance.stringId && $0.method == .put }
self._deleteCalls(existingPuts)
default:
break
}
let call: ApiCall<T> = self._createCall(method, instance: instance, transactionId: transactionId)
self._addCallToWaitingList(call)
}
/// deletes an array of ApiCall by id
fileprivate func _deleteCalls(_ calls: [ApiCall<T>]) {
for call in calls {
self.deleteById(call.id)
}
}
/// we want to avoid sending the same GET twice
fileprivate func _createGetCallIfNonExistent(_ parameters: [String : String]?, clear: Bool) -> ApiCall<T>? {
if let _ = self.items.first(where: { $0.method == .get && $0.urlParameters == parameters }) {
return nil
}
let option: CallOption? = !clear ? .additive : nil
let call = self._createCall(.get, instance: nil, option: option)
call.urlParameters = parameters
return call
}
/// Creates an API call for the Storable [instance] and an HTTP [method]
fileprivate func _createCall(_ method: HTTPMethod, instance: T?, transactionId: String? = nil, option: CallOption? = nil) -> ApiCall<T> {
if let instance {
return ApiCall(method: method, data: instance, transactionId: transactionId, option: option)
} else {
return ApiCall(method: .get, data: nil, option: option)
}
}
/// Prepares a call for execution by updating its properties and adding it to its collection for storage
fileprivate func _addCallToWaitingList(_ apiCall: ApiCall<T>) {
apiCall.lastAttemptDate = Date()
apiCall.attemptsCount += 1
self.addOrUpdate(apiCall)
}
/// Sends a GET request with an URLParameterConvertible [instance]
func sendGetRequest(instance: URLParameterConvertible) async throws {
let parameters = instance.queryParameters(storeCenter: self.storeCenter)
try await self._sendGetRequest(parameters: parameters)
}
/// Sends a GET request with an optional [storeId]
func sendGetRequest(storeId: String?, clear: Bool = true) async throws {
var parameters: [String : String]? = nil
if let storeId {
parameters = [Services.storeIdURLParameter : storeId]
}
try await self._sendGetRequest(parameters: parameters, clear: clear)
}
/// Sends an insert api call for the provided [instance]
fileprivate func _sendGetRequest(parameters: [String : String]?, clear: Bool = true) async throws {
if let getCall = self._createGetCallIfNonExistent(parameters, clear: clear) {
do {
try await self._prepareAndSendGetCall(getCall)
} catch {
self.rescheduleApiCallsIfNecessary()
Logger.error(error)
}
} else {
self.rescheduleImmediately()
}
}
/// Creates and execute the ApiCalls corresponding to the [batch]
func executeBatch(_ batch: OperationBatch<T>) {
self._prepareCalls(batch: batch)
self.rescheduleImmediately()
}
func singleBatchExecution(_ batch: OperationBatch<T>) async {
self._prepareCalls(batch: batch)
await self._batchExecution()
}
func executeSingleGet(instance: T) async throws -> Data where T : URLParameterConvertible {
let call = self._createCall(.get, instance: instance, option: .none)
call.urlParameters = instance.queryParameters(storeCenter: self.storeCenter)
self._addCallToWaitingList(call)
return try await self._executeGetCall(apiCall: call)
}
fileprivate func _prepareCalls(batch: OperationBatch<T>) {
let transactionId = Store.randomId()
for insert in batch.inserts {
self._prepareCall(instance: insert, method: .post, transactionId: transactionId)
}
for update in batch.updates {
self._prepareCall(instance: update, method: .put, transactionId: transactionId)
}
for delete in batch.deletes {
self._prepareCall(instance: delete, method: .delete, transactionId: transactionId)
}
}
/// Prepares and executes a GET call
fileprivate func _prepareAndSendGetCall(_ apiCall: ApiCall<T>) async throws {
self._addCallToWaitingList(apiCall)
try await self._executeGetCall(apiCall: apiCall)
}
/// Executes an API call
/// For POST requests, potentially copies additional data coming from the server during the insert
fileprivate func _executeApiCalls(_ apiCalls: [ApiCall<T>]) async throws -> [OperationResult<T>] {
let results = try await self.storeCenter.execute(apiCalls: apiCalls)
for result in results {
switch result.status {
case 200..<300:
self.deleteById(result.apiCallId)
default:
break
}
}
return results
}
/// Returns the content of the API call file as a String
func contentOfFile() -> String? {
guard let fileURL = try? self._urlForJSONFile() else { return nil }
if FileManager.default.fileExists(atPath: fileURL.path()) {
return try? FileUtils.readFile(fileURL: fileURL)
}
return nil
}
/// Returns if the API call collection is not empty
func hasPendingCalls() -> Bool {
// print("\(T.resourceName()) calls = \(self.items.count)")
return self.items.isNotEmpty
}
/// returns the list of API calls in the collection
func apiCalls() -> [ApiCall<T>] {
return self.items
}
func type() async -> any Storable.Type { return T.self }
func resourceName() async -> String { return T.resourceName() }
// MARK: - Testing
func sendInsertion(_ instance: T) async throws {
let batch = OperationBatch<T>()
batch.addInsert(instance)
self.executeBatch(batch)
}
func sendUpdate(_ instance: T) async throws {
let batch = OperationBatch<T>()
batch.addUpdate(instance)
self.executeBatch(batch)
}
func sendDeletion(_ instance: T) async throws {
let batch = OperationBatch<T>()
batch.addDelete(instance)
self.executeBatch(batch)
}
}