fix: crawler update to new MessagesInBlock event (#59)

# Description
`crawler` service was stlil using the deprected `MessagesStored` event.
Updated/refactored to use common logic with `scanner` to look for
`MessagesInBlock` events.

Also changed all occurrences of Schema IDs and Block Numbers to JS
`number` types, as SchemaId is `u16` and Block Number is `u32` (no need
to handle `u64`/`bigint`, so no need to transmit as strings)

Author: JoeCap08055

apps/api/src/metadata.ts

@@ -1,94 +1,9 @@
 /* eslint-disable */
 export default async () => {
-  const t = {
-    ['../../../libs/common/src/dtos/activity.dto']: await import('../../../libs/common/src/dtos/activity.dto'),
-    ['../../../libs/common/src/dtos/announcement.dto']: await import('../../../libs/common/src/dtos/announcement.dto'),
-  };
-  return {
-    '@nestjs/swagger': {
-      models: [
-        [
-          import('../../../libs/common/src/dtos/common.dto'),
-          {
-            DsnpUserIdParam: { userDsnpId: { required: true, type: () => String } },
-            AnnouncementResponseDto: { referenceId: { required: true, type: () => String } },
-            UploadResponseDto: { assetIds: { required: true, type: () => [String] } },
-            FilesUploadDto: { files: { required: true, type: () => [Object] } },
-          },
-        ],
-        [
-          import('../../../libs/common/src/dtos/activity.dto'),
-          {
-            LocationDto: {
-              name: { required: true, type: () => String, minLength: 1 },
-              accuracy: { required: false, type: () => Number, minimum: 0, maximum: 100 },
-              altitude: { required: false, type: () => Number },
-              latitude: { required: false, type: () => Number },
-              longitude: { required: false, type: () => Number },
-              radius: { required: false, type: () => Number, minimum: 0 },
-              units: { required: false, enum: t['../../../libs/common/src/dtos/activity.dto'].UnitTypeDto },
-            },
-            AssetReferenceDto: {
-              referenceId: { required: true, type: () => String, minLength: 1 },
-              height: { required: false, type: () => Number, minimum: 1 },
-              width: { required: false, type: () => Number, minimum: 1 },
-              duration: { required: false, type: () => String, pattern: 'DURATION_REGEX' },
-            },
-            TagDto: {
-              type: { required: true, enum: t['../../../libs/common/src/dtos/activity.dto'].TagTypeDto },
-              name: { required: false, type: () => String, minLength: 1 },
-              mentionedId: { required: false, type: () => String, minLength: 1, pattern: 'DSNP_USER_URI_REGEX' },
-            },
-            AssetDto: {
-              type: { required: true, enum: t['../../../libs/common/src/dtos/activity.dto'].AttachmentTypeDto },
-              references: { required: false, type: () => [t['../../../libs/common/src/dtos/activity.dto'].AssetReferenceDto] },
-              name: { required: false, type: () => String, minLength: 1 },
-              href: { required: false, type: () => String, minLength: 1 },
-            },
-            BaseActivityDto: {
-              name: { required: false, type: () => String },
-              tag: { required: false, type: () => [t['../../../libs/common/src/dtos/activity.dto'].TagDto] },
-              location: { required: false, type: () => t['../../../libs/common/src/dtos/activity.dto'].LocationDto },
-            },
-            NoteActivityDto: {
-              content: { required: true, type: () => String, minLength: 1 },
-              published: { required: true, type: () => String, pattern: 'ISO8601_REGEX' },
-              assets: { required: false, type: () => [t['../../../libs/common/src/dtos/activity.dto'].AssetDto] },
-            },
-            ProfileActivityDto: {
-              icon: { required: false, type: () => [t['../../../libs/common/src/dtos/activity.dto'].AssetReferenceDto] },
-              summary: { required: false, type: () => String },
-              published: { required: false, type: () => String, pattern: 'ISO8601_REGEX' },
-            },
-          },
-        ],
-        [
-          import('../../../libs/common/src/dtos/announcement.dto'),
-          {
-            BroadcastDto: { content: { required: true, type: () => t['../../../libs/common/src/dtos/activity.dto'].NoteActivityDto } },
-            ReplyDto: {
-              inReplyTo: { required: true, type: () => String, pattern: 'DSNP_CONTENT_URI_REGEX' },
-              content: { required: true, type: () => t['../../../libs/common/src/dtos/activity.dto'].NoteActivityDto },
-            },
-            TombstoneDto: {
-              targetContentHash: { required: true, type: () => String, pattern: 'DSNP_CONTENT_HASH_REGEX' },
-              targetAnnouncementType: { required: true, enum: t['../../../libs/common/src/dtos/announcement.dto'].ModifiableAnnouncementTypeDto },
-            },
-            UpdateDto: {
-              targetContentHash: { required: true, type: () => String, pattern: 'DSNP_CONTENT_HASH_REGEX' },
-              targetAnnouncementType: { required: true, enum: t['../../../libs/common/src/dtos/announcement.dto'].ModifiableAnnouncementTypeDto },
-              content: { required: true, type: () => t['../../../libs/common/src/dtos/activity.dto'].NoteActivityDto },
-            },
-            ReactionDto: {
-              emoji: { required: true, type: () => String, minLength: 1, pattern: 'DSNP_EMOJI_REGEX' },
-              apply: { required: true, type: () => Number, minimum: 0, maximum: 255 },
-              inReplyTo: { required: true, type: () => String, pattern: 'DSNP_CONTENT_URI_REGEX' },
-            },
-            ProfileDto: { profile: { required: true, type: () => t['../../../libs/common/src/dtos/activity.dto'].ProfileActivityDto } },
-          },
-        ],
-      ],
-      controllers: [[import('./api.controller'), { ApiController: { health: {} } }]],
-    },
-  };
-};
+    const t = {
+        ["../../../libs/common/src/dtos/activity.dto"]: await import("../../../libs/common/src/dtos/activity.dto"),
+        ["../../../libs/common/src/dtos/announcement.dto"]: await import("../../../libs/common/src/dtos/announcement.dto"),
+        ["../../../libs/common/src/dtos/chain.watch.dto"]: await import("../../../libs/common/src/dtos/chain.watch.dto")
+    };
+    return { "@nestjs/swagger": { "models": [[import("../../../libs/common/src/dtos/common.dto"), { "DsnpUserIdParam": { userDsnpId: { required: true, type: () => String } }, "AnnouncementResponseDto": { referenceId: { required: true, type: () => String } }, "UploadResponseDto": { assetIds: { required: true, type: () => [String] } }, "FilesUploadDto": { files: { required: true, type: () => [Object] } }, "ResetScannerDto": { blockNumber: { required: true, type: () => Number } } }], [import("../../../libs/common/src/dtos/activity.dto"), { "LocationDto": { name: { required: true, type: () => String, minLength: 1 }, accuracy: { required: false, type: () => Number, minimum: 0, maximum: 100 }, altitude: { required: false, type: () => Number }, latitude: { required: false, type: () => Number }, longitude: { required: false, type: () => Number }, radius: { required: false, type: () => Number, minimum: 0 }, units: { required: false, enum: t["../../../libs/common/src/dtos/activity.dto"].UnitTypeDto } }, "AssetReferenceDto": { referenceId: { required: true, type: () => String, minLength: 1 }, height: { required: false, type: () => Number, minimum: 1 }, width: { required: false, type: () => Number, minimum: 1 }, duration: { required: false, type: () => String, pattern: "DURATION_REGEX" } }, "TagDto": { type: { required: true, enum: t["../../../libs/common/src/dtos/activity.dto"].TagTypeDto }, name: { required: false, type: () => String, minLength: 1 }, mentionedId: { required: false, type: () => String, minLength: 1, pattern: "DSNP_USER_URI_REGEX" } }, "AssetDto": { type: { required: true, enum: t["../../../libs/common/src/dtos/activity.dto"].AttachmentTypeDto }, references: { required: false, type: () => [t["../../../libs/common/src/dtos/activity.dto"].AssetReferenceDto] }, name: { required: false, type: () => String, minLength: 1 }, href: { required: false, type: () => String, minLength: 1 } }, "BaseActivityDto": { name: { required: false, type: () => String }, tag: { required: false, type: () => [t["../../../libs/common/src/dtos/activity.dto"].TagDto] }, location: { required: false, type: () => t["../../../libs/common/src/dtos/activity.dto"].LocationDto } }, "NoteActivityDto": { content: { required: true, type: () => String, minLength: 1 }, published: { required: true, type: () => String, pattern: "ISO8601_REGEX" }, assets: { required: false, type: () => [t["../../../libs/common/src/dtos/activity.dto"].AssetDto] } }, "ProfileActivityDto": { icon: { required: false, type: () => [t["../../../libs/common/src/dtos/activity.dto"].AssetReferenceDto] }, summary: { required: false, type: () => String }, published: { required: false, type: () => String, pattern: "ISO8601_REGEX" } } }], [import("../../../libs/common/src/dtos/announcement.dto"), { "BroadcastDto": { content: { required: true, type: () => t["../../../libs/common/src/dtos/activity.dto"].NoteActivityDto } }, "ReplyDto": { inReplyTo: { required: true, type: () => String, pattern: "DSNP_CONTENT_URI_REGEX" }, content: { required: true, type: () => t["../../../libs/common/src/dtos/activity.dto"].NoteActivityDto } }, "TombstoneDto": { targetContentHash: { required: true, type: () => String, pattern: "DSNP_CONTENT_HASH_REGEX" }, targetAnnouncementType: { required: true, enum: t["../../../libs/common/src/dtos/announcement.dto"].ModifiableAnnouncementTypeDto } }, "UpdateDto": { targetContentHash: { required: true, type: () => String, pattern: "DSNP_CONTENT_HASH_REGEX" }, targetAnnouncementType: { required: true, enum: t["../../../libs/common/src/dtos/announcement.dto"].ModifiableAnnouncementTypeDto }, content: { required: true, type: () => t["../../../libs/common/src/dtos/activity.dto"].NoteActivityDto } }, "ReactionDto": { emoji: { required: true, type: () => String, minLength: 1, pattern: "DSNP_EMOJI_REGEX" }, apply: { required: true, type: () => Number, minimum: 0, maximum: 255 }, inReplyTo: { required: true, type: () => String, pattern: "DSNP_CONTENT_URI_REGEX" } }, "ProfileDto": { profile: { required: true, type: () => t["../../../libs/common/src/dtos/activity.dto"].ProfileActivityDto } } }], [import("../../../libs/common/src/dtos/chain.watch.dto"), { "ChainWatchOptionsDto": { schemaIds: { required: true, type: () => [Number] }, dsnpIds: { required: true, type: () => [String] } } }], [import("../../../libs/common/src/dtos/request-job.dto"), { "ContentSearchRequestDto": { id: { required: true, type: () => String }, startBlock: { required: true, type: () => Number, minimum: 1 }, endBlock: { required: true, type: () => Number, minimum: 1 }, filters: { required: true, type: () => t["../../../libs/common/src/dtos/chain.watch.dto"].ChainWatchOptionsDto } } }], [import("../../../libs/common/src/dtos/subscription.webhook.dto"), { "WebhookRegistrationDto": { url: { required: true, type: () => String }, announcementTypes: { required: true, type: () => [String] } } }]], "controllers": [[import("./api.controller"), { "ApiController": { "health": {}, "resetScanner": { type: String }, "setWatchOptions": {}, "pauseScanner": {}, "startScanner": {}, "search": {}, "registerWebhook": {}, "clearAllWebHooks": {}, "getRegisteredWebhooks": {} } }]] } };
+};

content-announcement.openapi.json

@@ -61,7 +61,7 @@
             "description": "An optional identifier for the request, may be used for tracking or correlation"
           },
           "schemaId": {
-            "type": "string",
+            "type": "integer",
             "description": "Identifier for the schema being used or referenced"
           },
           "blockNumber": {

libs/common/src/blockchain/blockchain.module.ts

@@ -5,11 +5,12 @@ https://docs.nestjs.com/modules
 import { Module } from '@nestjs/common';
 import { BlockchainService } from './blockchain.service';
 import { ConfigModule } from '../config/config.module';
+import { ChainEventProcessorService } from './chain-event-processor.service';
 
 @Module({
   imports: [ConfigModule],
   controllers: [],
-  providers: [BlockchainService],
-  exports: [BlockchainService],
+  providers: [BlockchainService, ChainEventProcessorService],
+  exports: [BlockchainService, ChainEventProcessorService],
 })
 export class BlockchainModule {}

libs/common/src/blockchain/chain-event-processor.service.ts

@@ -0,0 +1,111 @@
+import { Injectable } from '@nestjs/common';
+import { BlockchainService } from './blockchain.service';
+import { ChainWatchOptionsDto } from '../dtos/chain.watch.dto';
+import { ApiDecoration } from '@polkadot/api/types';
+import { FrameSystemEventRecord } from '@polkadot/types/lookup';
+import { Vec } from '@polkadot/types';
+import { BlockPaginationResponseMessage, MessageResponse } from '@frequency-chain/api-augment/interfaces';
+import { MessageResponseWithSchemaId } from '../interfaces/message_response_with_schema_id';
+import { createIPFSQueueJob } from '../interfaces/ipfs.job.interface';
+import { Queue } from 'bullmq';
+
+@Injectable()
+export class ChainEventProcessorService {
+  constructor(private readonly blockchainService: BlockchainService) {}
+
+  public async getMessagesInBlock(blockNumber: number, filter?: ChainWatchOptionsDto): Promise<MessageResponseWithSchemaId[]> {
+    const blockHash = await this.blockchainService.getBlockHash(blockNumber);
+    if (blockHash.isEmpty) {
+        return [];
+    }
+    const apiAt = await this.blockchainService.apiPromise.at(blockHash);
+    const events = await apiAt.query.system.events();
+    return this.getMessagesFromEvents(apiAt, blockNumber, events, filter);
+  }
+
+  private async getMessagesFromEvents(
+    apiAt: ApiDecoration<'promise'>,
+    blockNumber: number,
+    events: Vec<FrameSystemEventRecord>,
+    filter?: ChainWatchOptionsDto,
+  ): Promise<MessageResponseWithSchemaId[]> {
+    const hasMessages = events.some(({ event }) => apiAt.events.messages.MessagesInBlock.is(event));
+    if (!hasMessages) {
+      return [];
+    }
+
+    const keys = await apiAt.query.messages.messagesV2.keys(blockNumber);
+    let schemaIds = keys.map((key) => key.args[1].toNumber());
+    schemaIds = Array.from(new Set<number>(schemaIds));
+    const filteredEvents: (MessageResponseWithSchemaId | null)[] = await Promise.all(
+      schemaIds.map(async (schemaId) => {
+        if (filter && filter?.schemaIds?.length > 0 && !filter.schemaIds.includes(schemaId)) {
+          return null;
+        }
+        let paginationRequest = {
+          from_block: blockNumber,
+          from_index: 0,
+          page_size: 1000,
+          to_block: blockNumber + 1,
+        };
+
+        let messageResponse: BlockPaginationResponseMessage = await this.blockchainService.apiPromise.rpc.messages.getBySchemaId(schemaId, paginationRequest);
+        const messages: Vec<MessageResponse> = messageResponse.content;
+        while (messageResponse.has_next.toHuman()) {
+          paginationRequest = {
+            from_block: blockNumber,
+            from_index: messageResponse.next_index.isSome ? messageResponse.next_index.unwrap().toNumber() : 0,
+            page_size: 1000,
+            to_block: blockNumber + 1,
+          };
+          // eslint-disable-next-line no-await-in-loop
+          messageResponse = await this.blockchainService.apiPromise.rpc.messages.getBySchemaId(schemaId, paginationRequest);
+          if (messageResponse.content.length > 0) {
+            messages.push(...messageResponse.content);
+          }
+        }
+        const messagesWithSchemaId: MessageResponseWithSchemaId = {
+          schemaId: schemaId,
+          messages,
+        };
+        return messagesWithSchemaId;
+      }),
+    );
+    const collectedMessages: MessageResponseWithSchemaId[] = [];
+    filteredEvents.forEach((event) => {
+      if (event) {
+        collectedMessages.push(event);
+      }
+    });
+    return collectedMessages;
+  }
+
+  public async queueIPFSJobs(messages: MessageResponseWithSchemaId[], queue: Queue, requestId?: string): Promise<void> {
+    const jobs = messages.flatMap((messageResponse) =>
+      messageResponse.messages
+        .filter((message) => message.cid && message.cid.isSome)
+        .map((message) => {
+          const job = createIPFSQueueJob(
+            message.block_number.toNumber(),
+            message.msa_id.isNone ? message.provider_msa_id.toString() : message.msa_id.unwrap().toString(),
+            message.provider_msa_id.toString(),
+            messageResponse.schemaId,
+            message.cid.unwrap().toString(),
+            message.index.toNumber(),
+            requestId
+          );
+
+          return {
+            name: `IPFS Job: ${job.key}`,
+            data: job.data,
+            opts: { jobId: job.key },
+          };
+        }),
+    );
+
+    if (jobs && jobs.length > 0) {
+      await queue.addBulk(jobs);
+    }
+  }
+
+}