thread an explicit dispatch url through the video upload service auth

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/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/view/com/composer/Composer.tsx

@@ -285,6 +285,12 @@ export const ComposePost = ({
   const pdsClient = usePdsClient()
   const queryClient = useQueryClient()
   const currentDid = currentAccount!.did
+  /*
+   * The host the video service-auth token is minted for. This is the same value
+   * that seeds the session's PDS routing, so the audience always matches the host
+   * the upload actually reaches; a mismatch would 401 the upload.
+   */
+  const currentDispatchUrl = currentAccount!.pdsUrl ?? currentAccount!.service
   const {closeComposer} = useComposerControls()
   const {t: l, i18n} = useLingui()
   const requireAltTextEnabled = useRequireAltTextEnabled()
@@ -479,7 +485,8 @@ export const ComposePost = ({
             },
           })
         },
-        agent,
+        pdsClient,
+        currentDispatchUrl,
         currentDid,
         abortController.signal,
         i18n,
@@ -489,7 +496,8 @@ export const ComposePost = ({
     [
       l,
       i18n,
-      agent,
+      pdsClient,
+      currentDispatchUrl,
       currentDid,
       composerDispatch,
       ax.metric,
@@ -661,7 +669,8 @@ export const ComposePost = ({
               },
             })
           },
-          agent,
+          pdsClient,
+          currentDispatchUrl,
           currentDid,
           abortController.signal,
           i18n,
@@ -677,7 +686,8 @@ export const ComposePost = ({
     [
       l,
       i18n,
-      agent,
+      pdsClient,
+      currentDispatchUrl,
       currentDid,
       composerDispatch,
       ax.metric,

src/view/com/composer/state/video.ts

@@ -1,5 +1,5 @@
 import {type ImagePickerAsset} from 'expo-image-picker'
-import {type AppBskyVideoDefs, type AtpAgent, type BlobRef} from '@atproto/api'
+import {type BlobRef, type Client} from '@atproto/lex'
 import {type I18n} from '@lingui/core'
 import {msg} from '@lingui/core/macro'
 
@@ -14,9 +14,10 @@ import {
 import {type VideoTelemetry} from '#/lib/media/video/telemetry'
 import {type CompressedVideo} from '#/lib/media/video/types'
 import {uploadVideo} from '#/lib/media/video/upload'
-import {createVideoAgent} from '#/lib/media/video/util'
+import {createTokenlessVideoServiceClient} from '#/lib/media/video/util'
 import {isNetworkError} from '#/lib/strings/errors'
 import {logger} from '#/logger'
+import {app} from '#/lexicons'
 import {
   advanceVideoProgress,
   didSkipVideoCompression,
@@ -56,7 +57,7 @@ export type VideoAction =
     }
   | {
       type: 'update_job_status'
-      jobStatus: AppBskyVideoDefs.JobStatus
+      jobStatus: app.bsky.video.defs.JobStatus
       signal: AbortSignal
     }
 
@@ -126,7 +127,7 @@ type ProcessingState = {
   asset: ImagePickerAsset
   video: CompressedVideo
   jobId: string
-  jobStatus: AppBskyVideoDefs.JobStatus | null
+  jobStatus: app.bsky.video.defs.JobStatus | null
   pendingPublish?: undefined
   telemetry: VideoTelemetry
   altText: string
@@ -295,7 +296,8 @@ function trunc2dp(num: number) {
 export async function processVideo(
   asset: ImagePickerAsset,
   dispatch: (action: VideoAction) => void,
-  agent: AtpAgent,
+  client: Client,
+  dispatchUrl: string | URL,
   did: string,
   signal: AbortSignal,
   i18n: I18n,
@@ -339,12 +341,13 @@ export async function processVideo(
     signal,
   })
 
-  let uploadResponse: AppBskyVideoDefs.JobStatus | undefined
+  let uploadResponse: app.bsky.video.defs.JobStatus | undefined
   try {
     telemetry.uploadStarted(video.size)
     uploadResponse = await uploadVideo({
       video,
-      agent,
+      client,
+      dispatchUrl,
       did,
       signal,
       i18n,
@@ -381,12 +384,14 @@ export async function processVideo(
       return // Exit async loop
     }
 
-    const videoAgent = createVideoAgent()
-    let status: AppBskyVideoDefs.JobStatus | undefined
+    const videoClient = createTokenlessVideoServiceClient()
+    let status: app.bsky.video.defs.JobStatus | undefined
     let blob: BlobRef | undefined
     try {
-      const response = await videoAgent.app.bsky.video.getJobStatus({jobId})
-      status = response.data.jobStatus
+      const response = await videoClient.call(app.bsky.video.getJobStatus, {
+        jobId,
+      })
+      status = response.jobStatus
       pollFailures = 0
 
       if (status.state === 'JOB_STATE_COMPLETED') {