diff --git a/apps/archiver/src/workers/createStemsArchive/createStemsArchive.test.ts b/apps/archiver/src/workers/createStemsArchive/createStemsArchive.test.ts new file mode 100644 index 0000000..c34cec5 --- /dev/null +++ b/apps/archiver/src/workers/createStemsArchive/createStemsArchive.test.ts @@ -0,0 +1,248 @@ +import { Job } from 'bullmq' +import archiver from 'archiver' +import fs from 'fs/promises' +import fsSync from 'fs' +import os from 'os' +import path from 'path' +import { Readable } from 'stream' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { Config } from '../../config' +import { + StemsArchiveJobData, + StemsArchiveJobResult +} from '../../jobs/createStemsArchive' +import { createSpaceManager } from '../spaceManager' +import { WorkerServices } from '../services' + +import { createStemsArchiveWorker } from './createStemsArchive' + +const PARENT_ID = 'PARENT' +const STEM_IDS = ['STEM1', 'STEM2'] + +type FetchBehavior = { + /** Track ids whose canonical download URL should 404 on every mirror. */ + brokenTrackIds?: string[] + /** Track ids whose /download endpoint should fail to resolve entirely. */ + unresolvableTrackIds?: string[] +} + +// Stand-in for node-fetch: /v1/tracks/{id}/download resolves to a 302 whose +// Location keeps the track id in the path (like the real signed cidstream +// URL), and mirror fetches stream a small body — or 404, per behavior. +const createMockFetch = (behavior: FetchBehavior = {}) => { + const mockFetch = async (url: string) => { + const downloadMatch = url.match(/\/v1\/tracks\/([^/]+)\/download/) + if (downloadMatch) { + const trackId = downloadMatch[1] + if (behavior.unresolvableTrackIds?.includes(trackId)) { + return { + status: 404, + statusText: 'Not Found', + headers: { get: () => null }, + body: undefined + } + } + return { + status: 302, + statusText: 'Found', + headers: { + get: (header: string) => + header === 'location' + ? `https://mirror-a.test/tracks/cidstream/${trackId}?signature=sig` + : null + }, + body: undefined + } + } + + const cidstreamMatch = url.match(/\/tracks\/cidstream\/([^?/]+)/) + if (cidstreamMatch) { + const trackId = cidstreamMatch[1] + if (behavior.brokenTrackIds?.includes(trackId)) { + return { + ok: false, + status: 404, + statusText: 'Not Found', + body: { resume: () => {} } + } + } + return { + ok: true, + status: 200, + statusText: 'OK', + body: Readable.from(Buffer.from(`audio-bytes-${trackId}`)) + } + } + + throw new Error(`Unexpected fetch in test: ${url}`) + } + return mockFetch as unknown as WorkerServices['fetch'] +} + +const createMockSdk = ({ + parentInspectFails = false +}: { parentInspectFails?: boolean } = {}) => { + const sdk = { + tracks: { + getTrack: vi.fn(async () => ({ + data: { + id: PARENT_ID, + title: 'Parent Track', + isDownloadable: true, + origFilename: 'parent.wav', + download: { mirrors: ['https://mirror-a.test'] } + } + })), + getTrackStems: vi.fn(async () => ({ + data: STEM_IDS.map((id) => ({ id, origFilename: `${id}.wav` })) + })), + inspectTrack: vi.fn(async ({ trackId }: { trackId: string }) => { + if (parentInspectFails && trackId === PARENT_ID) { + return { data: undefined } + } + return { data: { size: 64 } } + }), + getTrackDownloadUrl: vi.fn( + async ({ trackId }: { trackId: string }) => + `https://api.test/v1/tracks/${trackId}/download` + ) + } + } + return sdk as unknown as WorkerServices['sdk'] +} + +const createMockLogger = () => { + const mockLogger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + child: () => mockLogger + } + return mockLogger +} + +describe('createStemsArchiveWorker parent-track handling', () => { + let tmpDir: string + let mockLogger: ReturnType + + const createServices = ({ + fetchBehavior, + parentInspectFails + }: { + fetchBehavior?: FetchBehavior + parentInspectFails?: boolean + } = {}): WorkerServices => { + const config = { + archiverTmpDir: tmpDir, + maxDiskSpaceWaitSeconds: 5 + } as Config + return { + archiver, + config, + fetch: createMockFetch(fetchBehavior), + spaceManager: createSpaceManager({ + maxSpaceBytes: 1024 * 1024, + logger: mockLogger as unknown as WorkerServices['logger'] + }), + fs, + fsSync, + path, + sdk: createMockSdk({ parentInspectFails }), + logger: mockLogger as unknown as WorkerServices['logger'] + } + } + + const createJob = () => + ({ + data: { + jobId: 'test-job', + trackId: 123, + userId: 456, + messageHeader: 'message', + signatureHeader: 'signature', + includeParentTrack: true + } + }) as Job + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'archiver-test-')) + mockLogger = createMockLogger() + }) + + afterEach(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }) + }) + + it('includes the parent track when it downloads successfully', async () => { + const { processJob } = createStemsArchiveWorker(createServices()) + + const result = await processJob(createJob()) + + expect(result.outputFile).toMatch(/Parent Track\.zip$/) + expect(fsSync.existsSync(result.outputFile)).toBe(true) + expect(mockLogger.warn).not.toHaveBeenCalledWith( + expect.anything(), + expect.stringMatching(/Skipping parent track/) + ) + }) + + it('still produces a stems-only archive when every mirror 404s the parent', async () => { + const { processJob } = createStemsArchiveWorker( + createServices({ fetchBehavior: { brokenTrackIds: [PARENT_ID] } }) + ) + + const result = await processJob(createJob()) + + expect(fsSync.existsSync(result.outputFile)).toBe(true) + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ parentTrackId: PARENT_ID }), + 'Skipping parent track: download failed' + ) + }) + + it('still produces a stems-only archive when the parent download URL cannot be resolved', async () => { + const { processJob } = createStemsArchiveWorker( + createServices({ fetchBehavior: { unresolvableTrackIds: [PARENT_ID] } }) + ) + + const result = await processJob(createJob()) + + expect(fsSync.existsSync(result.outputFile)).toBe(true) + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ parentTrackId: PARENT_ID }), + 'Skipping parent track: download failed' + ) + }) + + it('skips the parent up front when its original file cannot be inspected', async () => { + const services = createServices({ parentInspectFails: true }) + const { processJob } = createStemsArchiveWorker(services) + + const result = await processJob(createJob()) + + expect(fsSync.existsSync(result.outputFile)).toBe(true) + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ parentTrackId: PARENT_ID }), + 'Skipping parent track: original file not available' + ) + // The parent must not be downloaded at all in this case + const sdk = services.sdk as unknown as { + tracks: { getTrackDownloadUrl: ReturnType } + } + for (const call of sdk.tracks.getTrackDownloadUrl.mock.calls) { + expect(call[0].trackId).not.toBe(PARENT_ID) + } + }) + + it('still fails the job when a stem download fails', async () => { + const { processJob } = createStemsArchiveWorker( + createServices({ fetchBehavior: { brokenTrackIds: [STEM_IDS[0]] } }) + ) + + // A stem failure aborts the job's controller to cancel sibling downloads, + // so the surfaced error is the abort, not the underlying download error. + await expect(processJob(createJob())).rejects.toThrow('Job aborted') + }) +}) diff --git a/apps/archiver/src/workers/createStemsArchive/createStemsArchive.ts b/apps/archiver/src/workers/createStemsArchive/createStemsArchive.ts index 5d5e278..7d19de8 100644 --- a/apps/archiver/src/workers/createStemsArchive/createStemsArchive.ts +++ b/apps/archiver/src/workers/createStemsArchive/createStemsArchive.ts @@ -101,33 +101,52 @@ export const createStemsArchiveWorker = (services: WorkerServices) => { throw new Error('No stems found for track') } - const filesToDownload = + // The parent track is best-effort. Stem-only uploads can report + // isDownloadable while having no original file to serve (empty CID, so + // every mirror 404s) — the stems the user asked for shouldn't be held + // hostage by that. Any parent-track failure downgrades the job to a + // stems-only archive instead of failing it. + let parentTrackFile: { id: string; origFilename?: string } | null = includeParentTrack && track.isDownloadable - ? [ - ...stems, - { ...track, origFilename: track.origFilename ?? track.title } - ] - : stems - - logger.debug({ files: filesToDownload }, 'Getting file sizes') - - const fileSizes = await Promise.all( - filesToDownload.map(async (file: { id: string }) => { - const inspection = await sdk.tracks.inspectTrack( - { - trackId: file.id, - original: true - }, - sdkRequestInit - ) - if (!inspection.data?.size) { - throw new Error(`File size not found for ${file.id}`) + ? { ...track, origFilename: track.origFilename ?? track.title } + : null + + const inspectFileSize = async (file: { id: string }) => { + const inspection = await sdk.tracks.inspectTrack( + { + trackId: file.id, + original: true + }, + sdkRequestInit + ) + if (!inspection.data?.size) { + throw new Error(`File size not found for ${file.id}`) + } + return inspection.data.size + } + + logger.debug({ stems, parentTrackFile }, 'Getting file sizes') + + const fileSizes = await Promise.all(stems.map(inspectFileSize)) + + let parentSizeBytes = 0 + if (parentTrackFile) { + try { + parentSizeBytes = await inspectFileSize(parentTrackFile) + } catch (error) { + if (abortController.signal.aborted) { + throw error } - return inspection.data.size - }) - ) + logger.warn( + { err: error, parentTrackId: parentTrackFile.id }, + 'Skipping parent track: original file not available' + ) + parentTrackFile = null + } + } - const totalSizeBytes = fileSizes.reduce((sum, size) => sum + size, 0) + const totalSizeBytes = + fileSizes.reduce((sum, size) => sum + size, 0) + parentSizeBytes logger.debug({ totalSizeBytes }, 'Calculated required disk space') logger.debug({ stems }, 'Downloading stems') @@ -173,36 +192,59 @@ export const createStemsArchiveWorker = (services: WorkerServices) => { // swapping the host. The signed path is host-agnostic so any mirror // that holds the file can serve it; if none can, we fall back to the // archive node (creatornode2) which is guaranteed to. - const downloadPromises = filesToDownload.map( - async (stem: { id: string; origFilename?: string }) => { - const url = await sdk.tracks.getTrackDownloadUrl({ - trackId: stem.id, - userId: hashedUserId, - userSignature: signatureHeader, - userData: messageHeader, - filename: stem.origFilename ?? '' - }) + const downloadTrackFile = async (stem: { + id: string + origFilename?: string + }) => { + const url = await sdk.tracks.getTrackDownloadUrl({ + trackId: stem.id, + userId: hashedUserId, + userSignature: signatureHeader, + userData: messageHeader, + filename: stem.origFilename ?? '' + }) + + const filePath = path.join(jobTempDir, stem.origFilename ?? 'file') + return downloadFile({ + url, + filePath, + jobId, + mirrors: trackMirrors, + signal: abortController.signal + }) + } - const filePath = path.join(jobTempDir, stem.origFilename ?? 'file') - return downloadFile({ - url, - filePath, - jobId, - mirrors: trackMirrors, - signal: abortController.signal + const downloadPromises = stems.map(downloadTrackFile) + // The parent download never rejects — it resolves to null on failure so + // a broken or missing original can't fail the stems that did download. + // (On abort it also resolves null; the stem promises reject in that case + // and fail the job through the catch below.) + const parentDownloadPromise = parentTrackFile + ? downloadTrackFile(parentTrackFile).catch((error) => { + if (!abortController.signal.aborted) { + logger.warn( + { err: error, parentTrackId: parentTrackFile?.id }, + 'Skipping parent track: download failed' + ) + } + return null }) - } - ) + : Promise.resolve(null) let downloadedFiles: string[] try { downloadedFiles = await Promise.all(downloadPromises) } catch (error) { abortController.abort() - await Promise.allSettled(downloadPromises) + await Promise.allSettled([...downloadPromises, parentDownloadPromise]) throw error } + const parentFilePath = await parentDownloadPromise + if (parentFilePath) { + downloadedFiles.push(parentFilePath) + } + logger.debug( { files: downloadedFiles }, 'Successfully downloaded all stems'