diff --git a/packages/adapters/src/media/transcription.postgres.test.ts b/packages/adapters/src/media/transcription.postgres.test.ts new file mode 100644 index 0000000..2c28a68 --- /dev/null +++ b/packages/adapters/src/media/transcription.postgres.test.ts @@ -0,0 +1,167 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { AdapterContext, JobPublisher } from "@rakazo/adapter-kit"; +import type { MediaJobBlock, MediaSettings, MessageBlock } from "@rakazo/contracts"; +import { createDb, type PrismaClient } from "@rakazo/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { storeArtifactFromFile } from "../artifact-files.js"; +import { LocalArtifactStore } from "../artifacts.js"; +import { createMediaRuntime, type MediaOperationRegistration } from "../media-runtime.js"; +import { createOpenAiCompatibleTranscriptionProvider } from "../openai-compatible-transcription.js"; +import { OpenAiCompatibleTranscriptionServerEmulator } from "../transcription-emulator.js"; +import { createTranscriptionOperations } from "./transcription.js"; + +const describePostgres = + process.env.VERIFY_DATABASE && process.env.DATABASE_URL ? describe.sequential : describe.skip; + +const WAV = new Uint8Array([82, 73, 70, 70, 36, 0, 0, 0, 87, 65, 86, 69, 1, 2, 3, 4]); +const SETTINGS: MediaSettings = { + comfyui: null, + transcription: { baseUrl: "http://127.0.0.1:8080/v1", model: undefined, apiKey: undefined }, +}; + +describePostgres("transcribe through the media runtime (PostgreSQL)", () => { + let prisma: PrismaClient; + let db: ReturnType; + let root: string; + const users: string[] = []; + + beforeAll(async () => { + db = createDb(process.env.DATABASE_URL!); + prisma = db.prisma; + root = await mkdtemp(path.join(tmpdir(), "transcribe-runtime-test-")); + }); + afterAll(async () => { + if (!db) return; + await prisma.organization.deleteMany({ where: { id: { in: users } } }); + await prisma.user.deleteMany({ where: { id: { in: users } } }); + await prisma.$disconnect(); + await db.pool.end(); + await rm(root, { recursive: true, force: true }); + }); + + async function setup() { + const id = randomUUID(); + users.push(id); + await prisma.user.create({ + data: { id, name: "Transcribe test", email: `${id}@example.test`, emailVerified: true }, + }); + await prisma.organization.create({ + data: { id, name: "Transcribe test", slug: id, createdAt: new Date() }, + }); + await prisma.space.create({ + data: { id, organizationId: id, name: "Transcribe test", isDefault: true }, + }); + await prisma.member.create({ + data: { id, organizationId: id, userId: id, role: "owner", createdAt: new Date() }, + }); + const bot = await prisma.bot.create({ + data: { spaceId: id, userId: id, name: "Studio", color: "test-color" }, + }); + const thread = await prisma.thread.create({ data: { spaceId: id, userId: id, botId: bot.id } }); + const artifacts = new LocalArtifactStore(path.join(root, id)); + const jobs: JobPublisher = { + enqueue: async () => undefined, + cancel: async () => undefined, + close: async () => undefined, + }; + const server = new OpenAiCompatibleTranscriptionServerEmulator(); + const runtime = createMediaRuntime( + { + prisma, + events: { notify: async () => undefined }, + jobs, + artifacts, + registrations: createTranscriptionOperations({ + loadSettings: async () => SETTINGS, + createProvider: (settings) => + createOpenAiCompatibleTranscriptionProvider({ ...settings, fetch: server.fetch }), + }) as MediaOperationRegistration[], + }, + { tempRoot: path.join(root, id, "tmp"), heartbeatMs: 50, progressIntervalMs: 0 }, + ); + const file = path.join(root, `${randomUUID()}.wav`); + await writeFile(file, WAV); + const source = await storeArtifactFromFile( + { prisma, artifacts }, + { + path: file, + name: "voice note.wav", + mimeType: "audio/wav", + spaceId: id, + userId: id, + botId: bot.id, + groupId: null, + runId: null, + }, + ); + const start = (input: Record) => + runtime.start({ + operation: "transcribe", + input, + spaceId: id, + userId: id, + botId: bot.id, + groupId: null, + threadId: thread.id, + runId: null, + }); + const card = async (jobId: string): Promise => { + const job = await prisma.mediaJob.findUniqueOrThrow({ where: { id: jobId } }); + const message = await prisma.message.findUniqueOrThrow({ where: { id: job.messageId! } }); + return (message.blocks as MessageBlock[]).find( + (block): block is MediaJobBlock => block.kind === "media", + )!; + }; + const context: AdapterContext = { + operationId: "test", + traceId: "test", + spaceId: id, + userId: id, + signal: new AbortController().signal, + }; + return { id, runtime, server, source, start, card, artifacts, context }; + } + + it("turns an audio artifact into subtitle files on the media card", async () => { + const ctx = await setup(); + expect((await ctx.runtime.capabilities(ctx.id)).transcribe).toBe(true); + const started = await ctx.start({ artifactId: ctx.source.id }); + expect(await ctx.card(started.jobId)).toMatchObject({ + operation: "transcribe", + status: "queued", + title: "Transcript", + }); + + await ctx.runtime.runJob(started.jobId); + + const job = await prisma.mediaJob.findUniqueOrThrow({ where: { id: started.jobId } }); + expect(job.status).toBe("done"); + const block = await ctx.card(started.jobId); + expect(block.status).toBe("done"); + expect(block.outputs.map(({ kind, name, mimeType }) => [kind, name, mimeType])).toEqual([ + ["file", "voice note.srt", "application/x-subrip"], + ["file", "voice note.vtt", "text/vtt"], + ]); + expect(block.outputs.every((output) => output.parentArtifactId === ctx.source.id)).toBe(true); + expect(ctx.server.requests).toEqual([ + expect.objectContaining({ fileName: "voice_note.wav", fileType: "audio/wav" }), + ]); + const vtt = await prisma.artifact.findUniqueOrThrow({ + where: { id: block.outputs[1]!.artifactId }, + }); + const body = new TextDecoder().decode(await ctx.artifacts.get(vtt.storageKey, ctx.context)); + expect(body).toBe( + "WEBVTT\n\n00:00:00.000 --> 00:00:02.400\nWelcome to the studio.\n\n00:00:02.400 --> 00:00:05.100\nToday we cut a vertical clip.\n", + ); + }); + + it("refuses an artifact of another user before a job exists", async () => { + const owner = await setup(); + const other = await setup(); + await expect(other.start({ artifactId: owner.source.id })).rejects.toThrow("Unknown artifact."); + expect(await prisma.mediaJob.count({ where: { userId: other.id } })).toBe(0); + }); +}); diff --git a/packages/testkit/src/cli/harness.ts b/packages/testkit/src/cli/harness.ts index 2b1141a..6e26f8b 100644 --- a/packages/testkit/src/cli/harness.ts +++ b/packages/testkit/src/cli/harness.ts @@ -117,6 +117,7 @@ async function main() { "packages/adapters/src/realtime.postgres.test.ts", "packages/adapters/src/job-reconciler.postgres.test.ts", "packages/adapters/src/cloud-agent.postgres.test.ts", + "packages/adapters/src/media/transcription.postgres.test.ts", "packages/adapters/src/builtin-memory-provider.postgres.test.ts", "packages/adapters/src/profile-consolidation.postgres.test.ts", ];