feat: attach optional identity token to push subscription requests

[ioannisj]

💡 Motivation and Context

PostHog/posthog#72488 (draft) adds opt-in identity verification to /api/push_subscriptions: your backend mints a short-lived HS256 JWT (claims sub = distinct_id, app_id, aud = posthog:push_identity, exp) signed with the project's secret API key, and the endpoint verifies it on both POST and DELETE. Modes per integration: disabled (default), optional, required. Without SDK support there's no way to attach the token, so required mode can't be used. The provider shape was agreed with Dan in #team-messaging.

This adds the SDK half:

• New pushIdentityProvider: (distinctId, appId, completion) -> Unit constructor param on PostHogConfig (core module, no Firebase dependency). The app passes a backend-minted token (or null) to completion, from any thread, exactly once.
• The token goes out as identity_token on the register POST and unregister DELETE. When absent the key is omitted entirely, so the body stays byte-identical to today's 5-field contract and nothing changes for apps that don't opt in.
• Minting is on demand, not per attempt: the token is cached in memory per exact (distinctId, appId), reused across in-session backoff retries, and re-minted on identity change, on the next launch, and once after a 401.
• A 401 with a provider drops the cache, re-mints, and retries once; a second 401 halts for the session (record kept, next launch retries). Without a provider it's terminal, with a debug log pointing at the config hook.
• Register and unregister share one verification state: the reset() DELETE resolves a token for the old identity, the re-register POST for the new anonymous id.
• Opt-out clears the cached token, and the send guards are re-checked after the async mint.

Mirrors PostHog/posthog-ios#735. Field names, provider invocation counts, and 401 semantics are kept identical between the two SDKs.

💚 How did you test it?

Encoded the shared cross-SDK vectors 9-14 in the suite (token on both legs, key omission when absent, reset-leg tokens and invocation counts, 500-retry cache reuse, 401 refresh then terminal, 401 without a provider), plus throwing-provider, exactly-once, and foreign-thread completion tests. make testJava green (857 tests), make api dumps regenerated, ktlint clean.

📝 Checklist

• I reviewed the submitted code.
• I added tests to verify the changes.
• I updated the docs if needed.
• No breaking change or entry added to the changelog.

If releasing new changes

• Ran pnpm changeset to generate a changeset file (deliberately omitted: feature branch, not releasing yet)

🤖 Agent context

DRI: @ioannisj
Autonomy: Human-driven (agent-assisted)

[posthog]

🦔 ReviewHog reviewed this pull request

Found 1 must fix, 2 should fix, 0 consider.

Published 3 findings (view the review).

[posthog]

ReviewHog Alpha 🦔 If you find any issues helpful - please reply "valid", "invalid", etc., for evaluation purposes 🙏

[posthog]

ReviewHog Report

Changes

Issues: 3 issues

Files (5)

• posthog/api/posthog.api
• posthog/src/main/java/com/posthog/PostHogConfig.kt
• posthog/src/main/java/com/posthog/internal/PostHogApi.kt
• posthog/src/main/java/com/posthog/internal/PostHogPushSubscriptionManager.kt
• posthog/src/main/java/com/posthog/internal/PostHogPushSubscriptionRequest.kt

Other findings (outside the changed lines)

Valid issues on this PR's files that sit on lines GitHub won't let us comment on inline.

performSend sends a stale record/distinctId snapshot captured before the async identity-token resolves, silently dropping or corrupting concurrent registration updates

Priority: must_fix | File: posthog/src/main/java/com/posthog/internal/PostHogPushSubscriptionManager.kt:296-339, 96-122 | Category: bug

Why we think it's a valid issue

• Checked: The executor model (PostHog.kt:245 passes a single pushExecutor; the test uses Executors.newSingleThreadExecutor, test line 30), and traced attempt() (lines 296-304), performSend (309-338), performRegister (112-121), unregisterCurrent (234-243), resolveIdentityToken's async re-entry (449-466), and retryPending's skip guard (126-129).
• Found: attempt() claims isSending then calls resolveIdentityToken(distinctId, record.appId) { performSend(record, distinctId, token) } with record/distinctId captured in the closure (302-304). For a provider path the completion re-enters as a separate executor task via executor.executeSafely { onResolved(token) } (451), and isSending is deliberately held across the mint (comment 300-301). performSend guards only closed || config.optOut (314) — it never compares the captured record/distinctId to currentRecord()/distinctIdProvider(), and on success does pendingRecord = record.copy(deliveredForDistinctId = distinctId) and rewrites the file (329-333).
• Found: During the open mint window a concurrent register() runs performRegister → sets pendingRecord = record2, then its attempt() no-ops because the isSending CAS fails (296-298), leaving record2 unsent with no timer scheduled. When the original mint completes, performSend clobbers pendingRecord back to record1.copy(delivered=…). A later retryPending() then hits the deliveredForDistinctId == distinctIdProvider() skip (127) — since the distinctId is unchanged, the newer token is silently never sent (non-self-healing). The unregisterCurrent() variant is worse: it runs clearRecord() (pendingRecord=null, file deleted, 246-251), but the in-flight performSend then re-POSTs and revives the record as delivered, undoing the unregister.
• Found: No existing test covers this. The nearest one (opt-out during a pending mint, test lines 338-361) only exercises the optOut guard, which performSend does check — it does not exercise a record-mutating register()/unregisterCurrent() during the mint with optOut still false.
• Impact: A genuine race that corrupts shared registration state (Data loss/corruption + correctness under the keep bar), with a concrete trigger and consequence. It is gated on an async pushIdentityProvider — but that is exactly this PR's target usage (a backend JWT minted over the network is inherently async), and the concurrent triggers (FCM onNewToken refresh, logout/unregisterCurrent) are ordinary SDK operations landing in a network-sized window. The pre-PR synchronous inline send had no such window, so this is a regression the change introduces. Severity must_fix is defensible for a silent, potentially persistent registration loss / undone unregister.

Issue description

Before this PR, attempt() sent the record synchronously inline, so on the single-thread executor a concurrent register()/resendIfDistinctIdChanged() call always ran to completion only after the previous send finished (isSending released) and would then re-read currentRecord() fresh — no staleness was possible. This PR inserts an async gap: attempt() captures record and distinctId in a closure (line 302-304) and hands them to resolveIdentityToken, whose completion may fire much later from another thread (per pushIdentityProvider's documented 'any thread' / on-demand-mint contract). While that gap is open, isSending stays claimed, so if register() is called again in the meantime (e.g. an FCM token refresh, or unregisterCurrent()), performRegister sets the new pendingRecord but its nested attempt() call immediately no-ops because isSending.compareAndSet(false, true) fails (line 296-298) — the new record is never sent by that call. Worse, when the original async token mint eventually completes, performSend(record, distinctId, identityToken) runs with the stale closure values (not currentRecord()), POSTs the outdated device token/distinctId, and then overwrites pendingRecord/the pending file with record.copy(deliveredForDistinctId = distinctId) (line 330-333) — marking the stale registration as delivered. Any later retryPending()/resendIfDistinctIdChanged() will then see deliveredForDistinctId matching and skip re-sending, so the real, newer registration (or an intervening unregisterCurrent()) is silently lost until some other explicit trigger happens to fire again.

Suggested fix

Before performSend actually calls api.pushSubscription, re-validate against live state instead of trusting the captured closure: compare currentRecord() to record and distinctIdProvider() to distinctId, and if either has changed (or the record was cleared), abort this send and re-invoke attempt() against the fresh state instead of sending/persisting the stale snapshot. Alternatively, have the async completion always re-enter through attempt() (which re-reads currentRecord()/distinctIdProvider() at the top) rather than calling performSend(record, distinctId, ...) with pre-captured arguments.