diff --git a/.changeset/push-open-dedupe.md b/.changeset/push-open-dedupe.md new file mode 100644 index 000000000..beb0f02dc --- /dev/null +++ b/.changeset/push-open-dedupe.md @@ -0,0 +1,6 @@ +--- +'posthog': minor +'posthog-android': minor +--- + +Change `capturePushNotificationOpened` to skip a PostHog-sent notification open already captured in the last 5 minutes (same `invocation_id` and `action_id`), unless the payload's `google.message_id` differs. diff --git a/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt b/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt index 3910401b6..56c71a94f 100644 --- a/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt +++ b/posthog-android/src/main/java/com/posthog/android/PostHogAndroidConfig.kt @@ -31,7 +31,7 @@ import com.posthog.internal.PostHogQueue * extra). A warm-start tap arrives at `Activity.onNewIntent`, which is not observable here — * forward that intent to [PostHogAndroid.capturePushNotificationOpened], which is deduped against * this path. Foreground data messages and push delivered outside FCM need - * [PostHog.capturePushNotificationOpened], which is not. Also gates + * [PostHog.capturePushNotificationOpened], which dedupes only notifications sent by PostHog. Also gates * [PostHogAndroid.capturePushNotificationOpened]. Default: `true`. */ public open class PostHogAndroidConfig diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt index ea7cf3753..7e2ec926d 100644 --- a/posthog/src/main/java/com/posthog/PostHog.kt +++ b/posthog/src/main/java/com/posthog/PostHog.kt @@ -48,6 +48,20 @@ import java.util.concurrent.Executors private const val PUSH_NOTIFICATION_OPENED_EVENT = "\$push_notification_opened" +// A duplicate report of one tap arrives within the same launch: milliseconds after a warm tap, seconds +// after a cold start while the host's JS/Dart handlers register. Finite so that a re-send carrying no +// delivery id to tell it apart still counts once the window has passed. +private const val PUSH_OPEN_DEDUPE_WINDOW_MILLIS = 5 * 60 * 1000L + +// Only needs the opens of the last few minutes; the cap bounds memory for hosts that call this in bulk. +private const val MAX_RECENT_PUSH_OPENS = 20 + +// FCM stamps every message it delivers with its own id and puts it on the launch intent, so a payload +// built from that intent's extras identifies the delivery, not just the workflow step. `RemoteMessage.getData()` +// strips the `google.` prefix keys, so a foreground data message relayed by the host carries none — which is +// exactly the repeat this dedupe exists for. +private const val PUSH_DELIVERY_ID_KEY = "google.message_id" + public class PostHog private constructor( private val queueExecutor: ExecutorService = Executors.newSingleThreadScheduledExecutor( @@ -117,6 +131,15 @@ public class PostHog private constructor( private var logsRateCapWindowStartMillis: Long = 0 private var logsRateCapWindowCount: Int = 0 + // Recently captured PostHog push opens, keyed by `invocation_id/action_id`, oldest first. In memory + // only: both reports of one tap happen in the same launch. + private val recentPushOpens = LinkedHashMap() + + private class RecentPushOpen( + val capturedAt: Long, + val deliveryId: String?, + ) + private val remoteConfig: PostHogRemoteConfig? get() = config?.remoteConfigHolder @@ -2112,20 +2135,58 @@ public class PostHog private constructor( return } + val posthogPayload = payload?.get("posthog")?.let { posthogPayloadMap(it) } + if (!recordPushOpen(posthogPayload, payload?.get(PUSH_DELIVERY_ID_KEY) as? String)) { + return + } + val props = mutableMapOf() title?.takeIf { it.isNotEmpty() }?.let { props["\$notification_title"] = it } body?.takeIf { it.isNotEmpty() }?.let { props["\$notification_body"] = it } action?.takeIf { it.isNotEmpty() }?.let { props["\$notification_action"] = it } - payload?.get("posthog")?.let { raw -> - posthogPayloadMap(raw)?.forEach { (key, value) -> - value?.let { props["\$notification_$key"] = it } - } + posthogPayload?.forEach { (key, value) -> + value?.let { props["\$notification_$key"] = it } } capture(PUSH_NOTIFICATION_OPENED_EVENT, properties = props) } + private fun recordPushOpen( + posthogPayload: Map?, + deliveryId: String?, + ): Boolean { + val invocationId = posthogPayload?.get("invocation_id") as? String + if (invocationId.isNullOrEmpty()) { + return true + } + // Every step of one workflow run shares the run's invocation_id, so action_id tells the steps apart. + val key = "$invocationId/${posthogPayload?.get("action_id") as? String ?: ""}" + val now = config?.dateProvider?.currentTimeMillis() ?: return true + + synchronized(recentPushOpens) { + val previous = recentPushOpens[key] + if (previous != null) { + // A negative gap means the wall clock moved back; capture rather than risk dropping an open. + val insideWindow = now - previous.capturedAt in 0 until PUSH_OPEN_DEDUPE_WINDOW_MILLIS + // A re-send of the same workflow step reuses the key, so only delivery ids that are present + // on both reports and disagree prove a second notification rather than a second report of + // one tap. + val resent = previous.deliveryId != null && deliveryId != null && previous.deliveryId != deliveryId + if (insideWindow && !resent) { + config?.logger?.log("Skipped \$push_notification_opened: notification $key was already captured.") + return false + } + recentPushOpens.remove(key) + } + recentPushOpens[key] = RecentPushOpen(now, deliveryId) + if (recentPushOpens.size > MAX_RECENT_PUSH_OPENS) { + recentPushOpens.remove(recentPushOpens.keys.first()) + } + return true + } + } + /** * Coerces the `posthog` entry of a push payload into a map. FCM data maps are string→string, * so the value is accepted either as a nested [Map] or as a JSON string. Parse failures are diff --git a/posthog/src/main/java/com/posthog/PostHogInterface.kt b/posthog/src/main/java/com/posthog/PostHogInterface.kt index c19e68c47..275eda098 100644 --- a/posthog/src/main/java/com/posthog/PostHogInterface.kt +++ b/posthog/src/main/java/com/posthog/PostHogInterface.kt @@ -453,6 +453,17 @@ public interface PostHogInterface : PostHogCoreInterface { * Each key of `payload["posthog"]` (accepted as a `Map` or a JSON string) is attached as a * `$notification_` property. * + * A notification sent by PostHog is captured once: when `payload["posthog"]` carries an + * `invocation_id`, a repeat with the same `invocation_id` and `action_id` within 5 minutes of the + * first capture is skipped, whether that first capture came from this method or from the SDK's + * automatic capture. Payloads without a `posthog.invocation_id` are always captured. + * + * A rerun of that workflow, or a loop back to its push step, sends the pair again as a new + * notification, and its open counts separately: a payload whose `google.message_id` differs from the + * one captured first is captured. Forward the tapped intent's extras (which carry that id) to keep + * those apart — a payload without one, such as an FCM foreground `message.data`, is treated as a + * repeat report of the tap already captured. + * * @param title the notification title, attached as `$notification_title` when non-empty * @param body the notification body, attached as `$notification_body` when non-empty * @param payload the notification data payload; its `posthog` entry is spread into `$notification_*` props diff --git a/posthog/src/test/java/com/posthog/PostHogTest.kt b/posthog/src/test/java/com/posthog/PostHogTest.kt index 334203330..4c157c5b4 100644 --- a/posthog/src/test/java/com/posthog/PostHogTest.kt +++ b/posthog/src/test/java/com/posthog/PostHogTest.kt @@ -2,6 +2,8 @@ package com.posthog import com.posthog.internal.PostHogBatchEvent import com.posthog.internal.PostHogContext +import com.posthog.internal.PostHogDateProvider +import com.posthog.internal.PostHogDeviceDateProvider import com.posthog.internal.PostHogMemoryPreferences import com.posthog.internal.PostHogNetworkStatus import com.posthog.internal.PostHogPreferences @@ -27,10 +29,13 @@ import com.posthog.internal.PostHogThreadFactory import com.posthog.internal.errortracking.PostHogThrowable import com.posthog.vendor.uuid.TimeBasedEpochGenerator import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer import org.junit.Rule import org.junit.rules.TemporaryFolder import java.io.File +import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit import kotlin.collections.get import kotlin.test.AfterTest import kotlin.test.Test @@ -120,6 +125,7 @@ internal class PostHogTest { @AfterTest fun `set down`() { + pushOpenHttp?.shutdown() tmpDir.root.deleteRecursively() } @@ -4727,6 +4733,245 @@ internal class PostHogTest { sut.close() } + private val stepOne = """{"workflow_id":"wf-1","invocation_id":"inv-1","action_id":"step-1"}""" + private val stepTwo = """{"workflow_id":"wf-1","invocation_id":"inv-1","action_id":"step-2"}""" + private val pushOpens = CopyOnWriteArrayList() + private var pushOpenHttp: MockWebServer? = null + + private fun getPushOpenSut(optOut: Boolean = false): PostHogInterface = + getSut( + mockHttp().also { pushOpenHttp = it }.url("/").toString(), + optOut = optOut, + preloadFeatureFlags = false, + reloadFeatureFlags = false, + beforeSend = { event -> + if (event.event == "\$push_notification_opened") pushOpens.add(event) + null + }, + ) + + private fun PostHogInterface.captureAutomaticPushOpen( + posthog: Any?, + messageId: String = "m-1", + ) = capturePushNotificationOpened(payload = mapOf("google.message_id" to messageId, "posthog" to posthog)) + + private fun PostHogInterface.captureManualPushOpen(posthog: Any?) = + capturePushNotificationOpened(title = "Hello", body = "World", payload = mapOf("posthog" to posthog)) + + @Test + fun `capturePushNotificationOpened skips a manual repeat of an automatically captured PostHog push`() { + val sut = getPushOpenSut() + + sut.captureAutomaticPushOpen(stepOne) + sut.captureManualPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(1, pushOpens.size) + assertNull(pushOpens.single().properties!!["\$notification_title"]) + assertEquals("inv-1", pushOpens.single().properties!!["\$notification_invocation_id"]) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened skips an automatic repeat of a manually captured PostHog push`() { + val sut = getPushOpenSut() + + sut.captureManualPushOpen(mapOf("workflow_id" to "wf-1", "invocation_id" to "inv-1", "action_id" to "step-1")) + sut.captureAutomaticPushOpen(stepOne) + sut.captureManualPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(1, pushOpens.size) + assertEquals("Hello", pushOpens.single().properties!!["\$notification_title"]) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened captures a resend of the same workflow step`() { + val sut = getPushOpenSut() + + sut.captureAutomaticPushOpen(stepOne, messageId = "m-1") + sut.captureAutomaticPushOpen(stepOne, messageId = "m-2") + sut.captureManualPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(2, pushOpens.size) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened skips a repeat report of the same delivery`() { + val sut = getPushOpenSut() + + sut.captureAutomaticPushOpen(stepOne, messageId = "m-1") + sut.captureAutomaticPushOpen(stepOne, messageId = "m-1") + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(1, pushOpens.size) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened skips a resend when the first capture carried no delivery id`() { + val sut = getPushOpenSut() + + sut.captureManualPushOpen(stepOne) + sut.captureAutomaticPushOpen(stepOne, messageId = "m-2") + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(1, pushOpens.size) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened keys PostHog pushes by invocation and action`() { + val sut = getPushOpenSut() + val noAction = """{"workflow_id":"wf-1","invocation_id":"inv-1"}""" + val otherRun = """{"workflow_id":"wf-1","invocation_id":"inv-2","action_id":"step-1"}""" + + listOf(stepOne, stepTwo, noAction, otherRun).forEach { sut.captureAutomaticPushOpen(it) } + listOf(stepOne, stepTwo, noAction, otherRun).forEach { sut.captureManualPushOpen(it) } + queueExecutor.shutdownAndAwaitTermination() + + assertEquals( + listOf("inv-1" to "step-1", "inv-1" to "step-2", "inv-1" to null, "inv-2" to "step-1"), + pushOpens.map { + it.properties!!["\$notification_invocation_id"] to it.properties!!["\$notification_action_id"] + }, + ) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened never dedupes a push without a posthog entry`() { + val sut = getPushOpenSut() + + sut.capturePushNotificationOpened(payload = mapOf("google.message_id" to "m-1")) + sut.capturePushNotificationOpened(title = "Hello", payload = mapOf("google.message_id" to "m-1")) + sut.capturePushNotificationOpened() + sut.capturePushNotificationOpened() + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(4, pushOpens.size) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened never dedupes a push with a malformed posthog entry`() { + val sut = getPushOpenSut() + val malformed = + listOf( + "{not json", + """{"action_id":"step-1"}""", + """{"invocation_id":""}""", + """{"invocation_id":42}""", + """["inv-1"]""", + 42, + ) + + malformed.forEach { + sut.captureAutomaticPushOpen(it) + sut.captureManualPushOpen(it) + } + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(malformed.size * 2, pushOpens.size) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened captures a repeat once the dedupe window has passed`() { + val sut = getPushOpenSut() + var millis = 0L + config.dateProvider = + object : PostHogDateProvider by PostHogDeviceDateProvider() { + override fun currentTimeMillis(): Long = millis + } + + sut.captureAutomaticPushOpen(stepOne) + millis = TimeUnit.MINUTES.toMillis(5) - 1 + sut.captureManualPushOpen(stepOne) + millis = TimeUnit.MINUTES.toMillis(5) + sut.captureManualPushOpen(stepOne) + sut.captureAutomaticPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(listOf(null, "Hello"), pushOpens.map { it.properties!!["\$notification_title"] }) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened captures a repeat once the wall clock has moved backwards`() { + val sut = getPushOpenSut() + var millis = TimeUnit.MINUTES.toMillis(5) + config.dateProvider = + object : PostHogDateProvider by PostHogDeviceDateProvider() { + override fun currentTimeMillis(): Long = millis + } + + sut.captureAutomaticPushOpen(stepOne) + millis -= TimeUnit.SECONDS.toMillis(60) + sut.captureManualPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(listOf(null, "Hello"), pushOpens.map { it.properties!!["\$notification_title"] }) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened evicts the oldest open at the cap`() { + val sut = getPushOpenSut() + + repeat(21) { + sut.captureAutomaticPushOpen("""{"invocation_id":"inv-$it","action_id":"step-1"}""") + } + sut.captureManualPushOpen("""{"invocation_id":"inv-0","action_id":"step-1"}""") + sut.captureManualPushOpen("""{"invocation_id":"inv-20","action_id":"step-1"}""") + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(22, pushOpens.size) + assertEquals("inv-0", pushOpens.last().properties!!["\$notification_invocation_id"]) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened does not record a push skipped while opted out`() { + val sut = getPushOpenSut(optOut = true) + + sut.captureAutomaticPushOpen(stepOne) + sut.optIn() + sut.captureManualPushOpen(stepOne) + sut.captureAutomaticPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(listOf("Hello"), pushOpens.map { it.properties!!["\$notification_title"] }) + + sut.close() + } + + @Test + fun `capturePushNotificationOpened is a no-op after close`() { + val sut = getPushOpenSut() + sut.close() + + sut.captureAutomaticPushOpen(stepOne) + sut.captureManualPushOpen(stepOne) + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(0, pushOpens.size) + } + @Test fun `flush retries a push subscription registration deferred while offline`() { val http = mockHttp()