qstash publish receive

src/app/api/batches/route.ts

@@ -1,7 +1,9 @@
+import { Prisma } from '@prisma/client'
 import { prisma } from '@/lib/prisma'
 import { randomUUID } from 'crypto'
 import { NextResponse } from 'next/server'
 import { z } from 'zod'
+import { publishMessage } from '@/lib/qstash'
 
 const createBatchSchema = z.object({
   files: z
@@ -24,24 +26,42 @@ export async function POST(request: Request) {
   const batchId = randomUUID()
 
   try {
-    await prisma.uploadBatch.create({
-      data: {
-        id: batchId,
-        videos: {
-          createMany: {
-            data: files.map((file, index) => {
-              return {
-                uploadOrder: index + 1,
-                title: file.title,
-                storageKey: `inputs/${file.id}.mp4`,
-                audioStorageKey: `inputs/${file.id}.mp3`,
-                sizeInBytes: file.sizeInBytes,
-                duration: file.duration,
-              }
-            }),
+    await prisma.$transaction(async (tx) => {
+      const videos: Prisma.VideoCreateManyUploadBatchInput[] = files.map(
+        (file, index) => {
+          return {
+            id: randomUUID(),
+            uploadOrder: index + 1,
+            title: file.title,
+            storageKey: `inputs/${file.id}.mp4`,
+            audioStorageKey: `inputs/${file.id}.mp3`,
+            sizeInBytes: file.sizeInBytes,
+            duration: file.duration,
+          }
+        },
+      )
+
+      await tx.uploadBatch.create({
+        data: {
+          id: batchId,
+          videos: {
+            createMany: {
+              data: videos,
+            },
           },
         },
-      },
+      })
+
+      await Promise.all(
+        videos.map(async (video) => {
+          await publishMessage({
+            topic: 'jupiter.upload-created',
+            body: {
+              videoId: video.id,
+            },
+          })
+        }),
+      )
     })
 
     return NextResponse.json({ batchId })

src/app/api/webhooks/create-video-transcription/route.ts

@@ -6,6 +6,7 @@ import FormData from 'form-data'
 import axios from 'axios'
 import { z } from 'zod'
 import { env } from '@/env'
+import { validateQStashSignature } from '@/lib/qstash'
 
 const createTranscriptionBodySchema = z.object({
   videoId: z.string().uuid(),
@@ -20,9 +21,11 @@ interface OpenAITranscriptionResponse {
 }
 
 export async function POST(request: Request) {
-  const { videoId } = createTranscriptionBodySchema.parse(await request.json())
-
   try {
+    const { bodyAsJSON } = await validateQStashSignature({ request })
+
+    const { videoId } = createTranscriptionBodySchema.parse(bodyAsJSON)
+
     const video = await prisma.video.findUniqueOrThrow({
       where: {
         id: videoId,
@@ -107,7 +110,12 @@ export async function POST(request: Request) {
     })
 
     return new Response()
-  } catch (err) {
-    console.log(err)
+  } catch (err: any) {
+    console.error(err)
+
+    return NextResponse.json(
+      { message: 'Error processing video.', error: err?.message || '' },
+      { status: 401 },
+    )
   }
 }

src/app/api/webhooks/process-video/route.ts

@@ -3,63 +3,29 @@ import { r2 } from '@/lib/cloudflare-r2'
 import { prisma } from '@/lib/prisma'
 import { CopyObjectCommand } from '@aws-sdk/client-s3'
 import { NextResponse } from 'next/server'
-import { Receiver } from '@upstash/qstash/nodejs'
 import { z } from 'zod'
+import { publishMessage, validateQStashSignature } from '@/lib/qstash'
 
 const processVideoBodySchema = z.object({
   videoId: z.string().uuid(),
 })
 
 export async function POST(request: Request) {
-  if (env.NODE_ENV === 'production') {
-    const signature = request.headers.get('upstash-signature')
-
-    const receiver = new Receiver({
-      currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
-      nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
-    })
-
-    if (!signature) {
-      return NextResponse.json(
-        { message: 'QStash signature not found.' },
-        { status: 401 },
-      )
-    }
-
-    const isValid = await receiver
-      .verify({
-        signature,
-        body: await request.text(),
-      })
-      .catch((err) => {
-        console.error(err)
-        return false
-      })
-
-    if (!isValid) {
-      return NextResponse.json(
-        { message: 'QStash signature is invalid.' },
-        { status: 401 },
-      )
-    }
-  }
+  try {
+    const { bodyAsJSON } = await validateQStashSignature({ request })
 
-  const { videoId } = processVideoBodySchema.parse(await request.json())
+    const { videoId } = processVideoBodySchema.parse(bodyAsJSON)
 
-  try {
     const video = await prisma.video.findUniqueOrThrow({
       where: {
         id: videoId,
       },
     })
 
     if (video.processedAt) {
-      return NextResponse.json(
-        { message: 'Video has already been processed.' },
-        {
-          status: 409,
-        },
-      )
+      return NextResponse.json({
+        message: 'Video has already been processed.',
+      })
     }
 
     const bucket = env.CLOUDFLARE_BUCKET_NAME
@@ -97,12 +63,20 @@ export async function POST(request: Request) {
       },
     })
 
-    /**
-     * TODO: Add transcription and external provider upload to QStash
-     */
+    await publishMessage({
+      topic: 'jupiter.upload-processed',
+      body: {
+        videoId,
+      },
+    })
 
     return new Response()
-  } catch (err) {
-    console.log(err)
+  } catch (err: any) {
+    console.error(err)
+
+    return NextResponse.json(
+      { message: 'Error processing video.', error: err?.message || '' },
+      { status: 401 },
+    )
   }
 }

src/app/api/webhooks/update-external-provider-status/route.ts

@@ -4,7 +4,6 @@ import { z } from 'zod'
 const pandaWebhookBodySchema = z.object({
   action: z.enum(['video.changeStatus']),
   video_id: z.string().uuid(),
-  folder_id: z.string().uuid().optional(),
   status: z.enum(['DRAFT', 'CONVERTING', 'CONVERTED', 'FAILED']),
   video_external_id: z.string().uuid(),
 })
@@ -20,13 +19,11 @@ export async function POST(request: Request) {
       },
     })
 
-    if (video.externalProviderId) {
+    if (!video || video.externalProviderId) {
       /**
-       * TODO: It would be cool to store the external provider status instead
-       * of just storing the external ID.
-       *
        * Here we return a success response as the webhook can be called with
-       * videos that were not stored on jupiter.
+       * videos that were not stored on jupiter or the video could already had
+       * been updated.
        */
       return new Response()
     }

src/app/api/webhooks/upload-to-external-provider/route.ts

@@ -2,19 +2,21 @@ import { r2 } from '@/lib/cloudflare-r2'
 import { prisma } from '@/lib/prisma'
 import { GetObjectCommand } from '@aws-sdk/client-s3'
 import { NextResponse } from 'next/server'
-import FormData from 'form-data'
 import axios from 'axios'
 import { z } from 'zod'
 import { env } from '@/env'
+import { validateQStashSignature } from '@/lib/qstash'
 
 const createTranscriptionBodySchema = z.object({
   videoId: z.string().uuid(),
 })
 
 export async function POST(request: Request) {
-  const { videoId } = createTranscriptionBodySchema.parse(await request.json())
-
   try {
+    const { bodyAsJSON } = await validateQStashSignature({ request })
+
+    const { videoId } = createTranscriptionBodySchema.parse(bodyAsJSON)
+
     const video = await prisma.video.findUniqueOrThrow({
       where: {
         id: videoId,
@@ -50,19 +52,6 @@ export async function POST(request: Request) {
       return
     }
 
-    // const formData = new FormData()
-
-    // formData.append('file', videoFile.Body, {
-    //   contentType: videoFile.ContentType,
-    //   knownLength: videoFile.ContentLength,
-    //   filename: video.audioStorageKey,
-    // })
-
-    // formData.append('model', 'whisper-1')
-    // formData.append('response_format', 'json')
-    // formData.append('temperature', '0')
-    // formData.append('language', 'pt')
-
     const response = await axios.post(
       'https://uploader-us01.pandavideo.com.br/files',
       videoFile.Body,
@@ -81,7 +70,12 @@ export async function POST(request: Request) {
     )
 
     return NextResponse.json({ data: response.data })
-  } catch (err) {
-    console.log(err)
+  } catch (err: any) {
+    console.error(err)
+
+    return NextResponse.json(
+      { message: 'Error uploading video.', error: err?.message || '' },
+      { status: 401 },
+    )
   }
 }

src/env.ts

@@ -26,6 +26,7 @@ export const env = createEnv({
     NEXTAUTH_SECRET: z.string().min(1),
     GOOGLE_CLIENT_ID: z.string().min(1),
     GOOGLE_CLIENT_SECRET: z.string().min(1),
+    QSTASH_TOKEN: z.string().refine(requiredOnEnv('production')),
     QSTASH_CURRENT_SIGNING_KEY: z.string().refine(requiredOnEnv('production')),
     QSTASH_NEXT_SIGNING_KEY: z.string().refine(requiredOnEnv('production')),
   },

src/lib/qstash.ts

@@ -0,0 +1,73 @@
+import { env } from '@/env'
+import { Receiver, Client } from '@upstash/qstash/nodejs'
+
+const qstash = new Client({
+  token: env.QSTASH_TOKEN,
+})
+
+export async function publishMessage<T = any>({
+  topic,
+  body,
+  runInDev = false,
+}: {
+  topic: string
+  body: T
+  runInDev?: boolean
+}) {
+  if (env.NODE_ENV === 'development' && runInDev === false) {
+    console.log(`[Skipped] Publish to "${topic}: ${JSON.stringify(body)}"`)
+
+    return
+  }
+
+  await qstash.publishJSON({
+    topic,
+    contentBasedDeduplication: true,
+    body,
+  })
+}
+
+export async function validateQStashSignature({
+  request,
+  runInDev = false,
+}: {
+  request: Request
+  runInDev?: boolean
+}) {
+  const requestBodyAsText = await request.text()
+
+  if (env.NODE_ENV === 'development' && runInDev === false) {
+    return {
+      bodyAsJSON: JSON.parse(requestBodyAsText),
+    }
+  }
+
+  const signature = request.headers.get('upstash-signature')
+
+  const receiver = new Receiver({
+    currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
+    nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
+  })
+
+  if (!signature) {
+    throw new Error('QStash signature not found.')
+  }
+
+  const isValid = await receiver
+    .verify({
+      signature,
+      body: requestBodyAsText,
+    })
+    .catch((err) => {
+      console.error(err)
+      return false
+    })
+
+  if (!isValid) {
+    throw new Error('QStash signature is invalid.')
+  }
+
+  return {
+    bodyAsJSON: JSON.parse(requestBodyAsText),
+  }
+}