fix: changes stream (#297)

1. 修复 registry/sync 初始化时没有传递 registryId 问题
2. 修复 changesStream 获取时，如果没有需要同步的任务会导致 task#since 无法更新

app/core/entity/Task.ts

@@ -136,7 +136,7 @@ export class Task<T extends TaskBaseData = TaskBaseData> extends Entity {
     return task;
   }
 
-  public static createChangesStream(targetName: string, since = ''): ChangesStreamTask {
+  public static createChangesStream(targetName: string, registryId = '', since = ''): ChangesStreamTask {
     const data = {
       type: TaskType.ChangesStream,
       state: TaskState.Waiting,
@@ -146,6 +146,7 @@ export class Task<T extends TaskBaseData = TaskBaseData> extends Entity {
       data: {
         // task execute worker
         taskWorker: '',
+        registryId,
         since,
       },
     };

app/core/service/ChangesStreamService.ts

@@ -164,10 +164,10 @@ export class ChangesStreamService extends AbstractService {
     for await (const change of stream) {
       const { fullname, seq } = change as ChangesStreamChange;
       lastPackage = fullname;
+      lastSince = seq;
       const valid = await this.needSync(registry, fullname);
       if (valid) {
         taskCount++;
-        lastSince = seq;
         await this.packageSyncerService.createTask(fullname, {
           authorIp: HOST_NAME,
           authorId: 'ChangesStreamService',
@@ -185,6 +185,17 @@ export class ChangesStreamService extends AbstractService {
       }
     }
 
+    // 如果没有需要同步的任务，也更新一下 taskData 里的信息
+    if (taskCount === 0) {
+      // 实时更新 task 信息
+      task.updateSyncData({
+        lastSince,
+        lastPackage,
+        taskCount,
+      });
+      await this.taskRepository.saveTask(task);
+    }
+
     return { lastSince, taskCount };
   }
 }

app/core/service/RegistryManagerService.ts

@@ -55,7 +55,7 @@ export class RegistryManagerService extends AbstractService {
 
     // 启动 changeStream
     const targetName = `${registry.name.toUpperCase()}_WORKER`;
-    await this.taskService.createTask(Task.createChangesStream(targetName, since), false);
+    await this.taskService.createTask(Task.createChangesStream(targetName, registryId, since), false);
   }
 
   async createRegistry(createCmd: CreateRegistryCmd): Promise<Registry> {

app/repository/model/HistoryTask.ts

@@ -1,5 +1,5 @@
 import { Attribute, Model } from '@eggjs/tegg-orm-decorator';
-import { DataTypes, Bone } from 'leoric';
+import { DataTypes, Bone, LENGTH_VARIANTS } from 'leoric';
 import { TaskState, TaskType } from '../../common/enum/Task';
 
 @Model()
@@ -48,6 +48,6 @@ export class HistoryTask extends Bone {
   @Attribute(DataTypes.INTEGER)
   attempts: number;
 
-  @Attribute(DataTypes.TEXT('long'))
+  @Attribute(DataTypes.TEXT(LENGTH_VARIANTS.long))
   error: string;
 }

app/repository/model/PackageVersionBlock.ts

@@ -1,5 +1,5 @@
 import { Attribute, Model } from '@eggjs/tegg-orm-decorator';
-import { DataTypes, Bone } from 'leoric';
+import { DataTypes, Bone, LENGTH_VARIANTS } from 'leoric';
 
 @Model()
 export class PackageVersionBlock extends Bone {
@@ -26,6 +26,6 @@ export class PackageVersionBlock extends Bone {
   @Attribute(DataTypes.STRING(256))
   version: string;
 
-  @Attribute(DataTypes.TEXT('long'))
+  @Attribute(DataTypes.TEXT(LENGTH_VARIANTS.long))
   reason: string;
 }

app/repository/model/Task.ts

@@ -1,5 +1,5 @@
 import { Attribute, Model } from '@eggjs/tegg-orm-decorator';
-import { DataTypes, Bone } from 'leoric';
+import { DataTypes, Bone, LENGTH_VARIANTS } from 'leoric';
 import { TaskState, TaskType } from '../../common/enum/Task';
 
 @Model()
@@ -48,7 +48,7 @@ export class Task extends Bone {
   @Attribute(DataTypes.INTEGER)
   attempts: number;
 
-  @Attribute(DataTypes.TEXT('long'))
+  @Attribute(DataTypes.TEXT(LENGTH_VARIANTS.long))
   error: string;
 
   @Attribute(DataTypes.STRING(48), {

test/core/service/ChangesStreamService.test.ts

@@ -27,7 +27,7 @@ describe('test/core/service/ChangesStreamService.test.ts', () => {
     registryManagerService = await ctx.getEggObject(RegistryManagerService);
     scopeManagerService = await ctx.getEggObject(ScopeManagerService);
     assert(changesStreamService);
-    task = Task.createChangesStream('GLOBAL_WORKER', '9527');
+    task = Task.createChangesStream('GLOBAL_WORKER', '', '9527');
     taskService.createTask(task, false);
 
     // create default registry
@@ -163,6 +163,24 @@ describe('test/core/service/ChangesStreamService.test.ts', () => {
       assert(changes.taskCount === 2);
       assert(changes.lastSince === '3');
     });
+
+    it('should update since even not taskCount', async () => {
+      mock(ChangesStreamService.prototype, 'needSync', async () => {
+        return false;
+      });
+      mock(ctx.httpclient, 'request', async () => {
+        return {
+          res: Readable.from(`
+            {"seq":2,"id":"backbone.websql.deferred","changes":[{"rev":"4-f5150b238ab62cd890211fb57fc9eca5"}],"deleted":true},
+            {"seq":3,"id":"backbone2.websql.deferred","changes":[{"rev":"4-f6150b238ab62cd890211fb57fc9eca5"}],"deleted":true},
+            `),
+        };
+      });
+      const changes = await changesStreamService.executeSync('1', task);
+      assert(changes.taskCount === 0);
+      assert(changes.lastSince === '3');
+      assert(task.data.since === '3');
+    });
   });
 
 });

test/core/service/RegistryManagerService/index.test.ts

@@ -127,6 +127,7 @@ describe('test/core/service/RegistryManagerService/index.test.ts', () => {
         const targetName = 'CUSTOM_WORKER';
         const task = await taskRepository.findTaskByTargetName(targetName, TaskType.ChangesStream);
         assert(task);
+        assert((task.data as ChangesStreamTaskData).registryId === registry.registryId);
       });
 
       it('should preCheck registry', async () => {