Refactor video uploads (#5570)

* Remove unused video field

* Stop exposing video dispatch

* Move cancellation out of the reducer

* Make useUploadStatusQuery controlled by jobId

* Rename SetStatus to SetProcessing

This action only has one callsite and it's always passing "processing".

* Move jobId into video reducer state

* Make cancellation scoped

* Inline useCompressVideoMutation

* Move processVideo down the file

* Extract getErrorMessage

* useServiceAuthToken -> getServiceAuthToken

* useVideoAgent -> createVideoAgent

* useVideoUploadLimits -> getVideoUploadLimits

* useUploadVideoMutation -> uploadVideo

* Use async/await in processVideo

* Inline onVideoCompressed into processVideo

* Use async/await for uploadVideo

* Factor out error messages

* Guard dispatch with signal

This lets us remove the scattered signal checks around dispatch.

* Move job polling out of RQ

* Handle poll failures

* Remove unnecessary guards

* Slightly more accurate condition

* Move initVideoUri handling out of the hook

* Remove dead argument

It wasn't being used before either.

* Remove unused detailed status

This isn't being used because we're only respecting that state variable when isProcessing=true, but isProcessing is always false during video upload.

If we want to re-add this later, it should really just be derived from the reducer state.

* Harden the video reducer

* Tie all spawned work to a signal

* Preserve asset/media for nicer error state

* Rename actions to match states

* Inline useUploadVideo

This abstraction is getting in the way of some future work.

* Move MIME check to the only place that handles it

src/state/queries/video/compress-video.ts

@@ -1,39 +0,0 @@
-import {ImagePickerAsset} from 'expo-image-picker'
-import {useMutation} from '@tanstack/react-query'
-
-import {cancelable} from '#/lib/async/cancelable'
-import {CompressedVideo} from '#/lib/media/video/types'
-import {compressVideo} from 'lib/media/video/compress'
-
-export function useCompressVideoMutation({
-  onProgress,
-  onSuccess,
-  onError,
-  signal,
-}: {
-  onProgress: (progress: number) => void
-  onError: (e: any) => void
-  onSuccess: (video: CompressedVideo) => void
-  signal: AbortSignal
-}) {
-  return useMutation({
-    mutationKey: ['video', 'compress'],
-    mutationFn: cancelable(
-      (asset: ImagePickerAsset) =>
-        compressVideo(asset, {
-          onProgress: num => onProgress(trunc2dp(num)),
-          signal,
-        }),
-      signal,
-    ),
-    onError,
-    onSuccess,
-    onMutate: () => {
-      onProgress(0)
-    },
-  })
-}
-
-function trunc2dp(num: number) {
-  return Math.trunc(num * 100) / 100
-}

src/state/queries/video/util.ts

@@ -1,4 +1,3 @@
-import {useMemo} from 'react'
 import {AtpAgent} from '@atproto/api'
 
 import {SupportedMimeTypes, VIDEO_SERVICE} from '#/lib/constants'
@@ -17,12 +16,10 @@ export const createVideoEndpointUrl = (
   return url.href
 }
 
-export function useVideoAgent() {
+export function createVideoAgent() {
-  return useMemo(() => {
+  return new AtpAgent({
-    return new AtpAgent({
+    service: VIDEO_SERVICE,
-      service: VIDEO_SERVICE,
+  })
-    })
-  }, [])
 }
 
 export function mimeToExt(mimeType: SupportedMimeTypes | (string & {})) {

src/state/queries/video/video-upload.shared.ts

@@ -1,73 +1,61 @@
-import {useCallback} from 'react'
+import {BskyAgent} from '@atproto/api'
+import {I18n} from '@lingui/core'
 import {msg} from '@lingui/macro'
-import {useLingui} from '@lingui/react'
 
 import {VIDEO_SERVICE_DID} from '#/lib/constants'
 import {UploadLimitError} from '#/lib/media/video/errors'
 import {getServiceAuthAudFromUrl} from '#/lib/strings/url-helpers'
-import {useAgent} from '#/state/session'
+import {createVideoAgent} from './util'
-import {useVideoAgent} from './util'
 
-export function useServiceAuthToken({
+export async function getServiceAuthToken({
+  agent,
   aud,
   lxm,
   exp,
 }: {
+  agent: BskyAgent
   aud?: string
   lxm: string
   exp?: number
 }) {
-  const agent = useAgent()
+  const pdsAud = getServiceAuthAudFromUrl(agent.dispatchUrl)
-
+  if (!pdsAud) {
-  return useCallback(async () => {
+    throw new Error('Agent does not have a PDS URL')
-    const pdsAud = getServiceAuthAudFromUrl(agent.dispatchUrl)
+  }
-
+  const {data: serviceAuth} = await agent.com.atproto.server.getServiceAuth({
-    if (!pdsAud) {
+    aud: aud ?? pdsAud,
-      throw new Error('Agent does not have a PDS URL')
+    lxm,
-    }
+    exp,
-
+  })
-    const {data: serviceAuth} = await agent.com.atproto.server.getServiceAuth({
+  return serviceAuth.token
-      aud: aud ?? pdsAud,
-      lxm,
-      exp,
-    })
-
-    return serviceAuth.token
-  }, [agent, aud, lxm, exp])
 }
 
-export function useVideoUploadLimits() {
+export async function getVideoUploadLimits(agent: BskyAgent, _: I18n['_']) {
-  const agent = useVideoAgent()
+  const token = await getServiceAuthToken({
-  const getToken = useServiceAuthToken({
+    agent,
     lxm: 'app.bsky.video.getUploadLimits',
     aud: VIDEO_SERVICE_DID,
   })
-  const {_} = useLingui()
+  const videoAgent = createVideoAgent()
-
+  const {data: limits} = await videoAgent.app.bsky.video
-  return useCallback(async () => {
+    .getUploadLimits({}, {headers: {Authorization: `Bearer ${token}`}})
-    const {data: limits} = await agent.app.bsky.video
+    .catch(err => {
-      .getUploadLimits(
+      if (err instanceof Error) {
-        {},
+        throw new UploadLimitError(err.message)
-        {headers: {Authorization: `Bearer ${await getToken()}`}},
-      )
-      .catch(err => {
-        if (err instanceof Error) {
-          throw new UploadLimitError(err.message)
-        } else {
-          throw err
-        }
-      })
-
-    if (!limits.canUpload) {
-      if (limits.message) {
-        throw new UploadLimitError(limits.message)
       } else {
-        throw new UploadLimitError(
+        throw err
-          _(
-            msg`You have temporarily reached the limit for video uploads. Please try again later.`,
-          ),
-        )
       }
+    })
+
+  if (!limits.canUpload) {
+    if (limits.message) {
+      throw new UploadLimitError(limits.message)
+    } else {
+      throw new UploadLimitError(
+        _(
+          msg`You have temporarily reached the limit for video uploads. Please try again later.`,
+        ),
+      )
     }
-  }, [agent, _, getToken])
+  }
 }

src/state/queries/video/video-upload.ts

@@ -1,76 +1,79 @@
 import {createUploadTask, FileSystemUploadType} from 'expo-file-system'
-import {AppBskyVideoDefs} from '@atproto/api'
+import {AppBskyVideoDefs, BskyAgent} from '@atproto/api'
+import {I18n} from '@lingui/core'
 import {msg} from '@lingui/macro'
-import {useLingui} from '@lingui/react'
-import {useMutation} from '@tanstack/react-query'
 import {nanoid} from 'nanoid/non-secure'
 
-import {cancelable} from '#/lib/async/cancelable'
+import {AbortError} from '#/lib/async/cancelable'
 import {ServerError} from '#/lib/media/video/errors'
 import {CompressedVideo} from '#/lib/media/video/types'
 import {createVideoEndpointUrl, mimeToExt} from '#/state/queries/video/util'
-import {useSession} from '#/state/session'
+import {getServiceAuthToken, getVideoUploadLimits} from './video-upload.shared'
-import {useServiceAuthToken, useVideoUploadLimits} from './video-upload.shared'
 
-export const useUploadVideoMutation = ({
+export async function uploadVideo({
-  onSuccess,
+  video,
-  onError,
+  agent,
+  did,
   setProgress,
   signal,
+  _,
 }: {
-  onSuccess: (response: AppBskyVideoDefs.JobStatus) => void
+  video: CompressedVideo
-  onError: (e: any) => void
+  agent: BskyAgent
+  did: string
   setProgress: (progress: number) => void
   signal: AbortSignal
-}) => {
+  _: I18n['_']
-  const {currentAccount} = useSession()
+}) {
-  const getToken = useServiceAuthToken({
+  if (signal.aborted) {
+    throw new AbortError()
+  }
+  await getVideoUploadLimits(agent, _)
+
+  const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
+    did,
+    name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+  })
+
+  if (signal.aborted) {
+    throw new AbortError()
+  }
+  const token = await getServiceAuthToken({
+    agent,
     lxm: 'com.atproto.repo.uploadBlob',
     exp: Date.now() / 1000 + 60 * 30, // 30 minutes
   })
-  const checkLimits = useVideoUploadLimits()
+  const uploadTask = createUploadTask(
-  const {_} = useLingui()
+    uri,
+    video.uri,
+    {
+      headers: {
+        'content-type': video.mimeType,
+        Authorization: `Bearer ${token}`,
+      },
+      httpMethod: 'POST',
+      uploadType: FileSystemUploadType.BINARY_CONTENT,
+    },
+    p => setProgress(p.totalBytesSent / p.totalBytesExpectedToSend),
+  )
 
-  return useMutation({
+  if (signal.aborted) {
-    mutationKey: ['video', 'upload'],
+    throw new AbortError()
-    mutationFn: cancelable(async (video: CompressedVideo) => {
+  }
-      await checkLimits()
+  const res = await uploadTask.uploadAsync()
 
-      const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
+  if (!res?.body) {
-        did: currentAccount!.did,
+    throw new Error('No response')
-        name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+  }
-      })
 
-      const uploadTask = createUploadTask(
+  const responseBody = JSON.parse(res.body) as AppBskyVideoDefs.JobStatus
-        uri,
-        video.uri,
-        {
-          headers: {
-            'content-type': video.mimeType,
-            Authorization: `Bearer ${await getToken()}`,
-          },
-          httpMethod: 'POST',
-          uploadType: FileSystemUploadType.BINARY_CONTENT,
-        },
-        p => setProgress(p.totalBytesSent / p.totalBytesExpectedToSend),
-      )
-      const res = await uploadTask.uploadAsync()
 
-      if (!res?.body) {
+  if (!responseBody.jobId) {
-        throw new Error('No response')
+    throw new ServerError(responseBody.error || _(msg`Failed to upload video`))
-      }
+  }
 
-      const responseBody = JSON.parse(res.body) as AppBskyVideoDefs.JobStatus
+  if (signal.aborted) {
-
+    throw new AbortError()
-      if (!responseBody.jobId) {
+  }
-        throw new ServerError(
+  return responseBody
-          responseBody.error || _(msg`Failed to upload video`),
-        )
-      }
-
-      return responseBody
-    }, signal),
-    onError,
-    onSuccess,
-  })
 }

src/state/queries/video/video-upload.web.ts

@@ -1,86 +1,95 @@
 import {AppBskyVideoDefs} from '@atproto/api'
+import {BskyAgent} from '@atproto/api'
+import {I18n} from '@lingui/core'
 import {msg} from '@lingui/macro'
-import {useLingui} from '@lingui/react'
-import {useMutation} from '@tanstack/react-query'
 import {nanoid} from 'nanoid/non-secure'
 
-import {cancelable} from '#/lib/async/cancelable'
+import {AbortError} from '#/lib/async/cancelable'
 import {ServerError} from '#/lib/media/video/errors'
 import {CompressedVideo} from '#/lib/media/video/types'
 import {createVideoEndpointUrl, mimeToExt} from '#/state/queries/video/util'
-import {useSession} from '#/state/session'
+import {getServiceAuthToken, getVideoUploadLimits} from './video-upload.shared'
-import {useServiceAuthToken, useVideoUploadLimits} from './video-upload.shared'
 
-export const useUploadVideoMutation = ({
+export async function uploadVideo({
-  onSuccess,
+  video,
-  onError,
+  agent,
+  did,
   setProgress,
   signal,
+  _,
 }: {
-  onSuccess: (response: AppBskyVideoDefs.JobStatus) => void
+  video: CompressedVideo
-  onError: (e: any) => void
+  agent: BskyAgent
+  did: string
   setProgress: (progress: number) => void
   signal: AbortSignal
-}) => {
+  _: I18n['_']
-  const {currentAccount} = useSession()
+}) {
-  const getToken = useServiceAuthToken({
+  if (signal.aborted) {
+    throw new AbortError()
+  }
+  await getVideoUploadLimits(agent, _)
+
+  const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
+    did,
+    name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+  })
+
+  let bytes = video.bytes
+  if (!bytes) {
+    if (signal.aborted) {
+      throw new AbortError()
+    }
+    bytes = await fetch(video.uri).then(res => res.arrayBuffer())
+  }
+
+  if (signal.aborted) {
+    throw new AbortError()
+  }
+  const token = await getServiceAuthToken({
+    agent,
     lxm: 'com.atproto.repo.uploadBlob',
     exp: Date.now() / 1000 + 60 * 30, // 30 minutes
   })
-  const checkLimits = useVideoUploadLimits()
-  const {_} = useLingui()
 
-  return useMutation({
+  if (signal.aborted) {
-    mutationKey: ['video', 'upload'],
+    throw new AbortError()
-    mutationFn: cancelable(async (video: CompressedVideo) => {
+  }
-      await checkLimits()
+  const xhr = new XMLHttpRequest()
-
+  const res = await new Promise<AppBskyVideoDefs.JobStatus>(
-      const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
+    (resolve, reject) => {
-        did: currentAccount!.did,
+      xhr.upload.addEventListener('progress', e => {
-        name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+        const progress = e.loaded / e.total
+        setProgress(progress)
       })
-
+      xhr.onloadend = () => {
-      let bytes = video.bytes
+        if (signal.aborted) {
-      if (!bytes) {
+          reject(new AbortError())
-        bytes = await fetch(video.uri).then(res => res.arrayBuffer())
+        } else if (xhr.readyState === 4) {
+          const uploadRes = JSON.parse(
+            xhr.responseText,
+          ) as AppBskyVideoDefs.JobStatus
+          resolve(uploadRes)
+        } else {
+          reject(new ServerError(_(msg`Failed to upload video`)))
+        }
       }
-
+      xhr.onerror = () => {
-      const token = await getToken()
+        reject(new ServerError(_(msg`Failed to upload video`)))
-
-      const xhr = new XMLHttpRequest()
-      const res = await new Promise<AppBskyVideoDefs.JobStatus>(
-        (resolve, reject) => {
-          xhr.upload.addEventListener('progress', e => {
-            const progress = e.loaded / e.total
-            setProgress(progress)
-          })
-          xhr.onloadend = () => {
-            if (xhr.readyState === 4) {
-              const uploadRes = JSON.parse(
-                xhr.responseText,
-              ) as AppBskyVideoDefs.JobStatus
-              resolve(uploadRes)
-            } else {
-              reject(new ServerError(_(msg`Failed to upload video`)))
-            }
-          }
-          xhr.onerror = () => {
-            reject(new ServerError(_(msg`Failed to upload video`)))
-          }
-          xhr.open('POST', uri)
-          xhr.setRequestHeader('Content-Type', video.mimeType)
-          xhr.setRequestHeader('Authorization', `Bearer ${token}`)
-          xhr.send(bytes)
-        },
-      )
-
-      if (!res.jobId) {
-        throw new ServerError(res.error || _(msg`Failed to upload video`))
       }
+      xhr.open('POST', uri)
+      xhr.setRequestHeader('Content-Type', video.mimeType)
+      xhr.setRequestHeader('Authorization', `Bearer ${token}`)
+      xhr.send(bytes)
+    },
+  )
 
-      return res
+  if (!res.jobId) {
-    }, signal),
+    throw new ServerError(res.error || _(msg`Failed to upload video`))
-    onError,
+  }
-    onSuccess,
+
-  })
+  if (signal.aborted) {
+    throw new AbortError()
+  }
+  return res
 }

src/state/queries/video/video.ts

@@ -1,12 +1,11 @@
-import React, {useCallback, useEffect} from 'react'
 import {ImagePickerAsset} from 'expo-image-picker'
-import {AppBskyVideoDefs, BlobRef} from '@atproto/api'
+import {AppBskyVideoDefs, BlobRef, BskyAgent} from '@atproto/api'
+import {JobStatus} from '@atproto/api/dist/client/types/app/bsky/video/defs'
+import {I18n} from '@lingui/core'
 import {msg} from '@lingui/macro'
-import {useLingui} from '@lingui/react'
-import {QueryClient, useQuery, useQueryClient} from '@tanstack/react-query'
 
 import {AbortError} from '#/lib/async/cancelable'
-import {SUPPORTED_MIME_TYPES, SupportedMimeTypes} from '#/lib/constants'
+import {compressVideo} from '#/lib/media/video/compress'
 import {
   ServerError,
   UploadLimitError,
@@ -14,338 +13,409 @@ import {
 } from '#/lib/media/video/errors'
 import {CompressedVideo} from '#/lib/media/video/types'
 import {logger} from '#/logger'
-import {isWeb} from '#/platform/detection'
+import {createVideoAgent} from '#/state/queries/video/util'
-import {useCompressVideoMutation} from '#/state/queries/video/compress-video'
+import {uploadVideo} from '#/state/queries/video/video-upload'
-import {useVideoAgent} from '#/state/queries/video/util'
-import {useUploadVideoMutation} from '#/state/queries/video/video-upload'
-
-type Status = 'idle' | 'compressing' | 'processing' | 'uploading' | 'done'
 
 type Action =
-  | {type: 'SetStatus'; status: Status}
+  | {type: 'to_idle'; nextController: AbortController}
-  | {type: 'SetProgress'; progress: number}
+  | {
-  | {type: 'SetError'; error: string | undefined}
+      type: 'idle_to_compressing'
-  | {type: 'Reset'}
+      asset: ImagePickerAsset
-  | {type: 'SetAsset'; asset: ImagePickerAsset}
+      signal: AbortSignal
-  | {type: 'SetDimensions'; width: number; height: number}
+    }
-  | {type: 'SetVideo'; video: CompressedVideo}
+  | {
-  | {type: 'SetJobStatus'; jobStatus: AppBskyVideoDefs.JobStatus}
+      type: 'compressing_to_uploading'
-  | {type: 'SetComplete'; blobRef: BlobRef}
+      video: CompressedVideo
+      signal: AbortSignal
+    }
+  | {
+      type: 'uploading_to_processing'
+      jobId: string
+      signal: AbortSignal
+    }
+  | {type: 'to_error'; error: string; signal: AbortSignal}
+  | {
+      type: 'to_done'
+      blobRef: BlobRef
+      signal: AbortSignal
+    }
+  | {type: 'update_progress'; progress: number; signal: AbortSignal}
+  | {
+      type: 'update_dimensions'
+      width: number
+      height: number
+      signal: AbortSignal
+    }
+  | {
+      type: 'update_job_status'
+      jobStatus: AppBskyVideoDefs.JobStatus
+      signal: AbortSignal
+    }
 
-export interface State {
+type IdleState = {
-  status: Status
+  status: 'idle'
-  progress: number
+  progress: 0
-  asset?: ImagePickerAsset
-  video: CompressedVideo | null
-  jobStatus?: AppBskyVideoDefs.JobStatus
-  blobRef?: BlobRef
-  error?: string
   abortController: AbortController
-  pendingPublish?: {blobRef: BlobRef; mutableProcessed: boolean}
+  asset?: undefined
+  video?: undefined
+  jobId?: undefined
+  pendingPublish?: undefined
 }
 
-export type VideoUploadDispatch = (action: Action) => void
+type ErrorState = {
+  status: 'error'
+  progress: 100
+  abortController: AbortController
+  asset: ImagePickerAsset | null
+  video: CompressedVideo | null
+  jobId: string | null
+  error: string
+  pendingPublish?: undefined
+}
 
-function reducer(queryClient: QueryClient) {
+type CompressingState = {
-  return (state: State, action: Action): State => {
+  status: 'compressing'
-    let updatedState = state
+  progress: number
-    if (action.type === 'SetStatus') {
+  abortController: AbortController
-      updatedState = {...state, status: action.status}
+  asset: ImagePickerAsset
-    } else if (action.type === 'SetProgress') {
+  video?: undefined
-      updatedState = {...state, progress: action.progress}
+  jobId?: undefined
-    } else if (action.type === 'SetError') {
+  pendingPublish?: undefined
-      updatedState = {...state, error: action.error}
+}
-    } else if (action.type === 'Reset') {
+
-      state.abortController.abort()
+type UploadingState = {
-      queryClient.cancelQueries({
+  status: 'uploading'
-        queryKey: ['video'],
+  progress: number
-      })
+  abortController: AbortController
-      updatedState = {
+  asset: ImagePickerAsset
-        status: 'idle',
+  video: CompressedVideo
-        progress: 0,
+  jobId?: undefined
-        video: null,
+  pendingPublish?: undefined
-        blobRef: undefined,
+}
-        abortController: new AbortController(),
+
-      }
+type ProcessingState = {
-    } else if (action.type === 'SetAsset') {
+  status: 'processing'
-      updatedState = {
+  progress: number
+  abortController: AbortController
+  asset: ImagePickerAsset
+  video: CompressedVideo
+  jobId: string
+  jobStatus: AppBskyVideoDefs.JobStatus | null
+  pendingPublish?: undefined
+}
+
+type DoneState = {
+  status: 'done'
+  progress: 100
+  abortController: AbortController
+  asset: ImagePickerAsset
+  video: CompressedVideo
+  jobId?: undefined
+  pendingPublish: {blobRef: BlobRef; mutableProcessed: boolean}
+}
+
+export type State =
+  | IdleState
+  | ErrorState
+  | CompressingState
+  | UploadingState
+  | ProcessingState
+  | DoneState
+
+export function createVideoState(
+  abortController: AbortController = new AbortController(),
+): IdleState {
+  return {
+    status: 'idle',
+    progress: 0,
+    abortController,
+  }
+}
+
+export function videoReducer(state: State, action: Action): State {
+  if (action.type === 'to_idle') {
+    return createVideoState(action.nextController)
+  }
+  if (action.signal.aborted || action.signal !== state.abortController.signal) {
+    // This action is stale and the process that spawned it is no longer relevant.
+    return state
+  }
+  if (action.type === 'to_error') {
+    return {
+      status: 'error',
+      progress: 100,
+      abortController: state.abortController,
+      error: action.error,
+      asset: state.asset ?? null,
+      video: state.video ?? null,
+      jobId: state.jobId ?? null,
+    }
+  } else if (action.type === 'update_progress') {
+    if (state.status === 'compressing' || state.status === 'uploading') {
+      return {
         ...state,
-        asset: action.asset,
+        progress: action.progress,
+      }
+    }
+  } else if (action.type === 'idle_to_compressing') {
+    if (state.status === 'idle') {
+      return {
         status: 'compressing',
-        error: undefined,
+        progress: 0,
+        abortController: state.abortController,
+        asset: action.asset,
       }
-    } else if (action.type === 'SetDimensions') {
+    }
-      updatedState = {
+  } else if (action.type === 'update_dimensions') {
+    if (state.asset) {
+      return {
         ...state,
-        asset: state.asset
+        asset: {...state.asset, width: action.width, height: action.height},
-          ? {...state.asset, width: action.width, height: action.height}
-          : undefined,
       }
-    } else if (action.type === 'SetVideo') {
+    }
-      updatedState = {...state, video: action.video, status: 'uploading'}
+  } else if (action.type === 'compressing_to_uploading') {
-    } else if (action.type === 'SetJobStatus') {
+    if (state.status === 'compressing') {
-      updatedState = {...state, jobStatus: action.jobStatus}
+      return {
-    } else if (action.type === 'SetComplete') {
+        status: 'uploading',
-      updatedState = {
+        progress: 0,
+        abortController: state.abortController,
+        asset: state.asset,
+        video: action.video,
+      }
+    }
+    return state
+  } else if (action.type === 'uploading_to_processing') {
+    if (state.status === 'uploading') {
+      return {
+        status: 'processing',
+        progress: 0,
+        abortController: state.abortController,
+        asset: state.asset,
+        video: state.video,
+        jobId: action.jobId,
+        jobStatus: null,
+      }
+    }
+  } else if (action.type === 'update_job_status') {
+    if (state.status === 'processing') {
+      return {
         ...state,
+        jobStatus: action.jobStatus,
+        progress:
+          action.jobStatus.progress !== undefined
+            ? action.jobStatus.progress / 100
+            : state.progress,
+      }
+    }
+  } else if (action.type === 'to_done') {
+    if (state.status === 'processing') {
+      return {
+        status: 'done',
+        progress: 100,
+        abortController: state.abortController,
+        asset: state.asset,
+        video: state.video,
         pendingPublish: {
           blobRef: action.blobRef,
           mutableProcessed: false,
         },
-        status: 'done',
       }
     }
-    return updatedState
   }
-}
+  console.error(
-
+    'Unexpected video action (' +
-export function useUploadVideo({
+      action.type +
-  setStatus,
+      ') while in ' +
-  initialVideoUri,
+      state.status +
-}: {
+      ' state',
-  setStatus: (status: string) => void
-  onSuccess: () => void
-  initialVideoUri?: string
-}) {
-  const {_} = useLingui()
-  const queryClient = useQueryClient()
-  const [state, dispatch] = React.useReducer(reducer(queryClient), {
-    status: 'idle',
-    progress: 0,
-    video: null,
-    abortController: new AbortController(),
-  })
-
-  const {setJobId} = useUploadStatusQuery({
-    onStatusChange: (status: AppBskyVideoDefs.JobStatus) => {
-      // This might prove unuseful, most of the job status steps happen too quickly to even be displayed to the user
-      // Leaving it for now though
-      dispatch({
-        type: 'SetJobStatus',
-        jobStatus: status,
-      })
-      setStatus(status.state.toString())
-    },
-    onSuccess: blobRef => {
-      dispatch({
-        type: 'SetComplete',
-        blobRef,
-      })
-    },
-    onError: useCallback(
-      error => {
-        logger.error('Error processing video', {safeMessage: error})
-        dispatch({
-          type: 'SetError',
-          error: _(msg`Video failed to process`),
-        })
-      },
-      [_],
-    ),
-  })
-
-  const {mutate: onVideoCompressed} = useUploadVideoMutation({
-    onSuccess: response => {
-      dispatch({
-        type: 'SetStatus',
-        status: 'processing',
-      })
-      setJobId(response.jobId)
-    },
-    onError: e => {
-      if (e instanceof AbortError) {
-        return
-      } else if (e instanceof ServerError || e instanceof UploadLimitError) {
-        let message
-        // https://github.com/bluesky-social/tango/blob/lumi/lumi/worker/permissions.go#L77
-        switch (e.message) {
-          case 'User is not allowed to upload videos':
-            message = _(msg`You are not allowed to upload videos.`)
-            break
-          case 'Uploading is disabled at the moment':
-            message = _(
-              msg`Hold up! We’re gradually giving access to video, and you’re still waiting in line. Check back soon!`,
-            )
-            break
-          case "Failed to get user's upload stats":
-            message = _(
-              msg`We were unable to determine if you are allowed to upload videos. Please try again.`,
-            )
-            break
-          case 'User has exceeded daily upload bytes limit':
-            message = _(
-              msg`You've reached your daily limit for video uploads (too many bytes)`,
-            )
-            break
-          case 'User has exceeded daily upload videos limit':
-            message = _(
-              msg`You've reached your daily limit for video uploads (too many videos)`,
-            )
-            break
-          case 'Account is not old enough to upload videos':
-            message = _(
-              msg`Your account is not yet old enough to upload videos. Please try again later.`,
-            )
-            break
-          default:
-            message = e.message
-            break
-        }
-        dispatch({
-          type: 'SetError',
-          error: message,
-        })
-      } else {
-        dispatch({
-          type: 'SetError',
-          error: _(msg`An error occurred while uploading the video.`),
-        })
-      }
-      logger.error('Error uploading video', {safeMessage: e})
-    },
-    setProgress: p => {
-      dispatch({type: 'SetProgress', progress: p})
-    },
-    signal: state.abortController.signal,
-  })
-
-  const {mutate: onSelectVideo} = useCompressVideoMutation({
-    onProgress: p => {
-      dispatch({type: 'SetProgress', progress: p})
-    },
-    onSuccess: (video: CompressedVideo) => {
-      dispatch({
-        type: 'SetVideo',
-        video,
-      })
-      onVideoCompressed(video)
-    },
-    onError: e => {
-      if (e instanceof AbortError) {
-        return
-      } else if (e instanceof VideoTooLargeError) {
-        dispatch({
-          type: 'SetError',
-          error: _(msg`The selected video is larger than 50MB.`),
-        })
-      } else {
-        dispatch({
-          type: 'SetError',
-          error: _(msg`An error occurred while compressing the video.`),
-        })
-        logger.error('Error compressing video', {safeMessage: e})
-      }
-    },
-    signal: state.abortController.signal,
-  })
-
-  const selectVideo = React.useCallback(
-    (asset: ImagePickerAsset) => {
-      // compression step on native converts to mp4, so no need to check there
-      if (isWeb) {
-        const mimeType = getMimeType(asset)
-        if (!SUPPORTED_MIME_TYPES.includes(mimeType as SupportedMimeTypes)) {
-          throw new Error(_(msg`Unsupported video type: ${mimeType}`))
-        }
-      }
-
-      dispatch({
-        type: 'SetAsset',
-        asset,
-      })
-      onSelectVideo(asset)
-    },
-    [_, onSelectVideo],
   )
-
+  return state
-  const clearVideo = () => {
-    dispatch({type: 'Reset'})
-  }
-
-  const updateVideoDimensions = useCallback((width: number, height: number) => {
-    dispatch({
-      type: 'SetDimensions',
-      width,
-      height,
-    })
-  }, [])
-
-  // Whenever we receive an initial video uri, we should immediately run compression if necessary
-  useEffect(() => {
-    if (initialVideoUri) {
-      selectVideo({uri: initialVideoUri} as ImagePickerAsset)
-    }
-  }, [initialVideoUri, selectVideo])
-
-  return {
-    state,
-    dispatch,
-    selectVideo,
-    clearVideo,
-    updateVideoDimensions,
-  }
 }
 
-const useUploadStatusQuery = ({
+function trunc2dp(num: number) {
-  onStatusChange,
+  return Math.trunc(num * 100) / 100
-  onSuccess,
+}
-  onError,
-}: {
-  onStatusChange: (status: AppBskyVideoDefs.JobStatus) => void
-  onSuccess: (blobRef: BlobRef) => void
-  onError: (error: Error) => void
-}) => {
-  const videoAgent = useVideoAgent()
-  const [enabled, setEnabled] = React.useState(true)
-  const [jobId, setJobId] = React.useState<string>()
 
-  const {error} = useQuery({
+export async function processVideo(
-    queryKey: ['video', 'upload status', jobId],
+  asset: ImagePickerAsset,
-    queryFn: async () => {
+  dispatch: (action: Action) => void,
-      if (!jobId) return // this won't happen, can ignore
+  agent: BskyAgent,
+  did: string,
+  signal: AbortSignal,
+  _: I18n['_'],
+) {
+  dispatch({
+    type: 'idle_to_compressing',
+    asset,
+    signal,
+  })
+
+  let video: CompressedVideo | undefined
+  try {
+    video = await compressVideo(asset, {
+      onProgress: num => {
+        dispatch({type: 'update_progress', progress: trunc2dp(num), signal})
+      },
+      signal,
+    })
+  } catch (e) {
+    const message = getCompressErrorMessage(e, _)
+    if (message !== null) {
+      dispatch({
+        type: 'to_error',
+        error: message,
+        signal,
+      })
+    }
+    return
+  }
+  dispatch({
+    type: 'compressing_to_uploading',
+    video,
+    signal,
+  })
+
+  let uploadResponse: AppBskyVideoDefs.JobStatus | undefined
+  try {
+    uploadResponse = await uploadVideo({
+      video,
+      agent,
+      did,
+      signal,
+      _,
+      setProgress: p => {
+        dispatch({type: 'update_progress', progress: p, signal})
+      },
+    })
+  } catch (e) {
+    const message = getUploadErrorMessage(e, _)
+    if (message !== null) {
+      dispatch({
+        type: 'to_error',
+        error: message,
+        signal,
+      })
+    }
+    return
+  }
+
+  const jobId = uploadResponse.jobId
+  dispatch({
+    type: 'uploading_to_processing',
+    jobId,
+    signal,
+  })
+
+  let pollFailures = 0
+  while (true) {
+    if (signal.aborted) {
+      return // Exit async loop
+    }
+
+    const videoAgent = createVideoAgent()
+    let status: JobStatus | undefined
+    let blob: BlobRef | undefined
+    try {
+      const response = await videoAgent.app.bsky.video.getJobStatus({jobId})
+      status = response.data.jobStatus
+      pollFailures = 0
 
-      const {data} = await videoAgent.app.bsky.video.getJobStatus({jobId})
-      const status = data.jobStatus
       if (status.state === 'JOB_STATE_COMPLETED') {
-        setEnabled(false)
+        blob = status.blob
-        if (!status.blob)
+        if (!blob) {
           throw new Error('Job completed, but did not return a blob')
-        onSuccess(status.blob)
+        }
       } else if (status.state === 'JOB_STATE_FAILED') {
         throw new Error(status.error ?? 'Job failed to process')
       }
-      onStatusChange(status)
+    } catch (e) {
-      return status
+      if (!status) {
-    },
+        pollFailures++
-    enabled: Boolean(jobId && enabled),
+        if (pollFailures < 50) {
-    refetchInterval: 1500,
+          await new Promise(resolve => setTimeout(resolve, 5000))
-  })
+          continue // Continue async loop
+        }
+      }
 
-  useEffect(() => {
+      logger.error('Error processing video', {safeMessage: e})
-    if (error) {
+      dispatch({
-      onError(error)
+        type: 'to_error',
-      setEnabled(false)
+        error: _(msg`Video failed to process`),
+        signal,
+      })
+      return // Exit async loop
     }
-  }, [error, onError])
 
-  return {
+    if (blob) {
-    setJobId: (_jobId: string) => {
+      dispatch({
-      setJobId(_jobId)
+        type: 'to_done',
-      setEnabled(true)
+        blobRef: blob,
-    },
+        signal,
+      })
+    } else {
+      dispatch({
+        type: 'update_job_status',
+        jobStatus: status,
+        signal,
+      })
+    }
+
+    if (
+      status.state !== 'JOB_STATE_COMPLETED' &&
+      status.state !== 'JOB_STATE_FAILED'
+    ) {
+      await new Promise(resolve => setTimeout(resolve, 1500))
+      continue // Continue async loop
+    }
+
+    return // Exit async loop
   }
 }
 
-function getMimeType(asset: ImagePickerAsset) {
+function getCompressErrorMessage(e: unknown, _: I18n['_']): string | null {
-  if (isWeb) {
+  if (e instanceof AbortError) {
-    const [mimeType] = asset.uri.slice('data:'.length).split(';base64,')
+    return null
-    if (!mimeType) {
-      throw new Error('Could not determine mime type')
-    }
-    return mimeType
   }
-  if (!asset.mimeType) {
+  if (e instanceof VideoTooLargeError) {
-    throw new Error('Could not determine mime type')
+    return _(msg`The selected video is larger than 50MB.`)
   }
-  return asset.mimeType
+  logger.error('Error compressing video', {safeMessage: e})
+  return _(msg`An error occurred while compressing the video.`)
+}
+
+function getUploadErrorMessage(e: unknown, _: I18n['_']): string | null {
+  if (e instanceof AbortError) {
+    return null
+  }
+  logger.error('Error uploading video', {safeMessage: e})
+  if (e instanceof ServerError || e instanceof UploadLimitError) {
+    // https://github.com/bluesky-social/tango/blob/lumi/lumi/worker/permissions.go#L77
+    switch (e.message) {
+      case 'User is not allowed to upload videos':
+        return _(msg`You are not allowed to upload videos.`)
+      case 'Uploading is disabled at the moment':
+        return _(
+          msg`Hold up! We’re gradually giving access to video, and you’re still waiting in line. Check back soon!`,
+        )
+      case "Failed to get user's upload stats":
+        return _(
+          msg`We were unable to determine if you are allowed to upload videos. Please try again.`,
+        )
+      case 'User has exceeded daily upload bytes limit':
+        return _(
+          msg`You've reached your daily limit for video uploads (too many bytes)`,
+        )
+      case 'User has exceeded daily upload videos limit':
+        return _(
+          msg`You've reached your daily limit for video uploads (too many videos)`,
+        )
+      case 'Account is not old enough to upload videos':
+        return _(
+          msg`Your account is not yet old enough to upload videos. Please try again later.`,
+        )
+      default:
+        return e.message
+    }
+  }
+  return _(msg`An error occurred while uploading the video.`)
 }

src/view/com/composer/Composer.tsx

@@ -36,6 +36,7 @@ import Animated, {
   ZoomOut,
 } from 'react-native-reanimated'
 import {useSafeAreaInsets} from 'react-native-safe-area-context'
+import {ImagePickerAsset} from 'expo-image-picker'
 import {
   AppBskyFeedDefs,
   AppBskyFeedGetPostThread,
@@ -82,9 +83,10 @@ import {Gif} from '#/state/queries/tenor'
 import {ThreadgateAllowUISetting} from '#/state/queries/threadgate'
 import {threadgateViewToAllowUISetting} from '#/state/queries/threadgate/util'
 import {
+  createVideoState,
+  processVideo,
   State as VideoUploadState,
-  useUploadVideo,
+  videoReducer,
-  VideoUploadDispatch,
 } from '#/state/queries/video/video'
 import {useAgent, useSession} from '#/state/session'
 import {useComposerControls} from '#/state/shell/composer'
@@ -147,7 +149,8 @@ export const ComposePost = ({
 }) => {
   const {currentAccount} = useSession()
   const agent = useAgent()
-  const {data: currentProfile} = useProfileQuery({did: currentAccount!.did})
+  const currentDid = currentAccount!.did
+  const {data: currentProfile} = useProfileQuery({did: currentDid})
   const {isModalActive} = useModals()
   const {closeComposer} = useComposerControls()
   const pal = usePalette('default')
@@ -189,21 +192,50 @@ export const ComposePost = ({
   const [videoAltText, setVideoAltText] = useState('')
   const [captions, setCaptions] = useState<{lang: string; file: File}[]>([])
 
-  const {
+  const [videoUploadState, videoDispatch] = useReducer(
-    selectVideo,
+    videoReducer,
-    clearVideo,
+    undefined,
-    state: videoUploadState,
+    createVideoState,
-    updateVideoDimensions,
+  )
-    dispatch: videoUploadDispatch,
+
-  } = useUploadVideo({
+  const selectVideo = React.useCallback(
-    setStatus: setProcessingState,
+    (asset: ImagePickerAsset) => {
-    onSuccess: () => {
+      processVideo(
-      if (publishOnUpload) {
+        asset,
-        onPressPublish(true)
+        videoDispatch,
-      }
+        agent,
+        currentDid,
+        videoUploadState.abortController.signal,
+        _,
+      )
     },
-    initialVideoUri: initVideoUri,
+    [_, videoUploadState.abortController, videoDispatch, agent, currentDid],
-  })
+  )
+
+  // Whenever we receive an initial video uri, we should immediately run compression if necessary
+  useEffect(() => {
+    if (initVideoUri) {
+      selectVideo({uri: initVideoUri} as ImagePickerAsset)
+    }
+  }, [initVideoUri, selectVideo])
+
+  const clearVideo = React.useCallback(() => {
+    videoUploadState.abortController.abort()
+    videoDispatch({type: 'to_idle', nextController: new AbortController()})
+  }, [videoUploadState.abortController, videoDispatch])
+
+  const updateVideoDimensions = useCallback(
+    (width: number, height: number) => {
+      videoDispatch({
+        type: 'update_dimensions',
+        width,
+        height,
+        signal: videoUploadState.abortController.signal,
+      })
+    },
+    [videoUploadState.abortController],
+  )
+
   const hasVideo = Boolean(videoUploadState.asset || videoUploadState.video)
 
   const [publishOnUpload, setPublishOnUpload] = useState(false)
@@ -400,19 +432,18 @@ export const ComposePost = ({
             postgate,
             onStateChange: setProcessingState,
             langs: toPostLanguages(langPrefs.postLanguage),
-            video: videoUploadState.pendingPublish?.blobRef
+            video:
-              ? {
+              videoUploadState.status === 'done'
-                  blobRef: videoUploadState.pendingPublish.blobRef,
+                ? {
-                  altText: videoAltText,
+                    blobRef: videoUploadState.pendingPublish.blobRef,
-                  captions: captions,
+                    altText: videoAltText,
-                  aspectRatio: videoUploadState.asset
+                    captions: captions,
-                    ? {
+                    aspectRatio: {
-                        width: videoUploadState.asset?.width,
+                      width: videoUploadState.asset.width,
-                        height: videoUploadState.asset?.height,
+                      height: videoUploadState.asset.height,
-                      }
+                    },
-                    : undefined,
+                  }
-                }
+                : undefined,
-              : undefined,
           })
         ).uri
         try {
@@ -694,7 +725,7 @@ export const ComposePost = ({
             error={error}
             videoUploadState={videoUploadState}
             clearError={() => setError('')}
-            videoUploadDispatch={videoUploadDispatch}
+            clearVideo={clearVideo}
           />
         </Animated.View>
         <Animated.ScrollView
@@ -1083,25 +1114,25 @@ function ErrorBanner({
   error: standardError,
   videoUploadState,
   clearError,
-  videoUploadDispatch,
+  clearVideo,
 }: {
   error: string
   videoUploadState: VideoUploadState
   clearError: () => void
-  videoUploadDispatch: VideoUploadDispatch
+  clearVideo: () => void
 }) {
   const t = useTheme()
   const {_} = useLingui()
 
   const videoError =
-    videoUploadState.status !== 'idle' ? videoUploadState.error : undefined
+    videoUploadState.status === 'error' ? videoUploadState.error : undefined
   const error = standardError || videoError
 
   const onClearError = () => {
     if (standardError) {
       clearError()
     } else {
-      videoUploadDispatch({type: 'Reset'})
+      clearVideo()
     }
   }
 
@@ -1136,7 +1167,7 @@ function ErrorBanner({
             <ButtonIcon icon={X} />
           </Button>
         </View>
-        {videoError && videoUploadState.jobStatus?.jobId && (
+        {videoError && videoUploadState.jobId && (
           <NewText
             style={[
               {paddingLeft: 28},
@@ -1145,7 +1176,7 @@ function ErrorBanner({
               a.leading_snug,
               t.atoms.text_contrast_low,
             ]}>
-            <Trans>Job ID: {videoUploadState.jobStatus.jobId}</Trans>
+            <Trans>Job ID: {videoUploadState.jobId}</Trans>
           </NewText>
         )}
       </View>
@@ -1174,9 +1205,7 @@ function ToolbarWrapper({
 function VideoUploadToolbar({state}: {state: VideoUploadState}) {
   const t = useTheme()
   const {_} = useLingui()
-  const progress = state.jobStatus?.progress
+  const progress = state.progress
-    ? state.jobStatus.progress / 100
-    : state.progress
   const shouldRotate =
     state.status === 'processing' && (progress === 0 || progress === 1)
   let wheelProgress = shouldRotate ? 0.33 : progress
@@ -1212,16 +1241,15 @@ function VideoUploadToolbar({state}: {state: VideoUploadState}) {
     case 'processing':
       text = _('Processing video...')
       break
+    case 'error':
+      text = _('Error')
+      wheelProgress = 100
+      break
     case 'done':
       text = _('Video uploaded')
       break
   }
 
-  if (state.error) {
-    text = _('Error')
-    wheelProgress = 100
-  }
-
   return (
     <ToolbarWrapper style={[a.flex_row, a.align_center, {paddingVertical: 5}]}>
       <Animated.View style={[animatedStyle]}>
@@ -1229,7 +1257,11 @@ function VideoUploadToolbar({state}: {state: VideoUploadState}) {
           size={30}
           borderWidth={1}
           borderColor={t.atoms.border_contrast_low.borderColor}
-          color={state.error ? t.palette.negative_500 : t.palette.primary_500}
+          color={
+            state.status === 'error'
+              ? t.palette.negative_500
+              : t.palette.primary_500
+          }
           progress={wheelProgress}
         />
       </Animated.View>

src/view/com/composer/videos/SelectVideoBtn.tsx

@@ -9,12 +9,14 @@ import {
 import {msg} from '@lingui/macro'
 import {useLingui} from '@lingui/react'
 
+import {SUPPORTED_MIME_TYPES, SupportedMimeTypes} from '#/lib/constants'
+import {BSKY_SERVICE} from '#/lib/constants'
 import {useVideoLibraryPermission} from '#/lib/hooks/usePermissions'
+import {getHostnameFromUrl} from '#/lib/strings/url-helpers'
+import {isWeb} from '#/platform/detection'
 import {isNative} from '#/platform/detection'
 import {useModalControls} from '#/state/modals'
 import {useSession} from '#/state/session'
-import {BSKY_SERVICE} from 'lib/constants'
-import {getHostnameFromUrl} from 'lib/strings/url-helpers'
 import {atoms as a, useTheme} from '#/alf'
 import {Button} from '#/components/Button'
 import {VideoClip_Stroke2_Corner0_Rounded as VideoClipIcon} from '#/components/icons/VideoClip'
@@ -58,16 +60,25 @@ export function SelectVideoBtn({onSelectVideo, disabled, setError}: Props) {
           UIImagePickerPreferredAssetRepresentationMode.Current,
       })
       if (response.assets && response.assets.length > 0) {
-        if (isNative) {
+        const asset = response.assets[0]
-          if (typeof response.assets[0].duration !== 'number')
-            throw Error('Asset is not a video')
-          if (response.assets[0].duration > VIDEO_MAX_DURATION) {
-            setError(_(msg`Videos must be less than 60 seconds long`))
-            return
-          }
-        }
         try {
-          onSelectVideo(response.assets[0])
+          if (isWeb) {
+            // compression step on native converts to mp4, so no need to check there
+            const mimeType = getMimeType(asset)
+            if (
+              !SUPPORTED_MIME_TYPES.includes(mimeType as SupportedMimeTypes)
+            ) {
+              throw Error(_(msg`Unsupported video type: ${mimeType}`))
+            }
+          } else {
+            if (typeof asset.duration !== 'number') {
+              throw Error('Asset is not a video')
+            }
+            if (asset.duration > VIDEO_MAX_DURATION) {
+              throw Error(_(msg`Videos must be less than 60 seconds long`))
+            }
+          }
+          onSelectVideo(asset)
         } catch (err) {
           if (err instanceof Error) {
             setError(err.message)
@@ -132,3 +143,17 @@ function VerifyEmailPrompt({control}: {control: Prompt.PromptControlProps}) {
     />
   )
 }
+
+function getMimeType(asset: ImagePickerAsset) {
+  if (isWeb) {
+    const [mimeType] = asset.uri.slice('data:'.length).split(';base64,')
+    if (!mimeType) {
+      throw new Error('Could not determine mime type')
+    }
+    return mimeType
+  }
+  if (!asset.mimeType) {
+    throw new Error('Could not determine mime type')
+  }
+  return asset.mimeType
+}