[Video] Uploads (#4754)

* state for video uploads

* get upload working

* add a debug log

* add post progress

* progress

* fetch data

* add some progress info, web uploads

* post on finished uploading (wip)

* add a note

* add some todos

* clear video

* merge some stuff

* convert to `createUploadTask`

* patch expo modules core

* working native upload progress

* platform fork

* upload progress for web

* cleanup

* cleanup

* more tweaks

* simplify

* fix type errors

---------

Co-authored-by: Samuel Newman <10959775+mozzius@users.noreply.github.com>

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

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

src/state/queries/video/util.ts

@@ -0,0 +1,15 @@
+const UPLOAD_ENDPOINT = process.env.EXPO_PUBLIC_VIDEO_ROOT_ENDPOINT ?? ''
+
+export const createVideoEndpointUrl = (
+  route: string,
+  params?: Record<string, string>,
+) => {
+  const url = new URL(`${UPLOAD_ENDPOINT}`)
+  url.pathname = route
+  if (params) {
+    for (const key in params) {
+      url.searchParams.set(key, params[key])
+    }
+  }
+  return url.href
+}

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

@@ -0,0 +1,59 @@
+import {createUploadTask, FileSystemUploadType} from 'expo-file-system'
+import {useMutation} from '@tanstack/react-query'
+import {nanoid} from 'nanoid/non-secure'
+
+import {CompressedVideo} from 'lib/media/video/compress'
+import {UploadVideoResponse} from 'lib/media/video/types'
+import {createVideoEndpointUrl} from 'state/queries/video/util'
+import {useSession} from 'state/session'
+const UPLOAD_HEADER = process.env.EXPO_PUBLIC_VIDEO_HEADER ?? ''
+
+export const useUploadVideoMutation = ({
+  onSuccess,
+  onError,
+  setProgress,
+}: {
+  onSuccess: (response: UploadVideoResponse) => void
+  onError: (e: any) => void
+  setProgress: (progress: number) => void
+}) => {
+  const {currentAccount} = useSession()
+
+  return useMutation({
+    mutationFn: async (video: CompressedVideo) => {
+      const uri = createVideoEndpointUrl('/upload', {
+        did: currentAccount!.did,
+        name: `${nanoid(12)}.mp4`, // @TODO what are we limiting this to?
+      })
+
+      const uploadTask = createUploadTask(
+        uri,
+        video.uri,
+        {
+          headers: {
+            'dev-key': UPLOAD_HEADER,
+            'content-type': 'video/mp4', // @TODO same question here. does the compression step always output mp4?
+          },
+          httpMethod: 'POST',
+          uploadType: FileSystemUploadType.BINARY_CONTENT,
+        },
+        p => {
+          setProgress(p.totalBytesSent / p.totalBytesExpectedToSend)
+        },
+      )
+      const res = await uploadTask.uploadAsync()
+
+      if (!res?.body) {
+        throw new Error('No response')
+      }
+
+      // @TODO rm, useful for debugging/getting video cid
+      console.log('[VIDEO]', res.body)
+      const responseBody = JSON.parse(res.body) as UploadVideoResponse
+      onSuccess(responseBody)
+      return responseBody
+    },
+    onError,
+    onSuccess,
+  })
+}

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

@@ -0,0 +1,66 @@
+import {useMutation} from '@tanstack/react-query'
+import {nanoid} from 'nanoid/non-secure'
+
+import {CompressedVideo} from 'lib/media/video/compress'
+import {UploadVideoResponse} from 'lib/media/video/types'
+import {createVideoEndpointUrl} from 'state/queries/video/util'
+import {useSession} from 'state/session'
+const UPLOAD_HEADER = process.env.EXPO_PUBLIC_VIDEO_HEADER ?? ''
+
+export const useUploadVideoMutation = ({
+  onSuccess,
+  onError,
+  setProgress,
+}: {
+  onSuccess: (response: UploadVideoResponse) => void
+  onError: (e: any) => void
+  setProgress: (progress: number) => void
+}) => {
+  const {currentAccount} = useSession()
+
+  return useMutation({
+    mutationFn: async (video: CompressedVideo) => {
+      const uri = createVideoEndpointUrl('/upload', {
+        did: currentAccount!.did,
+        name: `${nanoid(12)}.mp4`, // @TODO what are we limiting this to?
+      })
+
+      const bytes = await fetch(video.uri).then(res => res.arrayBuffer())
+
+      const xhr = new XMLHttpRequest()
+      const res = (await new Promise((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 UploadVideoResponse
+            resolve(uploadRes)
+            onSuccess(uploadRes)
+          } else {
+            reject()
+            onError(new Error('Failed to upload video'))
+          }
+        }
+        xhr.onerror = () => {
+          reject()
+          onError(new Error('Failed to upload video'))
+        }
+        xhr.open('POST', uri)
+        xhr.setRequestHeader('Content-Type', 'video/mp4') // @TODO how we we set the proper content type?
+        // @TODO remove this header for prod
+        xhr.setRequestHeader('dev-key', UPLOAD_HEADER)
+        xhr.send(bytes)
+      })) as UploadVideoResponse
+
+      // @TODO rm for prod
+      console.log('[VIDEO]', res)
+      return res
+    },
+    onError,
+    onSuccess,
+  })
+}

src/state/queries/video/video.ts

@@ -0,0 +1,212 @@
+import React from 'react'
+import {ImagePickerAsset} from 'expo-image-picker'
+import {msg} from '@lingui/macro'
+import {useLingui} from '@lingui/react'
+import {useQuery} from '@tanstack/react-query'
+
+import {logger} from '#/logger'
+import {CompressedVideo} from 'lib/media/video/compress'
+import {VideoTooLargeError} from 'lib/media/video/errors'
+import {JobState, JobStatus} from 'lib/media/video/types'
+import {useCompressVideoMutation} from 'state/queries/video/compress-video'
+import {createVideoEndpointUrl} 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: 'SetProgress'
+      progress: number
+    }
+  | {
+      type: 'SetError'
+      error: string | undefined
+    }
+  | {type: 'Reset'}
+  | {type: 'SetAsset'; asset: ImagePickerAsset}
+  | {type: 'SetVideo'; video: CompressedVideo}
+  | {type: 'SetJobStatus'; jobStatus: JobStatus}
+
+export interface State {
+  status: Status
+  progress: number
+  asset?: ImagePickerAsset
+  video: CompressedVideo | null
+  jobStatus?: JobStatus
+  error?: string
+}
+
+function reducer(state: State, action: Action): State {
+  let updatedState = state
+  if (action.type === 'SetStatus') {
+    updatedState = {...state, status: action.status}
+  } else if (action.type === 'SetProgress') {
+    updatedState = {...state, progress: action.progress}
+  } else if (action.type === 'SetError') {
+    updatedState = {...state, error: action.error}
+  } else if (action.type === 'Reset') {
+    updatedState = {
+      status: 'idle',
+      progress: 0,
+      video: null,
+    }
+  } else if (action.type === 'SetAsset') {
+    updatedState = {...state, asset: action.asset}
+  } else if (action.type === 'SetVideo') {
+    updatedState = {...state, video: action.video}
+  } else if (action.type === 'SetJobStatus') {
+    updatedState = {...state, jobStatus: action.jobStatus}
+  }
+  return updatedState
+}
+
+export function useUploadVideo({
+  setStatus,
+  onSuccess,
+}: {
+  setStatus: (status: string) => void
+  onSuccess: () => void
+}) {
+  const {_} = useLingui()
+  const [state, dispatch] = React.useReducer(reducer, {
+    status: 'idle',
+    progress: 0,
+    video: null,
+  })
+
+  const {setJobId} = useUploadStatusQuery({
+    onStatusChange: (status: 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: () => {
+      dispatch({
+        type: 'SetStatus',
+        status: 'idle',
+      })
+      onSuccess()
+    },
+  })
+
+  const {mutate: onVideoCompressed} = useUploadVideoMutation({
+    onSuccess: response => {
+      dispatch({
+        type: 'SetStatus',
+        status: 'processing',
+      })
+      setJobId(response.job_id)
+    },
+    onError: e => {
+      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})
+    },
+  })
+
+  const {mutate: onSelectVideo} = useCompressVideoMutation({
+    onProgress: p => {
+      dispatch({type: 'SetProgress', progress: p})
+    },
+    onError: e => {
+      if (e instanceof VideoTooLargeError) {
+        dispatch({
+          type: 'SetError',
+          error: _(msg`The selected video is larger than 100MB.`),
+        })
+      } else {
+        dispatch({
+          type: 'SetError',
+          // @TODO better error message from server, left untranslated on purpose
+          error: 'An error occurred while compressing the video.',
+        })
+        logger.error('Error compressing video', {safeMessage: e})
+      }
+    },
+    onSuccess: (video: CompressedVideo) => {
+      dispatch({
+        type: 'SetVideo',
+        video,
+      })
+      dispatch({
+        type: 'SetStatus',
+        status: 'uploading',
+      })
+      onVideoCompressed(video)
+    },
+  })
+
+  const selectVideo = (asset: ImagePickerAsset) => {
+    dispatch({
+      type: 'SetAsset',
+      asset,
+    })
+    dispatch({
+      type: 'SetStatus',
+      status: 'compressing',
+    })
+    onSelectVideo(asset)
+  }
+
+  const clearVideo = () => {
+    // @TODO cancel any running jobs
+    dispatch({type: 'Reset'})
+  }
+
+  return {
+    state,
+    dispatch,
+    selectVideo,
+    clearVideo,
+  }
+}
+
+const useUploadStatusQuery = ({
+  onStatusChange,
+  onSuccess,
+}: {
+  onStatusChange: (status: JobStatus) => void
+  onSuccess: () => void
+}) => {
+  const [enabled, setEnabled] = React.useState(true)
+  const [jobId, setJobId] = React.useState<string>()
+
+  const {isLoading, isError} = useQuery({
+    queryKey: ['video-upload'],
+    queryFn: async () => {
+      const url = createVideoEndpointUrl(`/job/${jobId}/status`)
+      const res = await fetch(url)
+      const status = (await res.json()) as JobStatus
+      if (status.state === JobState.JOB_STATE_COMPLETED) {
+        setEnabled(false)
+        onSuccess()
+      }
+      onStatusChange(status)
+      return status
+    },
+    enabled: Boolean(jobId && enabled),
+    refetchInterval: 1500,
+  })
+
+  return {
+    isLoading,
+    isError,
+    setJobId: (_jobId: string) => {
+      setJobId(_jobId)
+    },
+  }
+}

src/state/shell/post-progress.tsx

@@ -0,0 +1,18 @@
+import React from 'react'
+
+interface PostProgressState {
+  progress: number
+  status: 'pending' | 'success' | 'error' | 'idle'
+  error?: string
+}
+
+const PostProgressContext = React.createContext<PostProgressState>({
+  progress: 0,
+  status: 'idle',
+})
+
+export function Provider() {}
+
+export function usePostProgress() {
+  return React.useContext(PostProgressContext)
+}