Feat/update owner for each cw721 activity ( develop )

migrations/20230830024631_cw721_activity_add_current_owner.ts

@@ -0,0 +1,46 @@
+import { Knex } from 'knex';
+import CW721Activity from '../src/models/cw721_tx';
+import _, { Dictionary } from 'lodash';
+import { CW721_ACTION } from '../src/services/cw721/cw721.service';
+
+export async function up(knex: Knex): Promise<void> {
+  await knex('cw721_activity').update({
+    sender: knex.ref('from'),
+  });
+  await knex.transaction(async (trx) => {
+    const activities = await CW721Activity.query()
+      .joinRelated('event')
+      .orderBy('event.id', 'asc')
+      .transacting(trx);
+    const latestOwners: Dictionary<string> = {};
+    activities.forEach((activity) => {
+      const latestOwner =
+        latestOwners[
+          activity.cw721_contract_id + '_' + activity.cw721_token_id
+        ];
+      if (latestOwner) {
+        activity.from = latestOwner;
+      } else {
+        activity.from = null;
+      }
+      if (
+        activity.action === CW721_ACTION.MINT ||
+        activity.action === CW721_ACTION.TRANSFER ||
+        activity.action === CW721_ACTION.SEND_NFT
+      ) {
+        latestOwners[
+          activity.cw721_contract_id + '_' + activity.cw721_token_id
+        ] = activity.to;
+      }
+    });
+    if (activities.length > 0) {
+      await CW721Activity.query()
+        .insert(activities)
+        .onConflict(['id'])
+        .merge()
+        .transacting(trx);
+    }
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {}

src/common/constant.ts

@@ -238,6 +238,10 @@ export const SERVICE = {
         path: 'v1.ReAssignMsgIndexToEvent',
       },
     },
+    CrawlIBCTaoService: {
+      key: 'CrawlIBCTaoService',
+      name: 'v1.CrawlIBCTaoService',
+    },
     DailyStatisticsService: {
       key: 'DailyStatisticsService',
       name: 'v1.DailyStatisticsService',
@@ -258,10 +262,6 @@ export const SERVICE = {
       key: 'DailyStatsJobsService',
       name: 'v1.DailyStatsJobsService',
     },
-    CrawlIBCTaoService: {
-      key: 'CrawlIBCTaoService',
-      name: 'v1.CrawlIBCTaoService',
-    },
     Cw20ReindexingService: {
       key: 'Cw20ReindexingService',
       name: 'v1.Cw20ReindexingService',

src/models/cw721_tx.ts

@@ -13,23 +13,23 @@ export default class CW721Activity extends BaseModel {
 
   id!: number;
 
-  action?: string;
+  action!: string;
 
   sender?: string;
 
   tx_hash!: string;
 
   cw721_contract_id!: number;
 
-  cw721_token_id?: number;
+  cw721_token_id!: number;
 
   created_at?: Date;
 
   updated_at?: Date;
 
-  from?: string;
+  from!: string | null;
 
-  to?: string;
+  to!: string;
 
   height!: number;
 

src/models/event_attribute.ts

@@ -96,8 +96,6 @@ export class EventAttribute extends BaseModel {
     GRANTER: 'granter',
     GRANTEE: 'grantee',
     FROM: 'from',
-    FEE: 'fee',
-    FEE_PAYER: 'fee_payer',
     CLIENT_ID: 'client_id',
     CLIENT_TYPE: 'client_type',
     CONNECTION_ID: 'connection_id',
@@ -108,6 +106,8 @@ export class EventAttribute extends BaseModel {
     COUNTERPARTY_PORT_ID: 'counterparty_port_id',
     COUNTERPARTY_CHANNEL_ID: 'counterparty_channel_id',
     MINTER: 'minter',
+    FEE: 'fee',
+    FEE_PAYER: 'fee_payer',
     DATA_HEX: 'packet_data_hex',
     SEQUENCE: 'packet_sequence',
     SRC_PORT: 'packet_src_port',

src/services/cw721/cw721.service.ts

@@ -3,7 +3,7 @@ import {
   Action,
   Service,
 } from '@ourparentcenter/moleculer-decorators-extended';
-import _ from 'lodash';
+import _, { Dictionary } from 'lodash';
 import { Context, ServiceBroker } from 'moleculer';
 import { Queue } from 'bullmq';
 import config from '../../../config.json' assert { type: 'json' };
@@ -296,9 +296,12 @@ export default class Cw721HandlerService extends BullableService {
       ),
       'address'
     );
+    const orderedEvents = _.orderBy(cw721Events, ['height'], ['asc']);
+    const cw721Activities: CW721Activity[] = [];
+    const latestOwners: Dictionary<string> = {};
     await knex.transaction(async (trx) => {
-      const queries: any[] = [];
-      cw721Events.forEach((cw721Event) => {
+      // eslint-disable-next-line no-restricted-syntax
+      for (const cw721Event of orderedEvents) {
         // find the cw721 Event's smart contract id
         const cw721Contract =
           cw721ContractDbRecords[cw721Event.contractAddress];
@@ -321,49 +324,85 @@ export default class Cw721HandlerService extends BullableService {
               );
             }
           }
-          queries.push(
-            CW721Activity.query()
-              .insert(
-                CW721Activity.fromJson({
-                  action: cw721Event.action,
-                  sender: cw721Event.sender,
-                  tx_hash: cw721Event.hash,
-                  cw721_contract_id: cw721Contract.id,
-                  cw721_token_id: cw721TokenId,
-                  height: cw721Event.height,
-                  smart_contract_event_id: cw721Event.smart_contract_event_id,
-                  from: getAttributeFrom(
-                    cw721Event.attributes,
-                    EventAttribute.ATTRIBUTE_KEY.SENDER
-                  ),
-                  to:
-                    getAttributeFrom(
-                      cw721Event.attributes,
-                      EventAttribute.ATTRIBUTE_KEY.OWNER
-                    ) ||
-                    getAttributeFrom(
-                      cw721Event.attributes,
-                      EventAttribute.ATTRIBUTE_KEY.RECIPIENT
-                    ),
-                })
-              )
-              .onConflict(['smart_contract_event_id'])
-              .merge()
-              .transacting(trx)
+          // eslint-disable-next-line no-await-in-loop
+          const from = await this.getLatestOwnerForToken(
+            cw721Contract.id,
+            cw721TokenId,
+            latestOwners
           );
+          const to =
+            getAttributeFrom(
+              cw721Event.attributes,
+              EventAttribute.ATTRIBUTE_KEY.OWNER
+            ) ||
+            getAttributeFrom(
+              cw721Event.attributes,
+              EventAttribute.ATTRIBUTE_KEY.RECIPIENT
+            );
+          const cw721Activity = CW721Activity.fromJson({
+            action: cw721Event.action,
+            sender: getAttributeFrom(
+              cw721Event.attributes,
+              EventAttribute.ATTRIBUTE_KEY.SENDER
+            ),
+            tx_hash: cw721Event.hash,
+            cw721_contract_id: cw721Contract.id,
+            cw721_token_id: cw721TokenId,
+            height: cw721Event.height,
+            smart_contract_event_id: cw721Event.smart_contract_event_id,
+            to,
+            from,
+          });
+          // push to update records
+          cw721Activities.push(cw721Activity);
+          // update latestActivity for this token
+          if (
+            cw721Activity.action === CW721_ACTION.MINT ||
+            cw721Activity.action === CW721_ACTION.TRANSFER ||
+            cw721Activity.action === CW721_ACTION.SEND_NFT
+          ) {
+            latestOwners[
+              `${cw721Activity.cw721_contract_id}_${cw721Activity.cw721_token_id}`
+            ] = cw721Activity.to;
+          }
         }
-      });
-      if (queries.length > 0) {
-        await Promise.all(queries) // Once every query is written
-          .then(trx.commit) // Try to execute all of them
-          .catch((e) => {
-            this.logger.error(e);
-            trx.rollback();
-          }); // And rollback in case any of them goes wrong
+      }
+      if (cw721Activities.length > 0) {
+        await CW721Activity.query()
+          .insert(cw721Activities)
+          .onConflict(['smart_contract_event_id'])
+          .merge()
+          .transacting(trx);
       }
     });
   }
 
+  async getLatestOwnerForToken(
+    cw721ContractId: number,
+    cw721TokenId: number | null,
+    latestOwners: Dictionary<string>
+  ) {
+    const latestOwner = latestOwners[`${cw721ContractId}_${cw721TokenId}`];
+    if (latestOwner) {
+      return latestOwner;
+    }
+    const latestActivityDB = await CW721Activity.query()
+      .joinRelated('event')
+      .whereIn('cw721_activity.action', [
+        CW721_ACTION.MINT,
+        CW721_ACTION.TRANSFER,
+        CW721_ACTION.SEND_NFT,
+      ])
+      .andWhere('cw721_contract_id', cw721ContractId)
+      .andWhere('cw721_token_id', cw721TokenId)
+      .orderBy('event.block_height', 'desc')
+      .first();
+    if (latestActivityDB) {
+      return latestActivityDB.to;
+    }
+    return null;
+  }
+
   // handle Instantiate Msgs
   async handleInstantiateMsgs(msgsInstantiate: SmartContractEvent[]) {
     const cw721Contracts: any[] = await SmartContract.query()