diff --git a/.changeset/v4-capture-ai.md b/.changeset/v4-capture-ai.md new file mode 100644 index 000000000..cc73f66f7 --- /dev/null +++ b/.changeset/v4-capture-ai.md @@ -0,0 +1,5 @@ +--- +'posthog-ios': minor +--- + +Add `captureAi(...)` to send AI events such as `$ai_generation` to PostHog's AI endpoint (`/i/v1/ai/events`), which accepts events up to 8 MiB. It returns the event UUID. diff --git a/PostHog/PostHogApi.swift b/PostHog/PostHogApi.swift index ac04345a5..cc75e72fb 100644 --- a/PostHog/PostHogApi.swift +++ b/PostHog/PostHogApi.swift @@ -68,7 +68,8 @@ class PostHogApi { /// Guards the capture V1 state below. In memory only: a new request ID /// after an app restart is fine, since the server only logs it. private let captureV1Lock = NSLock() - private var captureV1PendingRetry: CaptureV1RequestIdentity? + // Keyed by endpoint path, so analytics and AI retries keep separate identities. + private var captureV1PendingRetry: [String: CaptureV1RequestIdentity] = [:] // Set when V1 returns 404 so the rest of this session uses `/batch`. // Remove with /batch. private var captureV1Unavailable = false @@ -222,8 +223,21 @@ class PostHogApi { guard usesCaptureV1 else { return batch(events: events, completion: completion) } + postCaptureV1(path: Self.captureV1AnalyticsPath, events: events, completion: completion) + } + + /// POSTs AI events to the capture V1 AI endpoint (`/i/v1/ai/events`). + /// Same wire format and per-event results as `captureV1`, but a 404 is + /// terminal: AI events never fall back to `/batch`. + func captureAi(events: [PostHogEvent], completion: @escaping (PostHogUploadInfo) -> Void) { + postCaptureV1(path: Self.captureV1AiPath, events: events, completion: completion) + } + + static let captureV1AnalyticsPath = "/i/v1/analytics/events" + static let captureV1AiPath = "/i/v1/ai/events" - guard let url = getEndpointURL("/i/v1/analytics/events", relativeTo: config.host) else { + private func postCaptureV1(path: String, events: [PostHogEvent], completion: @escaping (PostHogUploadInfo) -> Void) { + guard let url = getEndpointURL(path, relativeTo: config.host) else { hedgeLog("Malformed capture URL error.") return completion(PostHogUploadInfo(statusCode: nil, error: nil)) } @@ -232,7 +246,7 @@ class PostHogApi { var seenUuids = Set() let uniqueEvents = events.filter { seenUuids.insert($0.uuid.postHogUuidString).inserted } let uuids = uniqueEvents.map(\.uuid.postHogUuidString) - let identity = nextCaptureV1Identity(uuids: Set(uuids)) + let identity = nextCaptureV1Identity(path: path, uuids: Set(uuids)) let toSend: [String: Any] = [ "created_at": identity.createdAt, @@ -257,10 +271,10 @@ class PostHogApi { guard let self else { return completion(info) } // iOS-only fallback for hosts without capture V1. Remove with /batch. - if info.statusCode == 404 { + if info.statusCode == 404, path == Self.captureV1AnalyticsPath { let firstFallback = self.captureV1Lock.withLock { defer { self.captureV1Unavailable = true } - self.captureV1PendingRetry = nil + self.captureV1PendingRetry[path] = nil return !self.captureV1Unavailable } if firstFallback { @@ -270,7 +284,8 @@ class PostHogApi { } completion(self.captureV1Result( - info, data: data, uuids: uuids, identity: identity, duplicates: events.count - uniqueEvents.count + info, path: path, data: data, uuids: uuids, identity: identity, + duplicates: events.count - uniqueEvents.count )) } }.resume() @@ -279,9 +294,9 @@ class PostHogApi { /// Reuses the request ID with the next attempt number when `uuids` is the /// set of events the previous request left to retry. Otherwise starts a new /// request at attempt 1. - private func nextCaptureV1Identity(uuids: Set) -> CaptureV1RequestIdentity { + private func nextCaptureV1Identity(path: String, uuids: Set) -> CaptureV1RequestIdentity { captureV1Lock.withLock { - if let pending = captureV1PendingRetry, pending.uuids == uuids { + if let pending = captureV1PendingRetry[path], pending.uuids == uuids { return CaptureV1RequestIdentity( uuids: uuids, requestId: pending.requestId, attempt: pending.attempt + 1, createdAt: pending.createdAt ) @@ -296,6 +311,7 @@ class PostHogApi { /// the next request retries under the same identity. private func captureV1Result( _ info: PostHogUploadInfo, + path: String, data: Data?, uuids: [String], identity: CaptureV1RequestIdentity, @@ -335,7 +351,7 @@ class PostHogApi { } captureV1Lock.withLock { - captureV1PendingRetry = retryUuids.isEmpty ? nil : CaptureV1RequestIdentity( + captureV1PendingRetry[path] = retryUuids.isEmpty ? nil : CaptureV1RequestIdentity( uuids: retryUuids, requestId: identity.requestId, attempt: identity.attempt, createdAt: identity.createdAt ) } diff --git a/PostHog/PostHogFileBackedQueue.swift b/PostHog/PostHogFileBackedQueue.swift index 9206ca0a0..4a4d4ee3b 100644 --- a/PostHog/PostHogFileBackedQueue.swift +++ b/PostHog/PostHogFileBackedQueue.swift @@ -15,16 +15,23 @@ class PostHogFileBackedQueue { let queue: URL private let maxSize: Int? + /// FIFO limit on the total size of the stored files, on top of `maxSize`. + /// The newest entry is always kept, even when it's larger on its own. + private let maxBytes: Int? private var items = [String]() + /// File sizes by id, tracked only when `maxBytes` is set. Guarded by `itemsLock`. + private var sizes = [String: Int]() + private var totalBytes = 0 private let itemsLock = NSLock() var depth: Int { itemsLock.withLock { items.count } } - init(queue: URL, oldQueues: [URL] = [], maxSize: Int? = nil) { + init(queue: URL, oldQueues: [URL] = [], maxSize: Int? = nil, maxBytes: Int? = nil) { self.queue = queue self.maxSize = maxSize.map { max(1, $0) } + self.maxBytes = maxBytes.map { max(1, $0) } setup(oldQueues: oldQueues) } @@ -67,7 +74,7 @@ class PostHogFileBackedQueue { func delete(index: Int) { let removed: String? = itemsLock.withLock { guard index < items.count else { return nil } - return items.remove(at: index) + return untrack(items.remove(at: index)) } if let removed { @@ -84,6 +91,7 @@ class PostHogFileBackedQueue { let removed: [String] = itemsLock.withLock { let removed = items.filter { ids.contains($0) } items.removeAll { ids.contains($0) } + removed.forEach { untrack($0) } return removed } @@ -93,32 +101,34 @@ class PostHogFileBackedQueue { } /// Persists one entry and optionally enforces a FIFO capacity in the same - /// critical section. Returning an evicted id lets the queue report + /// critical section. Returning the evicted ids lets the queue report /// backpressure without racing a separate depth check against other adds. @discardableResult - func add(_ contents: Data, maxSize: Int? = nil) -> (success: Bool, evicted: String?) { + func add(_ contents: Data, maxSize: Int? = nil) -> (success: Bool, evicted: [String]) { do { let filename = UUID.v7String() let effectiveMaxSize = maxSize.map { max(1, $0) } ?? self.maxSize - var evicted: String? + var evicted = [String]() try itemsLock.withLock { try contents.write(to: queue.appendingPathComponent(filename)) if let effectiveMaxSize, items.count >= effectiveMaxSize { - evicted = items.removeFirst() - if let evicted { - deleteSafely(queue.appendingPathComponent(evicted)) - } + evicted.append(untrack(items.removeFirst())) } items.append(filename) + track(filename, size: contents.count) + evicted += evictOverByteLimit() } + for id in evicted { + deleteSafely(queue.appendingPathComponent(id)) + } return (true, evicted) } catch { hedgeLog("Could not write file \(error)") - return (false, nil) + return (false, []) } } @@ -148,7 +158,15 @@ class PostHogFileBackedQueue { let sortedItems = try FileManager.default.contentsOfDirectory(at: queue, sortedBy: .creationDateKey) let overflow = maxSize.map { max(0, sortedItems.count - $0) } ?? 0 items = Array(sortedItems.dropFirst(overflow)) - return Array(sortedItems.prefix(overflow)) + sizes = [:] + totalBytes = 0 + if maxBytes != nil { + for item in items { + let attributes = try? FileManager.default.attributesOfItem(atPath: queue.appendingPathComponent(item).path) + track(item, size: (attributes?[.size] as? NSNumber)?.intValue ?? 0) + } + } + return Array(sortedItems.prefix(overflow)) + evictOverByteLimit() } for item in dropped { @@ -194,7 +212,10 @@ class PostHogFileBackedQueue { } if !skipped.isEmpty { - itemsLock.withLock { items.removeAll { skipped.contains($0) } } + itemsLock.withLock { + items.removeAll { skipped.contains($0) } + skipped.forEach { untrack($0) } + } } return results @@ -216,11 +237,38 @@ class PostHogFileBackedQueue { return false } + /// Caller must hold `itemsLock`. + private func track(_ id: String, size: Int) { + guard maxBytes != nil else { return } + sizes[id] = size + totalBytes += size + } + + /// Caller must hold `itemsLock`. Returns `id` for chaining. + @discardableResult + private func untrack(_ id: String) -> String { + if let size = sizes.removeValue(forKey: id) { + totalBytes -= size + } + return id + } + + /// Drops the oldest items until the total fits `maxBytes`, keeping the + /// newest. Caller must hold `itemsLock` and delete the returned files. + private func evictOverByteLimit() -> [String] { + guard let maxBytes else { return [] } + var evicted = [String]() + while totalBytes > maxBytes, items.count > 1 { + evicted.append(untrack(items.removeFirst())) + } + return evicted + } + private func deleteFiles(_ count: Int) { for _ in 0 ..< count { let removed: String? = itemsLock.withLock { guard !items.isEmpty else { return nil } - return items.remove(at: 0) // We always remove from the top of the queue + return untrack(items.remove(at: 0)) // We always remove from the top of the queue } guard let removed else { return } diff --git a/PostHog/PostHogQueue.swift b/PostHog/PostHogQueue.swift index 7ace934fa..a1d55f24d 100644 --- a/PostHog/PostHogQueue.swift +++ b/PostHog/PostHogQueue.swift @@ -118,7 +118,8 @@ class PostHogQueue { fileQueue = PostHogFileBackedQueue( queue: storage.url(forKey: endpoint.storageKey), oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) }, - maxSize: configuredMaxQueueSize + maxSize: configuredMaxQueueSize, + maxBytes: endpoint.maxQueueBytes ) dispatchQueue = DispatchQueue(label: endpoint.dispatchQueueLabel, target: .global(qos: .utility)) } @@ -134,7 +135,8 @@ class PostHogQueue { fileQueue = PostHogFileBackedQueue( queue: storage.url(forKey: endpoint.storageKey), oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) }, - maxSize: configuredMaxQueueSize + maxSize: configuredMaxQueueSize, + maxBytes: endpoint.maxQueueBytes ) dispatchQueue = DispatchQueue(label: endpoint.dispatchQueueLabel, target: .global(qos: .utility)) } @@ -363,8 +365,8 @@ class PostHogQueue { let result = fileQueue.add(data, maxSize: configuredMaxQueueSize) guard result.success else { return false } - if result.evicted != nil { - hedgeLog("Queue is full, dropping oldest record") + if !result.evicted.isEmpty { + hedgeLog("Queue is full, dropped \(result.evicted.count) oldest record(s)") } hedgeLog("Queued \(endpoint.describe(record)). Depth: \(fileQueue.depth)") flushIfOverThreshold() @@ -461,6 +463,7 @@ class PostHogQueue { var selectedIds: [String] = [] // Entry ID -> `endpoint.recordId`, for removing all but retried records. var recordIds: [String: String] = [:] + var batchBytes = 0 var next = start while next < entries.count { let entry = entries[next] @@ -471,6 +474,12 @@ class PostHogQueue { { break } + if !processing.isEmpty, let maxBatchBytes = endpoint.maxBatchBytes, + batchBytes + entry.data.count > maxBatchBytes + { + break + } + batchBytes += entry.data.count processing.append(record) recordIds[entry.id] = endpoint.recordId?(record) } diff --git a/PostHog/PostHogSDK.swift b/PostHog/PostHogSDK.swift index 370b1c40c..e4149e1c6 100644 --- a/PostHog/PostHogSDK.swift +++ b/PostHog/PostHogSDK.swift @@ -85,6 +85,7 @@ let maxRetryDelay = 30.0 } private(set) var logsQueue: PostHogQueue? + private(set) var aiQueue: PostHogQueue? private(set) var storage: PostHogStorage? private var surveyIdentityGeneration: String? #if !os(watchOS) @@ -278,10 +279,12 @@ let maxRetryDelay = 30.0 queue = PostHogQueue(config, theStorage, .batch(api: api), reachability) replayQueue = PostHogReplayQueue(config, theStorage, api, reachability) logsQueue = PostHogQueue(config, theStorage, logsEndpoint, reachability) + aiQueue = PostHogQueue(config, theStorage, .ai(api: api), reachability) #else queue = PostHogQueue(config, theStorage, .batch(api: api)) replayQueue = PostHogReplayQueue(config, theStorage, api) logsQueue = PostHogQueue(config, theStorage, logsEndpoint) + aiQueue = PostHogQueue(config, theStorage, .ai(api: api)) #endif queue?.start(disableReachabilityForTesting: config.disableReachabilityForTesting, @@ -293,6 +296,9 @@ let maxRetryDelay = 30.0 logsQueue?.start(disableReachabilityForTesting: config.disableReachabilityForTesting, disableQueueTimerForTesting: config.disableQueueTimerForTesting) + aiQueue?.start(disableReachabilityForTesting: config.disableReachabilityForTesting, + disableQueueTimerForTesting: config.disableQueueTimerForTesting) + // Create session manager instance for this PostHogSDK instance sessionManager.setup(config: config) sessionManager.startSession() @@ -803,8 +809,8 @@ let maxRetryDelay = 30.0 return props } - /// Trigger an immediate flush of every queue — events, session-replay - /// snapshots, and structured logs. + /// Trigger an immediate flush of every queue — events, AI events, + /// session-replay snapshots, and structured logs. @objc public func flush() { if !isEnabled() { return @@ -813,6 +819,7 @@ let maxRetryDelay = 30.0 queue?.flush() replayQueue?.flush() logsQueue?.flush() + aiQueue?.flush() pushSubscriptionHandler?.retryIfNeeded() } @@ -1408,6 +1415,53 @@ let maxRetryDelay = 30.0 ) } + /// Captures an AI observability event, such as `$ai_generation`, `$ai_span`, `$ai_trace` or + /// `$ai_embedding`, and sends it to PostHog's AI endpoint, which accepts much larger events than + /// `capture`. + /// + /// Takes the same arguments as `capture(_:distinctId:properties:userProperties:userPropertiesSetOnce:groups:timestamp:options:)` + /// and goes through the same `beforeSend` hooks. The event is queued on disk separately from other + /// events and sent unchanged: the SDK doesn't redact or truncate prompts, outputs or media. + /// An event over 8 MiB is dropped. PostHog drops an event sent here whose name doesn't start + /// with `$ai_`. `capture` never sends events here, even ones named `$ai_*`. + /// + /// - Parameters: + /// - event: AI event name, such as `$ai_generation`. + /// - distinctId: Optional distinct ID override. Defaults to the current SDK distinct ID. + /// - properties: Event properties attached only to this event. + /// - userProperties: Person properties to set. Existing values are overwritten. + /// - userPropertiesSetOnce: Person properties to set only if they do not already exist. + /// - groups: Group type/key pairs to attach to this event. + /// - timestamp: Optional event timestamp. Defaults to the current time. + /// - options: Optional settings that tell PostHog how to process this event. Same keys as + /// `capture`'s `options`. + /// + /// - Returns: The event's UUID, or `nil` if the event wasn't queued: for example, the SDK is + /// opted out or disabled, `beforeSend` dropped it, or it's over 8 MiB. + @discardableResult + @objc(captureAiWithEvent:distinctId:properties:userProperties:userPropertiesSetOnce:groups:timestamp:options:) + public func captureAi(_ event: String, + distinctId: String? = nil, + properties: [String: Any]? = nil, + userProperties: [String: Any]? = nil, + userPropertiesSetOnce: [String: Any]? = nil, + groups: [String: String]? = nil, + timestamp: Date? = nil, + options: [String: Any]? = nil) -> String? + { + captureInternal( + event, + distinctId: distinctId, + properties: properties, + userProperties: userProperties, + userPropertiesSetOnce: userPropertiesSetOnce, + groups: groups, + timestamp: timestamp, + options: options, + aiLane: true + ) + } + // A separate Swift name keeps this out of Swift overload resolution. /// Captures a custom event for an explicit distinct ID and timestamp from Objective-C. @@ -1584,6 +1638,8 @@ let maxRetryDelay = 30.0 /// - Parameters: /// - skipBuildProperties: When true, skips buildProperties call and uses properties as-is. /// Used by crash reporting to capture events with pre-built crash-time context. + /// - Returns: The queued event's UUID, or `nil` when the event isn't queued. + @discardableResult func captureInternal( _ event: String, distinctId: String? = nil, @@ -1595,18 +1651,19 @@ let maxRetryDelay = 30.0 options: [String: Any]? = nil, skipBuildProperties: Bool = false, propertyAllowlist: Set? = nil, - deduplicatePersonProperties: Bool = false - ) { + deduplicatePersonProperties: Bool = false, + aiLane: Bool = false + ) -> String? { if !isEnabled() { - return + return nil } if isOptOutState() { - return + return nil } - guard let queue else { - return + guard let queue = aiLane ? aiQueue : queue else { + return nil } // $exception_list only ever comes from the caller, so raw properties are sufficient here @@ -1616,11 +1673,11 @@ let maxRetryDelay = 30.0 PostHogErrorTrackingAutoCaptureIntegration.exceptionListMatchesIgnoredTypes(properties ?? [:], ignoredTypes: ignored) { hedgeLog("$exception skipped: exception type is in errorTrackingConfig.ignoredExceptionTypes") - return + return nil } } - var isSnapshotEvent = event == "$snapshot" + var isSnapshotEvent = !aiLane && event == "$snapshot" let eventTimestamp = timestamp ?? now() let eventDistinctId = distinctId ?? getDistinctId() @@ -1681,16 +1738,17 @@ let maxRetryDelay = 30.0 ) guard let posthogEvent else { - return + return nil } - // Reevaluate if this is a snapshot event because the event might have been updated by the beforeSend hook - isSnapshotEvent = posthogEvent.event == "$snapshot" + // Reevaluate if this is a snapshot event because the event might have been updated by the beforeSend hook. + // The AI lane always uses its own queue. + isSnapshotEvent = !aiLane && posthogEvent.event == "$snapshot" // if this is a $snapshot event and $session_id is missing, don't process then event if isSnapshotEvent, posthogEvent.properties["$session_id"] == nil { releaseClaimIfCarried(posthogEvent) - return + return nil } // Automatically set person properties for feature flags during capture event @@ -1703,9 +1761,10 @@ let maxRetryDelay = 30.0 releaseClaimIfCarried(posthogEvent) replayQueue?.add(posthogEvent) onEventCaptured.invoke(posthogEvent) - } else { - queueEvent(posthogEvent, queue: queue, deduplicatePersonProperties: deduplicatePersonProperties) + return replayQueue == nil ? nil : posthogEvent.uuid.postHogUuidString } + let stored = queueEvent(posthogEvent, queue: queue, deduplicatePersonProperties: deduplicatePersonProperties) + return stored ? posthogEvent.uuid.postHogUuidString : nil } /// Records a screen view by capturing a `$screen` event. @@ -1995,7 +2054,9 @@ let maxRetryDelay = 30.0 return resultEvent } - private func queueEvent(_ event: PostHogEvent, queue: PostHogQueue, deduplicatePersonProperties: Bool = false) { + /// Returns whether the event was stored in `queue`. + @discardableResult + private func queueEvent(_ event: PostHogEvent, queue: PostHogQueue, deduplicatePersonProperties: Bool = false) -> Bool { let storedInQueue: Bool let userProperties = event.properties["$set"] as? [String: Any] let userPropertiesSetOnce = event.properties["$set_once"] as? [String: Any] @@ -2026,7 +2087,7 @@ let maxRetryDelay = 30.0 releaseClaimIfCarried(event) // Only a suppressed duplicate skips the callback below: subscribers such as replay // triggers and event-activated surveys don't depend on this queue reaching disk. - return + return false } storedInQueue = stored } else { @@ -2040,6 +2101,7 @@ let maxRetryDelay = 30.0 } } onEventCaptured.invoke(event) + return storedInQueue } /// Associates subsequent events with a group. @@ -2869,10 +2931,12 @@ let maxRetryDelay = 30.0 queue?.stop() replayQueue?.stop() logsQueue?.stop() + aiQueue?.stop() queue = nil replayQueue = nil logsQueue = nil + aiQueue = nil pushSubscriptionHandler = nil replayDebugPropertiesLock.withLock { lastReplayDebugPropertiesAt = nil diff --git a/PostHog/PostHogStorage.swift b/PostHog/PostHogStorage.swift index 26ae00203..386e54135 100644 --- a/PostHog/PostHogStorage.swift +++ b/PostHog/PostHogStorage.swift @@ -236,6 +236,7 @@ class PostHogStorage { case oldReplayQueue = "posthog.replayFolder" // replay queue with legacy timestamp filenames case replayBufferQueue = "posthog.replayBufferFolder" case logsQueue = "posthog.logsFolder" + case aiQueue = "posthog.aiFolder" case enabledFeatureFlags = "posthog.enabledFeatureFlags" case enabledFeatureFlagPayloads = "posthog.enabledFeatureFlagPayloads" case flags = "posthog.flags" @@ -497,7 +498,7 @@ class PostHogStorage { if !keepAnonymousId { deleteSafely(url(forKey: .anonymousId)) } - // .queue, .replayQeueue, .logsQueue not deleted here — each queue manages its own + // .queue, .replayQeueue, .logsQueue, .aiQueue not deleted here — each queue manages its own // disk state via clear() and the per-record distinctId captured at enqueue time // (see PostHogLogRecord) lets in-flight telemetry survive an identity change. deleteSafely(url(forKey: .oldQueueFolder)) diff --git a/PostHog/QueueEndpoint+Factories.swift b/PostHog/QueueEndpoint+Factories.swift index c10f35e19..ebaf633ff 100644 --- a/PostHog/QueueEndpoint+Factories.swift +++ b/PostHog/QueueEndpoint+Factories.swift @@ -18,6 +18,17 @@ func isCaptureV1RetriableStatusCode(_ code: Int) -> Bool { [408, 500, 502, 503, 504].contains(code) } +/// PostHog's AI endpoint drops events whose properties exceed 8 MiB. The +/// stored event also holds its envelope, so allow 64 KiB on top, as +/// posthog-python and posthog-go do. +let aiMaxEventBytes = 8 * 1024 * 1024 + 64 * 1024 +/// Soft request size for the AI endpoint, matching the other PostHog SDKs. +let aiMaxBatchBytes = 5 * 1024 * 1024 +/// AI events can be megabytes, so keep fewer of them on disk than analytics events. +let aiMaxQueueSize = 100 +/// Disk budget for queued AI events; the oldest are dropped first. +let aiMaxQueueBytes = 50 * 1024 * 1024 + extension QueueEndpoint where Record == PostHogEvent { /// Analytics events endpoint: capture V1 (`/i/v1/analytics/events`), or /// `/batch` after V1 returned 404 this session. @@ -46,6 +57,41 @@ extension QueueEndpoint where Record == PostHogEvent { ) } + /// Capture V1 AI endpoint (`/i/v1/ai/events`) for events sent with + /// `captureAi`. Same retry policy as capture V1, with no `/batch` fallback. + /// Batches by bytes as well as count, since AI events can be megabytes. + static func ai(api: PostHogApi) -> QueueEndpoint { + QueueEndpoint( + storageKey: .aiQueue, + oldStorageKeys: [], + dispatchQueueLabel: "com.posthog.AiQueue", + initialCap: { $0.maxBatchSize }, + initialFlushAt: { $0.flushAt }, + maxQueueSize: { _ in aiMaxQueueSize }, + flushIntervalSeconds: { $0.flushIntervalSeconds }, + rateCapMax: { _ in 0 }, + rateCapWindowSeconds: { _ in 0 }, + encode: { event in + guard let data = toJSONData(event.toJSON()) else { return nil } + // Name and size only: AI properties can hold prompts and media. + if data.count > aiMaxEventBytes { + hedgeLog("Dropping AI event '\(event.event)' of \(data.count) bytes, over the \(aiMaxEventBytes) byte limit.") + return nil + } + return data + }, + decode: { data in PostHogEvent.fromJSON(data) }, + describe: { event in "AI event '\(event.event)'" }, + recordId: { event in event.uuid.postHogUuidString }, + maxBatchBytes: aiMaxBatchBytes, + maxQueueBytes: aiMaxQueueBytes, + send: { events, completion in + api.captureAi(events: events, completion: completion) + }, + isRetriableStatusCode: isCaptureV1RetriableStatusCode + ) + } + /// `/snapshot` endpoint for session-replay snapshots. Shares its retry /// policy with `/batch`. static func snapshot(api: PostHogApi) -> QueueEndpoint { diff --git a/PostHog/QueueEndpoint.swift b/PostHog/QueueEndpoint.swift index b02884300..689898b21 100644 --- a/PostHog/QueueEndpoint.swift +++ b/PostHog/QueueEndpoint.swift @@ -67,6 +67,15 @@ struct QueueEndpoint { /// When set, the queue sends only a compatible FIFO prefix. var canBatchTogether: ((Record, Record) -> Bool)? + /// Soft byte limit for one request, measured on the encoded records. A + /// record that doesn't fit starts the next request; a record larger than + /// the limit goes alone. `nil` batches by count only. + var maxBatchBytes: Int? + + /// FIFO limit on the total size of queued records on disk, on top of + /// `maxQueueSize`. `nil` limits by count only. + var maxQueueBytes: Int? + // MARK: Send /// Build the wire payload from a list of records and POST it. The queue diff --git a/PostHogTests/PostHogCaptureV1Test.swift b/PostHogTests/PostHogCaptureV1Test.swift index 3aac4bd43..1d469f512 100644 --- a/PostHogTests/PostHogCaptureV1Test.swift +++ b/PostHogTests/PostHogCaptureV1Test.swift @@ -605,3 +605,140 @@ final class PostHogCaptureV1QueueTest { } } } + +@Suite("Capture V1 AI lane", .serialized, .resetsGlobalState) +final class PostHogCaptureAiTest { + private let server: MockPostHogServer + + init() { + deleteSafely(applicationSupportDirectoryURL()) + server = MockPostHogServer() + server.start(batchCount: 0) + } + + deinit { + server.stop() + } + + private func makeSDK(beforeSend: BeforeSendBlock? = nil) -> PostHogSDK { + let config = PostHogConfig(projectToken: "phc_ai_\(UUID().uuidString)", host: "http://localhost:9001") + config.flushAt = 1 + config.preloadFeatureFlags = false + config.sendFeatureFlagEvent = false + config.disableReachabilityForTesting = true + config.disableQueueTimerForTesting = true + config.disableFlushOnBackgroundForTesting = true + config.captureApplicationLifecycleEvents = false + if let beforeSend { + config.setBeforeSend(beforeSend) + } + server.batchProjectToken = config.projectToken + return PostHogSDK.with(config) + } + + @Test("captureAi sends to /i/v1/ai/events through beforeSend, and capture keeps $ai_ events on analytics") + func routesByMethod() async throws { + let sut = makeSDK { event in + event.properties["edited"] = true + return event + } + defer { sut.close() } + + let uuid = sut.captureAi("$ai_generation", distinctId: "user", properties: ["$ai_model": "gpt"], options: ["cookieless_mode": true]) + await waitUntil { self.server.aiRequests.count == 1 } + + let request = try #require(server.aiRequests.first) + #expect(request.url?.path == "/i/v1/ai/events") + #expect(request.value(forHTTPHeaderField: "Authorization") == "Bearer \(sut.config.projectToken)") + #expect(request.value(forHTTPHeaderField: "PostHog-Attempt") == "1") + #expect(request.value(forHTTPHeaderField: "PostHog-Request-Id") != nil) + let event = try #require(server.parsePostHogEvents(request).first) + #expect(event.event == "$ai_generation") + #expect(event.uuid.postHogUuidString == uuid) + #expect(event.distinctId == "user") + #expect(event.properties["$ai_model"] as? String == "gpt") + #expect(event.properties["edited"] as? Bool == true) + #expect(event.options["cookieless_mode"] as? Bool == true) + #expect(server.batchRequests.isEmpty) + + sut.capture("$ai_generation") + await waitUntil { self.server.batchRequests.count == 1 } + #expect(server.parsePostHogEvents(try #require(server.batchRequests.first)).map(\.event) == ["$ai_generation"]) + #expect(server.aiRequests.count == 1) + } + + @Test("captureAi returns nil when the event isn't queued") + func returnsNilWhenNotQueued() { + let sut = makeSDK { event in event.event == "$ai_span" ? nil : event } + defer { sut.close() } + + #expect(sut.captureAi("$ai_span") == nil) + #expect(sut.captureAi("$ai_generation", properties: ["$ai_input": String(repeating: "a", count: aiMaxEventBytes)]) == nil) + sut.optOut() + #expect(sut.captureAi("$ai_generation") == nil) + } + + @Test("a 404 from the AI endpoint drops the batch without falling back to /batch") + func notFoundIsTerminal() async throws { + server.aiResponseHandler = { _, _ in HTTPStubsResponse(jsonObject: [:], statusCode: 404, headers: nil) } + let config = PostHogConfig(projectToken: "phc_ai_404", host: "http://localhost:9001") + let storage = PostHogStorage(config) + let api = PostHogApi(config) + let queue = PostHogQueue(config, storage, .ai(api: api), nil) + defer { + queue.stop() + queue.clear() + deleteSafely(storage.appFolderUrl) + } + + queue.add(PostHogEvent(event: "$ai_generation", distinctId: "user")) + queue.flush() + await waitUntil { queue.depth == 0 } + + #expect(queue.depth == 0) + #expect(server.aiRequests.count == 1) + #expect(server.batchRequests.isEmpty) + #expect(api.usesCaptureV1) + } + + @Test("drops an AI event over the size limit when it is queued") + func dropsOversizedEvent() { + let config = PostHogConfig(projectToken: "phc_ai_big", host: "http://localhost:9001") + let storage = PostHogStorage(config) + let queue = PostHogQueue(config, storage, .ai(api: PostHogApi(config)), nil) + defer { + queue.stop() + queue.clear() + deleteSafely(storage.appFolderUrl) + } + + #expect(QueueEndpoint.ai(api: PostHogApi(config)).maxQueueBytes == aiMaxQueueBytes) + let big = String(repeating: "a", count: aiMaxEventBytes) + #expect(!queue.add(PostHogEvent(event: "$ai_generation", distinctId: "user", properties: ["$ai_input": big]))) + #expect(queue.depth == 0) + } + + @Test("splits AI requests at the batch byte limit") + func batchesByBytes() async throws { + let config = PostHogConfig(projectToken: "phc_ai_bytes", host: "http://localhost:9001") + config.flushAt = 100 + let storage = PostHogStorage(config) + let queue = PostHogQueue(config, storage, .ai(api: PostHogApi(config)), nil) + defer { + queue.stop() + queue.clear() + deleteSafely(storage.appFolderUrl) + } + + // Each event is a little over 2 MiB, so two fit under 5 MiB and the third goes next. + let input = String(repeating: "a", count: 2 * 1024 * 1024) + for _ in 0 ..< 3 { + queue.add(PostHogEvent(event: "$ai_generation", distinctId: "user", properties: ["$ai_input": input])) + } + queue.flush() + await waitUntil(timeout: 10) { queue.depth == 0 } + + #expect(queue.depth == 0) + #expect(server.aiRequests.map { server.parsePostHogEvents($0).count } == [2, 1]) + } +} diff --git a/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift b/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift index a6b262957..27a60fa2c 100644 --- a/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift +++ b/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift @@ -57,7 +57,7 @@ struct PostHogFileBackedQueueAlignmentTest { let first = queue.add(Data("A".utf8), maxSize: 2) #expect(first.success) - #expect(first.evicted == nil) + #expect(first.evicted.isEmpty) queue.add(Data("B".utf8), maxSize: 2) let originalIds = queue.peekEntries(2).map(\.id) try FileManager.default.setAttributes([.posixPermissions: 0o555], ofItemAtPath: dir.path) @@ -67,7 +67,7 @@ struct PostHogFileBackedQueueAlignmentTest { let failed = queue.add(Data("C".utf8), maxSize: 2) #expect(!failed.success) - #expect(failed.evicted == nil) + #expect(failed.evicted.isEmpty) #expect(queue.depth == 2) #expect(queue.peekEntries(2).map(\.id) == originalIds) #expect(decode(queue.peek(2)) == ["A", "B"]) @@ -76,7 +76,7 @@ struct PostHogFileBackedQueueAlignmentTest { try FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: dir.path) let retried = queue.add(Data("C".utf8), maxSize: 2) #expect(retried.success) - #expect(retried.evicted == originalIds.first) + #expect(retried.evicted == [originalIds[0]]) #expect(queue.depth == 2) #expect(decode(queue.peek(2)) == ["B", "C"]) let reloaded = PostHogFileBackedQueue(queue: dir, maxSize: 2) @@ -84,6 +84,53 @@ struct PostHogFileBackedQueueAlignmentTest { #expect(reloaded.depth == 2) } + @Test("evicts the oldest records to stay under the byte limit") + func evictsOldestOverByteLimit() throws { + let dir = FileManager.default.temporaryDirectory + .appendingPathComponent("ph-queue-bytes-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: dir) } + let queue = PostHogFileBackedQueue(queue: dir, maxBytes: 10) + + queue.add(Data("AAAA".utf8)) + let first = queue.peekEntries(1).map(\.id) + queue.add(Data("BBBB".utf8)) + let added = queue.add(Data("CCCC".utf8)) + + #expect(added.evicted == first) + #expect(decode(queue.peek(10)) == ["BBBB", "CCCC"]) + #expect(try FileManager.default.contentsOfDirectory(atPath: dir.path).count == 2) + + queue.pop(1) + queue.add(Data("DDDD".utf8)) + #expect(decode(queue.peek(10)) == ["CCCC", "DDDD"]) + } + + @Test("keeps the newest record even when it is over the byte limit alone") + func keepsOversizedNewestRecord() throws { + let dir = FileManager.default.temporaryDirectory + .appendingPathComponent("ph-queue-bytes-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: dir) } + let queue = PostHogFileBackedQueue(queue: dir, maxBytes: 4) + + queue.add(Data("AA".utf8)) + let added = queue.add(Data("BBBBBB".utf8)) + + #expect(added.evicted.count == 1) + #expect(decode(queue.peek(10)) == ["BBBBBB"]) + } + + @Test("trims the oldest records over the byte limit when reloading from disk") + func trimsByteLimitOnLoad() throws { + let (queue, dir) = makeQueue() + defer { try? FileManager.default.removeItem(at: dir) } + enqueue(["AAAA", "BBBB", "CCCC"], into: queue) + + let reloaded = PostHogFileBackedQueue(queue: dir, maxBytes: 8) + + #expect(decode(reloaded.peek(10)) == ["BBBB", "CCCC"]) + #expect(try FileManager.default.contentsOfDirectory(atPath: dir.path).count == 2) + } + @Test("delivers every record once in FIFO order on the happy path") func happyPath() throws { let (queue, dir) = makeQueue() diff --git a/PostHogTests/TestUtils/MockPostHogServer.swift b/PostHogTests/TestUtils/MockPostHogServer.swift index c2c422cec..63988a277 100644 --- a/PostHogTests/TestUtils/MockPostHogServer.swift +++ b/PostHogTests/TestUtils/MockPostHogServer.swift @@ -17,6 +17,7 @@ class MockPostHogServer { private var recordedSnapshotRequests = [URLRequest]() private var recordedLogsRequests = [URLRequest]() private var recordedFlagsRequests = [URLRequest]() + private var recordedAiRequests = [URLRequest]() var batchRequests: [URLRequest] { get { requestsLock.withLock { recordedBatchRequests } } @@ -34,6 +35,14 @@ class MockPostHogServer { get { requestsLock.withLock { recordedFlagsRequests } } set { requestsLock.withLock { recordedFlagsRequests = newValue } } } + /// Requests to the capture V1 AI endpoint (`/i/v1/ai/events`), with buffered bodies. + var aiRequests: [URLRequest] { + get { requestsLock.withLock { recordedAiRequests } } + set { requestsLock.withLock { recordedAiRequests = newValue } } + } + /// Optional override for `/i/v1/ai/events` responses, given the request and its + /// 1-based number. When `nil`, every event gets an `ok` result. + var aiResponseHandler: ((URLRequest, Int) -> HTTPStubsResponse)? var batchExpectation: XCTestExpectation? var snapshotExpectation: XCTestExpectation? var logsExpectation: XCTestExpectation? @@ -448,6 +457,21 @@ class MockPostHogServer { return HTTPStubsResponse(jsonObject: ["results": results], statusCode: 200, headers: nil) }) + stubDescriptors.append(stub(condition: pathEndsWith("/i/v1/ai/events")) { request in + var request = request + request.httpBody = request.body() + let index = self.requestsLock.withLock { () -> Int in + self.recordedAiRequests.append(request) + return self.recordedAiRequests.count + } + if let handler = self.aiResponseHandler { + return handler(request, index) + } + let uuids = (self.parseRequest(request)?["batch"] as? [[String: Any]] ?? []).compactMap { $0["uuid"] as? String } + let results = Dictionary(uniqueKeysWithValues: uuids.map { ($0, ["result": "ok"]) }) + return HTTPStubsResponse(jsonObject: ["results": results], statusCode: 200, headers: nil) + }) + stubDescriptors.append(stub(condition: pathEndsWith("/s")) { _ in if self.return500 { HTTPStubsResponse(jsonObject: [], statusCode: 500, headers: nil) @@ -640,6 +664,8 @@ class MockPostHogServer { snapshotRequests = [] logsRequests = [] flagsRequests = [] + aiRequests = [] + aiResponseHandler = nil pushSubscriptionRequests = [] batchExpectation = XCTestExpectation(description: "\(batchCount) batch requests to occur") snapshotExpectation = XCTestExpectation(description: "\(snapshotCount) snapshot requests to occur") @@ -672,7 +698,7 @@ class MockPostHogServer { } static func isCaptureV1(_ request: URLRequest) -> Bool { - request.url?.path.hasSuffix("/i/v1/analytics/events") == true + request.url.map { $0.path.hasSuffix("/i/v1/analytics/events") || $0.path.hasSuffix("/i/v1/ai/events") } == true } /// Decodes the events of a `/batch` or capture V1 request. For V1 it puts diff --git a/api/posthog-ios.public-api.txt b/api/posthog-ios.public-api.txt index 73e115d62..71c2942c0 100644 --- a/api/posthog-ios.public-api.txt +++ b/api/posthog-ios.public-api.txt @@ -257,6 +257,7 @@ PostHog | PostHogSDK.capture(_:properties:) | method | @objc(captureWithEvent:pr PostHog | PostHogSDK.capture(_:properties:userProperties:) | method | @objc(captureWithEvent:properties:userProperties:) func capture(_ event: String, properties: [String : Any]? = nil, userProperties: [String : Any]? = nil) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureWithEvent:properties:userProperties: PostHog | PostHogSDK.capture(_:properties:userProperties:userPropertiesSetOnce:) | method | @objc(captureWithEvent:properties:userProperties:userPropertiesSetOnce:) func capture(_ event: String, properties: [String : Any]? = nil, userProperties: [String : Any]? = nil, userPropertiesSetOnce: [String : Any]? = nil) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureWithEvent:properties:userProperties:userPropertiesSetOnce: PostHog | PostHogSDK.capture(_:properties:userProperties:userPropertiesSetOnce:groups:) | method | @objc(captureWithEvent:properties:userProperties:userPropertiesSetOnce:groups:) func capture(_ event: String, properties: [String : Any]? = nil, userProperties: [String : Any]? = nil, userPropertiesSetOnce: [String : Any]? = nil, groups: [String : String]? = nil) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureWithEvent:properties:userProperties:userPropertiesSetOnce:groups: +PostHog | PostHogSDK.captureAi(_:distinctId:properties:userProperties:userPropertiesSetOnce:groups:timestamp:options:) | method | @discardableResult @objc(captureAiWithEvent:distinctId:properties:userProperties:userPropertiesSetOnce:groups:timestamp:options:) func captureAi(_ event: String, distinctId: String? = nil, properties: [String : Any]? = nil, userProperties: [String : Any]? = nil, userPropertiesSetOnce: [String : Any]? = nil, groups: [String : String]? = nil, timestamp: Date? = nil, options: [String : Any]? = nil) -> String? | c:@M@PostHog@objc(cs)PostHogSDK(im)captureAiWithEvent:distinctId:properties:userProperties:userPropertiesSetOnce:groups:timestamp:options: PostHog | PostHogSDK.captureDeepLink(openURLContexts:) | method | @objc func captureDeepLink(openURLContexts: Set) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureDeepLinkWithOpenURLContexts: PostHog | PostHogSDK.captureDeepLink(url:) | method | @objc func captureDeepLink(url: URL) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureDeepLinkWithUrl: PostHog | PostHogSDK.captureDeepLink(url:options:) | method | @objc func captureDeepLink(url: URL, options: [UIApplication.OpenURLOptionsKey : Any]) | c:@M@PostHog@objc(cs)PostHogSDK(im)captureDeepLinkWithUrl:options: diff --git a/sdk_compliance_adapter/Sources/RequestInterceptor.swift b/sdk_compliance_adapter/Sources/RequestInterceptor.swift index fe0cabbd2..416a9cea3 100644 --- a/sdk_compliance_adapter/Sources/RequestInterceptor.swift +++ b/sdk_compliance_adapter/Sources/RequestInterceptor.swift @@ -105,7 +105,8 @@ class RequestInterceptor: URLProtocol { // Intercept capture endpoints, but not /flags/ or /config let urlString = url.absoluteString - if urlString.contains("/i/v1/analytics/events") || urlString.contains("/batch") + if urlString.contains("/i/v1/analytics/events") || urlString.contains("/i/v1/ai/events") + || urlString.contains("/batch") || urlString.contains("/e/") || urlString.contains("/s/"), !urlString.contains("/flags"), !urlString.contains("/config") diff --git a/sdk_compliance_adapter/Sources/main.swift b/sdk_compliance_adapter/Sources/main.swift index 4f58259c7..d110314f2 100644 --- a/sdk_compliance_adapter/Sources/main.swift +++ b/sdk_compliance_adapter/Sources/main.swift @@ -56,7 +56,7 @@ app.get("health") { req async throws -> Response in // Declares which test suites apply. The iOS SDK posts events to // /i/v1/analytics/events (capture_v1) with gzip. Without this, the // harness skips the capability-gated capture suites entirely. - "capabilities": ["capture_v1", "encoding_gzip", "event_options"], + "capabilities": ["capture_v1", "capture_ai_v1", "encoding_gzip", "event_options"], ] print("[ADAPTER] GET /health") @@ -202,6 +202,41 @@ app.post("capture") { req async throws -> Response in return try await result.encodeResponse(for: req) } +// AI capture endpoint: same request as /capture, sent with captureAi to /i/v1/ai/events. +app.post("capture_ai") { req async throws -> Response in + struct CaptureAiRequest: Content { + let event: String + let distinctId: String? + let properties: [String: AnyCodable]? + let options: [String: AnyCodable]? + + enum CodingKeys: String, CodingKey { + case event + case distinctId = "distinct_id" + case properties + case options + } + } + + let captureReq = try req.content.decode(CaptureAiRequest.self) + print("[ADAPTER] POST /capture_ai - event: \(captureReq.event), distinct_id: \(captureReq.distinctId ?? "nil")") + + guard let sdk = state.posthogSDK else { + throw Abort(.badRequest, reason: "SDK not initialized. Call /init first.") + } + + let uuid = sdk.captureAi( + captureReq.event, + distinctId: captureReq.distinctId, + properties: captureReq.properties?.mapValues(\.value), + options: captureReq.options?.mapValues(\.value) + ) + + var result = ["status": "ok"] + result["uuid"] = uuid + return try await result.encodeResponse(for: req) +} + func fetchFlagsWithRetry(flagsURL: URL, payload: [String: Any]) async throws -> Data { let body = try JSONSerialization.data(withJSONObject: payload) var lastStatus = 0