Ffmpeg processing offload (#720)

Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Cursor Agent <cursoragent@cursor.com>

Author: kentcdodds

.env.example

@@ -147,6 +147,36 @@ R2_SECRET_ACCESS_KEY=MOCKR2SECRETACCESSKEY
 # Mocked: yes (MSW intercepts the S3-compatible API) when MOCKS=true and the
 # access key id opts in (starts with `MOCK`).
 CALL_KENT_R2_BUCKET=MOCK_CALL_KENT_R2_BUCKET
+# Feature: /calls/admin (offloaded FFmpeg audio generation from the app)
+# Required (Fly app env):
+# - Cloudflare Queue identifier for `kcd-call-kent-audio`.
+# - Get it from Cloudflare dashboard -> Queues -> queue details.
+CALL_KENT_AUDIO_CF_QUEUE_ID=replace-with-cloudflare-queue-id
+# Optional (Fly app env; defaults to this value)
+CALL_KENT_AUDIO_CF_API_BASE_URL=https://api.cloudflare.com/client/v4
+# Required (Fly app env):
+# - HMAC secret used to verify worker/container callback signatures.
+# - Must exactly match Cloudflare Worker's CALL_KENT_AUDIO_CALLBACK_SECRET.
+# - Generate: openssl rand -hex 32
+# - Set directly in Fly secrets as CALL_KENT_AUDIO_PROCESSOR_CALLBACK_SECRET.
+CALL_KENT_AUDIO_PROCESSOR_CALLBACK_SECRET=replace-with-openssl-random-hex
+
+# Feature: call-kent-audio-worker runtime (Cloudflare Worker vars; not read by app)
+# Set these in call-kent-audio-worker/wrangler.jsonc (vars) or Cloudflare Worker
+# settings. Keep secrets out of plain-text vars.
+# - URL for the FFmpeg container service endpoint.
+CALL_KENT_AUDIO_CONTAINER_URL=https://replace.example.com
+# - Shared bearer token between worker and container.
+# - Generate: openssl rand -hex 32
+# - Store as a Cloudflare Worker secret named CALL_KENT_AUDIO_CONTAINER_TOKEN.
+CALL_KENT_AUDIO_CONTAINER_TOKEN=replace-with-openssl-random-hex
+# - Public callback endpoint on the app.
+CALL_KENT_AUDIO_CALLBACK_URL=https://kentcdodds.com/resources/calls/episode-audio-callback
+# - Shared callback HMAC secret (same value as app's
+#   CALL_KENT_AUDIO_PROCESSOR_CALLBACK_SECRET).
+# - Generate: openssl rand -hex 32
+# - Store as a Cloudflare Worker secret named CALL_KENT_AUDIO_CALLBACK_SECRET.
+CALL_KENT_AUDIO_CALLBACK_SECRET=replace-with-openssl-random-hex
 # Ignore list object key (defaults to manifests/ignore-list.json)
 SEMANTIC_SEARCH_IGNORE_LIST_KEY=manifests/ignore-list.json
 

.github/workflows/deploy-call-kent-audio-worker.yml

@@ -0,0 +1,51 @@
+name: ☁️ Deploy Call Kent Audio Worker
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - call-kent-audio-worker/**
+      - .github/workflows/deploy-call-kent-audio-worker.yml
+  workflow_dispatch:
+
+concurrency:
+  group: deploy-call-kent-audio-worker-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  deploy:
+    name: 🚀 Deploy Call Kent Audio Worker
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - name: ⬇️ Checkout repo
+        uses: actions/checkout@v6
+
+      - name: ⎔ Setup node
+        uses: actions/setup-node@v6
+        with:
+          node-version: 24
+          cache: npm
+          cache-dependency-path: call-kent-audio-worker/package-lock.json
+
+      - name: 📥 Install audio worker deps
+        run: npm ci
+        working-directory: call-kent-audio-worker
+
+      - name: 🔬 Lint audio worker
+        run: npm run lint
+        working-directory: call-kent-audio-worker
+
+      - name: ʦ Typecheck audio worker
+        run: npx tsc --noEmit
+        working-directory: call-kent-audio-worker
+
+      - name: ☁️ Deploy via Wrangler
+        uses: cloudflare/wrangler-action@v3
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          workingDirectory: call-kent-audio-worker
+          command: deploy

app/routes/resources/calls/__tests__/episode-audio-callback.test.ts

@@ -0,0 +1,66 @@
+import { expect, test, vi } from 'vitest'
+
+vi.mock('#app/utils/call-kent-audio-processor-callback.server.ts', () => ({
+	handleCallKentAudioProcessorEvent: vi.fn(),
+	parseCallKentAudioProcessorEvent: vi.fn(),
+	verifyCallKentAudioProcessorCallbackSignature: vi.fn(),
+}))
+
+import {
+	handleCallKentAudioProcessorEvent,
+	parseCallKentAudioProcessorEvent,
+	verifyCallKentAudioProcessorCallbackSignature,
+} from '#app/utils/call-kent-audio-processor-callback.server.ts'
+import { action } from '../episode-audio-callback.ts'
+
+test('episode-audio-callback rejects unsigned cloudflare callback', async () => {
+	vi.clearAllMocks()
+	process.env.CALL_KENT_AUDIO_PROCESSOR_CALLBACK_SECRET = 'callback-secret'
+	const request = new Request(
+		'http://localhost/resources/calls/episode-audio-callback',
+		{
+			method: 'POST',
+			headers: { 'Content-Type': 'application/json' },
+			body: JSON.stringify({
+				type: 'audio_generation_started',
+				draftId: 'draft-1',
+			}),
+		},
+	)
+	const response = await action({ request })
+	expect(response.status).toBe(401)
+	expect(verifyCallKentAudioProcessorCallbackSignature).not.toHaveBeenCalled()
+})
+
+test('episode-audio-callback validates, parses, and handles callback payload', async () => {
+	vi.clearAllMocks()
+	process.env.CALL_KENT_AUDIO_PROCESSOR_CALLBACK_SECRET = 'callback-secret'
+	vi.mocked(verifyCallKentAudioProcessorCallbackSignature).mockReturnValue(true)
+	vi.mocked(parseCallKentAudioProcessorEvent).mockReturnValue({
+		type: 'audio_generation_started',
+		draftId: 'draft-1',
+	})
+	const request = new Request(
+		'http://localhost/resources/calls/episode-audio-callback',
+		{
+			method: 'POST',
+			headers: {
+				'Content-Type': 'application/json',
+				'x-call-kent-audio-timestamp': '1710000000',
+				'x-call-kent-audio-signature': 'abcd',
+			},
+			body: JSON.stringify({
+				type: 'audio_generation_started',
+				draftId: 'draft-1',
+			}),
+		},
+	)
+	const response = await action({ request })
+	expect(response.status).toBe(200)
+	expect(verifyCallKentAudioProcessorCallbackSignature).toHaveBeenCalled()
+	expect(parseCallKentAudioProcessorEvent).toHaveBeenCalled()
+	expect(handleCallKentAudioProcessorEvent).toHaveBeenCalledWith({
+		type: 'audio_generation_started',
+		draftId: 'draft-1',
+	})
+})

app/routes/resources/calls/episode-audio-callback.ts

@@ -0,0 +1,45 @@
+import {
+	handleCallKentAudioProcessorEvent,
+	parseCallKentAudioProcessorEvent,
+	verifyCallKentAudioProcessorCallbackSignature,
+} from '#app/utils/call-kent-audio-processor-callback.server.ts'
+
+const callbackTimestampHeader = 'x-call-kent-audio-timestamp'
+const callbackSignatureHeader = 'x-call-kent-audio-signature'
+
+export async function action({ request }: { request: Request }) {
+	if (request.method !== 'POST') {
+		return new Response('Method not allowed', { status: 405 })
+	}
+	const rawBody = await request.text()
+	const timestamp = request.headers.get(callbackTimestampHeader)
+	const signature = request.headers.get(callbackSignatureHeader)
+	if (!timestamp || !signature) {
+		return new Response('Missing callback signature', { status: 401 })
+	}
+	const isValid = verifyCallKentAudioProcessorCallbackSignature({
+		timestamp,
+		signature,
+		rawBody,
+	})
+	if (!isValid) {
+		return new Response('Invalid callback signature', { status: 401 })
+	}
+
+	let payload: unknown
+	try {
+		payload = rawBody ? JSON.parse(rawBody) : null
+	} catch {
+		return new Response('Invalid JSON body', { status: 400 })
+	}
+
+	let event
+	try {
+		event = parseCallKentAudioProcessorEvent(payload)
+	} catch {
+		return new Response('Invalid callback payload', { status: 400 })
+	}
+
+	await handleCallKentAudioProcessorEvent(event)
+	return Response.json({ ok: true })
+}

app/routes/resources/calls/save.tsx

@@ -5,10 +5,12 @@ import { type RecordingFormData } from '#app/components/calls/recording-form.tsx
 import {
 	deleteAudioObject,
 	getAudioBuffer,
+	parseBase64DataUrl,
 	putCallAudioFromDataUrl,
+	putEpisodeDraftResponseAudioFromBuffer,
 } from '#app/utils/call-kent-audio-storage.server.ts'
 import { startCallKentCallerTranscriptProcessing } from '#app/utils/call-kent-caller-transcript.server.ts'
-import { startCallKentEpisodeDraftProcessing } from '#app/utils/call-kent-episode-draft.server.ts'
+import { requestCallKentEpisodeAudioGeneration } from '#app/utils/call-kent-audio-processor.server.ts'
 import { getPublishedCallKentEpisodeEmail } from '#app/utils/call-kent-published-email.ts'
 import {
 	getErrorForAudio,
@@ -355,9 +357,13 @@ async function publishCall({
 		}
 
 		// Best-effort cleanup of stored audio blobs after publish.
-		const keysToDelete = [call.audioKey, draft.episodeAudioKey].filter(
-			(k): k is string => typeof k === 'string' && k.length > 0,
-		)
+		const keysToDelete = [
+			call.audioKey,
+			draft.episodeAudioKey,
+			draft.responseAudioKey,
+			draft.callerSegmentAudioKey,
+			draft.responseSegmentAudioKey,
+		].filter((k): k is string => typeof k === 'string' && k.length > 0)
 		await Promise.all(
 			keysToDelete.map(async (key) =>
 				deleteAudioObject({ key }).catch(() => {}),
@@ -434,11 +440,24 @@ async function createEpisodeDraft({
 	// If we're replacing a draft, clean up the old stored audio blob.
 	const existingDraft = await prisma.callKentEpisodeDraft.findFirst({
 		where: { callId },
-		select: { episodeAudioKey: true },
+		select: {
+			episodeAudioKey: true,
+			responseAudioKey: true,
+			callerSegmentAudioKey: true,
+			responseSegmentAudioKey: true,
+		},
 	})
-	if (existingDraft?.episodeAudioKey) {
-		await deleteAudioObject({ key: existingDraft.episodeAudioKey }).catch(
-			() => {},
+	const existingKeysToDelete = [
+		existingDraft?.episodeAudioKey,
+		existingDraft?.responseAudioKey,
+		existingDraft?.callerSegmentAudioKey,
+		existingDraft?.responseSegmentAudioKey,
+	].filter((key): key is string => typeof key === 'string' && key.length > 0)
+	if (existingKeysToDelete.length) {
+		await Promise.all(
+			existingKeysToDelete.map(async (key) =>
+				deleteAudioObject({ key }).catch(() => {}),
+			),
 		)
 	}
 
@@ -452,9 +471,47 @@ async function createEpisodeDraft({
 		}),
 	])
 
-	void startCallKentEpisodeDraftProcessing(draft.id, {
-		responseBase64: responseAudio!,
-	})
+	try {
+		if (!call.audioKey) {
+			throw new Error('Call audio is missing (audioKey is null).')
+		}
+		const parsedResponseAudio = parseBase64DataUrl(responseAudio!)
+		const storedResponseAudio = await putEpisodeDraftResponseAudioFromBuffer({
+			draftId: draft.id,
+			audio: parsedResponseAudio.buffer,
+			contentType: parsedResponseAudio.contentType,
+		})
+		await prisma.callKentEpisodeDraft.updateMany({
+			where: { id: draft.id, status: 'PROCESSING' },
+			data: {
+				responseAudioKey: storedResponseAudio.key,
+				responseAudioContentType: storedResponseAudio.contentType,
+				responseAudioSize: storedResponseAudio.size,
+				step: 'GENERATING_AUDIO',
+				errorMessage: null,
+			},
+		})
+		await requestCallKentEpisodeAudioGeneration({
+			draftId: draft.id,
+			callAudioKey: call.audioKey,
+			responseAudioKey: storedResponseAudio.key,
+		})
+	} catch (error: unknown) {
+		await prisma.callKentEpisodeDraft.updateMany({
+			where: { id: draft.id, status: 'PROCESSING' },
+			data: {
+				status: 'ERROR',
+				errorMessage: getErrorMessage(error),
+				step: 'DONE',
+			},
+		})
+		const searchParams = new URLSearchParams()
+		searchParams.set(
+			'error',
+			`Unable to start draft audio generation: ${getErrorMessage(error)}`,
+		)
+		return redirect(`/calls/admin/${callId}?${searchParams.toString()}`)
+	}
 
 	return redirect(`/calls/admin/${callId}`)
 }
@@ -547,14 +604,26 @@ async function undoEpisodeDraft({
 
 	const drafts = await prisma.callKentEpisodeDraft.findMany({
 		where: { callId },
-		select: { episodeAudioKey: true },
+		select: {
+			episodeAudioKey: true,
+			responseAudioKey: true,
+			callerSegmentAudioKey: true,
+			responseSegmentAudioKey: true,
+		},
 	})
-	if (drafts.some((d) => d.episodeAudioKey)) {
+	const keysToDelete = drafts
+		.flatMap((draft) => [
+			draft.episodeAudioKey,
+			draft.responseAudioKey,
+			draft.callerSegmentAudioKey,
+			draft.responseSegmentAudioKey,
+		])
+		.filter((k): k is string => typeof k === 'string' && k.length > 0)
+	if (keysToDelete.length) {
 		await Promise.all(
-			drafts
-				.map((d) => d.episodeAudioKey)
-				.filter((k): k is string => typeof k === 'string' && k.length > 0)
-				.map(async (key) => deleteAudioObject({ key }).catch(() => {})),
+			keysToDelete.map(async (key) =>
+				deleteAudioObject({ key }).catch(() => {}),
+			),
 		)
 	}
 	await prisma.callKentEpisodeDraft.deleteMany({ where: { callId } })
@@ -652,7 +721,14 @@ async function deleteCall({
 		select: {
 			id: true,
 			audioKey: true,
-			episodeDraft: { select: { episodeAudioKey: true } },
+			episodeDraft: {
+				select: {
+					episodeAudioKey: true,
+					responseAudioKey: true,
+					callerSegmentAudioKey: true,
+					responseSegmentAudioKey: true,
+				},
+			},
 		},
 	})
 	if (!call) {
@@ -663,6 +739,9 @@ async function deleteCall({
 	const keysToDelete = [
 		call.audioKey,
 		call.episodeDraft?.episodeAudioKey,
+		call.episodeDraft?.responseAudioKey,
+		call.episodeDraft?.callerSegmentAudioKey,
+		call.episodeDraft?.responseSegmentAudioKey,
 	].filter((k): k is string => typeof k === 'string' && k.length > 0)
 	if (keysToDelete.length) {
 		await Promise.all(