Feat/ibc app ( develop )

ci/config.json.ci

@@ -179,5 +179,10 @@
     "key": "crawlIbcTao",
     "millisecondRepeatJob": 2000,
     "blocksPerCall": 100
+  },
+  "crawlIbcApp": {
+    "key": "crawlIbcApp",
+    "millisecondRepeatJob": 2000,
+    "blocksPerCall": 100
   }
 }

config.json

@@ -179,5 +179,10 @@
     "key": "crawlIbcTao",
     "millisecondRepeatJob": 2000,
     "blocksPerCall": 100
+  },
+  "crawlIbcApp": {
+    "key": "crawlIbcApp",
+    "millisecondRepeatJob": 2000,
+    "blocksPerCall": 100
   }
 }

migrations/20230810094108_create_ibc_tx_model.ts

@@ -0,0 +1,23 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  await knex.schema.createTable('ibc_message', (table) => {
+    table.increments();
+    table.integer('transaction_message_id').index();
+    table.string('src_channel_id').notNullable().index();
+    table.string('src_port_id').notNullable().index();
+    table.string('dst_channel_id').notNullable().index();
+    table.string('dst_port_id').notNullable().index();
+    table.string('type').notNullable().index();
+    table.integer('sequence').notNullable().index();
+    table.string('sequence_key').notNullable().index();
+    table.jsonb('data');
+    table
+      .foreign('transaction_message_id')
+      .references('transaction_message.id');
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  await knex.schema.dropTableIfExists('ibc_message');
+}

src/common/constant.ts

@@ -75,6 +75,7 @@ export const BULL_JOB_NAME = {
   CRAWL_GENESIS_IBC_TAO: 'crawl:genesis-ibc-tao',
   REINDEX_CW20_CONTRACT: 'reindex:cw20-contract',
   REINDEX_CW20_HISTORY: 'reindex:cw20-history',
+  CRAWL_IBC_APP: 'crawl:ibc-app',
 };
 
 export const SERVICE = {
@@ -264,6 +265,10 @@ export const SERVICE = {
         path: 'v1.Cw20ReindexingService.reindexing',
       },
     },
+    CrawlIBCAppService: {
+      key: 'CrawlIBCAppService',
+      name: 'v1.CrawlIBCAppService',
+    },
   },
 };
 

src/models/event.ts

@@ -20,6 +20,8 @@ export class Event extends BaseModel {
 
   source!: string;
 
+  attributes!: EventAttribute[];
+
   static get tableName() {
     return 'event';
   }

src/models/event_attribute.ts

@@ -108,6 +108,12 @@ export class EventAttribute extends BaseModel {
     COUNTERPARTY_PORT_ID: 'counterparty_port_id',
     COUNTERPARTY_CHANNEL_ID: 'counterparty_channel_id',
     MINTER: 'minter',
+    DATA_HEX: 'packet_data_hex',
+    SEQUENCE: 'packet_sequence',
+    SRC_PORT: 'packet_src_port',
+    SRC_CHANNEL: 'packet_src_channel',
+    DST_PORT: 'packet_dst_port',
+    DST_CHANNEL: 'packet_dst_channel',
   };
 
   static ATTRIBUTE_COMPOSITE_KEY = {

src/models/ibc_message.ts

@@ -0,0 +1,96 @@
+/* eslint-disable import/no-cycle */
+import { Model } from 'objection';
+import BaseModel from './base';
+import { IbcChannel } from './ibc_channel';
+import { TransactionMessage } from './transaction_message';
+
+export class IbcMessage extends BaseModel {
+  [relation: string]: any | any[];
+
+  id!: number;
+
+  transaction_message_id!: number;
+
+  src_channel_id!: string;
+
+  src_port_id!: string;
+
+  dst_channel_id!: string;
+
+  dst_port_id!: string;
+
+  type!: string;
+
+  sequence!: number;
+
+  sequence_key!: string;
+
+  data!: any;
+
+  static get tableName() {
+    return 'ibc_message';
+  }
+
+  static get jsonSchema() {
+    return {
+      type: 'object',
+      required: [
+        'transaction_message_id',
+        'src_channel_id',
+        'src_port_id',
+        'dst_channel_id',
+        'dst_port_id',
+        'type',
+        'sequence',
+        'sequence_key',
+      ],
+      properties: {
+        transaction_message_id: { type: 'number' },
+        src_channel_id: { type: 'string' },
+        src_port_id: { type: 'string' },
+        dst_channel_id: { type: 'string' },
+        dst_port_id: { type: 'string' },
+        type: { type: 'string' },
+        sequence: { type: 'number' },
+        sequence_key: { type: 'string' },
+        status: { type: 'boolean' },
+      },
+    };
+  }
+
+  static get relationMappings() {
+    return {
+      message: {
+        relation: Model.BelongsToOneRelation,
+        modelClass: TransactionMessage,
+        join: {
+          from: 'ibc_message.transaction_message_id',
+          to: 'transaction_message.id',
+        },
+      },
+      src_channel: {
+        relation: Model.BelongsToOneRelation,
+        modelClass: IbcChannel,
+        join: {
+          from: 'ibc_message.src_channel_id',
+          to: 'ibc_channel.id',
+        },
+      },
+      dst_channel: {
+        relation: Model.BelongsToOneRelation,
+        modelClass: IbcChannel,
+        join: {
+          from: 'ibc_message.dst_channel_id',
+          to: 'ibc_channel.id',
+        },
+      },
+    };
+  }
+
+  static EVENT_TYPE = {
+    SEND_PACKET: 'send_packet',
+    RECV_PACKET: 'recv_packet',
+    ACKNOWLEDGE_PACKET: 'acknowledge_packet',
+    TIMEOUT_PACKET: 'timeout_packet',
+  };
+}

src/models/index.ts

@@ -28,3 +28,4 @@ export * from './cw20_total_holder_stats';
 export * from './ibc_client';
 export * from './ibc_connection';
 export * from './ibc_channel';
+export * from './ibc_message';

src/services/api-gateways/api_gateway.service.ts

@@ -39,6 +39,12 @@ import { bullBoardMixin } from '../../mixins/bullBoard/bullBoard.mixin';
         mappingPolicy: 'restrict', // allow action called with exact method
         whitelist: ['v1.cw20-admin.*'],
       },
+      {
+        path: '/admin',
+        autoAliases: true, // allow generate rest info (GET/PUT/POST...) in the services
+        mappingPolicy: 'restrict', // allow action called with exact method
+        whitelist: ['v1.cw20-admin.*'],
+      },
     ],
     // empty cors object will have moleculer to generate handler for preflight request and CORS header which allow all origin
     cors: {},

src/services/ibc/crawl_ibc_app.service.ts

@@ -0,0 +1,125 @@
+import { fromHex, fromUtf8 } from '@cosmjs/encoding';
+import { Service } from '@ourparentcenter/moleculer-decorators-extended';
+import { Knex } from 'knex';
+import { ServiceBroker } from 'moleculer';
+import knex from '../../common/utils/db_connection';
+import config from '../../../config.json' assert { type: 'json' };
+import BullableService, { QueueHandler } from '../../base/bullable.service';
+import { BULL_JOB_NAME, SERVICE } from '../../common';
+import { getAttributeFrom } from '../../common/utils/smart_contract';
+import {
+  BlockCheckpoint,
+  Event,
+  EventAttribute,
+  IbcMessage,
+} from '../../models';
+
+@Service({
+  name: SERVICE.V1.CrawlIBCAppService.key,
+  version: 1,
+})
+export default class CrawlIbcAppService extends BullableService {
+  public constructor(public broker: ServiceBroker) {
+    super(broker);
+  }
+
+  @QueueHandler({
+    queueName: BULL_JOB_NAME.CRAWL_IBC_APP,
+    jobName: BULL_JOB_NAME.CRAWL_IBC_APP,
+  })
+  public async crawlIbcApp(): Promise<void> {
+    const [startHeight, endHeight, updateBlockCheckpoint] =
+      await BlockCheckpoint.getCheckpoint(
+        BULL_JOB_NAME.CRAWL_IBC_APP,
+        [BULL_JOB_NAME.CRAWL_IBC_TAO],
+        config.crawlIbcTao.key
+      );
+    this.logger.info(
+      `Handle IBC/APP, startHeight: ${startHeight}, endHeight: ${endHeight}`
+    );
+    if (startHeight > endHeight) return;
+    const events = await Event.query()
+      .withGraphFetched('attributes')
+      .joinRelated('message')
+      .select('event.id', 'event.type', 'message.id as message_id')
+      .whereIn('event.type', [
+        IbcMessage.EVENT_TYPE.ACKNOWLEDGE_PACKET,
+        IbcMessage.EVENT_TYPE.RECV_PACKET,
+        IbcMessage.EVENT_TYPE.SEND_PACKET,
+        IbcMessage.EVENT_TYPE.TIMEOUT_PACKET,
+      ])
+      .andWhere('event.block_height', '>', startHeight)
+      .andWhere('event.block_height', '<=', endHeight)
+      .orderBy('event.id');
+    await knex.transaction(async (trx) => {
+      await this.handleIbcMessage(events, trx);
+      updateBlockCheckpoint.height = endHeight;
+      await BlockCheckpoint.query()
+        .transacting(trx)
+        .insert(updateBlockCheckpoint)
+        .onConflict('job_name')
+        .merge();
+    });
+  }
+
+  async handleIbcMessage(events: Event[], trx: Knex.Transaction) {
+    const ibcMessage = events.map((event) => {
+      const srcChannel = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.SRC_CHANNEL
+      );
+      const srcPort = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.SRC_PORT
+      );
+      const dstChannel = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.DST_CHANNEL
+      );
+      const dstPort = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.DST_PORT
+      );
+      const sequence = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.SEQUENCE
+      );
+      const dataHex = getAttributeFrom(
+        event.attributes,
+        EventAttribute.ATTRIBUTE_KEY.DATA_HEX
+      );
+      return IbcMessage.fromJson({
+        transaction_message_id: event.message_id,
+        src_channel_id: srcChannel,
+        src_port_id: srcPort,
+        dst_channel_id: dstChannel,
+        dst_port_id: dstPort,
+        type: event.type,
+        sequence,
+        sequence_key: `${srcChannel}.${srcPort}.${dstChannel}.${dstPort}.${sequence}`,
+        data: dataHex ? fromUtf8(fromHex(dataHex)) : null,
+      });
+    });
+    if (ibcMessage.length > 0) {
+      await IbcMessage.query().insert(ibcMessage).transacting(trx);
+    }
+  }
+
+  async _start(): Promise<void> {
+    await this.createJob(
+      BULL_JOB_NAME.CRAWL_IBC_APP,
+      BULL_JOB_NAME.CRAWL_IBC_APP,
+      {},
+      {
+        removeOnComplete: true,
+        removeOnFail: {
+          count: 3,
+        },
+        repeat: {
+          every: config.crawlIbcApp.millisecondRepeatJob,
+        },
+      }
+    );
+    return super._start();
+  }
+}