diff --git a/ApiCore-Backend/test/voice-flow.e2e-spec.ts b/ApiCore-Backend/test/voice-flow.e2e-spec.ts new file mode 100644 index 0000000..c4027f7 --- /dev/null +++ b/ApiCore-Backend/test/voice-flow.e2e-spec.ts @@ -0,0 +1,416 @@ +import { INestApplication, Logger } from '@nestjs/common'; +import * as bcrypt from 'bcryptjs'; +import { mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import request from 'supertest'; +import { App } from 'supertest/types'; +import { Prisma } from '../generated/prisma'; +import { berlinToday } from '../src/common/dates'; +import { VoiceTranscribeJob } from '../src/jobs/voice-transcribe.job'; +import { PrismaService } from '../src/prisma/prisma.service'; +import { claimVoiceCaptures } from '../src/voice/claims/voice-claims'; +import { openTranscript } from '../src/voice/core/transcript'; +import { FakeStructuredModel } from '../src/voice/extraction/fake-structured-model'; +import { VoiceExtractionService } from '../src/voice/extraction/voice-extraction.service'; +import { login } from './utils/api'; +import { createTestApp } from './utils/create-test-app'; +import { drainDomainEvents } from './utils/domain-events'; + +const WEBM = Buffer.concat([ + Buffer.from([0x1a, 0x45, 0xdf, 0xa3]), + Buffer.from('opus-bytes-of-a-spoken-note'), +]); +// Words that must never show up in a log line. +const SPOKEN = + 'Königin gesehen, Sperrbrutzellen sichtbar, vier Brutwaben belegt'; +const QUOTE = 'Königin gesehen'; +const FORBIDDEN_IN_LOGS = ['Königin', 'Sperrbrutzellen', 'Brutwaben']; +const ANSWER = { + findings: [ + { field: 'queenSeen', value: true, quote: QUOTE, corrected: false }, + ], + unclear: [], +}; + +interface IdBody { + id: string; + status?: string; +} + +/** + * The whole voice flow against the test database: upload, fetch, a stand-in + * for Whisper, ingest, extraction with a FakeStructuredModel (no network, no + * key), the ready notification and the confirmation in POST /inspections. + * Shares the database with parallel e2e files: own users/roles/colonies with + * a unique suffix, every claim/extraction scoped by `filter`, cleanup by suffix. + */ +describe('Voice flow (e2e)', () => { + let app: INestApplication; + let prisma: PrismaService; + let job: VoiceTranscribeJob; + let workDir: string; + const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`; + const password = 'SuperSecret123!'; + const emailOf = (name: string) => `voice-flow-${name}-${suffix}@example.com`; + const tokens: Record = {}; + const userIds: Record = {}; + const roleKey = `VOICE_FLOW_ONLY_VOICE_${suffix}`; + let apiaryId: string; + let colonyId: string; + let keyCounter = 0; + let addressCounter = 0; + const originalRecording = process.env.VOICE_RECORDING_ENABLED; + const originalLimit = process.env.VOICE_DAILY_EXTRACTIONS; + const originalWorkDir = process.env.VOICE_WORK_DIR; + + const server = () => app.getHttpServer(); + const as = (name: string) => ({ Authorization: `Bearer ${tokens[name]}` }); + const nextKey = () => + `voice-flow-${suffix}-${(keyCounter += 1).toString().padStart(4, '0')}`; + const mine = Prisma.sql`"ownerUserId" IN (SELECT "id" FROM "User" WHERE "email" LIKE ${`%${suffix}%`})`; + const row = (id: string) => + prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + const audioCount = (id: string) => + prisma.voiceCaptureAudio.count({ where: { captureId: id } }); + const readyNotifications = (id: string) => + prisma.notification.findMany({ + where: { relatedType: 'VoiceCapture', relatedId: id }, + }); + const extract = (model: FakeStructuredModel) => + new VoiceExtractionService(prisma, model).runOnce(undefined, { + filter: mine, + }); + + function upload(who: string) { + addressCounter += 1; + return request(server()) + .post('/api/v1/voice-captures') + .set(as(who)) + .set('X-Forwarded-For', `10.7.${addressCounter % 250}.9`) + .set('Idempotency-Key', nextKey()) + .field('colonyId', colonyId) + .field('inspectionDate', berlinToday()) + .field('durationSec', '30') + .attach('audio', WEBM, { + filename: 'note.webm', + contentType: 'audio/webm', + }); + } + + const sendText = (who: string, text = SPOKEN) => + request(server()) + .post('/api/v1/voice-captures/text') + .set(as(who)) + .set('Idempotency-Key', nextKey()) + .send({ colonyId, inspectionDate: berlinToday(), text }); + + const createInspection = (who: string, body: Record) => + request(server()) + .post('/api/v1/inspections') + .set(as(who)) + .set('Idempotency-Key', nextKey()) + .send({ colonyId, date: berlinToday(), ...body }); + + async function whisperOut(id: string, text = SPOKEN) { + await writeFile( + path.join(workDir, 'out', `${id}.json`), + JSON.stringify({ + v: 1, + captureId: id, + text, + durationSec: 28.4, + language: 'de', + languageProbability: 0.98, + segments: [{ start: 0, end: 6, text }], + }), + ); + } + + async function createUser(name: string, role: { key?: string; id?: string }) { + const user = await prisma.user.create({ + data: { + email: emailOf(name), + passwordHash: await bcrypt.hash(password, 4), + name: `Voice flow ${name}`, + role: { connect: role.key ? { key: role.key } : { id: role.id } }, + }, + }); + userIds[name] = user.id; + tokens[name] = (await login(app, emailOf(name), password)).accessToken; + } + + /** fetch (own rows, private work dir) -> Whisper stand-in -> ingest. */ + async function transcribe(id: string): Promise { + const fetched = await job.runFetchPhase({ filter: mine }); + expect(fetched).toMatchObject({ claimed: 1, written: 1 }); + expect((await readdir(path.join(workDir, 'in'))).sort()).toEqual( + [`${id}.webm`, 'manifest.json'].sort(), + ); + await whisperOut(id); + await job.ingest(); + } + + async function cleanOwn(): Promise { + const ids = Object.values(userIds); + const captures = await prisma.voiceCapture.findMany({ + where: { ownerUserId: { in: ids } }, + select: { id: true }, + }); + const captureIds = captures.map((c) => c.id); + await prisma.domainEvent.deleteMany({ + where: { aggregateType: 'VoiceCapture', aggregateId: { in: captureIds } }, + }); + await prisma.notification.deleteMany({ where: { userId: { in: ids } } }); + await prisma.voiceExtractionCall.deleteMany({ + where: { captureId: { in: captureIds } }, + }); + await prisma.voiceCapture.deleteMany({ + where: { ownerUserId: { in: ids } }, + }); + } + + async function cleanInspections(): Promise { + const inspections = await prisma.inspection.findMany({ + where: { colonyId }, + select: { id: true }, + }); + const ids = inspections.map((i) => i.id); + await prisma.domainEvent.deleteMany({ + where: { aggregateType: 'Inspection', aggregateId: { in: ids } }, + }); + await prisma.calendarSyncOutboxEntry.deleteMany({ + where: { sourceType: 'INSPECTION', sourceId: { in: ids } }, + }); + await prisma.inspection.deleteMany({ where: { colonyId } }); + } + + beforeAll(async () => { + workDir = await mkdtemp(path.join(tmpdir(), 'voice-flow-e2e-')); + process.env.VOICE_WORK_DIR = workDir; + // Notes are only accepted while recording is on; this file turns it on. + process.env.VOICE_RECORDING_ENABLED = 'true'; + process.env.VOICE_DAILY_EXTRACTIONS = '10000'; + app = await createTestApp(); + prisma = app.get(PrismaService); + job = new VoiceTranscribeJob(prisma); + + const onlyVoice = await prisma.role.create({ + data: { key: roleKey, name: 'Nur Sprache' }, + }); + const permission = await prisma.permission.findUniqueOrThrow({ + where: { key: 'voice-captures:use' }, + }); + await prisma.rolePermission.create({ + data: { roleId: onlyVoice.id, permissionId: permission.id }, + }); + await createUser('alice', { key: 'BEEKEEPER' }); + await createUser('bob', { key: 'BEEKEEPER' }); + await createUser('readonly', { key: 'READONLY' }); + await createUser('apiclient', { key: 'API_CLIENT' }); + await createUser('onlyvoice', { id: onlyVoice.id }); + apiaryId = ( + await prisma.apiary.create({ + data: { name: `Voice Flow Stand ${suffix}` }, + }) + ).id; + colonyId = ( + await prisma.colony.create({ + data: { + code: `VFLOW-${suffix}`, + apiaryId, + colonyType: 'PRODUCTION_COLONY', + }, + }) + ).id; + }); + + beforeEach(async () => { + await cleanOwn(); + await cleanInspections(); + }); + + afterAll(async () => { + // By suffix, so a failed setup leaks nothing. + const users = await prisma.user.findMany({ + where: { email: { contains: suffix } }, + select: { id: true }, + }); + const ids = users.map((u) => u.id); + users.forEach((u, index) => { + userIds[`cleanup-${index}`] = u.id; + }); + await cleanOwn(); + await cleanInspections(); + await prisma.suggestion.deleteMany({ where: { colonyId } }); + await prisma.auditLog.deleteMany({ where: { userId: { in: ids } } }); + await prisma.idempotencyRecord.deleteMany({ + where: { userId: { in: ids } }, + }); + await prisma.refreshToken.deleteMany({ where: { userId: { in: ids } } }); + await prisma.user.deleteMany({ where: { id: { in: ids } } }); + await prisma.role.deleteMany({ where: { key: roleKey } }); + await prisma.colony.deleteMany({ where: { code: `VFLOW-${suffix}` } }); + await prisma.apiary.deleteMany({ + where: { name: `Voice Flow Stand ${suffix}` }, + }); + await app.close(); + const restore = (key: string, value: string | undefined) => { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + }; + restore('VOICE_RECORDING_ENABLED', originalRecording); + restore('VOICE_DAILY_EXTRACTIONS', originalLimit); + restore('VOICE_WORK_DIR', originalWorkDir); + await rm(workDir, { recursive: true, force: true }); + }); + + it('runs upload to confirmation, deletes the content and never logs it', async () => { + const loggerSpies = ( + ['log', 'warn', 'error', 'debug', 'verbose', 'fatal'] as const + ).map((level) => jest.spyOn(Logger.prototype, level).mockImplementation()); + const consoleSpies = ( + ['log', 'warn', 'error', 'info', 'debug'] as const + ).map((level) => jest.spyOn(console, level).mockImplementation()); + try { + // Upload -> QUEUED + const up = await upload('alice').expect(201); + const id = (up.body as IdBody).id; + expect(up.body).toEqual({ id, status: 'QUEUED' }); + expect(await audioCount(id)).toBe(1); + + // fetch -> file + manifest; Whisper stand-in; ingest -> TRANSCRIBED + await transcribe(id); + const transcribed = await row(id); + expect(transcribed.status).toBe('TRANSCRIBED'); + expect(await audioCount(id)).toBe(0); + expect(openTranscript(id, transcribed.transcriptEnc as string).text).toBe( + SPOKEN, + ); + expect(transcribed.transcriptEnc).not.toContain('Königin'); + expect(await readdir(path.join(workDir, 'in'))).toEqual([]); + expect(await readdir(path.join(workDir, 'out'))).toEqual([]); + + // extraction -> READY, exactly one "ist fertig" notification + const model = new FakeStructuredModel([FakeStructuredModel.ok(ANSWER)]); + await extract(model); + expect((await row(id)).status).toBe('READY'); + expect(model.requests).toHaveLength(1); + await drainDomainEvents(app, [id]); + await drainDomainEvents(app, [id]); + const notifications = await readyNotifications(id); + expect(notifications).toHaveLength(1); + expect(notifications[0]).toMatchObject({ + userId: userIds.alice, + message: `Sprachnotiz für VFLOW-${suffix} ist fertig`, + }); + + // foreign user: 404; wrong roles: 403 + await request(server()) + .get(`/api/v1/voice-captures/${id}`) + .set(as('bob')) + .expect(404); + for (const who of ['readonly', 'apiclient', 'onlyvoice']) { + await request(server()) + .get(`/api/v1/voice-captures/${id}`) + .set(as(who)) + .expect(403); + } + const owner = await request(server()) + .get(`/api/v1/voice-captures/${id}`) + .set(as('alice')) + .expect(200); + expect(owner.body).toMatchObject({ + status: 'READY', + transcript: SPOKEN, + suggestions: [{ field: 'queenSeen', value: true, quote: QUOTE }], + }); + + // confirmation in POST /inspections wipes the content + const created = await createInspection('alice', { + voiceCaptureId: id, + colonyStrength: 12, + }).expect(201); + const confirmed = await row(id); + expect(confirmed).toMatchObject({ + status: 'CONFIRMED', + inspectionId: (created.body as IdBody).id, + transcriptEnc: null, + suggestionEnc: null, + }); + expect(confirmed.contentDeletedAt).not.toBeNull(); + expect(await audioCount(id)).toBe(0); + + // text path -> TRANSCRIBED -> READY + const textId = ((await sendText('alice').expect(201)).body as IdBody).id; + expect((await row(textId)).status).toBe('TRANSCRIBED'); + await extract(new FakeStructuredModel([FakeStructuredModel.ok(ANSWER)])); + expect((await row(textId)).status).toBe('READY'); + + const logged = JSON.stringify([ + ...loggerSpies.flatMap((spy) => spy.mock.calls), + ...consoleSpies.flatMap((spy) => spy.mock.calls), + ]); + for (const word of FORBIDDEN_IN_LOGS) { + expect(logged).not.toContain(word); + } + } finally { + [...loggerSpies, ...consoleSpies].forEach((spy) => spy.mockRestore()); + } + }); + + it('refuses READONLY and API_CLIENT on upload and text', async () => { + for (const who of ['readonly', 'apiclient']) { + expect((await upload(who)).status).toBe(403); + expect((await sendText(who)).status).toBe(403); + } + expect( + await prisma.voiceCapture.count({ + where: { ownerUserId: { in: Object.values(userIds) } }, + }), + ).toBe(0); + }); + + it('claims every capture exactly once across parallel fetch runs', async () => { + const ids: string[] = []; + for (let n = 0; n < 6; n += 1) { + const id = ((await upload('alice').expect(201)).body as IdBody).id; + ids.push(id); + } + const options = { limit: 2, maxTotalSec: 100_000, filter: mine }; + const batches = await Promise.all( + [0, 1, 2, 3].map(async () => { + const got: string[] = []; + for (;;) { + const claims = await claimVoiceCaptures( + prisma, + 'transcribe', + options, + ); + if (claims.length === 0) { + return got; + } + got.push(...claims.map((claim) => claim.id)); + } + }), + ); + const all = batches.flat(); + expect(all).toHaveLength(ids.length); + expect([...all].sort()).toEqual([...ids].sort()); + const rows = await prisma.voiceCapture.findMany({ + where: { id: { in: ids } }, + }); + expect(new Set(rows.map((r) => r.claimToken)).size).toBe(ids.length); + expect(rows.every((r) => r.claimedAt !== null)).toBe(true); + }); + + it('text path reaches READY without audio', async () => { + const id = ((await sendText('alice').expect(201)).body as IdBody).id; + expect(await audioCount(id)).toBe(0); + await extract(new FakeStructuredModel([FakeStructuredModel.ok(ANSWER)])); + expect((await row(id)).status).toBe('READY'); + }); +});