📦 NEW: Impl sync package like cnpmjs.org (#23)

- Task list base on MySQL
2021-12-12 12:50:41 +08:00
parent faa37a431d
commit bdeadf9c74
32 changed files with 1782 additions and 37 deletions
--- a/README.md
+++ b/README.md
@@ -2,6 +2,7 @@

 [![Node.js CI](https://github.com/cnpm/cnpmcore/actions/workflows/nodejs.yml/badge.svg)](https://github.com/cnpm/cnpmcore/actions/workflows/nodejs.yml)
 [![Test coverage](https://img.shields.io/codecov/c/github/cnpm/cnpmcore.svg?style=flat-square)](https://codecov.io/gh/cnpm/cnpmcore)
+[![emoji-log](https://cdn.rawgit.com/ahmadawais/stuff/ca97874/emoji-log/flat.svg)](https://github.com/ahmadawais/Emoji-Log/)

 Reimplementation based on [cnpmjs.org](https://github.com/cnpm/cnpmjs.org) with TypeScript.

--- a/app/common/adapter/NFSAdapter.ts
+++ b/app/common/adapter/NFSAdapter.ts
@@ -21,13 +21,15 @@ export class NFSAdapter {
  @Inject()
  private readonly logger: EggLogger;

-  @AsyncTimer(INSTANCE_NAME)
  async uploadBytes(storeKey: string, bytes: Uint8Array) {
    this.logger.info('[%s:uploadBytes] key: %s, bytes: %d', INSTANCE_NAME, storeKey, bytes.length);
    await this.nfsClientAdapter.client.uploadBuffer(bytes, { key: storeKey });
  }

-  @AsyncTimer(INSTANCE_NAME)
+  async appendBytes(storeKey: string, bytes: Uint8Array) {
+    await this.nfsClientAdapter.client.appendBuffer(bytes, { key: storeKey });
+  }
+
  async uploadFile(storeKey: string, file: string) {
    this.logger.info('[%s:uploadFile] key: %s, file: %s', INSTANCE_NAME, storeKey, file);
    await this.nfsClientAdapter.client.upload(file, { key: storeKey });
@@ -39,12 +41,10 @@ export class NFSAdapter {
    await this.nfsClientAdapter.client.remove(storeKey);
  }

-  @AsyncTimer(INSTANCE_NAME)
  async getStream(storeKey: string): Promise<Readable | undefined> {
    return await this.nfsClientAdapter.client.createDownloadStream(storeKey);
  }

-  @AsyncTimer(INSTANCE_NAME)
  async getBytes(storeKey: string): Promise<Uint8Array | undefined> {
    return await this.nfsClientAdapter.client.readBytes(storeKey);
  }
--- a/app/common/adapter/NPMRegistry.ts
+++ b/app/common/adapter/NPMRegistry.ts
@@ -0,0 +1,89 @@
+import { createWriteStream } from 'fs';
+import { mkdir } from 'fs/promises';
+import path from 'path';
+import { randomBytes } from 'crypto';
+import {
+  ContextProto,
+  AccessLevel,
+  Inject,
+} from '@eggjs/tegg';
+import {
+  EggLogger,
+  EggContextHttpClient,
+  EggAppConfig,
+} from 'egg';
+import { HttpMethod } from 'urllib';
+import dayjs from '../dayjs';
+
+const INSTANCE_NAME = 'npmRegistry';
+
+@ContextProto({
+  name: INSTANCE_NAME,
+  accessLevel: AccessLevel.PUBLIC,
+})
+export class NPMRegistry {
+  @Inject()
+  private readonly logger: EggLogger;
+  @Inject()
+  private readonly httpclient: EggContextHttpClient;
+  @Inject()
+  private config: EggAppConfig;
+  private timeout = 10000;
+
+  get registry(): string {
+    return this.config.cnpmcore.sourceRegistry;
+  }
+
+  public async getFullManifests(fullname: string) {
+    const url = `${this.registry}/${encodeURIComponent(fullname)}`;
+    return await this.request('GET', url);
+  }
+
+  public async downloadTarball(tarball: string) {
+    const uri = new URL(tarball);
+    const tmpfile = path.join(this.config.dataDir, 'downloads', dayjs().format('YYYY/MM/DD'),
+      `${randomBytes(10).toString('hex')}-${path.basename(uri.pathname)}`);
+    await mkdir(path.dirname(tmpfile), { recursive: true });
+    const writeStream = createWriteStream(tmpfile);
+    const result = await this.request('GET', tarball, undefined, { timeout: 120000, writeStream });
+    return {
+      ...result,
+      tmpfile,
+    };
+  }
+
+  // app.put('/:name/sync', sync.sync);
+  public async createSyncTask(fullname: string) {
+    const url = `${this.registry}/${encodeURIComponent(fullname)}/sync?sync_upstream=true&nodeps=true`;
+    // {
+    //   ok: true,
+    //   logId: logId
+    // };
+    return await this.request('PUT', url);
+  }
+
+  // app.get('/:name/sync/log/:id', sync.getSyncLog);
+  public async getSyncTask(fullname: string, id: string, offset: number) {
+    const url = `${this.registry}/${encodeURIComponent(fullname)}/sync/log/${id}?offset=${offset}`;
+    // { ok: true, syncDone: syncDone, log: log }
+    return await this.request('GET', url);
+  }
+
+  private async request(method: HttpMethod, url: string, params?: object, options?: object) {
+    const res = await this.httpclient.request(url, {
+      method,
+      data: params,
+      dataType: 'json',
+      timing: true,
+      timeout: this.timeout,
+      followRedirect: true,
+      ...options,
+    });
+    this.logger.info('[NPMRegistry:request] %s %s, status: %s', method, url, res.status);
+    return {
+      method,
+      url,
+      ...res,
+    };
+  }
+}
--- a/app/common/enum/Task.ts
+++ b/app/common/enum/Task.ts
@@ -0,0 +1,11 @@
+export enum TaskType {
+  SyncPackage = 'sync_package',
+}
+
+export enum TaskState {
+  Waiting = 'waiting',
+  Processing = 'processing',
+  Success = 'success',
+  Fail = 'fail',
+  Timeout = 'timeout'
+}
--- a/app/core/entity/Task.ts
+++ b/app/core/entity/Task.ts
@@ -0,0 +1,64 @@
+import { Entity, EntityData } from './Entity';
+import { EasyData, EntityUtil } from '../util/EntityUtil';
+import { TaskType, TaskState } from '../../common/enum/Task';
+import dayjs from '../../common/dayjs';
+
+interface TaskData extends EntityData {
+  taskId: string;
+  type: TaskType;
+  state: TaskState;
+  targetName: string;
+  authorId: string;
+  authorIp: string;
+  data: unknown;
+  logPath?: string;
+}
+
+export type SyncPackageTaskOptions = {
+  authorId?: string;
+  authorIp?: string;
+  tips?: string;
+  skipDependencies?: boolean;
+};
+
+export class Task extends Entity {
+  taskId: string;
+  type: TaskType;
+  state: TaskState;
+  targetName: string;
+  authorId: string;
+  authorIp: string;
+  data: unknown;
+  logPath: string;
+
+  constructor(data: TaskData) {
+    super(data);
+    this.taskId = data.taskId;
+    this.type = data.type;
+    this.state = data.state;
+    this.targetName = data.targetName;
+    this.authorId = data.authorId;
+    this.authorIp = data.authorIp;
+    this.data = data.data;
+    this.logPath = data.logPath ?? '';
+  }
+
+  private static create(data: EasyData<TaskData, 'taskId'>): Task {
+    const newData = EntityUtil.defaultData(data, 'taskId');
+    return new Task(newData);
+  }
+
+  public static createSyncPackage(fullname: string, options?: SyncPackageTaskOptions): Task {
+    const data = {
+      type: TaskType.SyncPackage,
+      state: TaskState.Waiting,
+      targetName: fullname,
+      authorId: options?.authorId ?? '',
+      authorIp: options?.authorIp ?? '',
+      data: { tips: options?.tips, skipDependencies: options?.skipDependencies },
+    };
+    const task = this.create(data);
+    task.logPath = `/packages/${fullname}/syncs/${dayjs().format('YYYY/MM/DDHHMM')}-${task.taskId}.log`;
+    return task;
+  }
+}
--- a/app/core/service/PackageManagerService.ts
+++ b/app/core/service/PackageManagerService.ts
@@ -18,6 +18,7 @@ import { Package } from '../entity/Package';
 import { PackageVersion } from '../entity/PackageVersion';
 import { PackageTag } from '../entity/PackageTag';
 import { User } from '../entity/User';
+import { Dist } from '../entity/Dist';
 import {
  PACKAGE_MAINTAINER_CHANGED,
  PACKAGE_VERSION_ADDED,
@@ -26,7 +27,6 @@ import {
  PACKAGE_TAG_CHANGED,
  PACKAGE_TAG_REMOVED,
 } from '../event';
-import { Dist } from '../entity/Dist';
 import { AbstractService } from './AbstractService';

 export interface PublishPackageCmd {
@@ -44,10 +44,11 @@ export interface PublishPackageCmd {
    content?: Uint8Array;
    // sync worker will use localFile field
    localFile?: string;
-    meta?: object;
  }, 'content' | 'localFile'>;
  tag?: string;
  isPrivate: boolean;
+  // only use on sync package
+  publishTime?: Date;
 }

@ContextProto({
@@ -146,6 +147,7 @@ export class PackageManagerService extends AbstractService {
      devDependencies: cmd.packageJson.devDependencies,
      bundleDependencies: cmd.packageJson.bundleDependencies,
      peerDependencies: cmd.packageJson.peerDependencies,
+      peerDependenciesMeta: cmd.packageJson.peerDependenciesMeta,
      bin: cmd.packageJson.bin,
      os: cmd.packageJson.os,
      cpu: cmd.packageJson.cpu,
@@ -166,7 +168,7 @@ export class PackageManagerService extends AbstractService {
    pkgVersion = PackageVersion.create({
      packageId: pkg.packageId,
      version: cmd.version,
-      publishTime: new Date(),
+      publishTime: cmd.publishTime || new Date(),
      manifestDist: pkg.createManifest(cmd.version, {
        size: manifestDistBytes.length,
        shasum: manifestDistIntegrity.shasum,
@@ -205,6 +207,14 @@ export class PackageManagerService extends AbstractService {
    this.eventBus.emit(PACKAGE_MAINTAINER_CHANGED, pkg.packageId);
  }

+  async savePackageMaintainer(pkg: Package, maintainer: User) {
+    const newRecord = await this.packageRepository.savePackageMaintainer(pkg.packageId, maintainer.userId);
+    if (newRecord) {
+      await this.refreshPackageManifestsToDists(pkg);
+      this.eventBus.emit(PACKAGE_MAINTAINER_CHANGED, pkg.packageId);
+    }
+  }
+
  async listPackageFullManifests(scope: string, name: string, expectEtag: string | undefined) {
    return await this._listPacakgeFullOrAbbreviatedManifests(scope, name, expectEtag, true);
  }
@@ -269,6 +279,12 @@ export class PackageManagerService extends AbstractService {
    await this.refreshPackageManifestsToDists(pkg);
  }

+  public async savePackageVersionManifest(pkg: Package, pkgVersion: PackageVersion, mergeManifest: object, mergeAbbreviated: object) {
+    await this.mergeManifestDist(pkgVersion.manifestDist, mergeManifest);
+    await this.mergeManifestDist(pkgVersion.abbreviatedDist, mergeAbbreviated);
+    await this.refreshPackageManifestsToDists(pkg);
+  }
+
  public async removePackageVersion(pkg: Package, pkgVersion: PackageVersion) {
    // remove nfs dists
    await Promise.all([
--- a/app/core/service/PackageSyncerService.ts
+++ b/app/core/service/PackageSyncerService.ts
@@ -0,0 +1,351 @@
+import {
+  AccessLevel,
+  ContextProto,
+  Inject,
+} from '@eggjs/tegg';
+import { setTimeout } from 'timers/promises';
+import { rm } from 'fs/promises';
+import { NFSAdapter } from '../../common/adapter/NFSAdapter';
+import { NPMRegistry } from '../../common/adapter/NPMRegistry';
+import { getScopeAndName } from '../../common/PackageUtil';
+import { TaskState, TaskType } from '../../common/enum/Task';
+import { TaskRepository } from '../../repository/TaskRepository';
+import { PackageRepository } from '../../repository/PackageRepository';
+import { Task, SyncPackageTaskOptions } from '../entity/Task';
+import { AbstractService } from './AbstractService';
+import { UserService } from './UserService';
+import { PackageManagerService } from './PackageManagerService';
+import { User } from '../entity/User';
+
+function isoNow() {
+  return new Date().toISOString();
+}
+
+@ContextProto({
+  accessLevel: AccessLevel.PUBLIC,
+})
+export class PackageSyncerService extends AbstractService {
+  @Inject()
+  private readonly taskRepository: TaskRepository;
+  @Inject()
+  private readonly packageRepository: PackageRepository;
+  @Inject()
+  private readonly nfsAdapter: NFSAdapter;
+  @Inject()
+  private readonly npmRegistry: NPMRegistry;
+  @Inject()
+  private readonly userService: UserService;
+  @Inject()
+  private readonly packageManagerService: PackageManagerService;
+
+  public async createTask(fullname: string, options?: SyncPackageTaskOptions) {
+    const task = Task.createSyncPackage(fullname, options);
+    await this.taskRepository.saveTask(task);
+    return task;
+  }
+
+  public async findTask(taskId: string) {
+    const task = await this.taskRepository.findTask(taskId);
+    return task;
+  }
+
+  public async findTaskLog(task: Task) {
+    return await this.nfsAdapter.getDownloadUrlOrStream(task.logPath);
+  }
+
+  public async findExecuteTask() {
+    const task = await this.taskRepository.executeWaitingTask(TaskType.SyncPackage);
+    if (task && task.attempts > 3) {
+      task.state = TaskState.Timeout;
+      task.attempts -= 1;
+      await this.taskRepository.saveTaskToHistory(task);
+      return null;
+    }
+    return task;
+  }
+
+  private async syncUpstream(task: Task) {
+    const registry = this.npmRegistry.registry;
+    const fullname = task.targetName;
+    let logs: string[] = [];
+    let logId = '';
+    logs.push(`[${isoNow()}][UP] 🚧🚧🚧🚧🚧 Waiting sync "${fullname}" task on ${registry} 🚧🚧🚧🚧🚧`);
+    const failEnd = `❌❌❌❌❌ Sync ${registry}/${fullname} 🚮 give up 🚮 ❌❌❌❌❌`;
+    try {
+      const { data, status, res } = await this.npmRegistry.createSyncTask(fullname);
+      logs.push(`[${isoNow()}][UP] HTTP [${status}] timing: ${JSON.stringify(res.timing)}, data: ${JSON.stringify(data)}`);
+      logId = data.logId;
+    } catch (err: any) {
+      const status = err.status || 'unknow';
+      logs.push(`[${isoNow()}][UP] ❌ Sync ${fullname} fail, create sync task error: ${err}, status: ${status}`);
+      logs.push(`[${isoNow()}][UP] ${failEnd}`);
+      await this.appendTaskLog(task, logs.join('\n'));
+      return;
+    }
+    if (!logId) {
+      logs.push(`[${isoNow()}][UP] ❌ Sync ${fullname} fail, missing logId`);
+      logs.push(`[${isoNow()}][UP] ${failEnd}`);
+      await this.appendTaskLog(task, logs.join('\n'));
+      return;
+    }
+    const startTime = Date.now();
+    const maxTimeout = this.config.cnpmcore.sourceRegistrySyncTimeout;
+    let logUrl = '';
+    let offset = 0;
+    let useTime = Date.now() - startTime;
+    while (useTime < maxTimeout) {
+      // sleep 1s ~ 6s in random
+      await setTimeout(1000 + Math.random() * 5000);
+      try {
+        const { data, status, url } = await this.npmRegistry.getSyncTask(fullname, logId, offset);
+        useTime = Date.now() - startTime;
+        if (!logUrl) {
+          logUrl = url;
+        }
+        const log = data && data.log || '';
+        offset += log.length;
+        if (data && data.syncDone) {
+          logs.push(`[${isoNow()}][UP] 🟢 Sync ${fullname} success [${useTime}ms], log: ${logUrl}, offset: ${offset}`);
+          logs.push(`[${isoNow()}][UP] 🟢🟢🟢🟢🟢 ${registry}/${fullname} 🟢🟢🟢🟢🟢`);
+          await this.appendTaskLog(task, logs.join('\n'));
+          return;
+        }
+        logs.push(`[${isoNow()}][UP] 🚧 HTTP [${status}] [${useTime}ms], offset: ${offset}`);
+        await this.appendTaskLog(task, logs.join('\n'));
+        logs = [];
+      } catch (err: any) {
+        useTime = Date.now() - startTime;
+        const status = err.status || 'unknow';
+        logs.push(`[${isoNow()}][UP] 🚧 HTTP [${status}] [${useTime}ms] error: ${err}`);
+      }
+    }
+    // timeout
+    logs.push(`[${isoNow()}][UP] ❌ Sync ${fullname} fail, timeout, log: ${logUrl}, offset: ${offset}`);
+    logs.push(`[${isoNow()}][UP] ${failEnd}`);
+    await this.appendTaskLog(task, logs.join('\n'));
+  }
+
+  public async executeTask(task: Task) {
+    const fullname = task.targetName;
+    const { tips, skipDependencies } = task.data as SyncPackageTaskOptions;
+    const registry = this.npmRegistry.registry;
+    if (this.config.cnpmcore.sourceRegistryIsCNpm) {
+      // create sync task on sourceRegistry and skipDependencies = true
+      await this.syncUpstream(task);
+    }
+    let logs: string[] = [];
+    if (tips) {
+      logs.push(`[${isoNow()}] 👉👉👉👉👉 Tips: ${tips} 👈👈👈👈👈`);
+    }
+    logs.push(`[${isoNow()}] 🚧🚧🚧🚧🚧 Start sync "${fullname}" from ${registry}, skipDependencies: ${!!skipDependencies} 🚧🚧🚧🚧🚧`);
+    const logUrl = `${this.config.cnpmcore.registry}/-/package/${fullname}/syncs/${task.taskId}/log`;
+    let result;
+    try {
+      result = await this.npmRegistry.getFullManifests(fullname);
+    } catch (err: any) {
+      const status = err.status || 'unknow';
+      logs.push(`[${isoNow()}] ❌ Synced ${fullname} fail, request manifests error: ${err}, status: ${status}, log: ${logUrl}`);
+      logs.push(`[${isoNow()}] ❌❌❌❌❌ ${fullname} ❌❌❌❌❌`);
+      await this.finishTask(task, TaskState.Fail, logs.join('\n'));
+      return;
+    }
+    const { url, data, headers, res, status } = result;
+    const readme: string = data.readme;
+    // "time": {
+    //   "created": "2021-03-27T12:30:23.891Z",
+    //   "0.0.2": "2021-03-27T12:30:24.349Z",
+    //   "modified": "2021-12-08T14:59:57.264Z",
+    const timeMap = data.time || {};
+    const failEnd = `❌❌❌❌❌ ${url || fullname} ❌❌❌❌❌`;
+    logs.push(`[${isoNow()}] HTTP [${status}] content-length: ${headers['content-length']}, timing: ${JSON.stringify(res.timing)}`);
+
+    // 1. save maintainers
+    // maintainers: [
+    //   { name: 'bomsy', email: 'b4bomsy@gmail.com' },
+    //   { name: 'jasonlaster11', email: 'jason.laster.11@gmail.com' }
+    // ],
+    const maintainers = data.maintainers;
+    const users: User[] = [];
+    if (Array.isArray(maintainers) && maintainers.length > 0) {
+      logs.push(`[${isoNow()}] Syncing maintainers: ${JSON.stringify(maintainers)}`);
+      for (const maintainer of maintainers) {
+        if (maintainer.name && maintainer.email) {
+          const user = await this.userService.savePublicUser(maintainer.name, maintainer.email);
+          users.push(user);
+          logs.push(`[${isoNow()}] Synced ${maintainer.name} => ${user.name}(${user.userId})`);
+        }
+      }
+    }
+
+    if (users.length === 0) {
+      // invalid maintainers, sync fail
+      logs.push(`[${isoNow()}] ❌ Invalid maintainers: ${JSON.stringify(maintainers)}, log: ${logUrl}`);
+      logs.push(`[${isoNow()}] ${failEnd}`);
+      await this.finishTask(task, TaskState.Fail, logs.join('\n'));
+      return;
+    }
+
+    const dependenciesSet = new Set<string>();
+
+    const [ scope, name ] = getScopeAndName(fullname);
+    const { data: existsData } = await this.packageManagerService.listPackageFullManifests(scope, name, undefined);
+    const existsVersionMap = existsData && existsData.versions || {};
+    const existsVersionCount = Object.keys(existsVersionMap).length;
+    // 2. save versions
+    const versions = Object.values<any>(data.versions || {});
+    logs.push(`[${isoNow()}] Syncing versions ${existsVersionCount} => ${versions.length}`);
+    let syncVersionCount = 0;
+    const differentMetas: any[] = [];
+    for (const item of versions) {
+      const version: string = item.version;
+      if (!version) continue;
+      const existsItem = existsVersionMap[version];
+      if (existsItem) {
+        // check metaDataKeys, if different value, override exists one
+        // https://github.com/cnpm/cnpmjs.org/issues/1667
+        const metaDataKeys = [ 'peerDependenciesMeta', 'os', 'cpu', 'workspaces', 'hasInstallScript', 'deprecated' ];
+        let diffMeta;
+        for (const key of metaDataKeys) {
+          if (JSON.stringify(item[key]) !== JSON.stringify(existsItem[key])) {
+            if (!diffMeta) diffMeta = {};
+            diffMeta[key] = item[key];
+          }
+        }
+        if (diffMeta) {
+          differentMetas.push([ existsItem, diffMeta ]);
+        }
+        continue;
+      }
+      const description: string = item.description;
+      // "dist": {
+      //   "shasum": "943e0ec03df00ebeb6273a5b94b916ba54b47581",
+      //   "tarball": "https://registry.npmjs.org/foo/-/foo-1.0.0.tgz"
+      // },
+      const dist = item.dist;
+      const tarball: string = dist && dist.tarball;
+      if (!tarball) {
+        logs.push(`[${isoNow()}] ❌ Synced version ${version} fail, missing tarball, dist: ${JSON.stringify(dist)}`);
+        await this.appendTaskLog(task, logs.join('\n'));
+        logs = [];
+        continue;
+      }
+      const publishTimeISO = timeMap[version];
+      const publishTime = publishTimeISO ? new Date(publishTimeISO) : new Date();
+      const delay = Date.now() - publishTime.getTime();
+      logs.push(`[${isoNow()}] 🚧 Syncing version ${version}, delay: ${delay}ms [${publishTimeISO}], tarball: ${tarball}`);
+      let localFile: string;
+      try {
+        const { tmpfile, status, headers, res } = await this.npmRegistry.downloadTarball(tarball);
+        localFile = tmpfile;
+        logs.push(`[${isoNow()}] HTTP [${status}] content-length: ${headers['content-length']}, timing: ${JSON.stringify(res.timing)} => ${localFile}`);
+        if (status !== 200) {
+          logs.push(`[${isoNow()}] ❌ Synced version ${version} fail, download tarball status error: ${status}`);
+          await this.appendTaskLog(task, logs.join('\n'));
+          logs = [];
+          if (localFile) {
+            await rm(localFile, { force: true });
+          }
+          continue;
+        }
+      } catch (err: any) {
+        const status = err.status || 'unknow';
+        logs.push(`[${isoNow()}] ❌ Synced version ${version} fail, download tarball error: ${err}, status: ${status}`);
+        await this.appendTaskLog(task, logs.join('\n'));
+        logs = [];
+        continue;
+      }
+      const pkgVersion = await this.packageManagerService.publish({
+        scope,
+        name,
+        version,
+        description,
+        packageJson: item,
+        readme,
+        dist: {
+          localFile,
+        },
+        isPrivate: false,
+        publishTime,
+      }, users[0]);
+      syncVersionCount++;
+      logs.push(`[${isoNow()}] 🟢 Synced version ${version} success, packageVersionId: ${pkgVersion.packageVersionId}, db id: ${pkgVersion.id}`);
+      await this.appendTaskLog(task, logs.join('\n'));
+      logs = [];
+      await rm(localFile, { force: true });
+      if (!skipDependencies) {
+        const dependencies = item.dependencies || {};
+        for (const dependencyName in dependencies) {
+          dependenciesSet.add(dependencyName);
+        }
+      }
+    }
+    const pkg = await this.packageRepository.findPackage(scope, name);
+    if (!pkg) {
+      // sync all versions fail in the first time
+      logs.push(`[${isoNow()}] ❌ All versions sync fail, package not exists, log: ${logUrl}`);
+      logs.push(`[${isoNow()}] ${failEnd}`);
+      await this.finishTask(task, TaskState.Fail, logs.join('\n'));
+      return;
+    }
+
+    // 2.1 save differentMetas
+    for (const [ existsItem, diffMeta ] of differentMetas) {
+      const pkgVersion = await this.packageRepository.findPackageVersion(pkg.packageId, existsItem.version);
+      await this.packageManagerService.savePackageVersionManifest(pkg, pkgVersion!, diffMeta, diffMeta);
+      syncVersionCount++;
+      logs.push(`[${isoNow()}] 🟢 Synced version ${existsItem.version} success, different meta: ${JSON.stringify(diffMeta)}`);
+    }
+
+    if (syncVersionCount > 0) {
+      logs.push(`[${isoNow()}] 🟢 Synced ${syncVersionCount} versions`);
+    }
+
+    // 3. update tags
+    // "dist-tags": {
+    //   "latest": "0.0.7"
+    // },
+    const distTags = data['dist-tags'] || {};
+    for (const tag in distTags) {
+      const version = distTags[tag];
+      await this.packageManagerService.savePackageTag(pkg, tag, version);
+    }
+    logs.push(`[${isoNow()}] 🟢 Synced tags: ${JSON.stringify(distTags)}`);
+
+    // 4. add package maintainers
+    for (const user of users) {
+      await this.packageManagerService.savePackageMaintainer(pkg!, user);
+    }
+
+    // 5. add deps sync task
+    for (const dependencyName of dependenciesSet) {
+      const existsTask = await this.taskRepository.findTaskByTargetName(fullname, TaskType.SyncPackage, TaskState.Waiting);
+      if (existsTask) {
+        logs.push(`[${isoNow()}] Has dependency "${dependencyName}" sync task: ${existsTask.taskId}, db id: ${existsTask.id}`);
+        continue;
+      }
+      const tips = `Sync cause by "${fullname}" dependencies, parent task: ${task.taskId}`;
+      const dependencyTask = await this.createTask(dependencyName, {
+        authorId: task.authorId,
+        authorIp: task.authorIp,
+        tips,
+      });
+      logs.push(`[${isoNow()}] Add dependency "${dependencyName}" sync task: ${dependencyTask.taskId}, db id: ${dependencyTask.id}`);
+    }
+    logs.push(`[${isoNow()}] log: ${logUrl}`);
+    logs.push(`[${isoNow()}] 🟢🟢🟢🟢🟢 ${url} 🟢🟢🟢🟢🟢`);
+    await this.finishTask(task, TaskState.Success, logs.join('\n'));
+  }
+
+  private async appendTaskLog(task: Task, appendLog: string) {
+    await this.nfsAdapter.appendBytes(task.logPath, Buffer.from(appendLog + '\n'));
+    task.updatedAt = new Date();
+    await this.taskRepository.saveTask(task);
+  }
+
+  private async finishTask(task: Task, taskState: TaskState, appendLog: string) {
+    // console.log(appendLog);
+    await this.nfsAdapter.appendBytes(task.logPath, Buffer.from(appendLog + '\n'));
+    task.state = taskState;
+    await this.taskRepository.saveTaskToHistory(task);
+  }
+}
--- a/app/core/service/UserService.ts
+++ b/app/core/service/UserService.ts
@@ -70,6 +70,27 @@ export class UserService extends AbstractService {
    return { user: userEntity, token };
  }

+  async savePublicUser(name: string, email: string) {
+    const storeName = name.startsWith('name:') ? name : `npm:${name}`;
+    let user = await this.userRepository.findUserByName(storeName);
+    if (!user) {
+      const passwordSalt = crypto.randomBytes(20).toString('hex');
+      const passwordIntegrity = integrity(passwordSalt);
+      user = UserEntity.create({
+        name: storeName,
+        email,
+        ip: '',
+        passwordSalt,
+        passwordIntegrity,
+        isPrivate: false,
+      });
+    } else {
+      user.email = email;
+    }
+    await this.userRepository.saveUser(user);
+    return user;
+  }
+
  async createToken(userId: string, options: CreateTokenOptions = {}) {
    // https://github.blog/2021-09-23-announcing-npms-new-access-token-format/
    // https://github.blog/2021-04-05-behind-githubs-new-authentication-token-formats/
--- a/app/port/UserRoleManager.ts
+++ b/app/port/UserRoleManager.ts
@@ -62,6 +62,7 @@ export class UserRoleManager {
    if (authorizedUserAndToken) {
      this.currentAuthorizedToken = authorizedUserAndToken.token;
      this.currentAuthorizedUser = authorizedUserAndToken.user;
+      ctx.userId = authorizedUserAndToken.user.userId;
    }
    return authorizedUserAndToken;
  }
@@ -101,6 +102,10 @@ export class UserRoleManager {
  }

  public async requiredPackageMaintainer(pkg: PackageEntity, user: UserEntity) {
+    // should be private package
+    if (!pkg.isPrivate) {
+      throw new ForbiddenError(`Can\'t modify npm public package "${pkg.fullname}"`);
+    }
    const maintainers = await this.packageRepository.listPackageMaintainers(pkg.packageId);
    const maintainer = maintainers.find(m => m.userId === user.userId);
    if (!maintainer) {
--- a/app/port/controller/PackageSyncController.ts
+++ b/app/port/controller/PackageSyncController.ts
@@ -0,0 +1,94 @@
+import {
+  HTTPController,
+  HTTPMethod,
+  HTTPMethodEnum,
+  HTTPParam,
+  HTTPBody,
+  Context,
+  EggContext,
+  Inject,
+} from '@eggjs/tegg';
+import { ForbiddenError, NotFoundError } from 'egg-errors';
+import { AbstractController } from './AbstractController';
+import { FULLNAME_REG_STRING, getScopeAndName } from '../../common/PackageUtil';
+import { PackageSyncerService } from '../../core/service/PackageSyncerService';
+import { TaskState } from '../../common/enum/Task';
+import { SyncPackageTaskRule, SyncPackageTaskType } from '../typebox';
+
+@HTTPController()
+export class PackageSyncController extends AbstractController {
+  @Inject()
+  private packageSyncerService: PackageSyncerService;
+
+  @HTTPMethod({
+    // PUT /-/package/:fullname/syncs
+    path: `/-/package/:fullname(${FULLNAME_REG_STRING})/syncs`,
+    method: HTTPMethodEnum.PUT,
+  })
+  async createSyncTask(@Context() ctx: EggContext, @HTTPParam() fullname: string, @HTTPBody() data: SyncPackageTaskType) {
+    const params = { fullname, tips: data.tips || '', skipDependencies: !!data.skipDependencies };
+    ctx.tValidate(SyncPackageTaskRule, params);
+    const [ scope, name ] = getScopeAndName(params.fullname);
+    const packageEntity = await this.packageRepository.findPackage(scope, name);
+    if (packageEntity?.isPrivate) {
+      throw new ForbiddenError(`Can\'t sync private package "${params.fullname}"`);
+    }
+    const authorized = await this.userRoleManager.getAuthorizedUserAndToken(ctx);
+    const task = await this.packageSyncerService.createTask(params.fullname, {
+      authorIp: ctx.ip,
+      authorId: authorized?.user.userId,
+      tips: params.tips,
+    });
+    ctx.status = 201;
+    return {
+      ok: true,
+      id: task.taskId,
+      type: task.type,
+      state: task.state,
+    };
+  }
+
+  // TODO: no-cache for CDN if task state is processing or timeout
+  @HTTPMethod({
+    // GET /-/package/:fullname/syncs/:syncId
+    path: `/-/package/:fullname(${FULLNAME_REG_STRING})/syncs/:taskId`,
+    method: HTTPMethodEnum.GET,
+  })
+  async showSyncTask(@Context() ctx: EggContext, @HTTPParam() fullname: string, @HTTPParam() taskId: string) {
+    const task = await this.packageSyncerService.findTask(taskId);
+    if (!task) throw new NotFoundError(`Package "${fullname}" sync task "${taskId}" not found`);
+    let logUrl: URL | undefined;
+    if (task.state !== TaskState.Waiting) {
+      logUrl = new URL(ctx.href);
+      logUrl.pathname = `${logUrl.pathname}/log`;
+    }
+    return {
+      ok: true,
+      id: task.taskId,
+      type: task.type,
+      state: task.state,
+      logUrl: logUrl?.toString(),
+    };
+  }
+
+  // TODO: no-cache for CDN if task state is processing or timeout
+  @HTTPMethod({
+    // GET /-/package/:fullname/syncs/:syncId/log
+    path: `/-/package/:fullname(${FULLNAME_REG_STRING})/syncs/:taskId/log`,
+    method: HTTPMethodEnum.GET,
+  })
+  async showSyncTaskLog(@Context() ctx: EggContext, @HTTPParam() fullname: string, @HTTPParam() taskId: string) {
+    const task = await this.packageSyncerService.findTask(taskId);
+    if (!task) throw new NotFoundError(`Package "${fullname}" sync task "${taskId}" not found`);
+    if (task.state === TaskState.Waiting) throw new NotFoundError(`Package "${fullname}" sync task "${taskId}" log not found`);
+
+    const logUrlOrStream = await this.packageSyncerService.findTaskLog(task);
+    if (!logUrlOrStream) throw new NotFoundError(`Package "${fullname}" sync task "${taskId}" log not found`);
+    if (typeof logUrlOrStream === 'string') {
+      ctx.redirect(logUrlOrStream);
+      return;
+    }
+    ctx.type = 'log';
+    return logUrlOrStream;
+  }
+}
--- a/app/port/typebox.ts
+++ b/app/port/typebox.ts
@@ -1,4 +1,4 @@
-import { Type } from '@sinclair/typebox';
+import { Type, Static } from '@sinclair/typebox';
 import semver from 'semver';

 export const Name = Type.String({
@@ -32,6 +32,16 @@ export const TagWithVersionRule = Type.Object({
  version: Version,
 });

+export const SyncPackageTaskRule = Type.Object({
+  fullname: Name,
+  tips: Type.String({
+    transform: [ 'trim' ],
+    maxLength: 1024,
+  }),
+  skipDependencies: Type.Boolean(),
+});
+export type SyncPackageTaskType = Static<typeof SyncPackageTaskRule>;
+
 // https://github.com/xiekw2010/egg-typebox-validate#%E5%A6%82%E4%BD%95%E5%86%99%E8%87%AA%E5%AE%9A%E4%B9%89%E6%A0%A1%E9%AA%8C%E8%A7%84%E5%88%99
 // add custom validate to ajv
 export function patchAjv(ajv: any) {
--- a/app/repository/PackageRepository.ts
+++ b/app/repository/PackageRepository.ts
@@ -69,12 +69,14 @@ export class PackageRepository extends AbstractRepository {
  }

  // Package Maintainers
-  async savePackageMaintainer(packageId: string, userId: string): Promise<void> {
+  // return true meaning create new record
+  async savePackageMaintainer(packageId: string, userId: string): Promise<undefined | true> {
    let model = await MaintainerModel.findOne({ packageId, userId });
    if (!model) {
      model = await MaintainerModel.create({ packageId, userId });
      this.logger.info('[PackageRepository:addPackageMaintainer:new] id: %s, packageId: %s, userId: %s',
        model.id, model.packageId, model.userId);
+      return true;
    }
  }

@@ -111,6 +113,7 @@ export class PackageRepository extends AbstractRepository {
  async createPackageVersion(pkgVersionEntity: PackageVersionEntity) {
    await PackageVersionModel.transaction(async function(transaction) {
      await Promise.all([
+        // FIXME: transaction is not the options
        ModelConvertor.convertEntityToModel(pkgVersionEntity, PackageVersionModel, transaction),
        ModelConvertor.convertEntityToModel(pkgVersionEntity.manifestDist, DistModel, transaction),
        ModelConvertor.convertEntityToModel(pkgVersionEntity.tarDist, DistModel, transaction),
--- a/app/repository/TaskRepository.ts
+++ b/app/repository/TaskRepository.ts
@@ -0,0 +1,79 @@
+import { AccessLevel, ContextProto } from '@eggjs/tegg';
+import { ModelConvertor } from './util/ModelConvertor';
+import { Task as TaskModel } from './model/Task';
+import { HistoryTask as HistoryTaskModel } from './model/HistoryTask';
+import { Task as TaskEntity } from '../core/entity/Task';
+import { AbstractRepository } from './AbstractRepository';
+import { TaskType, TaskState } from '../../app/common/enum/Task';
+
+@ContextProto({
+  accessLevel: AccessLevel.PUBLIC,
+})
+export class TaskRepository extends AbstractRepository {
+  async saveTask(task: TaskEntity): Promise<void> {
+    if (task.id) {
+      const model = await TaskModel.findOne({ id: task.id });
+      if (!model) return;
+      await ModelConvertor.saveEntityToModel(task, model);
+    } else {
+      await ModelConvertor.convertEntityToModel(task, TaskModel);
+    }
+  }
+
+  async saveTaskToHistory(task: TaskEntity): Promise<void> {
+    const model = await TaskModel.findOne({ id: task.id });
+    if (!model) return;
+    const history = await HistoryTaskModel.findOne({ taskId: task.taskId });
+    if (history) {
+      await ModelConvertor.saveEntityToModel(task, history);
+    } else {
+      await ModelConvertor.convertEntityToModel(task, HistoryTaskModel);
+    }
+    await model.remove();
+  }
+
+  async findTask(taskId: string) {
+    const task = await TaskModel.findOne({ taskId });
+    if (task) {
+      return ModelConvertor.convertModelToEntity(task, TaskEntity);
+    }
+    // try to read from history
+    const history = await HistoryTaskModel.findOne({ taskId });
+    if (history) {
+      return ModelConvertor.convertModelToEntity(history, TaskEntity);
+    }
+  }
+
+  async findTaskByTargetName(targetName: string, type: TaskType, state: TaskState) {
+    return await TaskModel.findOne({ targetName, type, state });
+  }
+
+  async executeWaitingTask(taskType: TaskType) {
+    // https://zhuanlan.zhihu.com/p/20293493?refer=alsotang
+    // Task list impl from MySQL
+    const GET_WAITING_TASK_SQL = `UPDATE tasks SET gmt_modified=now(3), state=?, attempts=attempts+1, id=LAST_INSERT_ID(id)
+WHERE type=? AND state=? ORDER BY gmt_modified ASC LIMIT 1`;
+    let result = await TaskModel.driver.query(GET_WAITING_TASK_SQL,
+      [ TaskState.Processing, taskType, TaskState.Waiting ]);
+    // if has task, affectedRows > 0 and insertId > 0
+    if (result.affectedRows && result.affectedRows > 0 && result.insertId && result.insertId > 0) {
+      this.logger.info('[TaskRepository:executeWaitingTask:waiting] type: %s, result: %j', taskType, result);
+      return await TaskModel.findOne({ id: result.insertId });
+    }
+
+    // try to find timeout task, 5 mins
+    const timeoutDate = new Date();
+    timeoutDate.setTime(timeoutDate.getTime() - 60000 * 5);
+    const GET_TIMEOUT_TASK_SQL = `UPDATE tasks SET gmt_modified=now(3), state=?, attempts=attempts+1, id=LAST_INSERT_ID(id)
+WHERE type=? AND state=? AND gmt_modified<? ORDER BY gmt_modified ASC LIMIT 1`;
+    result = await TaskModel.driver.query(GET_TIMEOUT_TASK_SQL,
+      [ TaskState.Processing, taskType, TaskState.Processing, timeoutDate ]);
+    // if has task, affectedRows > 0 and insertId > 0
+    if (result.affectedRows && result.affectedRows > 0 && result.insertId && result.insertId > 0) {
+      this.logger.info('[TaskRepository:executeWaitingTask:timeout] type: %s, result: %j, timeout: %j',
+        taskType, result, timeoutDate);
+      return await TaskModel.findOne({ id: result.insertId });
+    }
+    return null;
+  }
+}
--- a/app/repository/model/HistoryTask.ts
+++ b/app/repository/model/HistoryTask.ts
@@ -0,0 +1,47 @@
+import { Attribute, Model } from '@eggjs/tegg-orm-decorator';
+import { DataTypes, Bone } from 'leoric';
+import { TaskState, TaskType } from '../../common/enum/Task';
+
+@Model()
+export class HistoryTask extends Bone {
+  @Attribute(DataTypes.BIGINT, {
+    primary: true,
+    autoIncrement: true,
+  })
+  id: bigint;
+
+  @Attribute(DataTypes.DATE, { name: 'gmt_create' })
+  createdAt: Date;
+
+  @Attribute(DataTypes.DATE, { name: 'gmt_modified' })
+  updatedAt: Date;
+
+  @Attribute(DataTypes.STRING(24), {
+    unique: true,
+  })
+  taskId: string;
+
+  @Attribute(DataTypes.STRING(20))
+  type: TaskType;
+
+  @Attribute(DataTypes.STRING(20))
+  state: TaskState;
+
+  @Attribute(DataTypes.STRING(214))
+  targetName: string;
+
+  @Attribute(DataTypes.STRING(24))
+  authorId: string;
+
+  @Attribute(DataTypes.STRING(100))
+  authorIp: string;
+
+  @Attribute(DataTypes.JSONB)
+  data: object;
+
+  @Attribute(DataTypes.STRING(512))
+  logPath: string;
+
+  @Attribute(DataTypes.INTEGER)
+  attempts: number;
+}
--- a/app/repository/model/Package.ts
+++ b/app/repository/model/Package.ts
@@ -22,7 +22,7 @@ export class Package extends Bone {
  packageId: string;

  @Attribute(DataTypes.STRING(214))
-  scope?: string;
+  scope: string;

  // https://github.com/npm/npm/issues/8077#issuecomment-97258418
  // https://docs.npmjs.com/cli/v7/configuring-npm/package-json#name
--- a/app/repository/model/Task.ts
+++ b/app/repository/model/Task.ts
@@ -0,0 +1,47 @@
+import { Attribute, Model } from '@eggjs/tegg-orm-decorator';
+import { DataTypes, Bone } from 'leoric';
+import { TaskState, TaskType } from '../../common/enum/Task';
+
+@Model()
+export class Task extends Bone {
+  @Attribute(DataTypes.BIGINT, {
+    primary: true,
+    autoIncrement: true,
+  })
+  id: bigint;
+
+  @Attribute(DataTypes.DATE, { name: 'gmt_create' })
+  createdAt: Date;
+
+  @Attribute(DataTypes.DATE, { name: 'gmt_modified' })
+  updatedAt: Date;
+
+  @Attribute(DataTypes.STRING(24), {
+    unique: true,
+  })
+  taskId: string;
+
+  @Attribute(DataTypes.STRING(20))
+  type: TaskType;
+
+  @Attribute(DataTypes.STRING(20))
+  state: TaskState;
+
+  @Attribute(DataTypes.STRING(214))
+  targetName: string;
+
+  @Attribute(DataTypes.STRING(24))
+  authorId: string;
+
+  @Attribute(DataTypes.STRING(100))
+  authorIp: string;
+
+  @Attribute(DataTypes.JSONB)
+  data: object;
+
+  @Attribute(DataTypes.STRING(512))
+  logPath: string;
+
+  @Attribute(DataTypes.INTEGER)
+  attempts: number;
+}
--- a/app/repository/util/ModelConvertor.ts
+++ b/app/repository/util/ModelConvertor.ts
@@ -42,7 +42,7 @@ export class ModelConvertor {
    for (const attributeMeta of metadata.attributes) {
      const modelPropertyName = attributeMeta.propertyName;
      const entityPropertyName = ModelConvertorUtil.getEntityPropertyName(ModelClazz, modelPropertyName);
-      if (entityPropertyName === 'UPDATED_AT' || entityPropertyName === 'CREATED_AT') continue;
+      if (entityPropertyName === 'CREATED_AT') continue;
      const attributeValue = _.get(entity, entityPropertyName);
      model[modelPropertyName] = attributeValue;
    }
--- a/app/schedule/SyncPackageWorker.ts
+++ b/app/schedule/SyncPackageWorker.ts
@@ -0,0 +1,34 @@
+import { PackageSyncerService } from '../core/service/PackageSyncerService';
+import { Subscription } from 'egg';
+
+const cnpmcoreCore = 'cnpmcoreCore';
+
+let executing = false;
+export default class SyncPackageWorker extends Subscription {
+  static get schedule() {
+    return {
+      interval: 1000,
+      type: 'all',
+    };
+  }
+
+  async subscribe() {
+    if (executing) return;
+    const { ctx } = this;
+    await ctx.beginModuleScope(async () => {
+      const packageSyncerService: PackageSyncerService = ctx.module[cnpmcoreCore].packageSyncerService;
+      executing = true;
+      try {
+        const task = await packageSyncerService.findExecuteTask();
+        if (!task) {
+          return;
+        }
+        ctx.logger.info('[SyncPackageWorker:subscribe:executeTask] taskId: %s, params: %j',
+          task.taskId, task.data);
+        await packageSyncerService.executeTask(task);
+      } finally {
+        executing = false;
+      }
+    });
+  }
+}
--- a/config/config.default.ts
+++ b/config/config.default.ts
@@ -9,6 +9,11 @@ export default (/* appInfo: EggAppConfig */) => {
  config.cnpmcore = {
    name: 'cnpm',
    sourceRegistry: 'https://registry.npmjs.com',
+    // upstream registry is base on cnpm/cnpmjs.org or not
+    // if your upstream is official npm registry, please turn it off
+    sourceRegistryIsCNpm: false,
+    // 3 mins
+    sourceRegistrySyncTimeout: 180000,
    registry: 'http://localhost:7001',
    // https://docs.npmjs.com/cli/v6/using-npm/config#always-auth npm <= 6
    // if `alwaysAuth=true`, all api request required access token
--- a/package.json
+++ b/package.json
@@ -52,7 +52,7 @@
    "egg-opentracing": "^1.1.1",
    "egg-typebox-validate": "^2.0.0",
    "fresh": "^0.5.2",
-    "fs-cnpm": "^2.3.0",
+    "fs-cnpm": "^2.4.0",
    "leoric": "^1.15.0",
    "mysql": "^2.18.1",
    "mysql2": "^2.3.0",
--- a/sql/init.sql
+++ b/sql/init.sql
@@ -190,3 +190,40 @@ CREATE TABLE IF NOT EXISTS `maintainers` (
 KEY `idx_package_id` (`package_id`),
 KEY `idx_user_id` (`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='package maintainers';
+
+CREATE TABLE IF NOT EXISTS `tasks` (
+  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'primary key',
+  `gmt_create` datetime(3) NOT NULL COMMENT 'create time',
+  `gmt_modified` datetime(3) NOT NULL COMMENT 'modified time',
+  `task_id` varchar(24) NOT NULL COMMENT 'task id',
+  `type` varchar(20) NOT NULL COMMENT 'task type',
+  `state` varchar(20) NOT NULL COMMENT 'task state',
+  `target_name` varchar(214) NOT NULL COMMENT 'target name, like package name / user name',
+  `author_id` varchar(24) NOT NULL COMMENT 'create task user id',
+  `author_ip` varchar(100) NOT NULL COMMENT 'create task user request ip',
+  `data` json NULL COMMENT 'task params',
+  `log_path` varchar(512) NOT NULL COMMENT 'access path',
+  `attempts` int unsigned DEFAULT 0 COMMENT 'task execute attempts times',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_task_id` (`task_id`),
+  KEY `idx_type_state_target_name` (`target_name`, `type`, `state`),
+  KEY `idx_type_state_gmt_modified` (`type`, `state`, `gmt_modified`),
+  KEY `idx_gmt_modified` (`gmt_modified`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='task info';
+
+CREATE TABLE IF NOT EXISTS `history_tasks` (
+  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'primary key',
+  `gmt_create` datetime(3) NOT NULL COMMENT 'create time',
+  `gmt_modified` datetime(3) NOT NULL COMMENT 'modified time',
+  `task_id` varchar(24) NOT NULL COMMENT 'task id',
+  `type` varchar(20) NOT NULL COMMENT 'task type',
+  `state` varchar(20) NOT NULL COMMENT 'task state',
+  `target_name` varchar(214) NOT NULL COMMENT 'target name, like package name / user name',
+  `author_id` varchar(24) NOT NULL COMMENT 'create task user id',
+  `author_ip` varchar(100) NOT NULL COMMENT 'create task user request ip',
+  `data` json NULL COMMENT 'task params',
+  `log_path` varchar(512) NOT NULL COMMENT 'access path',
+  `attempts` int unsigned DEFAULT 0 COMMENT 'task execute attempts times',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_task_id` (`task_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='history task info';
--- a/test/TestUtil.ts
+++ b/test/TestUtil.ts
@@ -2,6 +2,18 @@ import * as fs from 'fs/promises';
 import mysql from 'mysql';
 import path from 'path';
 import crypto from 'crypto';
+import { getScopeAndName } from '../app/common/PackageUtil';
+
+type PackageOptions = {
+  name?: string;
+  version?: string;
+  versionObject?: object;
+  attachment?: object;
+  dist?: object;
+  readme?: string | null;
+  distTags?: object | null;
+  isPrivate?: boolean;
+};

 export class TestUtil {
  private static connection;
@@ -89,19 +101,20 @@ export class TestUtil {
    await Promise.all(tables.map(table => this.query(`TRUNCATE TABLE ${database}.${table};`)));
  }

+  static get app() {
+    if (!this._app) {
+      /* eslint @typescript-eslint/no-var-requires: "off" */
+      const bootstrap = require('egg-mock/bootstrap');
+      this._app = bootstrap.app;
+    }
+    return this._app;
+  }
+
  static getFixtures(name?: string): string {
    return path.join(__dirname, 'fixtures', name ?? '');
  }

-  static async getFullPackage(options?: {
-    name?: string;
-    version?: string;
-    versionObject?: object;
-    attachment?: object;
-    dist?: object;
-    readme?: string | null;
-    distTags?: object | null;
-  }): Promise<any> {
+  static async getFullPackage(options?: PackageOptions): Promise<any> {
    const fullJSONFile = this.getFixtures('exampleFullPackage.json');
    const pkg = JSON.parse((await fs.readFile(fullJSONFile)).toString());
    if (options) {
@@ -147,13 +160,21 @@ export class TestUtil {
    return pkg;
  }

-  static get app() {
-    if (!this._app) {
-      /* eslint @typescript-eslint/no-var-requires: "off" */
-      const bootstrap = require('egg-mock/bootstrap');
-      this._app = bootstrap.app;
+  static async createPackage(options?: PackageOptions) {
+    const pkg = await this.getFullPackage(options);
+    const user = await this.createUser();
+    await this.app.httpRequest()
+      .put(`/${pkg.name}`)
+      .set('authorization', user.authorization)
+      .set('user-agent', user.ua)
+      .send(pkg)
+      .expect(201);
+    if (options?.isPrivate === false) {
+      const [ scope, name ] = getScopeAndName(pkg.name);
+      const { Package: PackageModel } = require('../app/repository/model/Package');
+      await PackageModel.update({ scope, name }, { isPrivate: false });
    }
-    return this._app;
+    return { user, pkg };
  }

  static async createUser(user?: {
--- a/test/core/service/PackageManagerService/publish.test.ts
+++ b/test/core/service/PackageManagerService/publish.test.ts
@@ -1,12 +1,12 @@
 import assert from 'assert';
 import { app, mock } from 'egg-mock/bootstrap';
 import { Context } from 'egg';
-import { PackageManagerService } from '../../../app/core/service/PackageManagerService';
-import { UserService } from '../../../app/core/service/UserService';
-import { PackageRepository } from '../../../app/repository/PackageRepository';
-import { TestUtil } from '../../TestUtil';
+import { PackageManagerService } from '../../../../app/core/service/PackageManagerService';
+import { UserService } from '../../../../app/core/service/UserService';
+import { PackageRepository } from '../../../../app/repository/PackageRepository';
+import { TestUtil } from '../../../TestUtil';

-describe('test/core/service/PackageManagerService.test.ts', () => {
+describe('test/core/service/PackageManagerService/publish.test.ts', () => {
  let ctx: Context;
  let packageManagerService: PackageManagerService;
  let userService: UserService;
--- a/test/core/service/PackageSyncerService/executeTask.test.ts
+++ b/test/core/service/PackageSyncerService/executeTask.test.ts
@@ -0,0 +1,434 @@
+import assert from 'assert';
+import { Readable } from 'stream';
+import { app, mock } from 'egg-mock/bootstrap';
+import { Context } from 'egg';
+import { PackageSyncerService } from 'app/core/service/PackageSyncerService';
+import { PackageManagerService } from 'app/core/service/PackageManagerService';
+import { Package as PackageModel } from 'app/repository/model/Package';
+import { Task as TaskModel } from 'app/repository/model/Task';
+import { HistoryTask as HistoryTaskModel } from 'app/repository/model/HistoryTask';
+import { TestUtil } from 'test/TestUtil';
+import { NPMRegistry } from 'app/common/adapter/NPMRegistry';
+
+describe('test/core/service/PackageSyncerService/executeTask.test.ts', () => {
+  let ctx: Context;
+  let packageSyncerService: PackageSyncerService;
+  let packageManagerService: PackageManagerService;
+
+  beforeEach(async () => {
+    ctx = await app.mockModuleContext();
+    packageSyncerService = await ctx.getEggObject(PackageSyncerService);
+    packageManagerService = await ctx.getEggObject(PackageManagerService);
+  });
+
+  afterEach(async () => {
+    app.destroyModuleContext(ctx);
+  });
+
+  describe('executeTask()', () => {
+    it('should execute foo task', async () => {
+      await packageSyncerService.createTask('foo', { skipDependencies: true });
+      let task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      assert(!await TaskModel.findOne({ taskId: task.taskId }));
+      assert(await HistoryTaskModel.findOne({ taskId: task.taskId }));
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      const model = await PackageModel.findOne({ scope: '', name: 'foo' });
+      assert.equal(model!.isPrivate, false);
+      assert(log.includes(', skipDependencies: true'));
+
+      // sync again
+      await packageSyncerService.createTask('foo');
+      task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      // stream = await packageSyncerService.findTaskLog(task) as Readable;
+      // assert(stream);
+      // for await (const chunk of stream) {
+      //   process.stdout.write(chunk);
+      // }
+
+      const manifests = await packageManagerService.listPackageFullManifests('', 'foo', undefined);
+      // console.log(JSON.stringify(manifests, null, 2));
+      // should have 2 maintainers
+      assert.equal(manifests.data.maintainers.length, 2);
+      const abbreviatedManifests = await packageManagerService.listPackageAbbreviatedManifests('', 'foo', undefined);
+      // console.log(JSON.stringify(abbreviatedManifests, null, 2));
+      assert.equal(abbreviatedManifests.data.name, manifests.data.name);
+    });
+
+    it('should execute @node-rs/xxhash task, contains optionalDependencies', async () => {
+      await packageSyncerService.createTask('@node-rs/xxhash');
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      assert(!await TaskModel.findOne({ taskId: task.taskId }));
+      assert(await HistoryTaskModel.findOne({ taskId: task.taskId }));
+      // const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      // assert(stream);
+      // for await (const chunk of stream) {
+      //   process.stdout.write(chunk);
+      // }
+
+      const manifests = await packageManagerService.listPackageFullManifests('@node-rs', 'xxhash', undefined);
+      // console.log(JSON.stringify(manifests, null, 2));
+      // assert.equal(manifests.data.maintainers.length, 2);
+      const abbreviatedManifests = await packageManagerService.listPackageAbbreviatedManifests('@node-rs', 'xxhash', undefined);
+      // console.log(JSON.stringify(abbreviatedManifests, null, 2));
+      assert.equal(abbreviatedManifests.data.name, manifests.data.name);
+      assert(abbreviatedManifests.data.versions['1.0.1'].optionalDependencies);
+    });
+
+    it('should sync cnpmcore-test-sync-deprecated', async () => {
+      const name = 'cnpmcore-test-sync-deprecated';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      // const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      // assert(stream);
+      // for await (const chunk of stream) {
+      //   process.stdout.write(chunk);
+      // }
+      const manifests = await packageManagerService.listPackageFullManifests('', name, undefined);
+      assert.equal(manifests.data.versions['0.0.0'].deprecated, 'only test for cnpmcore');
+      assert.equal(manifests.data.versions['0.0.0']._hasShrinkwrap, false);
+      const abbreviatedManifests = await packageManagerService.listPackageAbbreviatedManifests('', name, undefined);
+      assert.equal(abbreviatedManifests.data.versions['0.0.0'].deprecated, 'only test for cnpmcore');
+      assert.equal(abbreviatedManifests.data.versions['0.0.0']._hasShrinkwrap, false);
+    });
+
+    it('should sync cnpmcore-test-sync-dependencies => cnpmcore-test-sync-deprecated', async () => {
+      let name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      let task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      let stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      let chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      let log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('] Add dependency "cnpmcore-test-sync-deprecated" sync task: '));
+
+      // will sync cnpmcore-test-sync-deprecated
+      name = 'cnpmcore-test-sync-deprecated';
+      await packageSyncerService.createTask(name);
+      task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      chunks = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('Sync cause by "cnpmcore-test-sync-dependencies" dependencies, parent task: '));
+    });
+
+    it('should sync sourceRegistryIsCNpm = true', async () => {
+      mock(app.config.cnpmcore, 'sourceRegistry', 'https://r.cnpmjs.org');
+      mock(app.config.cnpmcore, 'sourceRegistryIsCNpm', true);
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('] Add dependency "cnpmcore-test-sync-deprecated" sync task: '));
+      assert(log.includes('][UP] 🚧🚧🚧🚧🚧 Waiting sync "cnpmcore-test-sync-dependencies" task on https://r.cnpmjs.org 🚧'));
+      assert(log.includes('][UP] 🟢🟢🟢🟢🟢 https://r.cnpmjs.org/cnpmcore-test-sync-dependencies 🟢'));
+    });
+
+    it('should sync sourceRegistryIsCNpm = true and mock createSyncTask error', async () => {
+      mock(app.config.cnpmcore, 'sourceRegistry', 'https://r.cnpmjs.org');
+      mock(app.config.cnpmcore, 'sourceRegistryIsCNpm', true);
+      mock.error(NPMRegistry.prototype, 'createSyncTask');
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('🚮 give up 🚮 ❌❌❌❌❌'));
+      assert(log.includes(`][UP] ❌ Sync ${name} fail, create sync task error:`));
+    });
+
+    it('should sync sourceRegistryIsCNpm = true and mock createSyncTask return missing logId', async () => {
+      mock(app.config.cnpmcore, 'sourceRegistry', 'https://r.cnpmjs.org');
+      mock(app.config.cnpmcore, 'sourceRegistryIsCNpm', true);
+      mock.data(NPMRegistry.prototype, 'createSyncTask', { data: { ok: true }, res: {} });
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('🚮 give up 🚮 ❌❌❌❌❌'));
+      assert(log.includes(`][UP] ❌ Sync ${name} fail, missing logId`));
+    });
+
+    it('should sync sourceRegistryIsCNpm = true and mock getSyncTask syncDone = false', async () => {
+      mock(app.config.cnpmcore, 'sourceRegistry', 'https://r.cnpmjs.org');
+      mock(app.config.cnpmcore, 'sourceRegistryIsCNpm', true);
+      mock(app.config.cnpmcore, 'sourceRegistrySyncTimeout', 10000);
+      let first = true;
+      mock(NPMRegistry.prototype, 'getSyncTask', async () => {
+        if (!first) {
+          throw new Error('mock error');
+        }
+        first = false;
+        return { data: { syncDone: false }, res: {}, status: 200 };
+      });
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('🚮 give up 🚮 ❌❌❌❌❌'));
+      assert(log.includes('][UP] 🚧 HTTP [200]'));
+      assert.match(log, /\]\[UP\] 🚧 HTTP \[unknow\] \[\d+ms\] error: /);
+    });
+
+    it('should sync sourceRegistryIsCNpm = true and mock sync upstream timeout', async () => {
+      mock(app.config.cnpmcore, 'sourceRegistry', 'https://r.cnpmjs.org');
+      mock(app.config.cnpmcore, 'sourceRegistryIsCNpm', true);
+      mock(app.config.cnpmcore, 'sourceRegistrySyncTimeout', -1);
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('🚮 give up 🚮 ❌❌❌❌❌'));
+      assert(log.includes(`][UP] ❌ Sync ${name} fail, timeout`));
+    });
+
+    it('should mock getFullManifests error', async () => {
+      mock.error(NPMRegistry.prototype, 'getFullManifests');
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes(`❌❌❌❌❌ ${name} ❌❌❌❌❌`));
+      assert(log.includes(`❌ Synced ${name} fail, request manifests error`));
+    });
+
+    it('should mock getFullManifests Invalid maintainers error', async () => {
+      mock.data(NPMRegistry.prototype, 'getFullManifests', {
+        data: {
+          maintainers: [{ name: 'foo' }],
+        },
+        res: {},
+        headers: {},
+      });
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes(`❌❌❌❌❌ ${name} ❌❌❌❌❌`));
+      assert(log.includes('❌ Invalid maintainers: '));
+    });
+
+    it('should mock getFullManifests missing tarball error and downloadTarball error', async () => {
+      mock.error(NPMRegistry.prototype, 'downloadTarball');
+      mock.data(NPMRegistry.prototype, 'getFullManifests', {
+        data: {
+          maintainers: [{ name: 'fengmk2', email: 'fengmk2@gmai.com' }],
+          versions: {
+            '1.0.0': {
+              version: '1.0.0',
+              dist: { tarball: '' },
+            },
+            '2.0.0': {
+              version: '2.0.0',
+              dist: { tarball: 'https://foo.com/a.tgz' },
+            },
+          },
+        },
+        res: {},
+        headers: {},
+      });
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes(`❌❌❌❌❌ ${name} ❌❌❌❌❌`));
+      assert(log.includes('❌ Synced version 1.0.0 fail, missing tarball, dist: '));
+      assert(log.includes('❌ All versions sync fail, package not exists'));
+      assert(log.includes('❌ Synced version 2.0.0 fail, download tarball error'));
+    });
+
+    it('should mock downloadTarball status !== 200', async () => {
+      mock.data(NPMRegistry.prototype, 'downloadTarball', {
+        status: 404,
+        res: {},
+        headers: {},
+        localFile: __filename + '__not_exists',
+      });
+      mock.data(NPMRegistry.prototype, 'getFullManifests', {
+        data: {
+          maintainers: [{ name: 'fengmk2', email: 'fengmk2@gmai.com' }],
+          versions: {
+            '2.0.0': {
+              version: '2.0.0',
+              dist: { tarball: 'https://foo.com/a.tgz' },
+            },
+          },
+        },
+        res: {},
+        headers: {},
+      });
+      const name = 'cnpmcore-test-sync-dependencies';
+      await packageSyncerService.createTask(name);
+      const task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.targetName, name);
+      await packageSyncerService.executeTask(task);
+      const stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      const chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      const log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes(`❌❌❌❌❌ ${name} ❌❌❌❌❌`));
+      assert(log.includes('❌ All versions sync fail, package not exists'));
+      assert(log.includes('❌ Synced version 2.0.0 fail, download tarball status error'));
+    });
+
+    it('should sync mk2test-module-cnpmsync with different metas', async () => {
+      const name = 'mk2test-module-cnpmsync';
+      mock(app.config.cnpmcore, 'allowPublishNonScopePackage', true);
+      await TestUtil.createPackage({ name, version: '2.0.0', isPrivate: false });
+      await packageSyncerService.createTask(name, { tips: 'sync test tips here' });
+      let task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      let stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      let chunks: any[] = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      let log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(log.includes('🟢 Synced version 2.0.0 success, different meta: {"peerDependenciesMeta":{"bufferutil":{"optional":true},"utf-8-validate":{"optional":true}},"os":["linux"],"cpu":["x64"]}'));
+      assert(log.includes('Z] 👉👉👉👉👉 Tips: sync test tips here 👈👈👈👈👈'));
+      assert(log.includes(', skipDependencies: false'));
+      const manifests = await packageManagerService.listPackageFullManifests('', name, undefined);
+      assert.equal(manifests.data.versions['2.0.0'].peerDependenciesMeta.bufferutil.optional, true);
+      assert.equal(manifests.data.versions['2.0.0'].os[0], 'linux');
+      assert.equal(manifests.data.versions['2.0.0'].cpu[0], 'x64');
+      // publishTime
+      assert.equal(manifests.data.time['1.0.0'], '2021-09-27T08:10:48.747Z');
+      const abbreviatedManifests = await packageManagerService.listPackageAbbreviatedManifests('', name, undefined);
+      // console.log(JSON.stringify(abbreviatedManifests.data, null, 2));
+      assert.equal(abbreviatedManifests.data.versions['2.0.0'].peerDependenciesMeta.bufferutil.optional, true);
+      assert.equal(abbreviatedManifests.data.versions['2.0.0'].os[0], 'linux');
+      assert.equal(abbreviatedManifests.data.versions['2.0.0'].cpu[0], 'x64');
+
+      // again should skip sync different metas
+      await packageSyncerService.createTask(name);
+      task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      await packageSyncerService.executeTask(task);
+      stream = await packageSyncerService.findTaskLog(task) as Readable;
+      assert(stream);
+      chunks = [];
+      for await (const chunk of stream) {
+        chunks.push(chunk);
+      }
+      log = Buffer.concat(chunks).toString();
+      // console.log(log);
+      assert(!log.includes('🟢 Synced version 2.0.0 success, different meta:'));
+    });
+  });
+});
--- a/test/core/service/PackageSyncerService/findExecuteTask.test.ts
+++ b/test/core/service/PackageSyncerService/findExecuteTask.test.ts
@@ -0,0 +1,81 @@
+import assert from 'assert';
+import { app } from 'egg-mock/bootstrap';
+import { Context } from 'egg';
+import { PackageSyncerService } from 'app/core/service/PackageSyncerService';
+import { Task as TaskModel } from 'app/repository/model/Task';
+import { HistoryTask as HistoryTaskModel } from 'app/repository/model/HistoryTask';
+
+describe('test/core/service/PackageSyncerService/findExecuteTask.test.ts', () => {
+  let ctx: Context;
+  let packageSyncerService: PackageSyncerService;
+
+  beforeEach(async () => {
+    ctx = await app.mockModuleContext();
+    packageSyncerService = await ctx.getEggObject(PackageSyncerService);
+  });
+
+  afterEach(async () => {
+    app.destroyModuleContext(ctx);
+  });
+
+  describe('findExecuteTask()', () => {
+    it('should get a task to execute', async () => {
+      let task = await packageSyncerService.findExecuteTask();
+      assert(!task);
+
+      await packageSyncerService.createTask('foo');
+      await packageSyncerService.createTask('foo');
+      await packageSyncerService.createTask('foo');
+      const task1 = await packageSyncerService.findExecuteTask();
+      assert(task1);
+      assert.equal(task1.state, 'processing');
+      assert(task1.updatedAt > task1.createdAt);
+      assert.equal(task1.attempts, 1);
+      // console.log(task1, task1.updatedAt.getTime() - task1.createdAt.getTime());
+      const task2 = await packageSyncerService.findExecuteTask();
+      assert(task2);
+      assert.equal(task2.state, 'processing');
+      assert(task2.updatedAt > task2.createdAt);
+      assert.equal(task1.attempts, 1);
+      // console.log(task2, task2.updatedAt.getTime() - task2.createdAt.getTime());
+      const task3 = await packageSyncerService.findExecuteTask();
+      assert(task3);
+      assert.equal(task3.state, 'processing');
+      assert(task3.updatedAt > task3.createdAt);
+      assert.equal(task1.attempts, 1);
+      // console.log(task3, task3.updatedAt.getTime() - task3.createdAt.getTime());
+      assert(task3.id > task2.id);
+      assert(task2.id > task1.id);
+
+      // again will empty
+      task = await packageSyncerService.findExecuteTask();
+      assert(!task);
+
+      // mock timeout
+      await TaskModel.update({ id: task3.id }, { updatedAt: new Date(task3.updatedAt.getTime() - 60000 * 5 - 1) });
+      task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.id, task3.id);
+      assert(task.updatedAt > task3.updatedAt);
+      assert.equal(task.attempts, 2);
+
+      // again will empty
+      task = await packageSyncerService.findExecuteTask();
+      assert(!task);
+
+      // attempts > 3 will be set to timeout task and save to history
+      await TaskModel.update({ id: task3.id }, { updatedAt: new Date(task3.updatedAt.getTime() - 60000 * 5 - 1) });
+      task = await packageSyncerService.findExecuteTask();
+      assert(task);
+      assert.equal(task.attempts, 3);
+      await TaskModel.update({ id: task3.id }, { updatedAt: new Date(task3.updatedAt.getTime() - 60000 * 5 - 1) });
+      task = await packageSyncerService.findExecuteTask();
+      assert(!task);
+      const history = await HistoryTaskModel.findOne({ taskId: task3.taskId });
+      assert(history);
+      assert.equal(history.state, 'timeout');
+      assert.equal(history.attempts, 3);
+      assert(!await TaskModel.findOne({ id: task3.id }));
+    });
+  });
+});
--- a/test/port/controller/DownloadController/showDownloads.test.ts
+++ b/test/port/controller/DownloadController/showDownloads.test.ts
@@ -58,7 +58,7 @@ describe('test/port/controller/DownloadController/showDownloads.test.ts', () =>
        .expect(200)
        .expect('content-type', 'application/json; charset=utf-8');
      const data = res.body;
-      console.log(data);
+      // console.log(data);
      assert.equal(data.start, start);
      assert.equal(data.end, end);
      assert.equal(data.package, '@cnpm/koa');
--- a/test/port/controller/HomeController/ping.test.ts
+++ b/test/port/controller/HomeController/ping.test.ts
@@ -18,7 +18,7 @@ describe('test/port/controller/PackageController/ping.test.ts', () => {
        .get('/-/ping')
        .expect(200);
      assert.equal(res.body.pong, true);
-      console.log(res.body, res.headers['x-readtime']);
+      // console.log(res.body, res.headers['x-readtime']);
    });
  });
 });
--- a/test/port/controller/PackageController/saveVersion.test.ts
+++ b/test/port/controller/PackageController/saveVersion.test.ts
@@ -19,6 +19,18 @@ describe('test/port/controller/PackageController/saveVersion.test.ts', () => {
  });

  describe('[PUT /:fullname] saveVersion()', () => {
+    it('should 403 when package is public', async () => {
+      const { pkg, user } = await TestUtil.createPackage({ isPrivate: false, version: '1.0.0' });
+      const pkg2 = await TestUtil.getFullPackage({ name: pkg.name, version: '2.0.0' });
+      const res = await app.httpRequest()
+        .put(`/${pkg2.name}`)
+        .set('authorization', user.authorization)
+        .set('user-agent', user.ua)
+        .send(pkg2)
+        .expect(403);
+      assert.equal(res.body.error, `[FORBIDDEN] Can\'t modify npm public package "${pkg2.name}"`);
+    });
+
    it('should add new version success on scoped package', async () => {
      const name = '@cnpm/publish-package-test';
      const pkg = await TestUtil.getFullPackage({ name, version: '0.0.0' });
--- a/test/port/controller/PackageSyncController/createSyncTask.test.ts
+++ b/test/port/controller/PackageSyncController/createSyncTask.test.ts
@@ -0,0 +1,73 @@
+import { strict as assert } from 'assert';
+import { Context } from 'egg';
+import { app, mock } from 'egg-mock/bootstrap';
+import { TestUtil } from 'test/TestUtil';
+
+describe('test/port/controller/PackageSyncController/createSyncTask.test.ts', () => {
+  let publisher;
+  let ctx: Context;
+  beforeEach(async () => {
+    publisher = await TestUtil.createUser();
+    ctx = await app.mockModuleContext();
+  });
+
+  afterEach(() => {
+    app.destroyModuleContext(ctx);
+  });
+
+  describe('[PUT /-/package/:fullname/syncs] createSyncTask()', () => {
+    it('should 401 if user not login when alwaysAuth = true', async () => {
+      mock(app.config.cnpmcore, 'alwaysAuth', true);
+      const res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .expect(401);
+      assert.equal(res.body.error, '[UNAUTHORIZED] Login first');
+    });
+
+    it('should 403 if when sync private package', async () => {
+      const pkg = await TestUtil.getFullPackage({ name: '@cnpm/koa', version: '1.0.0' });
+      await app.httpRequest()
+        .put(`/${pkg.name}`)
+        .set('authorization', publisher.authorization)
+        .set('user-agent', publisher.ua)
+        .send(pkg)
+        .expect(201);
+      const res = await app.httpRequest()
+        .put(`/-/package/${pkg.name}/syncs`)
+        .expect(403);
+      assert.equal(res.body.error, '[FORBIDDEN] Can\'t sync private package "@cnpm/koa"');
+    });
+
+    it('should 201 if user login when alwaysAuth = true', async () => {
+      mock(app.config.cnpmcore, 'alwaysAuth', true);
+      const res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .set('authorization', publisher.authorization)
+        .expect(201);
+      assert.equal(res.body.ok, true);
+      assert.equal(res.body.type, 'sync_package');
+      assert.equal(res.body.state, 'waiting');
+      assert(res.body.id);
+    });
+
+    it('should 201 if user login when alwaysAuth = false', async () => {
+      mock(app.config.cnpmcore, 'alwaysAuth', false);
+      const res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .set('authorization', publisher.authorization)
+        .expect(201);
+      assert.equal(res.body.ok, true);
+      assert.equal(res.body.state, 'waiting');
+      assert(res.body.id);
+    });
+
+    it('should 201 if user not login when alwaysAuth = false', async () => {
+      const res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .expect(201);
+      assert.equal(res.body.ok, true);
+      assert.equal(res.body.state, 'waiting');
+      assert(res.body.id);
+    });
+  });
+});
--- a/test/port/controller/PackageSyncController/showSyncTask.test.ts
+++ b/test/port/controller/PackageSyncController/showSyncTask.test.ts
@@ -0,0 +1,115 @@
+import { strict as assert } from 'assert';
+import { Context } from 'egg';
+import { app, mock } from 'egg-mock/bootstrap';
+import { TestUtil } from 'test/TestUtil';
+import { TaskRepository } from '../../../../app/repository/TaskRepository';
+import { TaskState } from '../../../../app/common/enum/Task';
+
+describe('test/port/controller/PackageSyncController/showSyncTask.test.ts', () => {
+  let publisher;
+  let ctx: Context;
+  let taskRepository: TaskRepository;
+  beforeEach(async () => {
+    publisher = await TestUtil.createUser();
+    ctx = await app.mockModuleContext();
+    taskRepository = await ctx.getEggObject(TaskRepository);
+  });
+
+  afterEach(() => {
+    app.destroyModuleContext(ctx);
+  });
+
+  describe('[GET /-/package/:fullname/syncs/:taskId] showSyncTask()', () => {
+    it('should 401 if user not login when alwaysAuth = true', async () => {
+      const pkg = await TestUtil.getFullPackage({ name: '@cnpm/koa', version: '1.0.0' });
+      await app.httpRequest()
+        .put(`/${pkg.name}`)
+        .set('authorization', publisher.authorization)
+        .set('user-agent', publisher.ua)
+        .send(pkg)
+        .expect(201);
+
+      mock(app.config.cnpmcore, 'alwaysAuth', true);
+      const res = await app.httpRequest()
+        .get(`/-/package/${pkg.name}/syncs/mock-task-id`)
+        .expect(401);
+      assert.equal(res.body.error, '[UNAUTHORIZED] Login first');
+    });
+
+    it('should 404 when task not exists', async () => {
+      const res = await app.httpRequest()
+        .get('/-/package/koa/syncs/mock-task-id')
+        .expect(404);
+      assert.equal(res.body.error, '[NOT_FOUND] Package "koa" sync task "mock-task-id" not found');
+    });
+
+    it('should 200', async () => {
+      let res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .expect(201);
+      assert(res.body.id);
+      const task = await taskRepository.findTask(res.body.id);
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}`)
+        .expect(200);
+      assert(res.body.id);
+      // waiting state logUrl is not exists
+      assert(!res.body.logUrl);
+
+      task!.state = TaskState.Processing;
+      await taskRepository.saveTask(task!);
+
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}`)
+        .expect(200);
+      assert(res.body.id);
+      assert(res.body.logUrl);
+      assert.match(res.body.logUrl, /^http:\/\//);
+      assert.match(res.body.logUrl, /\/log$/);
+    });
+
+    it('should get sucess task after schedule run', async () => {
+      const name = 'mk2test-module-cnpmsync-issue-1667';
+      let res = await app.httpRequest()
+        .put(`/-/package/${name}/syncs`)
+        .expect(201);
+      const taskId = res.body.id;
+      assert(taskId);
+      res = await app.httpRequest()
+        .get(`/-/package/${name}/syncs/${taskId}`)
+        .expect(200);
+      // waiting state logUrl is not exists
+      assert(!res.body.logUrl);
+      await app.runSchedule('SyncPackageWorker');
+      // again should work
+      await app.runSchedule('SyncPackageWorker');
+
+      res = await app.httpRequest()
+        .get(`/-/package/${name}/syncs/${taskId}`)
+        .expect(200);
+      assert.equal(res.body.state, TaskState.Success);
+      assert(res.body.logUrl);
+
+      res = await app.httpRequest()
+        .get(`/-/package/${name}/syncs/${taskId}/log`)
+        .expect(200);
+      // console.log(res.text);
+      assert.match(res.text, /🟢🟢🟢🟢🟢/);
+
+      // check hasInstallScript
+      res = await app.httpRequest()
+        .get(`/${name}`)
+        .expect(200);
+      let pkg = res.body.versions['3.0.0'];
+      assert(!('hasInstallScript' in pkg));
+      assert(pkg.scripts);
+      res = await app.httpRequest()
+        .get(`/${name}`)
+        .set('accept', 'application/vnd.npm.install-v1+json')
+        .expect(200);
+      pkg = res.body.versions['3.0.0'];
+      assert.equal(pkg.hasInstallScript, true);
+      assert(!pkg.scripts);
+    });
+  });
+});
--- a/test/port/controller/PackageSyncController/showSyncTaskLog.test.ts
+++ b/test/port/controller/PackageSyncController/showSyncTaskLog.test.ts
@@ -0,0 +1,86 @@
+import { strict as assert } from 'assert';
+import { Context } from 'egg';
+import { app, mock } from 'egg-mock/bootstrap';
+import { TestUtil } from 'test/TestUtil';
+import { TaskRepository } from 'app/repository/TaskRepository';
+import { TaskState } from 'app/common/enum/Task';
+import { NFSAdapter } from 'app/common/adapter/NFSAdapter';
+
+describe('test/port/controller/PackageSyncController/showSyncTaskLog.test.ts', () => {
+  let publisher;
+  let ctx: Context;
+  let taskRepository: TaskRepository;
+  let nfsAdapter: NFSAdapter;
+  beforeEach(async () => {
+    publisher = await TestUtil.createUser();
+    ctx = await app.mockModuleContext();
+    taskRepository = await ctx.getEggObject(TaskRepository);
+    nfsAdapter = await ctx.getEggObject(NFSAdapter);
+  });
+
+  afterEach(() => {
+    app.destroyModuleContext(ctx);
+  });
+
+  describe('[GET /-/package/:fullname/syncs/:taskId/log] showSyncTaskLog()', () => {
+    it('should 401 if user not login when alwaysAuth = true', async () => {
+      const pkg = await TestUtil.getFullPackage({ name: '@cnpm/koa' });
+      await app.httpRequest()
+        .put(`/${pkg.name}`)
+        .set('authorization', publisher.authorization)
+        .set('user-agent', publisher.ua)
+        .send(pkg)
+        .expect(201);
+
+      mock(app.config.cnpmcore, 'alwaysAuth', true);
+      const res = await app.httpRequest()
+        .get(`/-/package/${pkg.name}/syncs/mock-task-id/log`)
+        .expect(401);
+      assert.equal(res.body.error, '[UNAUTHORIZED] Login first');
+    });
+
+    it('should 404 when task not exists', async () => {
+      const res = await app.httpRequest()
+        .get('/-/package/koa/syncs/mock-task-id/log')
+        .expect(404);
+      assert.equal(res.body.error, '[NOT_FOUND] Package "koa" sync task "mock-task-id" not found');
+    });
+
+    it('should 200 and 302', async () => {
+      let res = await app.httpRequest()
+        .put('/-/package/koa/syncs')
+        .expect(201);
+      assert(res.body.id);
+      const task = await taskRepository.findTask(res.body.id);
+      // waiting state logUrl is not exists
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}/log`)
+        .expect(404);
+      assert.equal(res.body.error, `[NOT_FOUND] Package "koa" sync task "${task!.taskId}" log not found`);
+
+      task!.state = TaskState.Processing;
+      await taskRepository.saveTask(task!);
+
+      // log file not exists
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}/log`)
+        .expect(404);
+      assert.equal(res.body.error, `[NOT_FOUND] Package "koa" sync task "${task!.taskId}" log not found`);
+
+      // save log file
+      await nfsAdapter.uploadBytes(task!.logPath, Buffer.from('hello log file 😄\nsencod line here'));
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}/log`)
+        .expect('content-type', 'text/plain; charset=utf-8')
+        .expect(200);
+      assert.equal(res.text, 'hello log file 😄\nsencod line here');
+
+      // mock redirect
+      mock.data(nfsAdapter.constructor.prototype, 'getDownloadUrlOrStream', 'http://mock.com/some.log');
+      res = await app.httpRequest()
+        .get(`/-/package/koa/syncs/${task!.taskId}/log`)
+        .expect('location', 'http://mock.com/some.log')
+        .expect(302);
+    });
+  });
+});
--- a/test/port/controller/PackageTagController/saveTag.test.ts
+++ b/test/port/controller/PackageTagController/saveTag.test.ts
@@ -128,6 +128,18 @@ describe('test/port/controller/PackageTagController/saveTag.test.ts', () => {
      assert.equal(res.body.error, '[INVALID_PARAM] tag: must match format "semver-tag"');
    });

+    it('should 403 when package is public', async () => {
+      const { pkg, user } = await TestUtil.createPackage({ isPrivate: false, version: '1.0.0' });
+      const res = await app.httpRequest()
+        .put(`/-/package/${pkg.name}/dist-tags/beta`)
+        .set('authorization', user.authorization)
+        .set('user-agent', user.ua)
+        .set('content-type', 'application/json')
+        .send(JSON.stringify('1.0.0'))
+        .expect(403);
+      assert.equal(res.body.error, `[FORBIDDEN] Can\'t modify npm public package "${pkg.name}"`);
+    });
+
    it('should 200', async () => {
      const pkg = await TestUtil.getFullPackage({ name: '@cnpm/koa', version: '1.0.0' });
      await app.httpRequest()
@@ -184,7 +196,6 @@ describe('test/port/controller/PackageTagController/saveTag.test.ts', () => {
        .put(`/${pkg.name}`)
        .set('authorization', publisher.authorization)
        .set('user-agent', publisher.ua)
-        .set('user-agent', publisher.ua)
        .send(pkg)
        .expect(201);
      const userAutomation = await TestUtil.createTokenByUser({
@@ -196,7 +207,6 @@ describe('test/port/controller/PackageTagController/saveTag.test.ts', () => {
        .put(`/-/package/${pkg.name}/dist-tags/automation`)
        .set('authorization', userAutomation.authorization)
        .set('user-agent', publisher.ua)
-        .set('user-agent', publisher.ua)
        .set('content-type', 'application/json')
        .send(JSON.stringify('1.0.0'))
        .expect(200);
@@ -212,7 +222,6 @@ describe('test/port/controller/PackageTagController/saveTag.test.ts', () => {
        .put(`/-/package/${pkg.name}/dist-tags/latest-3`)
        .set('authorization', userAutomation.authorization)
        .set('user-agent', publisher.ua)
-        .set('user-agent', publisher.ua)
        .set('content-type', 'application/json')
        .send(JSON.stringify('1.0.0'))
        .expect(200);