feat: add tokenId and token to crosschainPayloads (#275)

Fixes #274

Author: filo87

ponder.schema.ts

@@ -1102,6 +1102,7 @@ const CrosschainPayloadColumns = (t: PgColumnsBuilders) => ({
   toCentrifugeId: t.text().notNull(),
   rawData: t.hex().notNull(),
   poolId: t.bigint(),
+  tokenId: t.hex(),
   status: CrosschainPayloadStatus("crosschain_payload_status").notNull(),
   gasLimit: t.bigint(),
   gasPaid: t.bigint(),
@@ -1142,6 +1143,10 @@ export const CrosschainPayloadRelations = relations(CrosschainPayload, ({ one, m
     fields: [CrosschainPayload.poolId],
     references: [Pool.id],
   }),
+  token: one(Token, {
+    fields: [CrosschainPayload.tokenId],
+    references: [Token.id],
+  }),
   adapterParticipations: many(AdapterParticipation, {
     relationName: "adapterParticipations",
   }),
@@ -1162,6 +1167,7 @@ const CrosschainMessageColumns = (t: PgColumnsBuilders) => ({
   id: t.hex().notNull(),
   index: t.integer().notNull().default(0),
   poolId: t.bigint(),
+  tokenId: t.hex(),
   payloadId: t.hex(),
   payloadIndex: t.integer(),
   messageType: t.text().notNull(),
@@ -1200,6 +1206,10 @@ export const CrosschainMessageRelations = relations(CrosschainMessage, ({ one })
     fields: [CrosschainMessage.poolId],
     references: [Pool.id],
   }),
+  token: one(Token, {
+    fields: [CrosschainMessage.tokenId],
+    references: [Token.id],
+  }),
   fromBlockchain: one(Blockchain, {
     fields: [CrosschainMessage.fromCentrifugeId],
     references: [Blockchain.centrifugeId],

src/handlers/gatewayHandlers.ts

@@ -36,13 +36,15 @@ multiMapper("gateway:PrepareMessage", async ({ event, context }) => {
 
   const rawData = `0x${Buffer.from(messageBuffer.toString("hex"))}` as `0x${string}`;
   const data = decodeMessage(messageType, messagePayload, versionIndex);
+  const tokenId = data && "scId" in data ? (data.scId! as `0x${string}`) : null;
 
   const _crosschainMessage = (await CrosschainMessageService.insert(
     context,
     {
       id: messageId,
       index: messageCount,
       poolId: poolId || null,
+      tokenId,
       fromCentrifugeId,
       toCentrifugeId: toCentrifugeId.toString(),
       messageType: messageType,
@@ -77,6 +79,7 @@ multiMapper("gateway:UnderpaidBatch", async ({ event, context }) => {
   });
 
   const poolIdSet = new Set<bigint>();
+  const tokenIdSet = new Set<`0x${string}`>();
   const messages = extractMessagesFromPayload(batch, versionIndex);
   for (const message of messages) {
     const messageBuffer = Buffer.from(message.substring(2), "hex");
@@ -116,12 +119,16 @@ multiMapper("gateway:UnderpaidBatch", async ({ event, context }) => {
     const poolId = "poolId" in data ? BigInt(data.poolId!) : null;
     if (poolId) poolIdSet.add(poolId);
 
+    const tokenId = "scId" in data ? (data.scId as `0x${string}`) : null;
+    if (tokenId) tokenIdSet.add(tokenId);
+
     const _crosschainMessage = (await CrosschainMessageService.insert(
       context,
       {
         id: messageId,
         index: messageIndex,
         poolId,
+        tokenId,
         fromCentrifugeId,
         toCentrifugeId: toCentrifugeId.toString(),
         messageType: messageType,
@@ -142,13 +149,15 @@ multiMapper("gateway:UnderpaidBatch", async ({ event, context }) => {
     return;
   }
   const poolId = Array.from(poolIdSet).pop() ?? null;
+  const tokenId = Array.from(tokenIdSet).pop() ?? null;
 
   const crosschainPayload = (await CrosschainPayloadService.insert(
     context,
     {
       id: payloadId,
       index: payloadIndex,
       poolId,
+      tokenId,
       rawData: batch,
       toCentrifugeId: toCentrifugeId.toString(),
       fromCentrifugeId: fromCentrifugeId,

src/handlers/multiAdapterHandlers.ts

@@ -52,7 +52,7 @@ multiMapper("multiAdapter:SendPayload", async ({ event, context }) => {
     const payloadIndex = await CrosschainPayloadService.count(context, {
       id: payloadId,
     });
-    const poolId = await CrosschainMessageService.linkMessagesToPayload(
+    const [poolId, tokenId] = await CrosschainMessageService.linkMessagesToPayload(
       context,
       event,
       messageIds,
@@ -69,6 +69,7 @@ multiMapper("multiAdapter:SendPayload", async ({ event, context }) => {
         toCentrifugeId: toCentrifugeId.toString(),
         fromCentrifugeId: fromCentrifugeId,
         poolId,
+        tokenId,
         gasLimit,
         gasPaid,
         ...timestamper("prepared", event),

src/services/CrosschainMessageService.ts

@@ -139,9 +139,10 @@ export class CrosschainMessageService extends mixinCommonStatics(
     messageIds: `0x${string}`[],
     payloadId: `0x${string}`,
     payloadIndex: number
-  ) {
+  ): Promise<[poolId: bigint | null, tokenId: `0x${string}` | null]> {
     const crosschainMessageSaves = [];
     const poolIdSet = new Set<bigint>();
+    const tokenIdSet = new Set<`0x${string}`>();
     for (const messageId of messageIds) {
       const crosschainMessages = (await CrosschainMessageService.query(context, {
         id: messageId,
@@ -154,15 +155,17 @@ export class CrosschainMessageService extends mixinCommonStatics(
         continue;
       }
       const crosschainMessage = crosschainMessages.shift()!;
-      const { poolId } = crosschainMessage.read();
+      const { poolId, tokenId } = crosschainMessage.read();
       crosschainMessage.setPayloadId(payloadId, payloadIndex);
       crosschainMessageSaves.push(crosschainMessage.save(event));
       if (poolId) poolIdSet.add(poolId);
+      if (tokenId) tokenIdSet.add(tokenId);
     }
     await Promise.all(crosschainMessageSaves);
     const poolIds = Array.from(poolIdSet);
+    const tokenIds = Array.from(tokenIdSet);
     if (poolIds.length > 1) throw new Error("Multiple pools found among messages");
-    return poolIds.pop() ?? null;
+    return [poolIds.pop() ?? null, tokenIds.pop() ?? null];
   }
 
   /**