feat: implemented dynamic datasource removing themseleves and adding them at end blocks

packages/node-core/src/indexer/dynamic-ds.service.spec.ts

@@ -15,15 +15,16 @@ class TestDynamicDsService extends DynamicDsService<BaseDataSource, ISubqueryPro
   }
 
   // Make it public
-  getTemplate(templateName: string, startBlock?: number): BaseDataSource {
-    return super.getTemplate(templateName, startBlock);
+  getTemplate(templateName: string, startBlock?: number | undefined, endBlock?: number | undefined): BaseDataSource {
+    return super.getTemplate(templateName, startBlock, endBlock);
   }
 }
 
 const testParam1 = {templateName: 'Test', startBlock: 1};
 const testParam2 = {templateName: 'Test', startBlock: 2};
 const testParam3 = {templateName: 'Test', startBlock: 3};
 const testParam4 = {templateName: 'Test', startBlock: 4};
+const testParamOther = {templateName: 'Other', startBlock: 5};
 
 const mockMetadata = (initData: DatasourceParams[] = []) => {
   let datasourceParams: DatasourceParams[] = initData;
@@ -40,7 +41,7 @@ const mockMetadata = (initData: DatasourceParams[] = []) => {
 describe('DynamicDsService', () => {
   let service: TestDynamicDsService;
   const project = {
-    templates: [{name: 'Test'}],
+    templates: [{name: 'Test'}, {name: 'Other'}],
   } as any as ISubqueryProject;
 
   beforeEach(() => {
@@ -70,6 +71,69 @@ describe('DynamicDsService', () => {
     ]);
   });
 
+  it('can destroy a dynamic datasource', async () => {
+    const meta = mockMetadata([testParam1, testParam2]);
+    await service.init(meta);
+
+    // Destroy specific datasource by index
+    await service.destroyDynamicDatasource('Test', 50, 0);
+
+    const updatedParams = (service as any)._datasourceParams;
+    expect(updatedParams[0]).toEqual({...testParam1, endBlock: 50});
+    expect(updatedParams[1]).toEqual(testParam2);
+
+    const datasources = (service as any)._datasources;
+    expect(datasources[0].endBlock).toBe(50);
+  });
+
+  it('throws error when destroying non-existent datasource', async () => {
+    const meta = mockMetadata([testParam1]);
+    await service.init(meta);
+
+    await expect(service.destroyDynamicDatasource('NonExistent', 50, 0)).rejects.toThrow(
+      'Datasource at index 0 has template name "Test", not "NonExistent"'
+    );
+  });
+
+  it('throws error when destroying already destroyed datasource', async () => {
+    const destroyedParam = {...testParam1, endBlock: 30};
+    const meta = mockMetadata([destroyedParam]);
+    await service.init(meta);
+
+    await expect(service.destroyDynamicDatasource('Test', 50, 0)).rejects.toThrow(
+      'Dynamic datasource at index 0 is already destroyed'
+    );
+  });
+
+  it('allows creating new datasource after destroying existing one', async () => {
+    const meta = mockMetadata([testParam1]);
+    await service.init(meta);
+
+    expect((service as any)._datasourceParams).toEqual([testParam1]);
+
+    // Destroy by index
+    await service.destroyDynamicDatasource('Test', 50, 0);
+
+    const paramsAfterDestroy = (service as any)._datasourceParams;
+    expect(paramsAfterDestroy[0]).toEqual({...testParam1, endBlock: 50});
+
+    const newParam = {templateName: 'Test', startBlock: 60};
+    await service.createDynamicDatasource(newParam);
+
+    const finalParams = (service as any)._datasourceParams;
+    const destroyedCount = finalParams.filter((p) => p.endBlock !== undefined).length;
+    const activeCount = finalParams.filter((p) => p.endBlock === undefined).length;
+
+    expect(destroyedCount).toBeGreaterThanOrEqual(1);
+    expect(activeCount).toBeGreaterThanOrEqual(1);
+
+    const destroyedParam = finalParams.find((p) => p.startBlock === 1 && p.endBlock === 50);
+    expect(destroyedParam).toBeDefined();
+
+    const newParamFound = finalParams.find((p) => p.startBlock === 60 && !p.endBlock);
+    expect(newParamFound).toBeDefined();
+  });
+
   it('resets dynamic datasources', async () => {
     const meta = mockMetadata([testParam1, testParam2, testParam3, testParam4]);
     await service.init(meta);
@@ -83,6 +147,26 @@ describe('DynamicDsService', () => {
     ]);
   });
 
+  it('handles reset after datasource destruction correctly', async () => {
+    const params = [testParam1, testParam2, testParam3, testParam4];
+    const meta = mockMetadata(params);
+    await service.init(meta);
+
+    // Destroy only the first datasource by index
+    await service.destroyDynamicDatasource('Test', 25, 0);
+
+    const paramsAfterDestroy = (service as any)._datasourceParams;
+    expect(paramsAfterDestroy[0]).toEqual({...testParam1, endBlock: 25});
+
+    // Reset to block 2 (should keep testParam1 and testParam2)
+    await service.resetDynamicDatasource(2, null as any);
+
+    const paramsAfterReset = (service as any)._datasourceParams;
+    expect(paramsAfterReset).toHaveLength(2);
+    expect(paramsAfterReset[0]).toEqual({...testParam1, endBlock: 25});
+    expect(paramsAfterReset[1]).toEqual(testParam2);
+  });
+
   it('getDynamicDatasources with force reloads from metadata', async () => {
     const meta = mockMetadata([testParam1, testParam2]);
     await service.init(meta);
@@ -107,6 +191,30 @@ describe('DynamicDsService', () => {
     ]);
   });
 
+  it('loads destroyed datasources with endBlock correctly', async () => {
+    const destroyedParam = {...testParam1, endBlock: 100};
+    const meta = mockMetadata([destroyedParam, testParam2]);
+    await service.init(meta);
+
+    const datasources = await service.getDynamicDatasources();
+    expect(datasources).toHaveLength(2);
+    expect((datasources[0] as any).endBlock).toBe(100);
+    expect((datasources[1] as any).endBlock).toBeUndefined();
+  });
+
+  it('updates metadata correctly when destroying datasource', async () => {
+    const meta = mockMetadata([testParam1, testParam2]);
+    await service.init(meta);
+
+    // Destroy first datasource by index
+    await service.destroyDynamicDatasource('Test', 75, 0);
+
+    const metadataParams = await meta.find('dynamicDatasources');
+    expect(metadataParams).toBeDefined();
+    expect(metadataParams![0]).toEqual({...testParam1, endBlock: 75});
+    expect(metadataParams![1]).toEqual(testParam2);
+  });
+
   it('can find a template and cannot mutate the template', () => {
     const template1 = service.getTemplate('Test', 1);
     const template2 = service.getTemplate('Test', 2);
@@ -119,4 +227,232 @@ describe('DynamicDsService', () => {
 
     expect(project.templates![0]).toEqual({name: 'Test'});
   });
+
+  it('can create template with endBlock', () => {
+    const template = service.getTemplate('Test', 1, 100);
+
+    expect(template.startBlock).toBe(1);
+    expect((template as any).endBlock).toBe(100);
+    expect((template as any).name).toBeUndefined();
+  });
+
+  it('handles multiple templates with same name during destruction', async () => {
+    const param1 = {templateName: 'Test', startBlock: 1};
+    const param2 = {templateName: 'Test', startBlock: 5};
+    const param3 = {templateName: 'Other', startBlock: 3};
+
+    const meta = mockMetadata([param1, param2, param3]);
+    await service.init(meta);
+
+    // Should destroy the first matching one by index
+    await service.destroyDynamicDatasource('Test', 10, 0);
+
+    const updatedParams = (service as any)._datasourceParams;
+    expect(updatedParams[0]).toEqual({...param1, endBlock: 10});
+    expect(updatedParams[1]).toEqual(param2); // Not destroyed
+    expect(updatedParams[2]).toEqual(param3); // Not destroyed
+  });
+
+  it('throws error when service not initialized for destruction', async () => {
+    await expect(service.destroyDynamicDatasource('Test', 50, 0)).rejects.toThrow(
+      'DynamicDsService has not been initialized'
+    );
+  });
+
+  describe('getDynamicDatasourcesByTemplate', () => {
+    it('returns list of active datasources for a template', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParam3, testParamOther]);
+      await service.init(meta);
+
+      const testDatasources = service.getDynamicDatasourcesByTemplate('Test');
+
+      expect(testDatasources).toHaveLength(3);
+      expect(testDatasources[0]).toEqual({
+        index: 0,
+        templateName: 'Test',
+        startBlock: 1,
+        endBlock: undefined,
+        args: undefined,
+      });
+      expect(testDatasources[1]).toEqual({
+        index: 1,
+        templateName: 'Test',
+        startBlock: 2,
+        endBlock: undefined,
+        args: undefined,
+      });
+      expect(testDatasources[2]).toEqual({
+        index: 2,
+        templateName: 'Test',
+        startBlock: 3,
+        endBlock: undefined,
+        args: undefined,
+      });
+    });
+
+    it('excludes destroyed datasources from list', async () => {
+      const destroyedParam = {...testParam1, endBlock: 50};
+      const meta = mockMetadata([destroyedParam, testParam2, testParam3]);
+      await service.init(meta);
+
+      const datasources = service.getDynamicDatasourcesByTemplate('Test');
+
+      expect(datasources).toHaveLength(2);
+      expect(datasources[0].index).toBe(1); // Global index
+      expect(datasources[0].startBlock).toBe(2);
+      expect(datasources[1].index).toBe(2); // Global index
+      expect(datasources[1].startBlock).toBe(3);
+    });
+
+    it('returns empty array when no datasources match template', async () => {
+      const meta = mockMetadata([testParamOther]);
+      await service.init(meta);
+
+      const datasources = service.getDynamicDatasourcesByTemplate('Test');
+
+      expect(datasources).toEqual([]);
+    });
+
+    it('includes args in datasource info when present', async () => {
+      const paramWithArgs = {...testParam1, args: {address: '0x123', tokenId: 1}};
+      const meta = mockMetadata([paramWithArgs]);
+      await service.init(meta);
+
+      const datasources = service.getDynamicDatasourcesByTemplate('Test');
+
+      expect(datasources).toHaveLength(1);
+      expect(datasources[0].args).toEqual({address: '0x123', tokenId: 1});
+    });
+
+    it('throws error when service not initialized', () => {
+      expect(() => service.getDynamicDatasourcesByTemplate('Test')).toThrow(
+        'DynamicDsService has not been initialized'
+      );
+    });
+  });
+
+  describe('destroyDynamicDatasource with index', () => {
+    it('destroys specific datasource by index', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParam3, testParamOther]);
+      await service.init(meta);
+
+      await service.destroyDynamicDatasource('Test', 50, 1);
+
+      const updatedParams = (service as any)._datasourceParams;
+      expect(updatedParams[0]).toEqual(testParam1); // Not destroyed
+      expect(updatedParams[1]).toEqual({...testParam2, endBlock: 50}); // Destroyed
+      expect(updatedParams[2]).toEqual(testParam3); // Not destroyed
+      expect(updatedParams[3]).toEqual(testParamOther); // Not destroyed
+    });
+
+    it('throws error when index is out of bounds', async () => {
+      const meta = mockMetadata([testParam1, testParam2]);
+      await service.init(meta);
+
+      await expect(service.destroyDynamicDatasource('Test', 50, 5)).rejects.toThrow(
+        'Index 5 is out of bounds. There are 2 datasource(s) in total'
+      );
+    });
+
+    it('throws error when index is negative', async () => {
+      const meta = mockMetadata([testParam1, testParam2]);
+      await service.init(meta);
+
+      await expect(service.destroyDynamicDatasource('Test', 50, -1)).rejects.toThrow(
+        'Index -1 is out of bounds. There are 2 datasource(s) in total'
+      );
+    });
+
+    it('throws error when trying to destroy already destroyed datasource', async () => {
+      const destroyedParam = {...testParam1, endBlock: 30};
+      const meta = mockMetadata([destroyedParam]);
+      await service.init(meta);
+
+      await expect(service.destroyDynamicDatasource('Test', 50, 0)).rejects.toThrow(
+        'Dynamic datasource at index 0 is already destroyed'
+      );
+    });
+
+    it('correctly handles global index after some datasources are destroyed', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParam3, testParam4]);
+      await service.init(meta);
+
+      // Destroy the first one using global index 0
+      await service.destroyDynamicDatasource('Test', 40, 0);
+
+      // Now only 3 active datasources for 'Test' template, with global indices 1, 2, 3
+      const activeDatasources = service.getDynamicDatasourcesByTemplate('Test');
+      expect(activeDatasources).toHaveLength(3);
+      expect(activeDatasources[0].index).toBe(1); // Global index
+      expect(activeDatasources[0].startBlock).toBe(2);
+      expect(activeDatasources[1].index).toBe(2); // Global index
+      expect(activeDatasources[1].startBlock).toBe(3);
+      expect(activeDatasources[2].index).toBe(3); // Global index
+      expect(activeDatasources[2].startBlock).toBe(4);
+
+      // Destroy using global index 2 (testParam3)
+      await service.destroyDynamicDatasource('Test', 60, 2);
+
+      const updatedParams = (service as any)._datasourceParams;
+      expect(updatedParams[0]).toEqual({...testParam1, endBlock: 40});
+      expect(updatedParams[1]).toEqual(testParam2); // Still active
+      expect(updatedParams[2]).toEqual({...testParam3, endBlock: 60});
+      expect(updatedParams[3]).toEqual(testParam4); // Still active
+    });
+
+    it('updates datasources in memory correctly when destroying by index', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParam3]);
+      await service.init(meta);
+
+      await service.destroyDynamicDatasource('Test', 100, 1);
+
+      const datasources = (service as any)._datasources;
+      expect(datasources[0].endBlock).toBeUndefined();
+      expect(datasources[1].endBlock).toBe(100);
+      expect(datasources[2].endBlock).toBeUndefined();
+    });
+
+    it('allows destroying datasources from different templates independently', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParamOther]);
+      await service.init(meta);
+
+      await service.destroyDynamicDatasource('Test', 50, 0);
+      await service.destroyDynamicDatasource('Other', 60, 2);
+
+      const updatedParams = (service as any)._datasourceParams;
+      expect(updatedParams[0]).toEqual({...testParam1, endBlock: 50});
+      expect(updatedParams[1]).toEqual(testParam2); // Not destroyed
+      expect(updatedParams[2]).toEqual({...testParamOther, endBlock: 60});
+    });
+
+    it('throws error when template name does not match global index', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParamOther]);
+      await service.init(meta);
+
+      // Try to destroy 'Test' template with index 2, which is 'Other' template
+      await expect(service.destroyDynamicDatasource('Test', 50, 2)).rejects.toThrow(
+        'Datasource at index 2 has template name "Other", not "Test"'
+      );
+    });
+
+    it('sets endBlock correctly allowing in-place removal during block processing', async () => {
+      const meta = mockMetadata([testParam1, testParam2, testParam3]);
+      await service.init(meta);
+
+      // Destroy datasource at index 1 at block 50
+      await service.destroyDynamicDatasource('Test', 50, 1);
+
+      // Verify the datasource has endBlock set
+      const dsParam = service.getDatasourceParamByIndex(1);
+      expect(dsParam).toBeDefined();
+      expect(dsParam?.endBlock).toBe(50);
+      expect(dsParam?.startBlock).toBe(2);
+      expect(dsParam?.templateName).toBe('Test');
+
+      // Verify the internal _datasources array also has endBlock set
+      const datasources = (service as any)._datasources;
+      expect(datasources[1]).toBeDefined();
+      expect((datasources[1] as any).endBlock).toBe(50);
+    });
+  });
 });