diff --git a/src/app.js b/src/app.js index 94c8baa..c8cf67d 100644 --- a/src/app.js +++ b/src/app.js @@ -31,7 +31,7 @@ function registerAllConverters() { convertersRegistered = true; } -export function createApp(config, pool) { +export function createApp(config, prisma) { registerAllConverters(); const app = express(); @@ -131,7 +131,7 @@ export function createApp(config, pool) { } const expiresAt = new Date(Date.now() + config.retentionHours * 3600 * 1000); - await createJob(pool, { + await createJob(prisma, { uuid, family: registryEntry.family, sourceFormat, @@ -151,7 +151,7 @@ export function createApp(config, pool) { }); app.get('/api/jobs/:id', async (req, res) => { - const job = await getJobByUuid(pool, req.params.id); + const job = await getJobByUuid(prisma, req.params.id); if (!job) { return res.status(404).json({ error: 'Job not found' }); } @@ -173,7 +173,7 @@ export function createApp(config, pool) { } app.get('/api/jobs/:id/download', async (req, res) => { - const job = await getJobByUuid(pool, req.params.id); + const job = await getJobByUuid(prisma, req.params.id); if (!job) { return res.status(404).json({ error: 'Job not found' }); } diff --git a/src/cleanup.js b/src/cleanup.js index c1598f2..7889e0b 100644 --- a/src/cleanup.js +++ b/src/cleanup.js @@ -1,16 +1,16 @@ import { pathToFileURL } from 'node:url'; import { loadConfig } from './config.js'; -import { getPool, closePool } from './db.js'; +import { getPrismaClient, closePrismaClient } from './db.js'; import { uploadPath, outputPath, deleteIfExists } from './storage.js'; import { findExpiredJobs, markCleaned } from './jobs/jobRepository.js'; -export async function runCleanup(pool, config) { - const expiredJobs = await findExpiredJobs(pool); +export async function runCleanup(prisma, config) { + const expiredJobs = await findExpiredJobs(prisma); for (const job of expiredJobs) { await deleteIfExists(uploadPath(config, job.uuid, job.sourceFormat)); await deleteIfExists(outputPath(config, job.uuid, job.targetFormat)); - await markCleaned(pool, job.id); + await markCleaned(prisma, job.id); } return expiredJobs.length; @@ -18,10 +18,10 @@ export async function runCleanup(pool, config) { async function main() { const config = loadConfig(); - const pool = getPool(config); - const deletedCount = await runCleanup(pool, config); + const prisma = getPrismaClient(config); + const deletedCount = await runCleanup(prisma, config); console.log(`Cleanup: removed ${deletedCount} expired job(s).`); - await closePool(); + await closePrismaClient(); } if (import.meta.url === pathToFileURL(process.argv[1]).href) { diff --git a/src/server.js b/src/server.js index 8112859..8f0eac1 100644 --- a/src/server.js +++ b/src/server.js @@ -1,13 +1,13 @@ import { loadConfig } from './config.js'; -import { getPool } from './db.js'; +import { getPrismaClient } from './db.js'; import { ensureStorageDirs } from './storage.js'; import { createApp } from './app.js'; async function main() { const config = loadConfig(); await ensureStorageDirs(config); - const pool = getPool(config); - const app = createApp(config, pool); + const prisma = getPrismaClient(config); + const app = createApp(config, prisma); app.listen(config.port, () => { console.log(`File converter API listening on port ${config.port}`); diff --git a/src/worker.js b/src/worker.js index 499d28b..bd202af 100644 --- a/src/worker.js +++ b/src/worker.js @@ -1,7 +1,7 @@ import { pathToFileURL } from 'node:url'; import fs from 'node:fs/promises'; import { loadConfig } from './config.js'; -import { getPool } from './db.js'; +import { getPrismaClient } from './db.js'; import { ensureStorageDirs, uploadPath, outputPath, deleteIfExists } from './storage.js'; import { outputMimeType } from './mime.js'; import { resolve as resolveConverter } from './converters/registry.js'; @@ -20,8 +20,8 @@ function withTimeout(promise, ms) { return Promise.race([promise, timeout]).finally(() => clearTimeout(timeoutId)); } -async function processJob(pool, config, job) { - await markProcessing(pool, job.id); +async function processJob(prisma, config, job) { + await markProcessing(prisma, job.id); const inputFilePath = uploadPath(config, job.uuid, job.sourceFormat); const outputFilePath = outputPath(config, job.uuid, job.targetFormat); @@ -37,7 +37,7 @@ async function processJob(pool, config, job) { const conversionDurationSeconds = (Date.now() - startedAt) / 1000; const { size: outputSizeBytes } = await fs.stat(outputFilePath); - await markDone(pool, job.id, { + await markDone(prisma, job.id, { outputPath: `${job.uuid}.${job.targetFormat}`, outputMimeType: outputMimeType(job.targetFormat), outputSizeBytes, @@ -45,22 +45,22 @@ async function processJob(pool, config, job) { }); } catch (error) { await deleteIfExists(outputFilePath); - await markFailed(pool, job.id, { + await markFailed(prisma, job.id, { errorMessage: 'Conversion failed, please try again.', errorLog: error.stack ?? String(error), }); } } -export async function processPendingJobs(pool, config) { - const jobs = await findPendingJobs(pool, config.workerConcurrency); - await Promise.all(jobs.map((job) => processJob(pool, config, job))); +export async function processPendingJobs(prisma, config) { + const jobs = await findPendingJobs(prisma, config.workerConcurrency); + await Promise.all(jobs.map((job) => processJob(prisma, config, job))); return jobs.length; } -export function startWorker(pool, config) { +export function startWorker(prisma, config) { const interval = setInterval(() => { - processPendingJobs(pool, config).catch((error) => { + processPendingJobs(prisma, config).catch((error) => { console.error('Error while processing pending jobs:', error); }); }, config.workerPollIntervalMs); @@ -70,14 +70,14 @@ export function startWorker(pool, config) { async function main() { const config = loadConfig(); await ensureStorageDirs(config); - const pool = getPool(config); + const prisma = getPrismaClient(config); registerImageConverters(); registerImageToPdfConverter(); registerDocumentConverters(); - startWorker(pool, config); + startWorker(prisma, config); console.log(`Worker started, polling every ${config.workerPollIntervalMs}ms`); } -main(); \ No newline at end of file +main(); diff --git a/test/api/endToEnd.test.js b/test/api/endToEnd.test.js index a9b8e4f..b9a4983 100644 --- a/test/api/endToEnd.test.js +++ b/test/api/endToEnd.test.js @@ -5,34 +5,34 @@ import path from 'node:path'; import os from 'node:os'; import { Document, Paragraph, TextRun, Packer } from 'docx'; import { createApp } from '../../src/app.js'; -import { getPool, closePool } from '../../src/db.js'; +import { getPrismaClient, closePrismaClient } from '../../src/db.js'; import { loadConfig } from '../../src/config.js'; import { ensureStorageDirs } from '../../src/storage.js'; import { processPendingJobs } from '../../src/worker.js'; let app; -let pool; +let prisma; let config; beforeAll(async () => { config = { ...loadConfig(), storageDir: await fs.mkdtemp(path.join(os.tmpdir(), 'converter-e2e-')) }; await ensureStorageDirs(config); - pool = getPool(config); - app = createApp(config, pool); + prisma = getPrismaClient(config); + app = createApp(config, prisma); }); afterAll(async () => { - await closePool(); + await closePrismaClient(); await fs.rm(config.storageDir, { recursive: true, force: true }); }); beforeEach(async () => { - await pool.query('DELETE FROM conversion_jobs'); + await prisma.conversionJob.deleteMany(); }); async function waitForDone(id, maxAttempts = 20) { for (let attempt = 0; attempt < maxAttempts; attempt += 1) { - await processPendingJobs(pool, config); + await processPendingJobs(prisma, config); const response = await request(app).get(`/api/jobs/${id}`); if (response.body.status === 'done' || response.body.status === 'failed') { return response.body; diff --git a/test/api/jobStatus.test.js b/test/api/jobStatus.test.js index 65eea80..f46f4fb 100644 --- a/test/api/jobStatus.test.js +++ b/test/api/jobStatus.test.js @@ -4,29 +4,29 @@ import fs from 'node:fs/promises'; import path from 'node:path'; import os from 'node:os'; import { createApp } from '../../src/app.js'; -import { getPool, closePool } from '../../src/db.js'; +import { getPrismaClient, closePrismaClient } from '../../src/db.js'; import { loadConfig } from '../../src/config.js'; import { createJob, getJobByUuid, markDone } from '../../src/jobs/jobRepository.js'; import { ensureStorageDirs, outputPath } from '../../src/storage.js'; let app; -let pool; +let prisma; let config; beforeAll(async () => { config = { ...loadConfig(), storageDir: await fs.mkdtemp(path.join(os.tmpdir(), 'converter-status-')) }; await ensureStorageDirs(config); - pool = getPool(config); - app = createApp(config, pool); + prisma = getPrismaClient(config); + app = createApp(config, prisma); }); afterAll(async () => { - await closePool(); + await closePrismaClient(); await fs.rm(config.storageDir, { recursive: true, force: true }); }); beforeEach(async () => { - await pool.query('DELETE FROM conversion_jobs'); + await prisma.conversionJob.deleteMany(); }); function baseJob(uuid) { @@ -46,7 +46,7 @@ function baseJob(uuid) { describe('GET /api/jobs/:id', () => { it('returns job status without the error log field', async () => { const uuid = '66666666-6666-4666-8666-666666666666'; - await createJob(pool, baseJob(uuid)); + await createJob(prisma, baseJob(uuid)); const response = await request(app).get(`/api/jobs/${uuid}`); @@ -66,11 +66,11 @@ describe('GET /api/jobs/:id', () => { describe('GET /api/jobs/:id/download', () => { it('streams the converted file with correct headers once done', async () => { const uuid = '77777777-7777-4777-8777-777777777777'; - await createJob(pool, baseJob(uuid)); + await createJob(prisma, baseJob(uuid)); const filePath = outputPath(config, uuid, 'webp'); await fs.writeFile(filePath, Buffer.from('fake webp bytes')); - const created = await getJobByUuid(pool, uuid); - await markDone(pool, created.id, { + const created = await getJobByUuid(prisma, uuid); + await markDone(prisma, created.id, { outputPath: `${uuid}.webp`, outputMimeType: 'image/webp', outputSizeBytes: 16, @@ -87,7 +87,7 @@ describe('GET /api/jobs/:id/download', () => { it('returns 409 when the job is not done yet', async () => { const uuid = '88888888-8888-4888-8888-888888888888'; - await createJob(pool, baseJob(uuid)); + await createJob(prisma, baseJob(uuid)); const response = await request(app).get(`/api/jobs/${uuid}/download`); @@ -102,12 +102,12 @@ describe('GET /api/jobs/:id/download', () => { describe('unexpected server errors', () => { it('returns a generic 500 without leaking internal error details', async () => { - const originalQuery = pool.query.bind(pool); - pool.query = () => Promise.reject(new Error('connection reset by peer')); + const originalFindUnique = prisma.conversionJob.findUnique.bind(prisma.conversionJob); + prisma.conversionJob.findUnique = () => Promise.reject(new Error('connection reset by peer')); const response = await request(app).get('/api/jobs/99999999-9999-4999-8999-999999999999'); - pool.query = originalQuery; + prisma.conversionJob.findUnique = originalFindUnique; expect(response.status).toBe(500); expect(response.body.error).toBe('Internal server error'); diff --git a/test/api/jobs.test.js b/test/api/jobs.test.js index 7622b2e..ffed02b 100644 --- a/test/api/jobs.test.js +++ b/test/api/jobs.test.js @@ -4,29 +4,29 @@ import fs from 'node:fs/promises'; import path from 'node:path'; import os from 'node:os'; import { createApp } from '../../src/app.js'; -import { getPool, closePool } from '../../src/db.js'; +import { getPrismaClient, closePrismaClient } from '../../src/db.js'; import { loadConfig } from '../../src/config.js'; import { getJobByUuid } from '../../src/jobs/jobRepository.js'; import { ensureStorageDirs } from '../../src/storage.js'; let app; -let pool; +let prisma; let config; beforeAll(async () => { config = { ...loadConfig(), storageDir: await fs.mkdtemp(path.join(os.tmpdir(), 'converter-api-')) }; await ensureStorageDirs(config); - pool = getPool(config); - app = createApp(config, pool); + prisma = getPrismaClient(config); + app = createApp(config, prisma); }); afterAll(async () => { - await closePool(); + await closePrismaClient(); await fs.rm(config.storageDir, { recursive: true, force: true }); }); beforeEach(async () => { - await pool.query('DELETE FROM conversion_jobs'); + await prisma.conversionJob.deleteMany(); }); describe('GET /api/formats', () => { @@ -60,7 +60,7 @@ describe('POST /api/jobs', () => { expect(response.body.jobs[0].status).toBe('pending'); expect(response.body.jobs[0].file).toBe('photo.png'); - const job = await getJobByUuid(pool, response.body.jobs[0].id); + const job = await getJobByUuid(prisma, response.body.jobs[0].id); expect(job.status).toBe('pending'); expect(job.sourceFormat).toBe('png'); expect(job.targetFormat).toBe('webp'); @@ -82,7 +82,7 @@ describe('POST /api/jobs', () => { expect(response.status).toBe(201); expect(response.body.jobs[0].status).toBe('pending'); - const job = await getJobByUuid(pool, response.body.jobs[0].id); + const job = await getJobByUuid(prisma, response.body.jobs[0].id); expect(job.quality).toBe(45); }); diff --git a/test/cleanup.test.js b/test/cleanup.test.js index 4cadf59..32dbd53 100644 --- a/test/cleanup.test.js +++ b/test/cleanup.test.js @@ -2,28 +2,28 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import fs from 'node:fs/promises'; import path from 'node:path'; import os from 'node:os'; -import { getPool, closePool } from '../src/db.js'; +import { getPrismaClient, closePrismaClient } from '../src/db.js'; import { loadConfig } from '../src/config.js'; import { ensureStorageDirs, uploadPath, outputPath } from '../src/storage.js'; import { createJob, markDone, getJobByUuid } from '../src/jobs/jobRepository.js'; import { runCleanup } from '../src/cleanup.js'; -let pool; +let prisma; let config; beforeAll(async () => { config = { ...loadConfig(), storageDir: await fs.mkdtemp(path.join(os.tmpdir(), 'converter-cleanup-')) }; await ensureStorageDirs(config); - pool = getPool(config); + prisma = getPrismaClient(config); }); afterAll(async () => { - await closePool(); + await closePrismaClient(); await fs.rm(config.storageDir, { recursive: true, force: true }); }); beforeEach(async () => { - await pool.query('DELETE FROM conversion_jobs'); + await prisma.conversionJob.deleteMany(); }); describe('runCleanup', () => { @@ -31,7 +31,7 @@ describe('runCleanup', () => { const uuid = 'eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee'; await fs.writeFile(uploadPath(config, uuid, 'png'), 'input bytes'); await fs.writeFile(outputPath(config, uuid, 'webp'), 'output bytes'); - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat: 'png', @@ -42,18 +42,18 @@ describe('runCleanup', () => { inputSizeBytes: 11, expiresAt: new Date(Date.now() - 1000), }); - const created = await getJobByUuid(pool, uuid); - await markDone(pool, created.id, { + const created = await getJobByUuid(prisma, uuid); + await markDone(prisma, created.id, { outputPath: `${uuid}.webp`, outputMimeType: 'image/webp', outputSizeBytes: 12, conversionDurationSeconds: 0.5, }); - const cleanedCount = await runCleanup(pool, config); + const cleanedCount = await runCleanup(prisma, config); expect(cleanedCount).toBe(1); - const job = await getJobByUuid(pool, uuid); + const job = await getJobByUuid(prisma, uuid); expect(job).not.toBeNull(); expect(job.cleanedAt).not.toBeNull(); await expect(fs.stat(uploadPath(config, uuid, 'png'))).rejects.toThrow(); @@ -63,7 +63,7 @@ describe('runCleanup', () => { it('marks an expired pending job (with no output file) cleaned without throwing', async () => { const uuid = 'ffffffff-ffff-4fff-8fff-ffffffffffff'; await fs.writeFile(uploadPath(config, uuid, 'png'), 'input bytes'); - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat: 'png', @@ -75,17 +75,17 @@ describe('runCleanup', () => { expiresAt: new Date(Date.now() - 1000), }); - const cleanedCount = await runCleanup(pool, config); + const cleanedCount = await runCleanup(prisma, config); expect(cleanedCount).toBe(1); - const job = await getJobByUuid(pool, uuid); + const job = await getJobByUuid(prisma, uuid); expect(job).not.toBeNull(); expect(job.cleanedAt).not.toBeNull(); }); it('leaves non-expired jobs untouched', async () => { const uuid = '12121212-1212-4212-8212-121212121212'; - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat: 'png', @@ -97,17 +97,17 @@ describe('runCleanup', () => { expiresAt: new Date(Date.now() + 3600 * 1000), }); - const cleanedCount = await runCleanup(pool, config); + const cleanedCount = await runCleanup(prisma, config); expect(cleanedCount).toBe(0); - const job = await getJobByUuid(pool, uuid); + const job = await getJobByUuid(prisma, uuid); expect(job).not.toBeNull(); expect(job.cleanedAt).toBeNull(); }); it('does not re-process an already cleaned expired job', async () => { const uuid = '13131313-1313-4313-8313-131313131313'; - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat: 'png', @@ -119,8 +119,8 @@ describe('runCleanup', () => { expiresAt: new Date(Date.now() - 1000), }); - const firstRun = await runCleanup(pool, config); - const secondRun = await runCleanup(pool, config); + const firstRun = await runCleanup(prisma, config); + const secondRun = await runCleanup(prisma, config); expect(firstRun).toBe(1); expect(secondRun).toBe(0); diff --git a/test/worker.test.js b/test/worker.test.js index d5fedf6..63c42e3 100644 --- a/test/worker.test.js +++ b/test/worker.test.js @@ -3,30 +3,30 @@ import fs from 'node:fs/promises'; import path from 'node:path'; import os from 'node:os'; import sharp from 'sharp'; -import { getPool, closePool } from '../src/db.js'; +import { getPrismaClient, closePrismaClient } from '../src/db.js'; import { loadConfig } from '../src/config.js'; import { ensureStorageDirs, uploadPath, outputPath } from '../src/storage.js'; import { createJob, getJobByUuid, getJobErrorLog } from '../src/jobs/jobRepository.js'; import { registerImageConverters } from '../src/converters/image.js'; import { processPendingJobs } from '../src/worker.js'; -let pool; +let prisma; let config; beforeAll(async () => { registerImageConverters(); config = { ...loadConfig(), storageDir: await fs.mkdtemp(path.join(os.tmpdir(), 'converter-worker-')) }; await ensureStorageDirs(config); - pool = getPool(config); + prisma = getPrismaClient(config); }); afterAll(async () => { - await closePool(); + await closePrismaClient(); await fs.rm(config.storageDir, { recursive: true, force: true }); }); beforeEach(async () => { - await pool.query('DELETE FROM conversion_jobs'); + await prisma.conversionJob.deleteMany(); }); async function createPendingImageJob(uuid, sourceFormat, targetFormat) { @@ -35,7 +35,7 @@ async function createPendingImageJob(uuid, sourceFormat, targetFormat) { await fs.copyFile(fixturePath, inputFilePath); const { size: inputSizeBytes } = await fs.stat(inputFilePath); - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat, @@ -53,10 +53,10 @@ describe('processPendingJobs', () => { const uuid = '99999999-9999-4999-8999-999999999999'; await createPendingImageJob(uuid, 'png', 'webp'); - const processedCount = await processPendingJobs(pool, config); + const processedCount = await processPendingJobs(prisma, config); expect(processedCount).toBe(1); - const job = await getJobByUuid(pool, uuid); + const job = await getJobByUuid(prisma, uuid); expect(job.status).toBe('done'); expect(job.outputPath).toBe(`${uuid}.webp`); expect(job.outputMimeType).toBe('image/webp'); @@ -86,7 +86,7 @@ describe('processPendingJobs', () => { await fs.writeFile(uploadPath(config, uuid, 'png'), noisyBuffer); } - await createJob(pool, { + await createJob(prisma, { uuid: lowUuid, family: 'image', sourceFormat: 'png', @@ -98,7 +98,7 @@ describe('processPendingJobs', () => { expiresAt: new Date(Date.now() + 3600 * 1000), quality: 5, }); - await createJob(pool, { + await createJob(prisma, { uuid: defaultUuid, family: 'image', sourceFormat: 'png', @@ -110,10 +110,10 @@ describe('processPendingJobs', () => { expiresAt: new Date(Date.now() + 3600 * 1000), }); - await processPendingJobs(pool, { ...config, workerConcurrency: 2 }); + await processPendingJobs(prisma, { ...config, workerConcurrency: 2 }); - const lowJob = await getJobByUuid(pool, lowUuid); - const defaultJob = await getJobByUuid(pool, defaultUuid); + const lowJob = await getJobByUuid(prisma, lowUuid); + const defaultJob = await getJobByUuid(prisma, defaultUuid); expect(lowJob.status).toBe('done'); expect(defaultJob.status).toBe('done'); expect(lowJob.outputSizeBytes).toBeLessThan(defaultJob.outputSizeBytes); @@ -121,7 +121,7 @@ describe('processPendingJobs', () => { it('marks a job failed with a safe message and a detailed log when the converter throws', async () => { const uuid = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; - await createJob(pool, { + await createJob(prisma, { uuid, family: 'image', sourceFormat: 'png', @@ -134,13 +134,13 @@ describe('processPendingJobs', () => { }); // Note: input file is intentionally never written, so sharp will throw ENOENT. - await processPendingJobs(pool, config); + await processPendingJobs(prisma, config); - const job = await getJobByUuid(pool, uuid); + const job = await getJobByUuid(prisma, uuid); expect(job.status).toBe('failed'); expect(job.errorMessage).toBe('Conversion failed, please try again.'); - const errorLog = await getJobErrorLog(pool, job.id); + const errorLog = await getJobErrorLog(prisma, job.id); expect(errorLog).toMatch(/input file is missing/i); }); @@ -150,7 +150,7 @@ describe('processPendingJobs', () => { await createPendingImageJob('dddddddd-dddd-4ddd-8ddd-dddddddddddd', 'png', 'webp'); const limitedConfig = { ...config, workerConcurrency: 2 }; - const processedCount = await processPendingJobs(pool, limitedConfig); + const processedCount = await processPendingJobs(prisma, limitedConfig); expect(processedCount).toBe(2); });