feat: add tracker runtime and durable offline queue

flextrack/build.gradle.kts

@@ -49,7 +49,10 @@ android {
 
 dependencies {
     coreLibraryDesugaring(libs.desugar.jdk.libs)
+    implementation(libs.kotlinx.coroutines.core)
     testImplementation(libs.junit.jupiter)
+    testImplementation(libs.kotlinx.coroutines.test)
+    testImplementation(libs.json)
 }
 
 publishing {

flextrack/src/main/kotlin/dev/taghizadeh/flextrack/runtime/DispatchResult.kt

@@ -0,0 +1,22 @@
+package dev.taghizadeh.flextrack.runtime
+
+import dev.taghizadeh.flextrack.event.FlexEvent
+import dev.taghizadeh.flextrack.routing.RoutingResult
+
+public data class TrackerFailure(val trackerId: String, val cause: Throwable)
+
+public data class DispatchResult(
+    val event: FlexEvent,
+    val routing: RoutingResult,
+    val successfulTrackerIds: List<String>,
+    val failures: List<TrackerFailure>,
+    val queuedTrackerIds: List<String>,
+) {
+    public val wasQueued: Boolean get() = queuedTrackerIds.isNotEmpty()
+}
+
+public data class FlushResult(
+    val attemptedEvents: Int,
+    val deliveredEvents: Int,
+    val remainingEvents: Int,
+)

flextrack/src/main/kotlin/dev/taghizadeh/flextrack/runtime/EventQueue.kt

@@ -0,0 +1,47 @@
+package dev.taghizadeh.flextrack.runtime
+
+import dev.taghizadeh.flextrack.event.FlexEvent
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import java.time.Instant
+
+public data class QueuedEvent(
+    val id: String,
+    val event: FlexEvent,
+    val trackerIds: List<String>,
+    val attempts: Int = 0,
+    val queuedAt: Instant = Instant.now(),
+)
+
+/** Storage boundary for events waiting for another delivery attempt. */
+public interface EventQueue {
+    public suspend fun enqueue(item: QueuedEvent)
+    public suspend fun read(limit: Int): List<QueuedEvent>
+    public suspend fun replace(item: QueuedEvent)
+    public suspend fun remove(id: String)
+    public suspend fun size(): Int
+}
+
+/** Process-local queue useful for tests and apps that do not require persistence. */
+public class InMemoryEventQueue : EventQueue {
+    private val mutex: Mutex = Mutex()
+    private val items: LinkedHashMap<String, QueuedEvent> = linkedMapOf()
+
+    override suspend fun enqueue(item: QueuedEvent): Unit = mutex.withLock {
+        items.putIfAbsent(item.id, item)
+        Unit
+    }
+
+    override suspend fun read(limit: Int): List<QueuedEvent> = mutex.withLock {
+        require(limit > 0) { "limit must be positive" }
+        items.values.take(limit)
+    }
+
+    override suspend fun replace(item: QueuedEvent): Unit = mutex.withLock {
+        if (item.id in items) items[item.id] = item
+    }
+
+    override suspend fun remove(id: String): Unit = mutex.withLock { items.remove(id); Unit }
+
+    override suspend fun size(): Int = mutex.withLock { items.size }
+}

flextrack/src/main/kotlin/dev/taghizadeh/flextrack/runtime/FileEventQueue.kt

@@ -0,0 +1,152 @@
+package dev.taghizadeh.flextrack.runtime
+
+import android.content.Context
+import android.util.AtomicFile
+import dev.taghizadeh.flextrack.event.FlexEvent
+import dev.taghizadeh.flextrack.routing.EventCategory
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
+import org.json.JSONArray
+import org.json.JSONObject
+import java.io.File
+import java.time.Instant
+
+/** Durable JSON queue stored in the application's private files directory. */
+public class FileEventQueue(
+    context: Context,
+    fileName: String = "flextrack-event-queue.json",
+) : EventQueue {
+    private val file: File = File(context.applicationContext.filesDir, fileName)
+    private val atomicFile: AtomicFile = AtomicFile(file)
+    private val mutex: Mutex = Mutex()
+
+    init {
+        require(fileName.isNotBlank() && !fileName.contains(File.separatorChar)) {
+            "fileName must be a simple non-blank file name"
+        }
+    }
+
+    override suspend fun enqueue(item: QueuedEvent): Unit = mutate { items ->
+        if (items.none { it.id == item.id }) items += item
+    }
+
+    override suspend fun read(limit: Int): List<QueuedEvent> {
+        require(limit > 0) { "limit must be positive" }
+        return mutex.withLock { withContext(Dispatchers.IO) { load().take(limit) } }
+    }
+
+    override suspend fun replace(item: QueuedEvent): Unit = mutate { items ->
+        val index = items.indexOfFirst { it.id == item.id }
+        if (index >= 0) items[index] = item
+    }
+
+    override suspend fun remove(id: String): Unit = mutate { items ->
+        items.removeAll { it.id == id }
+    }
+
+    override suspend fun size(): Int = mutex.withLock {
+        withContext(Dispatchers.IO) { load().size }
+    }
+
+    private suspend fun mutate(block: (MutableList<QueuedEvent>) -> Unit) {
+        mutex.withLock {
+            withContext(Dispatchers.IO) {
+                val items = load()
+                block(items)
+                persist(items)
+            }
+        }
+    }
+
+    private fun load(): MutableList<QueuedEvent> {
+        if (!file.exists() || file.length() == 0L) return mutableListOf()
+        val array = JSONArray(atomicFile.readFully().toString(Charsets.UTF_8))
+        return MutableList(array.length()) { index -> array.getJSONObject(index).toQueuedEvent() }
+    }
+
+    private fun persist(items: List<QueuedEvent>) {
+        file.parentFile?.mkdirs()
+        val output = atomicFile.startWrite()
+        try {
+            output.write(JSONArray(items.map(QueuedEvent::toJson)).toString().toByteArray(Charsets.UTF_8))
+            atomicFile.finishWrite(output)
+        } catch (failure: Throwable) {
+            atomicFile.failWrite(output)
+            throw failure
+        }
+    }
+}
+
+private fun QueuedEvent.toJson(): JSONObject = JSONObject().apply {
+    put("id", id)
+    put("trackerIds", JSONArray(trackerIds))
+    put("attempts", attempts)
+    put("queuedAt", queuedAt.toString())
+    put("event", event.toJson())
+}
+
+private fun FlexEvent.toJson(): JSONObject = JSONObject().apply {
+    put("eventId", eventId)
+    put("timestamp", timestamp.toString())
+    put("name", name)
+    put("properties", properties?.let(::JSONObject) ?: JSONObject.NULL)
+    put("category", category?.name ?: JSONObject.NULL)
+    put("containsPII", containsPII)
+    put("requiresConsent", requiresConsent)
+    put("isHighVolume", isHighVolume)
+    put("isEssential", isEssential)
+    put("userId", userId ?: JSONObject.NULL)
+    put("sessionId", sessionId ?: JSONObject.NULL)
+}
+
+private fun JSONObject.toQueuedEvent(): QueuedEvent {
+    val trackerArray = getJSONArray("trackerIds")
+    return QueuedEvent(
+        id = getString("id"),
+        event = getJSONObject("event").toEvent(),
+        trackerIds = List(trackerArray.length()) { trackerArray.getString(it) },
+        attempts = getInt("attempts"),
+        queuedAt = Instant.parse(getString("queuedAt")),
+    )
+}
+
+private fun JSONObject.toEvent(): FlexEvent {
+    val source = this
+    return object : FlexEvent(
+        eventId = source.getString("eventId"),
+        timestamp = Instant.parse(source.getString("timestamp")),
+    ) {
+        override val name: String = source.getString("name")
+        override val properties: Map<String, Any?>? = source.optJSONObject("properties")?.toMap()
+        override val category: EventCategory? = source.nullableString("category")?.let(::EventCategory)
+        override val containsPII: Boolean = source.getBoolean("containsPII")
+        override val requiresConsent: Boolean = source.getBoolean("requiresConsent")
+        override val isHighVolume: Boolean = source.getBoolean("isHighVolume")
+        override val isEssential: Boolean = source.getBoolean("isEssential")
+        override val userId: String? = source.nullableString("userId")
+        override val sessionId: String? = source.nullableString("sessionId")
+    }
+}
+
+private fun JSONObject.nullableString(key: String): String? =
+    if (isNull(key)) null else getString(key)
+
+private fun JSONObject.toMap(): Map<String, Any?> = keys().asSequence().associateWith { key ->
+    when (val value = get(key)) {
+        JSONObject.NULL -> null
+        is JSONObject -> value.toMap()
+        is JSONArray -> value.toList()
+        else -> value
+    }
+}
+
+private fun JSONArray.toList(): List<Any?> = List(length()) { index ->
+    when (val value = get(index)) {
+        JSONObject.NULL -> null
+        is JSONObject -> value.toMap()
+        is JSONArray -> value.toList()
+        else -> value
+    }
+}

flextrack/src/main/kotlin/dev/taghizadeh/flextrack/runtime/FlexTrackClient.kt

@@ -0,0 +1,99 @@
+package dev.taghizadeh.flextrack.runtime
+
+import dev.taghizadeh.flextrack.event.FlexEvent
+import dev.taghizadeh.flextrack.event.TransformerPipeline
+import dev.taghizadeh.flextrack.routing.ConsentState
+import dev.taghizadeh.flextrack.routing.RoutingEngine
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.coroutineScope
+
+/** Runtime entry point for transforming, routing, delivering, and retrying events. */
+public class FlexTrackClient(
+    public val routingEngine: RoutingEngine,
+    public val registry: TrackerRegistry = TrackerRegistry(),
+    public val queue: EventQueue = InMemoryEventQueue(),
+    public val transformers: TransformerPipeline = TransformerPipeline(),
+    private val consentProvider: () -> ConsentState = { ConsentState() },
+    private val onlineProvider: () -> Boolean = { true },
+) {
+    public suspend fun start(): Unit = registry.start()
+
+    public suspend fun shutdown(): Unit = registry.shutdown()
+
+    public suspend fun register(tracker: Tracker): Unit = registry.register(tracker)
+
+    public suspend fun unregister(trackerId: String): Tracker? = registry.unregister(trackerId)
+
+    public suspend fun track(event: FlexEvent): DispatchResult {
+        val transformed = transformers.transform(event)
+        val trackers = registry.snapshot()
+        val routing = routingEngine.route(transformed, consentProvider(), trackers.keys)
+        val targets = routing.targetTrackers
+
+        if (!onlineProvider() && targets.isNotEmpty()) {
+            queue.enqueue(QueuedEvent(transformed.eventId, transformed, targets))
+            return DispatchResult(transformed, routing, emptyList(), emptyList(), targets)
+        }
+
+        val outcomes = deliver(transformed, targets, trackers)
+        val failures = outcomes.mapNotNull { it.failure }
+        val failedIds = failures.map(TrackerFailure::trackerId)
+        if (failedIds.isNotEmpty()) {
+            queue.enqueue(QueuedEvent(transformed.eventId, transformed, failedIds))
+        }
+        return DispatchResult(
+            transformed,
+            routing,
+            outcomes.filter { it.failure == null }.map(Outcome::trackerId),
+            failures,
+            failedIds,
+        )
+    }
+
+    public suspend fun flush(limit: Int = 100): FlushResult {
+        require(limit > 0) { "limit must be positive" }
+        if (!onlineProvider()) return FlushResult(0, 0, queue.size())
+
+        val items = queue.read(limit)
+        val trackers = registry.snapshot()
+        var delivered = 0
+        for (item in items) {
+            val failedIds = deliver(item.event, item.trackerIds, trackers)
+                .mapNotNull { it.failure?.trackerId }
+            if (failedIds.isEmpty()) {
+                queue.remove(item.id)
+                delivered++
+            } else {
+                queue.replace(item.copy(trackerIds = failedIds, attempts = item.attempts + 1))
+            }
+        }
+        return FlushResult(items.size, delivered, queue.size())
+    }
+
+    private suspend fun deliver(
+        event: FlexEvent,
+        trackerIds: List<String>,
+        trackers: Map<String, Tracker>,
+    ): List<Outcome> = coroutineScope {
+        trackerIds.map { id ->
+            async {
+                val tracker = trackers[id]
+                if (tracker == null) {
+                    Outcome(id, TrackerFailure(id, IllegalStateException("tracker '$id' is unavailable")))
+                } else {
+                    try {
+                        tracker.track(event)
+                        Outcome(id)
+                    } catch (failure: Throwable) {
+                        if (failure is CancellationException) throw failure
+                        Outcome(id, TrackerFailure(id, failure))
+                    }
+                }
+            }
+        }.awaitAll()
+    }
+
+    private data class Outcome(val trackerId: String, val failure: TrackerFailure? = null)
+}

flextrack/src/main/kotlin/dev/taghizadeh/flextrack/runtime/Tracker.kt

@@ -0,0 +1,14 @@
+package dev.taghizadeh.flextrack.runtime
+
+import dev.taghizadeh.flextrack.event.FlexEvent
+
+/** A destination adapter managed by [FlexTrackClient]. */
+public interface Tracker {
+    public val id: String
+
+    public suspend fun start(): Unit = Unit
+
+    public suspend fun track(event: FlexEvent)
+
+    public suspend fun shutdown(): Unit = Unit
+}