diff --git a/src/db.js b/src/db.js index bbc8af8..ce8d3dc 100644 --- a/src/db.js +++ b/src/db.js @@ -10,6 +10,7 @@ export function getPool(config) { password: config.db.password, database: config.db.database, connectionLimit: 10, + timezone: 'auto', }); } return pool; diff --git a/src/jobs/jobRepository.js b/src/jobs/jobRepository.js new file mode 100644 index 0000000..51f59af --- /dev/null +++ b/src/jobs/jobRepository.js @@ -0,0 +1,96 @@ +function toCamelJob(row) { + if (!row) return null; + return { + id: row.id, + status: row.status, + family: row.family, + sourceFormat: row.source_format, + targetFormat: row.target_format, + originalFilename: row.original_filename, + inputPath: row.input_path, + outputPath: row.output_path, + inputMimeType: row.input_mime_type, + outputMimeType: row.output_mime_type, + errorMessage: row.error_message, + createdAt: row.created_at, + updatedAt: row.updated_at, + expiresAt: row.expires_at, + }; +} + +export async function createJob(pool, job) { + await pool.query( + `INSERT INTO conversion_jobs + (id, status, family, source_format, target_format, original_filename, input_path, input_mime_type, expires_at) + VALUES (?, 'pending', ?, ?, ?, ?, ?, ?, ?)`, + [ + job.id, + job.family, + job.sourceFormat, + job.targetFormat, + job.originalFilename, + job.inputPath, + job.inputMimeType, + job.expiresAt, + ] + ); +} + +export async function getJobById(pool, id) { + const rows = await pool.query( + `SELECT id, status, family, source_format, target_format, original_filename, + input_path, output_path, input_mime_type, output_mime_type, + error_message, created_at, updated_at, expires_at + FROM conversion_jobs WHERE id = ?`, + [id] + ); + return toCamelJob(rows[0]); +} + +export async function getJobErrorLog(pool, id) { + const rows = await pool.query('SELECT error_log FROM conversion_jobs WHERE id = ?', [id]); + return rows[0]?.error_log ?? null; +} + +export async function markProcessing(pool, id) { + await pool.query("UPDATE conversion_jobs SET status = 'processing' WHERE id = ?", [id]); +} + +export async function markDone(pool, id, { outputPath, outputMimeType }) { + await pool.query( + "UPDATE conversion_jobs SET status = 'done', output_path = ?, output_mime_type = ? WHERE id = ?", + [outputPath, outputMimeType, id] + ); +} + +export async function markFailed(pool, id, { errorMessage, errorLog }) { + await pool.query( + "UPDATE conversion_jobs SET status = 'failed', error_message = ?, error_log = ? WHERE id = ?", + [errorMessage, errorLog, id] + ); +} + +export async function findPendingJobs(pool, limit) { + const rows = await pool.query( + `SELECT id, status, family, source_format, target_format, original_filename, + input_path, output_path, input_mime_type, output_mime_type, + error_message, created_at, updated_at, expires_at + FROM conversion_jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT ?`, + [limit] + ); + return rows.map(toCamelJob); +} + +export async function findExpiredJobs(pool) { + const rows = await pool.query( + `SELECT id, status, family, source_format, target_format, original_filename, + input_path, output_path, input_mime_type, output_mime_type, + error_message, created_at, updated_at, expires_at + FROM conversion_jobs WHERE expires_at < NOW()` + ); + return rows.map(toCamelJob); +} + +export async function deleteJob(pool, id) { + await pool.query('DELETE FROM conversion_jobs WHERE id = ?', [id]); +} diff --git a/test/jobs/jobRepository.test.js b/test/jobs/jobRepository.test.js new file mode 100644 index 0000000..e2b0b72 --- /dev/null +++ b/test/jobs/jobRepository.test.js @@ -0,0 +1,122 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { getPool, closePool } from '../../src/db.js'; +import { loadConfig } from '../../src/config.js'; +import { + createJob, + getJobById, + getJobErrorLog, + markProcessing, + markDone, + markFailed, + findPendingJobs, + findExpiredJobs, + deleteJob, +} from '../../src/jobs/jobRepository.js'; + +let pool; + +beforeAll(() => { + pool = getPool(loadConfig()); +}); + +afterAll(async () => { + await closePool(); +}); + +beforeEach(async () => { + await pool.query('DELETE FROM conversion_jobs'); +}); + +function baseJob(overrides = {}) { + return { + id: overrides.id ?? '11111111-1111-4111-8111-111111111111', + family: 'image', + sourceFormat: 'png', + targetFormat: 'webp', + originalFilename: 'photo.png', + inputPath: `${overrides.id ?? '11111111-1111-4111-8111-111111111111'}.png`, + inputMimeType: 'image/png', + expiresAt: new Date(Date.now() + 3600 * 1000), + ...overrides, + }; +} + +describe('jobRepository', () => { + it('creates and retrieves a pending job', async () => { + await createJob(pool, baseJob()); + + const job = await getJobById(pool, '11111111-1111-4111-8111-111111111111'); + + expect(job.status).toBe('pending'); + expect(job.family).toBe('image'); + expect(job.sourceFormat).toBe('png'); + expect(job.targetFormat).toBe('webp'); + expect(job.originalFilename).toBe('photo.png'); + expect(job.inputMimeType).toBe('image/png'); + expect(job.outputPath).toBeNull(); + expect(job.errorMessage).toBeNull(); + }); + + it('returns null for an unknown id', async () => { + const job = await getJobById(pool, '22222222-2222-4222-8222-222222222222'); + expect(job).toBeNull(); + }); + + it('transitions a job through processing to done', async () => { + await createJob(pool, baseJob()); + + await markProcessing(pool, '11111111-1111-4111-8111-111111111111'); + const processing = await getJobById(pool, '11111111-1111-4111-8111-111111111111'); + expect(processing.status).toBe('processing'); + + await markDone(pool, '11111111-1111-4111-8111-111111111111', { + outputPath: '11111111-1111-4111-8111-111111111111.webp', + outputMimeType: 'image/webp', + }); + const done = await getJobById(pool, '11111111-1111-4111-8111-111111111111'); + expect(done.status).toBe('done'); + expect(done.outputPath).toBe('11111111-1111-4111-8111-111111111111.webp'); + expect(done.outputMimeType).toBe('image/webp'); + }); + + it('marks a job failed with a short message and a separate detailed log', async () => { + await createJob(pool, baseJob()); + + await markFailed(pool, '11111111-1111-4111-8111-111111111111', { + errorMessage: 'Conversion failed, please try again', + errorLog: 'Error: sharp threw at line 42\n at convert (image.js:10:5)', + }); + + const job = await getJobById(pool, '11111111-1111-4111-8111-111111111111'); + expect(job.status).toBe('failed'); + expect(job.errorMessage).toBe('Conversion failed, please try again'); + expect(job.errorLog).toBeUndefined(); + + const errorLog = await getJobErrorLog(pool, '11111111-1111-4111-8111-111111111111'); + 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 () => { + await createJob(pool, baseJob({ id: '33333333-3333-4333-8333-333333333333' })); + await createJob(pool, baseJob({ id: '44444444-4444-4444-8444-444444444444' })); + await createJob(pool, baseJob({ id: '55555555-5555-4555-8555-555555555555' })); + + const jobs = await findPendingJobs(pool, 2); + + expect(jobs).toHaveLength(2); + expect(jobs[0].id).toBe('33333333-3333-4333-8333-333333333333'); + expect(jobs[1].id).toBe('44444444-4444-4444-8444-444444444444'); + }); + + it('finds expired jobs and allows deleting them', async () => { + await createJob(pool, baseJob({ expiresAt: new Date(Date.now() - 1000) })); + + const expired = await findExpiredJobs(pool); + expect(expired).toHaveLength(1); + expect(expired[0].id).toBe('11111111-1111-4111-8111-111111111111'); + + await deleteJob(pool, '11111111-1111-4111-8111-111111111111'); + const afterDelete = await getJobById(pool, '11111111-1111-4111-8111-111111111111'); + expect(afterDelete).toBeNull(); + }); +});