Ombrora/convert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rewrite the data access layer onto Prisma Client

Browse Source
This commit is contained in:
anthony
2026-07-31 01:27:18 +02:00
parent 439453be38
commit 3a04da3180
4 changed files with 119 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/db.js
+11 -16
View File
@@ -1,28 +1,23 @@
import mariadb from 'mariadb'; import { PrismaClient } from '@prisma/client';
let pool; let prisma;
export function buildDatabaseUrl(config) { export function buildDatabaseUrl(config) {
const { host, user, password, database } = config.db; const { host, user, password, database } = config.db;
return `mysql://${encodeURIComponent(user)}:${encodeURIComponent(password)}@${host}:3306/${database}?connection_limit=10`; return `mysql://${encodeURIComponent(user)}:${encodeURIComponent(password)}@${host}:3306/${database}?connection_limit=10`;
} }
export function getPool(config) { export function getPrismaClient(config) {
if (!pool) { if (!prisma) {
pool = mariadb.createPool({ process.env.DATABASE_URL = buildDatabaseUrl(config);
host: config.db.host, prisma = new PrismaClient();
user: config.db.user,
password: config.db.password,
database: config.db.database,
connectionLimit: 10,
});
} }
return pool; return prisma;
} }
export async function closePool() { export async function closePrismaClient() {
if (pool) { if (prisma) {
await pool.end(); await prisma.$disconnect();
pool = undefined; prisma = undefined;
} }
} }
src/jobs/jobRepository.js
+68 -91
View File
@@ -1,109 +1,86 @@
function toCamelJob(row) { const jobSelect = {
if (!row) return null; id: true,
return { uuid: true,
id: row.id, status: true,
uuid: row.uuid, family: true,
status: row.status, sourceFormat: true,
family: row.family, targetFormat: true,
sourceFormat: row.source_format, originalFilename: true,
targetFormat: row.target_format, inputPath: true,
originalFilename: row.original_filename, outputPath: true,
inputPath: row.input_path, inputMimeType: true,
outputPath: row.output_path, outputMimeType: true,
inputMimeType: row.input_mime_type, inputSizeBytes: true,
outputMimeType: row.output_mime_type, outputSizeBytes: true,
inputSizeBytes: row.input_size_bytes, quality: true,
outputSizeBytes: row.output_size_bytes, conversionDurationSeconds: true,
quality: row.quality, errorMessage: true,
conversionDurationSeconds: row.conversion_duration_seconds, createdAt: true,
errorMessage: row.error_message, updatedAt: true,
createdAt: row.created_at, expiresAt: true,
updatedAt: row.updated_at, cleanedAt: true,
expiresAt: row.expires_at,
cleanedAt: row.cleaned_at,
}; };
export async function createJob(prisma, job) {
await prisma.conversionJob.create({
data: {
uuid: job.uuid,
family: job.family,
sourceFormat: job.sourceFormat,
targetFormat: job.targetFormat,
originalFilename: job.originalFilename,
inputPath: job.inputPath,
inputMimeType: job.inputMimeType,
inputSizeBytes: job.inputSizeBytes,
expiresAt: job.expiresAt,
quality: job.quality ?? null,
},
});
} }
export async function createJob(pool, job) { export async function getJobByUuid(prisma, uuid) {
await pool.query( return prisma.conversionJob.findUnique({ where: { uuid }, select: jobSelect });
`INSERT INTO conversion_jobs
(uuid, status, family, source_format, target_format, original_filename, input_path, input_mime_type, input_size_bytes, expires_at, quality)
VALUES (?, 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[
job.uuid,
job.family,
job.sourceFormat,
job.targetFormat,
job.originalFilename,
job.inputPath,
job.inputMimeType,
job.inputSizeBytes,
job.expiresAt,
job.quality ?? null,
]
);
} }
export async function getJobByUuid(pool, uuid) { export async function getJobErrorLog(prisma, id) {
const rows = await pool.query( const job = await prisma.conversionJob.findUnique({ where: { id }, select: { errorLog: true } });
`SELECT id, uuid, status, family, source_format, target_format, original_filename, return job?.errorLog ?? null;
input_path, output_path, input_mime_type, output_mime_type,
input_size_bytes, output_size_bytes, quality, conversion_duration_seconds,
error_message, created_at, updated_at, expires_at, cleaned_at
FROM conversion_jobs WHERE uuid = ?`,
[uuid]
);
return toCamelJob(rows[0]);
} }
export async function getJobErrorLog(pool, id) { export async function markProcessing(prisma, id) {
const rows = await pool.query('SELECT error_log FROM conversion_jobs WHERE id = ?', [id]); await prisma.conversionJob.update({ where: { id }, data: { status: 'processing' } });
return rows[0]?.error_log ?? null;
} }
export async function markProcessing(pool, id) { export async function markDone(prisma, id, { outputPath, outputMimeType, outputSizeBytes, conversionDurationSeconds }) {
await pool.query("UPDATE conversion_jobs SET status = 'processing' WHERE id = ?", [id]); await prisma.conversionJob.update({
where: { id },
data: { status: 'done', outputPath, outputMimeType, outputSizeBytes, conversionDurationSeconds },
});
} }
export async function markDone(pool, id, { outputPath, outputMimeType, outputSizeBytes, conversionDurationSeconds }) { export async function markFailed(prisma, id, { errorMessage, errorLog }) {
await pool.query( await prisma.conversionJob.update({
`UPDATE conversion_jobs where: { id },
SET status = 'done', output_path = ?, output_mime_type = ?, output_size_bytes = ?, conversion_duration_seconds = ? data: { status: 'failed', errorMessage, errorLog },
WHERE id = ?`, });
[outputPath, outputMimeType, outputSizeBytes, conversionDurationSeconds, id]
);
} }
export async function markFailed(pool, id, { errorMessage, errorLog }) { export async function findPendingJobs(prisma, limit) {
await pool.query( return prisma.conversionJob.findMany({
"UPDATE conversion_jobs SET status = 'failed', error_message = ?, error_log = ? WHERE id = ?", where: { status: 'pending' },
[errorMessage, errorLog, id] orderBy: { createdAt: 'asc' },
); take: limit,
select: jobSelect,
});
} }
export async function findPendingJobs(pool, limit) { export async function findExpiredJobs(prisma) {
const rows = await pool.query( return prisma.conversionJob.findMany({
`SELECT id, uuid, status, family, source_format, target_format, original_filename, where: { expiresAt: { lt: new Date() }, cleanedAt: null },
input_path, output_path, input_mime_type, output_mime_type, select: jobSelect,
input_size_bytes, output_size_bytes, quality, conversion_duration_seconds, });
error_message, created_at, updated_at, expires_at, cleaned_at
FROM conversion_jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT ?`,
[limit]
);
return rows.map(toCamelJob);
} }
export async function findExpiredJobs(pool) { export async function markCleaned(prisma, id) {
const rows = await pool.query( await prisma.conversionJob.update({ where: { id }, data: { cleanedAt: new Date() } });
`SELECT id, uuid, status, family, source_format, target_format, original_filename,
input_path, output_path, input_mime_type, output_mime_type,
input_size_bytes, output_size_bytes, quality, conversion_duration_seconds,
error_message, created_at, updated_at, expires_at, cleaned_at
FROM conversion_jobs WHERE expires_at < NOW() AND cleaned_at IS NULL`
);
return rows.map(toCamelJob);
}
export async function markCleaned(pool, id) {
await pool.query('UPDATE conversion_jobs SET cleaned_at = NOW() WHERE id = ?', [id]);
} }
test/db.test.js
+10 -10
View File
@@ -1,5 +1,5 @@
import { describe, it, expect, afterAll } from 'vitest'; import { describe, it, expect, afterAll } from 'vitest';
import { getPool, closePool, buildDatabaseUrl } from '../src/db.js'; import { getPrismaClient, closePrismaClient, buildDatabaseUrl } from '../src/db.js';
import { loadConfig } from '../src/config.js'; import { loadConfig } from '../src/config.js';
describe('buildDatabaseUrl', () => { describe('buildDatabaseUrl', () => {
@@ -20,25 +20,25 @@ describe('buildDatabaseUrl', () => {
}); });
}); });
describe('getPool', () => { describe('getPrismaClient', () => {
afterAll(async () => { afterAll(async () => {
await closePool(); await closePrismaClient();
}); });
it('returns a working pool that can run a query', async () => { it('returns a working client that can run a query', async () => {
const config = loadConfig(); const config = loadConfig();
const pool = getPool(config); const prisma = getPrismaClient(config);
const rows = await pool.query('SELECT 1 AS value'); const rows = await prisma.$queryRaw`SELECT 1 AS value`;
expect(Number(rows[0].value)).toBe(1); expect(Number(rows[0].value)).toBe(1);
}); });
it('returns the same pool instance on repeated calls', () => { it('returns the same client instance on repeated calls', () => {
const config = loadConfig(); const config = loadConfig();
const poolA = getPool(config); const clientA = getPrismaClient(config);
const poolB = getPool(config); const clientB = getPrismaClient(config);
expect(poolA).toBe(poolB); expect(clientA).toBe(clientB);
}); });
}); });
test/jobs/jobRepository.test.js
+29 -29
View File
@@ -1,5 +1,5 @@
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
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 { import {
createJob, createJob,
@@ -13,18 +13,18 @@ import {
markCleaned, markCleaned,
} from '../../src/jobs/jobRepository.js'; } from '../../src/jobs/jobRepository.js';
let pool; let prisma;
beforeAll(() => { beforeAll(() => {
pool = getPool(loadConfig()); prisma = getPrismaClient(loadConfig());
}); });
afterAll(async () => { afterAll(async () => {
await closePool(); await closePrismaClient();
}); });
beforeEach(async () => { beforeEach(async () => {
await pool.query('DELETE FROM conversion_jobs'); await prisma.conversionJob.deleteMany();
}); });
function baseJob(overrides = {}) { function baseJob(overrides = {}) {
@@ -44,9 +44,9 @@ function baseJob(overrides = {}) {
describe('jobRepository', () => { describe('jobRepository', () => {
it('creates and retrieves a pending job', async () => { it('creates and retrieves a pending job', async () => {
await createJob(pool, baseJob()); await createJob(prisma, baseJob());
const job = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const job = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
expect(job.id).toEqual(expect.any(Number)); expect(job.id).toEqual(expect.any(Number));
expect(job.uuid).toBe('11111111-1111-4111-8111-111111111111'); expect(job.uuid).toBe('11111111-1111-4111-8111-111111111111');
@@ -66,31 +66,31 @@ describe('jobRepository', () => {
}); });
it('stores and retrieves a numeric quality value', async () => { it('stores and retrieves a numeric quality value', async () => {
await createJob(pool, baseJob({ uuid: '66666666-6666-4666-8666-666666666666', quality: 82 })); await createJob(prisma, baseJob({ uuid: '66666666-6666-4666-8666-666666666666', quality: 82 }));
const job = await getJobByUuid(pool, '66666666-6666-4666-8666-666666666666'); const job = await getJobByUuid(prisma, '66666666-6666-4666-8666-666666666666');
expect(job.quality).toBe(82); expect(job.quality).toBe(82);
}); });
it('returns null for an unknown uuid', async () => { it('returns null for an unknown uuid', async () => {
const job = await getJobByUuid(pool, '22222222-2222-4222-8222-222222222222'); const job = await getJobByUuid(prisma, '22222222-2222-4222-8222-222222222222');
expect(job).toBeNull(); expect(job).toBeNull();
}); });
it('transitions a job through processing to done', async () => { it('transitions a job through processing to done', async () => {
await createJob(pool, baseJob()); await createJob(prisma, baseJob());
await markProcessing(pool, (await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111')).id); await markProcessing(prisma, (await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111')).id);
const processing = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const processing = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
expect(processing.status).toBe('processing'); expect(processing.status).toBe('processing');
await markDone(pool, processing.id, { await markDone(prisma, processing.id, {
outputPath: '11111111-1111-4111-8111-111111111111.webp', outputPath: '11111111-1111-4111-8111-111111111111.webp',
outputMimeType: 'image/webp', outputMimeType: 'image/webp',
outputSizeBytes: 2048, outputSizeBytes: 2048,
conversionDurationSeconds: 1.5, conversionDurationSeconds: 1.5,
}); });
const done = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const done = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
expect(done.status).toBe('done'); expect(done.status).toBe('done');
expect(done.outputPath).toBe('11111111-1111-4111-8111-111111111111.webp'); expect(done.outputPath).toBe('11111111-1111-4111-8111-111111111111.webp');
expect(done.outputMimeType).toBe('image/webp'); expect(done.outputMimeType).toBe('image/webp');
@@ -99,29 +99,29 @@ describe('jobRepository', () => {
}); });
it('marks a job failed with a short message and a separate detailed log', async () => { it('marks a job failed with a short message and a separate detailed log', async () => {
await createJob(pool, baseJob()); await createJob(prisma, baseJob());
const created = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const created = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
await markFailed(pool, created.id, { await markFailed(prisma, created.id, {
errorMessage: 'Conversion failed, please try again', errorMessage: 'Conversion failed, please try again',
errorLog: 'Error: sharp threw at line 42\n at convert (image.js:10:5)', errorLog: 'Error: sharp threw at line 42\n at convert (image.js:10:5)',
}); });
const job = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const job = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
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');
expect(job.errorLog).toBeUndefined(); expect(job.errorLog).toBeUndefined();
const errorLog = await getJobErrorLog(pool, created.id); const errorLog = await getJobErrorLog(prisma, created.id);
expect(errorLog).toBe('Error: sharp threw at line 42\n at convert (image.js:10:5)'); expect(errorLog).toBe('Error: sharp threw at line 42\n at convert (image.js:10:5)');
}); });
it('finds pending jobs oldest first, up to a limit', async () => { it('finds pending jobs oldest first, up to a limit', async () => {
await createJob(pool, baseJob({ uuid: '33333333-3333-4333-8333-333333333333' })); await createJob(prisma, baseJob({ uuid: '33333333-3333-4333-8333-333333333333' }));
await createJob(pool, baseJob({ uuid: '44444444-4444-4444-8444-444444444444' })); await createJob(prisma, baseJob({ uuid: '44444444-4444-4444-8444-444444444444' }));
await createJob(pool, baseJob({ uuid: '55555555-5555-4555-8555-555555555555' })); await createJob(prisma, baseJob({ uuid: '55555555-5555-4555-8555-555555555555' }));
const jobs = await findPendingJobs(pool, 2); const jobs = await findPendingJobs(prisma, 2);
expect(jobs).toHaveLength(2); expect(jobs).toHaveLength(2);
expect(jobs[0].uuid).toBe('33333333-3333-4333-8333-333333333333'); expect(jobs[0].uuid).toBe('33333333-3333-4333-8333-333333333333');
@@ -129,19 +129,19 @@ describe('jobRepository', () => {
}); });
it('finds expired jobs and allows marking them cleaned without deleting the row', async () => { it('finds expired jobs and allows marking them cleaned without deleting the row', async () => {
await createJob(pool, baseJob({ expiresAt: new Date(Date.now() - 1000) })); await createJob(prisma, baseJob({ expiresAt: new Date(Date.now() - 1000) }));
const expired = await findExpiredJobs(pool); const expired = await findExpiredJobs(prisma);
expect(expired).toHaveLength(1); expect(expired).toHaveLength(1);
expect(expired[0].uuid).toBe('11111111-1111-4111-8111-111111111111'); expect(expired[0].uuid).toBe('11111111-1111-4111-8111-111111111111');
await markCleaned(pool, expired[0].id); await markCleaned(prisma, expired[0].id);
const stillPresent = await getJobByUuid(pool, '11111111-1111-4111-8111-111111111111'); const stillPresent = await getJobByUuid(prisma, '11111111-1111-4111-8111-111111111111');
expect(stillPresent).not.toBeNull(); expect(stillPresent).not.toBeNull();
expect(stillPresent.cleanedAt).not.toBeNull(); expect(stillPresent.cleanedAt).not.toBeNull();
const afterCleaning = await findExpiredJobs(pool); const afterCleaning = await findExpiredJobs(prisma);
expect(afterCleaning).toHaveLength(0); expect(afterCleaning).toHaveLength(0);
}); });
}); });