Add control ability server side.

Package.swift

@@ -18,7 +18,8 @@ let package = Package(
             targets: ["AnalyticsLive"])
     ],
     dependencies: [
-        .package(url: "https://github.com/segmentio/analytics-swift.git", from: "1.8.0"),
+        .package(url: "https://github.com/segmentio/analytics-swift.git", from: "1.9.1"),
+        //.package(path: "../analytics-swift"),
         .package(url: "https://github.com/segmentio/substrata-swift.git", from: "2.0.11"),
         //.package(path: "../substrata-swift")
     ],

Sources/AnalyticsLive/LivePlugins/LivePlugins.swift

@@ -81,7 +81,8 @@ public class LivePlugins: UtilityPlugin, WaitingPlugin {
             DispatchQueue.main.async { [weak self] in
                 guard let self else { return }
                 let useFallback = self.forceFallback || !success
-                self.loadEdgeFn(url: Bundler.getLocalBundleURL(bundleName: Constants.edgeFunctionFilename), useFallback: useFallback)
+                let url = Bundler.getLocalBundleURL(bundleName: Constants.edgeFunctionFilename)
+                self.loadEdgeFn(url: url, useFallback: useFallback)
                 analytics.resumeEventProcessing(plugin: self)
             }
         }

Sources/AnalyticsLive/Signals/Broadcasters/SegmentBroadcaster.swift

@@ -9,17 +9,8 @@ import Foundation
 import Segment
 
 public class SegmentBroadcaster: SignalBroadcaster {
-    public weak var analytics: Analytics? = nil {
-        didSet {
-            if sendToSegment {
-                guard let analytics else { return }
-                self.mini = MiniAnalytics(analytics: analytics)
-            }
-        }
-    }
-
-    internal let sendToSegment: Bool
-    internal let obfuscate: Bool
+    internal var sendToSegment: Bool
+    internal var obfuscate: Bool
     internal var mini: MiniAnalytics? = nil
 
     public func added(signal: any RawSignal) {
@@ -35,8 +26,16 @@ public class SegmentBroadcaster: SignalBroadcaster {
         }
     }
 
-    public init(sendToSegment: Bool = false, obfuscate: Bool = true) {
+    public init(sendToSegment: Bool = false, obfuscate: Bool = true, writeKey: String, apiHost: String) {
         self.obfuscate = obfuscate
         self.sendToSegment = sendToSegment
+        if sendToSegment {
+            self.mini = MiniAnalytics(writeKey: writeKey, apiHost: apiHost)
+        }
+    }
+
+    public func disable() {
+        self.obfuscate = true
+        self.sendToSegment = false
     }
 }

Sources/AnalyticsLive/Signals/Configuration.swift

@@ -26,6 +26,7 @@ public struct SignalsConfiguration {
     internal var broadcasters: [SignalBroadcaster]
     internal let sendDebugSignalsToSegment: Bool
     internal let obfuscateDebugSignals: Bool
+    internal let apiHost: String
     internal let useUIKitAutoSignal: Bool
     internal let useSwiftUIAutoSignal: Bool
     internal let useNetworkAutoSignal: Bool
@@ -40,6 +41,7 @@ public struct SignalsConfiguration {
         broadcasters: [SignalBroadcaster] = [],
         sendDebugSignalsToSegment: Bool = false,
         obfuscateDebugSignals: Bool = true,
+        apiHost: String = "signals.segment.io/v1",
         useUIKitAutoSignal: Bool = false,
         useSwiftUIAutoSignal: Bool = false,
         useNetworkAutoSignal: Bool = false,
@@ -53,20 +55,22 @@ public struct SignalsConfiguration {
         self.broadcasters = broadcasters
         self.sendDebugSignalsToSegment = sendDebugSignalsToSegment
         self.obfuscateDebugSignals = obfuscateDebugSignals
+        self.apiHost = apiHost
         self.useUIKitAutoSignal = useUIKitAutoSignal
         self.useSwiftUIAutoSignal = useSwiftUIAutoSignal
         self.useNetworkAutoSignal = useNetworkAutoSignal
         self.allowedNetworkHosts = allowedNetworkHosts
 
-        if !self.broadcasters.contains(where: { $0 is SegmentBroadcaster }) {
+        /*if !self.broadcasters.contains(where: { $0 is SegmentBroadcaster }) {
             if self.sendDebugSignalsToSegment {
                 let seg = SegmentBroadcaster(
                     sendToSegment: self.sendDebugSignalsToSegment,
-                    obfuscate: self.obfuscateDebugSignals
+                    obfuscate: self.obfuscateDebugSignals,
+                    apiHost: self.apiHost
                 )
                 self.broadcasters.append(seg)
             }
-        }
+        }*/
 
         var blocked = blockedNetworkHosts + Self.autoBlockedHosts
         // block the webhook if it's in use

Sources/AnalyticsLive/Signals/Minilytics/MiniAnalytics.swift

@@ -56,30 +56,41 @@ internal struct MiniTrackEvent: RawEvent {
 }
 
 internal class MiniAnalytics {
-    let analytics: Analytics
     let session: URLSession
+    let apiHost: String
     let storage: TransientDB
 
     @Atomic var flushing: Bool = false
+
+    struct LocalUserInfo {
+        var userId: String? = nil
+        var anonymousId: String? = nil
+    }
+    @Atomic var userInfo = LocalUserInfo()
+
     // used for testing only.
     internal static var observer: ((_ in: any RawSignal, _ out: MiniTrackEvent) -> Void)? = nil
 
-    init(analytics: Analytics) {
-        self.analytics = analytics
+    init(writeKey: String, apiHost: String) {
+        self.apiHost = apiHost
         self.session = Self.configuredSession()
 
         let fileStore = DirectoryStore(
             configuration:
                 DirectoryStore.Configuration(
-                    writeKey: analytics.writeKey,
-                    storageLocation: Self.signalStorageDirectory(writeKey: analytics.writeKey),
+                    writeKey: writeKey,
+                    storageLocation: Self.signalStorageDirectory(writeKey: writeKey),
                     baseFilename: "segment-signals",
                     maxFileSize: 475000,
                     indexKey: "signalFileIndex")
         )
         self.storage = TransientDB(store: fileStore, asyncAppend: true)
     }
 
+    func stateSubscriber(userInfo: UserInfo) {
+        _userInfo.set(LocalUserInfo(userId: userInfo.userId, anonymousId: userInfo.anonymousId))
+    }
+
     func track(signal: any RawSignal, obfuscate: Bool) {
         let input = signal
         var signal = signal
@@ -90,9 +101,10 @@ internal class MiniAnalytics {
 
         guard let props = try? JSON(with: signal) else { return }
 
-        let anonId = analytics.anonymousId
+        let userInfo = self.userInfo
+        let anonId = userInfo.anonymousId
+        let userId = userInfo.userId
         let messageId = UUID().uuidString
-        let userId = analytics.userId
         let timestamp = Date().iso8601()
 
         let track = MiniTrackEvent(
@@ -137,7 +149,7 @@ internal class MiniAnalytics {
                     break
                 }
 
-                analytics.log(message: "Processed: \(url.lastPathComponent)")
+                Analytics.segmentLog(message: "Processed: \(url.lastPathComponent)", kind: .debug)
                 group.leave()
             }
         }
@@ -150,7 +162,7 @@ internal class MiniAnalytics {
 
 extension MiniAnalytics {
     //private static let defaultAPIHost = "signals.segment.io/v1"
-    private static let defaultAPIHost = "signals.segment.build/v1"
+    //private static let defaultAPIHost = "signals.segment.build/v1"
 
     func segmentURL(for host: String, path: String) -> URL? {
         let s = "https://\(host)\(path)"
@@ -160,8 +172,8 @@ extension MiniAnalytics {
 
     @discardableResult
     func startBatchUpload(batch: URL, completion: @escaping (_ result: Result<Bool, Error>) -> Void) -> URLSessionDataTask? {
-        guard let uploadURL = segmentURL(for: Self.defaultAPIHost, path: "/b") else {
-            analytics.reportInternalError(HTTPClientErrors.failedToOpenBatch, fatal: false)
+        guard let uploadURL = segmentURL(for: self.apiHost, path: "/b") else {
+            Analytics.reportInternalError(HTTPClientErrors.failedToOpenBatch, fatal: false)
             completion(.failure(HTTPClientErrors.failedToOpenBatch))
             return nil
         }
@@ -181,7 +193,7 @@ extension MiniAnalytics {
         var request = URLRequest(url: url, cachePolicy: .reloadIgnoringLocalCacheData, timeoutInterval: 60)
         request.httpMethod = method
         request.addValue("application/json; charset=utf-8", forHTTPHeaderField: "Content-Type")
-        request.addValue("analytics-ios/\(analytics.version())", forHTTPHeaderField: "User-Agent")
+        request.addValue("analytics-ios/\(Analytics.version())", forHTTPHeaderField: "User-Agent")
         request.addValue("gzip", forHTTPHeaderField: "Accept-Encoding")
 
         return request
@@ -196,22 +208,22 @@ extension MiniAnalytics {
 
     private func handleResponse(data: Data?, response: URLResponse?, error: Error?, completion: @escaping (_ result: Result<Bool, Error>) -> Void) {
         if let error = error {
-            analytics.log(message: "Error uploading request \(error.localizedDescription).")
-            analytics.reportInternalError(AnalyticsError.networkUnknown(error), fatal: false)
+            Analytics.segmentLog(message: "Error uploading request \(error.localizedDescription).", kind: .error)
+            Analytics.reportInternalError(AnalyticsError.networkUnknown(error), fatal: false)
             completion(.failure(HTTPClientErrors.unknown(error: error)))
         } else if let httpResponse = response as? HTTPURLResponse {
             switch (httpResponse.statusCode) {
             case 1..<300:
                 completion(.success(true))
                 return
             case 300..<400:
-                analytics.reportInternalError(AnalyticsError.networkUnexpectedHTTPCode(httpResponse.statusCode), fatal: false)
+                Analytics.reportInternalError(AnalyticsError.networkUnexpectedHTTPCode(httpResponse.statusCode), fatal: false)
                 completion(.failure(HTTPClientErrors.statusCode(code: httpResponse.statusCode)))
             case 429:
-                analytics.reportInternalError(AnalyticsError.networkServerLimited(httpResponse.statusCode), fatal: false)
+                Analytics.reportInternalError(AnalyticsError.networkServerLimited(httpResponse.statusCode), fatal: false)
                 completion(.failure(HTTPClientErrors.statusCode(code: httpResponse.statusCode)))
             default:
-                analytics.reportInternalError(AnalyticsError.networkServerRejected(httpResponse.statusCode), fatal: false)
+                Analytics.reportInternalError(AnalyticsError.networkServerRejected(httpResponse.statusCode), fatal: false)
                 completion(.failure(HTTPClientErrors.statusCode(code: httpResponse.statusCode)))
             }
         }

Sources/AnalyticsLive/Signals/Signals.swift

@@ -22,12 +22,18 @@ public class Signals: Plugin {
     @Atomic internal var counter: Int = 0
     @Atomic internal var configuration: SignalsConfiguration = SignalsConfiguration(writeKey: "NONE")
 
+    struct AutoInstrumentationSettings: Decodable {
+        let disableTraffic: Bool?
+        let sampleRate: Double?
+    }
+
     struct QueuedSignal {
         let signal: any RawSignal
         let source: SignalSource
     }
     @Atomic internal var queuedSignals = [QueuedSignal]()
     @Atomic internal var ready = false
+    weak var segmentBroadcaster: SegmentBroadcaster? = nil
 
     public var anonymousId: String {
         get {
@@ -57,9 +63,37 @@ public class Signals: Plugin {
         if let e = analytics.find(pluginType: LivePlugins.self) {
             e.addDependent(plugin: self)
         }
-
-        for var b in broadcasters {
-            b.analytics = analytics
+
+        // only pull the lock once.
+        _configuration.mutate { config in
+            // if we don't have a segment broadcaster already ...
+            if !config.broadcasters.contains(where: { $0 is SegmentBroadcaster }) {
+                // and they turned this on, add it.
+                if config.sendDebugSignalsToSegment {
+                    let seg = SegmentBroadcaster(
+                        sendToSegment: true,
+                        obfuscate: config.obfuscateDebugSignals,
+                        writeKey: config.writeKey,
+                        apiHost: config.apiHost
+                    )
+                    if let mini = seg.mini {
+                        // let the miniAnalytics instance get user info updates.
+                        analytics.subscribeToUserInfo(handler: mini.stateSubscriber)
+                    }
+                    config.broadcasters.append(seg)
+                    segmentBroadcaster = seg
+                }
+            }
+        }
+    }
+
+    public func update(settings: Settings, type: UpdateType) {
+        guard type == .initial else { return }
+        guard let aiSettings: AutoInstrumentationSettings = settings.autoInstrumentation?.codableValue() else { return }
+        guard let disableTraffic = aiSettings.disableTraffic else { return }
+        guard let segmentBroadcaster else { return }
+        if disableTraffic {
+            segmentBroadcaster.disable()
         }
     }
 
@@ -92,10 +126,6 @@ public class Signals: Plugin {
 
         // Start swizzlers with new config
         startConfiguredSwizzlers()
-
-        for var b in broadcasters {
-            b.analytics = analytics
-        }
     }
 
     public func emit<T: RawSignal>(signal: T, source: SignalSource = .manual) {
@@ -300,7 +330,7 @@ extension Signals {
     }
 
     internal func isRepeating(event: RawEvent?) -> Bool {
-        let type: String? = event?.context?.value(forKeyPath: KeyPath("__eventOrigin.type"))
+        let type: String? = event?.context?.value(forKeyPath: JSONKeyPath("__eventOrigin.type"))
         guard let type else { return false }
         if type == "signals" {
             return true

Sources/AnalyticsLive/Signals/Types.swift

@@ -12,7 +12,6 @@ import Substrata
 // MARK: -- Signal Broadcaster
 
 public protocol SignalBroadcaster {
-    var analytics: Analytics? { get set }
     func added(signal: any RawSignal)
     func relay()
 }

Sources/AnalyticsLive/Signals/Utilities/Extensions.swift

@@ -6,9 +6,11 @@
 //
 
 import Foundation
+import Segment
 
 extension URL {
     func host() -> String? {
         return self.host
     }
 }
+