fix: unpkg lock (#629)

> When accessing the unpkg service, when the packages have not yet been
synchronized, will lead to multiple synchronization attempts
concurrently causing db insert errors.
* 🔒 Added a Redis lock for the `ensurePackageVersionFilesSync` function,
with a default timeout of 60 seconds.
* 🥸 Admin PUT requests and the package version auto sync process are not
restricted by this.

> 当访问 unpkg 服务时，如果访问存量未同步的包，可能导致多次同步并发报错
* 🔒 为 ensurePackageVersionFilesSync 添加 redis 锁，默认超时 60s
* 🥸 管理员手动 PUT 请求和包同步流程不受限制

app/core/service/PackageVersionFileService.ts

@@ -21,6 +21,8 @@ import { PackageVersionFile } from '../entity/PackageVersionFile';
 import { PackageVersion } from '../entity/PackageVersion';
 import { Package } from '../entity/Package';
 import { PackageManagerService } from './PackageManagerService';
+import { CacheAdapter } from '../../common/adapter/CacheAdapter';
+import { ConflictError } from 'egg-errors';
 
 @SingletonProto({
   accessLevel: AccessLevel.PUBLIC,
@@ -34,6 +36,8 @@ export class PackageVersionFileService extends AbstractService {
   private readonly distRepository: DistRepository;
   @Inject()
   private readonly packageManagerService: PackageManagerService;
+  @Inject()
+  private readonly cacheAdapter: CacheAdapter;
 
   async listPackageVersionFiles(pkgVersion: PackageVersion, directory: string) {
     await this.#ensurePackageVersionFilesSync(pkgVersion);
@@ -50,8 +54,16 @@ export class PackageVersionFileService extends AbstractService {
   async #ensurePackageVersionFilesSync(pkgVersion: PackageVersion) {
     const hasFiles = await this.packageVersionFileRepository.hasPackageVersionFiles(pkgVersion.packageVersionId);
     if (!hasFiles) {
-      await this.syncPackageVersionFiles(pkgVersion);
+      const lockRes = await this.cacheAdapter.usingLock(`${pkgVersion.packageVersionId}:syncFiles`, 60, async () => {
+        await this.syncPackageVersionFiles(pkgVersion);
+      });
+      // lock fail
+      if (!lockRes) {
+        this.logger.warn('[package:version:syncPackageVersionFiles] check lock fail');
+        throw new ConflictError('Package version file sync is currently in progress. Please try again later.');
+      }
     }
+
   }
 
   // 基于 latest version 同步 package readme

test/port/controller/PackageVersionFileController/listFiles.test.ts

@@ -1,6 +1,8 @@
 import { strict as assert } from 'node:assert';
+import { setTimeout } from 'node:timers/promises';
 import { app, mock } from 'egg-mock/bootstrap';
 import { TestUtil } from '../../../../test/TestUtil';
+import { PackageVersionFileService } from '../../../../app/core/service/PackageVersionFileService';
 
 describe('test/port/controller/PackageVersionFileController/listFiles.test.ts', () => {
   let publisher;
@@ -331,5 +333,24 @@ describe('test/port/controller/PackageVersionFileController/listFiles.test.ts',
       assert(!res.headers['cache-control']);
       assert.equal(res.body.error, '[NOT_FOUND] @cnpm/foonot-exists@1.0.40000404 not found');
     });
+
+    it('should conflict when syncing', async () => {
+      mock(app.config.cnpmcore, 'enableUnpkg', true);
+      const { pkg } = await TestUtil.createPackage({
+        name: '@cnpm/banana',
+        version: '1.0.0',
+        versionObject: {
+          description: 'mock mock',
+        },
+      });
+      let called = 0;
+      mock(PackageVersionFileService.prototype, 'syncPackageVersionFiles', async () => {
+        called++;
+        await setTimeout(50);
+      });
+      const resList = await Promise.all([ 0, 1 ].map(() => app.httpRequest().get(`/${pkg.name}/1.0.0/files/`)));
+      assert.equal(called, 1);
+      assert.equal(resList.filter(res => res.status === 409 && res.body.error === '[CONFLICT] Package version file sync is currently in progress. Please try again later.').length, 1);
+    });
   });
 });