feat: import job runner in-memory
- Añade ImportRunner en memoria con concurrencia configurable - Tests TDD para enqueue, concurrencia y comportamiento tras stop - Actualiza /api/import/scan para encolar jobs y registrar errores - Ajusta tsconfig.json para incluir tests en comprobaciones de tipo
This commit is contained in:
@@ -3,6 +3,7 @@ import cors from '@fastify/cors';
|
||||
import helmet from '@fastify/helmet';
|
||||
import rateLimit from '@fastify/rate-limit';
|
||||
import healthRoutes from './routes/health';
|
||||
import importRoutes from './routes/import';
|
||||
|
||||
export function buildApp(): FastifyInstance {
|
||||
const app: FastifyInstance = Fastify({
|
||||
@@ -13,6 +14,7 @@ export function buildApp(): FastifyInstance {
|
||||
void app.register(helmet);
|
||||
void app.register(rateLimit, { max: 1000, timeWindow: '1 minute' });
|
||||
void app.register(healthRoutes, { prefix: '/api' });
|
||||
void app.register(importRoutes, { prefix: '/api' });
|
||||
|
||||
return app;
|
||||
}
|
||||
|
||||
7
backend/src/config.ts
Normal file
7
backend/src/config.ts
Normal file
@@ -0,0 +1,7 @@
|
||||
import os from 'os';
|
||||
|
||||
const envVal = Number.parseInt(process.env.IMPORT_CONCURRENCY ?? '', 10);
|
||||
export const IMPORT_CONCURRENCY =
|
||||
Number.isFinite(envVal) && envVal > 0 ? envVal : Math.min(8, Math.max(1, os.cpus().length - 1));
|
||||
|
||||
export default IMPORT_CONCURRENCY;
|
||||
133
backend/src/jobs/importRunner.ts
Normal file
133
backend/src/jobs/importRunner.ts
Normal file
@@ -0,0 +1,133 @@
|
||||
import { IMPORT_CONCURRENCY } from '../config';
|
||||
|
||||
type Task<T = unknown> = {
|
||||
fn: () => Promise<T> | T;
|
||||
resolve: (value: T) => void;
|
||||
reject: (err: any) => void;
|
||||
promise?: Promise<T>;
|
||||
};
|
||||
|
||||
export class ImportRunner {
|
||||
private concurrency: number;
|
||||
private queue: Task[] = [];
|
||||
private runningCount = 0;
|
||||
private completedCount = 0;
|
||||
private isRunning = false;
|
||||
private stopped = false;
|
||||
|
||||
constructor(concurrency?: number) {
|
||||
this.concurrency = Math.max(1, concurrency ?? IMPORT_CONCURRENCY);
|
||||
}
|
||||
|
||||
start() {
|
||||
if (this.isRunning) return;
|
||||
this.isRunning = true;
|
||||
this.stopped = false;
|
||||
this._processQueue();
|
||||
}
|
||||
|
||||
async stopAndWait() {
|
||||
this.stop();
|
||||
|
||||
// wait until any running tasks finish
|
||||
while (this.runningCount > 0) {
|
||||
await new Promise((res) => setImmediate(res));
|
||||
}
|
||||
}
|
||||
stop() {
|
||||
if (this.stopped) return;
|
||||
|
||||
this.isRunning = false;
|
||||
this.stopped = true;
|
||||
|
||||
// reject and count all pending tasks (schedule rejection to avoid unhandled rejections)
|
||||
while (this.queue.length > 0) {
|
||||
const task = this.queue.shift()!;
|
||||
this.completedCount++;
|
||||
// attach a noop catch so Node doesn't treat the rejection as unhandled
|
||||
if (task.promise) {
|
||||
task.promise.catch(() => {});
|
||||
}
|
||||
setImmediate(() => {
|
||||
try {
|
||||
task.reject(new Error('ImportRunner stopped'));
|
||||
} catch (e) {
|
||||
// noop
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
enqueue<T = unknown>(fn: () => Promise<T> | T): Promise<T> {
|
||||
if (this.stopped) {
|
||||
return Promise.reject(new Error('ImportRunner stopped'));
|
||||
}
|
||||
|
||||
let resolveFn!: (v: T) => void;
|
||||
let rejectFn!: (e: any) => void;
|
||||
const p = new Promise<T>((res, rej) => {
|
||||
resolveFn = res;
|
||||
rejectFn = rej;
|
||||
});
|
||||
|
||||
this.queue.push({ fn, resolve: resolveFn, reject: rejectFn, promise: p });
|
||||
|
||||
// start or continue processing immediately so the first task begins right away
|
||||
if (!this.isRunning) {
|
||||
this.start();
|
||||
} else {
|
||||
this._processQueue();
|
||||
}
|
||||
|
||||
return p;
|
||||
}
|
||||
|
||||
getStatus() {
|
||||
return {
|
||||
queued: this.queue.length,
|
||||
running: this.runningCount,
|
||||
completed: this.completedCount,
|
||||
concurrency: this.concurrency,
|
||||
};
|
||||
}
|
||||
|
||||
private _processQueue() {
|
||||
if (!this.isRunning) return;
|
||||
|
||||
while (this.runningCount < this.concurrency && this.queue.length > 0) {
|
||||
const task = this.queue.shift()!;
|
||||
|
||||
const result = Promise.resolve().then(() => task.fn());
|
||||
|
||||
this.runningCount++;
|
||||
|
||||
result
|
||||
.then((res) => {
|
||||
this.runningCount--;
|
||||
this.completedCount++;
|
||||
try {
|
||||
task.resolve(res as any);
|
||||
} catch (e) {
|
||||
// noop
|
||||
}
|
||||
setImmediate(() => this._processQueue());
|
||||
})
|
||||
.catch((err) => {
|
||||
this.runningCount--;
|
||||
this.completedCount++;
|
||||
console.error(err);
|
||||
try {
|
||||
task.reject(err);
|
||||
} catch (e) {
|
||||
// noop
|
||||
}
|
||||
setImmediate(() => this._processQueue());
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const runner = new ImportRunner();
|
||||
runner.start();
|
||||
|
||||
export default runner;
|
||||
17
backend/src/lib/fileTypeDetector.ts
Normal file
17
backend/src/lib/fileTypeDetector.ts
Normal file
@@ -0,0 +1,17 @@
|
||||
import path from 'path';
|
||||
|
||||
export function detectFormat(filename: string): string {
|
||||
const ext = path.extname(filename || '').toLowerCase();
|
||||
|
||||
if (!ext) return 'bin';
|
||||
|
||||
const map: Record<string, string> = {
|
||||
'.zip': 'zip',
|
||||
'.7z': '7z',
|
||||
'.chd': 'chd',
|
||||
};
|
||||
|
||||
return map[ext] ?? ext.replace(/^\./, '');
|
||||
}
|
||||
|
||||
export default detectFormat;
|
||||
25
backend/src/routes/import.ts
Normal file
25
backend/src/routes/import.ts
Normal file
@@ -0,0 +1,25 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { runner } from '../jobs/importRunner';
|
||||
|
||||
export default async function importRoutes(app: FastifyInstance) {
|
||||
app.post('/import/scan', async (request, reply) => {
|
||||
const body = request.body as any;
|
||||
|
||||
// Encolar el job en background (placeholder)
|
||||
setImmediate(() => {
|
||||
// placeholder task: no persistencia, trabajo ligero en background
|
||||
runner
|
||||
.enqueue(async () => {
|
||||
// usar body en caso necesario; aquí sólo un placeholder
|
||||
void body;
|
||||
return true;
|
||||
})
|
||||
.catch((err) => {
|
||||
app.log.warn({ err }, 'Background import task failed');
|
||||
});
|
||||
});
|
||||
|
||||
// Responder inmediatamente
|
||||
reply.code(202).send({ status: 'queued' });
|
||||
});
|
||||
}
|
||||
62
backend/src/services/checksumService.ts
Normal file
62
backend/src/services/checksumService.ts
Normal file
@@ -0,0 +1,62 @@
|
||||
import fs from 'fs';
|
||||
import { createHash } from 'crypto';
|
||||
|
||||
function makeCRCTable(): Uint32Array {
|
||||
const table = new Uint32Array(256);
|
||||
for (let n = 0; n < 256; n++) {
|
||||
let c = n;
|
||||
for (let k = 0; k < 8; k++) {
|
||||
if (c & 1) c = 0xedb88320 ^ (c >>> 1);
|
||||
else c = c >>> 1;
|
||||
}
|
||||
table[n] = c >>> 0;
|
||||
}
|
||||
return table;
|
||||
}
|
||||
|
||||
const CRC_TABLE = makeCRCTable();
|
||||
|
||||
function updateCrc(crc: number, buf: Buffer): number {
|
||||
let c = crc >>> 0;
|
||||
for (let i = 0; i < buf.length; i++) {
|
||||
c = (CRC_TABLE[(c ^ buf[i]) & 0xff] ^ (c >>> 8)) >>> 0;
|
||||
}
|
||||
return c >>> 0;
|
||||
}
|
||||
|
||||
export async function computeHashes(filePath: string): Promise<{
|
||||
size: number;
|
||||
md5: string;
|
||||
sha1: string;
|
||||
crc32: string;
|
||||
}> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const md5 = createHash('md5');
|
||||
const sha1 = createHash('sha1');
|
||||
|
||||
let size = 0;
|
||||
let crc = 0xffffffff >>> 0;
|
||||
|
||||
const rs = fs.createReadStream(filePath);
|
||||
|
||||
rs.on('error', (err) => reject(err));
|
||||
|
||||
rs.on('data', (chunk: Buffer) => {
|
||||
md5.update(chunk);
|
||||
sha1.update(chunk);
|
||||
size += chunk.length;
|
||||
crc = updateCrc(crc, chunk);
|
||||
});
|
||||
|
||||
rs.on('end', () => {
|
||||
const md5sum = md5.digest('hex');
|
||||
const sha1sum = sha1.digest('hex');
|
||||
const final = (crc ^ 0xffffffff) >>> 0;
|
||||
const crcHex = final.toString(16).padStart(8, '0').toLowerCase();
|
||||
|
||||
resolve({ size, md5: md5sum, sha1: sha1sum, crc32: crcHex });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export default computeHashes;
|
||||
11
backend/src/services/datVerifier.ts
Normal file
11
backend/src/services/datVerifier.ts
Normal file
@@ -0,0 +1,11 @@
|
||||
export function parseDat(_xml: string): any {
|
||||
// Stub: el parseo completo no se implementa en esta fase.
|
||||
return {};
|
||||
}
|
||||
|
||||
export async function verifyRomAgainstDat(_romMeta: any, _parsedDat: any): Promise<any> {
|
||||
// Stub: verificación mínima para que los tests de integración puedan ser saltados.
|
||||
return {};
|
||||
}
|
||||
|
||||
export default { parseDat, verifyRomAgainstDat };
|
||||
42
backend/src/services/fsScanner.ts
Normal file
42
backend/src/services/fsScanner.ts
Normal file
@@ -0,0 +1,42 @@
|
||||
import path from 'path';
|
||||
import { promises as fsPromises } from 'fs';
|
||||
import { detectFormat } from '../lib/fileTypeDetector';
|
||||
|
||||
export async function scanDirectory(dirPath: string): Promise<any[]> {
|
||||
const results: any[] = [];
|
||||
|
||||
async function walk(dir: string) {
|
||||
const entries = await fsPromises.readdir(dir, { withFileTypes: true });
|
||||
|
||||
for (const entry of entries) {
|
||||
if (entry.name.startsWith('.')) continue; // ignore dotfiles
|
||||
|
||||
const full = path.join(dir, entry.name);
|
||||
|
||||
if (entry.isDirectory()) {
|
||||
await walk(full);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (entry.isFile()) {
|
||||
const stat = await fsPromises.stat(full);
|
||||
const format = detectFormat(entry.name);
|
||||
const isArchive = ['zip', '7z', 'chd'].includes(format);
|
||||
|
||||
results.push({
|
||||
path: full,
|
||||
filename: entry.name,
|
||||
name: entry.name,
|
||||
size: stat.size,
|
||||
format,
|
||||
isArchive,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await walk(dirPath);
|
||||
return results;
|
||||
}
|
||||
|
||||
export default scanDirectory;
|
||||
Reference in New Issue
Block a user