diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts index c3ec5f155958..9c06ec09e7ec 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts @@ -8,6 +8,7 @@ import Constants from "expo-constants"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import { Cookies, FetchHttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; import { ManagedRelay } from "@t3tools/client-runtime/relay"; @@ -391,26 +392,144 @@ describe("makeRelayDeviceRegistrationRequest", () => { ).toEqual({ liveActivitiesEnabled: true, baseFontSize: 18 }); }); - it.effect("registers at most one listener while a Live Activity push token is pending", () => { + it.effect("registers one push-token listener per Live Activity across wrappers", () => { registerAgentAwarenessConnection(savedConnection()); const addPushTokenListener = vi.fn(); - const activity = { + // getInstances() returns a new wrapper for the same native activity each call. + const wrapper = () => ({ + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve(null)), addPushTokenListener, - }; + }); return Effect.gen(function* () { - expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(false); - expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(false); + expect(yield* registerLiveActivityPushToken({ activity: wrapper() as never })).toBe(false); + expect(yield* registerLiveActivityPushToken({ activity: wrapper() as never })).toBe(false); - expect(activity.getPushToken).toHaveBeenCalledTimes(2); expect(addPushTokenListener).toHaveBeenCalledTimes(1); }).pipe(Effect.provide(layerRelayTest)); }); + it.effect("registers a Live Activity push token that rotates after registration", () => { + const registeredTokens: Array = []; + vi.stubGlobal("fetch", async (request: RequestInfo | URL, init?: RequestInit) => { + const url = request instanceof Request ? request.url : String(request); + if (url.endsWith("/v1/client/dpop-token")) { + return Response.json({ + access_token: "relay-dpop-token", + issued_token_type: "urn:ietf:params:oauth:token-type:access_token", + token_type: "DPoP", + expires_in: 300, + scope: "mobile:registration", + }); + } + if (url.endsWith("/v1/mobile/live-activities")) { + const body = request instanceof Request ? await request.text() : String(init?.body); + registeredTokens.push( + (JSON.parse(body) as { readonly activityPushToken: string }).activityPushToken, + ); + } + return Response.json({ ok: true }); + }); + Constants.expoConfig!.extra = { relay: { url: "https://relay.example.test/" } }; + setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk-token-user-a")); + let tokenListener: ((event: { readonly pushToken: string }) => void) | undefined; + const activity = { + getId: () => "activity-1", + getPushToken: vi.fn(() => Promise.resolve("activity-token")), + addPushTokenListener: vi.fn((listener: typeof tokenListener) => { + tokenListener = listener; + return { remove: vi.fn() }; + }), + }; + + return Effect.gen(function* () { + yield* runBackgroundOperations(); + expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(true); + + tokenListener?.({ pushToken: "rotated-activity-token" }); + yield* runBackgroundOperations(); + + expect(registeredTokens).toEqual(["activity-token", "rotated-activity-token"]); + }).pipe( + Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch), + Effect.provide(layerRelayTest), + ); + }); + + it.effect("registers a rotated Live Activity token only after an older registration", () => { + const arrivals: Array = []; + let openOldRequest: (() => void) | undefined; + const oldRequestGate = new Promise((resolve) => { + openOldRequest = resolve; + }); + let markOldRequestArrived: (() => void) | undefined; + const oldRequestArrived = new Promise((resolve) => { + markOldRequestArrived = resolve; + }); + vi.stubGlobal("fetch", async (request: RequestInfo | URL, init?: RequestInit) => { + const url = request instanceof Request ? request.url : String(request); + if (url.endsWith("/v1/client/dpop-token")) { + return Response.json({ + access_token: "relay-dpop-token", + issued_token_type: "urn:ietf:params:oauth:token-type:access_token", + token_type: "DPoP", + expires_in: 300, + scope: "mobile:registration", + }); + } + if (url.endsWith("/v1/mobile/live-activities")) { + const body = request instanceof Request ? await request.text() : String(init?.body); + const token = (JSON.parse(body) as { readonly activityPushToken: string }) + .activityPushToken; + arrivals.push(token); + if (token === "activity-token") { + markOldRequestArrived?.(); + await oldRequestGate; + } + } + return Response.json({ ok: true }); + }); + Constants.expoConfig!.extra = { relay: { url: "https://relay.example.test/" } }; + setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk-token-user-a")); + let tokenListener: ((event: { readonly pushToken: string }) => void) | undefined; + const activity = { + getId: () => "activity-1", + getPushToken: vi.fn(() => Promise.resolve("activity-token")), + addPushTokenListener: vi.fn((listener: typeof tokenListener) => { + tokenListener = listener; + return { remove: vi.fn() }; + }), + }; + + return Effect.gen(function* () { + yield* runBackgroundOperations(); + const lookup = yield* Effect.forkChild( + registerLiveActivityPushToken({ activity: activity as never }), + ); + yield* Effect.promise(() => oldRequestArrived); + + tokenListener?.({ pushToken: "rotated-activity-token" }); + const listenerRegistration = yield* Effect.forkChild(runBackgroundOperations()); + for (let pass = 0; pass < 5; pass++) { + yield* Effect.promise(() => new Promise((resolve) => setImmediate(resolve))); + } + expect(arrivals).toEqual(["activity-token"]); + + openOldRequest?.(); + yield* Fiber.join(lookup); + yield* Fiber.join(listenerRegistration); + expect(arrivals).toEqual(["activity-token", "rotated-activity-token"]); + }).pipe( + Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch), + Effect.provide(layerRelayTest), + ); + }); + it.effect("preserves Live Activity push-token lookup failures", () => { const cause = new Error("native token lookup failed"); const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.reject(cause)), addPushTokenListener: vi.fn(), }; @@ -435,6 +554,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { () => { registerAgentAwarenessConnection(savedConnection()); const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), }; @@ -449,6 +569,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { "registers APNS-started Live Activities for relay updates without mutating them locally", () => { const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), start: vi.fn(), @@ -473,6 +594,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { "re-registers active Live Activity tokens when the app returns to the foreground", () => { const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), }; @@ -502,6 +624,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { it("ends local Live Activities and stops foreground reconciliation on cloud sign-out", () => { const end = vi.fn(() => Promise.resolve()); const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), end, @@ -613,6 +736,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { it("releases the provider without ending activities or clearing the registration", () => { const end = vi.fn(() => Promise.resolve()); const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), end, @@ -708,6 +832,7 @@ describe("makeRelayDeviceRegistrationRequest", () => { }, }; const activity = { + getId: () => "activity-1", getPushToken: vi.fn(() => Promise.resolve("activity-token")), addPushTokenListener: vi.fn(), }; diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts index 8212b02965cc..cccb7056cb2b 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts @@ -7,6 +7,7 @@ import Constants from "expo-constants"; import * as Notifications from "expo-notifications"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import { AppState, Platform } from "react-native"; import type { EnvironmentId } from "@t3tools/contracts"; import { @@ -72,7 +73,12 @@ export class AgentAwarenessOperationError extends Schema.TaggedError(); -const activityPushTokenListeners = new WeakSet>(); +// Keyed by ActivityKit id: getAgentLiveActivities() wraps the same native +// activity in a new object on every call. The relay keeps only the latest token +// a device registers, so each activity's token reads and registrations run one +// at a time; otherwise a lookup that read the old token could land after the +// rotated one. +const activityPushTokenLocks = new Map(); // But the registration is not a pure no-op: the relay replays the current // aggregate to this device on every accepted registration, and that replay is // the foreground reconciliation that repairs drifted or orphaned activities. @@ -893,6 +899,7 @@ export function __resetAgentAwarenessRemoteRegistrationForTest(): void { androidDeviceReplayedAt = null; registrationStatusListeners.clear(); registeredActivityPushTokens.clear(); + activityPushTokenLocks.clear(); } export function unregisterAgentAwarenessDeviceForCurrentUser( @@ -928,52 +935,53 @@ export function registerLiveActivityPushToken(input: { return false; } - const activityPushToken = yield* Effect.tryPromise({ - try: () => input.activity.getPushToken(), - catch: (cause) => - new AgentAwarenessOperationError({ - operation: "read-live-activity-push-token", - cause, - }), - }); - if (!activityPushToken) { - if (activityPushTokenListeners.has(input.activity)) { - logRegistrationDebug( - "live activity push token not available yet; token listener already registered", - { - connectionCount: environmentConnections.size, - }, - ); - return false; - } - - logRegistrationDebug( - "live activity push token not available yet; listening for token event", - { - connectionCount: environmentConnections.size, - }, - ); - activityPushTokenListeners.add(input.activity); - input.activity.addPushTokenListener((event) => { - if (event.pushToken) { - logRegistrationDebug("live activity push token event received", { - tokenSuffix: event.pushToken.slice(-8), - }); - runRegistrationInBackground( - registerLiveActivityPushTokenValue({ - activityPushToken: event.pushToken, + const lock = observeLiveActivityPushTokens(input.activity); + return yield* lock.withPermit( + Effect.gen(function* () { + const activityPushToken = yield* Effect.tryPromise({ + try: () => input.activity.getPushToken(), + catch: (cause) => + new AgentAwarenessOperationError({ + operation: "read-live-activity-push-token", + cause, }), - "live activity token listener registration failed", + }); + if (!activityPushToken) { + logRegistrationDebug( + "live activity push token not available yet; waiting for token event", + { connectionCount: environmentConnections.size }, ); + return false; } - }); - return false; - } - return yield* registerLiveActivityPushTokenValue({ - activityPushToken, + return yield* registerLiveActivityPushTokenValue({ activityPushToken }); + }), + ); + }); +} + +// iOS can replace an activity's push token while it runs, not only deliver a +// late first one. The relay keeps pushing to the old token until APNs rejects +// it, so every activity needs a listener that registers each new token. +function observeLiveActivityPushTokens( + activity: LiveActivity, +): Semaphore.Semaphore { + const activityId = activity.getId(); + const existing = activityPushTokenLocks.get(activityId); + if (existing) return existing; + const lock = Semaphore.makeUnsafe(1); + activityPushTokenLocks.set(activityId, lock); + activity.addPushTokenListener((event) => { + if (!event.pushToken) return; + logRegistrationDebug("live activity push token event received", { + tokenSuffix: event.pushToken.slice(-8), }); + runRegistrationInBackground( + lock.withPermit(registerLiveActivityPushTokenValue({ activityPushToken: event.pushToken })), + "live activity token listener registration failed", + ); }); + return lock; } function registerLiveActivityPushTokenValue(input: {