diff --git a/e2e/src/api/specs/asset-upload.e2e-spec.ts b/e2e/src/api/specs/asset-upload.e2e-spec.ts index c4feb40333..3440d95bf9 100644 --- a/e2e/src/api/specs/asset-upload.e2e-spec.ts +++ b/e2e/src/api/specs/asset-upload.e2e-spec.ts @@ -1073,6 +1073,7 @@ describe('/upload', () => { expect(status).toBe(204); expect(headers['upload-offset']).toBe('512'); expect(headers['upload-complete']).toBe('?0'); + expect(headers['upload-length']).toBe('512'); expect(headers['upload-limit']).toEqual('min-size=1, max-age=259200'); expect(headers['cache-control']).toBe('no-store'); }); @@ -1092,6 +1093,7 @@ describe('/upload', () => { const { status, headers } = await request(app).options('/upload'); expect(status).toBe(204); + expect(headers['accept-patch']).toBe('application/partial-upload'); expect(headers['upload-limit']).toEqual('min-size=1, max-age=259200'); }); }); diff --git a/server/src/dtos/asset-upload.dto.ts b/server/src/dtos/asset-upload.dto.ts index 9f702e62a2..8e75f9f9f3 100644 --- a/server/src/dtos/asset-upload.dto.ts +++ b/server/src/dtos/asset-upload.dto.ts @@ -190,7 +190,7 @@ function isUploadComplete(obj: any) { return false; } else if (uploadIncomplete === STRUCTURED_FALSE) { return true; - } else if (uploadComplete !== undefined) { + } else if (uploadIncomplete !== undefined) { throw new BadRequestException('upload-incomplete must be a structured boolean value'); } } diff --git a/server/src/repositories/asset.repository.ts b/server/src/repositories/asset.repository.ts index 970958ef49..cf1973cb3c 100644 --- a/server/src/repositories/asset.repository.ts +++ b/server/src/repositories/asset.repository.ts @@ -440,8 +440,8 @@ export class AssetRepository { } @GenerateSql({ params: [DummyValue.UUID] }) - async setComplete(assetId: string) { - await this.db + setComplete(assetId: string) { + return this.db .updateTable('asset as complete_asset') .set((eb) => ({ status: sql.lit(AssetStatus.Active), @@ -459,7 +459,8 @@ export class AssetRepository { })) .where('id', '=', assetId) .where('status', '=', sql.lit(AssetStatus.Partial)) - .execute(); + .returningAll() + .executeTakeFirst(); } @GenerateSql({ params: [DummyValue.UUID] }) diff --git a/server/src/repositories/storage.repository.ts b/server/src/repositories/storage.repository.ts index 92ee163958..5f38f0f24c 100644 --- a/server/src/repositories/storage.repository.ts +++ b/server/src/repositories/storage.repository.ts @@ -62,12 +62,12 @@ export class StorageRepository { return fs.writeFile(filepath, buffer, { flag: 'wx' }); } - createWriteStream(filepath: string): Writable { - return createWriteStream(filepath, { flags: 'w', flush: true, highWaterMark: 1024 * 1024 }); + createWriteStream(filepath: string, { flush }: { flush: boolean } = { flush: true }): Writable { + return createWriteStream(filepath, { flags: 'w', flush, highWaterMark: 1024 * 1024 }); } - createOrAppendWriteStream(filepath: string): Writable { - return createWriteStream(filepath, { flags: 'a', flush: true, highWaterMark: 1024 * 1024 }); + createOrAppendWriteStream(filepath: string, { flush }: { flush: boolean } = { flush: true }): Writable { + return createWriteStream(filepath, { flags: 'a', flush, highWaterMark: 1024 * 1024 }); } createOrOverwriteFile(filepath: string, buffer: Buffer) { diff --git a/server/src/services/asset-upload.service.ts b/server/src/services/asset-upload.service.ts index 486dac64e5..f2b63a8140 100644 --- a/server/src/services/asset-upload.service.ts +++ b/server/src/services/asset-upload.service.ts @@ -97,8 +97,8 @@ export class AssetUploadService extends BaseService { let checksumBuffer: Buffer | undefined; const writeStream = asset.isDuplicate - ? this.storageRepository.createWriteStream(asset.path) - : this.storageRepository.createOrAppendWriteStream(asset.path); + ? this.storageRepository.createWriteStream(asset.path, { flush: isComplete }) + : this.storageRepository.createOrAppendWriteStream(asset.path, { flush: isComplete }); this.pipe(req, writeStream, contentLength); if (isComplete) { const hash = createHash('sha1'); @@ -159,7 +159,7 @@ export class AssetUploadService extends BaseService { return; } - const writeStream = this.storageRepository.createOrAppendWriteStream(path); + const writeStream = this.storageRepository.createOrAppendWriteStream(path, { flush: uploadComplete }); this.pipe(req, writeStream, contentLength); await new Promise((resolve, reject) => writeStream.on('close', resolve).on('error', reject)); this.setCompleteHeader(res, version, uploadComplete); @@ -213,18 +213,21 @@ export class AssetUploadService extends BaseService { const offset = await this.getCurrentOffset(asset.path); this.setCompleteHeader(res, version, asset.status !== AssetStatus.Partial); - res - .status(204) - .setHeader('Upload-Offset', offset.toString()) - .setHeader('Cache-Control', 'no-store') - .setHeader('Upload-Limit', this.getUploadLimits(backup)) - .send(); + res.status(204).setHeader('Upload-Offset', offset.toString()).setHeader('Cache-Control', 'no-store'); + if (asset.size) { + res.setHeader('Upload-Length', asset.size.toString()); + } + res.setHeader('Upload-Limit', this.getUploadLimits(backup)).send(); }); } async getUploadOptions(res: Response): Promise { const { backup } = await this.getConfig({ withCache: true }); - res.status(204).setHeader('Upload-Limit', this.getUploadLimits(backup)).send(); + res + .status(204) + .setHeader('Accept-Patch', 'application/partial-upload') + .setHeader('Upload-Limit', this.getUploadLimits(backup)) + .send(); } @OnJob({ name: JobName.PartialAssetCleanupQueueAll, queue: QueueName.BackgroundTask }) @@ -324,13 +327,19 @@ export class AssetUploadService extends BaseService { async onComplete({ id, path, fileModifiedAt }: { id: string; path: string; fileModifiedAt: Date }) { this.logger.log('Completing upload for asset', id); - const jobData = { name: JobName.AssetExtractMetadata, data: { id, source: 'upload' } } as const; - await withRetry(() => this.assetRepository.setComplete(id)); + const asset = await withRetry(() => this.assetRepository.setComplete(id)); + if (!asset) { + this.logger.error(`Failed to mark asset ${id} as complete: not found or already completed`); + return; + } + try { await withRetry(() => this.storageRepository.utimes(path, new Date(), fileModifiedAt)); + await this.eventRepository.emit('AssetCreate', { asset }); } catch (error: any) { - this.logger.error(`Failed to update times for ${path}: ${error.message}`); + this.logger.error(`onComplete error for ${path}: ${error.message}`); } + const jobData = { name: JobName.AssetExtractMetadata, data: { id, source: 'upload' } } as const; await withRetry(() => this.jobRepository.queue(jobData)); }