fix: subscription

Author: mschead

indexer/src/index.ts

@@ -34,7 +34,6 @@ async function main() {
     }
 
     if (options.streaming) {
-      await initializeDatabase();
       await startStreaming();
     } else if (options.guards) {
       await startGuardsBackfill();

indexer/src/jobs/publisher-job.ts

@@ -1,13 +1,9 @@
 import zod from "zod";
 import { getRequiredEnvString } from "../utils/helpers";
 
-const eventsToPublish: Array<DispatchInfo> = [];
-
 const KADENA_GRAPHQL_URL = getRequiredEnvString("KADENA_GRAPHQL_API_URL");
 const KADENA_GRAPHQL_PORT = getRequiredEnvString("KADENA_GRAPHQL_API_PORT");
 
-const DISPATCH_INTERVAL = 1000; // 1 second
-
 export const dispatchInfoSchema = zod.object({
   hash: zod.string(),
   chainId: zod.string(),
@@ -18,20 +14,7 @@ export const dispatchInfoSchema = zod.object({
 
 export type DispatchInfo = zod.infer<typeof dispatchInfoSchema>;
 
-export function startPublisher() {
-  setInterval(() => {
-    const [first] = eventsToPublish.splice(0, 1);
-    if (first) {
-      dispatch(first);
-    }
-  }, DISPATCH_INTERVAL);
-}
-
-export const addPublishEvents = (events: DispatchInfo[]) => {
-  // eventsToPublish.push(...events);
-};
-
-const dispatch = async (dispatchInfo: DispatchInfo) => {
+export const dispatch = async (dispatchInfo: DispatchInfo) => {
   try {
     const url = `${KADENA_GRAPHQL_URL}:${KADENA_GRAPHQL_PORT}/new-block`;
     const response = await fetch(url, {

indexer/src/services/sync/payload.ts

@@ -25,21 +25,21 @@ export async function processPayloadKey(
   block: BlockAttributes,
   payloadData: any,
   tx?: Transaction,
-) {
+): Promise<EventAttributes[]> {
   const transactions = payloadData.transactions || [];
 
   const transactionPromises = transactions.map((transactionArray: any) =>
     processTransaction(transactionArray, block, tx),
   );
 
-  await Promise.all(transactionPromises);
+  return (await Promise.all(transactionPromises)).flat();
 }
 
 export async function processTransaction(
   transactionArray: any,
   block: BlockAttributes,
   tx?: Transaction,
-) {
+): Promise<EventAttributes[]> {
   const transactionInfo = transactionArray[TRANSACTION_INDEX];
   const receiptInfo = transactionArray[RECEIPT_INDEX];
 
@@ -192,8 +192,11 @@ export async function processTransaction(
     });
 
     await saveGuards(insertedBalances ?? [], tx);
+
+    return eventsAttributes;
   } catch (error) {
     console.error(`Error saving transaction to the database: ${error}`);
+    return [];
   }
 }
 

indexer/src/services/sync/streaming.ts

@@ -1,7 +1,7 @@
 import { processPayloadKey } from "./payload";
-import { getDecoded, getRequiredEnvString } from "../../utils/helpers";
+import { delay, getDecoded, getRequiredEnvString } from "../../utils/helpers";
 import EventSource from "eventsource";
-import { DispatchInfo } from "../../jobs/publisher-job";
+import { dispatch, DispatchInfo } from "../../jobs/publisher-job";
 import { uint64ToInt64 } from "../../utils/int-uint-64";
 import Block, { BlockAttributes } from "../../models/block";
 import { sequelize } from "../../config/database";
@@ -46,13 +46,23 @@ export async function startStreaming() {
 
     const promises = blocksToProcess.map(async (block: any) => {
       try {
-        await saveBlock(block);
+        return saveBlock(block);
       } catch (error) {
         console.error("Error saving block:", error);
       }
     });
 
-    await Promise.all(promises);
+    const res = (await Promise.all(promises)).filter(
+      (r) => r !== null,
+    ) as DispatchInfo[];
+
+    const dispatches = res.map(async (r, index) => {
+      await dispatch(r);
+      await delay(500);
+      console.log("Dispatched block:", index);
+    });
+
+    await Promise.all(dispatches);
     console.log("Done processing blocks: ", blocksToProcess.length);
   }, 1000 * 10);
 
@@ -122,42 +132,27 @@ async function saveBlock(parsedData: any): Promise<DispatchInfo | null> {
       transaction: tx,
     });
 
-    await processPayloadKey(createdBlock, payloadData, tx);
-
-    const txs: Array<{
-      requestKey: string;
-      qualifiedEventNames: string[];
-    }> = transactions.map((t: any) => {
-      const qualifiedEventNames = t[1].events.map((e: any) => {
-        const module = e.module.namespace
-          ? `${e.module.namespace}.${e.module.name}`
-          : e.module.name;
-        const name = e.name;
-        return `${module}.${name}`;
-      });
-      return {
-        requestKey: t[1].reqKey,
-        qualifiedEventNames,
-      };
-    });
+    const eventsCreated = await processPayloadKey(
+      createdBlock,
+      payloadData,
+      tx,
+    );
+
+    const uniqueRequestKeys = new Set(
+      eventsCreated.map((t) => t.requestkey).filter(Boolean),
+    );
 
-    const events = transactions.flatMap((t: any) => t.qualifiedEventNames);
-    const qualifiedEventNamesSet = new Set();
-    const uniqueQualifiedEventNames = events.filter(
-      (qualifiedEventName: any) => {
-        const isDuplicate = qualifiedEventNamesSet.has(qualifiedEventName);
-        qualifiedEventNamesSet.add(qualifiedEventName);
-        return !isDuplicate;
-      },
+    const uniqueQualifiedEventNames = new Set(
+      eventsCreated.map((t) => `${t.module}.${t.name}`).filter(Boolean),
     );
 
     await tx.commit();
     return {
       hash: createdBlock.hash,
       chainId: createdBlock.chainId.toString(),
       height: createdBlock.height,
-      requestKeys: txs.map((t) => t.requestKey),
-      qualifiedEventNames: uniqueQualifiedEventNames,
+      requestKeys: Array.from(uniqueRequestKeys),
+      qualifiedEventNames: Array.from(uniqueQualifiedEventNames),
     };
   } catch (error) {
     await tx.rollback();