Capture Destination Metadata (#116)

* capture destination metadata

* add writeKey to batch files and switch to /b endpoint

Author: prayansh

Segment.xcodeproj/project.pbxproj

@@ -48,6 +48,7 @@
 		46FE4CE025A53FAD003A7362 /* Storage.swift in Sources */ = {isa = PBXBuildFile; fileRef = 46FE4CDF25A53FAD003A7362 /* Storage.swift */; };
 		46FE4CFB25A6C671003A7362 /* TestUtilities.swift in Sources */ = {isa = PBXBuildFile; fileRef = 46FE4CFA25A6C671003A7362 /* TestUtilities.swift */; };
 		46FE4D1D25A7A850003A7362 /* Storage_Tests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 46FE4D1C25A7A850003A7362 /* Storage_Tests.swift */; };
+		759D6CD127B48ABB00AB900A /* DestinationMetadataPlugin.swift in Sources */ = {isa = PBXBuildFile; fileRef = 759D6CD027B48ABB00AB900A /* DestinationMetadataPlugin.swift */; };
 		9620862C2575C0C800314F8D /* Events.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9620862B2575C0C800314F8D /* Events.swift */; };
 		96208650257AA83E00314F8D /* iOSLifecycleMonitor.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9620864F257AA83E00314F8D /* iOSLifecycleMonitor.swift */; };
 		96259F8326CEF526008AE301 /* LogTarget.swift in Sources */ = {isa = PBXBuildFile; fileRef = 96259F8226CEF526008AE301 /* LogTarget.swift */; };
@@ -147,6 +148,7 @@
 		46FE4CDF25A53FAD003A7362 /* Storage.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Storage.swift; sourceTree = "<group>"; };
 		46FE4CFA25A6C671003A7362 /* TestUtilities.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TestUtilities.swift; sourceTree = "<group>"; };
 		46FE4D1C25A7A850003A7362 /* Storage_Tests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Storage_Tests.swift; sourceTree = "<group>"; };
+		759D6CD027B48ABB00AB900A /* DestinationMetadataPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DestinationMetadataPlugin.swift; sourceTree = "<group>"; };
 		9620862B2575C0C800314F8D /* Events.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Events.swift; sourceTree = "<group>"; };
 		962086482579CCC200314F8D /* Cocoa.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = Cocoa.framework; path = System/Library/Frameworks/Cocoa.framework; sourceTree = SDKROOT; };
 		9620864F257AA83E00314F8D /* iOSLifecycleMonitor.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = iOSLifecycleMonitor.swift; sourceTree = "<group>"; };
@@ -293,6 +295,7 @@
 				9692726725A583A6009B5298 /* SegmentDestination.swift */,
 				46210835260BBEE400EBC4A8 /* DeviceToken.swift */,
 				46031D64266E7C10009BA540 /* StartupQueue.swift */,
+				759D6CD027B48ABB00AB900A /* DestinationMetadataPlugin.swift */,
 			);
 			path = Plugins;
 			sourceTree = "<group>";
@@ -562,6 +565,7 @@
 				460227422612987300A9E913 /* watchOSLifecycleEvents.swift in Sources */,
 				96259F8326CEF526008AE301 /* LogTarget.swift in Sources */,
 				46F7485E26C718710042798E /* ObjCConfiguration.swift in Sources */,
+				759D6CD127B48ABB00AB900A /* DestinationMetadataPlugin.swift in Sources */,
 				A31A162F2576B73F00C9CDDF /* State.swift in Sources */,
 				9692726825A583A6009B5298 /* SegmentDestination.swift in Sources */,
 				4602276C261E7BF900A9E913 /* iOSDelegation.swift in Sources */,

Sources/Segment/Plugins/DestinationMetadataPlugin.swift

@@ -0,0 +1,59 @@
+//
+//  DestinationMetadataPlugin.swift
+//  Segment
+//
+//  Created by Prayansh Srivastava on 2/9/22.
+//
+
+import Foundation
+
+/**
+ * DestinationMetadataPlugin adds `_metadata` information to payloads that Segment uses to
+ * determine delivery of events to cloud/device-mode destinations
+ */
+public class DestinationMetadataPlugin: Plugin {
+    public let type: PluginType = PluginType.enrichment
+    public var analytics: Analytics?
+    private var analyticsSettings: Settings? = nil
+    
+    public func update(settings: Settings, type: UpdateType) {
+        analyticsSettings = settings
+    }
+    
+    public func execute<T: RawEvent>(event: T?) -> T? {
+        guard var modified = event else {
+            return event
+        }
+        
+        guard let integrationSettings = analytics?.settings() else { return event }
+        guard let destinations = analytics?.timeline.plugins[.destination]?.plugins as? [DestinationPlugin] else { return event }
+        
+        // Mark all loaded and enabled destinations as bundled
+        var bundled = [String]()
+        for plugin in destinations {
+            // Skip processing for Segment.io
+            if (plugin is SegmentDestination) {
+                continue
+            }
+            let hasSettings = integrationSettings.hasIntegrationSettings(forPlugin: plugin)
+            if hasSettings {
+                // we have a device mode plugin installed.
+                bundled.append(plugin.key)
+            }
+        }
+        
+        // All unbundledIntegrations not in `bundled` are put in `unbundled`
+        var unbundled = [String]()
+        let segmentInfo = integrationSettings.integrationSettings(forKey: "Segment.io")
+        let unbundledIntegrations = segmentInfo?["unbundledIntegrations"] as? [String] ?? []
+        for integration in unbundledIntegrations {
+            if (!bundled.contains(integration)) {
+                unbundled.append(integration)
+            }
+        }
+        
+        modified._metadata = DestinationMetadata(bundled: bundled, unbundled: unbundled, bundledIds: [])
+        
+        return modified
+    }
+}

Sources/Segment/Plugins/SegmentDestination.swift

@@ -57,6 +57,8 @@ public class SegmentDestination: DestinationPlugin {
         flushTimer = QueueTimer(interval: analytics.configuration.values.flushInterval) {
             self.flush()
         }
+        // Add DestinationMetadata enrichment plugin
+        add(plugin: DestinationMetadataPlugin())
     }
 
     public func update(settings: Settings, type: UpdateType) {
@@ -72,8 +74,7 @@ public class SegmentDestination: DestinationPlugin {
     public func execute<T: RawEvent>(event: T?) -> T? {
         let result: T? = event
         if let r = result {
-            let modified = configureCloudDestinations(event: r)
-            queueEvent(event: modified)
+            queueEvent(event: r)
         }
         return result
     }
@@ -143,37 +144,6 @@ public class SegmentDestination: DestinationPlugin {
     }
 }
 
-// MARK: - Utility methods
-extension SegmentDestination {
-    internal func configureCloudDestinations<T: RawEvent>(event: T) -> T {
-        guard let integrationSettings = analytics?.settings() else { return event }
-        guard let plugins = analytics?.timeline.plugins[.destination]?.plugins as? [DestinationPlugin] else { return event }
-        guard let customerValues = event.integrations?.dictionaryValue else { return event }
-        
-        var merged = [String: Any]()
-        
-        // compare settings to loaded plugins
-        for plugin in plugins {
-            let hasSettings = integrationSettings.hasIntegrationSettings(forPlugin: plugin)
-            if hasSettings {
-                // we have a device mode plugin installed.
-                // tell segment not to send it via cloud mode.
-                merged[plugin.key] = false
-            }
-        }
-        
-        // apply customer values; the customer is always right!
-        for (key, value) in customerValues {
-            merged[key] = value
-        }
-        
-        var modified = event
-        modified.integrations = try? JSON(merged)
-        
-        return modified
-    }
-}
-
 // MARK: - Upload management
 
 extension SegmentDestination {

Sources/Segment/Types.swift

@@ -8,11 +8,17 @@
 import Foundation
 import Sovran
 
+// MARK: - Supplementary Types
+
+public struct DestinationMetadata: Codable {
+    var bundled: [String] = []
+    var unbundled: [String] = []
+    var bundledIds: [String] = []
+}
 
 // MARK: - Event Types
 
 public protocol RawEvent: Codable {
-    
     var type: String? { get set }
     var anonymousId: String? { get set }
     var messageId: String? { get set }
@@ -22,6 +28,7 @@ public protocol RawEvent: Codable {
     var context: JSON? { get set }
     var integrations: JSON? { get set }
     var metrics: [JSON]? { get set }
+    var _metadata: DestinationMetadata? { get set }
 }
 
 public struct TrackEvent: RawEvent {
@@ -33,6 +40,7 @@ public struct TrackEvent: RawEvent {
     public var context: JSON? = nil
     public var integrations: JSON? = nil
     public var metrics: [JSON]? = nil
+    public var _metadata: DestinationMetadata? = nil
 
     public var event: String
     public var properties: JSON?
@@ -57,9 +65,11 @@ public struct IdentifyEvent: RawEvent {
     public var context: JSON? = nil
     public var integrations: JSON? = nil
     public var metrics: [JSON]? = nil
+    public var _metadata: DestinationMetadata? = nil
 
     public var traits: JSON?
 
+    
     public init(userId: String? = nil, traits: JSON? = nil) {
         self.userId = userId
         self.traits = traits
@@ -80,6 +90,7 @@ public struct ScreenEvent: RawEvent {
     public var context: JSON? = nil
     public var integrations: JSON? = nil
     public var metrics: [JSON]? = nil
+    public var _metadata: DestinationMetadata? = nil
 
     public var name: String?
     public var category: String?
@@ -106,6 +117,7 @@ public struct GroupEvent: RawEvent {
     public var context: JSON? = nil
     public var integrations: JSON? = nil
     public var metrics: [JSON]? = nil
+    public var _metadata: DestinationMetadata? = nil
 
     public var groupId: String?
     public var traits: JSON?
@@ -129,6 +141,7 @@ public struct AliasEvent: RawEvent {
     public var context: JSON? = nil
     public var integrations: JSON? = nil
     public var metrics: [JSON]? = nil
+    public var _metadata: DestinationMetadata? = nil
 
     public var userId: String?
     public var previousId: String?
@@ -271,6 +284,7 @@ extension RawEvent {
             timestamp = e.timestamp
             context = e.context
             integrations = e.integrations
+            _metadata = e._metadata
         }
     }
 
@@ -284,6 +298,7 @@ extension RawEvent {
         result.messageId = UUID().uuidString
         result.timestamp = Date().iso8601()
         result.integrations = try? JSON([String: Any]())
+        result._metadata = DestinationMetadata()
 
         return result
     }

Sources/Segment/Utilities/HTTPClient.swift

@@ -65,7 +65,7 @@ public class HTTPClient {
     ///   - completion: The closure executed when done. Passes if the task should be retried or not if failed.
     @discardableResult
     func startBatchUpload(writeKey: String, batch: URL, completion: @escaping (_ result: Result<Bool, Error>) -> Void) -> URLSessionDataTask? {
-        guard let uploadURL = segmentURL(for: apiHost, path: "/batch") else {
+        guard let uploadURL = segmentURL(for: apiHost, path: "/b") else {
             completion(.failure(HTTPClientErrors.failedToOpenBatch))
             return nil
         }

Sources/Segment/Utilities/Storage.swift

@@ -294,7 +294,7 @@ extension Storage {
         let sentAt = Date().iso8601()
 
         // write it to the existing file
-        let fileEnding = "],\"sentAt\":\"\(sentAt)\"}"
+        let fileEnding = "],\"sentAt\":\"\(sentAt)\",\"writeKey\":\"\(writeKey)\"}"
         let endData = fileEnding.data(using: .utf8)
         if let endData = endData {
             fileHandle.seekToEndOfFile()

Tests/Segment-Tests/Analytics_Tests.swift

@@ -340,4 +340,51 @@ final class Analytics_Tests: XCTestCase {
 
         XCTAssertEqual(eventVersion, analyticsVersion)
     }
+    
+    class AnyDestination: DestinationPlugin {
+        var timeline: Timeline
+        let type: PluginType
+        let key: String
+        var analytics: Analytics?
+        
+        init(key: String) {
+            self.key = key
+            self.type = .destination
+            self.timeline = Timeline()
+        }
+    }
+
+    func testDestinationMetadata() {
+        let analytics = Analytics(configuration: Configuration(writeKey: "test"))
+        let mixpanel = AnyDestination(key: "Mixpanel")
+        let outputReader = OutputReaderPlugin()
+        analytics.add(plugin: outputReader)
+        analytics.add(plugin: mixpanel)
+        analytics.add(plugin: DestinationMetadataPlugin())
+        var settings = Settings(writeKey: "123")
+        let integrations = try? JSON([
+            "Segment.io": JSON([
+                "unbundledIntegrations":
+                    [
+                        "Customer.io",
+                        "Mixpanel",
+                        "Amplitude"
+                    ]
+                ]),
+            "Mixpanel": JSON(["someKey": "someVal"])
+        ])
+        settings.integrations = integrations
+        analytics.store.dispatch(action: System.UpdateSettingsAction(settings: settings))
+        
+        waitUntilStarted(analytics: analytics)
+
+        
+        analytics.track(name: "sampleEvent")
+        
+        let trackEvent: TrackEvent? = outputReader.lastEvent as? TrackEvent
+        let metadata = trackEvent?._metadata
+        
+        XCTAssertEqual(metadata?.bundled, ["Mixpanel"])
+        XCTAssertEqual(metadata?.unbundled, ["Customer.io", "Amplitude"])
+    }
 }