diff --git a/web-player/src/lib/state/uploadQueue.svelte.ts b/web-player/src/lib/state/uploadQueue.svelte.ts new file mode 100644 index 0000000..bf9d478 --- /dev/null +++ b/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; + readDuration: (file: File) => Promise; +} + +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 { + const delay = Math.min(3000 * 1.5 ** attempt, 15000); + return new Promise((resolve) => setTimeout(resolve, delay)); +} + +export class UploadQueue { + items = $state([]); + selectedGenres = $state([]); + cover = $state(null); + albumId = $state(''); + notice = $state(null); + running = $state(false); + + summary = $derived(summarizeQueue(this.items)); + active = $derived(hasActiveUploads(this.items)); + + private deps: QueueDeps; + private controllers = new Map(); + private presignController: AbortController | null = null; + private polling = false; + + constructor(deps: Partial = {}) { + 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(); diff --git a/web-player/src/lib/state/uploadQueue.test.ts b/web-player/src/lib/state/uploadQueue.test.ts new file mode 100644 index 0000000..0b9c761 --- /dev/null +++ b/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 = {}) { + 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': '*' }); + }); +}); diff --git a/web-player/src/lib/utils/uploadTransport.ts b/web-player/src/lib/utils/uploadTransport.ts new file mode 100644 index 0000000..97a97ea --- /dev/null +++ b/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; + put( + target: PresignTarget, + file: Blob, + onProgress: (fraction: number) => void, + signal: AbortSignal + ): Promise; + complete(request: CompleteRequest, signal: AbortSignal): Promise; + status(ids: string[]): Promise; +} + +export class TransportError extends Error {} + +export function signedHeaders(uploadUrl: string): Set { + const raw = new URL(uploadUrl).searchParams.get('X-Amz-SignedHeaders') ?? 'host'; + return new Set(raw.toLowerCase().split(';')); +} + +export function putHeaders(target: PresignTarget): Record { + const signed = signedHeaders(target.uploadUrl); + const headers: Record = {}; + 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(url: string, body: unknown, signal: AbortSignal): Promise { + 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 { + 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('/api/uploads', request, signal), + put: putWithProgress, + complete: (request, signal) => + postJson('/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; + } +};