From 7d824047cb722f25f532082af799e55927605bb9 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:34:09 +0000 Subject: [PATCH 01/11] refactor(web): share live transcription storage and claim helpers --- apps/web/lib/live-transcribe-runner.ts | 500 +++++++++++++++++++++++++ apps/web/workflows/live-transcribe.ts | 458 ++-------------------- 2 files changed, 540 insertions(+), 418 deletions(-) create mode 100644 apps/web/lib/live-transcribe-runner.ts diff --git a/apps/web/lib/live-transcribe-runner.ts b/apps/web/lib/live-transcribe-runner.ts new file mode 100644 index 0000000000..a725322912 --- /dev/null +++ b/apps/web/lib/live-transcribe-runner.ts @@ -0,0 +1,500 @@ +import { db } from "@cap/database"; +import { users, videoEdits, videos } from "@cap/database/schema"; +import { serverEnv } from "@cap/env"; +import { Storage } from "@cap/web-backend/src/Storage/index"; +import { type User, Video } from "@cap/web-domain"; +import { and, eq, isNull, type SQL, sql } from "drizzle-orm"; +import { type Effect, Either, Option, Schema } from "effect"; +import { isAiGenerationEnabledForUser } from "@/lib/ai-generation-entitlement"; +import { ASSEMBLYAI_SPEECH_MODELS } from "@/lib/assemblyai"; +import { + getEditTranscriptObjectKey, + parseEditTranscript, + serializeEditTranscript, +} from "@/lib/edit-transcript"; +import { encryptEditTranscriptObject } from "@/lib/edit-transcript-storage"; +import { startAiGeneration } from "@/lib/generate-ai"; +import { + applyChunkToLiveTranscript, + canPromoteLiveTranscript, + createEmptyLiveTranscript, + getLiveTranscriptObjectKey, + type LiveChunkWordInput, + type LiveTranscriptArtifact, + type LiveTranscriptState, + liveTranscriptToEditTranscript, + offsetChunkWords, + parseLiveTranscript, +} from "@/lib/live-transcribe-core"; +import { transcribeVideo } from "@/lib/transcribe"; +import { decodeStorageVideo } from "@/lib/video-storage"; +import { runWorkflowPromise } from "@/lib/workflow-runtime"; + +/** + * Storage and claim operations shared by the two live transcription runners: + * the liveTranscribe workflow and the media server session that reports + * through /api/webhooks/media-server/live-transcribe. + */ + +export type LiveBucket = Effect.Effect.Success< + ReturnType +>[0]; + +const CANONICAL_DONE_STATUSES = new Set(["COMPLETE", "SKIPPED", "NO_AUDIO"]); + +export function isCanonicalTranscriptDone(status: string | null | undefined) { + return status !== null && status !== undefined + ? CANONICAL_DONE_STATUSES.has(status) + : false; +} + +/** Only the session named in the claim may write; a replaced runner's late + * reports must not touch the artifact. */ +export function liveSessionMatches(session: string): SQL { + return sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.session')) = ${session}`; +} + +export async function readLiveTranscript( + bucket: LiveBucket, + ownerId: string, + videoId: string, +): Promise { + const existing = await bucket + .getObject(getLiveTranscriptObjectKey(ownerId, videoId)) + .pipe(runWorkflowPromise) + .catch(() => Option.none()); + return Option.isSome(existing) ? parseLiveTranscript(existing.value) : null; +} + +/** + * Record that the audio up to `lastAudioSegmentIndex` was skipped. Must land + * before the cursor moves past it, or a later promotion could ship a + * transcript that silently misses speech. Throws when it can't be recorded. + */ +export async function writeLiveChunkGap(options: { + bucket: LiveBucket; + ownerId: string; + videoId: string; + lastAudioSegmentIndex: number; + endMs: number; + session?: string; +}): Promise { + const { bucket, ownerId, videoId } = options; + const artifact = + (await readLiveTranscript(bucket, ownerId, videoId)) ?? + createEmptyLiveTranscript(new Date().toISOString()); + const updated: LiveTranscriptArtifact = { + ...artifact, + hasGaps: true, + lastAudioSegmentIndex: Math.max( + artifact.lastAudioSegmentIndex, + options.lastAudioSegmentIndex, + ), + transcribedDurationMs: Math.max( + artifact.transcribedDurationMs, + options.endMs, + ), + updatedAt: new Date().toISOString(), + }; + await bucket + .putObject( + getLiveTranscriptObjectKey(ownerId, videoId), + JSON.stringify(updated), + { contentType: "application/json" }, + ) + .pipe(runWorkflowPromise); + await touchLiveClaim(videoId, options.session); + return updated; +} + +/** + * Merge one transcribed chunk into `transcription.live.json`. Returns null + * without writing when the canonical transcript finished meanwhile (the write + * would re-create the artifact it just deleted, an orphan billed forever) or, + * for a session runner, when the claim moved to another session. + */ +export async function writeLiveChunk(options: { + bucket: LiveBucket; + ownerId: string; + videoId: string; + chunk: { + startMs: number; + durationMs: number; + contextMs: number; + lastAudioSegmentIndex: number; + }; + words: readonly LiveChunkWordInput[] | null | undefined; + languageCode: string | null; + session?: string; +}): Promise { + const { bucket, ownerId, videoId, chunk, session } = options; + const words = offsetChunkWords( + options.words, + chunk.startMs, + chunk.durationMs, + chunk.contextMs, + ); + const artifact = + (await readLiveTranscript(bucket, ownerId, videoId)) ?? + createEmptyLiveTranscript(new Date().toISOString()); + const updated = applyChunkToLiveTranscript(artifact, { + ...chunk, + words, + languageCode: options.languageCode, + nowIso: new Date().toISOString(), + }); + + const [current] = await db() + .select({ + transcriptionStatus: videos.transcriptionStatus, + session: sql< + string | null + >`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.session'))`, + }) + .from(videos) + .where(eq(videos.id, videoId as Video.VideoId)); + if ( + isCanonicalTranscriptDone(current?.transcriptionStatus) || + (session !== undefined && current?.session !== session) + ) { + return null; + } + + await bucket + .putObject( + getLiveTranscriptObjectKey(ownerId, videoId), + JSON.stringify(updated), + { contentType: "application/json" }, + ) + .pipe(runWorkflowPromise); + + await touchLiveClaim(videoId, session); + return updated; +} + +export function liveContextSeconds() { + const value = Number(serverEnv().LIVE_TRANSCRIBE_CONTEXT_SECONDS ?? 0); + return Number.isFinite(value) ? Math.min(Math.max(value, 0), 60) : 0; +} + +/** + * Re-stamp the live claim's freshness so other transcription triggers keep + * deferring to the running live transcription (lib/transcribe.ts checks the + * stamp age). A no-op once the claim was cleared, finished, or handed to + * another session. + */ +export async function touchLiveClaim( + videoId: string, + session?: string, +): Promise { + try { + await db() + .update(videos) + .set({ + metadata: sql`JSON_SET(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript.updatedAt', ${new Date().toISOString()})`, + updatedAt: sql`${videos.updatedAt}`, + }) + .where( + and( + eq(videos.id, videoId as Video.VideoId), + sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.status')) = 'active'`, + session === undefined ? undefined : liveSessionMatches(session), + ), + ); + } catch (error) { + console.warn( + `[liveTranscribe] Failed to stamp live claim for ${videoId}`, + error, + ); + } +} + +type PromotionResult = { promoted: boolean; reason?: string }; + +/** + * Promote the completed live transcript to the canonical transcript: write + * transcription.vtt + the encrypted edit transcript from the accumulated + * words, claim transcriptionStatus, queue AI generation, and clean up the + * provisional artifact. Never throws. Any failure releases the claim (if + * held) and queues the normal full-pass so the video still transcribes. + */ +export async function promoteLiveTranscript( + videoId: string, + userId: string, +): Promise { + try { + const [video] = await db() + .select() + .from(videos) + .where(eq(videos.id, videoId as Video.VideoId)); + + if (!video || video.ownerId !== userId) { + return { promoted: false, reason: "video not found" }; + } + if (video.transcriptionStatus !== null) { + return { + promoted: false, + reason: `canonical status is ${video.transcriptionStatus}`, + }; + } + + const [bucket] = await Storage.getAccessForVideo( + decodeStorageVideo(video), + ).pipe(runWorkflowPromise); + const segSource = new Video.SegmentsSource({ + videoId, + ownerId: video.ownerId, + }); + + const manifestContent = await bucket + .getObject(segSource.getManifestKey()) + .pipe(runWorkflowPromise); + const manifestJson = Option.getOrNull(manifestContent); + if (!manifestJson) { + return { promoted: false, reason: "manifest missing" }; + } + const decoded = Schema.decodeUnknownEither(Video.SegmentManifest)( + JSON.parse(manifestJson), + ); + if (Either.isLeft(decoded)) { + return { promoted: false, reason: "manifest invalid" }; + } + + const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); + const existing = await bucket + .getObject(artifactKey) + .pipe(runWorkflowPromise); + const artifact = Option.isSome(existing) + ? parseLiveTranscript(existing.value) + : null; + if (!artifact) { + return { promoted: false, reason: "live artifact missing" }; + } + + const eligible = canPromoteLiveTranscript(artifact, decoded.right); + if (!eligible.ok) { + await queueFullPassFallback(videoId, userId); + return { promoted: false, reason: eligible.reason }; + } + + const editTranscript = serializeEditTranscript( + liveTranscriptToEditTranscript(artifact, ASSEMBLYAI_SPEECH_MODELS[0]), + ); + if (!parseEditTranscript(editTranscript)) { + await queueFullPassFallback(videoId, userId); + return { + promoted: false, + reason: "live words are not a valid transcript", + }; + } + + const claim = await db() + .update(videos) + .set({ transcriptionStatus: "PROCESSING" }) + .where( + and( + eq(videos.id, videoId as Video.VideoId), + isNull(videos.transcriptionStatus), + ), + ); + const affectedRows = Array.isArray(claim) + ? ((claim[0] as { affectedRows?: number } | undefined)?.affectedRows ?? 0) + : ((claim as { affectedRows?: number } | undefined)?.affectedRows ?? 0); + if (affectedRows === 0) { + return { promoted: false, reason: "canonical claim held elsewhere" }; + } + + try { + await bucket + .putObject( + `${video.ownerId}/${videoId}/transcription.vtt`, + artifact.vtt, + { contentType: "text/vtt" }, + ) + .pipe(runWorkflowPromise); + + // Same rule as the full-pass save: the word transcript must describe + // the original media, so edited videos keep their own. + const [edit] = await db() + .select({ videoId: videoEdits.videoId }) + .from(videoEdits) + .where(eq(videoEdits.videoId, videoId as Video.VideoId)); + if (!edit) { + await bucket + .putObject( + getEditTranscriptObjectKey(video.ownerId, videoId), + encryptEditTranscriptObject(editTranscript, video.ownerId, videoId), + { contentType: "application/octet-stream" }, + ) + .pipe(runWorkflowPromise); + } + + await db() + .update(videos) + .set({ transcriptionStatus: "COMPLETE" }) + .where( + and( + eq(videos.id, videoId as Video.VideoId), + eq(videos.transcriptionStatus, "PROCESSING"), + ), + ); + } catch (error) { + // Release the claim so the fallback can transcribe from scratch. + await db() + .update(videos) + .set({ transcriptionStatus: null }) + .where( + and( + eq(videos.id, videoId as Video.VideoId), + eq(videos.transcriptionStatus, "PROCESSING"), + ), + ); + await queueFullPassFallback(videoId, userId); + return { + promoted: false, + reason: error instanceof Error ? error.message : String(error), + }; + } + + // Post-COMPLETE housekeeping: never fatal, never releases the claim. + try { + await bucket.deleteObject(artifactKey).pipe(runWorkflowPromise); + } catch (error) { + console.warn( + `[liveTranscribe] Failed to delete promoted artifact for ${videoId}`, + error, + ); + } + try { + await db() + .update(videos) + .set({ + metadata: sql`JSON_REMOVE(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript')`, + updatedAt: sql`${videos.updatedAt}`, + }) + .where(eq(videos.id, videoId as Video.VideoId)); + } catch (error) { + console.warn( + `[liveTranscribe] Failed to clear live flag for ${videoId}`, + error, + ); + } + try { + const [owner] = await db() + .select({ + email: users.email, + stripeSubscriptionStatus: users.stripeSubscriptionStatus, + thirdPartyStripeSubscriptionId: users.thirdPartyStripeSubscriptionId, + }) + .from(users) + .where(eq(users.id, userId as User.UserId)); + if (isAiGenerationEnabledForUser(owner)) { + await startAiGeneration(videoId as Video.VideoId, userId); + } + } catch (error) { + console.warn( + `[liveTranscribe] Failed to queue AI generation for ${videoId}`, + error, + ); + } + + return { promoted: true }; + } catch (error) { + return { + promoted: false, + reason: error instanceof Error ? error.message : String(error), + }; + } +} + +/** + * When promotion can't happen, immediately queue the normal full + * transcription (early-from-segments) instead of waiting for the post-mux + * queue, so a declined promotion costs seconds, not a minute. + */ +async function queueFullPassFallback( + videoId: string, + userId: string, +): Promise { + try { + const [owner] = await db() + .select({ + email: users.email, + stripeSubscriptionStatus: users.stripeSubscriptionStatus, + thirdPartyStripeSubscriptionId: users.thirdPartyStripeSubscriptionId, + }) + .from(users) + .where(eq(users.id, userId as User.UserId)); + + await transcribeVideo( + videoId as Video.VideoId, + userId, + isAiGenerationEnabledForUser(owner), + { earlyFromSegments: true }, + ); + } catch (error) { + console.warn( + `[liveTranscribe] Full-pass fallback queue failed for ${videoId}`, + error, + ); + } +} + +export async function finishLiveTranscription( + videoId: string, + userId: string, + state: LiveTranscriptState, +): Promise { + try { + await db() + .update(videos) + .set({ + metadata: sql`JSON_SET(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript.status', ${state}, '$.liveTranscript.updatedAt', ${new Date().toISOString()})`, + updatedAt: sql`${videos.updatedAt}`, + }) + .where(eq(videos.id, videoId as Video.VideoId)); + } catch (error) { + console.warn( + `[liveTranscribe] Failed to update metadata for ${videoId}`, + error, + ); + } + + try { + const [video] = await db() + .select() + .from(videos) + .where(eq(videos.id, videoId as Video.VideoId)); + if (!video || video.ownerId !== userId) return; + + const [bucket] = await Storage.getAccessForVideo( + decodeStorageVideo(video), + ).pipe(runWorkflowPromise); + const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); + const existing = await bucket + .getObject(artifactKey) + .pipe(runWorkflowPromise) + .catch(() => Option.none()); + if (Option.isNone(existing)) return; + + const artifact = parseLiveTranscript(existing.value); + if (!artifact || artifact.state === state) return; + + await bucket + .putObject( + artifactKey, + JSON.stringify({ + ...artifact, + state, + updatedAt: new Date().toISOString(), + }), + { contentType: "application/json" }, + ) + .pipe(runWorkflowPromise); + } catch (error) { + console.warn( + `[liveTranscribe] Failed to finalize artifact for ${videoId}`, + error, + ); + } +} diff --git a/apps/web/workflows/live-transcribe.ts b/apps/web/workflows/live-transcribe.ts index c419221d1b..a37ec08ab6 100644 --- a/apps/web/workflows/live-transcribe.ts +++ b/apps/web/workflows/live-transcribe.ts @@ -1,45 +1,36 @@ import { db } from "@cap/database"; -import { organizations, users, videoEdits, videos } from "@cap/database/schema"; +import { organizations, videos } from "@cap/database/schema"; import { serverEnv } from "@cap/env"; import { Storage } from "@cap/web-backend/src/Storage/index"; import { type AiGenerationLanguage, parseAiGenerationLanguage, - type User, Video, } from "@cap/web-domain"; -import { and, eq, isNull, sql } from "drizzle-orm"; +import { eq } from "drizzle-orm"; import { Either, Option, Schema } from "effect"; -import { isAiGenerationEnabledForUser } from "@/lib/ai-generation-entitlement"; import { - ASSEMBLYAI_SPEECH_MODELS, ASSEMBLYAI_SUPPORTED_LANGUAGES, createAssemblyAIClient, getAssemblyAITranscriptionOptions, } from "@/lib/assemblyai"; import { - getEditTranscriptObjectKey, - parseEditTranscript, - serializeEditTranscript, -} from "@/lib/edit-transcript"; -import { encryptEditTranscriptObject } from "@/lib/edit-transcript-storage"; -import { startAiGeneration } from "@/lib/generate-ai"; -import { - applyChunkToLiveTranscript, - canPromoteLiveTranscript, - createEmptyLiveTranscript, getLiveTranscriptObjectKey, isNoSpokenAudioError, LIVE_TRANSCRIBE, LIVE_TRANSCRIPT_NO_SEGMENTS, type LiveTranscriptState, - liveTranscriptToEditTranscript, - offsetChunkWords, parseLiveTranscript, planNextLiveChunk, } from "@/lib/live-transcribe-core"; +import { + finishLiveTranscription, + liveContextSeconds, + promoteLiveTranscript, + writeLiveChunk, + writeLiveChunkGap, +} from "@/lib/live-transcribe-runner"; import { downloadConcatenatedSegmentsToBuffer } from "@/lib/segments-audio-download"; -import { transcribeVideo } from "@/lib/transcribe"; import { decodeStorageVideo } from "@/lib/video-storage"; import { runWorkflowPromise } from "@/lib/workflow-runtime"; @@ -97,7 +88,7 @@ export async function liveTranscribeWorkflow( const init = await initLiveTranscription(videoId, userId); if (!init.ok) { - await finishLiveTranscription(videoId, userId, "stopped"); + await finishLiveTranscriptionStep(videoId, userId, "stopped"); return { success: true, message: `Live transcription skipped: ${init.reason}`, @@ -178,7 +169,7 @@ export async function liveTranscribeWorkflow( } catch (error) { // A step exhausted its retries; make sure the artifact and metadata // never advertise a live transcription that is no longer running. - await finishLiveTranscription(videoId, userId, "stopped"); + await finishLiveTranscriptionStep(videoId, userId, "stopped"); throw error; } @@ -186,7 +177,7 @@ export async function liveTranscribeWorkflow( // Full gap-free coverage: promote the live transcript to canonical and // skip the duplicate full transcription pass entirely. On any failure // the full-pass fallback is queued so transcription still lands fast. - const promotion = await promoteLiveTranscript(videoId, userId); + const promotion = await promoteLiveTranscriptStep(videoId, userId); if (promotion.promoted) { return { success: true, @@ -198,7 +189,7 @@ export async function liveTranscribeWorkflow( ); } - await finishLiveTranscription( + await finishLiveTranscriptionStep( videoId, userId, outcome === "done" ? "complete" : "stopped", @@ -375,34 +366,13 @@ async function processNextLiveChunk(options: { // The gap MUST be durably recorded before the cursor advances, or a // later promotion could ship a transcript that silently misses speech. try { - const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); - const existing = await bucket - .getObject(artifactKey) - .pipe(runWorkflowPromise) - .catch(() => Option.none()); - const artifact = - (Option.isSome(existing) - ? parseLiveTranscript(existing.value) - : null) ?? createEmptyLiveTranscript(new Date().toISOString()); - await bucket - .putObject( - artifactKey, - JSON.stringify({ - ...artifact, - hasGaps: true, - lastAudioSegmentIndex: Math.max( - artifact.lastAudioSegmentIndex, - chunkEndIndex, - ), - transcribedDurationMs: Math.max( - artifact.transcribedDurationMs, - decision.startMs + decision.durationMs, - ), - updatedAt: new Date().toISOString(), - }), - { contentType: "application/json" }, - ) - .pipe(runWorkflowPromise); + await writeLiveChunkGap({ + bucket, + ownerId: video.ownerId, + videoId, + lastAudioSegmentIndex: chunkEndIndex, + endMs: decision.startMs + decision.durationMs, + }); } catch (error) { return { outcome: "chunk-failed", @@ -412,7 +382,6 @@ async function processNextLiveChunk(options: { }`, }; } - await touchLiveClaim(videoId); return { outcome: "chunk", lastAudioSegmentIndex: chunkEndIndex, @@ -456,60 +425,25 @@ async function processNextLiveChunk(options: { throw new Error(transcript.error ?? "AssemblyAI chunk failed"); } - const words = noSpokenAudio - ? [] - : offsetChunkWords( - transcript.words, - decision.startMs, - decision.durationMs, - decision.contextMs, - ); - - const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); - const existing = await bucket - .getObject(artifactKey) - .pipe(runWorkflowPromise) - .catch(() => Option.none()); - const artifact = - (Option.isSome(existing) ? parseLiveTranscript(existing.value) : null) ?? - createEmptyLiveTranscript(new Date().toISOString()); - - const updated = applyChunkToLiveTranscript(artifact, { - startMs: decision.startMs, - durationMs: decision.durationMs, - contextMs: decision.contextMs, - lastAudioSegmentIndex: chunkEndIndex, - words, + // Polling + the AssemblyAI round trip can take minutes, so the write + // re-checks that the canonical pipeline didn't complete meanwhile. + const updated = await writeLiveChunk({ + bucket, + ownerId: video.ownerId, + videoId, + chunk: { + startMs: decision.startMs, + durationMs: decision.durationMs, + contextMs: decision.contextMs, + lastAudioSegmentIndex: chunkEndIndex, + }, + words: noSpokenAudio ? [] : transcript.words, languageCode: typeof transcript.language_code === "string" ? transcript.language_code : null, - nowIso: new Date().toISOString(), }); - - // Polling + the AssemblyAI round trip can take minutes; re-check that the - // canonical pipeline didn't complete meanwhile, or this write would - // re-create the artifact it just deleted (an orphan billed forever). - const [current] = await db() - .select({ transcriptionStatus: videos.transcriptionStatus }) - .from(videos) - .where(eq(videos.id, videoId as Video.VideoId)); - if ( - current?.transcriptionStatus === "COMPLETE" || - current?.transcriptionStatus === "SKIPPED" || - current?.transcriptionStatus === "NO_AUDIO" - ) { - return { outcome: "canonical-done" }; - } - - const body = JSON.stringify(updated); - await bucket - .putObject(artifactKey, body, { contentType: "application/json" }) - .pipe(runWorkflowPromise); - - // Freshness-stamp the claim so other transcription triggers (post-mux - // webhook, share-page retries) keep deferring to this workflow. - await touchLiveClaim(videoId); + if (!updated) return { outcome: "canonical-done" }; // If this chunk completed full coverage of a finished recording, tell // the workflow to promote right away instead of paying another poll @@ -523,7 +457,7 @@ async function processNextLiveChunk(options: { : ({ action: "wait" } as const); console.log( - `[liveTranscribe] ${videoId} chunk ${lastProcessedIndex + 1}..${chunkEndIndex} (${decision.durationMs}ms + ${decision.contextMs}ms context, ${words.length} words)`, + `[liveTranscribe] ${videoId} chunk ${lastProcessedIndex + 1}..${chunkEndIndex} (${decision.durationMs}ms + ${decision.contextMs}ms context, ${updated.words.length} words total)`, ); return { @@ -546,328 +480,16 @@ async function processNextLiveChunk(options: { } } -function liveContextSeconds() { - const value = Number(serverEnv().LIVE_TRANSCRIBE_CONTEXT_SECONDS ?? 0); - return Number.isFinite(value) ? Math.min(Math.max(value, 0), 60) : 0; -} - -/** - * Re-stamp the live claim's freshness so other transcription triggers keep - * deferring to this workflow (lib/transcribe.ts checks the stamp age). A - * no-op once the claim was cleared or finished. - */ -async function touchLiveClaim(videoId: string): Promise { - try { - await db() - .update(videos) - .set({ - metadata: sql`JSON_SET(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript.updatedAt', ${new Date().toISOString()})`, - updatedAt: sql`${videos.updatedAt}`, - }) - .where( - and( - eq(videos.id, videoId as Video.VideoId), - sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.status')) = 'active'`, - ), - ); - } catch (error) { - console.warn( - `[liveTranscribe] Failed to stamp live claim for ${videoId}`, - error, - ); - } -} - -type PromotionResult = { promoted: boolean; reason?: string }; - -/** - * Promote the completed live transcript to the canonical transcript: write - * transcription.vtt + the encrypted edit transcript from the accumulated - * words, claim transcriptionStatus, queue AI generation, and clean up the - * provisional artifact. Never throws. Any failure releases the claim (if - * held) and queues the normal full-pass so the video still transcribes. - */ -async function promoteLiveTranscript( - videoId: string, - userId: string, -): Promise { +async function promoteLiveTranscriptStep(videoId: string, userId: string) { "use step"; - - try { - const [video] = await db() - .select() - .from(videos) - .where(eq(videos.id, videoId as Video.VideoId)); - - if (!video || video.ownerId !== userId) { - return { promoted: false, reason: "video not found" }; - } - if (video.transcriptionStatus !== null) { - return { - promoted: false, - reason: `canonical status is ${video.transcriptionStatus}`, - }; - } - - const [bucket] = await Storage.getAccessForVideo( - decodeStorageVideo(video), - ).pipe(runWorkflowPromise); - const segSource = new Video.SegmentsSource({ - videoId, - ownerId: video.ownerId, - }); - - const manifestContent = await bucket - .getObject(segSource.getManifestKey()) - .pipe(runWorkflowPromise); - const manifestJson = Option.getOrNull(manifestContent); - if (!manifestJson) { - return { promoted: false, reason: "manifest missing" }; - } - const decoded = Schema.decodeUnknownEither(Video.SegmentManifest)( - JSON.parse(manifestJson), - ); - if (Either.isLeft(decoded)) { - return { promoted: false, reason: "manifest invalid" }; - } - - const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); - const existing = await bucket - .getObject(artifactKey) - .pipe(runWorkflowPromise); - const artifact = Option.isSome(existing) - ? parseLiveTranscript(existing.value) - : null; - if (!artifact) { - return { promoted: false, reason: "live artifact missing" }; - } - - const eligible = canPromoteLiveTranscript(artifact, decoded.right); - if (!eligible.ok) { - await queueFullPassFallback(videoId, userId); - return { promoted: false, reason: eligible.reason }; - } - - const editTranscript = serializeEditTranscript( - liveTranscriptToEditTranscript(artifact, ASSEMBLYAI_SPEECH_MODELS[0]), - ); - if (!parseEditTranscript(editTranscript)) { - await queueFullPassFallback(videoId, userId); - return { - promoted: false, - reason: "live words are not a valid transcript", - }; - } - - const claim = await db() - .update(videos) - .set({ transcriptionStatus: "PROCESSING" }) - .where( - and( - eq(videos.id, videoId as Video.VideoId), - isNull(videos.transcriptionStatus), - ), - ); - const affectedRows = Array.isArray(claim) - ? ((claim[0] as { affectedRows?: number } | undefined)?.affectedRows ?? 0) - : ((claim as { affectedRows?: number } | undefined)?.affectedRows ?? 0); - if (affectedRows === 0) { - return { promoted: false, reason: "canonical claim held elsewhere" }; - } - - try { - await bucket - .putObject( - `${video.ownerId}/${videoId}/transcription.vtt`, - artifact.vtt, - { contentType: "text/vtt" }, - ) - .pipe(runWorkflowPromise); - - // Same rule as the full-pass save: the word transcript must describe - // the original media, so edited videos keep their own. - const [edit] = await db() - .select({ videoId: videoEdits.videoId }) - .from(videoEdits) - .where(eq(videoEdits.videoId, videoId as Video.VideoId)); - if (!edit) { - await bucket - .putObject( - getEditTranscriptObjectKey(video.ownerId, videoId), - encryptEditTranscriptObject(editTranscript, video.ownerId, videoId), - { contentType: "application/octet-stream" }, - ) - .pipe(runWorkflowPromise); - } - - await db() - .update(videos) - .set({ transcriptionStatus: "COMPLETE" }) - .where( - and( - eq(videos.id, videoId as Video.VideoId), - eq(videos.transcriptionStatus, "PROCESSING"), - ), - ); - } catch (error) { - // Release the claim so the fallback can transcribe from scratch. - await db() - .update(videos) - .set({ transcriptionStatus: null }) - .where( - and( - eq(videos.id, videoId as Video.VideoId), - eq(videos.transcriptionStatus, "PROCESSING"), - ), - ); - await queueFullPassFallback(videoId, userId); - return { - promoted: false, - reason: error instanceof Error ? error.message : String(error), - }; - } - - // Post-COMPLETE housekeeping: never fatal, never releases the claim. - try { - await bucket.deleteObject(artifactKey).pipe(runWorkflowPromise); - } catch (error) { - console.warn( - `[liveTranscribe] Failed to delete promoted artifact for ${videoId}`, - error, - ); - } - try { - await db() - .update(videos) - .set({ - metadata: sql`JSON_REMOVE(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript')`, - updatedAt: sql`${videos.updatedAt}`, - }) - .where(eq(videos.id, videoId as Video.VideoId)); - } catch (error) { - console.warn( - `[liveTranscribe] Failed to clear live flag for ${videoId}`, - error, - ); - } - try { - const [owner] = await db() - .select({ - email: users.email, - stripeSubscriptionStatus: users.stripeSubscriptionStatus, - thirdPartyStripeSubscriptionId: users.thirdPartyStripeSubscriptionId, - }) - .from(users) - .where(eq(users.id, userId as User.UserId)); - if (isAiGenerationEnabledForUser(owner)) { - await startAiGeneration(videoId as Video.VideoId, userId); - } - } catch (error) { - console.warn( - `[liveTranscribe] Failed to queue AI generation for ${videoId}`, - error, - ); - } - - return { promoted: true }; - } catch (error) { - return { - promoted: false, - reason: error instanceof Error ? error.message : String(error), - }; - } + return promoteLiveTranscript(videoId, userId); } -/** - * When promotion can't happen, immediately queue the normal full - * transcription (early-from-segments) instead of waiting for the post-mux - * queue, so a declined promotion costs seconds, not a minute. - */ -async function queueFullPassFallback( - videoId: string, - userId: string, -): Promise { - try { - const [owner] = await db() - .select({ - email: users.email, - stripeSubscriptionStatus: users.stripeSubscriptionStatus, - thirdPartyStripeSubscriptionId: users.thirdPartyStripeSubscriptionId, - }) - .from(users) - .where(eq(users.id, userId as User.UserId)); - - await transcribeVideo( - videoId as Video.VideoId, - userId, - isAiGenerationEnabledForUser(owner), - { earlyFromSegments: true }, - ); - } catch (error) { - console.warn( - `[liveTranscribe] Full-pass fallback queue failed for ${videoId}`, - error, - ); - } -} - -async function finishLiveTranscription( +async function finishLiveTranscriptionStep( videoId: string, userId: string, state: LiveTranscriptState, -): Promise { +) { "use step"; - - try { - await db() - .update(videos) - .set({ - metadata: sql`JSON_SET(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript.status', ${state}, '$.liveTranscript.updatedAt', ${new Date().toISOString()})`, - updatedAt: sql`${videos.updatedAt}`, - }) - .where(eq(videos.id, videoId as Video.VideoId)); - } catch (error) { - console.warn( - `[liveTranscribe] Failed to update metadata for ${videoId}`, - error, - ); - } - - try { - const [video] = await db() - .select() - .from(videos) - .where(eq(videos.id, videoId as Video.VideoId)); - if (!video || video.ownerId !== userId) return; - - const [bucket] = await Storage.getAccessForVideo( - decodeStorageVideo(video), - ).pipe(runWorkflowPromise); - const artifactKey = getLiveTranscriptObjectKey(video.ownerId, videoId); - const existing = await bucket - .getObject(artifactKey) - .pipe(runWorkflowPromise) - .catch(() => Option.none()); - if (Option.isNone(existing)) return; - - const artifact = parseLiveTranscript(existing.value); - if (!artifact || artifact.state === state) return; - - await bucket - .putObject( - artifactKey, - JSON.stringify({ - ...artifact, - state, - updatedAt: new Date().toISOString(), - }), - { contentType: "application/json" }, - ) - .pipe(runWorkflowPromise); - } catch (error) { - console.warn( - `[liveTranscribe] Failed to finalize artifact for ${videoId}`, - error, - ); - } + await finishLiveTranscription(videoId, userId, state); } From ecb1d0b1ad10682eec61d9bf9fecf5322099e359 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:34:09 +0000 Subject: [PATCH 02/11] feat(web): rebuild a reported live chunk from the segment manifest --- .../unit/live-transcribe-core.test.ts | 77 +++++++++++++++++++ apps/web/lib/live-transcribe-core.ts | 65 ++++++++++++++++ 2 files changed, 142 insertions(+) diff --git a/apps/web/__tests__/unit/live-transcribe-core.test.ts b/apps/web/__tests__/unit/live-transcribe-core.test.ts index 9b79a7bf88..5112420c30 100644 --- a/apps/web/__tests__/unit/live-transcribe-core.test.ts +++ b/apps/web/__tests__/unit/live-transcribe-core.test.ts @@ -7,6 +7,7 @@ import { applyChunkToLiveTranscript, canPromoteLiveTranscript, createEmptyLiveTranscript, + describeLiveChunk, isNoSpokenAudioError, liveTranscriptToEditTranscript, offsetChunkWords, @@ -757,3 +758,79 @@ describe("live chunks with context", () => { expect(run(0).some((word) => word.text === "garbled")).toBe(true); }); }); + +describe("describeLiveChunk", () => { + it("rebuilds exactly what planNextLiveChunk planned", () => { + let seed = 7; + const random = () => { + seed = (seed * 1_103_515_245 + 12_345) % 2_147_483_648; + return seed / 2_147_483_648; + }; + let compared = 0; + for (let round = 0; round < 400; round++) { + const first = random() < 0.5 ? 0 : 1; + const manifest = { + ...baseManifest, + is_complete: random() < 0.3, + audio_segments: Array.from( + { length: 1 + Math.floor(random() * 40) }, + (_, i) => + seg(first + i, Math.round((0.4 + random() * 4) * 1000) / 1000), + ), + }; + const lastProcessedIndex = + random() < 0.2 ? -1 : first + Math.floor(random() * 30) - 1; + const contextSeconds = random() < 0.3 ? 0 : random() * 25; + const plan = planNextLiveChunk({ + manifest, + lastProcessedIndex, + targetSeconds: 5, + contextSeconds, + }); + if (plan.action !== "chunk") continue; + compared++; + const described = describeLiveChunk({ + manifest, + lastProcessedIndex, + lastIndex: plan.entries[plan.entries.length - 1]?.index ?? -1, + contextFromIndex: plan.context[0]?.index ?? null, + }); + expect(described).toEqual(plan); + } + expect(compared).toBeGreaterThan(100); + }); + + it("rejects ranges that don't continue the cursor or aren't contiguous", () => { + const manifest = { + ...baseManifest, + audio_segments: [1, 2, 3, 4, 6, 7].map((index) => seg(index)), + }; + const describe = ( + lastProcessedIndex: number, + lastIndex: number, + contextFromIndex: number | null = null, + ) => + describeLiveChunk({ + manifest, + lastProcessedIndex, + lastIndex, + contextFromIndex, + }); + + expect(describe(-1, 2)).toMatchObject({ startMs: 0, durationMs: 4000 }); + expect(describe(2, 4, 1)).toMatchObject({ + startMs: 4000, + durationMs: 4000, + contextMs: 4000, + }); + // Past the gap at index 5. + expect(describe(4, 6)).toBeNull(); + expect(describe(2, 7)).toBeNull(); + // Cursor beyond the contiguous audio, or behind the chunk start. + expect(describe(5, 6)).toBeNull(); + expect(describe(2, 2)).toBeNull(); + // Context must be audio before the cursor that exists. + expect(describe(2, 4, 3)).toBeNull(); + expect(describe(2, 4, 0)).toBeNull(); + }); +}); diff --git a/apps/web/lib/live-transcribe-core.ts b/apps/web/lib/live-transcribe-core.ts index 80dba5cc83..b3a3a4e86c 100644 --- a/apps/web/lib/live-transcribe-core.ts +++ b/apps/web/lib/live-transcribe-core.ts @@ -153,6 +153,71 @@ export function planNextLiveChunk(options: { }; } +/** + * Rebuild a chunk another runner transcribed from the segments it says it + * sent: `contextFromIndex..lastProcessedIndex` as context, then everything up + * to `lastIndex`. Offsets come from the same contiguous walk as + * planNextLiveChunk, so the runner never decides where words land. Null when + * the range doesn't describe contiguous audio right after the cursor. + */ +export function describeLiveChunk(options: { + manifest: Video.SegmentManifestType; + lastProcessedIndex: number; + lastIndex: number; + contextFromIndex: number | null; +}): Extract | null { + const { manifest, lastProcessedIndex, lastIndex, contextFromIndex } = options; + const plan = planSegmentsAudioExtraction(manifest, { + requireComplete: false, + }); + if (plan.status !== "ok") return null; + + let startMs = 0; + let contextSeconds = 0; + let durationSeconds = 0; + const context: NormalizedSegmentEntry[] = []; + const entries: NormalizedSegmentEntry[] = []; + let expectedIndex = plan.entries[0]?.index === 0 ? 0 : 1; + + for (const entry of plan.entries) { + if (entry.index !== expectedIndex || entry.index > lastIndex) break; + expectedIndex++; + + if (entry.index <= lastProcessedIndex) { + startMs += entry.duration * 1000; + if (contextFromIndex !== null && entry.index >= contextFromIndex) { + context.push(entry); + contextSeconds += entry.duration; + } + } else { + entries.push(entry); + durationSeconds += entry.duration; + } + } + + const first = entries[0]?.index; + if ( + first === undefined || + (lastProcessedIndex !== LIVE_TRANSCRIPT_NO_SEGMENTS && + first !== lastProcessedIndex + 1) || + entries[entries.length - 1]?.index !== lastIndex || + (contextFromIndex !== null && context[0]?.index !== contextFromIndex) || + (context.length > 0 && + context[context.length - 1]?.index !== lastProcessedIndex) + ) { + return null; + } + + return { + action: "chunk", + entries, + startMs: Math.round(startMs), + durationMs: Math.round(durationSeconds * 1000), + context, + contextMs: Math.round(contextSeconds * 1000), + }; +} + export interface LiveChunkWordInput { text?: unknown; start?: unknown; From 780e186f08fa353a1156e71a714d73ad019cea85 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:34:09 +0000 Subject: [PATCH 03/11] feat(media-server): run instant-mode live transcription sessions --- apps/media-server/Dockerfile | 1 + .../src/__tests__/lib/live-transcribe.test.ts | 426 +++++++++++ .../__tests__/routes/live-transcribe.test.ts | 105 +++ apps/media-server/src/app.ts | 4 + apps/media-server/src/index.ts | 19 +- apps/media-server/src/lib/live-transcribe.ts | 674 ++++++++++++++++++ .../src/routes/live-transcribe.ts | 51 ++ 7 files changed, 1278 insertions(+), 2 deletions(-) create mode 100644 apps/media-server/src/__tests__/lib/live-transcribe.test.ts create mode 100644 apps/media-server/src/__tests__/routes/live-transcribe.test.ts create mode 100644 apps/media-server/src/lib/live-transcribe.ts create mode 100644 apps/media-server/src/routes/live-transcribe.ts diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile index 83b944ecf5..19a7f950ca 100644 --- a/apps/media-server/Dockerfile +++ b/apps/media-server/Dockerfile @@ -27,6 +27,7 @@ RUN bun test src/__tests__/lib/audio-quality-policy.test.ts src/__tests__/lib/au && bun test src/__tests__/lib/lifecycle.test.ts src/__tests__/lib/temp-files.test.ts \ && bun test src/__tests__/routes/lifecycle.test.ts \ && bun test src/__tests__/routes/peer-jobs.test.ts \ + && bun test src/__tests__/lib/live-transcribe.test.ts src/__tests__/routes/live-transcribe.test.ts \ && bun test src/__tests__/lib/media-routes-real-world.integration.test.ts \ && bun test src/__tests__/lib/media-video.integration.test.ts diff --git a/apps/media-server/src/__tests__/lib/live-transcribe.test.ts b/apps/media-server/src/__tests__/lib/live-transcribe.test.ts new file mode 100644 index 0000000000..86b1ae9862 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/live-transcribe.test.ts @@ -0,0 +1,426 @@ +import { describe, expect, test } from "bun:test"; +import { + type LiveManifest, + LiveSession, + planLiveChunk, + type StartLiveSession, + transcribeAudio, +} from "../../lib/live-transcribe"; + +const seg = (index: number, duration = 2) => ({ index, duration }); +const manifestOf = ( + segments: LiveManifest["audio_segments"], + is_complete = false, +): LiveManifest => ({ + audio_init_uploaded: true, + audio_segments: segments, + is_complete, +}); +const everyUrl = () => true; + +describe("planLiveChunk", () => { + const plan = ( + manifest: LiveManifest, + lastProcessedIndex: number, + options: Partial[0]> = {}, + ) => + planLiveChunk({ + manifest, + lastProcessedIndex, + targetSeconds: 5, + contextSeconds: 0, + hasUrl: everyUrl, + ...options, + }); + + test("waits for enough new audio, then takes it all", () => { + expect(plan(manifestOf([seg(1), seg(2)]), -1)).toEqual({ action: "wait" }); + const chunk = plan(manifestOf([seg(1), seg(2), seg(3)]), -1); + expect(chunk).toMatchObject({ action: "chunk", context: [] }); + if (chunk.action === "chunk") { + expect(chunk.entries.map((entry) => entry.index)).toEqual([1, 2, 3]); + } + }); + + test("sends the remainder once the recording is complete, then is done", () => { + const manifest = manifestOf([seg(1), seg(2)], true); + expect(plan(manifest, -1)).toMatchObject({ action: "chunk" }); + expect(plan(manifest, 2)).toEqual({ action: "done" }); + expect(plan(manifestOf([], true), -1)).toEqual({ action: "no-audio" }); + expect(plan(manifestOf([]), -1)).toEqual({ action: "wait" }); + }); + + test("starts 0-based manifests at segment 0 and stops at gaps", () => { + const zero = plan(manifestOf([0, 1, 2], true), -1); + if (zero.action !== "chunk") throw new Error("expected a chunk"); + expect(zero.entries.map((entry) => entry.index)).toEqual([0, 1, 2]); + + const gap = plan(manifestOf([seg(1), seg(2), seg(4), seg(5)], true), -1); + if (gap.action !== "chunk") throw new Error("expected a chunk"); + expect(gap.entries.map((entry) => entry.index)).toEqual([1, 2]); + expect(plan(manifestOf([seg(1), seg(2), seg(4)], true), 2)).toEqual({ + action: "done", + }); + }); + + test("waits for the audio init while recording", () => { + expect( + plan( + { ...manifestOf([seg(1), seg(2), seg(3)]), audio_init_uploaded: false }, + -1, + ), + ).toEqual({ action: "wait" }); + }); + + test("caps one chunk at a minute of audio", () => { + const manifest = manifestOf( + Array.from({ length: 50 }, (_, i) => seg(i + 1, 3)), + ); + const chunk = plan(manifest, -1); + if (chunk.action !== "chunk") throw new Error("expected a chunk"); + expect(chunk.entries).toHaveLength(20); + }); + + test("adds earlier segments as context, only those it has URLs for", () => { + const manifest = manifestOf( + Array.from({ length: 20 }, (_, i) => seg(i + 1)), + ); + const chunk = plan(manifest, 10, { contextSeconds: 5 }); + if (chunk.action !== "chunk") throw new Error("expected a chunk"); + expect(chunk.context.map((entry) => entry.index)).toEqual([8, 9, 10]); + + const partial = plan(manifest, 10, { + contextSeconds: 5, + hasUrl: (index) => index >= 10, + }); + if (partial.action !== "chunk") throw new Error("expected a chunk"); + expect(partial.context.map((entry) => entry.index)).toEqual([10]); + }); + + test("asks for URLs when the next segment has none", () => { + const manifest = manifestOf([seg(1), seg(2), seg(3)], true); + expect(plan(manifest, -1, { hasUrl: () => false })).toEqual({ + action: "need-urls", + }); + const limited = plan(manifest, -1, { hasUrl: (index) => index <= 2 }); + if (limited.action !== "chunk") throw new Error("expected a chunk"); + expect(limited.entries.map((entry) => entry.index)).toEqual([1, 2]); + }); +}); + +type Report = Record & { action: string }; + +/** Storage, transcription API and the web app's report endpoint in one fake. */ +function createFakeCap() { + const segments = new Map>(); + let manifest: LiveManifest = manifestOf([]); + let manifestVersion = 0; + const reports: Report[] = []; + const uploads = new Map(); + const failAudio = new Set(); + let cursor = -1; + let stopAfter: string | null = null; + let manifestGets = 0; + let notModified = 0; + let server: ReturnType; + let asrSubmits = 0; + + const state = () => { + const urls: Record = {}; + for (let index = Math.max(0, cursor - 15); index <= cursor + 40; index++) { + urls[String(index)] = `${base()}/storage/segment/${index}`; + } + return { + lastAudioSegmentIndex: cursor, + transcribedDurationMs: 0, + transcriptionOptions: { language_code: "en" }, + urls: { + manifest: `${base()}/storage/manifest`, + audioInit: `${base()}/storage/init`, + segments: urls, + }, + }; + }; + const base = () => `http://127.0.0.1:${server.port}`; + + server = Bun.serve({ + port: 0, + async fetch(request) { + const url = new URL(request.url); + const path = url.pathname; + if (path === "/storage/manifest") { + manifestGets++; + const etag = `"m${manifestVersion}"`; + if (request.headers.get("if-none-match") === etag) { + notModified++; + return new Response(null, { status: 304 }); + } + return Response.json(manifest, { headers: { etag } }); + } + if (path === "/storage/init") return new Response(new Uint8Array([0])); + if (path.startsWith("/storage/segment/")) { + const bytes = segments.get(Number(path.split("/").pop())); + return bytes + ? new Response(bytes) + : new Response(null, { status: 404 }); + } + if (path === "/asr/v2/upload") { + expect(request.headers.get("authorization")).toBe("asr-key"); + const id = `u${uploads.size}`; + uploads.set(id, new Uint8Array(await request.arrayBuffer())); + return Response.json({ upload_url: `${base()}/asr/files/${id}` }); + } + if (path === "/asr/v2/transcript") { + asrSubmits++; + const body = (await request.json()) as { audio_url: string }; + const id = body.audio_url.split("/").pop() as string; + return Response.json({ id: `t-${id}`, status: "queued" }); + } + if (path.startsWith("/asr/v2/transcript/")) { + const id = (path.split("/").pop() as string).slice(2); + const audio = uploads.get(id) as Uint8Array; + const key = [...audio.slice(1)].join(","); + if (failAudio.has(key)) { + return Response.json({ id, status: "error", error: "boom" }); + } + // One word per audio byte after the init byte: its segment index. + return Response.json({ + id, + status: "completed", + language_code: "en", + words: [...audio.slice(1)].map((value, i) => ({ + text: `w${value}`, + start: i * 1000, + end: i * 1000 + 500, + confidence: 0.9, + speaker: null, + })), + }); + } + if (path === "/web/report") { + expect(request.headers.get("x-media-server-secret")).toBe("secret"); + const report = (await request.json()) as Report; + reports.push(report); + if (stopAfter && report.action === stopAfter) { + return Response.json({ continue: false }); + } + if (report.action === "chunk" || report.action === "gap") { + if (report.lastProcessedIndex === cursor) { + cursor = report.lastIndex as number; + } + } + if (report.action === "finish" || report.action === "release") { + return Response.json({ continue: false }); + } + return Response.json({ continue: true, state: state() }); + } + return new Response(null, { status: 404 }); + }, + }); + + return { + get base() { + return base(); + }, + reports, + state, + failAudio, + setCursor(value: number) { + cursor = value; + }, + stopOn(action: string) { + stopAfter = action; + }, + get manifestGets() { + return manifestGets; + }, + get notModified() { + return notModified; + }, + get asrSubmits() { + return asrSubmits; + }, + record(indices: number[], is_complete = false) { + for (const index of indices) segments.set(index, new Uint8Array([index])); + manifest = manifestOf( + [...segments.keys()].sort((a, b) => a - b).map((index) => seg(index)), + is_complete, + ); + manifestVersion++; + }, + stop() { + server.stop(true); + }, + }; +} + +const fastSleep = (ms: number, signal?: AbortSignal) => + new Promise((resolve) => { + const timer = setTimeout(resolve, Math.min(ms, 5)); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + }); + +function startRequest( + cap: ReturnType, + contextSeconds = 0, +): StartLiveSession { + return { + videoId: "video1", + session: "s1", + callbackUrl: `${cap.base}/web/report`, + asr: { baseUrl: `${cap.base}/asr`, apiKey: "asr-key" }, + contextSeconds, + state: cap.state(), + }; +} + +const deps = { sleep: fastSleep, secret: "secret" }; + +describe("LiveSession", () => { + test("transcribes a recording as it grows, then reports done", async () => { + const fresh = createFakeCap(); + const session = new LiveSession(startRequest(fresh, 4), deps); + const done = session.start(); + for (let index = 1; index <= 24; index++) { + fresh.record([index], index === 24); + await Bun.sleep(3); + } + await done; + fresh.stop(); + + const chunks = fresh.reports.filter((report) => report.action === "chunk"); + let expected = -1; + for (const chunk of chunks) { + expect(chunk.lastProcessedIndex).toBe(expected); + expected = chunk.lastIndex as number; + } + expect(expected).toBe(24); + // The first chunk is small, and later ones carry earlier audio as + // context, which the words prove was sent in order. + expect(chunks[0]?.contextFromIndex).toBeNull(); + const later = chunks.find((chunk) => chunk.contextFromIndex !== null); + expect(later).toBeDefined(); + if (later) { + const from = later.contextFromIndex as number; + const words = (later.words as { text: string }[]).map((w) => w.text); + expect(words[0]).toBe(`w${from}`); + expect(words.at(-1)).toBe(`w${later.lastIndex}`); + } + expect(fresh.reports.at(-1)).toMatchObject({ + action: "finish", + outcome: "done", + videoId: "video1", + session: "s1", + }); + expect(fresh.notModified).toBeGreaterThan(0); + }); + + test("retries a failing chunk without context, then reports a gap", async () => { + const fresh = createFakeCap(); + fresh.record([1, 2, 3, 4, 5], true); + fresh.failAudio.add("1,2,3,4,5"); + const session = new LiveSession(startRequest(fresh, 10), deps); + await session.start(); + fresh.stop(); + + expect(fresh.asrSubmits).toBe(2); + expect(fresh.reports.map((report) => report.action)).toEqual([ + "gap", + "finish", + ]); + expect(fresh.reports[0]).toMatchObject({ + lastProcessedIndex: -1, + lastIndex: 5, + }); + }); + + test("stops quietly when the web app says to", async () => { + const fresh = createFakeCap(); + fresh.record([1, 2, 3, 4]); + fresh.stopOn("chunk"); + await new LiveSession(startRequest(fresh), deps).start(); + fresh.stop(); + expect(fresh.reports.map((report) => report.action)).toEqual(["chunk"]); + }); + + test("ends with no-audio for a finished recording without audio", async () => { + const fresh = createFakeCap(); + fresh.record([], true); + await new LiveSession(startRequest(fresh), deps).start(); + fresh.stop(); + expect(fresh.reports).toEqual([ + expect.objectContaining({ action: "finish", outcome: "no-audio" }), + ]); + }); + + test("hands the recording back when released", async () => { + const fresh = createFakeCap(); + fresh.record([1]); + const session = new LiveSession(startRequest(fresh), deps); + session.start(); + await Bun.sleep(20); + await session.release(); + fresh.stop(); + expect(fresh.reports.at(-1)).toMatchObject({ action: "release" }); + expect( + fresh.reports.some((report) => report.action === "finish"), + ).toBeFalse(); + }); + + test("resyncs to the web app's cursor instead of resending audio", async () => { + const fresh = createFakeCap(); + fresh.record([1, 2, 3, 4, 5, 6], true); + const request = startRequest(fresh); + // The web app already holds 1..3 (say a report's answer was lost). + fresh.setCursor(3); + await new LiveSession(request, deps).start(); + fresh.stop(); + const chunks = fresh.reports.filter((report) => report.action === "chunk"); + expect( + chunks.map((chunk) => [chunk.lastProcessedIndex, chunk.lastIndex]), + ).toEqual([ + [-1, 6], + [3, 6], + ]); + expect(fresh.reports.at(-1)).toMatchObject({ outcome: "done" }); + }); +}); + +describe("transcribeAudio", () => { + test("treats speech-free audio as an empty chunk and surfaces errors", async () => { + const server = Bun.serve({ + port: 0, + async fetch(request) { + const path = new URL(request.url).pathname; + if (path === "/v2/upload") return Response.json({ upload_url: "x" }); + const body = (await request.json()) as { language_code: string }; + return Response.json({ + id: "t1", + status: "error", + error: + body.language_code === "quiet" + ? "No spoken audio detected" + : "decode failed", + }); + }, + }); + const call = (language_code: string) => + transcribeAudio({ + baseUrl: `http://127.0.0.1:${server.port}/`, + apiKey: "k", + audio: new Uint8Array([1]), + transcriptionOptions: { language_code }, + signal: new AbortController().signal, + }); + expect(await call("quiet")).toEqual({ words: [], languageCode: null }); + await expect(call("en")).rejects.toThrow("decode failed"); + server.stop(true); + }); +}); diff --git a/apps/media-server/src/__tests__/routes/live-transcribe.test.ts b/apps/media-server/src/__tests__/routes/live-transcribe.test.ts new file mode 100644 index 0000000000..b0ea23c533 --- /dev/null +++ b/apps/media-server/src/__tests__/routes/live-transcribe.test.ts @@ -0,0 +1,105 @@ +import { afterAll, describe, expect, test } from "bun:test"; +import app from "../../app"; +import { setDraining } from "../../lib/lifecycle"; + +const SECRET = "test-secret"; +process.env.MEDIA_SERVER_WEBHOOK_SECRET = SECRET; + +const reports: Record[] = []; +let reported: () => void = () => {}; +const firstReport = new Promise((resolve) => { + reported = resolve; +}); +const server = Bun.serve({ + port: 0, + async fetch(request) { + const path = new URL(request.url).pathname; + if (path === "/manifest") { + return Response.json({ + audio_init_uploaded: false, + audio_segments: [], + is_complete: true, + }); + } + if (path === "/report") { + reports.push(await request.json()); + reported(); + return Response.json({ continue: false }); + } + return new Response(null, { status: 404 }); + }, +}); +afterAll(() => server.stop(true)); + +const base = `http://127.0.0.1:${server.port}`; +const body = { + videoId: "video1", + session: "s1", + callbackUrl: `${base}/report`, + asr: { baseUrl: `${base}/asr`, apiKey: "k" }, + contextSeconds: 20, + state: { + lastAudioSegmentIndex: -1, + transcribedDurationMs: 0, + transcriptionOptions: {}, + urls: { + manifest: `${base}/manifest`, + audioInit: `${base}/init`, + segments: { "0": `${base}/0` }, + }, + }, +}; + +const post = (payload: unknown, secret: string | null = SECRET) => + app.fetch( + new Request("http://localhost/live-transcribe/sessions", { + method: "POST", + headers: { + "Content-Type": "application/json", + ...(secret ? { "x-media-server-secret": secret } : {}), + }, + body: JSON.stringify(payload), + }), + ); + +describe("POST /live-transcribe/sessions", () => { + test("rejects requests without the secret or with a bad body", async () => { + expect((await post(body, null)).status).toBe(401); + expect((await post(body, "wrong")).status).toBe(401); + expect((await post({ ...body, callbackUrl: "nope" })).status).toBe(400); + expect((await post({ ...body, contextSeconds: 600 })).status).toBe(400); + }); + + test("refuses new sessions while draining", async () => { + setDraining(true); + try { + const response = await post(body); + expect(response.status).toBe(503); + expect(response.headers.get("Retry-After")).toBeTruthy(); + } finally { + setDraining(false); + } + }); + + test("starts a session that reports back to the web app", async () => { + const response = await post(body); + expect(response.status).toBe(202); + await firstReport; + expect(reports).toEqual([ + { + action: "finish", + outcome: "no-audio", + videoId: "video1", + session: "s1", + }, + ]); + await Bun.sleep(20); + const status = await app.fetch( + new Request("http://localhost/live-transcribe/status"), + ); + expect(await status.json()).toEqual({ + sessions: 0, + canAcceptNewSession: true, + }); + }); +}); diff --git a/apps/media-server/src/app.ts b/apps/media-server/src/app.ts index 162a7652dd..96e7ca32cb 100644 --- a/apps/media-server/src/app.ts +++ b/apps/media-server/src/app.ts @@ -3,6 +3,7 @@ import { logger } from "hono/logger"; import { beginRequest } from "./lib/lifecycle"; import audio from "./routes/audio"; import health from "./routes/health"; +import liveTranscribe from "./routes/live-transcribe"; import video from "./routes/video"; const app = new Hono(); @@ -21,6 +22,7 @@ app.use("*", async (c, next) => { app.route("/health", health); app.route("/audio", audio); app.route("/video", video); +app.route("/live-transcribe", liveTranscribe); app.get("/", (c) => { return c.json({ @@ -42,6 +44,8 @@ app.get("/", (c) => { "/video/process/:jobId/cancel", "/video/cleanup", "/video/force-cleanup", + "/live-transcribe/status", + "/live-transcribe/sessions", ], }); }); diff --git a/apps/media-server/src/index.ts b/apps/media-server/src/index.ts index 05dbdac4fd..bbda6eef3c 100644 --- a/apps/media-server/src/index.ts +++ b/apps/media-server/src/index.ts @@ -1,6 +1,12 @@ import app from "./app"; import { abortAllJobs } from "./lib/job-manager"; -import { createShutdownHandler, getIdleTimeoutSeconds } from "./lib/lifecycle"; +import { + createShutdownHandler, + getIdleTimeoutSeconds, + setDraining, + trackBackgroundTask, +} from "./lib/lifecycle"; +import { releaseLiveSessions } from "./lib/live-transcribe"; import { cancelAllMediaOperations } from "./lib/media-operations"; import { cleanupMediaTransferCache, @@ -24,7 +30,8 @@ void Promise.allSettled([ } }); -const shutdown = createShutdownHandler(async () => { +const drainAndShutdown = createShutdownHandler(async () => { + await releaseLiveSessions(); const abortedJobs = await abortAllJobs(); if (abortedJobs > 0) { console.log(`[media-server] Aborted ${abortedJobs} active jobs`); @@ -35,6 +42,14 @@ const shutdown = createShutdownHandler(async () => { process.exit(0); }); +// Live transcription sessions last as long as a recording, so they move to +// another server as soon as draining starts instead of holding it open. +const shutdown = (signal: string) => { + setDraining(true); + void trackBackgroundTask(releaseLiveSessions()); + return drainAndShutdown(signal); +}; + process.on("SIGINT", () => { void shutdown("SIGINT"); }); diff --git a/apps/media-server/src/lib/live-transcribe.ts b/apps/media-server/src/lib/live-transcribe.ts new file mode 100644 index 0000000000..9bb9e4ee7b --- /dev/null +++ b/apps/media-server/src/lib/live-transcribe.ts @@ -0,0 +1,674 @@ +import { z } from "zod"; + +/** + * Live transcription sessions for instant-mode recordings. Cap's web app + * starts one per recording (apps/web/lib/live-transcribe-farm.ts). The session + * watches the segment manifest, transcribes new audio in chunks through the + * AssemblyAI-compatible API it was given, and reports each chunk's words to + * the web app, which owns the transcript artifact, the claim, and promotion. + * Every report answers with the authoritative cursor and fresh signed URLs, + * so this server holds no storage credentials and no transcript state of its + * own: a lost session costs only the chunk in flight. + */ + +export const LIVE = { + INITIAL_CHUNK_SECONDS: 5, + MAX_CHUNK_SECONDS: 10, + GROW_AFTER_CHUNKS: 2, + MAX_CHUNK_TAKE_SECONDS: 60, + MAX_TRANSCRIBED_MS: 60 * 60 * 1000, + MAX_CHUNKS: 420, + /** The manifest is re-uploaded after every ~3 s segment. */ + POLL_INTERVAL_MS: 1_000, + /** No new audio for this long ends the session (the workflow's 4 x 2 min). */ + MAX_IDLE_MS: 8 * 60 * 1000, + /** Reports at least this often while idle keep the claim fresh (the web + * app treats a claim older than 3 minutes as dead) and the URLs signed. */ + SYNC_INTERVAL_MS: 60_000, + MAX_SESSION_MS: 3 * 60 * 60 * 1000, + MAX_CHUNK_ATTEMPTS: 2, + MAX_SESSIONS: 200, + ASR_POLL_INTERVAL_MS: 500, + ASR_TIMEOUT_MS: 3 * 60 * 1000, + DOWNLOAD_TIMEOUT_MS: 30_000, + DOWNLOAD_CONCURRENCY: 8, + MAX_CHUNK_BYTES: 64 * 1024 * 1024, + REPORT_TIMEOUT_MS: 30_000, + FINISH_TIMEOUT_MS: 110_000, +} as const; + +const manifestEntrySchema = z.union([ + z.number(), + z.object({ index: z.number(), duration: z.number() }), +]); + +const manifestSchema = z.object({ + audio_init_uploaded: z.boolean(), + audio_segments: z.array(manifestEntrySchema), + is_complete: z.boolean(), +}); + +export type LiveManifest = z.infer; + +const stateSchema = z.object({ + lastAudioSegmentIndex: z.number().int(), + transcribedDurationMs: z.number(), + transcriptionOptions: z.record(z.unknown()), + urls: z.object({ + manifest: z.string().url(), + audioInit: z.string().url(), + segments: z.record(z.string().url()), + }), +}); + +export type LiveSessionState = z.infer; + +export const startLiveSessionSchema = z.object({ + videoId: z.string().min(1).max(64), + session: z.string().min(1).max(64), + callbackUrl: z.string().url(), + asr: z.object({ + baseUrl: z.string().url(), + apiKey: z.string().min(1), + }), + contextSeconds: z.number().min(0).max(60), + state: stateSchema, +}); + +export type StartLiveSession = z.infer; + +const reportAnswerSchema = z.union([ + z.object({ continue: z.literal(false) }), + z.object({ continue: z.literal(true), state: stateSchema }), +]); + +type Entry = { index: number; duration: number }; + +export type LiveChunkPlan = + | { action: "wait" } + | { action: "no-audio" } + | { action: "done" } + | { action: "need-urls" } + | { action: "chunk"; entries: Entry[]; context: Entry[] }; + +/** + * The web app's planNextLiveChunk (apps/web/lib/live-transcribe-core.ts), + * limited to segments this session has URLs for. Only decides which segments + * to send; the web app rebuilds the timing from the manifest itself and + * rejects a chunk that doesn't start right after its cursor. + */ +export function planLiveChunk(options: { + manifest: LiveManifest; + lastProcessedIndex: number; + targetSeconds: number; + contextSeconds: number; + hasUrl: (index: number) => boolean; +}): LiveChunkPlan { + const { manifest, lastProcessedIndex, targetSeconds, hasUrl } = options; + + if (manifest.audio_segments.length === 0) { + return manifest.is_complete ? { action: "no-audio" } : { action: "wait" }; + } + if (!manifest.is_complete && !manifest.audio_init_uploaded) { + return { action: "wait" }; + } + + const byIndex = new Map(); + for (const raw of manifest.audio_segments) { + const entry = typeof raw === "number" ? { index: raw, duration: 3 } : raw; + if ( + !Number.isFinite(entry.index) || + entry.index < 0 || + !Number.isFinite(entry.duration) || + entry.duration < 0 + ) { + return { action: "wait" }; + } + if (!byIndex.has(entry.index)) byIndex.set(entry.index, entry); + } + const sorted = [...byIndex.values()].sort((a, b) => a.index - b.index); + + const processed: Entry[] = []; + const pending: Entry[] = []; + let expectedIndex = sorted[0]?.index === 0 ? 0 : 1; + for (const entry of sorted) { + if (entry.index !== expectedIndex) break; + expectedIndex++; + if (entry.index <= lastProcessedIndex) processed.push(entry); + else pending.push(entry); + } + + if (pending.length === 0) { + return manifest.is_complete ? { action: "done" } : { action: "wait" }; + } + + const entries: Entry[] = []; + let takenSeconds = 0; + for (const entry of pending) { + if (!hasUrl(entry.index)) break; + if ( + entries.length > 0 && + takenSeconds + entry.duration > LIVE.MAX_CHUNK_TAKE_SECONDS + ) { + break; + } + entries.push(entry); + takenSeconds += entry.duration; + if (takenSeconds >= LIVE.MAX_CHUNK_TAKE_SECONDS) break; + } + if (entries.length === 0) return { action: "need-urls" }; + + const tookEverything = entries.length === pending.length; + if (!manifest.is_complete && tookEverything && takenSeconds < targetSeconds) { + return { action: "wait" }; + } + + const context: Entry[] = []; + let contextTaken = 0; + for (let i = processed.length - 1; i >= 0; i--) { + const entry = processed[i] as Entry; + if (contextTaken >= options.contextSeconds || !hasUrl(entry.index)) break; + context.unshift(entry); + contextTaken += entry.duration; + } + + return { action: "chunk", entries, context }; +} + +export function isNoSpokenAudioError(transcript: { + status?: unknown; + error?: unknown; +}) { + return ( + transcript.status === "error" && + typeof transcript.error === "string" && + /no spoken audio/i.test(transcript.error) + ); +} + +type AsrWord = { + text: string; + start: number; + end: number; + confidence: number | null; + speaker: string | null; +}; + +type AsrResult = { words: AsrWord[]; languageCode: string | null }; + +function pickWords(words: unknown): AsrWord[] { + if (!Array.isArray(words)) return []; + const result: AsrWord[] = []; + for (const word of words) { + if (!word || typeof word !== "object") continue; + const { text, start, end, confidence, speaker } = word as Record< + string, + unknown + >; + if ( + typeof text !== "string" || + typeof start !== "number" || + typeof end !== "number" + ) { + continue; + } + result.push({ + text: text.slice(0, 200), + start, + end, + confidence: typeof confidence === "number" ? confidence : null, + speaker: typeof speaker === "string" ? speaker.slice(0, 40) : null, + }); + } + return result; +} + +function sleep(ms: number, signal?: AbortSignal) { + return new Promise((resolve) => { + if (signal?.aborted) return resolve(); + const done = () => { + clearTimeout(timer); + signal?.removeEventListener("abort", done); + resolve(); + }; + const timer = setTimeout(done, ms); + signal?.addEventListener("abort", done, { once: true }); + }); +} + +async function readJson(response: Response, what: string) { + if (!response.ok) { + const detail = (await response.text().catch(() => "")).slice(0, 200); + throw new Error(`${what} failed: HTTP ${response.status} ${detail}`); + } + return (await response.json()) as Record; +} + +/** Upload, submit and wait, like the AssemblyAI SDK's transcripts.transcribe, + * but polling every 500 ms instead of 3 s (Cap's own service answers the + * submit with the finished transcript when it can). */ +export async function transcribeAudio(options: { + baseUrl: string; + apiKey: string; + audio: Uint8Array; + transcriptionOptions: Record; + signal: AbortSignal; +}): Promise { + const base = options.baseUrl.replace(/\/+$/, ""); + const deadline = AbortSignal.timeout(LIVE.ASR_TIMEOUT_MS); + const signal = AbortSignal.any([options.signal, deadline]); + const headers = { Authorization: options.apiKey }; + + const upload = await readJson( + await fetch(`${base}/v2/upload`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/octet-stream" }, + body: options.audio, + signal, + }), + "Transcription upload", + ); + if (typeof upload.upload_url !== "string") { + throw new Error("Transcription upload returned no upload_url"); + } + + let transcript = await readJson( + await fetch(`${base}/v2/transcript`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ + ...options.transcriptionOptions, + audio_url: upload.upload_url, + }), + signal, + }), + "Transcription submit", + ); + const id = transcript.id; + if (typeof id !== "string" || !/^[\w-]+$/.test(id)) { + throw new Error("Transcription submit returned no id"); + } + + while (transcript.status !== "completed" && transcript.status !== "error") { + await sleep(LIVE.ASR_POLL_INTERVAL_MS, signal); + if (signal.aborted) throw new Error("Transcription timed out"); + transcript = await readJson( + await fetch(`${base}/v2/transcript/${id}`, { headers, signal }), + "Transcription poll", + ); + } + + if (transcript.status === "error") { + if (isNoSpokenAudioError(transcript)) { + return { words: [], languageCode: null }; + } + throw new Error( + `Transcription failed: ${String(transcript.error ?? "unknown error").slice(0, 200)}`, + ); + } + + return { + words: pickWords(transcript.words), + languageCode: + typeof transcript.language_code === "string" + ? transcript.language_code.slice(0, 16) + : null, + }; +} + +type ReportAnswer = "ok" | "stop" | "failed"; + +export type LiveSessionDeps = { + sleep?: (ms: number, signal?: AbortSignal) => Promise; + secret?: string; +}; + +export class LiveSession { + readonly videoId: string; + readonly session: string; + private readonly request: StartLiveSession; + private state: LiveSessionState; + private readonly abort = new AbortController(); + private releasing = false; + private manifest: { etag: string | null; value: LiveManifest } | null = null; + private initAudio: Uint8Array | null = null; + /** Recently sent segments, so context audio isn't downloaded twice. */ + private readonly segmentCache = new Map(); + private lastReportAt: number; + private readonly sleep: (ms: number, signal?: AbortSignal) => Promise; + private readonly secret: string | undefined; + done: Promise = Promise.resolve(); + + constructor(request: StartLiveSession, deps: LiveSessionDeps = {}) { + this.videoId = request.videoId; + this.session = request.session; + this.request = request; + this.state = request.state; + this.sleep = deps.sleep ?? sleep; + this.secret = deps.secret ?? process.env.MEDIA_SERVER_WEBHOOK_SECRET; + this.lastReportAt = Date.now(); + } + + start() { + this.done = this.run(); + return this.done; + } + + /** Stop without reporting; the web app already moved on. */ + cancel() { + this.abort.abort(); + } + + /** Stop and ask the web app to run the recording elsewhere. */ + release() { + this.releasing = true; + this.abort.abort(); + return this.done; + } + + private async run() { + let outcome: string | null = null; + try { + outcome = await this.loop(); + } catch (error) { + console.warn( + `[live-transcribe] ${this.videoId} session failed:`, + error instanceof Error ? error.message : String(error), + ); + outcome = "session-error"; + } + if (this.releasing) { + await this.report({ action: "release" }, LIVE.FINISH_TIMEOUT_MS, 3); + } else if (outcome) { + await this.report( + { action: "finish", outcome }, + LIVE.FINISH_TIMEOUT_MS, + 5, + ); + } + this.segmentCache.clear(); + this.initAudio = null; + } + + private get signal() { + return this.abort.signal; + } + + private async loop(): Promise { + const now = Date.now; + const startedAt = now(); + let lastProgressAt = startedAt; + let chunkCount = + this.state.lastAudioSegmentIndex >= 0 ? LIVE.GROW_AFTER_CHUNKS : 0; + let failures = 0; + + while (!this.signal.aborted) { + if (now() - startedAt > LIVE.MAX_SESSION_MS) return "session-limit"; + if (chunkCount >= LIVE.MAX_CHUNKS) return "chunk-limit"; + if (this.state.transcribedDurationMs >= LIVE.MAX_TRANSCRIBED_MS) { + return "budget-exhausted"; + } + + const manifest = await this.readManifest(); + if (this.signal.aborted) break; + const plan = manifest + ? planLiveChunk({ + manifest, + lastProcessedIndex: this.state.lastAudioSegmentIndex, + targetSeconds: + chunkCount < LIVE.GROW_AFTER_CHUNKS + ? LIVE.INITIAL_CHUNK_SECONDS + : LIVE.MAX_CHUNK_SECONDS, + // A retry drops the context in case the earlier audio is + // what fails. + contextSeconds: failures === 0 ? this.request.contextSeconds : 0, + hasUrl: (index) => String(index) in this.state.urls.segments, + }) + : ({ action: "wait" } as const); + + if (plan.action === "done") return "done"; + if (plan.action === "no-audio") return "no-audio"; + + if (plan.action === "wait" || plan.action === "need-urls") { + if (now() - lastProgressAt > LIVE.MAX_IDLE_MS) return "stalled"; + const sinceReport = now() - this.lastReportAt; + if ( + sinceReport > + (plan.action === "need-urls" + ? LIVE.POLL_INTERVAL_MS * 5 + : LIVE.SYNC_INTERVAL_MS) + ) { + const answer = await this.report({ action: "sync" }); + if (answer === "stop") return null; + } + await this.sleep(LIVE.POLL_INTERVAL_MS, this.signal); + continue; + } + + const cursor = this.state.lastAudioSegmentIndex; + const range = `${plan.entries[0]?.index}..${plan.entries.at(-1)?.index}`; + const lastIndex = plan.entries.at(-1)?.index ?? cursor; + let report: Record; + if (failures >= LIVE.MAX_CHUNK_ATTEMPTS) { + console.warn( + `[live-transcribe] ${this.videoId} skipping chunk ${range}`, + ); + report = { action: "gap", lastProcessedIndex: cursor, lastIndex }; + } else { + try { + const audio = await this.loadAudio([ + ...plan.context, + ...plan.entries, + ]); + const result = await transcribeAudio({ + baseUrl: this.request.asr.baseUrl, + apiKey: this.request.asr.apiKey, + audio, + transcriptionOptions: this.state.transcriptionOptions, + signal: this.signal, + }); + report = { + action: "chunk", + lastProcessedIndex: cursor, + lastIndex, + contextFromIndex: plan.context[0]?.index ?? null, + words: result.words, + languageCode: result.languageCode, + }; + } catch (error) { + if (this.signal.aborted) break; + console.warn( + `[live-transcribe] ${this.videoId} chunk ${range} failed:`, + error instanceof Error ? error.message : String(error), + ); + failures++; + if (failures >= LIVE.MAX_CHUNK_ATTEMPTS * 3) return "chunk-abandoned"; + continue; + } + } + + const answer = await this.report(report); + if (answer === "stop") return null; + if (answer === "ok" && this.state.lastAudioSegmentIndex > cursor) { + failures = 0; + chunkCount++; + lastProgressAt = now(); + continue; + } + // Not accepted: the report failed, or the web app's cursor or + // manifest disagree with this chunk. Retry, then skip, then give up. + failures++; + if (failures >= LIVE.MAX_CHUNK_ATTEMPTS * 3) return "chunk-abandoned"; + await this.sleep(LIVE.POLL_INTERVAL_MS, this.signal); + } + return null; + } + + private async readManifest(): Promise { + try { + const response = await fetch(this.state.urls.manifest, { + headers: this.manifest?.etag + ? { "If-None-Match": this.manifest.etag } + : undefined, + signal: AbortSignal.any([ + this.signal, + AbortSignal.timeout(LIVE.DOWNLOAD_TIMEOUT_MS), + ]), + }); + if (response.status === 304 && this.manifest) { + return this.manifest.value; + } + if (!response.ok) { + await response.body?.cancel().catch(() => {}); + return null; + } + const parsed = manifestSchema.safeParse(await response.json()); + if (!parsed.success) return null; + this.manifest = { + etag: response.headers.get("etag"), + value: parsed.data, + }; + return parsed.data; + } catch { + return null; + } + } + + private async loadAudio(entries: Entry[]): Promise> { + const fetchBytes = async (url: string) => { + const response = await fetch(url, { + signal: AbortSignal.any([ + this.signal, + AbortSignal.timeout(LIVE.DOWNLOAD_TIMEOUT_MS), + ]), + }); + if (!response.ok) { + await response.body?.cancel().catch(() => {}); + throw new Error(`Segment not accessible: HTTP ${response.status}`); + } + return new Uint8Array(await response.arrayBuffer()); + }; + + this.initAudio ??= await fetchBytes(this.state.urls.audioInit); + const missing = entries.filter( + (entry) => !this.segmentCache.has(entry.index), + ); + for ( + let start = 0; + start < missing.length; + start += LIVE.DOWNLOAD_CONCURRENCY + ) { + const batch = missing.slice(start, start + LIVE.DOWNLOAD_CONCURRENCY); + const buffers = await Promise.all( + batch.map((entry) => { + const url = this.state.urls.segments[String(entry.index)]; + if (!url) throw new Error(`No URL for segment ${entry.index}`); + return fetchBytes(url); + }), + ); + for (const [offset, entry] of batch.entries()) { + this.segmentCache.set(entry.index, buffers[offset] as Uint8Array); + } + } + + const parts = [ + this.initAudio, + ...entries.map((entry) => this.segmentCache.get(entry.index)), + ] as Uint8Array[]; + const size = parts.reduce((sum, part) => sum + part.byteLength, 0); + if (size > LIVE.MAX_CHUNK_BYTES) { + throw new Error(`Chunk audio exceeds ${LIVE.MAX_CHUNK_BYTES} bytes`); + } + const audio = new Uint8Array(size); + let offset = 0; + for (const part of parts) { + audio.set(part, offset); + offset += part.byteLength; + } + + // Later chunks only reuse the segments just sent as context. + const keepFrom = (entries[entries.length - 1]?.index ?? 0) - 64; + for (const index of this.segmentCache.keys()) { + if (index < keepFrom) this.segmentCache.delete(index); + } + return audio; + } + + private async report( + body: Record, + timeoutMs: number = LIVE.REPORT_TIMEOUT_MS, + attempts = 3, + ): Promise { + for (let attempt = 0; attempt < attempts; attempt++) { + if (attempt > 0) await this.sleep(1_000 * 2 ** (attempt - 1)); + try { + const response = await fetch(this.request.callbackUrl, { + method: "POST", + headers: { + "Content-Type": "application/json", + ...(this.secret ? { "x-media-server-secret": this.secret } : {}), + }, + body: JSON.stringify({ + ...body, + videoId: this.videoId, + session: this.session, + }), + signal: AbortSignal.timeout(timeoutMs), + }); + if (response.status >= 400 && response.status < 500) { + await response.body?.cancel().catch(() => {}); + return "failed"; + } + if (!response.ok) { + await response.body?.cancel().catch(() => {}); + continue; + } + const answer = reportAnswerSchema.safeParse(await response.json()); + if (!answer.success) return "failed"; + this.lastReportAt = Date.now(); + if (!answer.data.continue) return "stop"; + this.state = answer.data.state; + return "ok"; + } catch {} + } + return "failed"; + } +} + +const sessions = new Map(); + +export function getLiveSessionCount() { + return sessions.size; +} + +export function canAcceptLiveSession() { + return sessions.size < LIVE.MAX_SESSIONS; +} + +/** Start a session, replacing any older one this server holds for the same + * recording (the web app has already fenced the older one off). */ +export function startLiveSession( + request: StartLiveSession, + deps?: LiveSessionDeps, +): LiveSession { + sessions.get(request.videoId)?.cancel(); + const session = new LiveSession(request, deps); + sessions.set(request.videoId, session); + void session.start().finally(() => { + if (sessions.get(request.videoId) === session) { + sessions.delete(request.videoId); + } + }); + return session; +} + +let releasing: Promise | null = null; + +/** Hand every session back to the web app, which restarts each on another + * server (or its workflow). Called once when this server starts draining. */ +export function releaseLiveSessions(): Promise { + releasing ??= Promise.all( + [...sessions.values()].map((session) => session.release()), + ).then(() => {}); + return releasing; +} diff --git a/apps/media-server/src/routes/live-transcribe.ts b/apps/media-server/src/routes/live-transcribe.ts new file mode 100644 index 0000000000..4aedcd1940 --- /dev/null +++ b/apps/media-server/src/routes/live-transcribe.ts @@ -0,0 +1,51 @@ +import { Hono } from "hono"; +import { canReadStatus, validateMediaServerSecret } from "../lib/auth"; +import { DRAIN_RETRY_AFTER_SECONDS, isDraining } from "../lib/lifecycle"; +import { + canAcceptLiveSession, + getLiveSessionCount, + startLiveSession, + startLiveSessionSchema, +} from "../lib/live-transcribe"; + +const liveTranscribe = new Hono(); + +liveTranscribe.get("/status", (c) => { + if (!canReadStatus(c)) { + return c.json({ error: "Unauthorized" }, 401); + } + return c.json({ + sessions: getLiveSessionCount(), + canAcceptNewSession: !isDraining() && canAcceptLiveSession(), + }); +}); + +liveTranscribe.post("/sessions", async (c) => { + if (!validateMediaServerSecret(c)) { + return c.json({ error: "Unauthorized" }, 401); + } + + const result = startLiveSessionSchema.safeParse( + await c.req.json().catch(() => null), + ); + if (!result.success) { + return c.json( + { + error: "Invalid request", + code: "INVALID_REQUEST", + details: result.error.message, + }, + 400, + ); + } + + if (isDraining() || !canAcceptLiveSession()) { + c.header("Retry-After", DRAIN_RETRY_AFTER_SECONDS.toString()); + return c.json({ error: "Server is busy", code: "SERVER_BUSY" }, 503); + } + + startLiveSession(result.data); + return c.json({ accepted: true }, 202); +}); + +export default liveTranscribe; From 48bc2c6a54f42ca2c047b4bf930c6599ada4a05c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:34:09 +0000 Subject: [PATCH 04/11] feat(web): run live transcription on the media server behind LIVE_TRANSCRIBE_RUNNER --- .../media-server/live-transcribe/route.ts | 232 +++++++++++++ apps/web/lib/live-transcribe-farm.ts | 310 ++++++++++++++++++ apps/web/lib/live-transcribe.ts | 41 ++- packages/env/server.ts | 6 + 4 files changed, 588 insertions(+), 1 deletion(-) create mode 100644 apps/web/app/api/webhooks/media-server/live-transcribe/route.ts create mode 100644 apps/web/lib/live-transcribe-farm.ts diff --git a/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts new file mode 100644 index 0000000000..4290811c41 --- /dev/null +++ b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts @@ -0,0 +1,232 @@ +import { createHash, timingSafeEqual } from "node:crypto"; +import { serverEnv } from "@cap/env"; +import { Video } from "@cap/web-domain"; +import { Either, Option, Schema } from "effect"; +import { type NextRequest, NextResponse } from "next/server"; +import { z } from "zod"; +import { dispatchLiveTranscription } from "@/lib/live-transcribe"; +import { + describeLiveChunk, + LIVE_TRANSCRIPT_NO_SEGMENTS, +} from "@/lib/live-transcribe-core"; +import { + buildLiveSessionState, + getFarmLiveBucket, + getLiveSession, + liveTranscriptionEnded, + loadLiveVideo, +} from "@/lib/live-transcribe-farm"; +import { + finishLiveTranscription, + type LiveBucket, + promoteLiveTranscript, + readLiveTranscript, + touchLiveClaim, + writeLiveChunk, + writeLiveChunkGap, +} from "@/lib/live-transcribe-runner"; +import { runWorkflowPromise } from "@/lib/workflow-runtime"; + +export const maxDuration = 120; + +const MAX_CHUNK_WORDS = 20_000; + +const wordSchema = z.object({ + text: z.string().max(200), + start: z.number(), + end: z.number(), + confidence: z.number().nullish(), + speaker: z.string().max(40).nullish(), +}); + +const reportSchema = z.discriminatedUnion("action", [ + z.object({ + action: z.literal("chunk"), + videoId: z.string(), + session: z.string(), + lastProcessedIndex: z.number().int(), + lastIndex: z.number().int(), + contextFromIndex: z.number().int().nullable(), + words: z.array(wordSchema).max(MAX_CHUNK_WORDS), + languageCode: z.string().max(16).nullable(), + }), + z.object({ + action: z.literal("gap"), + videoId: z.string(), + session: z.string(), + lastProcessedIndex: z.number().int(), + lastIndex: z.number().int(), + }), + z.object({ + action: z.literal("sync"), + videoId: z.string(), + session: z.string(), + }), + z.object({ + action: z.literal("finish"), + videoId: z.string(), + session: z.string(), + outcome: z.string().max(40), + }), + z.object({ + action: z.literal("release"), + videoId: z.string(), + session: z.string(), + }), +]); + +const stop = () => NextResponse.json({ continue: false }); + +function authorized(request: NextRequest) { + const secret = serverEnv().MEDIA_SERVER_WEBHOOK_SECRET; + const header = request.headers.get("x-media-server-secret"); + const digest = (value: string) => + createHash("sha256").update(value, "utf8").digest(); + return ( + Boolean(secret) && + Boolean(header) && + timingSafeEqual(digest(header ?? ""), digest(secret ?? "")) + ); +} + +async function readManifest( + bucket: LiveBucket, + ownerId: string, + videoId: string, +) { + const source = new Video.SegmentsSource({ videoId, ownerId }); + const content = await bucket + .getObject(source.getManifestKey()) + .pipe(runWorkflowPromise) + .catch(() => Option.none()); + if (Option.isNone(content)) return null; + try { + const decoded = Schema.decodeUnknownEither(Video.SegmentManifest)( + JSON.parse(content.value), + ); + return Either.isRight(decoded) ? decoded.right : null; + } catch { + return null; + } +} + +/** + * Reports from a media server live transcription session (see + * lib/live-transcribe-farm.ts). Only the session named in the claim is + * heard. Every answer carries the authoritative cursor and fresh signed URLs, + * or `continue: false` when the session should end. + */ +export async function POST(request: NextRequest) { + if (!authorized(request)) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const parsed = reportSchema.safeParse(await request.json().catch(() => null)); + if (!parsed.success) { + return NextResponse.json({ error: "Invalid report" }, { status: 400 }); + } + const report = parsed.data; + const { videoId, session } = report; + + const row = await loadLiveVideo(videoId); + if (!row || getLiveSession(row.video) !== session) return stop(); + const { video } = row; + + if (report.action === "release") { + await dispatchLiveTranscription(video.id, video.ownerId, session); + return stop(); + } + + if (report.action === "finish") { + if (report.outcome === "done") { + const promotion = await promoteLiveTranscript(videoId, video.ownerId); + if (promotion.promoted) return stop(); + console.warn( + `[liveTranscribe] Promotion declined for ${videoId}: ${promotion.reason}`, + ); + } + await finishLiveTranscription( + videoId, + video.ownerId, + report.outcome === "done" ? "complete" : "stopped", + ); + console.log(`[liveTranscribe] ${videoId} media server ${report.outcome}`); + return stop(); + } + + if (liveTranscriptionEnded(row)) { + await finishLiveTranscription(videoId, video.ownerId, "stopped"); + return stop(); + } + + const bucket = await getFarmLiveBucket(video); + if (!bucket) { + await finishLiveTranscription(videoId, video.ownerId, "stopped"); + return stop(); + } + let artifact = await readLiveTranscript(bucket, video.ownerId, videoId); + const cursor = artifact?.lastAudioSegmentIndex ?? LIVE_TRANSCRIPT_NO_SEGMENTS; + + if (report.action === "sync") { + await touchLiveClaim(videoId, session); + } else if (report.lastProcessedIndex === cursor) { + const manifest = await readManifest(bucket, video.ownerId, videoId); + const chunk = manifest + ? describeLiveChunk({ + manifest, + lastProcessedIndex: report.lastProcessedIndex, + lastIndex: report.lastIndex, + contextFromIndex: + report.action === "chunk" ? report.contextFromIndex : null, + }) + : null; + + if (chunk && report.action === "gap") { + console.warn( + `[liveTranscribe] Skipping poison chunk ${cursor + 1}..${report.lastIndex} for ${videoId}`, + ); + artifact = await writeLiveChunkGap({ + bucket, + ownerId: video.ownerId, + videoId, + lastAudioSegmentIndex: report.lastIndex, + endMs: chunk.startMs + chunk.durationMs, + session, + }); + } else if (chunk && report.action === "chunk") { + const updated = await writeLiveChunk({ + bucket, + ownerId: video.ownerId, + videoId, + chunk: { + startMs: chunk.startMs, + durationMs: chunk.durationMs, + contextMs: chunk.contextMs, + lastAudioSegmentIndex: report.lastIndex, + }, + words: report.words, + languageCode: report.languageCode, + session, + }); + if (!updated) return stop(); + artifact = updated; + console.log( + `[liveTranscribe] ${videoId} chunk ${cursor + 1}..${report.lastIndex} (${chunk.durationMs}ms + ${chunk.contextMs}ms context, ${report.words.length} words) via media server`, + ); + } else { + console.warn( + `[liveTranscribe] ${videoId} report for ${report.lastProcessedIndex + 1}..${report.lastIndex} does not match the manifest`, + ); + } + } + + return NextResponse.json({ + continue: true, + state: await buildLiveSessionState({ + bucket, + video, + orgSettings: row.orgSettings, + artifact, + }), + }); +} diff --git a/apps/web/lib/live-transcribe-farm.ts b/apps/web/lib/live-transcribe-farm.ts new file mode 100644 index 0000000000..2c6a9209af --- /dev/null +++ b/apps/web/lib/live-transcribe-farm.ts @@ -0,0 +1,310 @@ +import { randomUUID } from "node:crypto"; +import { db } from "@cap/database"; +import { organizations, videos } from "@cap/database/schema"; +import { serverEnv } from "@cap/env"; +import { Storage } from "@cap/web-backend/src/Storage/index"; +import { + type AiGenerationLanguage, + parseAiGenerationLanguage, + Video, +} from "@cap/web-domain"; +import { and, eq, sql } from "drizzle-orm"; +import { Option } from "effect"; +import { + ASSEMBLYAI_SUPPORTED_LANGUAGES, + getAssemblyAITranscriptionOptions, +} from "@/lib/assemblyai"; +import { + LIVE_TRANSCRIPT_NO_SEGMENTS, + type LiveTranscriptArtifact, +} from "@/lib/live-transcribe-core"; +import { + isCanonicalTranscriptDone, + type LiveBucket, + liveContextSeconds, + liveSessionMatches, + readLiveTranscript, +} from "@/lib/live-transcribe-runner"; +import { decodeStorageVideo } from "@/lib/video-storage"; +import { runWorkflowPromise } from "@/lib/workflow-runtime"; + +/** + * Live transcription on the media server. The server holds one session per + * recording: it watches the segment manifest, sends new audio to the + * transcription API, and reports each chunk's words to + * /api/webhooks/media-server/live-transcribe, which owns the artifact, the + * claim, and promotion. Every report answers with fresh state, so the server + * never needs storage credentials or database access. + */ + +/** Segments signed behind the cursor, enough for the context audio. */ +const SIGNED_SEGMENTS_BEHIND = 16; +/** Segments signed ahead of the cursor, more than one chunk can take. */ +const SIGNED_SEGMENTS_AHEAD = 40; +const SIGNED_URL_EXPIRES_SECONDS = 60 * 60; +const START_TIMEOUT_MS = 5_000; +const START_ATTEMPTS = 2; + +/** Stored in the claim instead of a session id once the workflow runs it, + * which also fences off any media server session that may have started. */ +export const WORKFLOW_LIVE_SESSION = "workflow"; + +export type LiveSessionState = { + lastAudioSegmentIndex: number; + transcribedDurationMs: number; + transcriptionOptions: Record; + urls: { + manifest: string; + audioInit: string; + segments: Record; + }; +}; + +type LiveVideoRow = { + video: typeof videos.$inferSelect; + orgSettings: (typeof organizations.$inferSelect)["settings"] | null; +}; + +export function isFarmLiveTranscriptionEnabled() { + const env = serverEnv(); + return ( + env.LIVE_TRANSCRIBE_RUNNER === "media-server" && + Boolean(env.MEDIA_SERVER_URL) && + Boolean(env.MEDIA_SERVER_WEBHOOK_SECRET) && + Boolean(env.ASSEMBLY_API_KEY) + ); +} + +export async function loadLiveVideo( + videoId: string, +): Promise { + const [row] = await db() + .select({ video: videos, orgSettings: organizations.settings }) + .from(videos) + .leftJoin(organizations, eq(videos.orgId, organizations.id)) + .where(eq(videos.id, videoId as Video.VideoId)); + return row ?? null; +} + +export function getLiveSession(video: typeof videos.$inferSelect) { + const live = video.metadata?.liveTranscript as + | { status?: string; session?: string } + | undefined; + return live?.status === "active" ? (live.session ?? null) : null; +} + +/** Why the recording can't be live-transcribed any more, if it can't. */ +export function liveTranscriptionEnded(row: LiveVideoRow): string | null { + if (row.video.source.type !== "desktopSegments") { + return "not a segmented recording"; + } + if ( + row.video.settings?.disableTranscript ?? + row.orgSettings?.disableTranscript + ) { + return "transcription disabled"; + } + if (isCanonicalTranscriptDone(row.video.transcriptionStatus)) { + return "canonical transcript already exists"; + } + return null; +} + +export async function getFarmLiveBucket( + video: typeof videos.$inferSelect, +): Promise { + const [bucket, customBucket] = await Storage.getAccessForVideo( + decodeStorageVideo(video), + ).pipe(runWorkflowPromise); + // Signed URLs for Cap's own bucket only; Drive and customer buckets stay + // on the workflow, which reads them through their own access. + return bucket.provider === "s3" && Option.isNone(customBucket) + ? bucket + : null; +} + +export async function buildLiveSessionState(options: { + bucket: LiveBucket; + video: typeof videos.$inferSelect; + orgSettings: LiveVideoRow["orgSettings"]; + artifact: LiveTranscriptArtifact | null; +}): Promise { + const { bucket, video, artifact } = options; + const source = new Video.SegmentsSource({ + videoId: video.id, + ownerId: video.ownerId, + }); + const lastAudioSegmentIndex = + artifact?.lastAudioSegmentIndex ?? LIVE_TRANSCRIPT_NO_SEGMENTS; + const sign = (key: string) => + bucket + .getInternalSignedObjectUrl(key, { + expiresIn: SIGNED_URL_EXPIRES_SECONDS, + }) + .pipe(runWorkflowPromise); + + const first = Math.max(0, lastAudioSegmentIndex + 1 - SIGNED_SEGMENTS_BEHIND); + const last = lastAudioSegmentIndex + SIGNED_SEGMENTS_AHEAD; + const indices = Array.from( + { length: last - first + 1 }, + (_, offset) => first + offset, + ); + const [manifest, audioInit, ...segmentUrls] = await Promise.all([ + sign(source.getManifestKey()), + sign(source.getAudioInitKey()), + ...indices.map((index) => sign(source.getAudioSegmentKey(index))), + ]); + const segments: Record = {}; + for (const [offset, index] of indices.entries()) { + const url = segmentUrls[offset]; + if (url) segments[String(index)] = url; + } + + const preferred = + artifact?.languageCode ?? + parseAiGenerationLanguage(options.orgSettings?.aiGenerationLanguage); + const language = ( + ASSEMBLYAI_SUPPORTED_LANGUAGES as readonly string[] + ).includes(preferred) + ? (preferred as AiGenerationLanguage) + : ("auto" as AiGenerationLanguage); + + return { + lastAudioSegmentIndex, + transcribedDurationMs: artifact?.transcribedDurationMs ?? 0, + transcriptionOptions: getAssemblyAITranscriptionOptions(language), + urls: { manifest: manifest ?? "", audioInit: audioInit ?? "", segments }, + }; +} + +const getAffectedRows = (result: unknown) => { + if (Array.isArray(result)) { + return ( + (result[0] as { affectedRows?: number } | undefined)?.affectedRows ?? 0 + ); + } + return (result as { affectedRows?: number } | undefined)?.affectedRows ?? 0; +}; + +/** + * Move the active claim from `from` (null: no session yet) to `to`. Exactly + * one caller wins, so two runners can never both own a recording. + */ +export async function rotateLiveSession( + videoId: string, + from: string | null, + to: string, +): Promise { + const result = await db() + .update(videos) + .set({ + metadata: sql`JSON_SET(COALESCE(${videos.metadata}, JSON_OBJECT()), '$.liveTranscript.session', ${to}, '$.liveTranscript.updatedAt', ${new Date().toISOString()})`, + updatedAt: sql`${videos.updatedAt}`, + }) + .where( + and( + eq(videos.id, videoId as Video.VideoId), + sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.status')) = 'active'`, + from === null + ? sql`JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.session') IS NULL` + : liveSessionMatches(from), + ), + ); + return getAffectedRows(result) > 0; +} + +export type FarmLiveStart = + | { status: "started" } + /** Another caller moved the claim first; it owns the recording now. */ + | { status: "lost" } + /** Not handed over. `session` is the claim's session now: a declined + * start after rotation leaves a fresh id that no server holds, so the + * workflow can take over without racing a server that accepted after + * the timeout. */ + | { status: "declined"; session: string | null }; + +/** Hand an active live transcription to the media server. */ +export async function startFarmLiveTranscription( + videoId: string, + from: string | null, +): Promise { + if (!isFarmLiveTranscriptionEnabled()) { + return { status: "declined", session: from }; + } + + const row = await loadLiveVideo(videoId); + if (!row || liveTranscriptionEnded(row)) { + return { status: "declined", session: from }; + } + const bucket = await getFarmLiveBucket(row.video); + if (!bucket) return { status: "declined", session: from }; + + const session = randomUUID(); + if (!(await rotateLiveSession(videoId, from, session))) { + return { status: "lost" }; + } + + const env = serverEnv(); + try { + const artifact = await readLiveTranscript( + bucket, + row.video.ownerId, + videoId, + ); + const state = await buildLiveSessionState({ + bucket, + video: row.video, + orgSettings: row.orgSettings, + artifact, + }); + const webhookBaseUrl = env.MEDIA_SERVER_WEBHOOK_URL || env.WEB_URL; + const body = JSON.stringify({ + videoId, + session, + callbackUrl: new URL( + "/api/webhooks/media-server/live-transcribe", + webhookBaseUrl, + ).toString(), + asr: { + baseUrl: env.ASSEMBLY_BASE_URL || "https://api.assemblyai.com", + apiKey: env.ASSEMBLY_API_KEY, + }, + contextSeconds: liveContextSeconds(), + state, + }); + + for (let attempt = 0; attempt < START_ATTEMPTS; attempt++) { + try { + const response = await fetch( + `${(env.MEDIA_SERVER_URL as string).replace(/\/$/, "")}/live-transcribe/sessions`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-media-server-secret": + env.MEDIA_SERVER_WEBHOOK_SECRET as string, + }, + body, + signal: AbortSignal.timeout(START_TIMEOUT_MS), + }, + ); + await response.body?.cancel().catch(() => {}); + if (response.ok) return { status: "started" }; + // A busy or draining server refused it, so another try under the + // same session can't duplicate it; the load balancer usually + // picks another server. + if (response.status !== 503) break; + } catch { + // A timed-out start may still have been accepted: never resend it. + break; + } + } + } catch (error) { + console.warn( + `[liveTranscribe] Media server start failed for ${videoId}`, + error, + ); + } + + return { status: "declined", session }; +} diff --git a/apps/web/lib/live-transcribe.ts b/apps/web/lib/live-transcribe.ts index 5759d39415..56a01276a1 100644 --- a/apps/web/lib/live-transcribe.ts +++ b/apps/web/lib/live-transcribe.ts @@ -4,6 +4,12 @@ import { serverEnv } from "@cap/env"; import type { Organisation, User, Video } from "@cap/web-domain"; import { and, eq, sql } from "drizzle-orm"; import { start } from "workflow/api"; +import { + type FarmLiveStart, + rotateLiveSession, + startFarmLiveTranscription, + WORKFLOW_LIVE_SESSION, +} from "@/lib/live-transcribe-farm"; import { liveTranscribeWorkflow } from "@/workflows/live-transcribe"; const getAffectedRows = (result: unknown) => { @@ -65,7 +71,7 @@ export async function maybeStartLiveTranscription({ } try { - await start(liveTranscribeWorkflow, [{ videoId, userId: ownerId }]); + await dispatchLiveTranscription(videoId, ownerId, null); } catch (error) { // Release the claim so a retried create can start the workflow. await db() @@ -84,3 +90,36 @@ export async function maybeStartLiveTranscription({ return "skipped"; } } + +/** + * Run an active live transcription on the media server when that's enabled + * and a server takes it, otherwise on the workflow, which resumes from the + * artifact. `from` is the claim's current session (null before the first). + */ +export async function dispatchLiveTranscription( + videoId: Video.VideoId, + ownerId: User.UserId, + from: string | null, +): Promise<"media-server" | "workflow" | "lost"> { + let farm: FarmLiveStart; + try { + farm = await startFarmLiveTranscription(videoId, from); + } catch (error) { + console.warn( + `[dispatchLiveTranscription] Media server path failed for ${videoId}`, + error, + ); + farm = { status: "declined", session: from }; + } + if (farm.status === "started") return "media-server"; + if (farm.status === "lost") return "lost"; + if ( + farm.session !== null && + !(await rotateLiveSession(videoId, farm.session, WORKFLOW_LIVE_SESSION)) + ) { + return "lost"; + } + + await start(liveTranscribeWorkflow, [{ videoId, userId: ownerId }]); + return "workflow"; +} diff --git a/packages/env/server.ts b/packages/env/server.ts index 24f4798c62..8f8376b670 100644 --- a/packages/env/server.ts +++ b/packages/env/server.ts @@ -142,6 +142,12 @@ function createServerEnv() { .describe( "Seconds of earlier audio sent with each instant-mode live transcription chunk (more accurate, more audio billed; default 0)", ), + LIVE_TRANSCRIBE_RUNNER: z + .string() + .optional() + .describe( + "Set to media-server to run instant-mode live transcription on the media server instead of a workflow (falls back to the workflow)", + ), ANTHROPIC_API_KEY: z.string().optional().describe("AI chat"), OPENAI_API_KEY: z.string().optional().describe("AI summaries"), GROQ_API_KEY: z.string().optional().describe("AI summaries"), From 4216200939d50937bfe31eda7fff3b7b39c7cfc4 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:49:09 +0000 Subject: [PATCH 05/11] feat(web-backend): read S3 objects with their ETag and allow conditional puts --- .../src/S3Buckets/S3BucketAccess.ts | 28 ++++++++++++++++++- packages/web-backend/src/Storage/index.ts | 2 ++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/packages/web-backend/src/S3Buckets/S3BucketAccess.ts b/packages/web-backend/src/S3Buckets/S3BucketAccess.ts index 7c0b1f5f22..09bfd7073c 100644 --- a/packages/web-backend/src/S3Buckets/S3BucketAccess.ts +++ b/packages/web-backend/src/S3Buckets/S3BucketAccess.ts @@ -295,10 +295,34 @@ export const createS3BucketAccess = Effect.gen(function* () { ), ), ), + getObjectWithETag: (key: string) => + wrapS3Promise( + provider.getInternal.pipe( + Effect.map(async (client) => { + const response = await client + .send( + new S3.GetObjectCommand({ Bucket: provider.bucket, Key: key }), + ) + .catch((e) => { + if (e instanceof S3.NoSuchKey) return null; + throw e; + }); + const body = await response?.Body?.transformToString(); + return body === undefined + ? Option.none() + : Option.some({ body, eTag: response?.ETag }); + }), + ), + ), putObject: ( key: string, body: string | Uint8Array | ArrayBuffer | Stream.Stream, - fields?: { contentType?: string; contentLength?: number }, + fields?: { + contentType?: string; + contentLength?: number; + ifMatch?: string; + ifNoneMatch?: string; + }, ) => wrapS3Promise( provider.getInternal.pipe( @@ -325,6 +349,8 @@ export const createS3BucketAccess = Effect.gen(function* () { Body: _body, ContentType: fields?.contentType, ContentLength: fields?.contentLength, + IfMatch: fields?.ifMatch, + IfNoneMatch: fields?.ifNoneMatch, }), ); }), diff --git a/packages/web-backend/src/Storage/index.ts b/packages/web-backend/src/Storage/index.ts index 6aae3c57b6..aa6c128fe0 100644 --- a/packages/web-backend/src/Storage/index.ts +++ b/packages/web-backend/src/Storage/index.ts @@ -434,6 +434,8 @@ const makeS3Access = (s3: S3BucketAccess) => ({ signingArgs?: Parameters[1], ) => mapStorageError(s3.getInternalSignedObjectUrl(key, signingArgs)), getObject: (key: string) => mapStorageError(s3.getObject(key)), + getObjectWithETag: (key: string) => + mapStorageError(s3.getObjectWithETag(key)), listObjects: (input: { prefix?: string; maxKeys?: number; From 077b07c94c81f915d73251e8a048f6de1814dd9e Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:49:09 +0000 Subject: [PATCH 06/11] fix(web): reject live transcript writes from outdated media server reports --- .../media-server/live-transcribe/route.ts | 49 +++-- apps/web/lib/live-transcribe-runner.ts | 193 ++++++++++++------ apps/web/workflows/live-transcribe.ts | 8 +- 3 files changed, 163 insertions(+), 87 deletions(-) diff --git a/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts index 4290811c41..fde24afd08 100644 --- a/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts +++ b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts @@ -7,7 +7,7 @@ import { z } from "zod"; import { dispatchLiveTranscription } from "@/lib/live-transcribe"; import { describeLiveChunk, - LIVE_TRANSCRIPT_NO_SEGMENTS, + type LiveTranscriptArtifact, } from "@/lib/live-transcribe-core"; import { buildLiveSessionState, @@ -19,6 +19,7 @@ import { import { finishLiveTranscription, type LiveBucket, + type LiveTranscriptUpdate, promoteLiveTranscript, readLiveTranscript, touchLiveClaim, @@ -164,12 +165,10 @@ export async function POST(request: NextRequest) { await finishLiveTranscription(videoId, video.ownerId, "stopped"); return stop(); } - let artifact = await readLiveTranscript(bucket, video.ownerId, videoId); - const cursor = artifact?.lastAudioSegmentIndex ?? LIVE_TRANSCRIPT_NO_SEGMENTS; - + let artifact: LiveTranscriptArtifact | null | undefined; if (report.action === "sync") { await touchLiveClaim(videoId, session); - } else if (report.lastProcessedIndex === cursor) { + } else { const manifest = await readManifest(bucket, video.ownerId, videoId); const chunk = manifest ? describeLiveChunk({ @@ -180,24 +179,28 @@ export async function POST(request: NextRequest) { report.action === "chunk" ? report.contextFromIndex : null, }) : null; + const range = `${report.lastProcessedIndex + 1}..${report.lastIndex}`; + const target = { + bucket, + ownerId: video.ownerId, + videoId, + session, + expectedCursor: report.lastProcessedIndex, + }; + let write: LiveTranscriptUpdate | null = null; if (chunk && report.action === "gap") { console.warn( - `[liveTranscribe] Skipping poison chunk ${cursor + 1}..${report.lastIndex} for ${videoId}`, + `[liveTranscribe] Skipping poison chunk ${range} for ${videoId}`, ); - artifact = await writeLiveChunkGap({ - bucket, - ownerId: video.ownerId, - videoId, + write = await writeLiveChunkGap({ + ...target, lastAudioSegmentIndex: report.lastIndex, endMs: chunk.startMs + chunk.durationMs, - session, }); } else if (chunk && report.action === "chunk") { - const updated = await writeLiveChunk({ - bucket, - ownerId: video.ownerId, - videoId, + write = await writeLiveChunk({ + ...target, chunk: { startMs: chunk.startMs, durationMs: chunk.durationMs, @@ -206,20 +209,22 @@ export async function POST(request: NextRequest) { }, words: report.words, languageCode: report.languageCode, - session, }); - if (!updated) return stop(); - artifact = updated; - console.log( - `[liveTranscribe] ${videoId} chunk ${cursor + 1}..${report.lastIndex} (${chunk.durationMs}ms + ${chunk.contextMs}ms context, ${report.words.length} words) via media server`, - ); + if (write.status === "written") { + console.log( + `[liveTranscribe] ${videoId} chunk ${range} (${chunk.durationMs}ms + ${chunk.contextMs}ms context, ${report.words.length} words) via media server`, + ); + } } else { console.warn( - `[liveTranscribe] ${videoId} report for ${report.lastProcessedIndex + 1}..${report.lastIndex} does not match the manifest`, + `[liveTranscribe] ${videoId} report for ${range} does not match the manifest`, ); } + if (write?.status === "stopped") return stop(); + if (write?.status === "written") artifact = write.artifact; } + artifact ??= await readLiveTranscript(bucket, video.ownerId, videoId); return NextResponse.json({ continue: true, state: await buildLiveSessionState({ diff --git a/apps/web/lib/live-transcribe-runner.ts b/apps/web/lib/live-transcribe-runner.ts index a725322912..d296be712d 100644 --- a/apps/web/lib/live-transcribe-runner.ts +++ b/apps/web/lib/live-transcribe-runner.ts @@ -66,24 +66,135 @@ export async function readLiveTranscript( return Option.isSome(existing) ? parseLiveTranscript(existing.value) : null; } +export type LiveTranscriptUpdate = + | { status: "written"; artifact: LiveTranscriptArtifact } + /** The canonical transcript finished (writing would re-create the artifact + * it just deleted, an orphan billed forever), or the claim moved to + * another session. */ + | { status: "stopped" } + /** The artifact isn't at `expectedCursor`, or changed after it was read. */ + | { status: "conflict" }; + +function isRejectedConditionalWrite(error: unknown) { + const pending = [error]; + for (let index = 0; index < pending.length && index < 16; index++) { + const value = pending[index]; + if (typeof value !== "object" || value === null) continue; + const record = value as Record; + const status = (record.$metadata as { httpStatusCode?: unknown } | null) + ?.httpStatusCode; + if ( + status === 412 || + status === 409 || + record.name === "PreconditionFailed" || + record.name === "ConditionalRequestConflict" + ) { + return true; + } + pending.push(record.cause, record.error); + } + return false; +} + +/** + * Read-modify-write of `transcription.live.json`. A session runner's reports + * can overlap (a retry can land while a stalled original is still running), + * so its writes are conditional on the copy it read: the stalled original + * then fails instead of putting older words back. Session runners only use + * Cap's own bucket; the workflow writes unconditionally as before, since + * customer S3-compatible buckets may not support conditional writes. + */ +async function updateLiveTranscript( + options: { + bucket: LiveBucket; + ownerId: string; + videoId: string; + session?: string; + expectedCursor?: number; + }, + update: (artifact: LiveTranscriptArtifact) => LiveTranscriptArtifact, +): Promise { + const { bucket, ownerId, videoId, session } = options; + const key = getLiveTranscriptObjectKey(ownerId, videoId); + + let artifact: LiveTranscriptArtifact | null; + let condition: { ifMatch?: string; ifNoneMatch?: string } = {}; + if (session !== undefined && bucket.provider === "s3") { + const existing = await bucket + .getObjectWithETag(key) + .pipe(runWorkflowPromise); + artifact = Option.isSome(existing) + ? parseLiveTranscript(existing.value.body) + : null; + if (Option.isSome(existing) && !existing.value.eTag) { + return { status: "conflict" }; + } + condition = Option.isSome(existing) + ? { ifMatch: existing.value.eTag } + : { ifNoneMatch: "*" }; + } else { + artifact = await readLiveTranscript(bucket, ownerId, videoId); + } + + const current = + artifact ?? createEmptyLiveTranscript(new Date().toISOString()); + if ( + options.expectedCursor !== undefined && + current.lastAudioSegmentIndex !== options.expectedCursor + ) { + return { status: "conflict" }; + } + const updated = update(current); + + const [row] = await db() + .select({ + transcriptionStatus: videos.transcriptionStatus, + session: sql< + string | null + >`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.session'))`, + }) + .from(videos) + .where(eq(videos.id, videoId as Video.VideoId)); + if ( + isCanonicalTranscriptDone(row?.transcriptionStatus) || + (session !== undefined && row?.session !== session) + ) { + return { status: "stopped" }; + } + + try { + await bucket + .putObject(key, JSON.stringify(updated), { + contentType: "application/json", + ...condition, + }) + .pipe(runWorkflowPromise); + } catch (error) { + if (session !== undefined && isRejectedConditionalWrite(error)) { + return { status: "conflict" }; + } + throw error; + } + + await touchLiveClaim(videoId, session); + return { status: "written", artifact: updated }; +} + /** * Record that the audio up to `lastAudioSegmentIndex` was skipped. Must land * before the cursor moves past it, or a later promotion could ship a * transcript that silently misses speech. Throws when it can't be recorded. */ -export async function writeLiveChunkGap(options: { +export function writeLiveChunkGap(options: { bucket: LiveBucket; ownerId: string; videoId: string; lastAudioSegmentIndex: number; endMs: number; session?: string; -}): Promise { - const { bucket, ownerId, videoId } = options; - const artifact = - (await readLiveTranscript(bucket, ownerId, videoId)) ?? - createEmptyLiveTranscript(new Date().toISOString()); - const updated: LiveTranscriptArtifact = { + expectedCursor?: number; +}): Promise { + return updateLiveTranscript(options, (artifact) => ({ ...artifact, hasGaps: true, lastAudioSegmentIndex: Math.max( @@ -95,25 +206,10 @@ export async function writeLiveChunkGap(options: { options.endMs, ), updatedAt: new Date().toISOString(), - }; - await bucket - .putObject( - getLiveTranscriptObjectKey(ownerId, videoId), - JSON.stringify(updated), - { contentType: "application/json" }, - ) - .pipe(runWorkflowPromise); - await touchLiveClaim(videoId, options.session); - return updated; + })); } -/** - * Merge one transcribed chunk into `transcription.live.json`. Returns null - * without writing when the canonical transcript finished meanwhile (the write - * would re-create the artifact it just deleted, an orphan billed forever) or, - * for a session runner, when the claim moved to another session. - */ -export async function writeLiveChunk(options: { +export function writeLiveChunk(options: { bucket: LiveBucket; ownerId: string; videoId: string; @@ -126,50 +222,23 @@ export async function writeLiveChunk(options: { words: readonly LiveChunkWordInput[] | null | undefined; languageCode: string | null; session?: string; -}): Promise { - const { bucket, ownerId, videoId, chunk, session } = options; + expectedCursor?: number; +}): Promise { + const { chunk } = options; const words = offsetChunkWords( options.words, chunk.startMs, chunk.durationMs, chunk.contextMs, ); - const artifact = - (await readLiveTranscript(bucket, ownerId, videoId)) ?? - createEmptyLiveTranscript(new Date().toISOString()); - const updated = applyChunkToLiveTranscript(artifact, { - ...chunk, - words, - languageCode: options.languageCode, - nowIso: new Date().toISOString(), - }); - - const [current] = await db() - .select({ - transcriptionStatus: videos.transcriptionStatus, - session: sql< - string | null - >`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.liveTranscript.session'))`, - }) - .from(videos) - .where(eq(videos.id, videoId as Video.VideoId)); - if ( - isCanonicalTranscriptDone(current?.transcriptionStatus) || - (session !== undefined && current?.session !== session) - ) { - return null; - } - - await bucket - .putObject( - getLiveTranscriptObjectKey(ownerId, videoId), - JSON.stringify(updated), - { contentType: "application/json" }, - ) - .pipe(runWorkflowPromise); - - await touchLiveClaim(videoId, session); - return updated; + return updateLiveTranscript(options, (artifact) => + applyChunkToLiveTranscript(artifact, { + ...chunk, + words, + languageCode: options.languageCode, + nowIso: new Date().toISOString(), + }), + ); } export function liveContextSeconds() { diff --git a/apps/web/workflows/live-transcribe.ts b/apps/web/workflows/live-transcribe.ts index a37ec08ab6..5c69e39c68 100644 --- a/apps/web/workflows/live-transcribe.ts +++ b/apps/web/workflows/live-transcribe.ts @@ -366,13 +366,14 @@ async function processNextLiveChunk(options: { // The gap MUST be durably recorded before the cursor advances, or a // later promotion could ship a transcript that silently misses speech. try { - await writeLiveChunkGap({ + const gap = await writeLiveChunkGap({ bucket, ownerId: video.ownerId, videoId, lastAudioSegmentIndex: chunkEndIndex, endMs: decision.startMs + decision.durationMs, }); + if (gap.status !== "written") return { outcome: "canonical-done" }; } catch (error) { return { outcome: "chunk-failed", @@ -427,7 +428,7 @@ async function processNextLiveChunk(options: { // Polling + the AssemblyAI round trip can take minutes, so the write // re-checks that the canonical pipeline didn't complete meanwhile. - const updated = await writeLiveChunk({ + const write = await writeLiveChunk({ bucket, ownerId: video.ownerId, videoId, @@ -443,7 +444,8 @@ async function processNextLiveChunk(options: { ? transcript.language_code : null, }); - if (!updated) return { outcome: "canonical-done" }; + if (write.status !== "written") return { outcome: "canonical-done" }; + const updated = write.artifact; // If this chunk completed full coverage of a finished recording, tell // the workflow to promote right away instead of paying another poll From a33d886a172c6655a2556dcfb2c1e4957ccb7373 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:49:09 +0000 Subject: [PATCH 07/11] chore: drop comments that restate the live transcription code --- apps/media-server/src/lib/live-transcribe.ts | 10 ++-------- apps/web/lib/live-transcribe-farm.ts | 2 -- 2 files changed, 2 insertions(+), 10 deletions(-) diff --git a/apps/media-server/src/lib/live-transcribe.ts b/apps/media-server/src/lib/live-transcribe.ts index 9bb9e4ee7b..d8176a1134 100644 --- a/apps/media-server/src/lib/live-transcribe.ts +++ b/apps/media-server/src/lib/live-transcribe.ts @@ -332,7 +332,6 @@ export class LiveSession { private releasing = false; private manifest: { etag: string | null; value: LiveManifest } | null = null; private initAudio: Uint8Array | null = null; - /** Recently sent segments, so context audio isn't downloaded twice. */ private readonly segmentCache = new Map(); private lastReportAt: number; private readonly sleep: (ms: number, signal?: AbortSignal) => Promise; @@ -354,12 +353,10 @@ export class LiveSession { return this.done; } - /** Stop without reporting; the web app already moved on. */ cancel() { this.abort.abort(); } - /** Stop and ask the web app to run the recording elsewhere. */ release() { this.releasing = true; this.abort.abort(); @@ -586,7 +583,6 @@ export class LiveSession { offset += part.byteLength; } - // Later chunks only reuse the segments just sent as context. const keepFrom = (entries[entries.length - 1]?.index ?? 0) - 64; for (const index of this.segmentCache.keys()) { if (index < keepFrom) this.segmentCache.delete(index); @@ -645,8 +641,8 @@ export function canAcceptLiveSession() { return sessions.size < LIVE.MAX_SESSIONS; } -/** Start a session, replacing any older one this server holds for the same - * recording (the web app has already fenced the older one off). */ +// An older session for the same recording was already fenced off by the web +// app when it rotated the claim, so it is cancelled without reporting. export function startLiveSession( request: StartLiveSession, deps?: LiveSessionDeps, @@ -664,8 +660,6 @@ export function startLiveSession( let releasing: Promise | null = null; -/** Hand every session back to the web app, which restarts each on another - * server (or its workflow). Called once when this server starts draining. */ export function releaseLiveSessions(): Promise { releasing ??= Promise.all( [...sessions.values()].map((session) => session.release()), diff --git a/apps/web/lib/live-transcribe-farm.ts b/apps/web/lib/live-transcribe-farm.ts index 2c6a9209af..f596e05117 100644 --- a/apps/web/lib/live-transcribe-farm.ts +++ b/apps/web/lib/live-transcribe-farm.ts @@ -93,7 +93,6 @@ export function getLiveSession(video: typeof videos.$inferSelect) { return live?.status === "active" ? (live.session ?? null) : null; } -/** Why the recording can't be live-transcribed any more, if it can't. */ export function liveTranscriptionEnded(row: LiveVideoRow): string | null { if (row.video.source.type !== "desktopSegments") { return "not a segmented recording"; @@ -223,7 +222,6 @@ export type FarmLiveStart = * the timeout. */ | { status: "declined"; session: string | null }; -/** Hand an active live transcription to the media server. */ export async function startFarmLiveTranscription( videoId: string, from: string | null, From 5f88a40e8db0f5c3efe94a3de286ac45295204e9 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:52:51 +0000 Subject: [PATCH 08/11] refactor(web): build the live transcription report webhook with HttpApi --- .../media-server/live-transcribe/route.ts | 193 +++++++++++------- 1 file changed, 117 insertions(+), 76 deletions(-) diff --git a/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts index fde24afd08..1a1ef0d9bf 100644 --- a/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts +++ b/apps/web/app/api/webhooks/media-server/live-transcribe/route.ts @@ -1,9 +1,15 @@ import { createHash, timingSafeEqual } from "node:crypto"; import { serverEnv } from "@cap/env"; import { Video } from "@cap/web-domain"; -import { Either, Option, Schema } from "effect"; -import { type NextRequest, NextResponse } from "next/server"; -import { z } from "zod"; +import { + HttpApi, + HttpApiBuilder, + HttpApiEndpoint, + HttpApiError, + HttpApiGroup, + HttpServerRequest, +} from "@effect/platform"; +import { Effect, Either, Layer, Option, Schema } from "effect"; import { dispatchLiveTranscription } from "@/lib/live-transcribe"; import { describeLiveChunk, @@ -26,68 +32,85 @@ import { writeLiveChunk, writeLiveChunkGap, } from "@/lib/live-transcribe-runner"; +import { apiToHandler } from "@/lib/server"; import { runWorkflowPromise } from "@/lib/workflow-runtime"; -export const maxDuration = 120; - const MAX_CHUNK_WORDS = 20_000; -const wordSchema = z.object({ - text: z.string().max(200), - start: z.number(), - end: z.number(), - confidence: z.number().nullish(), - speaker: z.string().max(40).nullish(), -}); - -const reportSchema = z.discriminatedUnion("action", [ - z.object({ - action: z.literal("chunk"), - videoId: z.string(), - session: z.string(), - lastProcessedIndex: z.number().int(), - lastIndex: z.number().int(), - contextFromIndex: z.number().int().nullable(), - words: z.array(wordSchema).max(MAX_CHUNK_WORDS), - languageCode: z.string().max(16).nullable(), - }), - z.object({ - action: z.literal("gap"), - videoId: z.string(), - session: z.string(), - lastProcessedIndex: z.number().int(), - lastIndex: z.number().int(), +const Session = { + videoId: Schema.String.pipe(Schema.maxLength(64)), + session: Schema.String.pipe(Schema.maxLength(64)), +}; +const SegmentIndex = Schema.Number.pipe(Schema.int()); + +const Report = Schema.Union( + Schema.Struct({ + action: Schema.Literal("chunk"), + ...Session, + lastProcessedIndex: SegmentIndex, + lastIndex: SegmentIndex, + contextFromIndex: Schema.NullOr(SegmentIndex), + words: Schema.Array( + Schema.Struct({ + text: Schema.String.pipe(Schema.maxLength(200)), + start: Schema.Number, + end: Schema.Number, + confidence: Schema.optional(Schema.NullOr(Schema.Number)), + speaker: Schema.optional( + Schema.NullOr(Schema.String.pipe(Schema.maxLength(40))), + ), + }), + ).pipe(Schema.maxItems(MAX_CHUNK_WORDS)), + languageCode: Schema.NullOr(Schema.String.pipe(Schema.maxLength(16))), }), - z.object({ - action: z.literal("sync"), - videoId: z.string(), - session: z.string(), + Schema.Struct({ + action: Schema.Literal("gap"), + ...Session, + lastProcessedIndex: SegmentIndex, + lastIndex: SegmentIndex, }), - z.object({ - action: z.literal("finish"), - videoId: z.string(), - session: z.string(), - outcome: z.string().max(40), + Schema.Struct({ action: Schema.Literal("sync"), ...Session }), + Schema.Struct({ + action: Schema.Literal("finish"), + ...Session, + outcome: Schema.String.pipe(Schema.maxLength(40)), }), - z.object({ - action: z.literal("release"), - videoId: z.string(), - session: z.string(), + Schema.Struct({ action: Schema.Literal("release"), ...Session }), +); + +const Answer = Schema.Union( + Schema.Struct({ continue: Schema.Literal(false) }), + Schema.Struct({ + continue: Schema.Literal(true), + state: Schema.Struct({ + lastAudioSegmentIndex: Schema.Number, + transcribedDurationMs: Schema.Number, + transcriptionOptions: Schema.Record({ + key: Schema.String, + value: Schema.Unknown, + }), + urls: Schema.Struct({ + manifest: Schema.String, + audioInit: Schema.String, + segments: Schema.Record({ key: Schema.String, value: Schema.String }), + }), + }), }), -]); - -const stop = () => NextResponse.json({ continue: false }); - -function authorized(request: NextRequest) { - const secret = serverEnv().MEDIA_SERVER_WEBHOOK_SECRET; - const header = request.headers.get("x-media-server-secret"); - const digest = (value: string) => - createHash("sha256").update(value, "utf8").digest(); - return ( - Boolean(secret) && - Boolean(header) && - timingSafeEqual(digest(header ?? ""), digest(secret ?? "")) - ); +); + +class Api extends HttpApi.make("LiveTranscribeReportApi").add( + HttpApiGroup.make("root").add( + HttpApiEndpoint.post("report")`/api/webhooks/media-server/live-transcribe` + .setPayload(Report) + .addSuccess(Answer) + .addError(HttpApiError.Unauthorized), + ), +) {} + +const stop = { continue: false } as const; + +function digest(value: string) { + return createHash("sha256").update(value, "utf8").digest(); } async function readManifest( @@ -117,31 +140,24 @@ async function readManifest( * heard. Every answer carries the authoritative cursor and fresh signed URLs, * or `continue: false` when the session should end. */ -export async function POST(request: NextRequest) { - if (!authorized(request)) { - return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); - } - - const parsed = reportSchema.safeParse(await request.json().catch(() => null)); - if (!parsed.success) { - return NextResponse.json({ error: "Invalid report" }, { status: 400 }); - } - const report = parsed.data; +async function handleReport( + report: typeof Report.Type, +): Promise { const { videoId, session } = report; const row = await loadLiveVideo(videoId); - if (!row || getLiveSession(row.video) !== session) return stop(); + if (!row || getLiveSession(row.video) !== session) return stop; const { video } = row; if (report.action === "release") { await dispatchLiveTranscription(video.id, video.ownerId, session); - return stop(); + return stop; } if (report.action === "finish") { if (report.outcome === "done") { const promotion = await promoteLiveTranscript(videoId, video.ownerId); - if (promotion.promoted) return stop(); + if (promotion.promoted) return stop; console.warn( `[liveTranscribe] Promotion declined for ${videoId}: ${promotion.reason}`, ); @@ -152,18 +168,18 @@ export async function POST(request: NextRequest) { report.outcome === "done" ? "complete" : "stopped", ); console.log(`[liveTranscribe] ${videoId} media server ${report.outcome}`); - return stop(); + return stop; } if (liveTranscriptionEnded(row)) { await finishLiveTranscription(videoId, video.ownerId, "stopped"); - return stop(); + return stop; } const bucket = await getFarmLiveBucket(video); if (!bucket) { await finishLiveTranscription(videoId, video.ownerId, "stopped"); - return stop(); + return stop; } let artifact: LiveTranscriptArtifact | null | undefined; if (report.action === "sync") { @@ -220,12 +236,12 @@ export async function POST(request: NextRequest) { `[liveTranscribe] ${videoId} report for ${range} does not match the manifest`, ); } - if (write?.status === "stopped") return stop(); + if (write?.status === "stopped") return stop; if (write?.status === "written") artifact = write.artifact; } artifact ??= await readLiveTranscript(bucket, video.ownerId, videoId); - return NextResponse.json({ + return { continue: true, state: await buildLiveSessionState({ bucket, @@ -233,5 +249,30 @@ export async function POST(request: NextRequest) { orgSettings: row.orgSettings, artifact, }), - }); + }; } + +const ApiLive = HttpApiBuilder.api(Api).pipe( + Layer.provide( + HttpApiBuilder.group(Api, "root", (handlers) => + handlers.handle("report", ({ payload }) => + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const secret = serverEnv().MEDIA_SERVER_WEBHOOK_SECRET; + const header = request.headers["x-media-server-secret"]; + if ( + !secret || + !header || + !timingSafeEqual(digest(header), digest(secret)) + ) + return yield* new HttpApiError.Unauthorized(); + return yield* Effect.promise(() => handleReport(payload)); + }), + ), + ), + ), +); + +export const POST = apiToHandler(ApiLive); +export const maxDuration = 120; +export const dynamic = "force-dynamic"; From f404709db15503899839cabb357c790327e8e290 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:59:39 +0000 Subject: [PATCH 09/11] fix(web): hand the live claim back when the workflow fails to start --- apps/web/lib/live-transcribe.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/apps/web/lib/live-transcribe.ts b/apps/web/lib/live-transcribe.ts index 56a01276a1..e5546dc25d 100644 --- a/apps/web/lib/live-transcribe.ts +++ b/apps/web/lib/live-transcribe.ts @@ -120,6 +120,17 @@ export async function dispatchLiveTranscription( return "lost"; } - await start(liveTranscribeWorkflow, [{ videoId, userId: ownerId }]); + try { + await start(liveTranscribeWorkflow, [{ videoId, userId: ownerId }]); + } catch (error) { + // Hand the claim back so a retried release can try again; a first start + // has no session to return to and its caller drops the claim instead. + if (from !== null) { + await rotateLiveSession(videoId, WORKFLOW_LIVE_SESSION, from).catch( + () => false, + ); + } + throw error; + } return "workflow"; } From ee07e53c119778bccda657284a2a018265549f98 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:59:39 +0000 Subject: [PATCH 10/11] fix(media-server): bound how long releasing live sessions can delay shutdown --- apps/media-server/src/lib/live-transcribe.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/apps/media-server/src/lib/live-transcribe.ts b/apps/media-server/src/lib/live-transcribe.ts index d8176a1134..aeb3dc3de9 100644 --- a/apps/media-server/src/lib/live-transcribe.ts +++ b/apps/media-server/src/lib/live-transcribe.ts @@ -35,6 +35,10 @@ export const LIVE = { MAX_CHUNK_BYTES: 64 * 1024 * 1024, REPORT_TIMEOUT_MS: 30_000, FINISH_TIMEOUT_MS: 110_000, + /** A release only restarts the session elsewhere, so it must not hold up + * a server that is shutting down. */ + RELEASE_TIMEOUT_MS: 20_000, + MAX_RELEASE_WAIT_MS: 45_000, } as const; const manifestEntrySchema = z.union([ @@ -375,7 +379,7 @@ export class LiveSession { outcome = "session-error"; } if (this.releasing) { - await this.report({ action: "release" }, LIVE.FINISH_TIMEOUT_MS, 3); + await this.report({ action: "release" }, LIVE.RELEASE_TIMEOUT_MS, 2); } else if (outcome) { await this.report( { action: "finish", outcome }, @@ -661,8 +665,9 @@ export function startLiveSession( let releasing: Promise | null = null; export function releaseLiveSessions(): Promise { - releasing ??= Promise.all( - [...sessions.values()].map((session) => session.release()), - ).then(() => {}); + releasing ??= Promise.race([ + Promise.all([...sessions.values()].map((session) => session.release())), + sleep(LIVE.MAX_RELEASE_WAIT_MS), + ]).then(() => {}); return releasing; } From 98be0c313b0fa51494b126b599f5675889b41aec Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:06:16 +0000 Subject: [PATCH 11/11] fix(media-server): cancel in-flight chunk reports when a session is released --- .../src/__tests__/lib/live-transcribe.test.ts | 28 +++++++++++++++++++ apps/media-server/src/lib/live-transcribe.ts | 23 +++++++++++++-- 2 files changed, 48 insertions(+), 3 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/live-transcribe.test.ts b/apps/media-server/src/__tests__/lib/live-transcribe.test.ts index 86b1ae9862..4820e5ec0a 100644 --- a/apps/media-server/src/__tests__/lib/live-transcribe.test.ts +++ b/apps/media-server/src/__tests__/lib/live-transcribe.test.ts @@ -120,6 +120,7 @@ function createFakeCap() { const failAudio = new Set(); let cursor = -1; let stopAfter: string | null = null; + let hangOn: string | null = null; let manifestGets = 0; let notModified = 0; let server: ReturnType; @@ -204,6 +205,9 @@ function createFakeCap() { if (stopAfter && report.action === stopAfter) { return Response.json({ continue: false }); } + if (hangOn === report.action) { + return new Promise(() => {}); + } if (report.action === "chunk" || report.action === "gap") { if (report.lastProcessedIndex === cursor) { cursor = report.lastIndex as number; @@ -228,6 +232,9 @@ function createFakeCap() { setCursor(value: number) { cursor = value; }, + hangOn(action: string) { + hangOn = action; + }, stopOn(action: string) { stopAfter = action; }, @@ -374,6 +381,27 @@ describe("LiveSession", () => { ).toBeFalse(); }); + test("hands over at once even while a chunk report is stuck", async () => { + const fresh = createFakeCap(); + fresh.record([1, 2, 3, 4]); + fresh.hangOn("chunk"); + const session = new LiveSession(startRequest(fresh), deps); + session.start(); + for (let i = 0; i < 400; i++) { + if (fresh.reports.some((report) => report.action === "chunk")) break; + await Bun.sleep(5); + } + const releasedAt = Date.now(); + await session.release(); + const waited = Date.now() - releasedAt; + fresh.stop(); + expect(waited).toBeLessThan(1_000); + expect(fresh.reports.map((report) => report.action)).toEqual([ + "chunk", + "release", + ]); + }); + test("resyncs to the web app's cursor instead of resending audio", async () => { const fresh = createFakeCap(); fresh.record([1, 2, 3, 4, 5, 6], true); diff --git a/apps/media-server/src/lib/live-transcribe.ts b/apps/media-server/src/lib/live-transcribe.ts index aeb3dc3de9..b1120f29c9 100644 --- a/apps/media-server/src/lib/live-transcribe.ts +++ b/apps/media-server/src/lib/live-transcribe.ts @@ -379,12 +379,18 @@ export class LiveSession { outcome = "session-error"; } if (this.releasing) { - await this.report({ action: "release" }, LIVE.RELEASE_TIMEOUT_MS, 2); + await this.report( + { action: "release" }, + LIVE.RELEASE_TIMEOUT_MS, + 2, + false, + ); } else if (outcome) { await this.report( { action: "finish", outcome }, LIVE.FINISH_TIMEOUT_MS, 5, + false, ); } this.segmentCache.clear(); @@ -594,13 +600,22 @@ export class LiveSession { return audio; } + /** + * Reports made from inside the loop stop as soon as the session is + * released or cancelled, so a slow chunk report can't delay the handoff. + * Abandoning one mid-flight is safe: the web app only applies a chunk that + * still matches its cursor and the artifact copy it read. + */ private async report( body: Record, timeoutMs: number = LIVE.REPORT_TIMEOUT_MS, attempts = 3, + cancellable = true, ): Promise { + const cancel = cancellable ? this.signal : undefined; for (let attempt = 0; attempt < attempts; attempt++) { - if (attempt > 0) await this.sleep(1_000 * 2 ** (attempt - 1)); + if (attempt > 0) await this.sleep(1_000 * 2 ** (attempt - 1), cancel); + if (cancel?.aborted) return "failed"; try { const response = await fetch(this.request.callbackUrl, { method: "POST", @@ -613,7 +628,9 @@ export class LiveSession { videoId: this.videoId, session: this.session, }), - signal: AbortSignal.timeout(timeoutMs), + signal: cancel + ? AbortSignal.any([cancel, AbortSignal.timeout(timeoutMs)]) + : AbortSignal.timeout(timeoutMs), }); if (response.status >= 400 && response.status < 500) { await response.body?.cancel().catch(() => {});