From 4f669812b98cde14a9d5106c3874a3badf6966e2 Mon Sep 17 00:00:00 2001 From: Samuel Newman Date: Tue, 4 Aug 2026 01:10:13 +0300 Subject: [PATCH] thread an explicit dispatch url through the video upload service auth Co-Authored-By: Claude Fable 5 --- src/lib/api/index.ts | 8 ++-- src/lib/api/legacy-blob.ts | 15 -------- src/lib/media/video/multipart/types.ts | 6 ++- src/lib/media/video/multipart/upload.ts | 32 ++++++++++------ src/lib/media/video/upload.shared.ts | 50 +++++++++++++++++-------- src/lib/media/video/upload.ts | 20 ++++++---- src/lib/media/video/upload.web.ts | 22 +++++++---- src/lib/media/video/util.ts | 24 ++++++++++-- src/view/com/composer/Composer.tsx | 18 +++++++-- src/view/com/composer/state/video.ts | 27 +++++++------ 10 files changed, 140 insertions(+), 82 deletions(-) diff --git a/src/lib/api/index.ts b/src/lib/api/index.ts index e52d4ec4fc..f13c227857 100644 --- a/src/lib/api/index.ts +++ b/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, diff --git a/src/lib/api/legacy-blob.ts b/src/lib/api/legacy-blob.ts index a98e375d20..b60acadfe4 100644 --- a/src/lib/api/legacy-blob.ts +++ b/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[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() -} diff --git a/src/lib/media/video/multipart/types.ts b/src/lib/media/video/multipart/types.ts index 16817778c3..a0cfd216e3 100644 --- a/src/lib/media/video/multipart/types.ts +++ b/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< diff --git a/src/lib/media/video/multipart/upload.ts b/src/lib/media/video/multipart/upload.ts index 2b45e6d2c6..d3e5210873 100644 --- a/src/lib/media/video/multipart/upload.ts +++ b/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 { +}): Promise { 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 signal: AbortSignal resendMissingParts: (receivedPartNumbers: number[]) => Promise -}): Promise { +}): Promise { let createdFailures = 0 let forceTokenRefresh = true while (true) { @@ -224,7 +228,7 @@ async function abortThenFallbackOrResolve( jobId: string, token: string, cause: unknown, -): Promise { +): Promise { 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 | 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, }) diff --git a/src/lib/media/video/upload.shared.ts b/src/lib/media/video/upload.shared.ts index f8aaa1249b..9e8d99a185 100644 --- a/src/lib/media/video/upload.shared.ts +++ b/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) diff --git a/src/lib/media/video/upload.ts b/src/lib/media/video/upload.ts index b91ad7a153..287c8f09c6 100644 --- a/src/lib/media/video/upload.ts +++ b/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( diff --git a/src/lib/media/video/upload.web.ts b/src/lib/media/video/upload.web.ts index cfefbd797d..90c10a0684 100644 --- a/src/lib/media/video/upload.web.ts +++ b/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( + const res = await new Promise( (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`))) diff --git a/src/lib/media/video/util.ts b/src/lib/media/video/util.ts index 236f0cff3e..7c9686a7f4 100644 --- a/src/lib/media/video/util.ts +++ b/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': diff --git a/src/view/com/composer/Composer.tsx b/src/view/com/composer/Composer.tsx index 1388ce0c45..3f44040444 100644 --- a/src/view/com/composer/Composer.tsx +++ b/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, diff --git a/src/view/com/composer/state/video.ts b/src/view/com/composer/state/video.ts index eeca7dca47..4ae825a7f1 100644 --- a/src/view/com/composer/state/video.ts +++ b/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') {