Add upload queue store with concurrent uploads, progress, cancel and retry

web-player/src/lib/state/uploadQueue.svelte.ts

@@ -0,0 +1,265 @@
+import {
+	UPLOAD_CONCURRENCY,
+	beginUpload,
+	chunkStartable,
+	failUpload,
+	finishCreate,
+	finishProcessing,
+	finishUpload,
+	genreConflictMessage,
+	hasActiveUploads,
+	isStartable,
+	readDuration,
+	reportProgress,
+	retryUpload,
+	summarizeQueue,
+	toQueuedUpload,
+	type QueuedUpload
+} from '$lib/data/uploadQueue';
+import type { PresignedAudio } from '$lib/data/uploadProtocol';
+import type { GenreOption } from '$lib/types';
+import { httpUploadTransport, type UploadTransport } from '$lib/utils/uploadTransport';
+
+const STATUS_BATCH = 20;
+
+interface QueueDeps {
+	transport: UploadTransport;
+	pollDelay: (attempt: number) => Promise<void>;
+	readDuration: (file: File) => Promise<number | null>;
+}
+
+interface BatchSettings {
+	tags: string[];
+	albumId: string | null;
+}
+
+function reasonOf(err: unknown): string {
+	if (err instanceof DOMException && err.name === 'AbortError') return 'Cancelada';
+	return err instanceof Error && err.message !== '' ? err.message : 'La subida falló.';
+}
+
+function backoff(attempt: number): Promise<void> {
+	const delay = Math.min(3000 * 1.5 ** attempt, 15000);
+	return new Promise((resolve) => setTimeout(resolve, delay));
+}
+
+export class UploadQueue {
+	items = $state<QueuedUpload[]>([]);
+	selectedGenres = $state<string[]>([]);
+	cover = $state<File | null>(null);
+	albumId = $state('');
+	notice = $state<string | null>(null);
+	running = $state(false);
+
+	summary = $derived(summarizeQueue(this.items));
+	active = $derived(hasActiveUploads(this.items));
+
+	private deps: QueueDeps;
+	private controllers = new Map<string, AbortController>();
+	private presignController: AbortController | null = null;
+	private polling = false;
+
+	constructor(deps: Partial<QueueDeps> = {}) {
+		this.deps = {
+			transport: deps.transport ?? httpUploadTransport,
+			pollDelay: deps.pollDelay ?? backoff,
+			readDuration: deps.readDuration ?? readDuration
+		};
+	}
+
+	enqueue(files: File[]) {
+		const added = files.map(toQueuedUpload);
+		this.items = [...this.items, ...added];
+		for (const item of added) {
+			if (item.error) continue;
+			this.deps.readDuration(item.file).then((duration) => {
+				this.update(item.id, (current) => ({ ...current, duration }));
+			});
+		}
+	}
+
+	setTitle(id: string, title: string) {
+		this.update(id, (item) => ({ ...item, title }));
+	}
+
+	remove(id: string) {
+		this.controllers.get(id)?.abort();
+		this.items = this.items.filter((item) => item.id !== id);
+	}
+
+	cancel(id: string) {
+		this.controllers.get(id)?.abort();
+	}
+
+	clear() {
+		this.presignController?.abort();
+		for (const controller of this.controllers.values()) controller.abort();
+		this.items = [];
+	}
+
+	retry(id: string) {
+		this.update(id, retryUpload);
+	}
+
+	async start(genreOptions: GenreOption[]) {
+		if (this.running) return;
+		this.notice = null;
+		if (this.selectedGenres.length === 0) {
+			this.notice = 'Elige al menos un género para toda la cola.';
+			return;
+		}
+		const conflict = genreConflictMessage(this.selectedGenres, genreOptions);
+		if (conflict) {
+			this.notice = conflict;
+			return;
+		}
+
+		const chunks = chunkStartable(this.items);
+		if (chunks.length === 0) return;
+
+		const settings: BatchSettings = {
+			tags: [...this.selectedGenres],
+			albumId: this.albumId === '' ? null : this.albumId
+		};
+		const cover = this.cover;
+		this.running = true;
+		try {
+			for (const chunk of chunks) await this.runChunk(chunk, cover, settings);
+		} finally {
+			this.running = false;
+		}
+	}
+
+	private update(id: string, change: (item: QueuedUpload) => QueuedUpload) {
+		this.items = this.items.map((item) => (item.id === id ? change(item) : item));
+	}
+
+	private find(id: string): QueuedUpload | undefined {
+		return this.items.find((item) => item.id === id);
+	}
+
+	private async runChunk(chunk: QueuedUpload[], cover: File | null, settings: BatchSettings) {
+		const current = chunk.filter((item) => {
+			const live = this.find(item.id);
+			return live !== undefined && isStartable(live);
+		});
+		if (current.length === 0) return;
+
+		const controller = new AbortController();
+		this.presignController = controller;
+
+		let audios: PresignedAudio[];
+		try {
+			const presigned = await this.deps.transport.presign(
+				{
+					cover: cover ? { fileName: cover.name, contentType: cover.type, size: cover.size } : null,
+					files: current.map(({ file }) => ({
+						fileName: file.name,
+						contentType: file.type,
+						size: file.size
+					}))
+				},
+				controller.signal
+			);
+			if (presigned.picture && cover) {
+				await this.deps.transport.put(presigned.picture, cover, () => {}, controller.signal);
+			}
+			audios = presigned.audios;
+		} catch (err) {
+			const reason = reasonOf(err);
+			for (const { id } of current)
+				this.update(id, (item) => failUpload(beginUpload(item), reason));
+			return;
+		} finally {
+			this.presignController = null;
+		}
+
+		const jobs = current.map((item, index) => ({ id: item.id, audio: audios[index] }));
+		const worker = async () => {
+			for (let job = jobs.shift(); job; job = jobs.shift()) {
+				await this.processOne(job.id, job.audio, settings);
+			}
+		};
+		await Promise.all(Array.from({ length: Math.min(UPLOAD_CONCURRENCY, jobs.length) }, worker));
+	}
+
+	private async processOne(id: string, audio: PresignedAudio, settings: BatchSettings) {
+		const item = this.find(id);
+		if (!item || !isStartable(item)) return;
+
+		const controller = new AbortController();
+		this.controllers.set(id, controller);
+		this.update(id, beginUpload);
+		try {
+			await this.deps.transport.put(
+				audio,
+				item.file,
+				(fraction) => this.update(id, (current) => reportProgress(current, fraction)),
+				controller.signal
+			);
+			this.update(id, finishUpload);
+			const title = this.find(id)?.title.trim() || item.title;
+			const created = await this.deps.transport.complete(
+				{
+					title,
+					audioIntentId: audio.intentId,
+					pictureIntentId: audio.pictureIntentId,
+					tags: settings.tags,
+					albumId: settings.albumId
+				},
+				controller.signal
+			);
+			this.update(id, (current) => finishCreate(current, created.trackId, created.note));
+			void this.watchProcessing();
+		} catch (err) {
+			const reason = reasonOf(err);
+			this.update(id, (current) => failUpload(current, reason));
+		} finally {
+			this.controllers.delete(id);
+		}
+	}
+
+	private async watchProcessing() {
+		if (this.polling) return;
+		this.polling = true;
+		try {
+			for (let attempt = 0; ; attempt++) {
+				const trackIds = this.items
+					.filter((item) => item.phase === 'processing' && item.trackId !== null)
+					.map((item) => item.trackId as string);
+				if (trackIds.length === 0) return;
+				await this.deps.pollDelay(attempt);
+				await this.refreshStatus(trackIds);
+			}
+		} finally {
+			this.polling = false;
+		}
+	}
+
+	private async refreshStatus(trackIds: string[]) {
+		const batches: string[][] = [];
+		for (let i = 0; i < trackIds.length; i += STATUS_BATCH) {
+			batches.push(trackIds.slice(i, i + STATUS_BATCH));
+		}
+		await Promise.all(
+			batches.map(async (ids) => {
+				try {
+					const statuses = await this.deps.transport.status(ids);
+					for (const item of this.items) {
+						const status = item.trackId ? statuses[item.trackId] : undefined;
+						if (status === 'Completed') this.update(item.id, finishProcessing);
+						if (status === 'Failed') {
+							this.update(item.id, (current) =>
+								failUpload(current, 'No se pudo procesar el audio.')
+							);
+						}
+					}
+				} catch {
+					return;
+				}
+			})
+		);
+	}
+}
+
+export const uploadQueue = new UploadQueue();

web-player/src/lib/state/uploadQueue.test.ts

@@ -0,0 +1,194 @@
+import { describe, expect, it, vi } from 'vitest';
+import { UploadQueue } from '$lib/state/uploadQueue.svelte';
+import type { UploadTransport } from '$lib/utils/uploadTransport';
+import { putHeaders } from '$lib/utils/uploadTransport';
+import type { GenreOption } from '$lib/types';
+
+const GENRES: GenreOption[] = [
+	{ genre: 'Rock', incompatibleWith: ['Classical'] },
+	{ genre: 'Classical', incompatibleWith: ['Rock'] },
+	{ genre: 'Jazz', incompatibleWith: [] }
+];
+
+function file(name: string) {
+	return new File([new Uint8Array(8)], name);
+}
+
+function fakeTransport(overrides: Partial<UploadTransport> = {}) {
+	let counter = 0;
+	const transport: UploadTransport = {
+		presign: vi.fn(async (request) => ({
+			picture: request.cover ? { uploadUrl: 'https://s3/pic', contentType: 'image/png' } : null,
+			audios: request.files.map(() => ({
+				intentId: `intent-${counter++}`,
+				pictureIntentId: request.cover ? 'pic' : null,
+				uploadUrl: 'https://s3/audio',
+				contentType: 'audio/wav'
+			}))
+		})),
+		put: vi.fn(async (_target, _file, onProgress) => onProgress(0.5)),
+		complete: vi.fn(async (request) => ({ trackId: `track-${request.audioIntentId}`, note: null })),
+		status: vi.fn(async (ids) =>
+			Object.fromEntries(ids.map((id: string) => [id, 'Completed' as const]))
+		),
+		...overrides
+	};
+	return transport;
+}
+
+function queue(transport: UploadTransport) {
+	const q = new UploadQueue({
+		transport,
+		pollDelay: async () => {},
+		readDuration: async () => 60
+	});
+	q.selectedGenres = ['Jazz'];
+	return q;
+}
+
+describe('UploadQueue', () => {
+	it('uploads every valid file, creates the tracks and finishes them', async () => {
+		const transport = fakeTransport();
+		const q = queue(transport);
+		q.enqueue([file('a.wav'), file('b.mp3'), file('c.psd')]);
+		await q.start(GENRES);
+		await vi.waitFor(() => expect(q.summary.done).toBe(2));
+		expect(q.summary.failed).toBe(1);
+		expect(transport.put).toHaveBeenCalledTimes(2);
+		expect(transport.complete).toHaveBeenCalledTimes(2);
+		expect(transport.presign).toHaveBeenCalledTimes(1);
+	});
+
+	it('uploads the shared cover once and passes the album and genres', async () => {
+		const transport = fakeTransport();
+		const q = queue(transport);
+		q.cover = new File([new Uint8Array(4)], 'c.png', { type: 'image/png' });
+		q.albumId = 'album-1';
+		q.enqueue([file('a.wav'), file('b.wav')]);
+		await q.start(GENRES);
+		const puts = vi.mocked(transport.put).mock.calls.map((call) => call[0].uploadUrl);
+		expect(puts.filter((url) => url === 'https://s3/pic')).toHaveLength(1);
+		expect(transport.complete).toHaveBeenCalledWith(
+			expect.objectContaining({ albumId: 'album-1', tags: ['Jazz'], pictureIntentId: 'pic' }),
+			expect.anything()
+		);
+	});
+
+	it('requires genres and rejects conflicting ones before sending anything', async () => {
+		const transport = fakeTransport();
+		const q = queue(transport);
+		q.enqueue([file('a.wav')]);
+		q.selectedGenres = [];
+		await q.start(GENRES);
+		expect(q.notice).toContain('género');
+		q.selectedGenres = ['Rock', 'Classical'];
+		await q.start(GENRES);
+		expect(q.notice).toContain('no se pueden combinar');
+		expect(transport.presign).not.toHaveBeenCalled();
+	});
+
+	it('never runs more than the concurrency limit at once', async () => {
+		let running = 0;
+		let peak = 0;
+		const transport = fakeTransport({
+			put: async () => {
+				running++;
+				peak = Math.max(peak, running);
+				await new Promise((resolve) => setTimeout(resolve, 5));
+				running--;
+			}
+		});
+		const q = queue(transport);
+		q.enqueue(Array.from({ length: 8 }, (_, i) => file(`${i}.wav`)));
+		await q.start(GENRES);
+		expect(peak).toBe(3);
+	});
+
+	it('marks a file as failed and lets it retry', async () => {
+		let fail = true;
+		const transport = fakeTransport({
+			put: async () => {
+				if (fail) throw new Error('boom');
+			}
+		});
+		const q = queue(transport);
+		q.enqueue([file('a.wav')]);
+		await q.start(GENRES);
+		expect(q.items[0]).toMatchObject({ phase: 'error', failure: 'boom' });
+		fail = false;
+		q.retry(q.items[0].id);
+		await q.start(GENRES);
+		await vi.waitFor(() => expect(q.items[0].phase).toBe('done'));
+	});
+
+	it('fails the whole batch when the presign request fails', async () => {
+		const transport = fakeTransport({
+			presign: async () => {
+				throw new Error('sin URLs');
+			}
+		});
+		const q = queue(transport);
+		q.enqueue([file('a.wav'), file('b.wav')]);
+		await q.start(GENRES);
+		expect(q.items.map((item) => item.failure)).toEqual(['sin URLs', 'sin URLs']);
+	});
+
+	it('cancels an upload in flight', async () => {
+		const transport = fakeTransport({
+			put: (_target, _file, _progress, signal) =>
+				new Promise((_resolve, reject) =>
+					signal.addEventListener('abort', () => reject(new DOMException('x', 'AbortError')))
+				)
+		});
+		const q = queue(transport);
+		q.enqueue([file('a.wav')]);
+		const started = q.start(GENRES);
+		await vi.waitFor(() => expect(q.items[0].phase).toBe('uploading'));
+		q.cancel(q.items[0].id);
+		await started;
+		expect(q.items[0]).toMatchObject({ phase: 'error', failure: 'Cancelada' });
+		expect(transport.complete).not.toHaveBeenCalled();
+	});
+
+	it('keeps the album note and turns a failed processing into an error', async () => {
+		const transport = fakeTransport({
+			complete: async () => ({ trackId: 't1', note: 'Sin álbum' }),
+			status: async () => ({ t1: 'Failed' })
+		});
+		const q = queue(transport);
+		q.enqueue([file('a.wav')]);
+		await q.start(GENRES);
+		await vi.waitFor(() => expect(q.items[0].phase).toBe('error'));
+		expect(q.items[0].note).toBe('Sin álbum');
+	});
+
+	it('reports active uploads and clears the queue', async () => {
+		const transport = fakeTransport({
+			put: (_target, _file, _progress, signal) =>
+				new Promise((_resolve, reject) =>
+					signal.addEventListener('abort', () => reject(new DOMException('x', 'AbortError')))
+				)
+		});
+		const q = queue(transport);
+		q.enqueue([file('a.wav')]);
+		const started = q.start(GENRES);
+		await vi.waitFor(() => expect(q.active).toBe(true));
+		q.clear();
+		await started;
+		expect(q.items).toEqual([]);
+		expect(q.active).toBe(false);
+	});
+});
+
+describe('putHeaders', () => {
+	it('sends only the signed headers', () => {
+		const target = { uploadUrl: 'https://s3/a?X-Amz-SignedHeaders=host', contentType: 'audio/wav' };
+		expect(putHeaders(target)).toEqual({});
+		expect(
+			putHeaders({
+				...target,
+				uploadUrl: 'https://s3/a?X-Amz-SignedHeaders=content-type%3Bhost%3Bif-none-match'
+			})
+		).toEqual({ 'Content-Type': 'audio/wav', 'If-None-Match': '*' });
+	});
+});

web-player/src/lib/utils/uploadTransport.ts

@@ -0,0 +1,91 @@
+import type {
+	CompleteRequest,
+	CompleteResponse,
+	PresignRequest,
+	PresignResponse,
+	PresignTarget,
+	StatusResponse
+} from '$lib/data/uploadProtocol';
+
+export interface UploadTransport {
+	presign(request: PresignRequest, signal: AbortSignal): Promise<PresignResponse>;
+	put(
+		target: PresignTarget,
+		file: Blob,
+		onProgress: (fraction: number) => void,
+		signal: AbortSignal
+	): Promise<void>;
+	complete(request: CompleteRequest, signal: AbortSignal): Promise<CompleteResponse>;
+	status(ids: string[]): Promise<StatusResponse>;
+}
+
+export class TransportError extends Error {}
+
+export function signedHeaders(uploadUrl: string): Set<string> {
+	const raw = new URL(uploadUrl).searchParams.get('X-Amz-SignedHeaders') ?? 'host';
+	return new Set(raw.toLowerCase().split(';'));
+}
+
+export function putHeaders(target: PresignTarget): Record<string, string> {
+	const signed = signedHeaders(target.uploadUrl);
+	const headers: Record<string, string> = {};
+	if (signed.has('content-type')) headers['Content-Type'] = target.contentType;
+	if (signed.has('if-none-match')) headers['If-None-Match'] = '*';
+	return headers;
+}
+
+async function postJson<T>(url: string, body: unknown, signal: AbortSignal): Promise<T> {
+	const response = await fetch(url, {
+		method: 'POST',
+		headers: { 'content-type': 'application/json' },
+		body: JSON.stringify(body),
+		signal
+	});
+	const data = await response.json().catch(() => null);
+	if (!response.ok) {
+		throw new TransportError(data?.message ?? 'La solicitud falló.');
+	}
+	return data as T;
+}
+
+function putWithProgress(
+	target: PresignTarget,
+	file: Blob,
+	onProgress: (fraction: number) => void,
+	signal: AbortSignal
+): Promise<void> {
+	return new Promise((resolve, reject) => {
+		if (signal.aborted) {
+			reject(new DOMException('Aborted', 'AbortError'));
+			return;
+		}
+		const xhr = new XMLHttpRequest();
+		xhr.open('PUT', target.uploadUrl);
+		for (const [name, value] of Object.entries(putHeaders(target))) {
+			xhr.setRequestHeader(name, value);
+		}
+		xhr.upload.onprogress = (event) => {
+			if (event.lengthComputable && event.total > 0) onProgress(event.loaded / event.total);
+		};
+		xhr.onload = () => {
+			if (xhr.status >= 200 && xhr.status < 300) resolve();
+			else reject(new TransportError(`El almacenamiento rechazó el archivo (${xhr.status}).`));
+		};
+		xhr.onerror = () => reject(new TransportError('No se pudo conectar con el almacenamiento.'));
+		xhr.onabort = () => reject(new DOMException('Aborted', 'AbortError'));
+		signal.addEventListener('abort', () => xhr.abort(), { once: true });
+		xhr.send(file);
+	});
+}
+
+export const httpUploadTransport: UploadTransport = {
+	presign: (request, signal) => postJson<PresignResponse>('/api/uploads', request, signal),
+	put: putWithProgress,
+	complete: (request, signal) =>
+		postJson<CompleteResponse>('/api/uploads/complete', request, signal),
+	async status(ids) {
+		const response = await fetch(`/api/uploads/status?ids=${encodeURIComponent(ids.join(','))}`);
+		if (!response.ok) throw new TransportError('No se pudo consultar el estado.');
+		return (await response.json()) as StatusResponse;
+	}
+};