ApiCore/ApiCore-Backend/test/voice-transcribe.e2e-spec.ts

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<string> {
    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<Prisma.VoiceCaptureUncheckedCreateInput> = {},
    audio: Buffer | null = Buffer.from('OggS-fake-audio-bytes'),
    audioCreatedAt?: Date,
  ): Promise<string> {
    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<string, unknown> = {}) {
    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<string>((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.JobRunUncheckedCreateInput, 'job'>) =>
      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']);
  });
});