update to latest draft, apply old upload changes

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');
     });
   });

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');
   }
 }

server/src/repositories/asset.repository.ts

@@ -440,8 +440,8 @@ export class AssetRepository {
   }
 
   @GenerateSql({ params: [DummyValue.UUID] })
-  async setComplete(assetId: string) {
+  setComplete(assetId: string) {
-    await this.db
+    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] })

server/src/repositories/storage.repository.ts

@@ -62,12 +62,12 @@ export class StorageRepository {
     return fs.writeFile(filepath, buffer, { flag: 'wx' });
   }
 
-  createWriteStream(filepath: string): Writable {
+  createWriteStream(filepath: string, { flush }: { flush: boolean } = { flush: true }): Writable {
-    return createWriteStream(filepath, { flags: 'w', flush: true, highWaterMark: 1024 * 1024 });
+    return createWriteStream(filepath, { flags: 'w', flush, highWaterMark: 1024 * 1024 });
   }
 
-  createOrAppendWriteStream(filepath: string): Writable {
+  createOrAppendWriteStream(filepath: string, { flush }: { flush: boolean } = { flush: true }): Writable {
-    return createWriteStream(filepath, { flags: 'a', flush: true, highWaterMark: 1024 * 1024 });
+    return createWriteStream(filepath, { flags: 'a', flush, highWaterMark: 1024 * 1024 });
   }
 
   createOrOverwriteFile(filepath: string, buffer: Buffer) {

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.createWriteStream(asset.path, { flush: isComplete })
-        : this.storageRepository.createOrAppendWriteStream(asset.path);
+        : 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
+      res.status(204).setHeader('Upload-Offset', offset.toString()).setHeader('Cache-Control', 'no-store');
-        .status(204)
+      if (asset.size) {
-        .setHeader('Upload-Offset', offset.toString())
+        res.setHeader('Upload-Length', asset.size.toString());
-        .setHeader('Cache-Control', 'no-store')
+      }
-        .setHeader('Upload-Limit', this.getUploadLimits(backup))
+      res.setHeader('Upload-Limit', this.getUploadLimits(backup)).send();
-        .send();
     });
   }
 
   async getUploadOptions(res: Response): Promise<void> {
     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;
+    const asset = await withRetry(() => this.assetRepository.setComplete(id));
-    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));
   }