From 2a4dc868d93a1ea8d9ab51ae0d2e503413afba79 Mon Sep 17 00:00:00 2001 From: Nick Szaunig Date: Fri, 16 Oct 2026 09:32:16 +0200 Subject: [PATCH] feat: Wiederholregel und Voice-Claims (T7 L2 Task 10) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RetryPolicy mit retryDelayMsFor (bestehende Aufrufer unverändert), Phasen transcribe/extract mit eigenen Wartezeiten (5/15 min bzw. 30 s/2 min, 3 Versuche) und Stale-Fenstern (30 bzw. 5 min). Claims per FOR UPDATE SKIP LOCKED mit claimToken als Compare-and-Set, nur Notizen mit gültigem Inhalt; Erfolg setzt den nächsten Status und nach dem Abschreiben attempts = 0; Fehler meldet 'lost', wenn die Notiz inzwischen bestätigt oder verworfen wurde; Aufschub mit Grund; Stale-Freigabe transaktionsfähig mit den beendeten Notizen als Rückgabe. Eine Quelle für die 14 Fehlercodes. Refs #27 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/common/retry-policy.spec.ts | 27 ++ ApiCore-Backend/src/common/retry-policy.ts | 30 +- .../src/voice/claims/voice-claims.spec.ts | 139 +++++++ .../src/voice/claims/voice-claims.ts | 304 +++++++++++++++ .../src/voice/claims/voice-phases.spec.ts | 27 ++ .../src/voice/claims/voice-phases.ts | 44 +++ .../src/voice/voice-error-codes.ts | 22 ++ ApiCore-Backend/test/voice-claims.e2e-spec.ts | 369 ++++++++++++++++++ 8 files changed, 959 insertions(+), 3 deletions(-) create mode 100644 ApiCore-Backend/src/voice/claims/voice-claims.spec.ts create mode 100644 ApiCore-Backend/src/voice/claims/voice-claims.ts create mode 100644 ApiCore-Backend/src/voice/claims/voice-phases.spec.ts create mode 100644 ApiCore-Backend/src/voice/claims/voice-phases.ts create mode 100644 ApiCore-Backend/src/voice/voice-error-codes.ts create mode 100644 ApiCore-Backend/test/voice-claims.e2e-spec.ts diff --git a/ApiCore-Backend/src/common/retry-policy.spec.ts b/ApiCore-Backend/src/common/retry-policy.spec.ts index b1bf9ad..bdf74b0 100644 --- a/ApiCore-Backend/src/common/retry-policy.spec.ts +++ b/ApiCore-Backend/src/common/retry-policy.spec.ts @@ -1,7 +1,9 @@ import { + DEFAULT_RETRY_POLICY, MAX_ATTEMPTS, RETRY_DELAYS_MS, retryDelayMs, + retryDelayMsFor, STALE_CLAIM_MS, } from './retry-policy'; @@ -26,4 +28,29 @@ describe('retry policy', () => { expect(() => retryDelayMs(attempts)).toThrow(RangeError); }, ); + + it('keeps retryDelayMs identical to the default policy', () => { + expect(DEFAULT_RETRY_POLICY).toEqual({ + delaysMs: RETRY_DELAYS_MS, + maxAttempts: MAX_ATTEMPTS, + staleClaimMs: STALE_CLAIM_MS, + }); + for (const attempts of [1, 2, 3, 4, 5, 6, 9]) { + expect(retryDelayMsFor(DEFAULT_RETRY_POLICY, attempts)).toBe( + retryDelayMs(attempts), + ); + } + }); + + it("retryDelayMsFor uses the policy's delays and repeats the last one", () => { + const policy = { + delaysMs: [1_000, 5_000], + maxAttempts: 4, + staleClaimMs: 60_000, + }; + expect([1, 2, 3].map((n) => retryDelayMsFor(policy, n))).toEqual([ + 1_000, 5_000, 5_000, + ]); + expect(() => retryDelayMsFor(policy, 0)).toThrow(RangeError); + }); }); diff --git a/ApiCore-Backend/src/common/retry-policy.ts b/ApiCore-Backend/src/common/retry-policy.ts index 6b6d866..6b95c98 100644 --- a/ApiCore-Backend/src/common/retry-policy.ts +++ b/ApiCore-Backend/src/common/retry-policy.ts @@ -12,12 +12,36 @@ export const RETRY_DELAYS_MS = [ export const MAX_ATTEMPTS = 8; export const STALE_CLAIM_MS = 300_000; -/** Delay after `attempts` failed attempts (1 = the first failure). */ -export function retryDelayMs(attempts: number): number { +/** + * A retry rule as data, for queues with their own pace (voice notes: 3 + * attempts, minutes instead of hours). `delaysMs[n - 1]` is the wait after + * the n-th failure, the last entry repeats; the `maxAttempts`-th failure + * ends in FAILED; a claim older than `staleClaimMs` belongs to a dead worker. + */ +export interface RetryPolicy { + readonly delaysMs: readonly number[]; + readonly maxAttempts: number; + readonly staleClaimMs: number; +} + +/** The rule of the outboxes (the constants above). */ +export const DEFAULT_RETRY_POLICY: RetryPolicy = { + delaysMs: RETRY_DELAYS_MS, + maxAttempts: MAX_ATTEMPTS, + staleClaimMs: STALE_CLAIM_MS, +}; + +/** Delay after `attempts` failed attempts (1 = the first failure) under `policy`. */ +export function retryDelayMsFor(policy: RetryPolicy, attempts: number): number { if (!Number.isInteger(attempts) || attempts < 1) { throw new RangeError( `Retry delay needs a positive number of failed attempts, got ${attempts}`, ); } - return RETRY_DELAYS_MS[Math.min(attempts, RETRY_DELAYS_MS.length) - 1]; + return policy.delaysMs[Math.min(attempts, policy.delaysMs.length) - 1]; +} + +/** Delay after `attempts` failed attempts (1 = the first failure). */ +export function retryDelayMs(attempts: number): number { + return retryDelayMsFor(DEFAULT_RETRY_POLICY, attempts); } diff --git a/ApiCore-Backend/src/voice/claims/voice-claims.spec.ts b/ApiCore-Backend/src/voice/claims/voice-claims.spec.ts new file mode 100644 index 0000000..66abe2c --- /dev/null +++ b/ApiCore-Backend/src/voice/claims/voice-claims.spec.ts @@ -0,0 +1,139 @@ +import { Prisma } from '../../../generated/prisma'; +import { PrismaService } from '../../prisma/prisma.service'; +import { + claimVoiceCaptures, + deferVoiceClaim, + releaseStaleVoiceClaims, + settleVoiceFailure, + settleVoiceSuccess, + VoiceClaim, +} from './voice-claims'; + +function fakeDb() { + const queryRaw = jest.fn().mockResolvedValue([]); + const executeRaw = jest.fn().mockResolvedValue(1); + return { + db: { + $queryRaw: queryRaw, + $executeRaw: executeRaw, + } as unknown as PrismaService, + queryRaw, + executeRaw, + }; +} + +const claim: VoiceClaim = { + id: 'c1', + claimToken: 't1', + attempts: 0, + ownerUserId: 'u1', + colonyId: 'k1', + audioMime: 'audio/webm', + durationSec: 30, +}; + +function sqlOf(call: unknown): Prisma.Sql { + return (call as unknown[])[0] as Prisma.Sql; +} + +describe('voice claims (SQL shape)', () => { + it('claims with SKIP LOCKED against the database clock', async () => { + const { db, queryRaw } = fakeDb(); + await claimVoiceCaptures(db, 'transcribe', { limit: 5, maxTotalSec: 600 }); + const text = sqlOf(queryRaw.mock.calls[0]).sql; + expect(text).toContain('FOR UPDATE SKIP LOCKED'); + expect(text).toContain('now()'); + expect(text).toContain('gen_random_uuid()'); + }); + + it.each([0, -1, 1.5])('refuses the claim limit %p', async (limit) => { + const { db } = fakeDb(); + await expect(claimVoiceCaptures(db, 'extract', { limit })).rejects.toThrow( + RangeError, + ); + }); + + it('refuses an unknown phase', async () => { + const { db } = fakeDb(); + await expect( + claimVoiceCaptures(db, 'drop' as never, { limit: 1 }), + ).rejects.toThrow(/Unknown voice phase/); + }); + + it('settles by compare-and-set on status and claim token', async () => { + const { db, executeRaw } = fakeDb(); + await expect(settleVoiceSuccess(db, 'transcribe', claim)).resolves.toBe( + true, + ); + const { sql, values } = sqlOf(executeRaw.mock.calls[0]); + expect(sql).toContain('"claimToken"'); + expect(values).toEqual( + expect.arrayContaining(['c1', 't1', 'TRANSCRIBING']), + ); + }); + + it('reports a lost claim when no row changed', async () => { + const { db, executeRaw } = fakeDb(); + executeRaw.mockResolvedValue(0); + await expect(settleVoiceSuccess(db, 'extract', claim)).resolves.toBe(false); + await expect(deferVoiceClaim(db, 'extract', claim, 1000)).resolves.toBe( + false, + ); + }); + + it('decides retry or failed from the attempts before', async () => { + const { db } = fakeDb(); + await expect( + settleVoiceFailure( + db, + 'transcribe', + { ...claim, attempts: 1 }, + 'TRANSCRIBE_FAILED', + ), + ).resolves.toBe('retry'); + await expect( + settleVoiceFailure( + db, + 'transcribe', + { ...claim, attempts: 2 }, + 'TRANSCRIBE_FAILED', + ), + ).resolves.toBe('failed'); + }); + + it('reports a lost failure settle', async () => { + const { db, executeRaw } = fakeDb(); + executeRaw.mockResolvedValue(0); + await expect( + settleVoiceFailure(db, 'extract', claim, 'EXTRACT_RETRY'), + ).resolves.toBe('lost'); + }); + + it('turns the last EXTRACT_RETRY into EXTRACT_FAILED', async () => { + const { db, executeRaw } = fakeDb(); + await settleVoiceFailure( + db, + 'extract', + { ...claim, attempts: 2 }, + 'EXTRACT_RETRY', + ); + const { values } = sqlOf(executeRaw.mock.calls[0]); + expect(values).toContain('EXTRACT_FAILED'); + expect(values).not.toContain('EXTRACT_RETRY'); + }); + + it.each([-1, 1.5])('refuses the defer delay %p', async (delay) => { + const { db } = fakeDb(); + await expect(deferVoiceClaim(db, 'extract', claim, delay)).rejects.toThrow( + RangeError, + ); + }); + + it('releases stale claims with the phase stale window and attempt limit', async () => { + const { db, queryRaw } = fakeDb(); + await releaseStaleVoiceClaims(db, 'extract'); + const { values } = sqlOf(queryRaw.mock.calls[0]); + expect(values).toContain(300_000); + expect(values).toContain(3); + }); +}); diff --git a/ApiCore-Backend/src/voice/claims/voice-claims.ts b/ApiCore-Backend/src/voice/claims/voice-claims.ts new file mode 100644 index 0000000..b561fad --- /dev/null +++ b/ApiCore-Backend/src/voice/claims/voice-claims.ts @@ -0,0 +1,304 @@ +import { Prisma } from '../../../generated/prisma'; +import { retryDelayMsFor } from '../../common/retry-policy'; +import { DbClient } from '../../prisma/db-client'; +import { PrismaService } from '../../prisma/prisma.service'; +import { VoiceErrorCode } from '../voice-error-codes'; +import { VOICE_PHASES, VoicePhase } from './voice-phases'; + +/** + * Claim, settle and stale release for voice notes - the same SQL form as + * outbox-claims.ts (SKIP LOCKED, database clock, closed table map), but with + * two phases that have their own states and retry rule, and a claimToken + * instead of claimedAt as the compare-and-set key: the claim travels across + * processes (fetch -> whisper -> ingest), so the key must be a value both + * halves hold. Settling is a compare-and-set on status + claimToken; a note + * that was confirmed, discarded or released in the meantime is not touched. + * Part 4 adds `VoiceUtterance` to VOICE_CLAIM_TABLES. + */ +export type VoiceClaimTable = 'VoiceCapture'; + +interface VoiceClaimTableSql { + table: Prisma.Sql; + statusType: Prisma.Sql; +} + +/** + * Table and enum names cannot be bound as parameters. They come only from + * this closed map - never from caller input - which is what makes + * Prisma.raw safe here. + */ +export const VOICE_CLAIM_TABLES: Readonly< + Record +> = { + VoiceCapture: { + table: Prisma.raw('"VoiceCapture"'), + statusType: Prisma.raw('"VoiceCaptureStatus"'), + }, +}; + +const TABLE = VOICE_CLAIM_TABLES.VoiceCapture; + +/** Longest wait deferVoiceClaim allows: a day. */ +const MAX_DEFER_MS = 24 * 60 * 60 * 1000; + +/** What a claimed note gives its worker: no content, only what the job needs. */ +export interface VoiceClaim { + id: string; + claimToken: string; + /** Attempts before this claim (0 on the first). */ + attempts: number; + ownerUserId: string; + colonyId: string; + audioMime: string | null; + durationSec: number | null; +} + +export interface VoiceClaimOptions { + limit: number; + /** Total reported seconds of audio per call; the first note is always claimed. */ + maxTotalSec?: number; + /** + * Extra condition, ANDed to the claim (e.g. an owner filter in tests). + * Trusted SQL only: never build it with Prisma.raw from input. + */ + filter?: Prisma.Sql; +} + +/** Codes a job may report for a failed attempt. */ +export type VoiceFailureCode = + 'TRANSCRIBE_FAILED' | 'EXTRACT_RETRY' | 'EXTRACT_INVALID'; + +/** Codes shown while a note waits without having failed. */ +export type VoiceDeferCode = 'EXTRACT_DISABLED' | 'DAILY_LIMIT'; + +/** A note a stale release touched; `code` is set when it ended FAILED. */ +export interface ReleasedVoiceClaim { + id: string; + status: string; + code: VoiceErrorCode | null; +} + +function phaseOf(phase: VoicePhase) { + if (!Object.hasOwn(VOICE_PHASES, phase)) { + throw new Error(`Unknown voice phase "${String(phase)}"`); + } + return VOICE_PHASES[phase]; +} + +function toClaim(row: VoiceClaim): VoiceClaim { + return { + id: row.id, + claimToken: row.claimToken, + attempts: row.attempts, + ownerUserId: row.ownerUserId, + colonyId: row.colonyId, + audioMime: row.audioMime, + durationSec: row.durationSec, + }; +} + +/** + * Claims up to `limit` due notes of the phase whose content has not + * expired, oldest first, each with its own fresh claimToken, and - with + * maxTotalSec - only as many as fit the audio budget (a running sum over + * the locked rows; the first note is always taken so a long note cannot + * starve). "availableAt" is TIMESTAMP(3): compare against now() rounded the + * same way, see claimBatch. Uses the root client: a claim must commit on + * its own. + */ +export async function claimVoiceCaptures( + db: PrismaService, + phase: VoicePhase, + options: VoiceClaimOptions, +): Promise { + const { from, working } = phaseOf(phase); + if (!Number.isInteger(options.limit) || options.limit < 1) { + throw new RangeError( + `Claim limit must be a positive integer, got ${options.limit}`, + ); + } + const budget = options.maxTotalSec; + if (budget !== undefined && (!Number.isInteger(budget) || budget < 1)) { + throw new RangeError( + `Claim audio budget must be a positive integer, got ${budget}`, + ); + } + const filter = options.filter + ? Prisma.sql`AND ${options.filter}` + : Prisma.empty; + const fits = + budget === undefined + ? Prisma.empty + : Prisma.sql`WHERE "position" = 1 OR "runningSec" <= ${budget}`; + const rows = await db.$queryRaw(Prisma.sql` + WITH locked AS ( + SELECT "id", "createdAt", COALESCE("durationSec", 0) AS "sec" + FROM ${TABLE.table} + WHERE "status" = CAST(${from} AS ${TABLE.statusType}) + AND "availableAt" <= CAST(now() AS timestamp(3)) + AND "contentExpiresAt" > now() ${filter} + ORDER BY "createdAt", "id" + LIMIT ${options.limit} + FOR UPDATE SKIP LOCKED + ), ranked AS ( + SELECT "id", + row_number() OVER (ORDER BY "createdAt", "id") AS "position", + sum("sec") OVER (ORDER BY "createdAt", "id") AS "runningSec" + FROM locked + ), claimed AS ( + UPDATE ${TABLE.table} + SET "status" = CAST(${working} AS ${TABLE.statusType}), + "claimedAt" = now(), + "updatedAt" = now(), + "claimToken" = gen_random_uuid()::text + WHERE "id" IN (SELECT "id" FROM ranked ${fits}) + RETURNING "id", "claimToken", "attempts", "ownerUserId", "colonyId", + "audioMime", "durationSec", "createdAt" + ) + SELECT "id", "claimToken", "attempts", "ownerUserId", "colonyId", + "audioMime", "durationSec" + FROM claimed ORDER BY "createdAt", "id"`); + return rows.map(toClaim); +} + +/** + * Settles a claimed note as done: sets the phase's next status + * (TRANSCRIBED or READY), releases the claim and - after transcribing - + * resets attempts to 0 (Ruling 6), so no caller has to remember it. `set` + * holds only the result columns (e.g. `"transcriptEnc" = ...`); it must not + * name status, attempts, claimedAt, claimToken or updatedAt. Pass the + * transaction client to settle together with the result's other writes. + * Returns false if the claim was lost (released as stale, claimed again, + * confirmed or discarded): the current state wins and nothing changed. + */ +export async function settleVoiceSuccess( + tx: DbClient, + phase: VoicePhase, + claim: Pick, + set?: Prisma.Sql, +): Promise { + const { working, done } = phaseOf(phase); + const result = set ? Prisma.sql`${set},` : Prisma.empty; + const attempts = + phase === 'transcribe' ? Prisma.sql`"attempts" = 0,` : Prisma.empty; + const count = await tx.$executeRaw(Prisma.sql` + UPDATE ${TABLE.table} + SET ${result} ${attempts} + "status" = CAST(${done} AS ${TABLE.statusType}), + "updatedAt" = now(), + "claimedAt" = NULL, + "claimToken" = NULL + WHERE "id" = ${claim.id} + AND "status" = CAST(${working} AS ${TABLE.statusType}) + AND "claimToken" = ${claim.claimToken}`); + return count === 1; +} + +/** + * Settles a claimed note after a failed attempt: back to the phase's + * waiting state after the policy delay (database clock), or FAILED once the + * policy's maxAttempts is reached. A last EXTRACT_RETRY ends as + * EXTRACT_FAILED. Returns 'lost' (nothing changed) if the claim is gone: the + * note was confirmed or discarded meanwhile, or a stale release already + * counted this attempt and decided its fate. + */ +export async function settleVoiceFailure( + db: DbClient, + phase: VoicePhase, + claim: VoiceClaim, + code: VoiceFailureCode, +): Promise<'retry' | 'failed' | 'lost'> { + const { from, working, failureCode, policy } = phaseOf(phase); + const attempts = claim.attempts + 1; + const failed = attempts >= policy.maxAttempts; + const storedCode = failed && code === 'EXTRACT_RETRY' ? failureCode : code; + const next = failed + ? Prisma.sql`"status" = CAST('FAILED' AS ${TABLE.statusType})` + : Prisma.sql`"status" = CAST(${from} AS ${TABLE.statusType}), + "availableAt" = now() + (${retryDelayMsFor(policy, attempts)} * interval '1 millisecond')`; + const count = await db.$executeRaw(Prisma.sql` + UPDATE ${TABLE.table} + SET ${next}, + "attempts" = "attempts" + 1, + "updatedAt" = now(), + "claimedAt" = NULL, + "claimToken" = NULL, + "lastErrorCode" = ${storedCode} + WHERE "id" = ${claim.id} + AND "status" = CAST(${working} AS ${TABLE.statusType}) + AND "claimToken" = ${claim.claimToken}`); + if (count !== 1) { + return 'lost'; + } + return failed ? 'failed' : 'retry'; +} + +/** + * Puts a claimed note back to the phase's waiting state, due after + * `delayMs` by the database clock, without counting an attempt: it did not + * fail, it has to wait (daily limit, key missing). `code` is stored in + * lastErrorCode so the user sees why meanwhile. Capped at 24 hours. Returns + * false if the claim was lost. + */ +export async function deferVoiceClaim( + db: DbClient, + phase: VoicePhase, + claim: Pick, + delayMs: number, + code?: VoiceDeferCode, +): Promise { + const { from, working } = phaseOf(phase); + if (!Number.isInteger(delayMs) || delayMs < 0) { + throw new RangeError( + `Defer delay must be a non-negative integer, got ${delayMs}`, + ); + } + const lastError = + code === undefined ? Prisma.empty : Prisma.sql`"lastErrorCode" = ${code},`; + const count = await db.$executeRaw(Prisma.sql` + UPDATE ${TABLE.table} + SET ${lastError} + "status" = CAST(${from} AS ${TABLE.statusType}), + "updatedAt" = now(), + "claimedAt" = NULL, + "claimToken" = NULL, + "availableAt" = now() + (${Math.min(delayMs, MAX_DEFER_MS)} * interval '1 millisecond') + WHERE "id" = ${claim.id} + AND "status" = CAST(${working} AS ${TABLE.statusType}) + AND "claimToken" = ${claim.claimToken}`); + return count === 1; +} + +/** + * Puts notes of the phase whose claim is older than the phase's stale + * window (their worker died) back to the waiting state right away with + * attempts + 1 - or to FAILED (phase failure code) once that reaches + * maxAttempts, so a note that kills its worker cannot loop. Takes any + * client so the release and the VoiceCaptureFailed events of the FAILED + * ones (status FAILED, `code` set) can share one transaction. + */ +export async function releaseStaleVoiceClaims( + db: DbClient, + phase: VoicePhase, + options: { filter?: Prisma.Sql } = {}, +): Promise { + const { from, working, failureCode, policy } = phaseOf(phase); + const filter = options.filter + ? Prisma.sql`AND ${options.filter}` + : Prisma.empty; + return db.$queryRaw(Prisma.sql` + UPDATE ${TABLE.table} + SET "status" = CAST( + CASE WHEN "attempts" + 1 >= ${policy.maxAttempts} THEN 'FAILED' ELSE ${from} END + AS ${TABLE.statusType}), + "attempts" = "attempts" + 1, + "updatedAt" = now(), + "claimedAt" = NULL, + "claimToken" = NULL, + "availableAt" = now(), + "lastErrorCode" = CASE WHEN "attempts" + 1 >= ${policy.maxAttempts} THEN ${failureCode} ELSE "lastErrorCode" END + WHERE "status" = CAST(${working} AS ${TABLE.statusType}) + AND "claimedAt" < now() - (${policy.staleClaimMs} * interval '1 millisecond') ${filter} + RETURNING "id", "status"::text AS "status", + CASE WHEN "status" = 'FAILED' THEN "lastErrorCode" END AS "code"`); +} diff --git a/ApiCore-Backend/src/voice/claims/voice-phases.spec.ts b/ApiCore-Backend/src/voice/claims/voice-phases.spec.ts new file mode 100644 index 0000000..e89408d --- /dev/null +++ b/ApiCore-Backend/src/voice/claims/voice-phases.spec.ts @@ -0,0 +1,27 @@ +import { retryDelayMsFor } from '../../common/retry-policy'; +import { VOICE_PHASES } from './voice-phases'; + +describe('voice phases', () => { + it('transcribes QUEUED to TRANSCRIBING with 3 attempts, 5 min / 15 min and a 30 min stale claim', () => { + const phase = VOICE_PHASES.transcribe; + expect([phase.from, phase.working]).toEqual(['QUEUED', 'TRANSCRIBING']); + expect(phase.policy).toEqual({ + delaysMs: [300_000, 900_000], + maxAttempts: 3, + staleClaimMs: 1_800_000, + }); + expect([1, 2].map((n) => retryDelayMsFor(phase.policy, n))).toEqual([ + 300_000, 900_000, + ]); + }); + + it('extracts TRANSCRIBED to EXTRACTING with 3 attempts, 30 s / 2 min and a 5 min stale claim', () => { + const phase = VOICE_PHASES.extract; + expect([phase.from, phase.working]).toEqual(['TRANSCRIBED', 'EXTRACTING']); + expect(phase.policy).toEqual({ + delaysMs: [30_000, 120_000], + maxAttempts: 3, + staleClaimMs: 300_000, + }); + }); +}); diff --git a/ApiCore-Backend/src/voice/claims/voice-phases.ts b/ApiCore-Backend/src/voice/claims/voice-phases.ts new file mode 100644 index 0000000..3cf4dbc --- /dev/null +++ b/ApiCore-Backend/src/voice/claims/voice-phases.ts @@ -0,0 +1,44 @@ +import { RetryPolicy } from '../../common/retry-policy'; +import { VoiceErrorCode } from '../voice-error-codes'; + +/** + * The two job phases of a voice note. Each has its own states, its own + * retry rule and its own code when the last attempt fails. Both phases share + * one `attempts` counter; settling the transcribe phase resets it to 0 + * (Ruling 6), so the extract phase starts with its own 3 attempts. + */ +export const VOICE_PHASES = { + transcribe: { + from: 'QUEUED', + working: 'TRANSCRIBING', + done: 'TRANSCRIBED', + failureCode: 'TRANSCRIBE_FAILED', + policy: { + delaysMs: [300_000, 900_000], + maxAttempts: 3, + staleClaimMs: 1_800_000, + }, + }, + extract: { + from: 'TRANSCRIBED', + working: 'EXTRACTING', + done: 'READY', + failureCode: 'EXTRACT_FAILED', + policy: { + delaysMs: [30_000, 120_000], + maxAttempts: 3, + staleClaimMs: 300_000, + }, + }, +} as const satisfies Record< + string, + { + from: string; + working: string; + done: string; + failureCode: VoiceErrorCode; + policy: RetryPolicy; + } +>; + +export type VoicePhase = keyof typeof VOICE_PHASES; diff --git a/ApiCore-Backend/src/voice/voice-error-codes.ts b/ApiCore-Backend/src/voice/voice-error-codes.ts new file mode 100644 index 0000000..ffebb85 --- /dev/null +++ b/ApiCore-Backend/src/voice/voice-error-codes.ts @@ -0,0 +1,22 @@ +/** + * The fixed error codes of a voice note. They land in lastErrorCode, in + * events and in API answers - never content, only one of these. + */ +export const VOICE_ERROR_CODES = [ + 'UNSUPPORTED_AUDIO', + 'TOO_LARGE', + 'TOO_MANY_OPEN', + 'NO_SPEECH', + 'TOO_LONG', + 'TRANSCRIBE_FAILED', + 'AUDIO_EXPIRED', + 'EXTRACT_DISABLED', + 'EXTRACT_RETRY', + 'EXTRACT_FAILED', + 'DAILY_LIMIT', + 'EXTRACT_INVALID', + 'REFUSED', + 'VOICE_CAPTURE_GONE', +] as const; + +export type VoiceErrorCode = (typeof VOICE_ERROR_CODES)[number]; diff --git a/ApiCore-Backend/test/voice-claims.e2e-spec.ts b/ApiCore-Backend/test/voice-claims.e2e-spec.ts new file mode 100644 index 0000000..8aa82ec --- /dev/null +++ b/ApiCore-Backend/test/voice-claims.e2e-spec.ts @@ -0,0 +1,369 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import * as bcrypt from 'bcryptjs'; +import { Prisma } from '../generated/prisma'; +import { PrismaModule } from '../src/prisma/prisma.module'; +import { PrismaService } from '../src/prisma/prisma.service'; +import { + claimVoiceCaptures, + deferVoiceClaim, + releaseStaleVoiceClaims, + settleVoiceFailure, + settleVoiceSuccess, +} from '../src/voice/claims/voice-claims'; + +/** + * Every query is scoped to this file's own owner (ownerUserId filter) and + * every row is created and removed here: other suites share the database. + */ +describe('voice claims (e2e)', () => { + const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`; + let moduleRef: TestingModule; + let prisma: PrismaService; + let roleId: string; + let userId: string; + let apiaryId: string; + let colonyId: string; + + const mine = () => Prisma.sql`"ownerUserId" = ${userId}`; + const stale = (phase: 'transcribe' | 'extract') => + releaseStaleVoiceClaims(prisma, phase, { filter: mine() }); + const claim = ( + phase: 'transcribe' | 'extract', + limit: number, + maxTotalSec?: number, + ) => + claimVoiceCaptures(prisma, phase, { limit, maxTotalSec, filter: mine() }); + const TEXT_WAITING = { + status: 'TRANSCRIBED', + source: 'TEXT', + audioMime: null, + durationSec: null, + } as const; + + async function seed( + count: number, + data: Partial = {}, + ): Promise { + const ids: string[] = []; + for (let i = 0; i < count; i++) { + const row = await prisma.voiceCapture.create({ + data: { + ownerUserId: userId, + colonyId, + inspectionDate: new Date('2026-10-16'), + source: 'AUDIO', + audioMime: 'audio/webm', + durationSec: 30, + contentExpiresAt: new Date(Date.now() + 7 * 86_400_000), + ...data, + }, + }); + ids.push(row.id); + } + return ids; + } + + const rowOf = (id: string) => + prisma.voiceCapture.findUniqueOrThrow({ where: { id } }); + + beforeAll(async () => { + moduleRef = await Test.createTestingModule({ + imports: [PrismaModule], + }).compile(); + await moduleRef.init(); + prisma = moduleRef.get(PrismaService); + const role = await prisma.role.create({ + data: { key: `VOICE_CLAIMS_${suffix}`, name: `Voice claims ${suffix}` }, + }); + roleId = role.id; + const user = await prisma.user.create({ + data: { + email: `voice-claims-${suffix}@example.com`, + passwordHash: await bcrypt.hash('SuperSecret123!', 4), + name: 'Voice claims', + roleId, + }, + }); + userId = user.id; + const apiary = await prisma.apiary.create({ + data: { name: `Voice claims ${suffix}` }, + }); + apiaryId = apiary.id; + const colony = await prisma.colony.create({ + data: { + code: `VC-${suffix}`, + apiaryId, + colonyType: 'PRODUCTION_COLONY', + }, + }); + colonyId = colony.id; + }); + + beforeEach(async () => { + await prisma.voiceCapture.deleteMany({ where: { ownerUserId: userId } }); + }); + + afterAll(async () => { + await prisma.voiceCapture.deleteMany({ where: { ownerUserId: userId } }); + await prisma.colony.deleteMany({ where: { id: colonyId } }); + await prisma.apiary.deleteMany({ where: { id: apiaryId } }); + await prisma.user.deleteMany({ where: { id: userId } }); + await prisma.role.deleteMany({ where: { id: roleId } }); + await moduleRef.close(); + }); + + it('fetch claims disjoint sets under SKIP LOCKED', async () => { + const ids = await seed(8); + const claimedBy: string[][] = [[], [], []]; + const worker = async (n: number) => { + for (let round = 0; round < 8; round++) { + const got = await claim('transcribe', 3); + if (got.length === 0) { + return; + } + claimedBy[n].push(...got.map((c) => c.id)); + } + }; + await Promise.all([worker(0), worker(1), worker(2)]); + const all = claimedBy.flat(); + expect(new Set(all).size).toBe(all.length); + expect([...all].sort()).toEqual([...ids].sort()); + const rows = await prisma.voiceCapture.findMany({ + where: { ownerUserId: userId }, + }); + expect(rows.every((r) => r.status === 'TRANSCRIBING')).toBe(true); + expect(rows.every((r) => r.claimToken && r.claimedAt)).toBe(true); + }); + + it('hands out a distinct claim token per capture and the data a phase needs', async () => { + await seed(2, { durationSec: 42 }); + const got = await claim('transcribe', 5); + expect(new Set(got.map((c) => c.claimToken)).size).toBe(2); + expect(got[0]).toMatchObject({ + attempts: 0, + ownerUserId: userId, + colonyId, + audioMime: 'audio/webm', + durationSec: 42, + }); + }); + + it('claims only captures that are due in the phase', async () => { + const [due] = await seed(1); + await seed(1, { availableAt: new Date(Date.now() + 3_600_000) }); + await seed(1, { status: 'TRANSCRIBED' }); + const got = await claim('transcribe', 10); + expect(got.map((c) => c.id)).toEqual([due]); + }); + + it('respects the total seconds of audio per run', async () => { + const ids = await seed(3, { durationSec: 200 }); + const got = await claim('transcribe', 5, 450); + expect(got.map((c) => c.id)).toEqual(ids.slice(0, 2)); + expect((await rowOf(ids[2])).status).toBe('QUEUED'); + }); + + it('still claims the first capture when it alone exceeds the seconds budget', async () => { + await seed(1, { durationSec: 240 }); + expect(await claim('transcribe', 5, 100)).toHaveLength(1); + }); + + it('does not claim a capture whose content has expired', async () => { + await seed(1, { contentExpiresAt: new Date(Date.now() - 1000) }); + expect(await claim('transcribe', 5)).toHaveLength(0); + }); + + it('defers with a code that stays visible and counts no attempt', async () => { + const [id] = await seed(1, TEXT_WAITING); + const [claimed] = await claim('extract', 1); + expect( + await deferVoiceClaim(prisma, 'extract', claimed, 60_000, 'DAILY_LIMIT'), + ).toBe(true); + expect(await rowOf(id)).toMatchObject({ + status: 'TRANSCRIBED', + attempts: 0, + lastErrorCode: 'DAILY_LIMIT', + }); + }); + + it('settles a success, resets attempts after transcribing and clears the claim', async () => { + const [id] = await seed(1, { attempts: 1 }); + const [claimed] = await claim('transcribe', 1); + const ok = await settleVoiceSuccess( + prisma, + 'transcribe', + claimed, + Prisma.sql`"transcriptEnc" = ${'ct'}`, + ); + expect(ok).toBe(true); + expect(await rowOf(id)).toMatchObject({ + status: 'TRANSCRIBED', + claimToken: null, + claimedAt: null, + attempts: 0, + }); + }); + + it('settle does nothing when the claim token no longer matches', async () => { + const [id] = await seed(1); + const [claimed] = await claim('transcribe', 1); + const wrong = { ...claimed, claimToken: 'someone-else' }; + expect(await settleVoiceSuccess(prisma, 'transcribe', wrong)).toBe(false); + expect( + await settleVoiceFailure( + prisma, + 'transcribe', + wrong, + 'TRANSCRIBE_FAILED', + ), + ).toBe('lost'); + expect(await deferVoiceClaim(prisma, 'transcribe', wrong, 1000)).toBe( + false, + ); + expect(await rowOf(id)).toMatchObject({ + status: 'TRANSCRIBING', + claimToken: claimed.claimToken, + attempts: 0, + lastErrorCode: null, + }); + }); + + it('settle does nothing on a capture confirmed meanwhile', async () => { + const [id] = await seed(1); + const [claimed] = await claim('transcribe', 1); + await prisma.voiceCapture.update({ + where: { id }, + data: { + status: 'CONFIRMED', + inspectionId: `insp-${suffix}`, + claimedAt: null, + claimToken: null, + }, + }); + expect(await settleVoiceSuccess(prisma, 'transcribe', claimed)).toBe(false); + expect( + await settleVoiceFailure( + prisma, + 'transcribe', + claimed, + 'TRANSCRIBE_FAILED', + ), + ).toBe('lost'); + expect( + await deferVoiceClaim(prisma, 'transcribe', claimed, 1000, 'DAILY_LIMIT'), + ).toBe(false); + expect(await rowOf(id)).toMatchObject({ + status: 'CONFIRMED', + attempts: 0, + lastErrorCode: null, + }); + }); + + it('retries after the policy delay and the third failure ends FAILED with the code', async () => { + const [id] = await seed(1); + const delays: number[] = []; + for (let round = 1; round <= 3; round++) { + await prisma.voiceCapture.update({ + where: { id }, + data: { availableAt: new Date(Date.now() - 1000) }, + }); + const [claimed] = await claim('transcribe', 1); + expect(claimed.attempts).toBe(round - 1); + const outcome = await settleVoiceFailure( + prisma, + 'transcribe', + claimed, + 'TRANSCRIBE_FAILED', + ); + expect(outcome).toBe(round < 3 ? 'retry' : 'failed'); + const row = await rowOf(id); + if (round < 3) { + delays.push(row.availableAt.getTime() - Date.now()); + expect(row.status).toBe('QUEUED'); + } else { + expect(row).toMatchObject({ + status: 'FAILED', + attempts: 3, + lastErrorCode: 'TRANSCRIBE_FAILED', + claimToken: null, + claimedAt: null, + }); + } + } + expect(delays[0]).toBeGreaterThan(290_000); + expect(delays[0]).toBeLessThanOrEqual(300_500); + expect(delays[1]).toBeGreaterThan(890_000); + expect(delays[1]).toBeLessThanOrEqual(900_500); + }); + + it('ends an extraction that kept being retryable as EXTRACT_FAILED', async () => { + const [id] = await seed(1, { ...TEXT_WAITING, attempts: 2 }); + const [claimed] = await claim('extract', 1); + expect( + await settleVoiceFailure(prisma, 'extract', claimed, 'EXTRACT_RETRY'), + ).toBe('failed'); + expect(await rowOf(id)).toMatchObject({ + status: 'FAILED', + lastErrorCode: 'EXTRACT_FAILED', + }); + }); + + it('defers a claim without counting an attempt', async () => { + const [id] = await seed(1, TEXT_WAITING); + const [claimed] = await claim('extract', 1); + expect(await deferVoiceClaim(prisma, 'extract', claimed, 600_000)).toBe( + true, + ); + const row = await rowOf(id); + expect(row).toMatchObject({ + status: 'TRANSCRIBED', + attempts: 0, + claimToken: null, + }); + expect(row.availableAt.getTime() - Date.now()).toBeGreaterThan(590_000); + expect(await claim('extract', 1)).toHaveLength(0); + }); + + it('stale TRANSCRIBING claims are released with attempts + 1', async () => { + const [id] = await seed(1); + await claim('transcribe', 1); + await prisma.$executeRaw`UPDATE "VoiceCapture" SET "claimedAt" = now() - interval '31 minutes' WHERE "id" = ${id}`; + expect(await stale('transcribe')).toEqual([ + { id, status: 'QUEUED', code: null }, + ]); + expect(await rowOf(id)).toMatchObject({ + status: 'QUEUED', + attempts: 1, + claimToken: null, + claimedAt: null, + }); + }); + + it('keeps a fresh claim and ends a worn-out stale one as FAILED', async () => { + const [fresh] = await seed(1); + const [worn] = await seed(1, { attempts: 2 }); + await claim('transcribe', 2); + await prisma.$executeRaw`UPDATE "VoiceCapture" SET "claimedAt" = now() - interval '29 minutes' WHERE "id" = ${worn}`; + expect(await stale('transcribe')).toEqual([]); + await prisma.$executeRaw`UPDATE "VoiceCapture" SET "claimedAt" = now() - interval '31 minutes' WHERE "id" = ${worn}`; + expect(await stale('transcribe')).toEqual([ + { id: worn, status: 'FAILED', code: 'TRANSCRIBE_FAILED' }, + ]); + expect(await rowOf(worn)).toMatchObject({ + attempts: 3, + lastErrorCode: 'TRANSCRIBE_FAILED', + }); + expect((await rowOf(fresh)).status).toBe('TRANSCRIBING'); + }); + + it('uses the shorter stale window of the extract phase', async () => { + const [id] = await seed(1, TEXT_WAITING); + await claim('extract', 1); + await prisma.$executeRaw`UPDATE "VoiceCapture" SET "claimedAt" = now() - interval '6 minutes' WHERE "id" = ${id}`; + expect(await stale('transcribe')).toEqual([]); + expect(await stale('extract')).toEqual([ + { id, status: 'TRANSCRIBED', code: null }, + ]); + expect((await rowOf(id)).attempts).toBe(1); + }); +});