/** * Intake Service * Handles document ingestion, OCR, classification, and routing */ import Fastify from 'fastify'; import fastifySwagger from '@fastify/swagger'; import fastifySwaggerUI from '@fastify/swagger-ui'; import { errorHandler, createLogger, registerSecurityPlugins, addCorrelationId, addRequestLogging, getEnv, createBodySchema, authenticateJWT, } from '@the-order/shared'; import { CreateDocumentSchema } from '@the-order/schemas'; // import { intakeWorkflow } from '@the-order/workflows'; // Not yet implemented import { WORMStorage } from '@the-order/storage'; import { healthCheck as dbHealthCheck, getPool, createDocument, updateDocument } from '@the-order/database'; import { OCRClient } from '@the-order/ocr'; import { randomUUID } from 'crypto'; const logger = createLogger('intake-service'); // eslint-disable-next-line @typescript-eslint/no-explicit-any const server: any = Fastify({ logger: logger as any, requestIdLogLabel: 'requestId', disableRequestLogging: false, }); // Initialize database pool const env = getEnv(); if (env.DATABASE_URL) { getPool({ connectionString: env.DATABASE_URL }); } // Initialize storage client (WORM mode for document retention) const storageClient = new WORMStorage({ provider: env.STORAGE_TYPE || 's3', bucket: env.STORAGE_BUCKET, region: env.STORAGE_REGION, }); // Initialize OCR client (for future use) const ocrClient = new OCRClient(storageClient); void ocrClient; // Suppress unused variable warning // Initialize server async function initializeServer(): Promise { // Register Swagger const swaggerUrl = env.SWAGGER_SERVER_URL || (env.NODE_ENV === 'development' ? 'http://localhost:4001' : undefined); if (!swaggerUrl) { logger.warn('SWAGGER_SERVER_URL not set, Swagger documentation will not be available'); } else { await server.register(fastifySwagger, { openapi: { info: { title: 'Intake Service API', description: 'Document ingestion, OCR, classification, and routing', version: '1.0.0', }, servers: [ { url: swaggerUrl, description: env.NODE_ENV || 'Development server', }, ], }, }); await server.register(fastifySwaggerUI, { routePrefix: '/docs', }); } await registerSecurityPlugins(server as any); addCorrelationId(server as any); addRequestLogging(server as any); server.setErrorHandler(errorHandler as any); } // Health check server.get( '/health', { schema: { description: 'Health check endpoint', tags: ['health'], response: { 200: { type: 'object', properties: { status: { type: 'string' }, service: { type: 'string' }, database: { type: 'string' }, storage: { type: 'string' }, }, }, }, }, }, async () => { const dbHealthy = await dbHealthCheck(); const storageHealthy = await storageClient.objectExists('health-check').catch(() => false); return { status: dbHealthy && storageHealthy ? 'ok' : 'degraded', service: 'intake', database: dbHealthy ? 'connected' : 'disconnected', storage: storageHealthy ? 'accessible' : 'unavailable', }; } ); // Ingest endpoint server.post( '/ingest', { preHandler: [authenticateJWT], schema: { ...createBodySchema(CreateDocumentSchema), description: 'Ingest a document for processing', tags: ['documents'], response: { 202: { type: 'object', properties: { documentId: { type: 'string', format: 'uuid' }, status: { type: 'string' }, message: { type: 'string' }, }, }, }, }, }, // eslint-disable-next-line @typescript-eslint/no-explicit-any async (request: any, reply: any) => { const body = request.body as { title: string; type: string; content?: string; fileUrl?: string; }; const documentId = randomUUID(); const userId = request.user?.id || 'system'; // Upload to WORM storage if content provided let fileUrl = body.fileUrl; let storageKey: string | undefined; if (body.content) { storageKey = `documents/${documentId}`; await storageClient.upload({ key: storageKey, content: Buffer.from(body.content), contentType: 'application/pdf', metadata: { title: body.title, type: body.type, userId, }, }); fileUrl = storageKey; } // Create document record const document = await createDocument({ title: body.title, type: body.type, file_url: fileUrl, storage_key: storageKey, user_id: userId, status: 'processing', }); // Trigger intake workflow (commented out until implemented) // const workflowResult = await intakeWorkflow( // { // documentId: document.id, // fileUrl: fileUrl || '', // userId, // }, // ocrClient, // storageClient // ); // Placeholder workflow result const workflowResult = { status: 'processing', classification: body.type || 'unknown', extractedData: body.content ? JSON.parse(body.content) : {}, }; // Update document with workflow results await updateDocument(document.id, { status: 'processed', classification: workflowResult.classification, ocr_text: typeof workflowResult.extractedData === 'object' && workflowResult.extractedData !== null ? (workflowResult.extractedData as { ocrText?: string }).ocrText : undefined, extracted_data: workflowResult.extractedData, }); return reply.status(202).send({ documentId: document.id, status: 'processing', message: 'Document ingestion started', classification: workflowResult.classification, }); } ); // Start server const start = async () => { try { await initializeServer(); const env = getEnv(); const port = env.PORT || 4001; await server.listen({ port, host: '0.0.0.0' }); logger.info({ port }, 'Intake service listening'); } catch (err) { logger.error({ err }, 'Failed to start server'); process.exit(1); } }; start();