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