feat: add conversion_jobs repository
Also configures the MariaDB pool with timezone: 'auto', since the default 'local' mode sends dates without timezone conversion and silently broke expires_at comparisons whenever the app host and DB server clocks differ (caught by the findExpiredJobs test). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,7 @@ export function getPool(config) {
|
|||||||
password: config.db.password,
|
password: config.db.password,
|
||||||
database: config.db.database,
|
database: config.db.database,
|
||||||
connectionLimit: 10,
|
connectionLimit: 10,
|
||||||
|
timezone: 'auto',
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
return pool;
|
return pool;
|
||||||
|
|||||||
@@ -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]);
|
||||||
|
}
|
||||||
@@ -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();
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user