/** * Translation Jobs Routes * * Asynchronous translation job management (for content up to 50000 characters) */ import { Router, Request, Response } from 'express'; import { z } from 'zod'; import { BulkContentQueuePayload, LanguageCodeSchema, ToneSchema } from '../types'; import { authenticateApiKey } from '../middleware/auth'; import { apiRateLimiter } from '../middleware/rateLimiter'; import { validate, validateParams, validateQuery } from '../middleware/validator'; import { transformWordPressJobPayload } from '../middleware/payloadTransform'; import { PrismaClient } from '@prisma/client'; import type { Prisma } from '@prisma/client'; import { translationService } from '../services/translationService'; import { ExceptionRulesSchema } from '../services/exceptionService'; import { hasSufficientCredits, getCurrentBalance } from '../services/creditService'; import { countSourceCharacters } from '../utils/tokenCalculation'; import { StoredBulkQueuePayload, drainBulkQueue } from '../queue/bulkQueueDispatcher'; import { logger } from '../utils/logger'; import { config } from '../config'; import { BULK_ASYNC_MAX_STRINGS, BULK_MAX_CHARS_PER_STRING, BULK_MAX_TARGET_LANGS, BULK_REQUEST_CHAR_BUDGET, } from '../config/batchLimits'; import { BULK_CONTENT_MAX_ITEMS } from '../types'; import { mergeStructuredFields, countStructuredCharacters, STRUCTURED_FIELD_KEY_PATTERN } from '../utils/structuredFields'; import type { StructuredFields } from '../utils/structuredFields'; import crypto from 'crypto'; import { normalizeClientJobId, SITE_CONTENT_CONTRACT_VERSION, SITE_CONTENT_MAX_SEGMENTS, validateSiteContentRequest, } from '../services/siteContentTranslation'; const prisma = new PrismaClient(); const router = Router(); async function resolveSubmissionIdForJob( userId: string, jobId: string, candidate?: string ): Promise { if (candidate) { return candidate; } const identity = await prisma.translationJob.findFirst({ where: { id: jobId, user_id: userId }, select: { submission_id: true }, }); return identity?.submission_id ?? undefined; } // Validation schemas const clientJobIdSchema = z.string().transform((value, ctx) => { try { return normalizeClientJobId(value); } catch (error) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: error instanceof Error ? error.message : 'Invalid clientJobId', }); return z.NEVER; } }); const legacyJobSubmitSchema = z.object({ sourceLang: LanguageCodeSchema, targetLang: LanguageCodeSchema, title: z.string().optional(), excerpt: z.string().optional(), content: z.string().optional(), fields: z.record( z.string().regex(STRUCTURED_FIELD_KEY_PATTERN, 'Invalid structured field key'), z.string() ).optional(), tone: ToneSchema.optional(), exceptions: ExceptionRulesSchema, callbackUrl: z.string().url().optional(), callbackSecret: z.string().optional(), clientJobId: clientJobIdSchema.optional(), resourceType: z.never().optional(), segments: z.never().optional(), }).superRefine((data, ctx) => { if (data.fields !== undefined && Object.keys(data.fields).length === 0 && !data.title && !data.excerpt && !data.content) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: 'At least one translatable field must be provided' }); return; } if (!data.title && !data.excerpt && !data.content && !data.fields) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: 'At least one of title, excerpt, content, or fields must be provided' }); return; } try { const mergedFields = mergeStructuredFields( { title: data.title, excerpt: data.excerpt, content: data.content }, data.fields, config.maxAsyncChars ); if (countStructuredCharacters(mergedFields) === 0) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: 'Content cannot be empty.' }); } } catch (error) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ['fields'], message: error instanceof Error ? error.message : 'Invalid structured fields', }); } }); export const siteContentJobSubmitSchema = z.object({ resourceType: z.literal('site_content'), contractVersion: z.literal(SITE_CONTENT_CONTRACT_VERSION), sourceLang: LanguageCodeSchema, targetLang: LanguageCodeSchema, sourceRevision: z.string().regex(/^[a-f0-9]{64}$/), attemptToken: z.string().regex(/^[a-f0-9]{64}$/), segments: z.array(z.object({ id: z.string().regex(/^seg_[A-Za-z0-9_-]{32}$/), text: z.string().min(1), context: z.record(z.string(), z.unknown()), }).strict()).min(1).max(SITE_CONTENT_MAX_SEGMENTS), tone: ToneSchema.optional(), exceptions: ExceptionRulesSchema, callbackUrl: z.string().url(), callbackSecret: z.string().min(16).max(256), clientJobId: clientJobIdSchema, }).strict().superRefine((data, ctx) => { try { validateSiteContentRequest(data, config.maxAsyncChars); } catch (error) { ctx.addIssue({ code: 'custom', message: error instanceof Error ? error.message : 'Invalid segmented Site Content request', }); } }); export const jobSubmitSchema = z.union([siteContentJobSubmitSchema, legacyJobSubmitSchema]); const jobIdParamSchema = z.object({ jobId: z.string().uuid('Invalid job ID format'), }); export const jobsListQuerySchema = z.object({ status: z.string().optional(), clientJobId: clientJobIdSchema.optional(), limit: z.coerce.number().min(1).max(100).optional(), offset: z.coerce.number().min(0).optional(), sortBy: z.enum(['created_at', 'updated_at']).optional(), sortOrder: z.enum(['asc', 'desc']).optional(), }); /** * POST /v1/jobs * Submit asynchronous translation job (up to 50000 characters) * * @requires API Key authentication * @requires Rate limiting based on subscription tier */ router.post( '/', transformWordPressJobPayload, authenticateApiKey, apiRateLimiter, validate(jobSubmitSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const jobRequest = req.body; logger.info('Async job submission request received', { userId, sourceLang: jobRequest.sourceLang, targetLang: jobRequest.targetLang, contentLength: jobRequest.content?.length || 0, fieldCount: jobRequest.fields ? Object.keys(jobRequest.fields).length : 0, resourceType: jobRequest.resourceType, segmentCount: Array.isArray(jobRequest.segments) ? jobRequest.segments.length : 0, hasCallback: !!jobRequest.callbackUrl, }); // Call translation service const result = await translationService.translateAsync( userId, jobRequest, req.user!.plugin, req.siteUrl ); return res.status(202).json({ success: true, data: result, timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Async job submission error', { error, userId: req.user?.userId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; // Handle insufficient credits error if (errorMessage.includes('Insufficient credits')) { return res.status(402).json({ error: true, code: 'INSUFFICIENT_CREDITS', message: 'Translation job not found.', timestamp: new Date().toISOString(), }); } // Handle content too long error if (errorMessage.includes('Content too long')) { return res.status(413).json({ error: true, code: 'CONTENT_TOO_LONG', message: errorMessage, timestamp: new Date().toISOString(), }); } // Generic error return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while submitting the job', timestamp: new Date().toISOString(), }); } } ); /** * GET /v1/jobs/:jobId * Get translation job status and result * * @requires API Key authentication */ router.get( '/:jobId', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { jobId } = req.params; logger.info('Job status request received', { userId, jobId, }); // Get job status const result = await translationService.getJobStatus(userId, jobId); const submissionId = await resolveSubmissionIdForJob(userId, jobId, result.submissionId); // For external API, expose customerCost as cost const { customerCost, cost: _internalCost, ...rest } = result; const externalResult = { ...rest, cost: customerCost ?? _internalCost, ...(submissionId ? { submissionId } : {}), }; return res.status(200).json({ success: true, data: externalResult, timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Job status retrieval error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; // Handle not found error if (errorMessage.includes('not found')) { return res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); } // Handle access denied error if (errorMessage.includes('Access denied')) { return res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); } // Generic error return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while retrieving the job', timestamp: new Date().toISOString(), }); } } ); /** * POST /v1/jobs/:jobId/cancel * Cancel a pending translation job * * @requires API Key authentication */ router.post( '/:jobId/cancel', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { jobId } = req.params; if (!req.siteUrl) { return res.status(403).json({ error: true, code: 'SITE_SCOPE_REQUIRED', message: 'Authenticated site scope is required to cancel a job.', timestamp: new Date().toISOString(), }); } logger.info('Job cancellation request received', { userId, jobId, }); // Cancel job const result = await translationService.cancelJob(userId, jobId, { plugin: req.user!.plugin, siteUrl: req.siteUrl, }); return res.status(200).json({ success: true, data: result, message: 'Job cancelled successfully', timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Job cancellation error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; // Handle not found error if (errorMessage.includes('not found')) { return res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); } // Handle access denied error if (errorMessage.includes('Access denied')) { return res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); } // Handle cannot cancel error if (errorMessage.includes('Cannot cancel')) { return res.status(400).json({ error: true, code: 'CANNOT_CANCEL', message: errorMessage, timestamp: new Date().toISOString(), }); } // Generic error return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while cancelling the job', timestamp: new Date().toISOString(), }); } } ); /** * GET /v1/jobs * List translation jobs for the authenticated user * * @requires API Key authentication */ router.get( '/', authenticateApiKey, validateQuery(jobsListQuerySchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { status, clientJobId, limit, offset, sortBy, sortOrder } = req.query as any; logger.info('Jobs list request received', { userId, status, clientJobId, limit, offset, }); // Parse status (can be comma-separated) let statusFilter: any = undefined; if (status) { const statuses = status.split(',').map((s: string) => s.trim()); statusFilter = statuses.length > 1 ? statuses : statuses[0]; } // Get jobs const result = await translationService.getJobs(userId, { status: statusFilter, clientJobId, siteUrl: req.siteUrl, plugin: req.user!.plugin, limit: limit ? parseInt(limit) : undefined, offset: offset ? parseInt(offset) : undefined, sortBy: sortBy as any, sortOrder: sortOrder as any, }); // For external API, expose customerCost as cost const externalJobs = result.jobs.map(({ customerCost, cost: internalCost, ...rest }) => ({ ...rest, cost: customerCost ?? internalCost, })); return res.status(200).json({ success: true, data: externalJobs, pagination: { total: result.total, limit: result.limit, offset: result.offset, hasMore: result.offset + result.jobs.length < result.total, }, timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Jobs list retrieval error', { error, userId: req.user?.userId, requestId: req.requestId, }); // Generic error return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while retrieving jobs', timestamp: new Date().toISOString(), }); } } ); /** * DELETE /v1/jobs/:jobId * Delete a translation job (only allowed for completed/failed/cancelled jobs) * * @requires API Key authentication */ router.delete( '/:jobId', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { jobId } = req.params; logger.info('Job deletion request received', { userId, jobId, }); // Get job to check status const job = await translationService.getJobStatus(userId, jobId); // Only allow deletion of terminal state jobs if (job.status !== 'completed' && job.status !== 'failed' && job.status !== 'cancelled') { return res.status(400).json({ error: true, code: 'CANNOT_DELETE', message: `Cannot delete job in ${job.status} status. Only completed, failed, or cancelled jobs can be deleted.`, timestamp: new Date().toISOString(), }); } // Delete job (soft delete by updating status to 'deleted') // Note: Actual implementation would call a deleteJob method logger.warn('Job deletion not yet fully implemented - returning success for terminal state job', { userId, jobId, status: job.status, }); return res.status(200).json({ success: true, message: 'Job deleted successfully', timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Job deletion error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; if (errorMessage.includes('not found')) { return res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); } if (errorMessage.includes('Access denied')) { return res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); } return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while deleting the job', timestamp: new Date().toISOString(), }); } } ); /** * GET /v1/jobs/:jobId/result * Get only the translation result for a completed job * * @requires API Key authentication */ router.get( '/:jobId/result', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { jobId } = req.params; logger.info('Job result request received', { userId, jobId, }); // Get job status const job = await translationService.getJobStatus(userId, jobId); // Check if job is completed if (job.status !== 'completed') { return res.status(400).json({ error: true, code: 'JOB_NOT_COMPLETED', message: `Job is in ${job.status} status. Only completed jobs have results.`, timestamp: new Date().toISOString(), }); } const submissionId = await resolveSubmissionIdForJob(userId, jobId, job.submissionId); // Return just the translation result return res.status(200).json({ success: true, data: { jobId: job.jobId, ...(submissionId ? { submissionId } : {}), translation: job.translation, translatedTitle: job.translatedTitle, translatedExcerpt: job.translatedExcerpt, translatedContent: job.translatedContent, translatedFields: job.translatedFields, sourceLang: job.sourceLang, targetLang: job.targetLang, charactersUsed: job.charactersUsed, cost: job.customerCost ?? job.cost, completedAt: job.completedAt, }, timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Job result retrieval error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; if (errorMessage.includes('not found')) { return res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); } if (errorMessage.includes('Access denied')) { return res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); } return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while retrieving the job result', timestamp: new Date().toISOString(), }); } } ); /** * POST /v1/jobs/:jobId/retry * Retry a failed translation job * * @requires API Key authentication */ router.post( '/:jobId/retry', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { jobId } = req.params; logger.info('Job retry request received', { userId, jobId, }); // Retry job const result = await translationService.retryJob(userId, jobId); return res.status(200).json({ success: true, data: result, message: 'Job queued for retry', timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Job retry error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; // Handle not found error if (errorMessage.includes('not found')) { return res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); } // Handle access denied error if (errorMessage.includes('Access denied')) { return res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); } // Handle cannot retry error if (errorMessage.includes('Cannot retry')) { return res.status(400).json({ error: true, code: 'CANNOT_RETRY', message: errorMessage, timestamp: new Date().toISOString(), }); } // Generic error return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while retrying the job', timestamp: new Date().toISOString(), }); } } ); /** * GET /v1/jobs/:jobId/status/stream * Stream job status updates using Server-Sent Events (SSE) * * @requires API Key authentication */ router.get( '/:jobId/status/stream', authenticateApiKey, validateParams(jobIdParamSchema), async (req: Request, res: Response): Promise => { try { const userId = req.user!.userId; const { jobId } = req.params; logger.info('Job status stream request received', { userId, jobId, }); // Verify job exists and belongs to user const initialStatus = await translationService.getJobStatus(userId, jobId); // Set up SSE headers res.setHeader('Content-Type', 'text/event-stream'); res.setHeader('Cache-Control', 'no-cache'); res.setHeader('Connection', 'keep-alive'); res.setHeader('X-Accel-Buffering', 'no'); // Disable nginx buffering const submissionId = await resolveSubmissionIdForJob(userId, jobId, initialStatus.submissionId); // For external API, expose customerCost as cost const { customerCost: initCC, cost: initIC, ...initRest } = initialStatus; const externalInitial = { ...initRest, cost: initCC ?? initIC, ...(submissionId ? { submissionId } : {}), }; // Send initial status res.write(`data: ${JSON.stringify(externalInitial)}\n\n`); // If job is already in a terminal state, close the connection if ( initialStatus.status === 'completed' || initialStatus.status === 'failed' || initialStatus.status === 'cancelled' ) { logger.info('Job already in terminal state, closing stream', { userId, jobId, status: initialStatus.status, }); res.end(); return; } // Poll for updates every 2 seconds const pollInterval = setInterval(async () => { try { const currentStatus = await translationService.getJobStatus(userId, jobId); // For external API, expose customerCost as cost const { customerCost: curCC, cost: curIC, ...curRest } = currentStatus; const externalCurrent = { ...curRest, cost: curCC ?? curIC, ...(submissionId ? { submissionId } : {}), }; // Send update res.write(`data: ${JSON.stringify(externalCurrent)}\n\n`); // If job reached terminal state, close connection if ( currentStatus.status === 'completed' || currentStatus.status === 'failed' || currentStatus.status === 'cancelled' ) { logger.info('Job reached terminal state, closing stream', { userId, jobId, status: currentStatus.status, }); clearInterval(pollInterval); res.end(); } } catch (error) { logger.error('Error polling job status in stream', { error, userId, jobId, }); clearInterval(pollInterval); res.write(`data: ${JSON.stringify({ error: 'Failed to poll job status' })}\n\n`); res.end(); } }, 2000); // Clean up on client disconnect req.on('close', () => { logger.info('Client disconnected from job status stream', { userId, jobId, }); clearInterval(pollInterval); res.end(); }); } catch (error) { logger.error('Job status stream error', { error, userId: req.user?.userId, jobId: req.params.jobId, requestId: req.requestId, }); const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; // Handle not found error if (errorMessage.includes('not found')) { res.status(404).json({ error: true, code: 'JOB_NOT_FOUND', message: errorMessage, timestamp: new Date().toISOString(), }); return; } // Handle access denied error if (errorMessage.includes('Access denied')) { res.status(403).json({ error: true, code: 'ACCESS_DENIED', message: errorMessage, timestamp: new Date().toISOString(), }); return; } // Generic error res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while streaming job status', timestamp: new Date().toISOString(), }); } } ); // Validation schema for async bulk-strings job export const bulkStringsJobSchema = z.object({ sourceLang: LanguageCodeSchema, targetLangs: z.array(LanguageCodeSchema).min(1).max( BULK_MAX_TARGET_LANGS, `Bulk strings job supports at most ${BULK_MAX_TARGET_LANGS} target languages` ), strings: z.array(z.object({ id: z.string().min(1), content: z.string().min(1).max( BULK_MAX_CHARS_PER_STRING, `Each string must not exceed ${BULK_MAX_CHARS_PER_STRING} characters` ), })).min(1).max( BULK_ASYNC_MAX_STRINGS, `Bulk strings job supports at most ${BULK_ASYNC_MAX_STRINGS} strings` ), callbackUrl: z.string().url(), callbackSecret: z.string().min(16), tone: ToneSchema.optional(), exceptions: ExceptionRulesSchema, clientJobId: clientJobIdSchema.optional(), }).refine( (data) => data.strings.reduce((total, string) => total + string.content.length, 0) <= BULK_REQUEST_CHAR_BUDGET, { path: ['strings'], message: `Total bulk string content must not exceed ${BULK_REQUEST_CHAR_BUDGET} characters per request`, } ); type BulkStringsJobRequest = z.infer; type BulkStringsSubmission = BulkStringsJobRequest & { userId: string; plugin: string; siteUrl?: string; }; function bulkStringsIdentity(submission: BulkStringsSubmission): { content: string; contentHash: string; identity: string } { const strings = submission.strings.map(({ id, content }) => ({ id: id.normalize('NFC'), content: content.normalize('NFC'), })); const targetLangs = submission.targetLangs.map((targetLang) => targetLang.trim().toLowerCase()); const tone = submission.tone || 'neutral'; const content = JSON.stringify(strings); const callbackSecretFingerprint = crypto.createHash('sha256') .update(submission.callbackSecret) .digest('hex'); const identity = JSON.stringify({ callback_secret_fingerprint: callbackSecretFingerprint, callback_url: submission.callbackUrl, client_job_id: submission.clientJobId === undefined ? null : normalizeClientJobId(submission.clientJobId), site_url: submission.siteUrl ?? null, input: strings, model: 'gemini-3.1-flash-lite', plugin: submission.plugin, source_lang: submission.sourceLang.trim().toLowerCase(), target_langs: targetLangs, tone, user_id: submission.userId, }); return { content, contentHash: crypto.createHash('sha256').update(identity).digest('hex'), identity: crypto.createHash('sha256').update(identity).digest('hex'), }; } function bulkQueuePayload(jobId: string, submission: BulkStringsSubmission): Prisma.InputJsonObject { return { type: 'bulk-strings', jobId, clientJobId: submission.clientJobId === undefined ? null : normalizeClientJobId(submission.clientJobId), userId: submission.userId, plugin: submission.plugin, strings: submission.strings.map(({ id, content }) => ({ id, content })), sourceLang: submission.sourceLang, targetLangs: [...submission.targetLangs], tone: submission.tone || 'neutral', exceptions: submission.exceptions ?? [], callbackUrl: submission.callbackUrl, } satisfies StoredBulkQueuePayload; } export async function submitBulkStringsJob(submission: BulkStringsSubmission): Promise { const { content, contentHash, identity } = bulkStringsIdentity(submission); const jobClaim = await prisma.$transaction(async (tx) => { await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${identity}, 0))`; const existing = await tx.translationJob.findFirst({ where: { user_id: submission.userId, plugin: submission.plugin, content_hash: contentHash, source_lang: submission.sourceLang, target_lang: submission.targetLangs[0] || 'xx', tone: submission.tone || 'neutral', status: { in: ['pending', 'processing'] }, client_job_id: submission.clientJobId === undefined ? null : normalizeClientJobId(submission.clientJobId), callback_url: submission.callbackUrl, callback_secret: submission.callbackSecret, site_url: submission.siteUrl ?? null, }, }); if (existing) { return { job: existing, created: false }; } const jobId = crypto.randomUUID(); const job = await tx.translationJob.create({ data: { id: jobId, user_id: submission.userId, client_job_id: submission.clientJobId === undefined ? undefined : normalizeClientJobId(submission.clientJobId), plugin: submission.plugin, status: 'pending', source_lang: submission.sourceLang, target_lang: submission.targetLangs[0] || 'xx', tone: submission.tone || 'neutral', content, content_hash: contentHash, callback_url: submission.callbackUrl, callback_secret: submission.callbackSecret, site_url: submission.siteUrl ?? null, queue_payload: bulkQueuePayload(jobId, submission), }, }); return { job, created: true }; }); try { await drainBulkQueue(jobClaim.job.id); } catch (error) { logger.error('Immediate bulk queue dispatch failed', { error, jobId: jobClaim.job.id }); } return jobClaim.job; } /** * POST /v1/jobs/bulk-strings * Queue an async bulk-strings translation job. * The worker processes all strings in batches of 50 per Gemini call, once per target language, * then delivers a single webhook with full results. * * @requires API Key authentication */ router.post( '/bulk-strings', authenticateApiKey, apiRateLimiter, validate(bulkStringsJobSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { strings, sourceLang, targetLangs, callbackUrl, callbackSecret, tone } = req.body; // Total characters = per-string chars × number of target languages const totalCharacters = strings.reduce( (sum: number, s: { content: string }) => sum + countSourceCharacters(s.content), 0 ) * targetLangs.length; logger.info('Async bulk-strings job submission received', { userId, stringCount: strings.length, targetLangs, totalCharacters, }); // Fail fast if insufficient credits const sufficient = await hasSufficientCredits(userId, totalCharacters); const currentBalance = await getCurrentBalance(userId); if (!sufficient) { return res.status(402).json({ error: true, code: 'INSUFFICIENT_CREDITS', message: `Insufficient credits. You need ${totalCharacters} credits but only have ${currentBalance}. ` + `Please upgrade your plan or wait for your monthly allocation.`, timestamp: new Date().toISOString(), }); } const job = await submitBulkStringsJob({ userId, plugin: req.user!.plugin, strings, sourceLang, targetLangs, callbackUrl, callbackSecret, tone, clientJobId: req.body.clientJobId, siteUrl: req.siteUrl, }); logger.info('Async bulk-strings job queued', { userId, jobId: job.id, stringCount: strings.length }); return res.status(202).json({ success: true, job_id: job.id, status: 'queued', total_strings: strings.length, target_langs: targetLangs, timestamp: new Date().toISOString(), }); } catch (error) { logger.error('Async bulk-strings job submission error', { error, userId: req.user?.userId, requestId: req.requestId, }); return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while submitting the bulk-strings job', timestamp: new Date().toISOString(), }); } } ); // Validation schema for a single (post, language) item within an async bulk-content job const BULK_CONTENT_MAX_REF_LENGTH = 100; const BULK_CONTENT_MAX_CALLBACK_URL_LENGTH = 1000; const bulkContentItemSchema = z.object({ ref: z.string().min(1).max( BULK_CONTENT_MAX_REF_LENGTH, `Bulk content item refs must not exceed ${BULK_CONTENT_MAX_REF_LENGTH} characters` ), targetLang: LanguageCodeSchema, title: z.string().optional(), excerpt: z.string().optional(), content: z.string().optional(), fields: z.record( z.string().regex(STRUCTURED_FIELD_KEY_PATTERN, 'Invalid structured field key'), z.string() ).optional(), }).strict().superRefine((data, ctx) => { const hasCoreField = data.title !== undefined || data.excerpt !== undefined || data.content !== undefined; const hasCustomField = data.fields !== undefined && Object.keys(data.fields).length > 0; if (!hasCoreField && !hasCustomField) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: `Item "${data.ref}": at least one translatable field must be provided`, }); return; } try { const mergedFields = mergeStructuredFields( { title: data.title, excerpt: data.excerpt, content: data.content }, data.fields, config.maxAsyncChars ); if (countStructuredCharacters(mergedFields) === 0) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: `Item "${data.ref}": content cannot be empty.` }); } } catch (error) { ctx.addIssue({ code: z.ZodIssueCode.custom, message: `Item "${data.ref}": ${error instanceof Error ? error.message : 'Invalid item fields'}`, }); } }); // Validation schema for async bulk-content job. Unlike bulk-strings, an ITEM is a // (post, language) pair, not a post: a post with 3 missing target languages submits // 3 items sharing the same source fields, one per targetLang. The callback secret // deliberately does not belong to the body; it is accepted only from the authenticated // X-IPZ-Callback-Secret header in the route below. export const bulkContentJobSchema = z.object({ submissionId: z.string().uuid('submissionId must be a UUID'), clientJobId: clientJobIdSchema, sourceLang: LanguageCodeSchema, items: z.array(bulkContentItemSchema).min(1).max( BULK_CONTENT_MAX_ITEMS, `Bulk content job supports at most ${BULK_CONTENT_MAX_ITEMS} items` ), callbackUrl: z.string().url().max( BULK_CONTENT_MAX_CALLBACK_URL_LENGTH, `callbackUrl must not exceed ${BULK_CONTENT_MAX_CALLBACK_URL_LENGTH} characters` ), tone: ToneSchema.optional(), }).strict().superRefine((data, ctx) => { const seenRefs = new Set(); data.items.forEach((item, index) => { const canonicalRef = item.ref.normalize('NFC'); if (seenRefs.has(canonicalRef)) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ['items', index, 'ref'], message: `Duplicate bulk content ref "${item.ref}"`, }); } seenRefs.add(canonicalRef); }); // The per-item schema applies the async item cap. Keep the request-level // capacity bound aligned with bulk strings as well, counting the canonical // structured fields rather than the legacy aliases twice. let totalCharacters = 0; for (const item of data.items) { try { totalCharacters += countStructuredCharacters(mergeStructuredFields( { title: item.title, excerpt: item.excerpt, content: item.content }, item.fields, config.maxAsyncChars )); } catch { // The item-level refinement reports the specific field validation error. // Avoid adding an unrelated aggregate error for the same invalid item. } } if (totalCharacters > BULK_REQUEST_CHAR_BUDGET) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ['items'], message: `Total bulk content must not exceed ${BULK_REQUEST_CHAR_BUDGET} characters per request`, }); } }); type BulkContentJobRequest = z.infer; type BulkContentSubmission = BulkContentJobRequest & { userId: string; plugin: string; siteUrl?: string; callbackSecret: string; }; type BulkContentSubmissionOptions = { ensureCredits?: (characters: number) => Promise; }; type CanonicalBulkContentItem = BulkContentQueuePayload['items'][number]; export class BulkContentSubmissionError extends Error { public readonly statusCode: 404 | 409; public readonly code: 'JOB_NOT_FOUND' | 'SUBMISSION_ID_CONFLICT'; constructor(statusCode: 404 | 409, code: 'JOB_NOT_FOUND' | 'SUBMISSION_ID_CONFLICT', message: string) { super(message); this.name = 'BulkContentSubmissionError'; this.statusCode = statusCode; this.code = code; } } function normalizeBulkContentItems(items: BulkContentJobRequest['items']): CanonicalBulkContentItem[] { return items.map((item) => { const mergedFields = mergeStructuredFields( { title: item.title, excerpt: item.excerpt, content: item.content }, item.fields, config.maxAsyncChars ); const fields = Object.fromEntries( Object.entries(mergedFields).map(([key, value]) => [key, value.normalize('NFC')]) ) as StructuredFields; return { ref: item.ref.normalize('NFC'), targetLang: item.targetLang.trim().toLowerCase(), fields, }; }); } function normalizeBulkContentSiteUrl(siteUrl: string | undefined): string | null { return siteUrl === undefined ? null : siteUrl.normalize('NFC'); } function sameBulkContentPrincipal( job: { user_id: string; plugin: string; site_url: string | null }, submission: BulkContentSubmission ): boolean { return job.user_id === submission.userId && job.plugin === submission.plugin && (job.site_url ?? null) === normalizeBulkContentSiteUrl(submission.siteUrl); } function bulkContentIdentity(submission: BulkContentSubmission): { content: string; contentHash: string; fingerprint: string; items: CanonicalBulkContentItem[]; } { const items = normalizeBulkContentItems(submission.items); const submissionId = submission.submissionId.toLowerCase(); const canonicalRequest = { callback_url: submission.callbackUrl.normalize('NFC'), client_job_id: normalizeClientJobId(submission.clientJobId), input: items, model: 'gemini-3.1-flash-lite', plugin: submission.plugin, site_url: normalizeBulkContentSiteUrl(submission.siteUrl), source_lang: submission.sourceLang.trim().toLowerCase(), submission_id: submissionId, tone: submission.tone || 'neutral', user_id: submission.userId, }; const content = JSON.stringify(canonicalRequest); const fingerprint = crypto.createHash('sha256').update(content).digest('hex'); return { content, contentHash: fingerprint, fingerprint, items, }; } function bulkContentQueuePayload( jobId: string, submission: BulkContentSubmission, items: CanonicalBulkContentItem[] ): Prisma.InputJsonObject { // StoredBulkQueuePayload intentionally omits the callback secret. The dispatcher // reads the first-creation secret from the job row only while reconstructing Bull // data, so a durable queue payload or its replay fingerprint can never contain it. return { type: 'bulk-content', jobId, submissionId: submission.submissionId.toLowerCase(), clientJobId: normalizeClientJobId(submission.clientJobId), userId: submission.userId, plugin: submission.plugin, items: items.map(({ ref, targetLang, fields }) => ({ ref, targetLang, fields: { ...fields } })), sourceLang: submission.sourceLang.trim().toLowerCase(), tone: submission.tone || 'neutral', callbackUrl: submission.callbackUrl.normalize('NFC'), } satisfies StoredBulkQueuePayload; } export async function submitBulkContentJob( submission: BulkContentSubmission, options: BulkContentSubmissionOptions = {} ): Promise { const { content, contentHash, fingerprint, items } = bulkContentIdentity(submission); const submissionId = submission.submissionId.toLowerCase(); const siteUrl = normalizeBulkContentSiteUrl(submission.siteUrl); const totalCharacters = items.reduce((sum, item) => sum + countStructuredCharacters(item.fields), 0); const jobClaim = await prisma.$transaction(async (tx) => { await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${submissionId}, 0))`; const existing = await tx.translationJob.findUnique({ where: { submission_id: submissionId }, }); if (existing) { if (!sameBulkContentPrincipal(existing, submission)) { throw new BulkContentSubmissionError(404, 'JOB_NOT_FOUND', 'Translation job not found.'); } if (existing.submission_fingerprint !== fingerprint) { throw new BulkContentSubmissionError( 409, 'SUBMISSION_ID_CONFLICT', 'The submissionId has already been used for a different request.' ); } return { job: existing, created: false }; } // Serialize reservations for one account. A submission lock alone cannot stop // distinct request UUIDs from spending the same balance concurrently. await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`credits:${submission.userId}`}, 0))`; const latestTransaction = await tx.creditTransaction.findFirst({ where: { user_id: submission.userId }, orderBy: { ledger_sequence: 'desc' }, select: { balance_after: true }, }); const reservations = await tx.translationJob.aggregate({ where: { user_id: submission.userId, status: { in: ['pending', 'processing'] }, reserved_characters: { gt: 0 }, }, _sum: { reserved_characters: true }, }); const currentBalance = latestTransaction?.balance_after ?? 0; const reserved = reservations?._sum?.reserved_characters ?? 0; if (currentBalance - reserved < totalCharacters) { throw new Error( `Insufficient credits. You need ${totalCharacters} credits but only ${Math.max(0, currentBalance - reserved)}. ` + 'Please upgrade your plan or wait for your monthly allocation.' ); } if (options.ensureCredits) { await options.ensureCredits(totalCharacters); } const jobId = crypto.randomUUID(); const job = await tx.translationJob.create({ data: { id: jobId, user_id: submission.userId, site_url: siteUrl, client_job_id: normalizeClientJobId(submission.clientJobId), submission_id: submissionId, submission_fingerprint: fingerprint, plugin: submission.plugin, status: 'pending', reserved_characters: totalCharacters, source_lang: submission.sourceLang.trim().toLowerCase(), target_lang: items[0]?.targetLang || 'xx', tone: submission.tone || 'neutral', content, content_hash: contentHash, callback_url: submission.callbackUrl.normalize('NFC'), // This is the only persistence location for the callback secret. It is // deliberately absent from content, submission_fingerprint, and queue_payload. callback_secret: submission.callbackSecret, queue_payload: bulkContentQueuePayload(jobId, submission, items), }, }); return { job, created: true }; }); try { void Promise.resolve(drainBulkQueue(jobClaim.job.id)).catch((error) => { logger.error('Immediate bulk queue dispatch failed', { error, jobId: jobClaim.job.id }); }); } catch (error) { logger.error('Immediate bulk queue dispatch failed', { error, jobId: jobClaim.job.id }); } // Preserve the helper's historical direct-call shape while exposing whether this // request created or replayed the immutable submission to the HTTP controller. const result = { ...jobClaim.job, created: jobClaim.created }; Object.defineProperty(result, 'created', { value: jobClaim.created, enumerable: false, writable: false, }); return result; } /** * POST /v1/jobs/bulk-content * Queue an async bulk-content translation job. Each item is a (post, language) pair sharing * the request's sourceLang; the worker translates each item with one Gemini structured call, * isolates per-item failures, and delivers a single webhook with full results. * * @requires API Key authentication */ router.post( '/bulk-content', authenticateApiKey, apiRateLimiter, validate(bulkContentJobSchema), async (req: Request, res: Response) => { try { const userId = req.user!.userId; const { items, sourceLang, callbackUrl, submissionId, clientJobId, tone } = req.body; const rawCallbackSecret = req.headers['x-ipz-callback-secret']; if (rawCallbackSecret === undefined) { return res.status(400).json({ error: true, code: 'CALLBACK_SECRET_REQUIRED', message: 'X-IPZ-Callback-Secret header is required.', timestamp: new Date().toISOString(), }); } if (typeof rawCallbackSecret !== 'string') { return res.status(400).json({ error: true, code: 'CALLBACK_SECRET_INVALID', message: 'X-IPZ-Callback-Secret must be a single 16–256 byte value.', timestamp: new Date().toISOString(), }); } const callbackSecretBytes = Buffer.byteLength(rawCallbackSecret, 'utf8'); if (callbackSecretBytes < 16 || callbackSecretBytes > 256) { return res.status(400).json({ error: true, code: 'CALLBACK_SECRET_INVALID', message: 'X-IPZ-Callback-Secret must be a single 16–256 byte value.', timestamp: new Date().toISOString(), }); } const normalizedItems = normalizeBulkContentItems(items); const totalCharacters = normalizedItems.reduce( (sum, item) => sum + countStructuredCharacters(item.fields), 0 ); logger.info('Async bulk-content job submission received', { userId, submissionId, itemCount: items.length, totalCharacters, }); const job = await submitBulkContentJob({ userId, plugin: req.user!.plugin, items, sourceLang, callbackUrl, submissionId, clientJobId, callbackSecret: rawCallbackSecret, tone, siteUrl: req.siteUrl, }); const created = job.created !== false; const responseSubmissionId = typeof job.submission_id === 'string' ? job.submission_id : submissionId; logger.info('Async bulk-content job queued', { userId, jobId: job.id, submissionId: responseSubmissionId, itemCount: items.length, replay: !created, }); return res.status(created ? 202 : 200).json({ success: true, job_id: job.id, submissionId: responseSubmissionId, status: created ? 'queued' : job.status, total_items: items.length, timestamp: new Date().toISOString(), }); } catch (error) { if (error instanceof BulkContentSubmissionError) { return res.status(error.statusCode).json({ error: true, code: error.code, message: error.statusCode === 404 ? 'Translation job not found.' : error.message, timestamp: new Date().toISOString(), }); } const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'; if (errorMessage.includes('Insufficient credits')) { return res.status(402).json({ error: true, code: 'INSUFFICIENT_CREDITS', message: errorMessage, timestamp: new Date().toISOString(), }); } logger.error('Async bulk-content job submission error', { userId: req.user?.userId, requestId: req.requestId, }); return res.status(500).json({ error: true, code: 'INTERNAL_ERROR', message: 'An internal error occurred while submitting the bulk-content job', timestamp: new Date().toISOString(), }); } } ); export default router;