[Video] Add uploaded video to post (#4884)

* video uploads!

* use video upload lexicons

* add missing postgate

* remove references to prerelease package

* fix scrubber showing a "0"

* Delete types.ts

* rm logs

* rm upload header

---------

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

src/lib/api/index.ts

@@ -3,12 +3,14 @@ import {
   AppBskyEmbedImages,
   AppBskyEmbedRecord,
   AppBskyEmbedRecordWithMedia,
+  AppBskyEmbedVideo,
   AppBskyFeedPostgate,
+  AtUri,
+  BlobRef,
   BskyAgent,
   ComAtprotoLabelDefs,
   RichText,
 } from '@atproto/api'
-import {AtUri} from '@atproto/api'
 
 import {logger} from '#/logger'
 import {writePostgateRecord} from '#/state/queries/postgate'
@@ -43,10 +45,7 @@ interface PostOpts {
     uri: string
     cid: string
   }
-  video?: {
+  video?: BlobRef
-    uri: string
-    cid: string
-  }
   extLink?: ExternalEmbedDraft
   images?: ImageModel[]
   labels?: string[]
@@ -61,18 +60,16 @@ export async function post(agent: BskyAgent, opts: PostOpts) {
     | AppBskyEmbedImages.Main
     | AppBskyEmbedExternal.Main
     | AppBskyEmbedRecord.Main
+    | AppBskyEmbedVideo.Main
     | AppBskyEmbedRecordWithMedia.Main
     | undefined
   let reply
-  let rt = new RichText(
+  let rt = new RichText({text: opts.rawText.trimEnd()}, {cleanNewlines: true})
-    {text: opts.rawText.trimEnd()},
-    {
-      cleanNewlines: true,
-    },
-  )
 
   opts.onStateChange?.('Processing...')
+
   await rt.detectFacets(agent)
+
   rt = shortenLinks(rt)
   rt = stripInvalidMentions(rt)
 
@@ -129,6 +126,25 @@ export async function post(agent: BskyAgent, opts: PostOpts) {
     }
   }
 
+  // add video embed if present
+  if (opts.video) {
+    if (opts.quote) {
+      embed = {
+        $type: 'app.bsky.embed.recordWithMedia',
+        record: embed,
+        media: {
+          $type: 'app.bsky.embed.video',
+          video: opts.video,
+        } as AppBskyEmbedVideo.Main,
+      } as AppBskyEmbedRecordWithMedia.Main
+    } else {
+      embed = {
+        $type: 'app.bsky.embed.video',
+        video: opts.video,
+      } as AppBskyEmbedVideo.Main
+    }
+  }
+
   // add external embed if present
   if (opts.extLink && !opts.images?.length) {
     if (opts.extLink.embed) {

src/lib/media/video/types.ts

@@ -1,36 +0,0 @@
-/**
- * TEMPORARY: THIS IS A TEMPORARY PLACEHOLDER. THAT MEANS IT IS TEMPORARY. I.E. WILL BE REMOVED. NOT TO USE IN PRODUCTION.
- * @temporary
- * PS: This is a temporary placeholder for the video types. It will be removed once the actual types are implemented.
- * Not joking, this is temporary.
- */
-
-export interface JobStatus {
-  jobId: string
-  did: string
-  cid: string
-  state: JobState
-  progress?: number
-  errorHuman?: string
-  errorMachine?: string
-}
-
-export enum JobState {
-  JOB_STATE_UNSPECIFIED = 'JOB_STATE_UNSPECIFIED',
-  JOB_STATE_CREATED = 'JOB_STATE_CREATED',
-  JOB_STATE_ENCODING = 'JOB_STATE_ENCODING',
-  JOB_STATE_ENCODED = 'JOB_STATE_ENCODED',
-  JOB_STATE_UPLOADING = 'JOB_STATE_UPLOADING',
-  JOB_STATE_UPLOADED = 'JOB_STATE_UPLOADED',
-  JOB_STATE_CDN_PROCESSING = 'JOB_STATE_CDN_PROCESSING',
-  JOB_STATE_CDN_PROCESSED = 'JOB_STATE_CDN_PROCESSED',
-  JOB_STATE_FAILED = 'JOB_STATE_FAILED',
-  JOB_STATE_COMPLETED = 'JOB_STATE_COMPLETED',
-}
-
-export interface UploadVideoResponse {
-  job_id: string
-  did: string
-  cid: string
-  state: JobState
-}

src/state/queries/video/util.ts

@@ -1,3 +1,6 @@
+import {useMemo} from 'react'
+import {AtpAgent} from '@atproto/api'
+
 const UPLOAD_ENDPOINT = process.env.EXPO_PUBLIC_VIDEO_ROOT_ENDPOINT ?? ''
 
 export const createVideoEndpointUrl = (
@@ -13,3 +16,11 @@ export const createVideoEndpointUrl = (
   }
   return url.href
 }
+
+export function useVideoAgent() {
+  return useMemo(() => {
+    return new AtpAgent({
+      service: UPLOAD_ENDPOINT,
+    })
+  }, [])
+}

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

@@ -1,20 +1,18 @@
 import {createUploadTask, FileSystemUploadType} from 'expo-file-system'
+import {AppBskyVideoDefs} from '@atproto/api'
 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 {useAgent, useSession} from '#/state/session'
 
-const UPLOAD_HEADER = process.env.EXPO_PUBLIC_VIDEO_HEADER ?? ''
-
 export const useUploadVideoMutation = ({
   onSuccess,
   onError,
   setProgress,
 }: {
-  onSuccess: (response: UploadVideoResponse) => void
+  onSuccess: (response: AppBskyVideoDefs.JobStatus) => void
   onError: (e: any) => void
   setProgress: (progress: number) => void
 }) => {
@@ -23,7 +21,7 @@ export const useUploadVideoMutation = ({
 
   return useMutation({
     mutationFn: async (video: CompressedVideo) => {
-      const uri = createVideoEndpointUrl('/upload', {
+      const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
         did: currentAccount!.did,
         name: `${nanoid(12)}.mp4`, // @TODO what are we limiting this to?
       })
@@ -33,19 +31,19 @@ export const useUploadVideoMutation = ({
         throw new Error('Agent does not have a PDS URL')
       }
 
-      const {data: serviceAuth} =
+      const {data: serviceAuth} = await agent.com.atproto.server.getServiceAuth(
-        await agent.api.com.atproto.server.getServiceAuth({
+        {
           aud: `did:web:${agent.pdsUrl.hostname}`,
           lxm: 'com.atproto.repo.uploadBlob',
-        })
+        },
+      )
 
       const uploadTask = createUploadTask(
         uri,
         video.uri,
         {
           headers: {
-            'dev-key': UPLOAD_HEADER,
+            'content-type': 'video/mp4',
-            'content-type': 'video/mp4', // @TODO same question here. does the compression step always output mp4?
             Authorization: `Bearer ${serviceAuth.token}`,
           },
           httpMethod: 'POST',
@@ -59,10 +57,7 @@ export const useUploadVideoMutation = ({
         throw new Error('No response')
       }
 
-      // @TODO rm, useful for debugging/getting video cid
+      const responseBody = JSON.parse(res.body) as AppBskyVideoDefs.JobStatus
-      console.log('[VIDEO]', res.body)
-      const responseBody = JSON.parse(res.body) as UploadVideoResponse
-      onSuccess(responseBody)
       return responseBody
     },
     onError,

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

@@ -1,19 +1,17 @@
+import {AppBskyVideoDefs} from '@atproto/api'
 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 {useAgent, useSession} from '#/state/session'
 
-const UPLOAD_HEADER = process.env.EXPO_PUBLIC_VIDEO_HEADER ?? ''
-
 export const useUploadVideoMutation = ({
   onSuccess,
   onError,
   setProgress,
 }: {
-  onSuccess: (response: UploadVideoResponse) => void
+  onSuccess: (response: AppBskyVideoDefs.JobStatus) => void
   onError: (e: any) => void
   setProgress: (progress: number) => void
 }) => {
@@ -22,9 +20,9 @@ export const useUploadVideoMutation = ({
 
   return useMutation({
     mutationFn: async (video: CompressedVideo) => {
-      const uri = createVideoEndpointUrl('/upload', {
+      const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
         did: currentAccount!.did,
-        name: `${nanoid(12)}.mp4`, // @TODO what are we limiting this to?
+        name: `${nanoid(12)}.mp4`, // @TODO: make sure it's always mp4'
       })
 
       // a logged-in agent should have this set, but we'll check just in case
@@ -32,16 +30,18 @@ export const useUploadVideoMutation = ({
         throw new Error('Agent does not have a PDS URL')
       }
 
-      const {data: serviceAuth} =
+      const {data: serviceAuth} = await agent.com.atproto.server.getServiceAuth(
-        await agent.api.com.atproto.server.getServiceAuth({
+        {
           aud: `did:web:${agent.pdsUrl.hostname}`,
           lxm: 'com.atproto.repo.uploadBlob',
-        })
+        },
+      )
 
       const bytes = await fetch(video.uri).then(res => res.arrayBuffer())
 
       const xhr = new XMLHttpRequest()
-      const res = (await new Promise((resolve, reject) => {
+      const res = await new Promise<AppBskyVideoDefs.JobStatus>(
+        (resolve, reject) => {
           xhr.upload.addEventListener('progress', e => {
             const progress = e.loaded / e.total
             setProgress(progress)
@@ -50,7 +50,7 @@ export const useUploadVideoMutation = ({
             if (xhr.readyState === 4) {
               const uploadRes = JSON.parse(
                 xhr.responseText,
-            ) as UploadVideoResponse
+              ) as AppBskyVideoDefs.JobStatus
               resolve(uploadRes)
               onSuccess(uploadRes)
             } else {
@@ -63,15 +63,12 @@ export const useUploadVideoMutation = ({
             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?
+          xhr.setRequestHeader('Content-Type', 'video/mp4')
-        // @TODO remove this header for prod
-        xhr.setRequestHeader('dev-key', UPLOAD_HEADER)
           xhr.setRequestHeader('Authorization', `Bearer ${serviceAuth.token}`)
           xhr.send(bytes)
-      })) as UploadVideoResponse
+        },
+      )
 
-      // @TODO rm for prod
-      console.log('[VIDEO]', res)
       return res
     },
     onError,

src/state/queries/video/video.ts

@@ -1,5 +1,6 @@
 import React from 'react'
 import {ImagePickerAsset} from 'expo-image-picker'
+import {AppBskyVideoDefs, BlobRef} from '@atproto/api'
 import {msg} from '@lingui/macro'
 import {useLingui} from '@lingui/react'
 import {useQuery} from '@tanstack/react-query'
@@ -7,37 +8,29 @@ 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 {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: 'SetStatus'
+  | {type: 'SetProgress'; progress: number}
-      status: Status
+  | {type: 'SetError'; error: string | undefined}
-    }
-  | {
-      type: 'SetProgress'
-      progress: number
-    }
-  | {
-      type: 'SetError'
-      error: string | undefined
-    }
   | {type: 'Reset'}
   | {type: 'SetAsset'; asset: ImagePickerAsset}
   | {type: 'SetVideo'; video: CompressedVideo}
-  | {type: 'SetJobStatus'; jobStatus: JobStatus}
+  | {type: 'SetJobStatus'; jobStatus: AppBskyVideoDefs.JobStatus}
+  | {type: 'SetBlobRef'; blobRef: BlobRef}
 
 export interface State {
   status: Status
   progress: number
   asset?: ImagePickerAsset
   video: CompressedVideo | null
-  jobStatus?: JobStatus
+  jobStatus?: AppBskyVideoDefs.JobStatus
+  blobRef?: BlobRef
   error?: string
 }
 
@@ -54,6 +47,7 @@ function reducer(state: State, action: Action): State {
       status: 'idle',
       progress: 0,
       video: null,
+      blobRef: undefined,
     }
   } else if (action.type === 'SetAsset') {
     updatedState = {...state, asset: action.asset}
@@ -61,6 +55,8 @@ function reducer(state: State, action: Action): State {
     updatedState = {...state, video: action.video}
   } else if (action.type === 'SetJobStatus') {
     updatedState = {...state, jobStatus: action.jobStatus}
+  } else if (action.type === 'SetBlobRef') {
+    updatedState = {...state, blobRef: action.blobRef}
   }
   return updatedState
 }
@@ -80,7 +76,7 @@ export function useUploadVideo({
   })
 
   const {setJobId} = useUploadStatusQuery({
-    onStatusChange: (status: JobStatus) => {
+    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({
@@ -89,7 +85,11 @@ export function useUploadVideo({
       })
       setStatus(status.state.toString())
     },
-    onSuccess: () => {
+    onSuccess: blobRef => {
+      dispatch({
+        type: 'SetBlobRef',
+        blobRef,
+      })
       dispatch({
         type: 'SetStatus',
         status: 'idle',
@@ -104,7 +104,7 @@ export function useUploadVideo({
         type: 'SetStatus',
         status: 'processing',
       })
-      setJobId(response.job_id)
+      setJobId(response.jobId)
     },
     onError: e => {
       dispatch({
@@ -179,21 +179,27 @@ const useUploadStatusQuery = ({
   onStatusChange,
   onSuccess,
 }: {
-  onStatusChange: (status: JobStatus) => void
+  onStatusChange: (status: AppBskyVideoDefs.JobStatus) => void
-  onSuccess: () => void
+  onSuccess: (blobRef: BlobRef) => void
 }) => {
+  const videoAgent = useVideoAgent()
   const [enabled, setEnabled] = React.useState(true)
   const [jobId, setJobId] = React.useState<string>()
 
   const {isLoading, isError} = useQuery({
-    queryKey: ['video-upload'],
+    queryKey: ['video-upload', jobId],
     queryFn: async () => {
-      const url = createVideoEndpointUrl(`/job/${jobId}/status`)
+      if (!jobId) return // this won't happen, can ignore
-      const res = await fetch(url)
+
-      const status = (await res.json()) as JobStatus
+      const {data} = await videoAgent.app.bsky.video.getJobStatus({jobId})
-      if (status.state === JobState.JOB_STATE_COMPLETED) {
+      const status = data.jobStatus
+      if (status.state === 'JOB_STATE_COMPLETED') {
         setEnabled(false)
-        onSuccess()
+        if (!status.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('Job failed to process')
       }
       onStatusChange(status)
       return status

src/view/com/composer/Composer.tsx

@@ -178,7 +178,7 @@ export const ComposePost = observer(function ComposePost({
     clearVideo,
     state: videoUploadState,
   } = useUploadVideo({
-    setStatus: (status: string) => setProcessingState(status),
+    setStatus: setProcessingState,
     onSuccess: () => {
       if (publishOnUpload) {
         onPressPublish(true)
@@ -348,6 +348,7 @@ export const ComposePost = observer(function ComposePost({
           postgate,
           onStateChange: setProcessingState,
           langs: toPostLanguages(langPrefs.postLanguage),
+          video: videoUploadState.blobRef,
         })
       ).uri
       try {

src/view/com/util/post-embeds/VideoEmbedInner/VideoWebControls.tsx

@@ -557,7 +557,7 @@ function Scrubber({
             {backgroundColor: 'rgba(255, 255, 255, 0.4)'},
             {height: hovered || scrubberActive ? 6 : 3},
           ]}>
-          {currentTime && duration && (
+          {currentTime > 0 && duration > 0 && (
             <View
               style={[
                 a.h_full,