refactor(core,server): migrate contracts to api-contracts and modularize job handlers

Move shared type contracts from `contracts` to new `api-contracts` module in `compute/core` for clearer API boundaries. Remove legacy `contracts`, `pdf`, and `whisper` index files. Update all imports to use `api-contracts`. Refactor server job logic by replacing `parsePdfJob.ts` with modular `user-pdf-layout-job.ts` and `user-whisper-align-job.ts`, updating all relevant API routes and compute integration. This improves maintainability and separation of concerns across compute and server layers.
This commit is contained in:
Richard R 2026-05-24 11:39:25 -06:00
parent e75114a943
commit 12c586d7be
17 changed files with 70 additions and 86 deletions

View file

@ -14,6 +14,7 @@
"exports": { "exports": {
".": "./src/index.ts", ".": "./src/index.ts",
"./local-runtime": "./src/local-runtime.ts", "./local-runtime": "./src/local-runtime.ts",
"./api-contracts": "./src/api-contracts/index.ts",
"./types": "./src/types/index.ts" "./types": "./src/types/index.ts"
} }
} }

View file

@ -18,10 +18,13 @@ export type {
export const ALIGN_QUEUE_NAME = 'whisper-align'; export const ALIGN_QUEUE_NAME = 'whisper-align';
export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout'; export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout';
export interface WhisperAlignJobRequest { export interface WhisperAlignJobBase {
text: string; text: string;
lang?: string; lang?: string;
cacheKey?: string; cacheKey?: string;
}
export interface WhisperAlignJobRequest extends WhisperAlignJobBase {
audioObjectKey: string; audioObjectKey: string;
} }
@ -30,9 +33,12 @@ export interface WhisperAlignJobResult {
timing?: WorkerJobTiming; timing?: WorkerJobTiming;
} }
export interface PdfLayoutJobRequest { export interface PdfLayoutJobBase {
documentId: string; documentId: string;
namespace: string | null; namespace: string | null;
}
export interface PdfLayoutJobRequest extends PdfLayoutJobBase {
documentObjectKey: string; documentObjectKey: string;
} }

View file

@ -1,4 +1,4 @@
export * from './contracts'; export * from './api-contracts';
export { export {
getComputeJobConcurrency, getComputeJobConcurrency,
getAvailableCpuCores, getAvailableCpuCores,

View file

@ -1,3 +0,0 @@
export { renderPage } from './render';
export { mergeTextWithRegions } from './merge';
export { stitchCrossPageBlocks } from './stitch';

View file

@ -15,21 +15,3 @@ export type {
PdfParseProgress, PdfParseProgress,
PdfParseStatus, PdfParseStatus,
} from './parsed-pdf'; } from './parsed-pdf';
export type {
PdfLayoutJobRequest,
PdfLayoutJobResult,
PdfLayoutProgress,
PdfLayoutProgressPhase,
PdfLayoutOperationRequest,
WhisperAlignJobRequest,
WhisperAlignJobResult,
WhisperAlignOperationRequest,
WorkerJobErrorShape,
WorkerJobState,
WorkerJobStatusResponse,
WorkerJobTiming,
WorkerOperationKind,
WorkerOperationRequest,
WorkerOperationState,
} from '../contracts';

View file

@ -1,18 +0,0 @@
export {
alignAudioWithText,
makeWhisperCacheKey,
type WhisperRequestBody,
} from './align';
export {
ensureWhisperModel,
ensureWhisperArtifacts,
createSingleflightRunner,
type WhisperArtifactSpec,
type WhisperStaticArtifactSpec,
type WhisperFetch,
} from './model';
export { mapWordsToSentenceOffsets, type WhisperWord } from './alignment-map';
export { buildGoertzelCoefficients, goertzelPower } from './spectral';
export { buildWordsFromTimestampedTokens, extractTokenStartTimestamps } from './token-timestamps';

View file

@ -33,18 +33,20 @@ import {
getOnnxThreadsPerJob, getOnnxThreadsPerJob,
withIdleTimeoutAndHardCap, withIdleTimeoutAndHardCap,
withTimeout, withTimeout,
type PdfLayoutJobRequest,
type PdfLayoutJobResult,
type WhisperAlignJobRequest,
type WhisperAlignJobResult,
type WorkerJobErrorShape,
type WorkerJobState,
type WorkerJobTiming,
type WorkerOperationKind,
type WorkerOperationRequest,
type WorkerOperationState,
type PdfLayoutProgress,
} from '@openreader/compute-core'; } from '@openreader/compute-core';
import type {
PdfLayoutJobRequest,
PdfLayoutJobResult,
WhisperAlignJobRequest,
WhisperAlignJobResult,
WorkerJobErrorShape,
WorkerJobState,
WorkerJobTiming,
WorkerOperationKind,
WorkerOperationRequest,
WorkerOperationState,
PdfLayoutProgress,
} from '@openreader/compute-core/api-contracts';
import { GetObjectCommand, PutObjectCommand, S3Client } from '@aws-sdk/client-s3'; import { GetObjectCommand, PutObjectCommand, S3Client } from '@aws-sdk/client-s3';
const JOBS_STREAM_NAME = 'compute_jobs'; const JOBS_STREAM_NAME = 'compute_jobs';

View file

@ -22,9 +22,10 @@ const eslintConfig = [
"@openreader/compute-core/*", "@openreader/compute-core/*",
"!@openreader/compute-core/local-runtime", "!@openreader/compute-core/local-runtime",
"!@openreader/compute-core/types", "!@openreader/compute-core/types",
"!@openreader/compute-core/api-contracts",
], ],
message: message:
"Use '@openreader/compute-core' root imports for light APIs. Allowed subpaths are '@openreader/compute-core/local-runtime' and '@openreader/compute-core/types'.", "Use '@openreader/compute-core' root imports for light APIs. Allowed subpaths are '@openreader/compute-core/local-runtime', '@openreader/compute-core/types', and '@openreader/compute-core/api-contracts'.",
}, },
], ],
}, },

View file

@ -3,7 +3,7 @@ import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db'; import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth'; import { requireAuthContext } from '@/lib/server/auth/auth';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3'; import { isS3Configured } from '@/lib/server/storage/s3';
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';

View file

@ -10,7 +10,7 @@ import {
isMissingBlobError, isMissingBlobError,
isValidDocumentId, isValidDocumentId,
} from '@/lib/server/documents/blobstore'; } from '@/lib/server/documents/blobstore';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { import {
normalizeParseStatus, normalizeParseStatus,
parseDocumentParseState, parseDocumentParseState,

View file

@ -9,7 +9,7 @@ import {
deleteDocumentPreviewRows, deleteDocumentPreviewRows,
enqueueDocumentPreview, enqueueDocumentPreview,
} from '@/lib/server/documents/previews'; } from '@/lib/server/documents/previews';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { deleteDocumentBlob, headDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; import { deleteDocumentBlob, headDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore';
import { import {
normalizeParseStatus, normalizeParseStatus,

View file

@ -31,7 +31,7 @@ import { getClientIp } from '@/lib/server/rate-limit/request-ip';
import { getOrCreateDeviceId, setDeviceIdCookie } from '@/lib/server/rate-limit/device-id'; import { getOrCreateDeviceId, setDeviceIdCookie } from '@/lib/server/rate-limit/device-id';
import { buildDailyQuotaExceededResponse } from '@/lib/server/rate-limit/problem-response'; import { buildDailyQuotaExceededResponse } from '@/lib/server/rate-limit/problem-response';
import { getUpstreamRetryAfterSeconds, getUpstreamStatus } from '@/lib/server/tts/upstream-response'; import { getUpstreamRetryAfterSeconds, getUpstreamStatus } from '@/lib/server/tts/upstream-response';
import { getCompute } from '@/lib/server/compute'; import { userWhisperAlignJob } from '@/lib/server/jobs/user-whisper-align-job';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { resolveTtsModelForProvider } from '@/lib/shared/tts-provider-policy'; import { resolveTtsModelForProvider } from '@/lib/shared/tts-provider-policy';
import { resolveSegmentAudioUrls } from '@/lib/server/tts/segment-audio-urls'; import { resolveSegmentAudioUrls } from '@/lib/server/tts/segment-audio-urls';
@ -154,11 +154,6 @@ export async function POST(request: NextRequest) {
let deviceIdToSet: string | null = null; let deviceIdToSet: string | null = null;
const requestId = randomUUID(); const requestId = randomUUID();
const requestStartedAt = Date.now(); const requestStartedAt = Date.now();
let computeBackendPromise: ReturnType<typeof getCompute> | null = null;
const getComputeBackend = async () => {
if (!computeBackendPromise) computeBackendPromise = getCompute();
return computeBackendPromise;
};
try { try {
if (!isS3Configured()) return s3NotConfiguredResponse(); if (!isS3Configured()) return s3NotConfiguredResponse();
@ -395,13 +390,12 @@ export async function POST(request: NextRequest) {
if (!alignment && !request.signal.aborted) { if (!alignment && !request.signal.aborted) {
try { try {
const alignStartedAt = Date.now(); const alignStartedAt = Date.now();
const computeBackend = await getComputeBackend(); alignment = await userWhisperAlignJob({
const aligned = (await computeBackend.alignWords({
audioObjectKey: existing.audioKey, audioObjectKey: existing.audioKey,
text: segment.text, text: segment.text,
})).alignments; sentenceIndex: segment.original.segmentIndex,
});
stageTimings.selfHealAlignMs = Date.now() - alignStartedAt; stageTimings.selfHealAlignMs = Date.now() - alignStartedAt;
alignment = aligned[0] ? { ...aligned[0], sentenceIndex: segment.original.segmentIndex } : null;
if (alignment) { if (alignment) {
await db await db
@ -649,13 +643,12 @@ export async function POST(request: NextRequest) {
try { try {
failedStage = 'whisper.align'; failedStage = 'whisper.align';
const alignStartedAt = Date.now(); const alignStartedAt = Date.now();
const computeBackend = await getComputeBackend(); alignment = await userWhisperAlignJob({
const aligned = (await computeBackend.alignWords({
audioObjectKey: audioKey, audioObjectKey: audioKey,
text: segment.text, text: segment.text,
})).alignments; sentenceIndex: segment.original.segmentIndex,
});
stageTimings.whisperAlignMs = Date.now() - alignStartedAt; stageTimings.whisperAlignMs = Date.now() - alignStartedAt;
alignment = aligned[0] ? { ...aligned[0], sentenceIndex: segment.original.segmentIndex } : null;
} catch (alignError) { } catch (alignError) {
const aborted = isAbortLikeError(alignError) || request.signal.aborted; const aborted = isAbortLikeError(alignError) || request.signal.aborted;
const log = aborted ? console.info : console.warn; const log = aborted ? console.info : console.warn;

View file

@ -6,8 +6,8 @@ import {
getComputeTimeoutConfig, getComputeTimeoutConfig,
withIdleTimeoutAndHardCap, withIdleTimeoutAndHardCap,
withTimeout, withTimeout,
type PdfLayoutProgress,
} from '@openreader/compute-core'; } from '@openreader/compute-core';
import type { PdfLayoutProgress } from '@openreader/compute-core/api-contracts';
import { import {
runPdfLayoutFromPdfBuffer, runPdfLayoutFromPdfBuffer,
runWhisperAlignmentFromAudioBuffer, runWhisperAlignmentFromAudioBuffer,

View file

@ -2,17 +2,14 @@ import type {
TTSAudioBuffer, TTSAudioBuffer,
TTSSentenceAlignment, TTSSentenceAlignment,
ParsedPdfDocument, ParsedPdfDocument,
PdfLayoutProgress,
} from '@openreader/compute-core/types'; } from '@openreader/compute-core/types';
import type { PdfLayoutProgress, WhisperAlignJobBase } from '@openreader/compute-core/api-contracts';
export type ComputeMode = 'local' | 'worker'; export type ComputeMode = 'local' | 'worker';
export interface WhisperAlignInput { export interface WhisperAlignInput extends WhisperAlignJobBase {
audioBuffer?: TTSAudioBuffer; audioBuffer?: TTSAudioBuffer;
audioObjectKey?: string; audioObjectKey?: string;
text: string;
cacheKey?: string;
lang?: string;
} }
export interface WhisperAlignResult { export interface WhisperAlignResult {

View file

@ -7,7 +7,7 @@ import type {
WhisperAlignJobRequest, WhisperAlignJobRequest,
WhisperAlignJobResult, WhisperAlignJobResult,
WorkerOperationState, WorkerOperationState,
} from '@openreader/compute-core/types'; } from '@openreader/compute-core/api-contracts';
class WorkerHttpError extends Error { class WorkerHttpError extends Error {
status: number; status: number;

View file

@ -5,22 +5,20 @@ import { documentKey, putParsedDocumentBlob } from '@/lib/server/documents/blobs
import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state'; import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state';
import { getCompute } from '@/lib/server/compute'; import { getCompute } from '@/lib/server/compute';
import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache'; import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache';
import type { PdfLayoutProgress } from '@openreader/compute-core/types'; import type { PdfLayoutJobBase, PdfLayoutProgress } from '@openreader/compute-core/api-contracts';
interface ParsePdfJobInput { type UserPdfLayoutJobRequest = PdfLayoutJobBase & {
documentId: string;
userId: string; userId: string;
namespace: string | null;
forceToken?: string; forceToken?: string;
} };
const running = new Set<string>(); const running = new Set<string>();
function keyFor(input: ParsePdfJobInput): string { function keyFor(input: UserPdfLayoutJobRequest): string {
return `${input.userId}:${input.documentId}:${input.namespace || ''}`; return `${input.userId}:${input.documentId}:${input.namespace || ''}`;
} }
export async function parsePdfJob(input: ParsePdfJobInput): Promise<void> { export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void> {
const key = keyFor(input); const key = keyFor(input);
if (running.has(key)) return; if (running.has(key)) return;
running.add(key); running.add(key);
@ -137,7 +135,7 @@ export async function parsePdfJob(input: ParsePdfJobInput): Promise<void> {
} }
} }
export function enqueueParsePdfJob(input: ParsePdfJobInput): void { export function enqueueParsePdfJob(input: UserPdfLayoutJobRequest): void {
Promise.resolve() Promise.resolve()
.then(() => parsePdfJob(input)) .then(() => parsePdfJob(input))
.catch((error) => { .catch((error) => {

View file

@ -0,0 +1,25 @@
import { getCompute } from '@/lib/server/compute';
import type { WhisperAlignJobRequest } from '@openreader/compute-core/api-contracts';
import type { TTSSentenceAlignment } from '@/types/tts';
export type UserWhisperAlignJobRequest = WhisperAlignJobRequest & {
sentenceIndex?: number;
};
export async function userWhisperAlignJob(input: UserWhisperAlignJobRequest): Promise<TTSSentenceAlignment | null> {
const compute = await getCompute();
const { alignments } = await compute.alignWords({
audioObjectKey: input.audioObjectKey,
text: input.text,
cacheKey: input.cacheKey,
lang: input.lang,
});
const first = alignments[0];
if (!first) return null;
if (typeof input.sentenceIndex === 'number' && Number.isFinite(input.sentenceIndex)) {
return { ...first, sentenceIndex: input.sentenceIndex };
}
return first;
}