Implement assigning args to runtime dynamic ds (#265)

* Implement assigning args to runtime dynamic ds

* Update changelog

packages/node/CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ### Added
 - Add monitor service to record block indexing actions in order to improve POI accuracy, and provide debug info for Admin api (#264)
+- The ability to specify filters when creating dynamic data sources (#265)
 
 ### Changed
 - Update dependencies (#264)

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

@@ -0,0 +1,153 @@
+// Copyright 2020-2024 SubQuery Pte Ltd authors & contributors
+// SPDX-License-Identifier: GPL-3.0
+
+import { CacheMetadataModel } from '@subql/node-core';
+import { CosmosDatasourceKind, CosmosHandlerKind } from '@subql/types-cosmos';
+import { SubqueryProject } from '../configure/SubqueryProject';
+import { DynamicDsService } from './dynamic-ds.service';
+
+function getMetadata(): CacheMetadataModel {
+  let dynamicDs: any[] = [];
+
+  const metadata = {
+    set: (key: string, data: any[]) => {
+      dynamicDs = data;
+    },
+    setNewDynamicDatasource: (newDs: any) => {
+      dynamicDs.push(newDs);
+    },
+    find: (key: string) => Promise.resolve(dynamicDs),
+  };
+
+  return metadata as CacheMetadataModel;
+}
+
+describe('Creating dynamic ds', () => {
+  let dynamiDsService: DynamicDsService;
+  let project: SubqueryProject;
+
+  beforeEach(async () => {
+    project = new SubqueryProject(
+      '',
+      '',
+      null,
+      [],
+      null,
+      [
+        {
+          name: 'cosmos',
+          kind: CosmosDatasourceKind.Runtime,
+          mapping: {
+            file: '',
+            handlers: [
+              {
+                handler: 'handleEvent',
+                kind: CosmosHandlerKind.Event,
+                filter: {
+                  type: 'execute',
+                  messageFilter: {
+                    type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+                  },
+                },
+              },
+              {
+                handler: 'handleMessage',
+                kind: CosmosHandlerKind.Message,
+                filter: {
+                  type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+                },
+              },
+            ],
+          },
+        },
+      ],
+      null,
+      null,
+      null,
+    );
+    dynamiDsService = new DynamicDsService(null, project);
+
+    await dynamiDsService.init(getMetadata());
+  });
+
+  // Cant test this because createDynamicDatasource calls process.exit on error
+  it.skip('Should validate the arguments', async () => {
+    await expect(
+      dynamiDsService.createDynamicDatasource({
+        templateName: 'cosmos',
+        startBlock: 100,
+        args: [] as any,
+      }),
+    ).rejects.toThrow();
+
+    await expect(
+      dynamiDsService.createDynamicDatasource({
+        templateName: 'cosmos',
+        startBlock: 100,
+        args: {
+          notValues: {},
+          attributes: [],
+        },
+      }),
+    ).rejects.toThrow();
+  });
+
+  it('Should be able to set an address for a cosmwasm contract', async () => {
+    const ds = await dynamiDsService.createDynamicDatasource({
+      templateName: 'cosmos',
+      startBlock: 100,
+      args: {
+        values: { contract: 'cosmos1' },
+        attributes: { _contract_address: 'cosmos_wasm' },
+      },
+    });
+
+    expect(ds).toEqual({
+      kind: CosmosDatasourceKind.Runtime,
+      startBlock: 100,
+      mapping: {
+        file: '',
+        handlers: [
+          {
+            handler: 'handleEvent',
+            kind: CosmosHandlerKind.Event,
+            filter: {
+              type: 'execute',
+              messageFilter: {
+                type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+                values: {
+                  contract: 'cosmos1',
+                },
+              },
+              attributes: {
+                _contract_address: 'cosmos_wasm',
+              },
+            },
+          },
+          {
+            handler: 'handleMessage',
+            kind: CosmosHandlerKind.Message,
+            filter: {
+              type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+              values: {
+                contract: 'cosmos1',
+              },
+            },
+          },
+        ],
+      },
+    });
+
+    // Check that the project templates don't get mutated
+    expect(project.templates[0].mapping.handlers[0].filter).toEqual({
+      type: 'execute',
+      messageFilter: {
+        type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+      },
+    });
+
+    expect(project.templates[0].mapping.handlers[1].filter).toEqual({
+      type: '/cosmwasm.wasm.v1.MsgExecuteContract',
+    });
+  });
+});

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

@@ -2,15 +2,49 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import { Inject, Injectable } from '@nestjs/common';
-import { isCustomCosmosDs, isRuntimeCosmosDs } from '@subql/common-cosmos';
+import {
+  CosmosRuntimeDataSourceImpl,
+  isCustomCosmosDs,
+  isRuntimeCosmosDs,
+} from '@subql/common-cosmos';
 import {
   DatasourceParams,
   DynamicDsService as BaseDynamicDsService,
 } from '@subql/node-core';
-import { CosmosDatasource } from '@subql/types-cosmos';
+import { CosmosDatasource, CosmosHandlerKind } from '@subql/types-cosmos';
+import { plainToClass, ClassConstructor } from 'class-transformer';
+import { validateSync, IsOptional, IsObject } from 'class-validator';
 import { SubqueryProject } from '../configure/SubqueryProject';
 import { DsProcessorService } from './ds-processor.service';
 
+class DataSourceArgs {
+  @IsOptional()
+  @IsObject()
+  values?: Record<string, string>;
+
+  @IsOptional()
+  @IsObject()
+  attributes?: Record<string, string>;
+}
+
+function validateType<T extends object>(
+  classtype: ClassConstructor<T>,
+  data: T,
+  errorPrefix: string,
+) {
+  const parsed = plainToClass(classtype, data);
+
+  const errors = validateSync(parsed, {
+    whitelist: true,
+    forbidNonWhitelisted: false,
+  });
+  if (errors.length) {
+    throw new Error(
+      `${errorPrefix}\n${errors.map((e) => e.toString()).join('\n')}`,
+    );
+  }
+}
+
 @Injectable()
 export class DynamicDsService extends BaseDynamicDsService<
   CosmosDatasource,
@@ -39,7 +73,61 @@ export class DynamicDsService extends BaseDynamicDsService<
         };
         await this.dsProcessorService.validateCustomDs([dsObj]);
       } else if (isRuntimeCosmosDs(dsObj)) {
-        // XXX add any modifications to the ds here
+        validateType(
+          DataSourceArgs,
+          params.args,
+          'Dynamic ds args are invalid',
+        );
+
+        dsObj.mapping.handlers = dsObj.mapping.handlers.map((handler) => {
+          switch (handler.kind) {
+            case CosmosHandlerKind.Event:
+              if (handler.filter) {
+                return {
+                  ...handler,
+                  filter: {
+                    ...handler.filter,
+                    messageFilter: {
+                      ...handler.filter.messageFilter,
+                      values: {
+                        ...handler.filter.messageFilter.values,
+                        ...(params.args.values as Record<string, string>),
+                      },
+                    },
+                    attributes: {
+                      ...handler.filter.attributes,
+                      ...(params.args.attributes as Record<string, string>),
+                    },
+                  },
+                };
+              }
+              return handler;
+            case CosmosHandlerKind.Message:
+              if (handler.filter) {
+                return {
+                  ...handler,
+                  filter: {
+                    ...handler.filter,
+                    values: {
+                      ...handler.filter.values,
+                      ...(params.args.values as Record<string, string>),
+                    },
+                  },
+                };
+              }
+              return handler;
+            case CosmosHandlerKind.Transaction:
+            case CosmosHandlerKind.Block:
+            default:
+              return handler;
+          }
+        });
+
+        validateType(
+          CosmosRuntimeDataSourceImpl,
+          dsObj,
+          'Dynamic ds is invalid',
+        );
       }
 
       return dsObj;