diff --git a/ApiCore-Backend/src/common/no-network-outside-loops.spec.ts b/ApiCore-Backend/src/common/no-network-outside-loops.spec.ts index 017db77..61400d0 100644 --- a/ApiCore-Backend/src/common/no-network-outside-loops.spec.ts +++ b/ApiCore-Backend/src/common/no-network-outside-loops.spec.ts @@ -23,7 +23,9 @@ interface SourceFile { const WEATHER_HOSTS = /open-meteo\.com/i; const TELEGRAM_HOST = /api\.telegram\.org/i; -const FETCH_USE = /(? { describe('with planted violations', () => { it.each([ ['colonies/colonies.service.ts', 'await fetch(url);'], + [ + 'colonies/colonies.service.ts', + "const a = '--phase=fetch'; await fetch(url);", + ], + ['colonies/colonies.service.ts', 'const a = --phase=fetch;'], ['colonies/colonies.service.ts', 'const f = fetch;'], ['colonies/colonies.service.ts', 'await globalThis.fetch(url);'], ['colonies/colonies.service.ts', "import https from 'node:https';"], @@ -248,6 +255,7 @@ describe('network calls stay inside their loop directories', () => { ['colonies/colonies.service.ts', '// fetch the data later\nconst a = 1;'], ['colonies/colonies.service.ts', "const scheme = 'https';"], ['colonies/colonies.service.ts', 'const o = { fetch: 1 };'], + ['jobs/job-names.ts', "const a = ['--phase=fetch', '--phase=ingest'];"], ])('lets %s pass: %s', (path, source) => { expect(violations([{ path, source }])).toEqual([]); }); diff --git a/ApiCore-Backend/src/domain-events/domain-event-types.ts b/ApiCore-Backend/src/domain-events/domain-event-types.ts index f85a06f..01462c9 100644 --- a/ApiCore-Backend/src/domain-events/domain-event-types.ts +++ b/ApiCore-Backend/src/domain-events/domain-event-types.ts @@ -12,6 +12,21 @@ import { TaskStatus, } from '../../generated/prisma'; +/** + * The codes a VoiceCaptureFailed event may carry: terminal failures of a + * note, plus EXTRACT_DISABLED (the operator notice). Waiting states + * (DAILY_LIMIT) and per-attempt codes (EXTRACT_RETRY) are no events. + */ +export type VoiceFailureEventCode = + | 'TRANSCRIBE_FAILED' + | 'NO_SPEECH' + | 'TOO_LONG' + | 'AUDIO_EXPIRED' + | 'EXTRACT_FAILED' + | 'EXTRACT_INVALID' + | 'REFUSED' + | 'EXTRACT_DISABLED'; + export interface ColonyStateSnapshot { status: ColonyStatus; healthStatus: ColonyCondition | null; @@ -154,6 +169,12 @@ export interface DomainEventPayloads { source: HiveChangeSource; inspectionId: string | null; }; + // T7, aggregate VoiceCapture: a voice note can be reviewed. Ids only, never + // the text, the quote or a suggestion. Not forwarded to webhooks. + VoiceCaptureReady: { captureId: string; colonyId: string }; + // T7, aggregate VoiceCapture: a voice note failed for good. The fixed + // error code only, never content. Not forwarded to webhooks. + VoiceCaptureFailed: { captureId: string; code: VoiceFailureEventCode }; } export type DomainEventType = keyof DomainEventPayloads; @@ -199,6 +220,8 @@ const CATALOG: Record = { ApiaryChanged: true, TskReferenceDayReached: true, HiveChanged: true, + VoiceCaptureReady: true, + VoiceCaptureFailed: true, }; export const DOMAIN_EVENT_TYPES: readonly DomainEventType[] = Object.freeze( @@ -223,7 +246,8 @@ export type AggregateType = | 'PhenologyReport' | 'WinterCheck' | 'Apiary' - | 'Compliance'; + | 'Compliance' + | 'VoiceCapture'; export interface NewDomainEvent { type: T; diff --git a/ApiCore-Backend/src/domain-events/domain-events.service.spec.ts b/ApiCore-Backend/src/domain-events/domain-events.service.spec.ts index 8d65165..79070a5 100644 --- a/ApiCore-Backend/src/domain-events/domain-events.service.spec.ts +++ b/ApiCore-Backend/src/domain-events/domain-events.service.spec.ts @@ -183,7 +183,7 @@ describe('DomainEventsService', () => { describe('domain event catalog', () => { it('lists every event type exactly once', () => { - expect(DOMAIN_EVENT_TYPES).toHaveLength(39); + expect(DOMAIN_EVENT_TYPES).toHaveLength(41); expect(new Set(DOMAIN_EVENT_TYPES).size).toBe(DOMAIN_EVENT_TYPES.length); expect(DOMAIN_EVENT_TYPES).toEqual( expect.arrayContaining([ @@ -207,6 +207,8 @@ describe('domain event catalog', () => { 'ApiaryChanged', 'TskReferenceDayReached', 'HiveChanged', + 'VoiceCaptureReady', + 'VoiceCaptureFailed', ]), ); }); diff --git a/ApiCore-Backend/src/inspections/dto/create-inspection.dto.ts b/ApiCore-Backend/src/inspections/dto/create-inspection.dto.ts index 5ba2ca1..ebdde2f 100644 --- a/ApiCore-Backend/src/inspections/dto/create-inspection.dto.ts +++ b/ApiCore-Backend/src/inspections/dto/create-inspection.dto.ts @@ -12,6 +12,7 @@ import { IsNumber, IsOptional, IsString, + Matches, Max, Min, ValidateNested, @@ -37,6 +38,15 @@ export class InspectionHiveChangeDto { } export class CreateInspectionDto { + @ApiPropertyOptional({ + description: + 'Sprachnotiz, die mit dem Speichern übernommen und gelöscht wird (gleiches Volk, gleicher Nutzer).', + }) + @IsOptional() + @IsString() + @Matches(/^[A-Za-z0-9_-]{1,40}$/) + voiceCaptureId?: string; + @ApiProperty() @IsString() colonyId!: string; diff --git a/ApiCore-Backend/src/inspections/inspections.module.ts b/ApiCore-Backend/src/inspections/inspections.module.ts index 41cc250..bda4b95 100644 --- a/ApiCore-Backend/src/inspections/inspections.module.ts +++ b/ApiCore-Backend/src/inspections/inspections.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { TasksModule } from '../tasks/tasks.module'; import { ColoniesModule } from '../colonies/colonies.module'; import { HiveModule } from '../hive/hive.module'; +import { VoiceConfirmationModule } from '../voice/captures/voice-confirmation.module'; import { InspectionsController } from './inspections.controller'; import { ColonyInspectionsController } from './colony-inspections.controller'; import { InspectionsService } from './inspections.service'; @@ -12,7 +13,7 @@ import { InspectionsService } from './inspections.service'; * PermissionResolverService comes from the global resolver module. */ @Module({ - imports: [TasksModule, ColoniesModule, HiveModule], + imports: [TasksModule, ColoniesModule, HiveModule, VoiceConfirmationModule], controllers: [InspectionsController, ColonyInspectionsController], providers: [InspectionsService], exports: [InspectionsService], diff --git a/ApiCore-Backend/src/inspections/inspections.service.spec.ts b/ApiCore-Backend/src/inspections/inspections.service.spec.ts index 37e32ad..73c6a8b 100644 --- a/ApiCore-Backend/src/inspections/inspections.service.spec.ts +++ b/ApiCore-Backend/src/inspections/inspections.service.spec.ts @@ -8,6 +8,7 @@ import { ColoniesService } from '../colonies/colonies.service'; import { PermissionResolverService } from '../auth/permissions/permission-resolver.service'; import { HiveService } from '../hive/hive.service'; import { Prisma } from '../../generated/prisma'; +import { VoiceConfirmationService } from '../voice/captures/voice-confirmation.service'; describe('InspectionsService', () => { let service: InspectionsService; @@ -37,6 +38,7 @@ describe('InspectionsService', () => { let colonies: { updateIn: jest.Mock }; let resolver: { getPermissions: jest.Mock }; let hive: { applyChange: jest.Mock }; + let voice: { confirmIn: jest.Mock }; const inspection = { id: 'inspection-1', @@ -77,6 +79,7 @@ describe('InspectionsService', () => { getPermissions: jest.fn().mockResolvedValue(new Set(['colonies:update'])), }; hive = { applyChange: jest.fn().mockResolvedValue({ version: 2 }) }; + voice = { confirmIn: jest.fn().mockResolvedValue(undefined) }; service = new InspectionsService( prisma as unknown as PrismaService, auditLog as unknown as AuditLogService, @@ -85,6 +88,7 @@ describe('InspectionsService', () => { colonies as unknown as ColoniesService, resolver as unknown as PermissionResolverService, hive as unknown as HiveService, + voice as unknown as VoiceConfirmationService, ); }); @@ -99,6 +103,7 @@ describe('InspectionsService', () => { ColoniesService, PermissionResolverService, HiveService, + VoiceConfirmationService, ]); }); @@ -152,6 +157,74 @@ describe('InspectionsService', () => { expect(tasks.createInTx).not.toHaveBeenCalled(); }); + describe('voiceCaptureId', () => { + it('confirms the capture in the same transaction, after the inspection and its event, and never stores the id on the inspection', async () => { + tx.inspection.create.mockResolvedValue(inspection); + + await service.create( + { colonyId: 'colony-1', date: '2026-01-01', voiceCaptureId: 'cap-1' }, + 'user-1', + ); + + expect(prisma.$transaction).toHaveBeenCalledTimes(1); + const createArg = ( + tx.inspection.create.mock.calls[0] as [{ data: object }] + )[0].data; + expect(createArg).not.toHaveProperty('voiceCaptureId'); + expect(voice.confirmIn).toHaveBeenCalledWith( + tx, + { kind: 'capture', id: 'cap-1' }, + { + userId: 'user-1', + colonyId: 'colony-1', + inspectionId: 'inspection-1', + }, + ); + expect(voice.confirmIn.mock.invocationCallOrder[0]).toBeGreaterThan( + events.record.mock.invocationCallOrder[0], + ); + }); + + it('confirms last, after a hive change (the capture lock comes last in the lock order)', async () => { + tx.inspection.create.mockResolvedValue(inspection); + + await service.create( + { + colonyId: 'colony-1', + date: '2026-01-01', + voiceCaptureId: 'cap-1', + hiveChange: { ops: [{}] } as never, + }, + 'user-1', + ); + + expect(voice.confirmIn.mock.invocationCallOrder[0]).toBeGreaterThan( + hive.applyChange.mock.invocationCallOrder[0], + ); + }); + + it('lets the whole transaction fail when the capture cannot be confirmed', async () => { + tx.inspection.create.mockResolvedValue(inspection); + voice.confirmIn.mockRejectedValue(new Error('gone')); + + await expect( + service.create( + { colonyId: 'colony-1', date: '2026-01-01', voiceCaptureId: 'cap-1' }, + 'user-1', + ), + ).rejects.toThrow('gone'); + }); + + it('does not touch the voice service without voiceCaptureId', async () => { + tx.inspection.create.mockResolvedValue(inspection); + await service.create( + { colonyId: 'colony-1', date: '2026-01-01' }, + 'user-1', + ); + expect(voice.confirmIn).not.toHaveBeenCalled(); + }); + }); + describe('applyToColony', () => { const saved = { ...inspection, diff --git a/ApiCore-Backend/src/inspections/inspections.service.ts b/ApiCore-Backend/src/inspections/inspections.service.ts index b03c61f..c6deda8 100644 --- a/ApiCore-Backend/src/inspections/inspections.service.ts +++ b/ApiCore-Backend/src/inspections/inspections.service.ts @@ -25,6 +25,7 @@ import { optionalDate } from '../common/patch-values'; import { berlinDayOf } from '../common/dates'; import type { HiveOp } from '../hive/hive-types'; import { HiveService, HiveSetupView } from '../hive/hive.service'; +import { VoiceConfirmationService } from '../voice/captures/voice-confirmation.service'; export type InspectionWithTakeover = Inspection & { colonyTakeover: ColonyTakeover; @@ -51,6 +52,7 @@ export class InspectionsService { private readonly colonies: ColoniesService, private readonly permissionResolver: PermissionResolverService, private readonly hiveService: HiveService, + private readonly voiceConfirmation: VoiceConfirmationService, ) {} async findAll(): Promise { @@ -72,7 +74,9 @@ export class InspectionsService { dto: CreateInspectionDto, actingUserId: string, ): Promise { - const { applyToColony = [], hiveChange, ...data } = dto; + // voiceCaptureId is not an Inspection column: it only triggers the + // confirmation of the note below. + const { applyToColony = [], hiveChange, voiceCaptureId, ...data } = dto; // Resolved before the transaction (it may query the database), once for // both features: a takeover only applies with colonies:update, a hive // change is refused without it. @@ -123,22 +127,36 @@ export class InspectionsService { canUpdateColony, actingUserId, ); - if (!hiveChange) { - return { ...inspection, colonyTakeover }; + const hive = hiveChange + ? await this.hiveService.applyChange( + tx, + dto.colonyId, + { + occurredOn: berlinDayOf(new Date(dto.date)), + source: 'INSPECTION', + // Plain objects only: the DTO instances carry class metadata. + ops: JSON.parse(JSON.stringify(hiveChange.ops)) as HiveOp[], + inspectionId: inspection.id, + }, + actingUserId, + ) + : undefined; + if (voiceCaptureId) { + // Last: VoiceCapture is the last table in the lock order. A note that + // cannot be confirmed fails the whole save. + await this.voiceConfirmation.confirmIn( + tx, + { kind: 'capture', id: voiceCaptureId }, + { + userId: actingUserId, + colonyId: inspection.colonyId, + inspectionId: inspection.id, + }, + ); } - const hive = await this.hiveService.applyChange( - tx, - dto.colonyId, - { - occurredOn: berlinDayOf(new Date(dto.date)), - source: 'INSPECTION', - // Plain objects only: the DTO instances carry class metadata. - ops: JSON.parse(JSON.stringify(hiveChange.ops)) as HiveOp[], - inspectionId: inspection.id, - }, - actingUserId, - ); - return { ...inspection, colonyTakeover, hive }; + return hive + ? { ...inspection, colonyTakeover, hive } + : { ...inspection, colonyTakeover }; }); } diff --git a/ApiCore-Backend/src/jobs/cleanup.job.spec.ts b/ApiCore-Backend/src/jobs/cleanup.job.spec.ts index 8e0a4f9..91df5e3 100644 --- a/ApiCore-Backend/src/jobs/cleanup.job.spec.ts +++ b/ApiCore-Backend/src/jobs/cleanup.job.spec.ts @@ -1,6 +1,7 @@ import { Logger } from '@nestjs/common'; import { IdempotencyCleanupService } from '../common/idempotency/idempotency-cleanup.service'; import { PrismaService } from '../prisma/prisma.service'; +import * as voiceRetention from '../voice/voice-retention'; import { JobRunService } from '../system/job-run.service'; import { WEATHER_RETENTION_YEARS } from '../weather/history/weather-history.config'; import { @@ -9,6 +10,7 @@ import { FAILED_RETENTION_DAYS, JOB_RUN_RETENTION_DAYS, JOB_RUN_STALE_HOURS, + VOICE_IDLE_RUN_RETENTION_HOURS, PENDING_DELIVERY_MAX_AGE_DAYS, WEBHOOK_DELIVERY_RETENTION_DAYS, } from './cleanup.job'; @@ -37,6 +39,13 @@ describe('CleanupJob', () => { }; beforeEach(() => { + jest.spyOn(voiceRetention, 'expireVoiceContent').mockResolvedValue({ + audioExpired: 1, + contentExpired: 2, + inactiveOwnersWiped: 3, + }); + jest.spyOn(voiceRetention, 'deleteOldVoiceRows').mockResolvedValue(4); + jest.spyOn(voiceRetention, 'deleteIdleVoiceRuns').mockResolvedValue(21); let count = 10; executeRaw = jest.fn, RawCall>(() => Promise.resolve(count++), @@ -62,6 +71,17 @@ describe('CleanupJob', () => { expect(JOB_RUN_STALE_HOURS).toBe(6); }); + it('cleanup runs the voice retention and prunes idle voice-transcribe runs only', async () => { + await job.run(); + + expect(voiceRetention.expireVoiceContent).toHaveBeenCalledTimes(1); + expect(voiceRetention.deleteOldVoiceRows).toHaveBeenCalledTimes(1); + expect(voiceRetention.deleteIdleVoiceRuns).toHaveBeenCalledWith( + expect.anything(), + VOICE_IDLE_RUN_RETENTION_HOURS, + ); + }); + it('cleanup deletes weather days before January 1st four years back', async () => { await job.run(); @@ -94,6 +114,11 @@ describe('CleanupJob', () => { 'jobRuns.aborted': 1, 'jobRuns.deleted': 19, weatherDays: 20, + 'voice.audioExpired': 1, + 'voice.contentExpired': 2, + 'voice.inactiveOwnersWiped': 3, + 'voice.rowsDeleted': 4, + 'jobRuns.voiceIdle': 21, }); }); diff --git a/ApiCore-Backend/src/jobs/cleanup.job.ts b/ApiCore-Backend/src/jobs/cleanup.job.ts index e33a571..bc6f054 100644 --- a/ApiCore-Backend/src/jobs/cleanup.job.ts +++ b/ApiCore-Backend/src/jobs/cleanup.job.ts @@ -2,6 +2,11 @@ import { Injectable, Logger } from '@nestjs/common'; import { IdempotencyCleanupService } from '../common/idempotency/idempotency-cleanup.service'; import { PrismaService } from '../prisma/prisma.service'; import { JobRunService } from '../system/job-run.service'; +import { + deleteIdleVoiceRuns, + deleteOldVoiceRows, + expireVoiceContent, +} from '../voice/voice-retention'; import { WEATHER_RETENTION_YEARS } from '../weather/history/weather-history.config'; import { Job, JobResult } from './job'; import { JobName } from './job-names'; @@ -18,6 +23,8 @@ export const JOB_RUN_RETENTION_DAYS = 90; export const PENDING_DELIVERY_MAX_AGE_DAYS = 7; /** Far beyond activeDeadlineSeconds (at most 1 h) of every job CronJob. */ export const JOB_RUN_STALE_HOURS = 6; +/** A voice-transcribe run (every 2 min) that did nothing is noise after a day. */ +export const VOICE_IDLE_RUN_RETENTION_HOURS = 24; /** * Daily housekeeping (CronJob 04:00). Every age is compared against now(), @@ -100,6 +107,17 @@ export class CleanupJob implements Job { DELETE FROM "ApiaryWeatherDay" WHERE "day" < make_date( extract(year FROM now() AT TIME ZONE 'Europe/Berlin')::int - ${WEATHER_RETENTION_YEARS}, 1, 1)`; + // The same retention rules the voice CronJob runs every 2 minutes (audio + // over 24 h, content over 7 days, deactivated users), plus the rows + // after 90 days. + const voiceRetention = await expireVoiceContent(this.prisma); + const voiceRowsDeleted = await deleteOldVoiceRows(this.prisma); + // 1440 runs a day, nearly all without work: keep only runs that did + // something (a counter above 0) or failed, for the usual 90 days. + const voiceIdleRuns = await deleteIdleVoiceRuns( + this.prisma, + VOICE_IDLE_RUN_RETENTION_HOURS, + ); return { idempotencyRecords, consumedTokens, @@ -115,6 +133,11 @@ export class CleanupJob implements Job { 'jobRuns.aborted': jobRunsAborted, 'jobRuns.deleted': jobRunsDeleted, weatherDays, + 'voice.audioExpired': voiceRetention.audioExpired, + 'voice.contentExpired': voiceRetention.contentExpired, + 'voice.inactiveOwnersWiped': voiceRetention.inactiveOwnersWiped, + 'voice.rowsDeleted': voiceRowsDeleted, + 'jobRuns.voiceIdle': voiceIdleRuns, }; } } diff --git a/ApiCore-Backend/src/jobs/job-names.spec.ts b/ApiCore-Backend/src/jobs/job-names.spec.ts index 662e04a..06e4235 100644 --- a/ApiCore-Backend/src/jobs/job-names.spec.ts +++ b/ApiCore-Backend/src/jobs/job-names.spec.ts @@ -28,12 +28,14 @@ describe('job names', () => { ).toEqual([]); }); - it('only rekey accepts an argument, and only --dry-run', () => { + it('only rekey and voice-transcribe accept arguments, and only their own', () => { expect(Object.keys(JOB_ALLOWED_ARGS)).toEqual([...JOB_NAMES]); for (const name of JOB_NAMES) { - expect(JOB_ALLOWED_ARGS[name]).toEqual( - name === 'rekey' ? ['--dry-run'] : [], - ); + const expected: Record = { + rekey: ['--dry-run'], + 'voice-transcribe': ['--phase=fetch', '--phase=ingest'], + }; + expect(JOB_ALLOWED_ARGS[name]).toEqual(expected[name] ?? []); } }); }); diff --git a/ApiCore-Backend/src/jobs/job-names.ts b/ApiCore-Backend/src/jobs/job-names.ts index 87840f6..8e5a4d3 100644 --- a/ApiCore-Backend/src/jobs/job-names.ts +++ b/ApiCore-Backend/src/jobs/job-names.ts @@ -7,6 +7,7 @@ export const JOB_NAMES = [ 'cleanup', 'app-backup', 'rekey', + 'voice-transcribe', ] as const; export type JobName = (typeof JOB_NAMES)[number]; @@ -18,6 +19,7 @@ export const JOB_LABELS_DE: Record = { cleanup: 'Aufräumen', 'app-backup': 'App-Backup', rekey: 'Schlüssel umstellen', + 'voice-transcribe': 'Sprachnotizen abschreiben', }; /** @@ -32,6 +34,7 @@ export const JOB_FAILURE_PERMISSION: Record = { cleanup: 'system:manage', 'app-backup': 'backup:configure', rekey: 'system:manage', + 'voice-transcribe': 'system:manage', }; /** @@ -45,6 +48,7 @@ export const JOB_ALLOWED_ARGS: Record = { cleanup: [], 'app-backup': [], rekey: ['--dry-run'], + 'voice-transcribe': ['--phase=fetch', '--phase=ingest'], }; export function isJobName(value: string): value is JobName { diff --git a/ApiCore-Backend/src/jobs/jobs.module.ts b/ApiCore-Backend/src/jobs/jobs.module.ts index 1fa6485..38d5afc 100644 --- a/ApiCore-Backend/src/jobs/jobs.module.ts +++ b/ApiCore-Backend/src/jobs/jobs.module.ts @@ -21,6 +21,7 @@ import { JobRegistry, JOBS } from './job-registry'; import { RekeyJob } from './rekey.job'; import { TimeEventsStep } from './steps/time-events.step'; import { TaskTemplatesJob } from './task-templates.job'; +import { VoiceTranscribeJob } from './voice-transcribe.job'; const JOB_CLASSES = [ DailyEvaluationJob, @@ -28,6 +29,7 @@ const JOB_CLASSES = [ CleanupJob, AppBackupJob, RekeyJob, + VoiceTranscribeJob, ]; /** diff --git a/ApiCore-Backend/src/jobs/voice-transcribe.job.spec.ts b/ApiCore-Backend/src/jobs/voice-transcribe.job.spec.ts new file mode 100644 index 0000000..011c994 --- /dev/null +++ b/ApiCore-Backend/src/jobs/voice-transcribe.job.spec.ts @@ -0,0 +1,79 @@ +import { mkdtemp, readdir, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { PrismaService } from '../prisma/prisma.service'; +import * as claims from '../voice/claims/voice-claims'; +import * as retention from '../voice/voice-retention'; +import { VoiceTranscribeJob } from './voice-transcribe.job'; + +describe('VoiceTranscribeJob', () => { + let dir: string; + let job: VoiceTranscribeJob; + let retentionSpy: jest.SpyInstance; + let staleSpy: jest.SpyInstance; + let claimSpy: jest.SpyInstance; + const tx = {}; + const prisma = { + $transaction: (fn: (t: typeof tx) => Promise) => fn(tx), + } as unknown as PrismaService; + + beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), 'voice-job-')); + process.env.VOICE_WORK_DIR = dir; + retentionSpy = jest + .spyOn(retention, 'expireVoiceContent') + .mockResolvedValue({ + audioExpired: 2, + contentExpired: 1, + inactiveOwnersWiped: 0, + }); + staleSpy = jest + .spyOn(claims, 'releaseStaleVoiceClaims') + .mockResolvedValue([]); + claimSpy = jest.spyOn(claims, 'claimVoiceCaptures').mockResolvedValue([]); + job = new VoiceTranscribeJob(prisma); + }); + afterEach(async () => { + delete process.env.VOICE_WORK_DIR; + jest.restoreAllMocks(); + await rm(dir, { recursive: true, force: true }); + }); + + it('is named voice-transcribe', () => { + expect(job.name).toBe('voice-transcribe'); + }); + + it('refuses a missing or unknown phase', async () => { + await expect(job.run([])).rejects.toThrow(/exactly one phase argument/); + await expect(job.run(['--phase=whisper'])).rejects.toThrow( + /exactly one phase argument/, + ); + await expect(job.run(['--phase=fetch', '--phase=ingest'])).rejects.toThrow( + /exactly one phase argument/, + ); + }); + + it('fetch ends quickly without work and writes no manifest', async () => { + const result = await job.run(['--phase=fetch']); + + expect(result).toMatchObject({ + claimed: 0, + audioExpired: 2, + contentExpired: 1, + }); + expect(retentionSpy).toHaveBeenCalledTimes(1); + expect(staleSpy).toHaveBeenCalledWith(tx, 'transcribe', {}); + expect(claimSpy).toHaveBeenCalledWith(prisma, 'transcribe', { + limit: 5, + maxTotalSec: 600, + }); + expect(await readdir(dir)).not.toContain('manifest.json'); + await expect(readdir(path.join(dir, 'in'))).resolves.toEqual([]); + }); + + it('ingest without a manifest does nothing', async () => { + const result = await job.run(['--phase=ingest']); + + expect(result).toEqual({ ingested: 0, failed: 0, retried: 0, lost: 0 }); + }); +}); diff --git a/ApiCore-Backend/src/jobs/voice-transcribe.job.ts b/ApiCore-Backend/src/jobs/voice-transcribe.job.ts new file mode 100644 index 0000000..8987039 --- /dev/null +++ b/ApiCore-Backend/src/jobs/voice-transcribe.job.ts @@ -0,0 +1,282 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { Prisma } from '../../generated/prisma'; +import { notifyWorker, WAKE_CHANNELS } from '../common/worker-wakeup'; +import { PrismaService } from '../prisma/prisma.service'; +import { + claimVoiceCaptures, + releaseStaleVoiceClaims, + settleVoiceFailure, + settleVoiceSuccess, + VoiceClaim, +} from '../voice/claims/voice-claims'; +import { sealTranscript } from '../voice/core/transcript'; +import { openVoiceBytes, voiceAad } from '../voice/core/voice-crypto'; +import { + clearWorkDir, + readManifest, + readResults, + WhisperResult, + workDirFromEnv, + writeInputs, +} from '../voice/transcribe/work-dir'; +import { recordVoiceCaptureFailed } from '../voice/voice-failure-hook'; +import { expireVoiceContent } from '../voice/voice-retention'; +import { Job, JobResult } from './job'; +import { JobName } from './job-names'; + +/** Per run: at most 5 notes and 600 seconds of audio (spec 4.2). */ +export const VOICE_FETCH_LIMIT = 5; +export const VOICE_FETCH_MAX_SEC = 600; +/** Whisper text shorter than this is no speech (spec 6.3). */ +const MIN_TEXT_CHARS = 5; +const MIN_DURATION_SEC = 1; +const MAX_DURATION_SEC = 240; + +type Phase = 'claim' | 'ingest'; +type ItemOutcome = 'ingested' | 'failed' | 'retried' | 'lost'; + +export interface VoiceFetchOptions { + /** Extra condition on "VoiceCapture" (tests scope their runs to own rows). */ + filter?: Prisma.Sql; +} + +function parsePhase(args: readonly string[]): Phase { + if (args.length === 1 && args[0] === '--phase=fetch') return 'claim'; + if (args.length === 1 && args[0] === '--phase=ingest') return 'ingest'; + throw new Error('voice-transcribe needs exactly one phase argument'); +} + +/** + * CronJob `apicore-job-voice-transcribe`, the two Node halves around the + * Whisper container. `fetch` runs the retention rules, claims notes, + * decrypts their audio in memory and hands it over in the RAM work dir; + * `ingest` takes Whisper's results, seals the text, deletes the audio and + * wakes the worker - each note in one transaction, guarded by its claim + * token. Content is never logged: only ids, counts and fixed codes. + */ +@Injectable() +export class VoiceTranscribeJob implements Job { + private readonly logger = new Logger(VoiceTranscribeJob.name); + readonly name: JobName = 'voice-transcribe'; + + constructor(private readonly prisma: PrismaService) {} + + async run(args: readonly string[]): Promise { + return parsePhase(args) === 'claim' ? this.runFetchPhase() : this.ingest(); + } + + async runFetchPhase(options: VoiceFetchOptions = {}): Promise { + const dir = workDirFromEnv(); + await clearWorkDir(dir); + const filter = options.filter ? { filter: options.filter } : {}; + const retention = await expireVoiceContent(this.prisma, filter); + const staleReleased = await this.releaseStale(filter); + const claims = await claimVoiceCaptures(this.prisma, 'transcribe', { + limit: VOICE_FETCH_LIMIT, + maxTotalSec: VOICE_FETCH_MAX_SEC, + ...filter, + }); + const inputs: { claim: VoiceClaim; audio: Buffer }[] = []; + for (const claim of claims) { + const audio = await this.decryptAudio(claim); + if (audio === null) { + await this.failAttempt(claim, 'TRANSCRIBE_FAILED'); + } else { + inputs.push({ claim, audio }); + } + } + if (inputs.length > 0) { + await writeInputs(dir, inputs); + } + return { + claimed: claims.length, + written: inputs.length, + staleReleased, + ...retention, + }; + } + + async ingest(): Promise { + const dir = workDirFromEnv(); + const counts: Record = { + ingested: 0, + failed: 0, + retried: 0, + lost: 0, + }; + try { + const manifest = await readManifest(dir); + if (manifest === null) { + return counts; + } + const results = await readResults(dir, manifest); + let errors = 0; + for (const item of manifest.items) { + try { + const outcome = await this.ingestOne( + item.captureId, + item.claimToken, + results.get(item.captureId) ?? null, + ); + counts[outcome] += 1; + } catch (error: unknown) { + errors += 1; + // The message of a failed seal or query holds no content, but + // keep to the id anyway. + this.logger.error( + `Ingest failed for voice note ${item.captureId}: ${error instanceof Error ? error.name : 'error'}`, + ); + } + } + if (errors > 0) { + throw new Error( + `voice ingest failed for ${errors} of ${manifest.items.length} notes (${JSON.stringify(counts)})`, + ); + } + return counts; + } finally { + await clearWorkDir(dir); + } + } + + private releaseStale(filter: VoiceFetchOptions): Promise { + return this.prisma.$transaction(async (tx) => { + const released = await releaseStaleVoiceClaims(tx, 'transcribe', filter); + for (const { id, code } of released) { + if (code !== null) { + await recordVoiceCaptureFailed(tx, id, 'TRANSCRIBE_FAILED'); + } + } + return released.length; + }); + } + + private async decryptAudio(claim: VoiceClaim): Promise { + const row = await this.prisma.voiceCaptureAudio.findUnique({ + where: { captureId: claim.id }, + }); + if (!row) { + return null; + } + try { + return openVoiceBytes( + Buffer.from(row.dataEnc), + voiceAad(claim.id, 'audio'), + ); + } catch { + this.logger.warn(`Audio of voice note ${claim.id} cannot be opened`); + return null; + } + } + + private failAttempt( + claim: VoiceClaim, + code: 'TRANSCRIBE_FAILED', + ): Promise<'retry' | 'failed' | 'lost'> { + return this.prisma.$transaction(async (tx) => { + const outcome = await settleVoiceFailure(tx, 'transcribe', claim, code); + if (outcome === 'failed') { + await recordVoiceCaptureFailed(tx, claim.id, code); + } + return outcome; + }); + } + + /** The claim as the database holds it now; null if it is gone. */ + private async loadClaim( + captureId: string, + claimToken: string, + ): Promise { + const rows = await this.prisma.$queryRaw(Prisma.sql` + SELECT "id", "claimToken", "attempts", "ownerUserId", "colonyId", + "audioMime", "durationSec" + FROM "VoiceCapture" + WHERE "id" = ${captureId} AND "claimToken" = ${claimToken} + AND "status" = 'TRANSCRIBING'`); + return rows[0] ?? null; + } + + private async ingestOne( + captureId: string, + claimToken: string, + result: WhisperResult | null, + ): Promise { + const claim = await this.loadClaim(captureId, claimToken); + if (claim === null) { + return 'lost'; + } + if (result === null || result.error === 'DECODE_FAILED') { + return this.outcomeOf(await this.failAttempt(claim, 'TRANSCRIBE_FAILED')); + } + const noSpeech = + result.error ?? + (result.text.trim().length < MIN_TEXT_CHARS ? 'NO_SPEECH' : null); + if (noSpeech === 'NO_SPEECH' || noSpeech === 'TOO_LONG') { + return this.finishFailed(claim, noSpeech); + } + if (result.durationSec > MAX_DURATION_SEC) { + return this.finishFailed(claim, 'TOO_LONG'); + } + return this.finishTranscribed(claim, result); + } + + private outcomeOf(settled: 'retry' | 'failed' | 'lost'): ItemOutcome { + return settled === 'retry' + ? 'retried' + : settled === 'failed' + ? 'failed' + : 'lost'; + } + + /** Terminal failure of the note itself: FAILED with the code, audio gone. */ + private finishFailed( + claim: VoiceClaim, + code: 'NO_SPEECH' | 'TOO_LONG', + ): Promise { + return this.prisma.$transaction(async (tx) => { + const changed = await tx.$executeRaw` + UPDATE "VoiceCapture" + SET "status" = 'FAILED', "lastErrorCode" = ${code}, + "attempts" = "attempts" + 1, "updatedAt" = now(), + "claimedAt" = NULL, "claimToken" = NULL + WHERE "id" = ${claim.id} AND "status" = 'TRANSCRIBING' + AND "claimToken" = ${claim.claimToken}`; + if (changed !== 1) { + return 'lost'; + } + await tx.voiceCaptureAudio.deleteMany({ where: { captureId: claim.id } }); + await recordVoiceCaptureFailed(tx, claim.id, code); + return 'failed'; + }); + } + + private finishTranscribed( + claim: VoiceClaim, + result: WhisperResult, + ): Promise { + const durationSec = Math.max( + MIN_DURATION_SEC, + Math.round(result.durationSec), + ); + const sealed = sealTranscript(claim.id, { + v: 1, + text: result.text, + durationSec: result.durationSec, + segments: result.segments, + }); + return this.prisma.$transaction(async (tx) => { + const settled = await settleVoiceSuccess( + tx, + 'transcribe', + claim, + Prisma.sql`"transcriptEnc" = ${sealed}, "durationSec" = ${durationSec}`, + ); + if (!settled) { + return 'lost'; + } + await tx.voiceCaptureAudio.deleteMany({ where: { captureId: claim.id } }); + await notifyWorker(tx, WAKE_CHANNELS.voiceExtraction); + return 'ingested'; + }); + } +} diff --git a/ApiCore-Backend/src/system/system-queue.service.spec.ts b/ApiCore-Backend/src/system/system-queue.service.spec.ts index ac6d1c4..3e5fdd3 100644 --- a/ApiCore-Backend/src/system/system-queue.service.spec.ts +++ b/ApiCore-Backend/src/system/system-queue.service.spec.ts @@ -1,6 +1,7 @@ import { ConflictException, NotFoundException } from '@nestjs/common'; import { AuditLogService } from '../audit/audit-log.service'; import { PrismaService } from '../prisma/prisma.service'; +import { JOB_NAMES } from '../jobs/job-names'; import { JobRunService } from './job-run.service'; import { SystemQueueService } from './system-queue.service'; @@ -67,7 +68,7 @@ describe('SystemQueueService', () => { oldestPendingAgeSeconds: 42.5, }); expect(overview.webhookDeliveries).toEqual({ pending: 4, failed: 1 }); - expect(overview.jobs).toHaveLength(5); + expect(overview.jobs).toHaveLength(JOB_NAMES.length); const sql = sqlText(prisma.$queryRaw.mock.calls[0]); expect(sql).toContain('EXTRACT(EPOCH FROM now() - MIN("occurredAt")'); }); diff --git a/ApiCore-Backend/src/voice/captures/voice-confirmation.module.ts b/ApiCore-Backend/src/voice/captures/voice-confirmation.module.ts new file mode 100644 index 0000000..6cc7f9e --- /dev/null +++ b/ApiCore-Backend/src/voice/captures/voice-confirmation.module.ts @@ -0,0 +1,13 @@ +import { Module } from '@nestjs/common'; +import { VoiceConfirmationService } from './voice-confirmation.service'; + +/** + * Only the confirmation step, so InspectionsModule can use it without + * pulling in the voice note API (controller, upload, limits). Audit comes + * from a global module. No network provider here. + */ +@Module({ + providers: [VoiceConfirmationService], + exports: [VoiceConfirmationService], +}) +export class VoiceConfirmationModule {} diff --git a/ApiCore-Backend/src/voice/captures/voice-confirmation.service.spec.ts b/ApiCore-Backend/src/voice/captures/voice-confirmation.service.spec.ts new file mode 100644 index 0000000..9bfd2fc --- /dev/null +++ b/ApiCore-Backend/src/voice/captures/voice-confirmation.service.spec.ts @@ -0,0 +1,156 @@ +import { HttpException } from '@nestjs/common'; +import { VoiceConfirmationService } from './voice-confirmation.service'; + +const CTX = { userId: 'user-1', colonyId: 'colony-1', inspectionId: 'insp-1' }; + +function codeOf(error: unknown): unknown { + return ( + (error as HttpException).getResponse() as { code?: string } | undefined + )?.code; +} + +async function failureOf(promise: Promise): Promise { + try { + await promise; + } catch (error) { + return error as HttpException; + } + throw new Error('expected the call to fail'); +} + +describe('VoiceConfirmationService', () => { + let service: VoiceConfirmationService; + let tx: { + $queryRaw: jest.Mock; + $executeRaw: jest.Mock; + voiceCapture: { findUnique: jest.Mock }; + voiceCaptureAudio: { deleteMany: jest.Mock }; + }; + let auditLog: { record: jest.Mock }; + let row: Record | null; + let locked: boolean; + let order: string[]; + + beforeEach(() => { + order = []; + locked = true; + row = { + ownerUserId: 'user-1', + colonyId: 'colony-1', + status: 'READY', + }; + tx = { + $queryRaw: jest + .fn() + .mockImplementation((query: { strings?: string[] }) => { + const text = (query.strings ?? []).join('?'); + if (text.includes('FOR UPDATE')) { + order.push('lock'); + return Promise.resolve(locked ? [{ id: 'cap-1' }] : []); + } + return Promise.resolve([]); + }), + $executeRaw: jest.fn().mockImplementation(() => { + order.push('update'); + return Promise.resolve(1); + }), + voiceCapture: { + findUnique: jest.fn().mockImplementation(() => { + order.push('read'); + return Promise.resolve(row); + }), + }, + voiceCaptureAudio: { + deleteMany: jest.fn().mockImplementation(() => { + order.push('audio'); + return Promise.resolve({ count: 1 }); + }), + }, + }; + auditLog = { record: jest.fn().mockResolvedValue(undefined) }; + service = new VoiceConfirmationService(auditLog as never); + }); + + const confirm = () => + service.confirmIn(tx as never, { kind: 'capture', id: 'cap-1' }, CTX); + + it('locks the capture, then wipes audio and content and confirms in the given transaction', async () => { + await confirm(); + + expect(order).toEqual(['lock', 'read', 'audio', 'update']); + expect(tx.voiceCaptureAudio.deleteMany).toHaveBeenCalledWith({ + where: { captureId: 'cap-1' }, + }); + const sql = ( + tx.$executeRaw.mock.calls[0] as [{ strings: string[] }] + )[0].strings.join('?'); + expect(sql).toContain(`"status" = 'CONFIRMED'`); + expect(sql).toContain('"transcriptEnc" = NULL'); + expect(sql).toContain('"suggestionEnc" = NULL'); + expect(sql).toContain('"claimToken" = NULL'); + expect(sql).toContain('"claimedAt" = NULL'); + expect(sql).toContain('"contentDeletedAt" = now()'); + }); + + it.each([ + 'QUEUED', + 'TRANSCRIBING', + 'TRANSCRIBED', + 'EXTRACTING', + 'READY', + 'FAILED', + ])('confirms a %s capture', async (status) => { + row = { ...row, status }; + await expect(confirm()).resolves.toBeUndefined(); + }); + + it.each(['CONFIRMED', 'DISCARDED', 'EXPIRED'])( + 'answers 409 VOICE_CAPTURE_GONE for a %s capture', + async (status) => { + row = { ...row, status }; + const error = await failureOf(confirm()); + expect(error.getStatus()).toBe(409); + expect(codeOf(error)).toBe('VOICE_CAPTURE_GONE'); + expect(tx.voiceCaptureAudio.deleteMany).not.toHaveBeenCalled(); + }, + ); + + it('answers 409 VOICE_CAPTURE_GONE for a missing capture', async () => { + locked = false; + row = null; + const error = await failureOf(confirm()); + expect(error.getStatus()).toBe(409); + expect(codeOf(error)).toBe('VOICE_CAPTURE_GONE'); + }); + + it('answers 409 VOICE_CAPTURE_GONE for a foreign capture without touching it', async () => { + row = { ...row, ownerUserId: 'someone-else' }; + const error = await failureOf(confirm()); + expect(codeOf(error)).toBe('VOICE_CAPTURE_GONE'); + expect(tx.voiceCaptureAudio.deleteMany).not.toHaveBeenCalled(); + expect(auditLog.record).not.toHaveBeenCalled(); + }); + + it('answers 409 VOICE_CAPTURE_COLONY_MISMATCH for another colony', async () => { + row = { ...row, colonyId: 'colony-2' }; + const error = await failureOf(confirm()); + expect(error.getStatus()).toBe(409); + expect(codeOf(error)).toBe('VOICE_CAPTURE_COLONY_MISMATCH'); + expect(tx.voiceCaptureAudio.deleteMany).not.toHaveBeenCalled(); + }); + + it('audits the status change without any content', async () => { + await confirm(); + expect(auditLog.record).toHaveBeenCalledWith( + { + entityType: 'VoiceCapture', + entityId: 'cap-1', + action: 'UPDATE', + userId: 'user-1', + beforeData: { status: 'READY' }, + afterData: { status: 'CONFIRMED', inspectionId: 'insp-1' }, + }, + tx, + ); + }); +}); diff --git a/ApiCore-Backend/src/voice/captures/voice-confirmation.service.ts b/ApiCore-Backend/src/voice/captures/voice-confirmation.service.ts new file mode 100644 index 0000000..895dd35 --- /dev/null +++ b/ApiCore-Backend/src/voice/captures/voice-confirmation.service.ts @@ -0,0 +1,108 @@ +import { ConflictException, HttpStatus, Injectable } from '@nestjs/common'; +import { Prisma } from '../../../generated/prisma'; +import { AuditLogService } from '../../audit/audit-log.service'; +import { lockRowForUpdate } from '../../prisma/row-lock'; +import { + VOICE_CAPTURE_COLONY_MISMATCH, + type VoiceErrorCode, +} from '../voice-error-codes'; + +/** What a note is confirmed against. Part 4 adds `{ kind: 'session'; id }`. */ +export type VoiceLink = { kind: 'capture'; id: string }; + +export interface VoiceConfirmContext { + userId: string; + colonyId: string; + inspectionId: string; +} + +/** States in which saving an inspection can no longer take the note over. */ +const CLOSED_STATUSES: ReadonlySet = new Set([ + 'CONFIRMED', + 'DISCARDED', + 'EXPIRED', +]); + +function gone(): ConflictException { + const code: VoiceErrorCode = 'VOICE_CAPTURE_GONE'; + return new ConflictException({ + statusCode: HttpStatus.CONFLICT, + code, + message: 'Die Sprachnotiz ist nicht mehr verfügbar.', + }); +} + +function otherColony(): ConflictException { + return new ConflictException({ + statusCode: HttpStatus.CONFLICT, + code: VOICE_CAPTURE_COLONY_MISMATCH, + message: 'Die Sprachnotiz gehört zu einem anderen Volk.', + }); +} + +/** + * Takes a voice note over when the inspection it led to is saved: status + * CONFIRMED, audio, transcript and suggestions gone, all in the caller's + * transaction. Anything that cannot be confirmed fails the whole save (spec + * 6.6). Ruling 1 (A1): any open state may be confirmed, so saving never leaves + * content behind; a running job loses its claim and can no longer settle. + */ +@Injectable() +export class VoiceConfirmationService { + constructor(private readonly auditLog: AuditLogService) {} + + /** + * Call it last in the transaction: VoiceCapture is the last table in the + * lock order (row-lock.ts). A foreign or missing note is answered like a + * gone one (409, not 404) so the client can retry without the id. + */ + async confirmIn( + tx: Prisma.TransactionClient, + link: VoiceLink, + ctx: VoiceConfirmContext, + ): Promise { + const locked = await lockRowForUpdate(tx, 'VoiceCapture', link.id); + if (!locked) { + throw gone(); + } + const row = await tx.voiceCapture.findUnique({ + where: { id: link.id }, + select: { ownerUserId: true, colonyId: true, status: true }, + }); + if ( + !row || + row.ownerUserId !== ctx.userId || + CLOSED_STATUSES.has(row.status) + ) { + throw gone(); + } + if (row.colonyId !== ctx.colonyId) { + throw otherColony(); + } + await tx.voiceCaptureAudio.deleteMany({ where: { captureId: link.id } }); + // Raw, so contentDeletedAt and updatedAt come from the database clock; + // CONFIRMED and inspectionId are set together (CHECK constraint). + await tx.$executeRaw(Prisma.sql` + UPDATE "VoiceCapture" + SET "status" = 'CONFIRMED', + "inspectionId" = ${ctx.inspectionId}, + "transcriptEnc" = NULL, + "suggestionEnc" = NULL, + "claimedAt" = NULL, + "claimToken" = NULL, + "contentDeletedAt" = now(), + "updatedAt" = now() + WHERE "id" = ${link.id}`); + await this.auditLog.record( + { + entityType: 'VoiceCapture', + entityId: link.id, + action: 'UPDATE', + userId: ctx.userId, + beforeData: { status: row.status }, + afterData: { status: 'CONFIRMED', inspectionId: ctx.inspectionId }, + }, + tx, + ); + } +} diff --git a/ApiCore-Backend/src/voice/transcribe/work-dir.spec.ts b/ApiCore-Backend/src/voice/transcribe/work-dir.spec.ts new file mode 100644 index 0000000..63cbf27 --- /dev/null +++ b/ApiCore-Backend/src/voice/transcribe/work-dir.spec.ts @@ -0,0 +1,187 @@ +import { + mkdtemp, + readFile, + readdir, + rm, + stat, + writeFile, +} from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { + MAX_INPUT_BYTES, + WhisperResult, + clearWorkDir, + readManifest, + readResults, + workDirFromEnv, + writeInputs, +} from './work-dir'; + +const CLAIM = (id: string) => ({ + id, + claimToken: `token-${id}`, + attempts: 0, + ownerUserId: 'u1', + colonyId: 'c1', + audioMime: 'audio/webm', + durationSec: 12, +}); + +const result = (captureId: string, over: Partial = {}) => ({ + v: 1, + captureId, + text: 'Volk ist ruhig', + durationSec: 11.5, + language: 'de', + languageProbability: 0.98, + segments: [{ start: 0, end: 2.5, text: 'Volk ist ruhig' }], + ...over, +}); + +describe('voice work dir', () => { + let dir: string; + + beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), 'voice-work-')); + }); + afterEach(async () => { + await rm(dir, { recursive: true, force: true }); + }); + + async function writeOut(id: string, body: unknown): Promise { + await writeFile( + path.join(dir, 'out', `${id}.json`), + typeof body === 'string' ? body : JSON.stringify(body), + ); + } + + it('reads the work dir from VOICE_WORK_DIR and defaults to /work', () => { + expect(workDirFromEnv({})).toBe('/work'); + expect(workDirFromEnv({ VOICE_WORK_DIR: '/tmp/x' })).toBe('/tmp/x'); + }); + + it('writes one file per capture and a manifest named after ids only', async () => { + const manifest = await writeInputs(dir, [ + { claim: CLAIM('ck1'), audio: Buffer.from('AAA') }, + { + claim: { ...CLAIM('ck2'), audioMime: 'audio/ogg', durationSec: null }, + audio: Buffer.from('BBBB'), + }, + ]); + + expect(manifest.items).toEqual([ + { + captureId: 'ck1', + claimToken: 'token-ck1', + file: 'ck1.webm', + mime: 'audio/webm', + reportedSec: 12, + }, + { + captureId: 'ck2', + claimToken: 'token-ck2', + file: 'ck2.ogg', + mime: 'audio/ogg', + reportedSec: null, + }, + ]); + expect((await readdir(path.join(dir, 'in'))).sort()).toEqual([ + 'ck1.webm', + 'ck2.ogg', + 'manifest.json', + ]); + expect(await readFile(path.join(dir, 'in', 'ck1.webm'))).toEqual( + Buffer.from('AAA'), + ); + expect(await readManifest(dir)).toEqual(manifest); + if (process.platform !== 'win32') { + expect((await stat(path.join(dir, 'in', 'ck1.webm'))).mode & 0o777).toBe( + 0o600, + ); + } + }); + + it('refuses audio that would not fit the memory volume', async () => { + const big = Buffer.alloc(MAX_INPUT_BYTES + 1); + await expect( + writeInputs(dir, [{ claim: CLAIM('ck1'), audio: big }]), + ).rejects.toThrow(/too large/); + expect(await readdir(dir)).toEqual([]); + }); + + it('refuses an id that is not a plain id and an unknown audio type', async () => { + await expect( + writeInputs(dir, [{ claim: CLAIM('../evil'), audio: Buffer.from('x') }]), + ).rejects.toThrow(/capture id/); + await expect( + writeInputs(dir, [ + { + claim: { ...CLAIM('ck1'), audioMime: 'video/mp4' }, + audio: Buffer.from('x'), + }, + ]), + ).rejects.toThrow(/audio type/); + }); + + it('returns null without a manifest and for a broken one', async () => { + expect(await readManifest(dir)).toBeNull(); + await writeInputs(dir, []); + await writeFile( + path.join(dir, 'in', 'manifest.json'), + '{"v":1,"items":[{"captureId":"../x"}]}', + ); + expect(await readManifest(dir)).toBeNull(); + await writeFile(path.join(dir, 'in', 'manifest.json'), 'not json'); + expect(await readManifest(dir)).toBeNull(); + }); + + it('reads a valid result and maps a missing or broken file to null', async () => { + const manifest = await writeInputs(dir, [ + { claim: CLAIM('ck1'), audio: Buffer.from('a') }, + { claim: CLAIM('ck2'), audio: Buffer.from('b') }, + { claim: CLAIM('ck3'), audio: Buffer.from('c') }, + { claim: CLAIM('ck4'), audio: Buffer.from('d') }, + ]); + await writeOut('ck1', result('ck1')); + await writeOut('ck3', '{"v":1,"captureId":"ck3","text":5}'); + await writeOut('ck4', result('ck4', { error: 'NO_SPEECH', text: '' })); + + const results = await readResults(dir, manifest); + + expect(results.get('ck1')).toEqual(result('ck1')); + expect(results.get('ck2')).toBeNull(); + expect(results.get('ck3')).toBeNull(); + expect(results.get('ck4')?.error).toBe('NO_SPEECH'); + }); + + it('rejects a result for another capture, a bad error and bad segments', async () => { + const manifest = await writeInputs(dir, [ + { claim: CLAIM('ck1'), audio: Buffer.from('a') }, + { claim: CLAIM('ck2'), audio: Buffer.from('b') }, + { claim: CLAIM('ck3'), audio: Buffer.from('c') }, + ]); + await writeOut('ck1', result('other')); + await writeOut('ck2', result('ck2', { error: 'WHATEVER' as never })); + await writeOut( + 'ck3', + result('ck3', { + segments: [{ start: 'a', end: 1, text: 'x' }] as never, + }), + ); + + const results = await readResults(dir, manifest); + + expect([...results.values()]).toEqual([null, null, null]); + }); + + it('clears input and output', async () => { + await writeInputs(dir, [{ claim: CLAIM('ck1'), audio: Buffer.from('a') }]); + await writeOut('ck1', result('ck1')); + + await clearWorkDir(dir); + + expect(await readdir(path.join(dir, 'in'))).toEqual([]); + expect(await readdir(path.join(dir, 'out'))).toEqual([]); + }); +}); diff --git a/ApiCore-Backend/src/voice/transcribe/work-dir.ts b/ApiCore-Backend/src/voice/transcribe/work-dir.ts new file mode 100644 index 0000000..62ba590 --- /dev/null +++ b/ApiCore-Backend/src/voice/transcribe/work-dir.ts @@ -0,0 +1,279 @@ +import { + lstat, + mkdir, + readFile, + rename, + rm, + writeFile, +} from 'node:fs/promises'; +import path from 'node:path'; +import type { VoiceAudioMime } from '../../common/file-signature'; +import type { VoiceClaim } from '../claims/voice-claims'; +import { MAX_TRANSCRIPT_CHARS, TranscriptSegment } from '../core/transcript'; + +/** + * The hand-over between the three containers of the voice CronJob: `fetch` + * writes decrypted audio and a manifest to `/in`, the Whisper container + * writes one result per capture to `/out`, `ingest` reads the results. + * The directory is a memory-backed emptyDir (32Mi), so decrypted audio never + * touches a disk. File names come from capture ids only, never from content. + */ +export const DEFAULT_WORK_DIR = '/work'; +const IN_DIR = 'in'; +const OUT_DIR = 'out'; +const MANIFEST_FILE = 'manifest.json'; + +/** Size of the memory volume (deploy/k8s voice component). */ +export const WORK_VOLUME_BYTES = 32 * 1024 * 1024; +/** Leaves room for the results of Whisper and the manifest on the volume. */ +export const MAX_INPUT_BYTES = WORK_VOLUME_BYTES - 8 * 1024 * 1024; +/** A result is a few kilobytes of text; anything bigger is not from Whisper. */ +const MAX_RESULT_BYTES = 1024 * 1024; +const MAX_ITEMS = 50; +const MAX_SEGMENTS = 5000; + +const ID_PATTERN = /^[A-Za-z0-9_-]{1,64}$/; +const TOKEN_PATTERN = /^[A-Za-z0-9_-]{1,64}$/; +const LANGUAGE_PATTERN = /^[A-Za-z-]{1,16}$/; + +const EXTENSIONS: Readonly> = { + 'audio/webm': 'webm', + 'audio/ogg': 'ogg', + 'audio/mp4': 'mp4', +}; + +export type WhisperError = 'NO_SPEECH' | 'TOO_LONG' | 'DECODE_FAILED'; +const WHISPER_ERRORS: readonly string[] = [ + 'NO_SPEECH', + 'TOO_LONG', + 'DECODE_FAILED', +]; + +export interface WorkManifestItem { + captureId: string; + claimToken: string; + file: string; + mime: VoiceAudioMime; + reportedSec: number | null; +} + +export interface WorkManifest { + v: 1; + items: WorkManifestItem[]; +} + +export interface WhisperResult { + v: 1; + captureId: string; + text: string; + durationSec: number; + language: string; + languageProbability: number; + segments: TranscriptSegment[]; + error?: WhisperError; +} + +export function workDirFromEnv( + env: Readonly> = process.env, +): string { + return env.VOICE_WORK_DIR || DEFAULT_WORK_DIR; +} + +function isAudioMime(value: unknown): value is VoiceAudioMime { + return typeof value === 'string' && Object.hasOwn(EXTENSIONS, value); +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +/** + * Writes the audio of the claimed captures and the manifest (last and by + * rename, so Whisper never sees a half-written manifest). Throws, with no + * content in the message, if an id or type is unusable or the audio would + * not fit the volume - checked before anything is written. + */ +export async function writeInputs( + dir: string, + items: readonly { claim: VoiceClaim; audio: Buffer }[], +): Promise { + const manifestItems = items.map(({ claim }): WorkManifestItem => { + if (!ID_PATTERN.test(claim.id)) { + throw new Error('voice work dir: unusable capture id'); + } + if (!isAudioMime(claim.audioMime)) { + throw new Error('voice work dir: unsupported audio type'); + } + if (!TOKEN_PATTERN.test(claim.claimToken)) { + throw new Error('voice work dir: unusable claim token'); + } + return { + captureId: claim.id, + claimToken: claim.claimToken, + file: `${claim.id}.${EXTENSIONS[claim.audioMime]}`, + mime: claim.audioMime, + reportedSec: claim.durationSec, + }; + }); + const total = items.reduce((sum, { audio }) => sum + audio.length, 0); + if (total > MAX_INPUT_BYTES) { + throw new Error('voice work dir: audio too large for the work volume'); + } + const inDir = path.join(dir, IN_DIR); + await mkdir(inDir, { recursive: true, mode: 0o700 }); + await mkdir(path.join(dir, OUT_DIR), { recursive: true, mode: 0o700 }); + for (const [index, { audio }] of items.entries()) { + await writeFile(path.join(inDir, manifestItems[index].file), audio, { + mode: 0o600, + }); + } + const manifest: WorkManifest = { v: 1, items: manifestItems }; + const temporary = path.join(inDir, `${MANIFEST_FILE}.tmp`); + await writeFile(temporary, JSON.stringify(manifest), { mode: 0o600 }); + await rename(temporary, path.join(inDir, MANIFEST_FILE)); + return manifest; +} + +function toManifestItem(value: unknown): WorkManifestItem | null { + if (!isRecord(value)) { + return null; + } + const { captureId, claimToken, file, mime, reportedSec } = value; + if ( + typeof captureId !== 'string' || + !ID_PATTERN.test(captureId) || + typeof claimToken !== 'string' || + !TOKEN_PATTERN.test(claimToken) || + typeof file !== 'string' || + !isAudioMime(mime) || + file !== `${captureId}.${EXTENSIONS[mime]}` || + !(reportedSec === null || Number.isFinite(reportedSec)) + ) { + return null; + } + return { + captureId, + claimToken, + file, + mime, + reportedSec: reportedSec as number | null, + }; +} + +/** The manifest of this run, or null if there is none or it is not valid. */ +export async function readManifest(dir: string): Promise { + let parsed: unknown; + try { + parsed = JSON.parse( + await readFile(path.join(dir, IN_DIR, MANIFEST_FILE), 'utf8'), + ); + } catch { + return null; + } + if (!isRecord(parsed) || parsed.v !== 1 || !Array.isArray(parsed.items)) { + return null; + } + if (parsed.items.length > MAX_ITEMS) { + return null; + } + const items = parsed.items.map(toManifestItem); + if (items.some((item) => item === null)) { + return null; + } + return { v: 1, items: items as WorkManifestItem[] }; +} + +function toSegment(value: unknown): TranscriptSegment | null { + if (!isRecord(value)) { + return null; + } + const { start, end, text } = value; + return typeof start === 'number' && + Number.isFinite(start) && + typeof end === 'number' && + Number.isFinite(end) && + typeof text === 'string' && + text.length <= MAX_TRANSCRIPT_CHARS + ? { start, end, text } + : null; +} + +/** Strict check of one result file; the content is never put into an error. */ +function toResult(value: unknown, captureId: string): WhisperResult | null { + if (!isRecord(value) || value.v !== 1 || value.captureId !== captureId) { + return null; + } + const { text, durationSec, language, languageProbability, segments, error } = + value; + if ( + typeof text !== 'string' || + text.length > MAX_TRANSCRIPT_CHARS || + typeof durationSec !== 'number' || + !Number.isFinite(durationSec) || + durationSec < 0 || + typeof language !== 'string' || + !LANGUAGE_PATTERN.test(language) || + typeof languageProbability !== 'number' || + !(languageProbability >= 0 && languageProbability <= 1) || + !Array.isArray(segments) || + segments.length > MAX_SEGMENTS || + !(error === undefined || WHISPER_ERRORS.includes(error as string)) + ) { + return null; + } + const checked = segments.map(toSegment); + if (checked.some((segment) => segment === null)) { + return null; + } + const result: WhisperResult = { + v: 1, + captureId, + text, + durationSec, + language, + languageProbability, + segments: checked as TranscriptSegment[], + }; + if (error !== undefined) { + result.error = error as WhisperError; + } + return result; +} + +async function readResult( + dir: string, + captureId: string, +): Promise { + const file = path.join(dir, OUT_DIR, `${captureId}.json`); + try { + // lstat: a symlink or device planted by the Whisper container is no result. + const info = await lstat(file); + if (!info.isFile() || info.size > MAX_RESULT_BYTES) { + return null; + } + return toResult(JSON.parse(await readFile(file, 'utf8')), captureId); + } catch { + return null; + } +} + +/** One entry per manifest item; null = no file, unreadable or not valid. */ +export async function readResults( + dir: string, + manifest: WorkManifest, +): Promise> { + const results = new Map(); + for (const item of manifest.items) { + results.set(item.captureId, await readResult(dir, item.captureId)); + } + return results; +} + +/** Removes everything of a run (audio, manifest, results) and recreates the dirs. */ +export async function clearWorkDir(dir: string): Promise { + for (const sub of [IN_DIR, OUT_DIR]) { + const target = path.join(dir, sub); + await rm(target, { recursive: true, force: true }); + await mkdir(target, { recursive: true, mode: 0o700 }); + } +} diff --git a/ApiCore-Backend/src/voice/voice-error-codes.ts b/ApiCore-Backend/src/voice/voice-error-codes.ts index ffebb85..72d01c9 100644 --- a/ApiCore-Backend/src/voice/voice-error-codes.ts +++ b/ApiCore-Backend/src/voice/voice-error-codes.ts @@ -20,3 +20,9 @@ export const VOICE_ERROR_CODES = [ ] as const; export type VoiceErrorCode = (typeof VOICE_ERROR_CODES)[number]; + +/** + * Answered by the confirmation when a note belongs to another colony. Not in + * VOICE_ERROR_CODES: it is never stored in lastErrorCode. + */ +export const VOICE_CAPTURE_COLONY_MISMATCH = 'VOICE_CAPTURE_COLONY_MISMATCH'; diff --git a/ApiCore-Backend/src/voice/voice-error-texts.ts b/ApiCore-Backend/src/voice/voice-error-texts.ts new file mode 100644 index 0000000..e27a069 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-error-texts.ts @@ -0,0 +1,30 @@ +import type { VoiceErrorCode } from './voice-error-codes'; + +/** + * One short German sentence per error code: what the user does next. Fixed + * texts only - a voice note's content never lands in a message. + */ +export const VOICE_ERROR_TEXTS_DE: Readonly> = { + UNSUPPORTED_AUDIO: 'Das Audioformat wird nicht unterstützt.', + TOO_LARGE: 'Die Aufnahme ist zu groß. Bitte kürzer aufnehmen.', + TOO_MANY_OPEN: + 'Zu viele offene Sprachnotizen. Erst auswerten oder verwerfen.', + NO_SPEECH: 'Es war nichts zu verstehen. Bitte noch einmal aufnehmen.', + TOO_LONG: 'Die Aufnahme ist zu lang. Bitte kürzer aufnehmen.', + TRANSCRIBE_FAILED: + 'Die Aufnahme konnte nicht abgeschrieben werden. Bitte noch einmal aufnehmen.', + AUDIO_EXPIRED: + 'Die Aufnahme wurde nach 24 Stunden gelöscht. Bitte noch einmal aufnehmen.', + EXTRACT_DISABLED: + 'Die Auswertung ist ausgeschaltet. Bitte die Durchsicht von Hand ausfüllen.', + EXTRACT_RETRY: 'Die Auswertung wird noch einmal versucht.', + EXTRACT_FAILED: + 'Die Auswertung hat nicht geklappt. Bitte erneut auswerten oder von Hand ausfüllen.', + DAILY_LIMIT: + 'Das Tageslimit für die Auswertung ist erreicht. Bitte morgen erneut auswerten.', + EXTRACT_INVALID: + 'Die Auswertung war nicht brauchbar. Bitte erneut auswerten oder von Hand ausfüllen.', + REFUSED: + 'Die Auswertung wurde abgelehnt. Bitte die Durchsicht von Hand ausfüllen.', + VOICE_CAPTURE_GONE: 'Die Sprachnotiz ist nicht mehr vorhanden.', +}; diff --git a/ApiCore-Backend/src/voice/voice-failure-hook.spec.ts b/ApiCore-Backend/src/voice/voice-failure-hook.spec.ts new file mode 100644 index 0000000..799515b --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-failure-hook.spec.ts @@ -0,0 +1,40 @@ +import type { Prisma } from '../../generated/prisma'; +import { + recordVoiceCaptureFailed, + recordVoiceCaptureReady, +} from './voice-failure-hook'; + +describe('voice event writers', () => { + let createMany: jest.Mock; + let tx: Prisma.TransactionClient; + + beforeEach(() => { + createMany = jest.fn().mockResolvedValue({ count: 1 }); + tx = { + domainEvent: { createMany }, + $executeRaw: jest.fn().mockResolvedValue(1), + } as unknown as Prisma.TransactionClient; + }); + + it('payloads carry only ids and the code', async () => { + await recordVoiceCaptureReady(tx, 'vc-1', 'col-1'); + await recordVoiceCaptureFailed(tx, 'vc-2', 'NO_SPEECH'); + + const rows = createMany.mock.calls.map( + ([arg]: [{ data: Record[] }]) => arg.data[0], + ); + expect(rows[0]).toMatchObject({ + type: 'VoiceCaptureReady', + aggregateType: 'VoiceCapture', + aggregateId: 'vc-1', + payload: { captureId: 'vc-1', colonyId: 'col-1' }, + actorUserId: null, + }); + expect(rows[1]).toMatchObject({ + type: 'VoiceCaptureFailed', + aggregateType: 'VoiceCapture', + aggregateId: 'vc-2', + payload: { captureId: 'vc-2', code: 'NO_SPEECH' }, + }); + }); +}); diff --git a/ApiCore-Backend/src/voice/voice-failure-hook.ts b/ApiCore-Backend/src/voice/voice-failure-hook.ts new file mode 100644 index 0000000..c1776e1 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-failure-hook.ts @@ -0,0 +1,41 @@ +import { Prisma } from '../../generated/prisma'; +import { DomainEventsService } from '../domain-events/domain-events.service'; +import type { VoiceFailureEventCode } from '../domain-events/domain-event-types'; + +export type { VoiceFailureEventCode }; + +// DomainEventsService has no dependencies: jobs, which have no Nest +// container, write events through the same code path as the API. +const events = new DomainEventsService(); + +/** + * Event writers for the voice jobs (Tasks 11 and 15). Call them in the same + * transaction that sets the voice note READY or FAILED: no commit, no event, + * and no commit without its event. Payloads carry ids and the fixed code + * only - never text, quote, suggestion or audio. + */ +export function recordVoiceCaptureReady( + tx: Prisma.TransactionClient, + captureId: string, + colonyId: string, +): Promise { + return events.record(tx, { + type: 'VoiceCaptureReady', + aggregateType: 'VoiceCapture', + aggregateId: captureId, + payload: { captureId, colonyId }, + }); +} + +export function recordVoiceCaptureFailed( + tx: Prisma.TransactionClient, + captureId: string, + code: VoiceFailureEventCode, +): Promise { + return events.record(tx, { + type: 'VoiceCaptureFailed', + aggregateType: 'VoiceCapture', + aggregateId: captureId, + payload: { captureId, code }, + }); +} diff --git a/ApiCore-Backend/src/voice/voice-handlers.module.ts b/ApiCore-Backend/src/voice/voice-handlers.module.ts new file mode 100644 index 0000000..5ebb098 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-handlers.module.ts @@ -0,0 +1,15 @@ +import { Module } from '@nestjs/common'; +import { NotificationsModule } from '../notifications/notifications.module'; +import { VoiceEventHandlers } from './voice.handlers'; + +/** + * The voice event handlers (T7). Only the worker dispatches events, so + * WorkerModule imports it; VoiceModule imports it too, so the API and the + * worker discover the same handlers. Kept apart from VoiceModule, whose + * controllers the worker does not serve. + */ +@Module({ + imports: [NotificationsModule], + providers: [VoiceEventHandlers], +}) +export class VoiceHandlersModule {} diff --git a/ApiCore-Backend/src/voice/voice-retention.spec.ts b/ApiCore-Backend/src/voice/voice-retention.spec.ts new file mode 100644 index 0000000..4e562dd --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-retention.spec.ts @@ -0,0 +1,76 @@ +import { PrismaService } from '../prisma/prisma.service'; +import * as failureHook from './voice-failure-hook'; +import { deleteOldVoiceRows, expireVoiceContent } from './voice-retention'; + +type RawCall = [readonly string[] | { sql: string }, ...unknown[]]; + +describe('voice retention', () => { + let queryRaw: jest.Mock, RawCall>; + let executeRaw: jest.Mock, RawCall>; + let hook: jest.SpyInstance; + let db: PrismaService; + + const sql = (call: RawCall) => + (Array.isArray(call[0]) + ? call[0].join('?') + : (call[0] as { sql: string }).sql + ).replace(/\s+/g, ' '); + + beforeEach(() => { + queryRaw = jest.fn, RawCall>(() => Promise.resolve([])); + executeRaw = jest.fn, RawCall>(() => Promise.resolve(0)); + const tx = { $queryRaw: queryRaw, $executeRaw: executeRaw }; + db = { + ...tx, + $transaction: (fn: (t: typeof tx) => Promise) => fn(tx), + } as unknown as PrismaService; + hook = jest + .spyOn(failureHook, 'recordVoiceCaptureFailed') + .mockResolvedValue(); + }); + afterEach(() => jest.restoreAllMocks()); + + it('measures the audio age against the database clock, 24 hours', async () => { + await expireVoiceContent(db); + + const text = queryRaw.mock.calls.map(sql).join(' | '); + expect(text).toContain(`"createdAt" < now() - interval '24 hours'`); + expect(text).toContain(`'AUDIO_EXPIRED'`); + }); + + it('calls the failure hook once per capture that ended FAILED/AUDIO_EXPIRED', async () => { + queryRaw.mockResolvedValueOnce([{ id: 'a' }, { id: 'b' }]); + + const counts = await expireVoiceContent(db); + + expect(counts.audioExpired).toBe(2); + expect(hook).toHaveBeenCalledTimes(2); + expect(hook).toHaveBeenCalledWith(expect.anything(), 'a', 'AUDIO_EXPIRED'); + }); + + it('counts wiped content of expired notes and of deactivated users separately', async () => { + executeRaw.mockResolvedValueOnce(3).mockResolvedValueOnce(4); + + const counts = await expireVoiceContent(db); + + expect(counts).toEqual({ + audioExpired: 0, + contentExpired: 3, + inactiveOwnersWiped: 4, + }); + const text = executeRaw.mock.calls.map(sql); + expect(text[0]).toContain(`"contentExpiresAt" < now()`); + expect(text[1]).toContain(`"isActive" = false`); + expect( + text.every((each) => each.includes('"contentDeletedAt" = now()')), + ).toBe(true); + }); + + it('deletes rows after 90 days by the database clock', async () => { + executeRaw.mockResolvedValueOnce(5); + + expect(await deleteOldVoiceRows(db)).toBe(5); + expect(sql(executeRaw.mock.calls[0])).toContain(`"createdAt" < now() - (`); + expect(executeRaw.mock.calls[0].slice(1)).toContain(90); + }); +}); diff --git a/ApiCore-Backend/src/voice/voice-retention.ts b/ApiCore-Backend/src/voice/voice-retention.ts new file mode 100644 index 0000000..86a0cc2 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-retention.ts @@ -0,0 +1,151 @@ +import { Prisma } from '../../generated/prisma'; +import { PrismaService } from '../prisma/prisma.service'; +import { recordVoiceCaptureFailed } from './voice-failure-hook'; + +/** Audio is kept until it is transcribed, 24 hours at most. */ +export const VOICE_AUDIO_MAX_AGE_HOURS = 24; +/** The row itself (no content) stays 90 days for the error and cost overview. */ +export const VOICE_ROW_RETENTION_DAYS = 90; + +export interface VoiceRetentionCounts { + audioExpired: number; + contentExpired: number; + inactiveOwnersWiped: number; +} + +export interface VoiceRetentionOptions { + /** + * Extra condition on "VoiceCapture" columns, ANDed to every statement + * (e.g. an owner filter in tests). Trusted SQL only. + */ + filter?: Prisma.Sql; +} + +const OPEN_CONTENT = Prisma.sql`"status" NOT IN ('CONFIRMED', 'DISCARDED', 'EXPIRED')`; + +/** + * Audio older than 24 hours goes, and a note that still waited for its + * transcript (QUEUED, TRANSCRIBING) ends FAILED/AUDIO_EXPIRED; its claim is + * cleared, so a job that still holds it loses the compare-and-set. A note in + * another state keeps its status (FAILED keeps its own code). + */ +async function expireAudio( + tx: Prisma.TransactionClient, + filter: Prisma.Sql, +): Promise { + const moved = await tx.$queryRaw<{ id: string }[]>(Prisma.sql` + WITH old AS ( + SELECT c."id" + FROM "VoiceCaptureAudio" a + JOIN "VoiceCapture" c ON c."id" = a."captureId" + WHERE a."createdAt" < now() - interval '24 hours' ${filter} + ORDER BY c."id" + FOR UPDATE OF c + ), gone AS ( + DELETE FROM "VoiceCaptureAudio" + WHERE "captureId" IN (SELECT "id" FROM old) + ) + UPDATE "VoiceCapture" + SET "status" = 'FAILED', + "lastErrorCode" = 'AUDIO_EXPIRED', + "claimedAt" = NULL, + "claimToken" = NULL, + "updatedAt" = now() + WHERE "id" IN (SELECT "id" FROM old) + AND "status" IN ('QUEUED', 'TRANSCRIBING') + RETURNING "id"`); + for (const { id } of moved) { + await recordVoiceCaptureFailed(tx, id, 'AUDIO_EXPIRED'); + } + return moved.length; +} + +/** + * Deletes audio, transcript and suggestions of the notes the `due` + * condition selects and marks them EXPIRED. Locks the rows first, so a + * confirm or discard in flight finishes before the wipe looks at the state. + */ +function wipeContent( + tx: Prisma.TransactionClient, + due: Prisma.Sql, + filter: Prisma.Sql, +): Promise { + return tx.$executeRaw(Prisma.sql` + WITH due AS ( + SELECT "id" FROM "VoiceCapture" + WHERE "contentDeletedAt" IS NULL AND ${OPEN_CONTENT} AND ${due} ${filter} + ORDER BY "id" + FOR UPDATE + ), gone AS ( + DELETE FROM "VoiceCaptureAudio" + WHERE "captureId" IN (SELECT "id" FROM due) + ) + UPDATE "VoiceCapture" + SET "status" = 'EXPIRED', + "transcriptEnc" = NULL, + "suggestionEnc" = NULL, + "contentDeletedAt" = now(), + "claimedAt" = NULL, + "claimToken" = NULL, + "updatedAt" = now() + WHERE "id" IN (SELECT "id" FROM due)`); +} + +/** + * The retention rules that must not wait for the daily cleanup: audio over + * 24 hours, content past its 7 days (contentExpiresAt), content of + * deactivated users. Pure SQL against the database clock, no key needed, so + * every CronJob run (every 2 minutes) can do it - the cleanup job calls the + * same function. Hooks and writes share one transaction. + */ +export function expireVoiceContent( + db: PrismaService, + options: VoiceRetentionOptions = {}, +): Promise { + const filter = options.filter + ? Prisma.sql`AND ${options.filter}` + : Prisma.empty; + return db.$transaction(async (tx) => { + const audioExpired = await expireAudio(tx, filter); + const contentExpired = await wipeContent( + tx, + Prisma.sql`"contentExpiresAt" < now()`, + filter, + ); + const inactiveOwnersWiped = await wipeContent( + tx, + Prisma.sql`"ownerUserId" IN (SELECT "id" FROM "User" WHERE "isActive" = false)`, + filter, + ); + return { audioExpired, contentExpired, inactiveOwnersWiped }; + }); +} + +/** + * Removes successful voice-transcribe runs older than `hours` in which every + * counter is 0 (no work): the CronJob runs every 2 minutes and would add + * 1440 rows a day. Runs with work, and failed runs, stay with the normal + * 90 days. `filter` narrows on "JobRun" columns (tests). + */ +export function deleteIdleVoiceRuns( + db: PrismaService, + hours: number, + filter?: Prisma.Sql, +): Promise { + return db.$executeRaw(Prisma.sql` + DELETE FROM "JobRun" + WHERE "job" = 'voice-transcribe' AND "status" = 'SUCCEEDED' + AND "startedAt" < now() - (${hours} * interval '1 hour') + AND "result" IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM jsonb_each_text("result"::jsonb) AS counter + WHERE counter.value <> '0') + ${filter ? Prisma.sql`AND ${filter}` : Prisma.empty}`); +} + +/** Deletes rows (and, by cascade, any audio) older than 90 days. */ +export function deleteOldVoiceRows(db: PrismaService): Promise { + return db.$executeRaw` + DELETE FROM "VoiceCapture" + WHERE "createdAt" < now() - (${VOICE_ROW_RETENTION_DAYS} * interval '1 day')`; +} diff --git a/ApiCore-Backend/src/voice/voice.handlers.spec.ts b/ApiCore-Backend/src/voice/voice.handlers.spec.ts new file mode 100644 index 0000000..934c56d --- /dev/null +++ b/ApiCore-Backend/src/voice/voice.handlers.spec.ts @@ -0,0 +1,194 @@ +import type { Prisma } from '../../generated/prisma'; +import type { AnyDomainEventEnvelope } from '../domain-events/domain-event-types'; +import { domainEventHandlerMetadata } from '../domain-events/on-domain-event.decorator'; +import type { NotificationsService } from '../notifications/notifications.service'; +import { VOICE_ERROR_CODES } from './voice-error-codes'; +import { VOICE_ERROR_TEXTS_DE } from './voice-error-texts'; +import { VoiceEventHandlers } from './voice.handlers'; + +function readyEvent(): AnyDomainEventEnvelope<'VoiceCaptureReady'> { + return { + id: 'ev-1', + type: 'VoiceCaptureReady', + aggregateType: 'VoiceCapture', + aggregateId: 'vc-1', + payload: { captureId: 'vc-1', colonyId: 'col-1' }, + actorUserId: null, + correlationId: 'corr-1', + occurredAt: new Date('2026-10-16T08:00:00.000Z'), + attempts: 0, + }; +} + +function failedEvent( + code: (typeof VOICE_ERROR_CODES)[number], +): AnyDomainEventEnvelope<'VoiceCaptureFailed'> { + return { + ...readyEvent(), + type: 'VoiceCaptureFailed', + payload: { captureId: 'vc-1', code }, + } as unknown as AnyDomainEventEnvelope<'VoiceCaptureFailed'>; +} + +describe('VoiceEventHandlers', () => { + let tx: { voiceCapture: { findUnique: jest.Mock } }; + let notifications: { + record: jest.Mock; + recordForPermissionOnce: jest.Mock; + }; + let handlers: VoiceEventHandlers; + const asTx = () => tx as unknown as Prisma.TransactionClient; + + beforeEach(() => { + tx = { + voiceCapture: { + findUnique: jest.fn().mockResolvedValue({ + ownerUserId: 'u-1', + colonyId: 'col-1', + status: 'READY', + colony: { code: 'Volk 7' }, + }), + }, + }; + notifications = { + record: jest.fn().mockResolvedValue({ id: 'n-1' }), + recordForPermissionOnce: jest.fn().mockResolvedValue([]), + }; + handlers = new VoiceEventHandlers( + notifications as unknown as NotificationsService, + ); + }); + + it('writes one HINWEIS for the owner only once per capture', async () => { + await handlers.notifyReady(readyEvent(), asTx()); + + expect(notifications.record).toHaveBeenCalledTimes(1); + expect(notifications.record).toHaveBeenCalledWith( + { + userId: 'u-1', + priority: 'HINWEIS', + category: 'SUGGESTION', + message: 'Sprachnotiz für Volk 7 ist fertig', + relatedType: 'VoiceCapture', + relatedId: 'vc-1', + link: '/inspections/new?colonyId=col-1&voice=vc-1', + requiredPermissions: ['voice-captures:use'], + dedupeKey: 'voice-ready:vc-1', + }, + tx, + ); + }); + + it('reads the colony name from the database, not from the payload', async () => { + await handlers.notifyReady(readyEvent(), asTx()); + + expect(tx.voiceCapture.findUnique).toHaveBeenCalledWith({ + where: { id: 'vc-1' }, + select: { + ownerUserId: true, + colonyId: true, + status: true, + colony: { select: { code: true } }, + }, + }); + expect(Object.keys(readyEvent().payload).sort()).toEqual([ + 'captureId', + 'colonyId', + ]); + }); + + function withStatus(status: string): void { + tx.voiceCapture.findUnique.mockResolvedValue({ + ownerUserId: 'u-1', + colonyId: 'col-1', + status, + colony: { code: 'Volk 7' }, + }); + } + + it.each(['CONFIRMED', 'DISCARDED', 'EXPIRED'])( + 'sends no ready notice for a %s note', + async (status) => { + withStatus(status); + await handlers.notifyReady(readyEvent(), asTx()); + expect(notifications.record).not.toHaveBeenCalled(); + }, + ); + + it.each(['CONFIRMED', 'DISCARDED', 'TRANSCRIBED'])( + 'sends no failure notice to the user for a %s note', + async (status) => { + withStatus(status); + await handlers.notifyFailed(failedEvent('NO_SPEECH'), asTx()); + expect(notifications.record).not.toHaveBeenCalled(); + }, + ); + + it('still tells operators when a waiting note reports EXTRACT_DISABLED', async () => { + withStatus('TRANSCRIBED'); + await handlers.notifyFailed(failedEvent('EXTRACT_DISABLED'), asTx()); + expect(notifications.record).not.toHaveBeenCalled(); + expect(notifications.recordForPermissionOnce).toHaveBeenCalledTimes(1); + }); + + it('writes nothing when the capture is gone', async () => { + tx.voiceCapture.findUnique.mockResolvedValue(null); + + await handlers.notifyReady(readyEvent(), asTx()); + await handlers.notifyFailed(failedEvent('NO_SPEECH'), asTx()); + + expect(notifications.record).not.toHaveBeenCalled(); + }); + + it('the failure text names the next step per code', async () => { + withStatus('FAILED'); + await handlers.notifyFailed(failedEvent('NO_SPEECH'), asTx()); + + const [input, passedTx] = notifications.record.mock.calls[0] as [ + Record, + unknown, + ]; + expect(passedTx).toBe(tx); + expect(input).toMatchObject({ + userId: 'u-1', + priority: 'WARNUNG', + category: 'SUGGESTION', + message: + 'Sprachnotiz für Volk 7 konnte nicht ausgewertet werden. Es war nichts zu verstehen. Bitte noch einmal aufnehmen.', + link: '/voice', + requiredPermissions: ['voice-captures:use'], + dedupeKey: 'voice-failed:vc-1:NO_SPEECH', + }); + }); + + it('has a non-empty text for every error code', () => { + for (const code of VOICE_ERROR_CODES) { + expect(VOICE_ERROR_TEXTS_DE[code].length).toBeGreaterThan(0); + } + }); + + it('tells operators only when the evaluation is off, without a user in the text', async () => { + withStatus('FAILED'); + await handlers.notifyFailed(failedEvent('NO_SPEECH'), asTx()); + expect(notifications.recordForPermissionOnce).not.toHaveBeenCalled(); + + await handlers.notifyFailed(failedEvent('EXTRACT_DISABLED'), asTx()); + + expect(notifications.recordForPermissionOnce).toHaveBeenCalledTimes(1); + const [permission, input] = notifications.recordForPermissionOnce.mock + .calls[0] as [string, { message: string }]; + expect(permission).toBe('system:manage'); + expect(input.message).not.toMatch(/Volk|vc-1|u-1/); + }); + + it('registers both handlers under stable ids', () => { + const ready = Reflect.get(VoiceEventHandlers.prototype, 'notifyReady'); + const failed = Reflect.get(VoiceEventHandlers.prototype, 'notifyFailed'); + expect(domainEventHandlerMetadata(ready as object)).toEqual([ + { types: ['VoiceCaptureReady'], handlerId: 'voice.notify-ready' }, + ]); + expect(domainEventHandlerMetadata(failed as object)).toEqual([ + { types: ['VoiceCaptureFailed'], handlerId: 'voice.notify-failed' }, + ]); + }); +}); diff --git a/ApiCore-Backend/src/voice/voice.handlers.ts b/ApiCore-Backend/src/voice/voice.handlers.ts new file mode 100644 index 0000000..5516220 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice.handlers.ts @@ -0,0 +1,108 @@ +import { Injectable } from '@nestjs/common'; +import type { Prisma } from '../../generated/prisma'; +import type { AnyDomainEventEnvelope } from '../domain-events/domain-event-types'; +import { OnDomainEvent } from '../domain-events/on-domain-event.decorator'; +import { NotificationsService } from '../notifications/notifications.service'; +import { VOICE_ERROR_TEXTS_DE } from './voice-error-texts'; + +/** + * Tells the owner of a voice note that it can be reviewed or has failed. + * Pure DB writer through the delivery transaction; the colony name and the + * owner come from the database, the payload carries ids only. Quiet when + * the note is gone; an owner without the voice right gets nothing (the + * notification service skips recipients who lack a required permission). + */ +@Injectable() +export class VoiceEventHandlers { + constructor(private readonly notifications: NotificationsService) {} + + @OnDomainEvent('VoiceCaptureReady', 'voice.notify-ready') + async notifyReady( + event: AnyDomainEventEnvelope<'VoiceCaptureReady'>, + tx: Prisma.TransactionClient, + ): Promise { + const { captureId } = event.payload; + const capture = await this.findCapture(tx, captureId); + // Confirmed or discarded meanwhile: the link would lead nowhere. + if (capture === null || capture.status !== 'READY') { + return; + } + await this.notifications.record( + { + userId: capture.ownerUserId, + priority: 'HINWEIS', + category: 'SUGGESTION', + message: `Sprachnotiz für ${capture.colony.code} ist fertig`, + relatedType: 'VoiceCapture', + relatedId: captureId, + link: `/inspections/new?colonyId=${capture.colonyId}&voice=${captureId}`, + requiredPermissions: ['voice-captures:use'], + dedupeKey: `voice-ready:${captureId}`, + }, + tx, + ); + } + + @OnDomainEvent('VoiceCaptureFailed', 'voice.notify-failed') + async notifyFailed( + event: AnyDomainEventEnvelope<'VoiceCaptureFailed'>, + tx: Prisma.TransactionClient, + ): Promise { + const { captureId, code } = event.payload; + const capture = await this.findCapture(tx, captureId); + if (capture === null) { + return; + } + if (code === 'EXTRACT_DISABLED') { + await this.notifyOperators(tx); + } + // Only a note that ended FAILED gets the user notice (a waiting note + // is not a failure; a confirmed or discarded one has no link target). + if (capture.status !== 'FAILED') { + return; + } + await this.notifications.record( + { + userId: capture.ownerUserId, + priority: 'WARNUNG', + category: 'SUGGESTION', + message: `Sprachnotiz für ${capture.colony.code} konnte nicht ausgewertet werden. ${VOICE_ERROR_TEXTS_DE[code]}`, + relatedType: 'VoiceCapture', + relatedId: captureId, + link: '/voice', + requiredPermissions: ['voice-captures:use'], + dedupeKey: `voice-failed:${captureId}:${code}`, + }, + tx, + ); + } + + // No user, colony or capture in the text: operators only learn that the + // evaluation is off. One unread notice at a time. + private async notifyOperators(tx: Prisma.TransactionClient): Promise { + await this.notifications.recordForPermissionOnce( + 'system:manage', + { + priority: 'HINWEIS', + category: 'SYSTEM', + message: + 'Die Auswertung von Sprachnotizen ist ausgeschaltet. Sprachnotizen bleiben nach dem Abschreiben stehen.', + relatedType: 'VoiceExtraction', + relatedId: 'disabled', + }, + tx, + ); + } + + private findCapture(tx: Prisma.TransactionClient, captureId: string) { + return tx.voiceCapture.findUnique({ + where: { id: captureId }, + select: { + ownerUserId: true, + colonyId: true, + status: true, + colony: { select: { code: true } }, + }, + }); + } +} diff --git a/ApiCore-Backend/src/voice/voice.module.ts b/ApiCore-Backend/src/voice/voice.module.ts index 1f111b6..fe07ea5 100644 --- a/ApiCore-Backend/src/voice/voice.module.ts +++ b/ApiCore-Backend/src/voice/voice.module.ts @@ -3,6 +3,7 @@ import { InspectionCreateGuard } from './captures/inspection-create.guard'; import { VoiceCapturesController } from './captures/voice-captures.controller'; import { VoiceCapturesStatsService } from './captures/voice-captures-stats.service'; import { VoiceCapturesService } from './captures/voice-captures.service'; +import { VoiceHandlersModule } from './voice-handlers.module'; /** * The voice note API (T7). Audit, the permission resolver and the Prisma @@ -10,6 +11,7 @@ import { VoiceCapturesService } from './captures/voice-captures.service'; * (task 8) and the handlers (task 9). */ @Module({ + imports: [VoiceHandlersModule], controllers: [VoiceCapturesController], providers: [ VoiceCapturesService, diff --git a/ApiCore-Backend/src/webhooks/webhook.handlers.spec.ts b/ApiCore-Backend/src/webhooks/webhook.handlers.spec.ts index eb1b961..263b428 100644 --- a/ApiCore-Backend/src/webhooks/webhook.handlers.spec.ts +++ b/ApiCore-Backend/src/webhooks/webhook.handlers.spec.ts @@ -87,7 +87,7 @@ describe('WebhookEventHandlers', () => { }); }); - it('none of the five new events is forwarded to webhooks', async () => { + it('none of the herd and voice events is forwarded to webhooks', async () => { const herdEvents = [ ['ColonyCreated', { apiaryId: 'a1' }], ['ColonyMoved', { fromApiaryId: 'a1', toApiaryId: 'a2' }], @@ -97,6 +97,8 @@ describe('WebhookEventHandlers', () => { 'HiveChanged', { colonyId: 'c1', version: 2, source: 'CARD', inspectionId: null }, ], + ['VoiceCaptureReady', { captureId: 'v1', colonyId: 'c1' }], + ['VoiceCaptureFailed', { captureId: 'v1', code: 'NO_SPEECH' }], ] as const; for (const [type, payload] of herdEvents) { expect(Object.keys(DOMAIN_EVENT_WEBHOOKS)).not.toContain(type); diff --git a/ApiCore-Backend/src/worker/worker.module.ts b/ApiCore-Backend/src/worker/worker.module.ts index d733830..199e8a8 100644 --- a/ApiCore-Backend/src/worker/worker.module.ts +++ b/ApiCore-Backend/src/worker/worker.module.ts @@ -14,6 +14,7 @@ import { RemindersModule } from '../reminders/reminders.module'; import { SuggestionsModule } from '../suggestions/suggestions.module'; import { TasksModule } from '../tasks/tasks.module'; import { TelegramModule } from '../telegram/telegram.module'; +import { VoiceHandlersModule } from '../voice/voice-handlers.module'; import { WebhooksModule } from '../webhooks/webhooks.module'; import { WithdrawalModule } from '../withdrawal/withdrawal.module'; import { WorkerHealthServer } from './worker-health-server.service'; @@ -45,6 +46,7 @@ import { WorkerRuntimeModule } from './worker-runtime.module'; ComplianceModule, RemindersModule, TelegramModule, + VoiceHandlersModule, WorkerRuntimeModule, ], providers: [WorkerHealthServer], diff --git a/ApiCore-Backend/test/system.e2e-spec.ts b/ApiCore-Backend/test/system.e2e-spec.ts index 9f0eb03..aa6ce49 100644 --- a/ApiCore-Backend/test/system.e2e-spec.ts +++ b/ApiCore-Backend/test/system.e2e-spec.ts @@ -3,6 +3,7 @@ import * as bcrypt from 'bcryptjs'; import request from 'supertest'; import { App } from 'supertest/types'; import { DomainEventDispatcher } from '../src/domain-events/domain-event-dispatcher'; +import { JOB_NAMES } from '../src/jobs/job-names'; import { PrismaService } from '../src/prisma/prisma.service'; import { createTestApp } from './utils/create-test-app'; import { login } from './utils/api'; @@ -104,7 +105,7 @@ describe('System API (e2e)', () => { .set(auth(adminToken)) .expect(200); const body = response.body as OverviewBody; - expect(body.jobs).toHaveLength(5); + expect(body.jobs).toHaveLength(JOB_NAMES.length); expect(typeof body.domainEvents.failed).toBe('number'); await request(server()) diff --git a/ApiCore-Backend/test/voice-confirmation.e2e-spec.ts b/ApiCore-Backend/test/voice-confirmation.e2e-spec.ts new file mode 100644 index 0000000..1feae5b --- /dev/null +++ b/ApiCore-Backend/test/voice-confirmation.e2e-spec.ts @@ -0,0 +1,364 @@ +import { INestApplication } from '@nestjs/common'; +import * as bcrypt from 'bcryptjs'; +import request from 'supertest'; +import { App } from 'supertest/types'; +import { berlinToday } from '../src/common/dates'; +import { AuditLogService } from '../src/audit/audit-log.service'; +import { PrismaService } from '../src/prisma/prisma.service'; +import { + sealVoiceBytes, + sealVoiceText, + voiceAad, +} from '../src/voice/core/voice-crypto'; +import { sealTranscript } from '../src/voice/core/transcript'; +import { login } from './utils/api'; +import { createTestApp } from './utils/create-test-app'; + +const SPOKEN = + 'Koenigin gesehen, Stifte da, vier Brutwaben, ruhig, Futter niedrig, Honigraum kommt naechste Woche drauf'; +const WEBM = Buffer.concat([ + Buffer.from([0x1a, 0x45, 0xdf, 0xa3]), + Buffer.from('opus-bytes-of-a-spoken-note'), +]); + +interface Body { + id?: string; + code?: string; +} + +/** + * Parallel-safe: own users, colonies and apiary found by the suffix; every + * query is scoped to them and afterAll deletes by suffix. + */ +describe('Voice capture confirmation in POST /inspections (e2e)', () => { + let app: INestApplication; + let prisma: PrismaService; + const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`; + const password = 'SuperSecret123!'; + const emailOf = (name: string) => `voice-conf-${name}-${suffix}@example.com`; + const tokens: Record = {}; + const userIds: Record = {}; + let apiaryId: string; + let colonyId: string; + let otherColonyId: string; + let keyCounter = 0; + + const nextKey = () => + `voice-conf-${suffix}-${(keyCounter += 1).toString().padStart(4, '0')}`; + const as = (name: string) => ({ Authorization: `Bearer ${tokens[name]}` }); + const colonyIds = () => [colonyId, otherColonyId]; + + async function createUser(name: string): Promise { + const user = await prisma.user.create({ + data: { + email: emailOf(name), + passwordHash: await bcrypt.hash(password, 12), + name: `Voice Conf ${name}`, + role: { connect: { key: 'BEEKEEPER' } }, + }, + }); + userIds[name] = user.id; + tokens[name] = (await login(app, emailOf(name), password)).accessToken; + } + + /** A capture with sealed content and an audio row, as the pipeline leaves it. */ + async function seedCapture( + owner: string, + status: string, + options: { colony?: string } = {}, + ): Promise { + const capture = await prisma.voiceCapture.create({ + data: { + ownerUserId: userIds[owner], + colonyId: options.colony ?? colonyId, + inspectionDate: new Date(`${berlinToday()}T00:00:00.000Z`), + source: 'AUDIO', + status: status as never, + audioMime: 'audio/webm', + audioBytes: WEBM.length, + durationSec: 30, + contentExpiresAt: new Date(Date.now() + 86_400_000), + claimedAt: + status === 'TRANSCRIBING' || status === 'EXTRACTING' + ? new Date() + : null, + claimToken: + status === 'TRANSCRIBING' || status === 'EXTRACTING' + ? 'running-token' + : null, + }, + }); + await prisma.voiceCapture.update({ + where: { id: capture.id }, + data: { + transcriptEnc: sealTranscript(capture.id, { v: 1, text: SPOKEN }), + suggestionEnc: sealVoiceText( + JSON.stringify({ suggestions: [], unclear: [], dropped: 0 }), + voiceAad(capture.id, 'suggestion'), + ), + }, + }); + await prisma.voiceCaptureAudio.create({ + data: { + captureId: capture.id, + dataEnc: new Uint8Array( + sealVoiceBytes(WEBM, voiceAad(capture.id, 'audio')), + ), + }, + }); + return capture.id; + } + + function createInspection( + who: string, + body: Record, + key = nextKey(), + ) { + return request(app.getHttpServer()) + .post('/api/v1/inspections') + .set(as(who)) + .set('Idempotency-Key', key) + .send({ colonyId, date: berlinToday(), ...body }); + } + + const inspectionsOfColony = () => + prisma.inspection.findMany({ where: { colonyId } }); + + beforeAll(async () => { + app = await createTestApp(); + prisma = app.get(PrismaService); + await createUser('alice'); + await createUser('bob'); + apiaryId = ( + await prisma.apiary.create({ + data: { name: `Voice Conf Stand ${suffix}` }, + }) + ).id; + colonyId = ( + await prisma.colony.create({ + data: { + code: `VCONF-${suffix}`, + apiaryId, + colonyType: 'PRODUCTION_COLONY', + }, + }) + ).id; + otherColonyId = ( + await prisma.colony.create({ + data: { + code: `VCONF-B-${suffix}`, + apiaryId, + colonyType: 'PRODUCTION_COLONY', + }, + }) + ).id; + }); + + afterAll(async () => { + const ids = Object.values(userIds); + const inspectionIds = ( + await prisma.inspection.findMany({ + where: { colonyId: { in: colonyIds() } }, + select: { id: true }, + }) + ).map((inspection) => inspection.id); + await prisma.voiceCapture.deleteMany({ + where: { ownerUserId: { in: ids } }, + }); + await prisma.domainEvent.deleteMany({ + where: { + aggregateType: 'Inspection', + aggregateId: { in: inspectionIds }, + }, + }); + await prisma.calendarSyncOutboxEntry.deleteMany({ + where: { sourceType: 'INSPECTION', sourceId: { in: inspectionIds } }, + }); + await prisma.inspection.deleteMany({ + where: { colonyId: { in: colonyIds() } }, + }); + await prisma.suggestion.deleteMany({ + where: { colonyId: { in: colonyIds() } }, + }); + 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.colony.deleteMany({ where: { id: { in: colonyIds() } } }); + await prisma.apiary.deleteMany({ where: { id: apiaryId } }); + await app.close(); + }); + + beforeEach(async () => { + await prisma.voiceCapture.deleteMany({ + where: { ownerUserId: { in: Object.values(userIds) } }, + }); + await prisma.inspection.deleteMany({ + where: { colonyId: { in: colonyIds() } }, + }); + }); + + it('confirms the voice capture and wipes its content in the same transaction', async () => { + const id = await seedCapture('alice', 'READY'); + + const response = await createInspection('alice', { + voiceCaptureId: id, + colonyStrength: 12, + }).expect(201); + + const inspectionId = (response.body as Body).id; + expect(inspectionId).toBeDefined(); + expect(response.body).not.toHaveProperty('voiceCaptureId'); + const row = await prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + expect(row).toMatchObject({ + status: 'CONFIRMED', + inspectionId, + transcriptEnc: null, + suggestionEnc: null, + claimToken: null, + claimedAt: null, + }); + expect(row.contentDeletedAt).not.toBeNull(); + expect( + await prisma.voiceCaptureAudio.count({ where: { captureId: id } }), + ).toBe(0); + const audit = await prisma.auditLog.findMany({ + where: { entityType: 'VoiceCapture', entityId: id }, + }); + expect(audit).toHaveLength(1); + expect(JSON.stringify(audit)).not.toContain('Koenigin'); + }); + + it.each(['QUEUED', 'TRANSCRIBED', 'FAILED'])( + 'confirms a %s capture too', + async (status) => { + const id = await seedCapture('alice', status); + await createInspection('alice', { voiceCaptureId: id }).expect(201); + const row = await prisma.voiceCapture.findUniqueOrThrow({ + where: { id }, + }); + expect(row.status).toBe('CONFIRMED'); + expect(row.transcriptEnc).toBeNull(); + }, + ); + + it.each(['TRANSCRIBING', 'EXTRACTING'])( + 'takes a %s capture away from a running job: its claim is gone', + async (status) => { + const id = await seedCapture('alice', status); + await createInspection('alice', { voiceCaptureId: id }).expect(201); + const row = await prisma.voiceCapture.findUniqueOrThrow({ + where: { id }, + }); + expect(row).toMatchObject({ status: 'CONFIRMED', claimToken: null }); + }, + ); + + it('answers 409 VOICE_CAPTURE_GONE for a discarded, expired or foreign capture and writes no inspection', async () => { + const discarded = await seedCapture('alice', 'DISCARDED'); + const expired = await seedCapture('alice', 'EXPIRED'); + const foreign = await seedCapture('bob', 'READY'); + for (const id of [discarded, expired, foreign, 'does-not-exist']) { + const response = await createInspection('alice', { + voiceCaptureId: id, + }).expect(409); + expect((response.body as Body).code).toBe('VOICE_CAPTURE_GONE'); + } + expect(await inspectionsOfColony()).toHaveLength(0); + const untouched = await prisma.voiceCapture.findUniqueOrThrow({ + where: { id: foreign }, + }); + expect(untouched.status).toBe('READY'); + expect(untouched.transcriptEnc).not.toBeNull(); + }); + + it('answers 409 when the capture belongs to another colony and keeps its content', async () => { + const id = await seedCapture('alice', 'READY', { colony: otherColonyId }); + const response = await createInspection('alice', { + voiceCaptureId: id, + }).expect(409); + expect((response.body as Body).code).toBe('VOICE_CAPTURE_COLONY_MISMATCH'); + expect(await inspectionsOfColony()).toHaveLength(0); + const row = await prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + expect(row.status).toBe('READY'); + expect(row.transcriptEnc).not.toBeNull(); + }); + + it('rolls back inspection and confirmation when the transaction fails after the confirmation ran', async () => { + const id = await seedCapture('alice', 'READY'); + const auditLog: AuditLogService = app.get(AuditLogService); + const original = auditLog.record.bind( + auditLog, + ) as AuditLogService['record']; + // The confirmation's own audit entry is its last write: failing there + // proves the earlier UPDATE and audio DELETE are rolled back with it. + const spy = jest + .spyOn(auditLog, 'record') + .mockImplementation((input, tx) => { + if (input.entityType === 'VoiceCapture') { + return Promise.reject(new Error('boom after confirm')); + } + return original(input, tx); + }); + try { + const response = await createInspection('alice', { + voiceCaptureId: id, + }); + expect(response.status).toBe(500); + } finally { + spy.mockRestore(); + } + expect(await inspectionsOfColony()).toHaveLength(0); + const row = await prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + expect(row).toMatchObject({ status: 'READY', inspectionId: null }); + expect(row.transcriptEnc).not.toBeNull(); + expect(row.contentDeletedAt).toBeNull(); + expect( + await prisma.voiceCaptureAudio.count({ where: { captureId: id } }), + ).toBe(1); + }); + + it('rejects a malformed voiceCaptureId with 400', async () => { + await createInspection('alice', { voiceCaptureId: 'a b/c' }).expect(400); + await createInspection('alice', { voiceCaptureId: 'x'.repeat(41) }).expect( + 400, + ); + }); + + it('replays the stored answer for the same Idempotency-Key without failing on the confirmed capture', async () => { + const id = await seedCapture('alice', 'READY'); + const key = nextKey(); + const first = await createInspection( + 'alice', + { voiceCaptureId: id }, + key, + ).expect(201); + const second = await createInspection( + 'alice', + { voiceCaptureId: id }, + key, + ).expect(201); + expect((second.body as Body).id).toBe((first.body as Body).id); + expect(await inspectionsOfColony()).toHaveLength(1); + }); + + it('answers 409 for a second inspection with a new key on the same capture', async () => { + const id = await seedCapture('alice', 'READY'); + await createInspection('alice', { voiceCaptureId: id }).expect(201); + const response = await createInspection('alice', { + voiceCaptureId: id, + }).expect(409); + expect((response.body as Body).code).toBe('VOICE_CAPTURE_GONE'); + expect(await inspectionsOfColony()).toHaveLength(1); + }); + + it('creates an inspection without voiceCaptureId exactly as before and leaves other captures alone', async () => { + const id = await seedCapture('alice', 'READY'); + await createInspection('alice', { colonyStrength: 10 }).expect(201); + expect(await inspectionsOfColony()).toHaveLength(1); + const row = await prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + expect(row.status).toBe('READY'); + }); +}); diff --git a/ApiCore-Backend/test/voice-transcribe.e2e-spec.ts b/ApiCore-Backend/test/voice-transcribe.e2e-spec.ts new file mode 100644 index 0000000..0cbf098 --- /dev/null +++ b/ApiCore-Backend/test/voice-transcribe.e2e-spec.ts @@ -0,0 +1,530 @@ +import { Logger } from '@nestjs/common'; +import * as bcrypt from 'bcryptjs'; +import { mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { Client } from 'pg'; +import { Prisma } from '../generated/prisma'; +import { WAKE_CHANNELS } from '../src/common/worker-wakeup'; +import { VoiceTranscribeJob } from '../src/jobs/voice-transcribe.job'; +import { runCli } from '../src/jobs/run-cli'; +import { dbSessionConfig } from '../src/prisma/db-session'; +import { PrismaService } from '../src/prisma/prisma.service'; +import { openTranscript } from '../src/voice/core/transcript'; +import { sealVoiceBytes, voiceAad } from '../src/voice/core/voice-crypto'; +import { + deleteIdleVoiceRuns, + deleteOldVoiceRows, +} from '../src/voice/voice-retention'; + +const HOUR_MS = 3_600_000; +const SECRET_TEXT = 'Geheimer Befund Volk sehr unruhig Sperrbrut sichtbar'; + +/** + * The job halves are called directly with a filter on this file's own + * users and a private work dir, so parallel e2e files sharing the database + * are never touched. Everything is created with a unique suffix and + * removed by it in afterAll. + */ +describe('voice-transcribe job (e2e)', () => { + const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`; + const startedAt = new Date(Date.now() - 1000); + let prisma: PrismaService; + let job: VoiceTranscribeJob; + let workDir: string; + let roleId: string; + let userId: string; + let apiaryId: string; + let colonyId: string; + const extraUserIds: string[] = []; + + const mine = Prisma.sql`"ownerUserId" IN (SELECT "id" FROM "User" WHERE "email" LIKE ${`%${suffix}%`})`; + const fetch = () => job.runFetchPhase({ filter: mine }); + const row = (id: string) => + prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + const failedEvents = (id: string) => + prisma.domainEvent.count({ + where: { + aggregateType: 'VoiceCapture', + aggregateId: id, + type: 'VoiceCaptureFailed', + }, + }); + const audioCount = (id: string) => + prisma.voiceCaptureAudio.count({ where: { captureId: id } }); + + async function newUser(label: string, isActive = true): Promise { + const user = await prisma.user.create({ + data: { + email: `voice-tr-${label}-${suffix}@example.com`, + passwordHash: await bcrypt.hash('SuperSecret123!', 4), + name: `Voice transcribe ${label}`, + roleId, + isActive, + }, + }); + extraUserIds.push(user.id); + return user.id; + } + + async function seed( + data: Partial = {}, + audio: Buffer | null = Buffer.from('OggS-fake-audio-bytes'), + audioCreatedAt?: Date, + ): Promise { + const created = await prisma.voiceCapture.create({ + data: { + ownerUserId: userId, + colonyId, + inspectionDate: new Date('2026-10-16'), + source: 'AUDIO', + audioMime: 'audio/ogg', + durationSec: 30, + contentExpiresAt: new Date(Date.now() + 7 * 24 * HOUR_MS), + ...data, + }, + }); + if (audio) { + await prisma.voiceCaptureAudio.create({ + data: { + captureId: created.id, + dataEnc: Uint8Array.from( + sealVoiceBytes(audio, voiceAad(created.id, 'audio')), + ), + ...(audioCreatedAt ? { createdAt: audioCreatedAt } : {}), + }, + }); + } + return created.id; + } + + async function whisperOut(id: string, over: Record = {}) { + await writeFile( + path.join(workDir, 'out', `${id}.json`), + JSON.stringify({ + v: 1, + captureId: id, + text: SECRET_TEXT, + durationSec: 21.4, + language: 'de', + languageProbability: 0.97, + segments: [{ start: 0, end: 4, text: SECRET_TEXT }], + ...over, + }), + ); + } + + beforeAll(async () => { + workDir = await mkdtemp(path.join(tmpdir(), 'voice-tr-e2e-')); + process.env.VOICE_WORK_DIR = workDir; + prisma = new PrismaService(); + await prisma.$connect(); + job = new VoiceTranscribeJob(prisma); + roleId = ( + await prisma.role.create({ + data: { key: `VOICE_TR_${suffix}`, name: `Voice tr ${suffix}` }, + }) + ).id; + userId = await newUser('main'); + apiaryId = ( + await prisma.apiary.create({ data: { name: `Voice tr ${suffix}` } }) + ).id; + colonyId = ( + await prisma.colony.create({ + data: { + code: `VTR-${suffix}`, + apiaryId, + colonyType: 'PRODUCTION_COLONY', + }, + }) + ).id; + }); + + beforeEach(async () => { + const stale = await prisma.voiceCapture.findMany({ + where: { ownerUserId: { in: [userId, ...extraUserIds] } }, + select: { id: true }, + }); + await prisma.domainEvent.deleteMany({ + where: { + aggregateType: 'VoiceCapture', + aggregateId: { in: stale.map((c) => c.id) }, + }, + }); + await prisma.voiceCapture.deleteMany({ + where: { ownerUserId: { in: [userId, ...extraUserIds] } }, + }); + }); + + afterAll(async () => { + const users = await prisma.user.findMany({ + where: { email: { contains: suffix } }, + select: { id: true }, + }); + const captures = await prisma.voiceCapture.findMany({ + where: { ownerUserId: { in: users.map((u) => u.id) } }, + select: { id: true }, + }); + await prisma.domainEvent.deleteMany({ + where: { + aggregateType: 'VoiceCapture', + aggregateId: { in: captures.map((c) => c.id) }, + }, + }); + await prisma.voiceCapture.deleteMany({ + where: { ownerUserId: { in: users.map((u) => u.id) } }, + }); + await prisma.user.deleteMany({ + where: { id: { in: users.map((u) => u.id) } }, + }); + await prisma.role.deleteMany({ where: { id: roleId } }); + await prisma.colony.deleteMany({ where: { id: colonyId } }); + await prisma.apiary.deleteMany({ where: { id: apiaryId } }); + await prisma.jobRun.deleteMany({ + where: { job: 'voice-transcribe', startedAt: { gte: startedAt } }, + }); + delete process.env.VOICE_WORK_DIR; + await rm(workDir, { recursive: true, force: true }); + await prisma.$disconnect(); + }); + + it('fetch writes one file per claimed capture and a manifest', async () => { + const a = await seed({}, Buffer.from('OggS-AAA')); + const b = await seed({}, Buffer.from('OggS-BBB')); + + const result = await fetch(); + + expect(result).toMatchObject({ claimed: 2, written: 2 }); + expect((await readdir(path.join(workDir, 'in'))).sort()).toEqual( + [`${a}.ogg`, `${b}.ogg`, 'manifest.json'].sort(), + ); + expect(await readFile(path.join(workDir, 'in', `${a}.ogg`))).toEqual( + Buffer.from('OggS-AAA'), + ); + expect((await row(a)).status).toBe('TRANSCRIBING'); + }); + + it('fetch ends quickly without work and writes no manifest', async () => { + const start = Date.now(); + const result = await fetch(); + + expect(result).toMatchObject({ claimed: 0, written: 0 }); + expect(Date.now() - start).toBeLessThan(5000); + expect(await readdir(path.join(workDir, 'in'))).toEqual([]); + }); + + it('fetch counts a missing audio row as a failed attempt', async () => { + const id = await seed({}, null); + + const result = await fetch(); + + expect(result).toMatchObject({ claimed: 1, written: 0 }); + const after = await row(id); + expect(after.status).toBe('QUEUED'); + expect(after.attempts).toBe(1); + expect(after.lastErrorCode).toBe('TRANSCRIBE_FAILED'); + }); + + it('ingest stores TRANSCRIBED, deletes the audio and wakes the worker in one transaction', async () => { + const id = await seed(); + await fetch(); + await whisperOut(id); + const listener = new Client(dbSessionConfig(process.env.DATABASE_URL)); + await listener.connect(); + const woken = new Promise((resolve) => + listener.on('notification', (message) => resolve(message.channel)), + ); + await listener.query(`LISTEN ${WAKE_CHANNELS.voiceExtraction}`); + + try { + const result = await job.ingest(); + + expect(result).toMatchObject({ ingested: 1, failed: 0, lost: 0 }); + expect(await woken).toBe(WAKE_CHANNELS.voiceExtraction); + } finally { + await listener.end(); + } + const after = await row(id); + expect(after.status).toBe('TRANSCRIBED'); + expect(after.attempts).toBe(0); + expect(after.claimToken).toBeNull(); + expect(after.durationSec).toBe(21); + expect(await audioCount(id)).toBe(0); + expect(openTranscript(id, after.transcriptEnc as string).text).toBe( + SECRET_TEXT, + ); + expect(after.transcriptEnc).not.toContain('Geheimer'); + expect(await readdir(path.join(workDir, 'in'))).toEqual([]); + expect(await readdir(path.join(workDir, 'out'))).toEqual([]); + }); + + it('ingest does nothing when the claim token no longer matches', async () => { + const id = await seed(); + await fetch(); + await whisperOut(id); + await prisma.voiceCapture.update({ + where: { id }, + data: { claimToken: 'another-token' }, + }); + + const result = await job.ingest(); + + expect(result).toMatchObject({ ingested: 0, lost: 1 }); + const after = await row(id); + expect(after.status).toBe('TRANSCRIBING'); + expect(after.transcriptEnc).toBeNull(); + expect(await audioCount(id)).toBe(1); + }); + + it('a confirmed capture is not overwritten by a late ingest', async () => { + const id = await seed(); + await fetch(); + await whisperOut(id); + await prisma.voiceCapture.update({ + where: { id }, + data: { status: 'DISCARDED', claimToken: null, claimedAt: null }, + }); + + const result = await job.ingest(); + + expect(result).toMatchObject({ ingested: 0, lost: 1 }); + const after = await row(id); + expect(after.status).toBe('DISCARDED'); + expect(after.transcriptEnc).toBeNull(); + }); + + it.each(['NO_SPEECH', 'TOO_LONG'] as const)( + '%s ends FAILED with the code and deletes the audio', + async (code) => { + const id = await seed(); + await fetch(); + await whisperOut(id, { error: code, text: '', segments: [] }); + + const result = await job.ingest(); + + expect(result).toMatchObject({ failed: 1 }); + const after = await row(id); + expect(after.status).toBe('FAILED'); + expect(after.lastErrorCode).toBe(code); + expect(await audioCount(id)).toBe(0); + expect(await failedEvents(id)).toBe(1); + }, + ); + + it('a missing result counts as a failed attempt', async () => { + const id = await seed(); + await fetch(); + + const result = await job.ingest(); + + expect(result).toMatchObject({ retried: 1 }); + const after = await row(id); + expect(after.status).toBe('QUEUED'); + expect(after.attempts).toBe(1); + expect(after.lastErrorCode).toBe('TRANSCRIBE_FAILED'); + expect(await audioCount(id)).toBe(1); + }); + + it('the third failed attempt ends FAILED with TRANSCRIBE_FAILED', async () => { + const id = await seed({ attempts: 2 }); + await fetch(); + await whisperOut(id, { error: 'DECODE_FAILED', text: '', segments: [] }); + + const result = await job.ingest(); + + expect(result).toMatchObject({ failed: 1 }); + expect((await row(id)).status).toBe('FAILED'); + expect((await row(id)).lastErrorCode).toBe('TRANSCRIBE_FAILED'); + expect(await failedEvents(id)).toBe(1); + }); + + it('writes no failure event for a retry or a lost claim', async () => { + const retried = await seed(); + const lost = await seed(); + await fetch(); + await whisperOut(lost); + await prisma.voiceCapture.update({ + where: { id: lost }, + data: { claimToken: 'another-token' }, + }); + + await job.ingest(); + + expect((await row(retried)).status).toBe('QUEUED'); + expect(await failedEvents(retried)).toBe(0); + expect(await failedEvents(lost)).toBe(0); + }); + + it('writes one failure event when a stale claim reaches its last attempt', async () => { + const id = await seed({ + status: 'TRANSCRIBING', + attempts: 2, + claimToken: 'old', + claimedAt: new Date(Date.now() - 40 * 60_000), + }); + + await fetch(); + await fetch(); + + const after = await row(id); + expect(after.status).toBe('FAILED'); + expect(after.lastErrorCode).toBe('TRANSCRIBE_FAILED'); + expect(await failedEvents(id)).toBe(1); + }); + + it('expires audio after 24 h within one CronJob run', async () => { + const old = new Date(Date.now() - 25 * HOUR_MS); + const id = await seed({ createdAt: old }, undefined, old); + + const result = await fetch(); + + expect(result).toMatchObject({ audioExpired: 1, claimed: 0 }); + const after = await row(id); + expect(after.status).toBe('FAILED'); + expect(after.lastErrorCode).toBe('AUDIO_EXPIRED'); + expect(await audioCount(id)).toBe(0); + expect(await failedEvents(id)).toBe(1); + }); + + it('wipes content after 7 days and marks EXPIRED', async () => { + const id = await seed( + { + status: 'READY', + transcriptEnc: 'x', + suggestionEnc: 'y', + contentExpiresAt: new Date(Date.now() - HOUR_MS), + }, + null, + ); + + const result = await fetch(); + + expect(result).toMatchObject({ contentExpired: 1 }); + const after = await row(id); + expect(after.status).toBe('EXPIRED'); + expect(after.transcriptEnc).toBeNull(); + expect(after.suggestionEnc).toBeNull(); + expect(after.contentDeletedAt).not.toBeNull(); + }); + + it('wipes content of deactivated users', async () => { + const inactive = await newUser('inactive', false); + const id = await seed( + { ownerUserId: inactive, status: 'TRANSCRIBED', transcriptEnc: 'x' }, + null, + ); + const keep = await seed( + { status: 'TRANSCRIBED', transcriptEnc: 'x' }, + null, + ); + + const result = await fetch(); + + expect(result).toMatchObject({ inactiveOwnersWiped: 1 }); + expect((await row(id)).transcriptEnc).toBeNull(); + expect((await row(id)).status).toBe('EXPIRED'); + expect((await row(keep)).transcriptEnc).toBe('x'); + }); + + it('releases a stale claim on the next fetch', async () => { + const id = await seed({ + status: 'TRANSCRIBING', + claimToken: 'old', + claimedAt: new Date(Date.now() - 40 * 60_000), + }); + + await fetch(); + + const after = await row(id); + expect(after.attempts).toBe(1); + expect(after.status).toBe('TRANSCRIBING'); + expect(after.claimToken).not.toBe('old'); + }); + + it('deletes rows after 90 days', async () => { + const id = await seed( + { createdAt: new Date(Date.now() - 91 * 24 * HOUR_MS) }, + null, + ); + + expect(await deleteOldVoiceRows(prisma)).toBeGreaterThanOrEqual(1); + expect(await prisma.voiceCapture.count({ where: { id } })).toBe(0); + }); + + it('prunes only old successful voice-transcribe runs without any work', async () => { + const old = new Date(Date.now() - 48 * HOUR_MS); + const make = (data: Omit) => + prisma.jobRun + .create({ data: { job: 'voice-transcribe', startedAt: old, ...data } }) + .then((r) => r.id); + const idle = await make({ + status: 'SUCCEEDED', + result: { claimed: 0, written: 0 }, + }); + const busy = await make({ + status: 'SUCCEEDED', + result: { claimed: 2, written: 0 }, + }); + const failed = await make({ status: 'FAILED', result: { claimed: 0 } }); + const recent = await prisma.jobRun + .create({ + data: { + job: 'voice-transcribe', + status: 'SUCCEEDED', + result: { claimed: 0 }, + }, + }) + .then((r) => r.id); + const ids = [idle, busy, failed, recent]; + + const removed = await deleteIdleVoiceRuns( + prisma, + 24, + Prisma.sql`"id" IN (${Prisma.join(ids)})`, + ); + + expect(removed).toBe(1); + const left = await prisma.jobRun.findMany({ where: { id: { in: ids } } }); + await prisma.jobRun.deleteMany({ where: { id: { in: ids } } }); + expect(left.map((r) => r.id).sort()).toEqual([busy, failed, recent].sort()); + }); + + it('a measured duration over 240 s ends TOO_LONG instead of being clamped', async () => { + const id = await seed(); + await fetch(); + await whisperOut(id, { durationSec: 240.6 }); + + await job.ingest(); + + expect((await row(id)).status).toBe('FAILED'); + expect((await row(id)).lastErrorCode).toBe('TOO_LONG'); + expect(await audioCount(id)).toBe(0); + }); + + it('ingest never logs the transcript', async () => { + const spies = (['log', 'warn', 'error', 'debug', 'verbose'] as const).map( + (level) => jest.spyOn(Logger.prototype, level).mockImplementation(), + ); + const id = await seed(); + await fetch(); + await whisperOut(id); + + try { + await job.ingest(); + const logged = JSON.stringify(spies.flatMap((s) => s.mock.calls)); + expect(logged).not.toContain('Geheimer'); + expect(logged).not.toContain('Sperrbrut'); + } finally { + spies.forEach((s) => s.mockRestore()); + } + }); + + it('runs through the CLI: ingest without a manifest succeeds and is recorded', async () => { + expect(await runCli(['voice-transcribe', '--phase=ingest'])).toBe(0); + expect(await runCli(['voice-transcribe'])).toBe(1); + const runs = await prisma.jobRun.findMany({ + where: { job: 'voice-transcribe', startedAt: { gte: startedAt } }, + }); + expect(runs.map((r) => r.status).sort()).toEqual(['FAILED', 'SUCCEEDED']); + }); +});