refactor(queue): generic PostHogQueue<Record> + EndpointSpec

turnipdabeets · claude · turnipdabeets · commit a03a1a0799c6 · 2026-05-18T22:17:08.000-04:00
Lift per-endpoint variations (codec, send, retry policy, runtime knobs) out
of inlined branches in PostHogQueue and into a single EndpointSpec object.
PostHogQueue becomes generic on Record so events, session replay, and future
endpoints share one queue implementation, three instances.

No behavior change for events or session replay. The recordUuid spec lambda
preserves the existing event.uuid filename matching.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/.changeset/generic-queue-refactor.md b/.changeset/generic-queue-refactor.md
@@ -0,0 +1,10 @@
+---
+"posthog": patch
+"posthog-android": patch
+"posthog-server": patch
+---
+
+Refactor `PostHogQueue` to be generic on `Record` and introduce `EndpointSpec`
+for per-endpoint codec, send, retry policy, and runtime knobs. No behavior
+change for events or session replay; sets up future log-record support without
+duplicating queue plumbing.
diff --git a/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt b/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt
@@ -3,6 +3,7 @@ package com.posthog.android
 import com.posthog.PostHogConfig
 import com.posthog.android.replay.PostHogReplayQueue
 import com.posthog.android.replay.PostHogSessionReplayConfig
+import com.posthog.internal.EndpointSpec
 import com.posthog.internal.PostHogApiEndpoint
 import com.posthog.internal.PostHogQueue
 
@@ -26,7 +27,12 @@ public open class PostHogAndroidConfig
             apiKey = apiKey,
             host = host,
             queueProvider = { config, api, endpoint, storagePrefix, executor ->
-                val defaultQueue = PostHogQueue(config, api, endpoint, storagePrefix, executor)
+                val spec =
+                    when (endpoint) {
+                        PostHogApiEndpoint.BATCH -> EndpointSpec.batch(config, api, storagePrefix)
+                        PostHogApiEndpoint.SNAPSHOT -> EndpointSpec.snapshot(config, api, storagePrefix)
+                    }
+                val defaultQueue = PostHogQueue(config, spec, executor)
                 if (endpoint == PostHogApiEndpoint.SNAPSHOT) {
                     val replayQueue = PostHogReplayQueue(config, defaultQueue, storagePrefix, executor)
                     (config as? PostHogAndroidConfig)?.replayQueueHolder = replayQueue
diff --git a/posthog-android/src/main/java/com/posthog/android/replay/PostHogReplayBufferQueue.kt b/posthog-android/src/main/java/com/posthog/android/replay/PostHogReplayBufferQueue.kt
@@ -90,8 +90,8 @@ internal class PostHogReplayBufferQueue(
      *
      * Returns the number of events successfully migrated.
      */
-    fun migrateAllTo(targetQueue: PostHogQueueInterface): Int {
-        if (targetQueue !is PostHogQueue) {
+    fun migrateAllTo(targetQueue: PostHogQueueInterface<PostHogEvent>): Int {
+        if (targetQueue !is PostHogQueue<*>) {
             config.logger.log("Replay buffer migration skipped: target queue is not PostHogQueue")
             return 0
         }
diff --git a/posthog-android/src/main/java/com/posthog/android/replay/PostHogReplayQueue.kt b/posthog-android/src/main/java/com/posthog/android/replay/PostHogReplayQueue.kt
@@ -19,10 +19,10 @@ import java.util.concurrent.ExecutorService
  */
 internal class PostHogReplayQueue internal constructor(
     private val config: PostHogConfig,
-    private val innerQueue: PostHogQueueInterface,
+    private val innerQueue: PostHogQueueInterface<PostHogEvent>,
     replayStoragePrefix: String?,
     private val executor: ExecutorService,
-) : PostHogQueueInterface {
+) : PostHogQueueInterface<PostHogEvent> {
     private val replayDir = replayStoragePrefix?.let { File(it, config.apiKey) }
 
     private val bufferQueue: PostHogReplayBufferQueue =
diff --git a/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayBufferQueueTest.kt b/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayBufferQueueTest.kt
@@ -3,8 +3,8 @@ package com.posthog.android.replay
 import com.posthog.PostHogConfig
 import com.posthog.PostHogEvent
 import com.posthog.android.API_KEY
+import com.posthog.internal.EndpointSpec
 import com.posthog.internal.PostHogApi
-import com.posthog.internal.PostHogApiEndpoint
 import com.posthog.internal.PostHogQueue
 import com.posthog.internal.PostHogQueueInterface
 import org.junit.Rule
@@ -33,7 +33,7 @@ internal class PostHogReplayBufferQueueTest {
 
     private val executors = mutableListOf<ExecutorService>()
 
-    private class FakeQueue : PostHogQueueInterface {
+    private class FakeQueue : PostHogQueueInterface<PostHogEvent> {
         val events = mutableListOf<PostHogEvent>()
 
         override fun add(event: PostHogEvent) {
@@ -143,12 +143,10 @@ internal class PostHogReplayBufferQueueTest {
         config: PostHogConfig,
         storagePrefix: String,
         executor: ExecutorService = createExecutor(),
-    ): PostHogQueue {
+    ): PostHogQueue<PostHogEvent> {
         return PostHogQueue(
             config,
-            PostHogApi(config),
-            PostHogApiEndpoint.SNAPSHOT,
-            storagePrefix,
+            EndpointSpec.snapshot(config, PostHogApi(config), storagePrefix),
             executor,
         )
     }
diff --git a/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayIntegrationTest.kt b/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayIntegrationTest.kt
@@ -44,7 +44,7 @@ internal class PostHogReplayIntegrationTest {
     private val context = mock<Context>()
     private val replayExecutors = mutableListOf<ExecutorService>()
 
-    private class FakeQueue : PostHogQueueInterface {
+    private class FakeQueue : PostHogQueueInterface<PostHogEvent> {
         override fun add(event: PostHogEvent) {
         }
 
diff --git a/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayQueueTest.kt b/posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayQueueTest.kt
@@ -3,8 +3,8 @@ package com.posthog.android.replay
 import com.posthog.PostHogConfig
 import com.posthog.PostHogEvent
 import com.posthog.android.API_KEY
+import com.posthog.internal.EndpointSpec
 import com.posthog.internal.PostHogApi
-import com.posthog.internal.PostHogApiEndpoint
 import com.posthog.internal.PostHogQueue
 import com.posthog.internal.PostHogQueueInterface
 import org.junit.Rule
@@ -34,7 +34,7 @@ internal class PostHogReplayQueueTest {
         executors.clear()
     }
 
-    private class FakeQueue : PostHogQueueInterface {
+    private class FakeQueue : PostHogQueueInterface<PostHogEvent> {
         val events = mutableListOf<PostHogEvent>()
         var flushCallCount = 0
         var startCallCount = 0
@@ -160,9 +160,7 @@ internal class PostHogReplayQueueTest {
         val innerQueue =
             PostHogQueue(
                 config,
-                PostHogApi(config),
-                PostHogApiEndpoint.SNAPSHOT,
-                storagePrefix,
+                EndpointSpec.snapshot(config, PostHogApi(config), storagePrefix),
                 executor,
             )
         val queue = PostHogReplayQueue(config, innerQueue, storagePrefix, executor)
diff --git a/posthog-server/src/main/java/com/posthog/server/internal/PostHogMemoryQueue.kt b/posthog-server/src/main/java/com/posthog/server/internal/PostHogMemoryQueue.kt
@@ -31,7 +31,7 @@ internal class PostHogMemoryQueue(
     private val executor: ExecutorService,
     private val retryDelaySeconds: Int = DEFAULT_RETRY_DELAY_SECONDS,
     private val maxRetryDelaySeconds: Int = DEFAULT_MAX_RETRY_DELAY_SECONDS,
-) : PostHogQueueInterface {
+) : PostHogQueueInterface<PostHogEvent> {
     private val events: ArrayDeque<PostHogEvent> = ArrayDeque()
     private val eventsLock = Any()
     private val timerLock = Any()
diff --git a/posthog/api/posthog.api b/posthog/api/posthog.api
@@ -488,6 +488,17 @@ public final class com/posthog/errortracking/PostHogErrorTrackingConfig {
 	public final fun setAutoCapture (Z)V
 }
 
+public final class com/posthog/internal/EndpointSpec {
+	public static final field Companion Lcom/posthog/internal/EndpointSpec$Companion;
+	public static final fun batch (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/PostHogApi;Ljava/lang/String;)Lcom/posthog/internal/EndpointSpec;
+	public static final fun snapshot (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/PostHogApi;Ljava/lang/String;)Lcom/posthog/internal/EndpointSpec;
+}
+
+public final class com/posthog/internal/EndpointSpec$Companion {
+	public final fun batch (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/PostHogApi;Ljava/lang/String;)Lcom/posthog/internal/EndpointSpec;
+	public final fun snapshot (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/PostHogApi;Ljava/lang/String;)Lcom/posthog/internal/EndpointSpec;
+}
+
 public final class com/posthog/internal/EvaluationReason {
 	public fun <init> (Ljava/lang/String;Ljava/lang/String;Ljava/lang/Integer;)V
 	public final fun component1 ()Ljava/lang/String;
@@ -832,8 +843,8 @@ public final class com/posthog/internal/PostHogPrintLogger : com/posthog/interna
 }
 
 public final class com/posthog/internal/PostHogQueue : com/posthog/internal/PostHogQueueInterface {
-	public fun <init> (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/PostHogApi;Lcom/posthog/internal/PostHogApiEndpoint;Ljava/lang/String;Ljava/util/concurrent/ExecutorService;)V
-	public fun add (Lcom/posthog/PostHogEvent;)V
+	public fun <init> (Lcom/posthog/PostHogConfig;Lcom/posthog/internal/EndpointSpec;Ljava/util/concurrent/ExecutorService;)V
+	public fun add (Ljava/lang/Object;)V
 	public fun clear ()V
 	public fun flush ()V
 	public final fun getQueueDirectory ()Ljava/io/File;
@@ -843,7 +854,7 @@ public final class com/posthog/internal/PostHogQueue : com/posthog/internal/Post
 }
 
 public abstract interface class com/posthog/internal/PostHogQueueInterface {
-	public abstract fun add (Lcom/posthog/PostHogEvent;)V
+	public abstract fun add (Ljava/lang/Object;)V
 	public abstract fun clear ()V
 	public abstract fun flush ()V
 	public abstract fun start ()V
diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt
@@ -62,7 +62,7 @@ public class PostHog private constructor(
     private val featureFlagsCalledLock = Any()
     private val cachedPersonPropertiesLock = Any()
 
-    private var replayQueue: PostHogQueueInterface? = null
+    private var replayQueue: PostHogQueueInterface<PostHogEvent>? = null
 
     private val remoteConfig: PostHogRemoteConfig?
         get() = config?.remoteConfigHolder
diff --git a/posthog/src/main/java/com/posthog/PostHogConfig.kt b/posthog/src/main/java/com/posthog/PostHogConfig.kt
@@ -1,6 +1,7 @@
 package com.posthog
 
 import com.posthog.errortracking.PostHogErrorTrackingConfig
+import com.posthog.internal.EndpointSpec
 import com.posthog.internal.PostHogApi
 import com.posthog.internal.PostHogApiEndpoint
 import com.posthog.internal.PostHogContext
@@ -258,8 +259,21 @@ public open class PostHogConfig(
     /**
      * Factory to instantiate a custom queue implementation.
      */
-    public val queueProvider: (PostHogConfig, PostHogApi, PostHogApiEndpoint, String?, ExecutorService) -> PostHogQueueInterface =
-        { config, api, endpoint, storagePrefix, executor -> PostHogQueue(config, api, endpoint, storagePrefix, executor) },
+    public val queueProvider: (
+        PostHogConfig,
+        PostHogApi,
+        PostHogApiEndpoint,
+        String?,
+        ExecutorService,
+    ) -> PostHogQueueInterface<PostHogEvent> =
+        { config, api, endpoint, storagePrefix, executor ->
+            val spec =
+                when (endpoint) {
+                    PostHogApiEndpoint.BATCH -> EndpointSpec.batch(config, api, storagePrefix)
+                    PostHogApiEndpoint.SNAPSHOT -> EndpointSpec.snapshot(config, api, storagePrefix)
+                }
+            PostHogQueue(config, spec, executor)
+        },
     /**
      * Configuration for PostHog Error Tracking feature.
      */
diff --git a/posthog/src/main/java/com/posthog/PostHogStateless.kt b/posthog/src/main/java/com/posthog/PostHogStateless.kt
@@ -39,7 +39,7 @@ public open class PostHogStateless protected constructor(
     protected var config: PostHogConfig? = null
 
     protected var featureFlags: PostHogFeatureFlagsInterface? = null
-    protected var queue: PostHogQueueInterface? = null
+    protected var queue: PostHogQueueInterface<PostHogEvent>? = null
     protected var memoryPreferences: PostHogPreferences = PostHogMemoryPreferences()
     protected val throwableCoercer: ThrowableCoercer = ThrowableCoercer()
 
diff --git a/posthog/src/main/java/com/posthog/internal/EndpointSpec.kt b/posthog/src/main/java/com/posthog/internal/EndpointSpec.kt
@@ -0,0 +1,92 @@
+package com.posthog.internal
+
+import com.posthog.PostHogConfig
+import com.posthog.PostHogEvent
+import com.posthog.PostHogInternal
+import java.io.InputStream
+import java.io.OutputStream
+import java.util.UUID
+
+/**
+ * Per-endpoint specification consumed by [PostHogQueue]. Carries
+ * everything that differs between PostHog endpoints (`/batch`, `/snapshot`,
+ * and any future endpoint): the codec, the send call, the retry policy,
+ * and the per-endpoint runtime knobs.
+ *
+ * The queue itself stays record-type-agnostic; everything that varies
+ * between endpoints lives here.
+ */
+@PostHogInternal
+public class EndpointSpec<Record> internal constructor(
+    internal val name: String,
+    internal val storagePrefix: String?,
+    internal val initialCap: Int,
+    internal val initialFlushAt: Int,
+    internal val maxQueueSize: Int,
+    internal val flushIntervalSeconds: Int,
+    internal val encode: (Record, OutputStream) -> Unit,
+    internal val decode: (InputStream) -> Record?,
+    internal val describe: (Record) -> String,
+    internal val send: (List<Record>) -> Unit,
+    internal val isRetriableStatusCode: (Int) -> Boolean,
+    internal val isFatalRecord: (Record) -> Boolean = { false },
+    internal val recordUuid: (Record) -> UUID? = { null },
+) {
+    public companion object {
+        @JvmStatic
+        public fun batch(
+            config: PostHogConfig,
+            api: PostHogApi,
+            storagePrefix: String?,
+        ): EndpointSpec<PostHogEvent> =
+            EndpointSpec(
+                name = "events",
+                storagePrefix = storagePrefix,
+                initialCap = config.maxBatchSize,
+                initialFlushAt = config.flushAt,
+                maxQueueSize = config.maxQueueSize,
+                flushIntervalSeconds = config.flushIntervalSeconds,
+                encode = { event, stream ->
+                    config.serializer.serialize(event, stream.writer().buffered())
+                },
+                decode = { stream ->
+                    config.serializer.deserialize<PostHogEvent?>(stream.reader().buffered())
+                },
+                describe = { event -> "Event ${event.event}" },
+                send = { events -> api.batch(events) },
+                isRetriableStatusCode = ::isEventsRetriableStatusCode,
+                isFatalRecord = { it.isFatalExceptionEvent() },
+                recordUuid = { it.uuid },
+            )
+
+        @JvmStatic
+        public fun snapshot(
+            config: PostHogConfig,
+            api: PostHogApi,
+            storagePrefix: String?,
+        ): EndpointSpec<PostHogEvent> =
+            EndpointSpec(
+                name = "snapshots",
+                storagePrefix = storagePrefix,
+                initialCap = config.maxBatchSize,
+                initialFlushAt = config.flushAt,
+                maxQueueSize = config.maxQueueSize,
+                flushIntervalSeconds = config.flushIntervalSeconds,
+                encode = { event, stream ->
+                    config.serializer.serialize(event, stream.writer().buffered())
+                },
+                decode = { stream ->
+                    config.serializer.deserialize<PostHogEvent?>(stream.reader().buffered())
+                },
+                describe = { _ -> "snapshot" },
+                send = { events -> api.snapshot(events) },
+                isRetriableStatusCode = ::isEventsRetriableStatusCode,
+                isFatalRecord = { it.isFatalExceptionEvent() },
+                recordUuid = { it.uuid },
+            )
+    }
+}
+
+internal val DEFAULT_EVENTS_RETRYABLE_STATUS_CODES: Set<Int> = setOf(429, 500, 502, 503, 504)
+
+internal fun isEventsRetriableStatusCode(code: Int): Boolean = code in DEFAULT_EVENTS_RETRYABLE_STATUS_CODES
diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt
diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueueInterface.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueueInterface.kt
diff --git a/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt b/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt
diff --git a/posthog/src/test/java/com/posthog/internal/EndpointSpecTest.kt b/posthog/src/test/java/com/posthog/internal/EndpointSpecTest.kt
diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt

Original file line number	Diff line number	Diff line change
`@@ -90,8 +90,8 @@ internal class PostHogReplayBufferQueue(`
`90`	`90`	`*`
`91`	`91`	`* Returns the number of events successfully migrated.`
`92`	`92`	`*/`
`93`		`- fun migrateAllTo(targetQueue: PostHogQueueInterface): Int {`
`94`		`- if (targetQueue !is PostHogQueue) {`
	`93`	`+ fun migrateAllTo(targetQueue: PostHogQueueInterface<PostHogEvent>): Int {`
	`94`	`+ if (targetQueue !is PostHogQueue<*>) {`
`95`	`95`	`config.logger.log("Replay buffer migration skipped: target queue is not PostHogQueue")`
`96`	`96`	`return 0`
`97`	`97`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ internal class PostHogReplayIntegrationTest {`
`44`	`44`	`private val context = mock<Context>()`
`45`	`45`	`private val replayExecutors = mutableListOf<ExecutorService>()`
`46`	`46`
`47`		`- private class FakeQueue : PostHogQueueInterface {`
	`47`	`+ private class FakeQueue : PostHogQueueInterface<PostHogEvent> {`
`48`	`48`	`override fun add(event: PostHogEvent) {`
`49`	`49`	`}`
`50`	`50`