feat: support event callback and fix missing insert_id

Sources/Amplitude/Amplitude.swift

@@ -35,7 +35,7 @@ public class Amplitude {
     }
 
     @discardableResult
-    public func track(event: BaseEvent, options: EventOptions? = nil, callback: EventCallBack? = nil) -> Amplitude {
+    public func track(event: BaseEvent, options: EventOptions? = nil, callback: EventCallback? = nil) -> Amplitude {
         if options != nil {
             event.mergeEventOptions(eventOptions: options!)
         }

Sources/Amplitude/Configuration.swift

@@ -18,7 +18,7 @@ public class Configuration {
     var loggerProvider: any Logger
     var minIdLength: Int?
     var partnerId: String?
-    var callback: EventCallBack?
+    var callback: EventCallback?
     var flushMaxRetries: Int
     var useBatch: Bool
     var serverZone: ServerZone
@@ -43,7 +43,7 @@ public class Configuration {
         loggerProvider: any Logger = ConsoleLogger(),
         minIdLength: Int? = nil,
         partnerId: String? = nil,
-        callback: EventCallBack? = nil,
+        callback: EventCallback? = nil,
         flushMaxRetries: Int = Constants.Configuration.FLUSH_MAX_RETRIES,
         useBatch: Bool = false,
         serverZone: ServerZone = ServerZone.US,

Sources/Amplitude/Events/BaseEvent.swift

@@ -25,6 +25,7 @@ public class BaseEvent: EventOptions, Codable {
         case timestamp = "time"
         case eventId = "event_id"
         case sessionId = "session_id"
+        case insertId = "insert_id"
         case locationLat = "location_lat"
         case locationLng = "location_lng"
         case appVersion = "app_version"
@@ -92,7 +93,7 @@ public class BaseEvent: EventOptions, Codable {
         productId: String? = nil,
         revenueType: String? = nil,
         extra: [String: Any]? = nil,
-        callback: EventCallBack? = nil,
+        callback: EventCallback? = nil,
         partnerId: String? = nil,
         eventType: String,
         eventProperties: [String: Any]? = nil,
@@ -203,6 +204,7 @@ public class BaseEvent: EventOptions, Codable {
         timestamp = try values.decodeIfPresent(Int64.self, forKey: .timestamp)
         eventId = try values.decodeIfPresent(Int64.self, forKey: .eventId)
         sessionId = try values.decodeIfPresent(Int64.self, forKey: .sessionId)
+        insertId = try values.decodeIfPresent(String.self, forKey: .insertId)
         locationLat = try values.decodeIfPresent(Double.self, forKey: .locationLat)
         locationLng = try values.decodeIfPresent(Double.self, forKey: .locationLng)
         appVersion = try values.decodeIfPresent(String.self, forKey: .appVersion)
@@ -246,6 +248,7 @@ public class BaseEvent: EventOptions, Codable {
         try container.encode(timestamp, forKey: .timestamp)
         try container.encode(eventId, forKey: .eventId)
         try container.encode(sessionId, forKey: .sessionId)
+        try container.encode(insertId, forKey: .insertId)
         try container.encode(locationLat, forKey: .locationLat)
         try container.encode(locationLng, forKey: .locationLng)
         try container.encode(appVersion, forKey: .appVersion)

Sources/Amplitude/Events/EventOptions.swift

@@ -43,7 +43,7 @@ public class EventOptions {
     var productId: String?
     var revenueType: String?
     var extra: [String: Any]?
-    var callback: EventCallBack?
+    var callback: EventCallback?
     var partnerId: String?
     internal var attempts: Int
 
@@ -83,7 +83,7 @@ public class EventOptions {
         productId: String? = nil,
         revenueType: String? = nil,
         extra: [String: Any]? = nil,
-        callback: EventCallBack? = nil,
+        callback: EventCallback? = nil,
         partnerId: String? = nil,
         attempts: Int = 0
     ) {

Sources/Amplitude/Storages/PersistentStorage.swift

@@ -16,12 +16,16 @@ class PersistentStorage: Storage {
     private var outputStream: OutputFileStream?
     internal weak var amplitude: Amplitude?
 
+    // Store event.callback in memory as it cannot be ser/deser in files.
+    private var eventCallbackMap: [String: EventCallback]
+
     let syncQueue = DispatchQueue(label: "syncPersistentStorage.amplitude.com")
 
     init(apiKey: String = "") {
         self.storagePrefix = "\(PersistentStorage.DEFAULT_STORAGE_PREFIX)-\(apiKey)"
         self.userDefaults = UserDefaults(suiteName: "\(PersistentStorage.AMP_STORAGE_PREFIX).\(storagePrefix)")
         self.fileManager = FileManager.default
+        self.eventCallbackMap = [String: EventCallback]()
     }
 
     func write(key: StorageKey, value: Any?) throws {
@@ -31,6 +35,9 @@ class PersistentStorage: Storage {
                 if let event = value as? BaseEvent {
                     let eventStoreFile = getCurrentFile()
                     self.storeEvent(toFile: eventStoreFile, event: event)
+                    if let eventCallback = event.callback, let eventInsertId = event.insertId {
+                        eventCallbackMap[eventInsertId] = eventCallback
+                    }
                 }
             default:
                 if isBasicType(value: value) {
@@ -97,7 +104,7 @@ class PersistentStorage: Storage {
         eventBlock: EventBlock,
         eventsString: String
     ) -> ResponseHandler {
-        return PersitentStorageResponseHandler(
+        return PersistentStorageResponseHandler(
             configuration: configuration,
             storage: self,
             eventPipeline: eventPipeline,
@@ -134,6 +141,14 @@ class PersistentStorage: Storage {
         }
     }
 
+    func getEventCallback(insertId: String) -> EventCallback? {
+        return eventCallbackMap[insertId]
+    }
+
+    func removeEventCallback(insertId: String) {
+        eventCallbackMap.removeValue(forKey: insertId)
+    }
+
     func isBasicType(value: Any?) -> Bool {
         var result = false
         if value == nil {

Sources/Amplitude/Types.swift

@@ -19,7 +19,7 @@ public struct IngestionMetadata: Codable {
     var sourceVersion: String?
 }
 
-public typealias EventCallBack = (BaseEvent, Int, String) -> Void
+public typealias EventCallback = (BaseEvent, Int, String) -> Void
 
 // Swift 5.7 supports any existential type.
 // The type of EventBlock has to be determined pre-runtime.

Sources/Amplitude/Utilities/PersitentStorageResponseHandler.swift → Sources/Amplitude/Utilities/PersistentStorageResponseHandler.swift

@@ -1,13 +1,13 @@
 //
-//  PersitentStorageResponseHandler.swift
+//  PersistentStorageResponseHandler.swift
 //
 //
 //  Created by Marvin Liu on 11/30/22.
 //
 
 import Foundation
 
-class PersitentStorageResponseHandler: ResponseHandler {
+class PersistentStorageResponseHandler: ResponseHandler {
     var configuration: Configuration
     var storage: PersistentStorage
     var eventPipeline: EventPipeline
@@ -29,21 +29,25 @@ class PersitentStorageResponseHandler: ResponseHandler {
     }
 
     func handleSuccessResponse(code: Int) {
-        if let events = BaseEvent.fromArrayString(jsonString: eventsString) {
-            triggerEventsCallBack(events: events, code: code, message: "Successfully send event")
+        guard let events = BaseEvent.fromArrayString(jsonString: eventsString) else {
+            storage.remove(eventBlock: eventBlock)
+            removeEventCallbackByEventsString(eventsString: eventsString)
+            return
         }
+        triggerEventsCallback(events: events, code: code, message: "Successfully send event")
         storage.remove(eventBlock: eventBlock)
     }
 
     func handleBadRequestResponse(data: [String: Any]) {
         guard let events = BaseEvent.fromArrayString(jsonString: eventsString) else {
             storage.remove(eventBlock: eventBlock)
+            removeEventCallbackByEventsString(eventsString: eventsString)
             return
         }
 
         if events.count == 1 {
             let error = data["error"] as? String ?? ""
-            triggerEventsCallBack(
+            triggerEventsCallback(
                 events: events,
                 code: HttpClient.HttpStatus.BAD_REQUEST.rawValue,
                 message: error
@@ -78,7 +82,7 @@ class PersitentStorageResponseHandler: ResponseHandler {
         }
 
         let error = data["error"] as? String ?? ""
-        triggerEventsCallBack(events: eventsToDrop, code: HttpClient.HttpStatus.BAD_REQUEST.rawValue, message: error)
+        triggerEventsCallback(events: eventsToDrop, code: HttpClient.HttpStatus.BAD_REQUEST.rawValue, message: error)
 
         eventsToRetry.forEach { event in
             eventPipeline.put(event: event)
@@ -90,11 +94,12 @@ class PersitentStorageResponseHandler: ResponseHandler {
     func handlePayloadTooLargeResponse(data: [String: Any]) {
         guard let events = BaseEvent.fromArrayString(jsonString: eventsString) else {
             storage.remove(eventBlock: eventBlock)
+            removeEventCallbackByEventsString(eventsString: eventsString)
             return
         }
         if events.count == 1 {
             let error = data["error"] as? String ?? ""
-            triggerEventsCallBack(
+            triggerEventsCallback(
                 events: events,
                 code: HttpClient.HttpStatus.PAYLOAD_TOO_LARGE.rawValue,
                 message: error
@@ -150,12 +155,31 @@ class PersitentStorageResponseHandler: ResponseHandler {
     }
 }
 
-extension PersitentStorageResponseHandler {
-    private func triggerEventsCallBack(events: [BaseEvent], code: Int, message: String) {
+extension PersistentStorageResponseHandler {
+    private func triggerEventsCallback(events: [BaseEvent], code: Int, message: String) {
         events.forEach { event in
             configuration.callback?(event, code, message)
-            // TODO: discuss whether to add event.callback support in here and storage for each individual event
-            // The map store event callbacks has to be in-memory, might be erased or cause memory leak issue
+            if let eventInsertId = event.insertId, let eventCallback = storage.getEventCallback(insertId: eventInsertId)
+            {
+                eventCallback(event, code, message)
+                storage.removeEventCallback(insertId: eventInsertId)
+            }
+        }
+    }
+
+    func removeEventCallbackByEventsString(eventsString: String) {
+        guard let regex = try? NSRegularExpression(pattern: #"\"insert_id\":\"(.{36})\","#) else {
+            return
+        }
+        let eventsNSString = NSString(string: eventsString)
+        regex.matches(in: eventsString, options: [], range: NSRange(location: 0, length: eventsNSString.length)).forEach
+        { match in
+            (1..<match.numberOfRanges).forEach {
+                if match.range(at: $0).location != NSNotFound {
+                    let eventInsertId = eventsNSString.substring(with: match.range(at: $0))
+                    storage.removeEventCallback(insertId: eventInsertId)
+                }
+            }
         }
     }
 }

Tests/AmplitudeTests/Supports/TestUtilities.swift

@@ -143,3 +143,16 @@ class FakeResponseHandler: ResponseHandler {
     func handleFailedResponse(data: [String: Any]) {
     }
 }
+
+class FakePersistentStorage: PersistentStorage {
+    // Array to store the method invocation history for testing verification purpose
+    var haveBeenCalledWith = [String]()
+
+    override func removeEventCallback(insertId: String) {
+        haveBeenCalledWith.append("removeEventCallback(insertId: \(insertId))")
+    }
+
+    override func remove(eventBlock: EventBlock) {
+        haveBeenCalledWith.append("remove(eventBlock: \(eventBlock.absoluteURL))")
+    }
+}

Tests/AmplitudeTests/Utilities/PersistentStorageResponseHandlerTests.swift

@@ -0,0 +1,130 @@
+//
+//  PersistentStorageResponseHandlerTests.swift
+//
+//
+//  Created by Marvin Liu on 12/2/22.
+//
+
+import XCTest
+
+@testable import Amplitude_Swift
+
+final class PersistentStorageResponseHandlerTests: XCTestCase {
+    private var configuration: Configuration!
+    private var amplitude: Amplitude!
+    private var storage: PersistentStorage!
+    private var eventPipeline: EventPipeline!
+    private var eventBlock: URL!
+    private var eventsString: String!
+
+    override func setUp() {
+        super.setUp()
+        configuration = Configuration(apiKey: "testApiKey")
+        amplitude = Amplitude(configuration: configuration)
+        storage = PersistentStorage(apiKey: "testApiKey")
+        eventPipeline = EventPipeline(amplitude: amplitude)
+        eventBlock = URL(string: "test")
+    }
+
+    func testInit() {
+        eventsString = """
+            [
+                {"event_type": "test"}
+            ]
+            """
+        let handler = PersistentStorageResponseHandler(
+            configuration: configuration,
+            storage: storage,
+            eventPipeline: eventPipeline,
+            eventBlock: eventBlock,
+            eventsString: eventsString
+        )
+
+        XCTAssertEqual(handler.eventsString, eventsString)
+    }
+
+    func testRemoveEventCallbackByEventsString_callsRemoveEventCallback() {
+        let eventsString = """
+            [
+              {"event_type":"test","insert_id":"e3e4488d-6877-4775-ae88-344df7ccd5d8","user_id":"test-user"},
+              {"event_type":"test","insert_id":"c8d58999-7539-4184-8a7d-54302697baf0","user_id":"test-user"}
+            ]
+            """
+        let fakePersistentStorage = FakePersistentStorage(apiKey: "testApiKey")
+        let handler = PersistentStorageResponseHandler(
+            configuration: configuration,
+            storage: fakePersistentStorage,
+            eventPipeline: eventPipeline,
+            eventBlock: eventBlock,
+            eventsString: eventsString
+        )
+
+        handler.removeEventCallbackByEventsString(eventsString: eventsString)
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith.count,
+            2
+        )
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith[0],
+            "removeEventCallback(insertId: e3e4488d-6877-4775-ae88-344df7ccd5d8)"
+        )
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith[1],
+            "removeEventCallback(insertId: c8d58999-7539-4184-8a7d-54302697baf0)"
+        )
+    }
+
+    func testRemoveEventCallbackByEventsString_notCallRemoveEventCallback() {
+        // insert_id format, missing insert_id
+        let eventsString = """
+            [
+              {"event_type":"wrong-insert_id-format","insert_id":"6877-4775-ae88-344df7ccd5d8","user_id":"test-user"},
+              {"event_type":"missing-insert_id","user_id":"test-user"}
+            ]
+            """
+
+        let fakePersistentStorage = FakePersistentStorage(apiKey: "testApiKey")
+        let handler = PersistentStorageResponseHandler(
+            configuration: configuration,
+            storage: fakePersistentStorage,
+            eventPipeline: eventPipeline,
+            eventBlock: eventBlock,
+            eventsString: eventsString
+        )
+
+        handler.removeEventCallbackByEventsString(eventsString: eventsString)
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith.count,
+            0
+        )
+    }
+
+    func testHandleSuccessResponseWithInvalidEventsString_removesEventBlockAndEventCallback() {
+        // valid event, invalid event
+        let eventsString = """
+            [
+              {"event_type":"valid-event","insert_id":"e3e4488d-6877-4775-ae88-344df7ccd5d8","user_id":"test-user"},
+              {"event_type":"invalid-event",user_id:test-user,xxx}
+            ]
+            """
+
+        let fakePersistentStorage = FakePersistentStorage(apiKey: "testApiKey")
+        let handler = PersistentStorageResponseHandler(
+            configuration: configuration,
+            storage: fakePersistentStorage,
+            eventPipeline: eventPipeline,
+            eventBlock: eventBlock,
+            eventsString: eventsString
+        )
+
+        handler.handleSuccessResponse(code: 200)
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith[0],
+            "remove(eventBlock: \(eventBlock.absoluteURL))"
+        )
+        XCTAssertEqual(
+            fakePersistentStorage.haveBeenCalledWith[1],
+            "removeEventCallback(insertId: e3e4488d-6877-4775-ae88-344df7ccd5d8)"
+        )
+    }
+}