[SDK] Add refreshSession and migrate the session-pinned infra (#11381)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/lib/api/index.ts

@@ -32,7 +32,6 @@ import {app, com} from '#/lexicons'
 import * as bsky from '#/types/bsky'
 import {createGIFDescription} from '../gif-alt-text'
 import {computeCid} from './computeCid'
-import {fromLegacyBlobRef} from './legacy-blob'
 import {uploadBlob} from './upload-blob'
 
 export {uploadBlob}
@@ -410,10 +409,11 @@ async function resolveMedia(
     return {
       $type: 'app.bsky.embed.video',
       /*
-       * The video pipeline still reads its blob off the legacy agent, so
-       * normalize it to the lex shape before it reaches the lex write.
+       * The video blob is a plain lex blob from the video pipeline
+       * (getJobStatus, in composer state/video). Its structural shape matches
+       * the lexicon blob field and the CID hasher (see computeCid).
        */
-      video: fromLegacyBlobRef(videoDraft.pendingPublish.blobRef),
+      video: videoDraft.pendingPublish.blobRef,
       alt: videoDraft.altText || undefined,
       captions: captions.length === 0 ? undefined : captions,
       aspectRatio,

src/lib/api/legacy-blob.ts

@@ -14,18 +14,3 @@ import {type BlobRef as LexBlobRef} from '@atproto/lex'
 export function toLegacyBlobRef(blob: LexBlobRef): BlobRef {
   return BlobRef.fromJsonRef(blob as Parameters<typeof BlobRef.fromJsonRef>[0])
 }
-
-/**
- * Normalize a legacy `BlobRef` class instance to the plain-JSON lex blob shape.
- *
- * Required for any blob that reaches a lex write: the lex serializer walks
- * plain objects, so a class instance goes on the wire with its internal
- * `original` field and no `$type`. `ipld()` yields exactly the lex shape, and
- * hashes identically (see `computeCid.test.ts` case 2b).
- *
- * Only the video pipeline still needs this - it reads its blob off the legacy
- * agent (`app.bsky.video.getJobStatus`). Drop it when the video client moves.
- */
-export function fromLegacyBlobRef(blob: BlobRef): LexBlobRef {
-  return blob.ipld()
-}

src/lib/api/resolve.ts

@@ -184,7 +184,7 @@ export async function resolveLink(
       view: res.data.starterPack,
     }
   }
-  return resolveExternal(agent, uri)
+  return resolveExternal(uri)
 
   // Forked from useGetPost. TODO: move into RQ.
   async function getPost({uri}: {uri: string}) {
@@ -258,11 +258,8 @@ function getFileSlug(url: string | undefined): string | undefined {
   return dotIndex > 0 ? filename.slice(0, dotIndex) : undefined
 }
 
-async function resolveExternal(
-  agent: AtpAgent,
-  uri: string,
-): Promise<ResolvedExternalLink> {
-  const result = await getLinkMeta(agent, uri)
+async function resolveExternal(uri: string): Promise<ResolvedExternalLink> {
+  const result = await getLinkMeta(uri)
   return {
     type: 'external',
     uri: result.url,

src/lib/constants.ts

@@ -279,9 +279,14 @@ export const DM_SERVICE_HEADERS = {
   'atproto-proxy': `${CHAT_PROXY_DID}#bsky_chat`,
 }
 
-export const BLUESKY_NOTIF_SERVICE_HEADERS = {
-  'atproto-proxy': `${BLUESKY_PROXY_DID}#bsky_notif`,
-}
+/**
+ * The notification service's proxy target, in the `did#service_id` form a lex
+ * client's per-call `service` option takes. Passing it emits `atproto-proxy:
+ * <this value>` on that one request, which is what routes push registration to
+ * the notification service (replaces the old
+ * `BLUESKY_NOTIF_SERVICE_HEADERS`).
+ */
+export const NOTIF_SERVICE: Service = `${BLUESKY_PROXY_DID}#bsky_notif`
 
 export const webLinks = {
   tos: `https://bsky.social/about/support/tos`,

src/lib/link-meta/link-meta.ts

@@ -1,4 +1,4 @@
-import {type AppBskyEmbedExternal, type AtpAgent} from '@atproto/api'
+import {type AppBskyEmbedExternal} from '@atproto/api'
 
 import {LINK_META_PROXY} from '#/lib/constants'
 import {getGiphyMetaUri} from '#/lib/strings/embed-player'
@@ -31,7 +31,6 @@ export interface LinkMeta {
 }
 
 export async function getLinkMeta(
-  agent: AtpAgent,
   url: string,
   timeout = 15e3,
 ): Promise<LinkMeta> {
@@ -80,9 +79,7 @@ export async function getLinkMeta(
 
   try {
     const response = await fetch(
-      `${LINK_META_PROXY(agent.serviceUrl.toString() || '')}${encodeURIComponent(
-        url,
-      )}`,
+      `${LINK_META_PROXY('')}${encodeURIComponent(url)}`,
       {signal: controller.signal},
     )
 

src/lib/media/video/multipart/types.ts

@@ -1,3 +1,5 @@
+import {type app} from '#/lexicons'
+
 /**
  * One part of a multipart upload. `partNumber` is 1-indexed to match the S3
  * convention the backend uses.
@@ -57,13 +59,13 @@ export type UploadStatusResponse = {
   expiresAt: string
   state: UploadState
   completedJobId?: string
-  jobStatus?: import('@atproto/api').AppBskyVideoDefs.JobStatus
+  jobStatus?: app.bsky.video.defs.JobStatus
   failureReason?: string
 }
 
 export type FinishUploadResponse = {
   completedJobId: string
-  jobStatus: import('@atproto/api').AppBskyVideoDefs.JobStatus
+  jobStatus: app.bsky.video.defs.JobStatus
 }
 
 export type AbortUploadResponse = Pick<

src/lib/media/video/multipart/upload.ts

@@ -1,9 +1,10 @@
-import {type AppBskyVideoDefs, type AtpAgent} from '@atproto/api'
+import {type Client} from '@atproto/lex'
 import {nanoid} from 'nanoid/non-secure'
 
 import {AbortError} from '#/lib/async/cancelable'
 import {type CompressedVideo} from '#/lib/media/video/types'
 import {shouldRetryError} from '#/lib/strings/errors'
+import {type app} from '#/lexicons'
 import {getServiceAuthToken} from '../upload.shared'
 import {mimeToExt} from '../util'
 import {
@@ -29,19 +30,22 @@ export class MultipartFallbackError extends Error {}
 
 export async function uploadVideoMultipart({
   video,
-  agent,
+  client,
+  dispatchUrl,
   setProgress,
   signal,
   onStarted,
 }: {
   video: CompressedVideo
-  agent: AtpAgent
+  client: Client
+  /** The account's PDS/dispatch URL, for the uploadBlob service-auth token. */
+  dispatchUrl: string | URL
   setProgress: (progress: number) => void
   signal: AbortSignal
   onStarted?: () => void
-}): Promise<AppBskyVideoDefs.JobStatus> {
+}): Promise<app.bsky.video.defs.JobStatus> {
   throwIfAborted(signal)
-  const tokenProvider = createTokenProvider(agent, signal)
+  const tokenProvider = createTokenProvider(client, dispatchUrl, signal)
   const token = await tokenProvider.get()
   const name = `${nanoid(12)}.${mimeToExt(video.mimeType)}`
   let session
@@ -134,7 +138,7 @@ async function finishAndRecover({
   getToken: (forceRefresh?: boolean) => Promise<string>
   signal: AbortSignal
   resendMissingParts: (receivedPartNumbers: number[]) => Promise<boolean>
-}): Promise<AppBskyVideoDefs.JobStatus> {
+}): Promise<app.bsky.video.defs.JobStatus> {
   let createdFailures = 0
   let forceTokenRefresh = true
   while (true) {
@@ -224,7 +228,7 @@ async function abortThenFallbackOrResolve(
   jobId: string,
   token: string,
   cause: unknown,
-): Promise<AppBskyVideoDefs.JobStatus> {
+): Promise<app.bsky.video.defs.JobStatus> {
   const result = await abortUploadWithRetry(jobId, token)
   if (result.state === 'aborted') {
     throw new MultipartFallbackError(
@@ -264,7 +268,11 @@ async function abortUploadWithRetry(jobId: string, token: string) {
   throw lastError
 }
 
-function createTokenProvider(agent: AtpAgent, signal: AbortSignal) {
+function createTokenProvider(
+  client: Client,
+  dispatchUrl: string | URL,
+  signal: AbortSignal,
+) {
   let token: string | undefined
   let expiresAt = 0
   let refresh: Promise<string> | undefined
@@ -273,7 +281,7 @@ function createTokenProvider(agent: AtpAgent, signal: AbortSignal) {
     if (!forceRefresh && token && Date.now() < expiresAt - 60_000) return token
     if (!refresh) {
       const exp = Math.floor(Date.now() / 1000) + 60 * 30
-      refresh = getServiceAuthTokenWithRetry(agent, exp, signal)
+      refresh = getServiceAuthTokenWithRetry(client, dispatchUrl, exp, signal)
         .then(nextToken => {
           token = nextToken
           expiresAt = exp * 1000
@@ -290,7 +298,8 @@ function createTokenProvider(agent: AtpAgent, signal: AbortSignal) {
 }
 
 async function getServiceAuthTokenWithRetry(
-  agent: AtpAgent,
+  client: Client,
+  dispatchUrl: string | URL,
   exp: number,
   signal: AbortSignal,
 ) {
@@ -299,7 +308,8 @@ async function getServiceAuthTokenWithRetry(
     throwIfAborted(signal)
     try {
       return await getServiceAuthToken({
-        agent,
+        client,
+        dispatchUrl,
         lxm: 'com.atproto.repo.uploadBlob',
         exp,
       })

src/lib/media/video/upload.shared.ts

@@ -1,44 +1,62 @@
-import {type AtpAgent} from '@atproto/api'
+import {type Client} from '@atproto/lex'
+import {type DidString, type NsidString} from '@atproto/syntax'
 import {type I18n} from '@lingui/core'
 import {msg} from '@lingui/core/macro'
 
 import {VIDEO_SERVICE_DID} from '#/lib/constants'
 import {UploadLimitError} from '#/lib/media/video/errors'
 import {getServiceAuthAudFromUrl} from '#/lib/strings/url-helpers'
-import {createVideoAgent} from './util'
+import {app, com} from '#/lexicons'
+import {createVideoServiceClient} from './util'
 
 export async function getServiceAuthToken({
-  agent,
+  client,
+  dispatchUrl,
   aud,
   lxm,
   exp,
 }: {
-  agent: AtpAgent
+  client: Client
+  /**
+   * The account's dispatch URL (the old `agent.dispatchUrl`: its PDS, falling
+   * back to the account service). Only needed when `aud` is omitted, so the
+   * default audience can be derived from the PDS host. A lex {@link Client} does
+   * not expose this - it resolves the PDS per request internally - so the caller,
+   * which holds the session, passes it in.
+   */
+  dispatchUrl?: string | URL
   aud?: string
-  lxm: string
+  lxm: NsidString
   exp?: number
 }) {
-  const pdsAud = getServiceAuthAudFromUrl(agent.dispatchUrl)
-  if (!pdsAud) {
-    throw new Error('Agent does not have a PDS URL')
+  let resolvedAud = aud
+  if (!resolvedAud) {
+    if (!dispatchUrl) {
+      throw new Error('Missing service auth audience: no aud or dispatchUrl')
+    }
+    const pdsAud = getServiceAuthAudFromUrl(dispatchUrl)
+    if (!pdsAud) {
+      throw new Error('Agent does not have a PDS URL')
+    }
+    resolvedAud = pdsAud
   }
-  const {data: serviceAuth} = await agent.com.atproto.server.getServiceAuth({
-    aud: aud ?? pdsAud,
+  const {token} = await client.call(com.atproto.server.getServiceAuth, {
+    aud: resolvedAud as DidString,
     lxm,
     exp,
   })
-  return serviceAuth.token
+  return token
 }
 
-export async function getVideoUploadLimits(agent: AtpAgent, i18n: I18n) {
+export async function getVideoUploadLimits(client: Client, i18n: I18n) {
   const token = await getServiceAuthToken({
-    agent,
+    client,
     lxm: 'app.bsky.video.getUploadLimits',
     aud: VIDEO_SERVICE_DID,
   })
-  const videoAgent = createVideoAgent()
-  const {data: limits} = await videoAgent.app.bsky.video
-    .getUploadLimits({}, {headers: {Authorization: `Bearer ${token}`}})
+  const videoClient = createVideoServiceClient(token)
+  const limits = await videoClient
+    .call(app.bsky.video.getUploadLimits)
     .catch(err => {
       if (err instanceof Error) {
         throw new UploadLimitError(err.message)

src/lib/media/video/upload.ts

@@ -1,5 +1,5 @@
 import {createUploadTask, FileSystemUploadType} from 'expo-file-system/legacy'
-import {type AppBskyVideoDefs, type AtpAgent} from '@atproto/api'
+import {type Client} from '@atproto/lex'
 import {type I18n} from '@lingui/core'
 import {msg} from '@lingui/core/macro'
 import {nanoid} from 'nanoid/non-secure'
@@ -11,13 +11,15 @@ import {
   type VideoUploadTransport,
 } from '#/lib/media/video/types'
 import {Features, features} from '#/analytics/features'
+import {type app} from '#/lexicons'
 import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload'
 import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared'
 import {createVideoEndpointUrl, mimeToExt} from './util'
 
 export async function uploadVideo({
   video,
-  agent,
+  client,
+  dispatchUrl,
   did,
   setProgress,
   signal,
@@ -25,7 +27,9 @@ export async function uploadVideo({
   onTransport,
 }: {
   video: CompressedVideo
-  agent: AtpAgent
+  client: Client
+  /** The account's PDS/dispatch URL, for the uploadBlob service-auth token. */
+  dispatchUrl: string | URL
   did: string
   setProgress: (progress: number) => void
   signal: AbortSignal
@@ -35,13 +39,14 @@ export async function uploadVideo({
   if (signal.aborted) {
     throw new AbortError()
   }
-  await getVideoUploadLimits(agent, i18n)
+  await getVideoUploadLimits(client, i18n)
 
   if (features.isOn(Features.VideoMultipartUploadEnable)) {
     try {
       return await uploadVideoMultipart({
         video,
-        agent,
+        client,
+        dispatchUrl,
         setProgress,
         signal,
         onStarted: () => onTransport?.('multipart'),
@@ -64,7 +69,8 @@ export async function uploadVideo({
     throw new AbortError()
   }
   const token = await getServiceAuthToken({
-    agent,
+    client,
+    dispatchUrl,
     lxm: 'com.atproto.repo.uploadBlob',
     exp: Date.now() / 1000 + 60 * 30, // 30 minutes
   })
@@ -91,7 +97,7 @@ export async function uploadVideo({
     throw new Error('No response')
   }
 
-  const responseBody = JSON.parse(res.body) as AppBskyVideoDefs.JobStatus
+  const responseBody = JSON.parse(res.body) as app.bsky.video.defs.JobStatus
 
   if (!responseBody.jobId) {
     throw new ServerError(

src/lib/media/video/upload.web.ts

@@ -1,4 +1,4 @@
-import {type AppBskyVideoDefs, type AtpAgent} from '@atproto/api'
+import {type Client} from '@atproto/lex'
 import {type I18n} from '@lingui/core'
 import {msg} from '@lingui/core/macro'
 import {nanoid} from 'nanoid/non-secure'
@@ -10,13 +10,15 @@ import {
   type VideoUploadTransport,
 } from '#/lib/media/video/types'
 import {Features, features} from '#/analytics/features'
+import {type app} from '#/lexicons'
 import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload'
 import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared'
 import {createVideoEndpointUrl, mimeToExt} from './util'
 
 export async function uploadVideo({
   video,
-  agent,
+  client,
+  dispatchUrl,
   did,
   setProgress,
   signal,
@@ -24,7 +26,9 @@ export async function uploadVideo({
   onTransport,
 }: {
   video: CompressedVideo
-  agent: AtpAgent
+  client: Client
+  /** The account's PDS/dispatch URL, for the uploadBlob service-auth token. */
+  dispatchUrl: string | URL
   did: string
   setProgress: (progress: number) => void
   signal: AbortSignal
@@ -34,13 +38,14 @@ export async function uploadVideo({
   if (signal.aborted) {
     throw new AbortError()
   }
-  await getVideoUploadLimits(agent, i18n)
+  await getVideoUploadLimits(client, i18n)
 
   if (features.isOn(Features.VideoMultipartUploadEnable)) {
     try {
       return await uploadVideoMultipart({
         video,
-        agent,
+        client,
+        dispatchUrl,
         setProgress,
         signal,
         onStarted: () => onTransport?.('multipart'),
@@ -71,7 +76,8 @@ export async function uploadVideo({
     throw new AbortError()
   }
   const token = await getServiceAuthToken({
-    agent,
+    client,
+    dispatchUrl,
     lxm: 'com.atproto.repo.uploadBlob',
     exp: Date.now() / 1000 + 60 * 30, // 30 minutes
   })
@@ -80,7 +86,7 @@ export async function uploadVideo({
     throw new AbortError()
   }
   const xhr = new XMLHttpRequest()
-  const res = await new Promise<AppBskyVideoDefs.JobStatus>(
+  const res = await new Promise<app.bsky.video.defs.JobStatus>(
     (resolve, reject) => {
       xhr.upload.addEventListener('progress', e => {
         const progress = e.loaded / e.total
@@ -92,7 +98,7 @@ export async function uploadVideo({
         } else if (xhr.readyState === 4) {
           const uploadRes = JSON.parse(
             xhr.responseText,
-          ) as AppBskyVideoDefs.JobStatus
+          ) as app.bsky.video.defs.JobStatus
           resolve(uploadRes)
         } else {
           reject(new ServerError(i18n._(msg`Failed to upload video`)))

src/lib/media/video/util.ts

@@ -1,6 +1,5 @@
-import {AtpAgent} from '@atproto/api'
-
 import {type SupportedMimeTypes, VIDEO_SERVICE} from '#/lib/constants'
+import {createLexClient} from '#/lib/lexClient'
 
 export const createVideoEndpointUrl = (
   route: string,
@@ -16,12 +15,29 @@ export const createVideoEndpointUrl = (
   return url.href
 }
 
-export function createVideoAgent() {
-  return new AtpAgent({
+/**
+ * A non-refreshing single-use lex {@link Client} scoped to the video service and
+ * authenticated by a per-call service-auth token. It has no session, so nothing
+ * can refresh it: requests go straight to the video service with the token as a
+ * static `authorization` header, which a raw client - unlike a session-backed
+ * one - is allowed to preset. Mirrors the scoped client in
+ * `#/ageAssurance/useBeginAgeAssurance`.
+ */
+export function createVideoServiceClient(token: string) {
+  return createLexClient({
     service: VIDEO_SERVICE,
+    headers: {authorization: `Bearer ${token}`},
   })
 }
 
+/**
+ * An unauthenticated lex {@link Client} scoped to the video service, for public
+ * reads like `getJobStatus` polling.
+ */
+export function createTokenlessVideoServiceClient() {
+  return createLexClient({service: VIDEO_SERVICE})
+}
+
 export function mimeToExt(mimeType: SupportedMimeTypes | (string & {})) {
   switch (mimeType) {
     case 'video/mp4':

src/lib/notifications/notifications.ts

@@ -2,33 +2,47 @@ import {useCallback, useEffect} from 'react'
 import {Platform} from 'react-native'
 import * as Notifications from 'expo-notifications'
 import {getBadgeCountAsync, setBadgeCountAsync} from 'expo-notifications'
-import {type AppBskyNotificationRegisterPush, type AtpAgent} from '@atproto/api'
+import {type Client} from '@atproto/lex'
 import debounce from 'lodash.debounce'
 
 import {
-  BLUESKY_NOTIF_SERVICE_HEADERS,
+  NOTIF_SERVICE,
   PUBLIC_APPVIEW_DID,
   PUBLIC_STAGING_APPVIEW_DID,
 } from '#/lib/constants'
 import {logger as notyLogger} from '#/lib/notifications/util'
 import {isNetworkError} from '#/lib/strings/errors'
-import {type SessionAccount, useAgent, useSession} from '#/state/session'
+import {type SessionAccount, usePdsClient, useSession} from '#/state/session'
 import BackgroundNotificationHandler from '#/../modules/expo-background-notification-handler'
 import {useAgeAssurance} from '#/ageAssurance'
 import {useAnalytics} from '#/analytics'
 import {IS_DEV, IS_NATIVE} from '#/env'
+import {app} from '#/lexicons'
+
+/**
+ * A resumed single-use account client paired with the account's service origin
+ * and handle. Produced by `createTemporaryClientsAndResume` (session util) and
+ * consumed by {@link unregisterPushToken}, which needs the service host to pick
+ * the appview DID and the handle for a debug log line without reaching into the
+ * session internals.
+ */
+export type TemporaryPushClient = {
+  client: Client
+  service: string
+  handle: string
+}
 
 /**
  * @private
  * Registers the device's push notification token with the Bluesky server.
  */
 async function _registerPushToken({
-  agent,
+  client,
   currentAccount,
   token,
   extra = {},
 }: {
-  agent: AtpAgent
+  client: Client
   currentAccount: SessionAccount
   token: Notifications.DevicePushToken
   extra?: {
@@ -36,7 +50,7 @@ async function _registerPushToken({
   }
 }) {
   try {
-    const payload: AppBskyNotificationRegisterPush.InputSchema = {
+    const payload: app.bsky.notification.registerPush.$InputBody = {
       serviceDid: currentAccount.service?.includes('staging')
         ? PUBLIC_STAGING_APPVIEW_DID
         : PUBLIC_APPVIEW_DID,
@@ -48,8 +62,8 @@ async function _registerPushToken({
 
     notyLogger.debug(`registerPushToken: registering`, {...payload})
 
-    await agent.app.bsky.notification.registerPush(payload, {
-      headers: BLUESKY_NOTIF_SERVICE_HEADERS,
+    await client.call(app.bsky.notification.registerPush, payload, {
+      service: NOTIF_SERVICE,
     })
 
     notyLogger.debug(`registerPushToken: success`)
@@ -74,7 +88,7 @@ const _registerPushTokenDebounced = debounce(_registerPushToken, 100)
  * `_registerPushTokenDebounced` directly.
  */
 export function useRegisterPushToken() {
-  const agent = useAgent()
+  const client = usePdsClient()
   const {currentAccount} = useSession()
 
   return useCallback(
@@ -87,7 +101,7 @@ export function useRegisterPushToken() {
     }) => {
       if (!currentAccount) return
       return _registerPushTokenDebounced({
-        agent,
+        client,
         currentAccount,
         token,
         extra: {
@@ -95,7 +109,7 @@ export function useRegisterPushToken() {
         },
       })
     },
-    [agent, currentAccount],
+    [client, currentAccount],
   )
 }
 
@@ -326,16 +340,17 @@ export async function resetBadgeCount() {
   await setBadgeCountAsync(0)
 }
 
-export async function unregisterPushToken(agents: AtpAgent[]) {
+export async function unregisterPushToken(clients: TemporaryPushClient[]) {
   if (!IS_NATIVE) return
 
   try {
     const token = await getPushToken()
     if (token) {
-      for (const agent of agents) {
-        await agent.app.bsky.notification.unregisterPush(
+      for (const {client, service, handle} of clients) {
+        await client.call(
+          app.bsky.notification.unregisterPush,
           {
-            serviceDid: agent.serviceUrl.hostname.includes('staging')
+            serviceDid: service.includes('staging')
               ? PUBLIC_STAGING_APPVIEW_DID
               : PUBLIC_APPVIEW_DID,
             platform: Platform.OS,
@@ -343,10 +358,10 @@ export async function unregisterPushToken(agents: AtpAgent[]) {
             appId: 'xyz.blueskyweb.app',
           },
           {
-            headers: BLUESKY_NOTIF_SERVICE_HEADERS,
+            service: NOTIF_SERVICE,
           },
         )
-        notyLogger.debug(`Push token unregistered for ${agent.session?.handle}`)
+        notyLogger.debug(`Push token unregistered for ${handle}`)
       }
     } else {
       notyLogger.debug('Tried to unregister push token, but could not find one')