fix: shorten token metadata pg notifications (#1143)

Author: rafaelcr

src/datastore/common.ts

@@ -389,7 +389,7 @@ export type DataStoreEventEmitter = StrictEventEmitter<
     addressUpdate: (address: string, blockHeight: number) => void;
     nameUpdate: (info: string) => void;
     tokensUpdate: (contractID: string) => void;
-    tokenMetadataUpdateQueued: (entry: TokenMetadataUpdateInfo) => void;
+    tokenMetadataUpdateQueued: (queueId: number) => void;
   }
 >;
 
@@ -964,6 +964,12 @@ export interface DataStore extends DataStoreEventEmitter {
     offset: number;
   }): Promise<{ results: DbNonFungibleTokenMetadata[]; total: number }>;
 
+  /**
+   * Returns a single entry from the `token_metadata_queue` table.
+   * @param queueId - queue entry id
+   */
+  getTokenMetadataQueueEntry(queueId: number): Promise<FoundOrNot<DbTokenMetadataQueueEntry>>;
+
   getTokenMetadataQueue(
     limit: number,
     excludingEntries: number[]

src/datastore/postgres-notifier.ts

@@ -21,7 +21,7 @@ export type PgAddressNotificationPayload = {
 };
 
 export type PgTokenMetadataNotificationPayload = {
-  entry: DbTokenMetadataQueueEntry;
+  queueId: number;
 };
 
 export type PgNameNotificationPayload = {

src/datastore/postgres-store.ts

@@ -797,7 +797,7 @@ export class PgDataStore
           break;
         case 'tokenMetadataUpdateQueued':
           const metadata = notification.payload as PgTokenMetadataNotificationPayload;
-          this.emit('tokenMetadataUpdateQueued', metadata.entry);
+          this.emit('tokenMetadataUpdateQueued', metadata.queueId);
           break;
       }
     });
@@ -1457,7 +1457,7 @@ export class PgDataStore
       }
       await this.emitAddressTxUpdates(data.txs);
       for (const tokenMetadataQueueEntry of tokenMetadataQueueEntries) {
-        await this.notifier.sendTokenMetadata({ entry: tokenMetadataQueueEntry });
+        await this.notifier.sendTokenMetadata({ queueId: tokenMetadataQueueEntry.queueId });
       }
     }
   }
@@ -3670,7 +3670,7 @@ export class PgDataStore
         `
           SELECT tx_id
           FROM txs
-          WHERE canonical = true AND microblock_canonical = true 
+          WHERE canonical = true AND microblock_canonical = true
           AND tx_id = ANY($1)
           AND block_height = $2
           `,
@@ -4856,6 +4856,31 @@ export class PgDataStore
     );
   }
 
+  async getTokenMetadataQueueEntry(
+    queueId: number
+  ): Promise<FoundOrNot<DbTokenMetadataQueueEntry>> {
+    const result = await this.query(async client => {
+      const queryResult = await client.query<DbTokenMetadataQueueEntryQuery>(
+        `SELECT * FROM token_metadata_queue WHERE queue_id = $1`,
+        [queueId]
+      );
+      return queryResult;
+    });
+    if (result.rowCount === 0) {
+      return { found: false };
+    }
+    const row = result.rows[0];
+    const entry: DbTokenMetadataQueueEntry = {
+      queueId: row.queue_id,
+      txId: bufferToHexPrefixString(row.tx_id),
+      contractId: row.contract_id,
+      contractAbi: JSON.parse(row.contract_abi),
+      blockHeight: row.block_height,
+      processed: row.processed,
+    };
+    return { found: true, result: entry };
+  }
+
   async getTokenMetadataQueue(
     limit: number,
     excludingEntries: number[]
@@ -6484,7 +6509,7 @@ export class PgDataStore
     const validZonefileHash = this.validateZonefileHash(zonefile_hash);
     await client.query(
       `
-        INSERT INTO zonefiles (zonefile, zonefile_hash) 
+        INSERT INTO zonefiles (zonefile, zonefile_hash)
         VALUES ($1, $2)
         `,
       [zonefile, validZonefileHash]
@@ -6880,10 +6905,10 @@ export class PgDataStore
           WHERE owner = $1
           AND block_height <= $2
           AND canonical = true AND microblock_canonical = true
-        )), 
+        )),
 
-      latest_names AS( 
-      ( 
+      latest_names AS(
+      (
         SELECT DISTINCT ON (names.name) names.name, address, registered_at as block_height, tx_index
         FROM names, address_names
         WHERE address_names.name = names.name

src/event-stream/tokens-contract-handler.ts

@@ -748,13 +748,13 @@ export class TokensProcessorQueue {
   /** The entries currently queued for processing in memory, keyed by the queue entry db id. */
   readonly queuedEntries: Map<number, TokenMetadataUpdateInfo> = new Map();
 
-  readonly onTokenMetadataUpdateQueued: (entry: TokenMetadataUpdateInfo) => void;
+  readonly onTokenMetadataUpdateQueued: (queueId: number) => void;
 
   constructor(db: DataStore, chainId: ChainID) {
     this.db = db;
     this.chainId = chainId;
     this.queue = new PQueue({ concurrency: TOKEN_METADATA_PARSING_CONCURRENCY_LIMIT });
-    this.onTokenMetadataUpdateQueued = entry => this.queueHandler(entry);
+    this.onTokenMetadataUpdateQueued = entry => this.queueNotificationHandler(entry);
     this.db.on('tokenMetadataUpdateQueued', this.onTokenMetadataUpdateQueued);
   }
 
@@ -800,6 +800,13 @@ export class TokensProcessorQueue {
     }
   }
 
+  async queueNotificationHandler(queueId: number) {
+    const queueEntry = await this.db.getTokenMetadataQueueEntry(queueId);
+    if (queueEntry.found) {
+      await this.queueHandler(queueEntry.result);
+    }
+  }
+
   async queueHandler(queueEntry: TokenMetadataUpdateInfo) {
     if (
       this.queuedEntries.has(queueEntry.queueId) ||