feature: Added Filtering and Rate Limiting

CustomerIOCommon.podspec

@@ -21,4 +21,6 @@ Pod::Spec.new do |spec|
   spec.exclude_files = "Sources/**/*{.md}"
 
   spec.module_name = "CioInternalCommon" # the `import X` name when using SDK in Swift files
+
+  spec.dependency 'SyncSqlCipher', '1.0.0'
 end

Package.swift

@@ -1,4 +1,4 @@
-// swift-tools-version:5.5
+// swift-tools-version:5.10
 
 /**
  Manifest file for Swift Package Manager. This file defines our Swift Package for customers to install our SDK modules into their app. 
@@ -49,12 +49,14 @@ let package = Package(
         .package(name: "CioAnalytics", url: "https://github.com/customerio/cdp-analytics-swift.git", .exact("1.7.3+cio.1")),
 
         // SSE (Server-Sent Events) client for real-time in-app messaging
-        .package(url: "https://github.com/LaunchDarkly/swift-eventsource.git", .upToNextMajor(from: "3.3.0"))
+        .package(url: "https://github.com/LaunchDarkly/swift-eventsource.git", .upToNextMajor(from: "3.3.0")),
+        .package(url: "https://github.com/customerio/SyncSqlCipher.git", from: "1.0.0")
     ],
     targets: [ 
         // Common - Code used by multiple modules in the SDK project.
-        // this module is *not* exposed to the public. It's used internally. 
+        // this module is *not* exposed to the public. It's used internally.
         .target(name: "CioInternalCommon",
+                dependencies: [.product(name: "SyncSqlCipher", package: "SyncSqlCipher")],
                 path: "Sources/Common"),
         .testTarget(name: "CommonTests",
                     dependencies: ["CioInternalCommon", "SharedTests"],

Sources/Common/DIGraphShared+StorageManager.swift

@@ -0,0 +1,49 @@
+import Foundation
+import SyncSqlCipher
+
+public extension DIGraphShared {
+    /// Creates, migrates, and registers the ``StorageManager`` for a given CDP API key.
+    ///
+    /// Follows the same pattern as ``registerPendingPushDeliveryStore(appGroupId:)``
+    /// — call this once from ``DataPipeline.initialize(moduleConfig:)`` before
+    /// creating ``DataPipelineImplementation``, so the graph has a live instance
+    /// ready for any code that calls ``storageManager``.
+    ///
+    /// Returns `nil` and logs an error if the database cannot be opened or
+    /// migrations fail. Callers that depend on storage must handle a nil result
+    /// gracefully; the aggregation engine will be a no-op in that case.
+    @discardableResult
+    func registerStorageManager(cdpApiKey: String) -> StorageManager? {
+        if let existing: StorageManager = getOptional(StorageManager.self) {
+            return existing
+        }
+
+        do {
+            let path = Self.databasePath(for: cdpApiKey)
+            let db = try Database(path: path, key: cdpApiKey)
+            let storage = StorageManager(db: db)
+            try storage.runMigrations()
+            register(storage, forType: StorageManager.self)
+            return storage
+        } catch {
+            logger.error("CIO StorageManager failed to initialize: \(error)")
+            return nil
+        }
+    }
+
+    /// The registered ``StorageManager``, or `nil` if ``registerStorageManager(cdpApiKey:)``
+    /// has not been called yet or failed.
+    var storageManager: StorageManager? {
+        getOptional(StorageManager.self)
+    }
+
+    private static func databasePath(for cdpApiKey: String) -> String {
+        let appSupport = FileManager.default
+            .urls(for: .applicationSupportDirectory, in: .userDomainMask)[0]
+        let dir = appSupport
+            .appendingPathComponent("io.customer", isDirectory: true)
+            .appendingPathComponent(cdpApiKey, isDirectory: true)
+        try? FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
+        return dir.appendingPathComponent("cio.db").path
+    }
+}

Sources/Common/Storage/StorageManager.swift

@@ -0,0 +1,89 @@
+import Foundation
+@preconcurrency import SyncSqlCipher
+
+/// Gateway for all encrypted on-disk storage used by the aggregation engine.
+///
+/// Implemented as a `struct` because `SyncSqlCipher.Database` is internally
+/// thread-safe (DispatchQueue-serialized), so `StorageManager` carries no
+/// mutable state of its own.
+public struct StorageManager: Sendable {
+    // public so per-module StorageManager+X.swift extensions can reach it directly.
+    // Restrict to `package` once CocoaPods support is dropped.
+    public let db: Database
+
+    public init(db: Database) {
+        self.db = db
+    }
+
+    /// Apply schema migrations. Must be called once before any other method.
+    /// Pass `extra` migrations from modules that extend the schema.
+    public func runMigrations(extra: [any Migration] = []) throws {
+        try db.migrate([CreateAggregationSchema()] + extra)
+    }
+
+    // MARK: - sdk_meta (utility key/value table)
+
+    public func getMetaValue(_ key: String) throws -> String? {
+        try db.fetchOne(SdkMetaRecord.self, id: key)?.value
+    }
+
+    public func setMetaValue(_ value: String?, for key: String) throws {
+        if let value {
+            _ = try db.save(SdkMetaRecord(key: key, value: value))
+        } else {
+            try db.execute("DELETE FROM sdk_meta WHERE key = ?", key)
+        }
+    }
+}
+
+// MARK: - Entity Records
+
+private struct SdkMetaRecord: Entity {
+    static let tableName = TableName("sdk_meta")
+    static let primaryKeyName = "key"
+    static let primaryKey: WritableKeyPath<SdkMetaRecord, String> & Sendable = \.key
+
+    var key: String
+    var value: String
+}
+
+// MARK: - Schema
+
+private struct CreateAggregationSchema: Migration {
+    let id = "001-create-aggregation-schema"
+
+    func up(_ ctx: MigrationContext) throws {
+        // Server-fetched aggregation ruleset (single row, id always = 1).
+        try ctx.execute(
+            """
+            CREATE TABLE IF NOT EXISTS aggregation_rules (
+                id         INTEGER PRIMARY KEY CHECK (id = 1),
+                payload    TEXT    NOT NULL,
+                fetched_at TEXT    NOT NULL
+            )
+            """
+        )
+        // Per-rule accumulator state (counters for rate limiting).
+        try ctx.execute(
+            """
+            CREATE TABLE IF NOT EXISTS aggregation_state (
+                rule_id         TEXT    NOT NULL PRIMARY KEY,
+                state_json      TEXT    NOT NULL,
+                last_flushed_at INTEGER NOT NULL DEFAULT 0,
+                scope           TEXT    NOT NULL DEFAULT 'profile'
+            )
+            """
+        )
+        // General-purpose SDK key/value metadata.
+        try ctx.execute(
+            """
+            CREATE TABLE IF NOT EXISTS sdk_meta (
+                key   TEXT NOT NULL PRIMARY KEY,
+                value TEXT NOT NULL
+            )
+            """
+        )
+    }
+
+    func down(_ ctx: MigrationContext) throws {}
+}

Sources/DataPipeline/DataPipeline.swift

@@ -59,6 +59,7 @@ public class DataPipeline: ModuleTopLevelObject<DataPipelineInstance>, DataPipel
         shared.initializeModuleIfNotAlready {
             Self.moduleConfig = moduleConfig
 
+            DIGraphShared.shared.registerStorageManager(cdpApiKey: moduleConfig.cdpApiKey)
             return DataPipelineImplementation(diGraph: DIGraphShared.shared, moduleConfig: moduleConfig)
         }
 

Sources/DataPipeline/DataPipelineImplementation.swift

@@ -1,5 +1,6 @@
 import CioAnalytics
 import CioInternalCommon
+import Foundation
 
 class DataPipelineImplementation: DataPipelineInstance, DataPipelineTracking {
     private let moduleConfig: DataPipelineConfigOptions
@@ -14,6 +15,8 @@ class DataPipelineImplementation: DataPipelineInstance, DataPipelineTracking {
     private let deviceInfo: DeviceInfo
     private let contextPlugin: Context
     private let profileStore: ProfileStore
+    let storageManager: StorageManager?
+    let eventPolicyEngine: EventPolicyEngine?
 
     /// Per-session tracker of the most recently identified userId. Used to dedup
     /// no-traits identify calls against the same userId within a single process
@@ -34,6 +37,19 @@ class DataPipelineImplementation: DataPipelineInstance, DataPipelineTracking {
         self.dateUtil = diGraph.dateUtil
         self.deviceInfo = diGraph.deviceInfo
         self.profileStore = diGraph.profileStore
+        self.storageManager = diGraph.storageManager
+
+        if let storage = diGraph.storageManager {
+            let engine = EventPolicyEngine(storage: storage)
+            if let config = try? storage.getAggregationConfig(),
+               let data = config.payload.data(using: .utf8),
+               let ruleset = try? JSONDecoder().decode(AggregationRuleset.self, from: data) {
+                engine.load(ruleset: ruleset)
+            }
+            self.eventPolicyEngine = engine
+        } else {
+            self.eventPolicyEngine = nil
+        }
 
         self.contextPlugin = Context(diGraph: diGraph)
 
@@ -70,6 +86,11 @@ class DataPipelineImplementation: DataPipelineInstance, DataPipelineTracking {
         // plugin to publish data pipeline events
         analytics.add(plugin: DataPipelinePublishedEvents(diGraph: diGraph))
 
+        // Add plugin to enforce server-driven filter and rate-limit rules
+        if let engine = eventPolicyEngine {
+            analytics.add(plugin: EventPolicyPlugin(engine: engine, storage: storageManager))
+        }
+
         // Add plugin to filter events based on SDK configuration
         analytics.add(plugin: ScreenFilterPlugin(screenViewUse: moduleConfig.screenViewUse))
 
@@ -222,6 +243,7 @@ class DataPipelineImplementation: DataPipelineInstance, DataPipelineTracking {
         // reset all to default state
         logger.debug("resetting user profile")
         analytics.reset()
+        try? storageManager?.deleteProfileScopedAggregationState()
 
         // Reset per-session identify dedup tracker so the next identify (even
         // for the same userId) takes the full path.

Sources/DataPipeline/EventPolicy/AggregationRuleset.swift

@@ -0,0 +1,49 @@
+import Foundation
+
+enum RuleScope: String, Codable, Sendable {
+    case profile
+    case device
+
+    init(from decoder: Decoder) throws {
+        let raw = try decoder.singleValueContainer().decode(String.self)
+        self = RuleScope(rawValue: raw) ?? .profile
+    }
+}
+
+struct FilterEntry: Codable, Sendable {
+    let eventType: String
+    let name: String
+    let scope: RuleScope
+
+    enum CodingKeys: String, CodingKey { case eventType, name, scope }
+
+    init(from decoder: Decoder) throws {
+        let c = try decoder.container(keyedBy: CodingKeys.self)
+        self.eventType = try c.decode(String.self, forKey: .eventType)
+        self.name = try c.decode(String.self, forKey: .name)
+        self.scope = try c.decodeIfPresent(RuleScope.self, forKey: .scope) ?? .profile
+    }
+}
+
+struct RateLimitEntry: Codable, Sendable {
+    let eventType: String
+    let name: String
+    let windowSeconds: Int
+    let scope: RuleScope
+
+    enum CodingKeys: String, CodingKey { case eventType, name, windowSeconds, scope }
+
+    init(from decoder: Decoder) throws {
+        let c = try decoder.container(keyedBy: CodingKeys.self)
+        self.eventType = try c.decode(String.self, forKey: .eventType)
+        self.name = try c.decode(String.self, forKey: .name)
+        self.windowSeconds = try c.decode(Int.self, forKey: .windowSeconds)
+        self.scope = try c.decodeIfPresent(RuleScope.self, forKey: .scope) ?? .profile
+    }
+}
+
+struct AggregationRuleset: Codable, Sendable {
+    let filters: [FilterEntry]?
+    let rateLimits: [RateLimitEntry]?
+    // rules reserved for next aggregation phase
+}

Sources/DataPipeline/EventPolicy/EventPolicyEngine.swift

@@ -0,0 +1,67 @@
+import CioInternalCommon
+import Foundation
+
+/// Synchronous, thread-safe engine that decides whether an event should be
+/// allowed through or dropped based on the active ``AggregationRuleset``.
+///
+/// `@unchecked Sendable` is safe here: all mutable state (`processed`) is
+/// protected by `Synchronized<>`, and `StorageManager.db` is serialized
+/// internally via a private DispatchQueue.
+final class EventPolicyEngine: Sendable {
+    /// Wire-format ruleset pre-processed into O(1) lookup structures.
+    private struct ProcessedRuleset {
+        /// Keys of the form `"\(eventType):\(name)"` for blocked events.
+        let filterKeys: Set<String>
+        /// Rate-limit rules keyed by `"\(eventType):\(name)"`. First rule wins on duplicates.
+        let rateLimitsByKey: [String: RateLimitEntry]
+
+        init(_ ruleset: AggregationRuleset) {
+            self.filterKeys = Set((ruleset.filters ?? []).map { "\($0.eventType):\($0.name)" })
+            self.rateLimitsByKey = Dictionary(
+                (ruleset.rateLimits ?? []).map { ("\($0.eventType):\($0.name)", $0) },
+                uniquingKeysWith: { first, _ in first }
+            )
+        }
+    }
+
+    private let storage: StorageManager
+    private let processed = Synchronized<ProcessedRuleset?>(nil)
+
+    init(storage: StorageManager) {
+        self.storage = storage
+    }
+
+    /// Replace the active ruleset. Call this after receiving updated config from the server.
+    func load(ruleset: AggregationRuleset?) {
+        processed.wrappedValue = ruleset.map { ProcessedRuleset($0) }
+    }
+
+    /// Returns `true` if the event is allowed through, `false` if it should be dropped.
+    func shouldAllow(eventType: String, name: String) -> Bool {
+        shouldAllow(eventType: eventType, name: name, now: Int64(Date().timeIntervalSince1970))
+    }
+
+    /// Time-injectable overload used by tests.
+    func shouldAllow(eventType: String, name: String, now: Int64) -> Bool {
+        guard let rs = processed.wrappedValue else { return true }
+
+        let exactKey = "\(eventType):\(name)"
+        let wildcardKey = "\(eventType):*"
+
+        if rs.filterKeys.contains(exactKey) || rs.filterKeys.contains(wildcardKey) {
+            return false
+        }
+
+        if let rl = rs.rateLimitsByKey[exactKey] ?? rs.rateLimitsByKey[wildcardKey] {
+            // Fail open on DB error so a storage failure doesn't silently discard events.
+            return (try? storage.checkAndUpdateRateLimit(
+                key: exactKey,
+                now: now,
+                windowSeconds: Int64(rl.windowSeconds),
+                scope: rl.scope.rawValue
+            )) ?? true
+        }
+
+        return true
+    }
+}