refactoring

src/dune.ts

@@ -0,0 +1,103 @@
+import axios from "axios";
+import { DUNE_API_BASE_URL, DUNE_API_KEY, DUNE_QUERY_ID_BASE_LATEST_BLOCK } from "./constants";
+import { logger } from "./logger";
+
+export default class DuneApi {
+    async fetchLatestBlock() {
+        logger.info('Fetching latest block');
+        const latestBlockDataRows = await this.executeDuneQueryAndGetResult(DUNE_QUERY_ID_BASE_LATEST_BLOCK);
+        return latestBlockDataRows[0]._col0;
+    }
+
+
+    async executeDuneQueryAndGetResult(queryId: string, blocknumber?: number): Promise<any> {
+        logger.info('Executing query ' + queryId);
+        logger.debug('trigger execution')
+        const executeQueryUrl = `${DUNE_API_BASE_URL}/v1/query/${queryId}/execute`;
+        let body = { "performance": "medium", "query_parameters": {} };
+        if (blocknumber) {
+            body = {
+                "performance": "medium",
+                "query_parameters": {"blocknumber": blocknumber}
+            };
+        }
+        const executeQueryUrlResponse = await axios.post(executeQueryUrl, body, 
+        {
+            headers: {
+                'x-dune-api-key': DUNE_API_KEY
+            }
+        });
+        const executionId = executeQueryUrlResponse.data.execution_id;
+        logger.info(`Execution ID: ${executionId}`);
+
+        // poll state until execution is done
+        const executeStatusUrl = `${DUNE_API_BASE_URL}/v1/execution/${executionId}/status`;
+        let state = 'QUERY_STATE_PENDING';
+        let totalRowCount = 0;
+        while (state === 'QUERY_STATE_PENDING' || state === 'QUERY_STATE_EXECUTING') {
+            const executeStatusUrlResponse = await axios.get(executeStatusUrl, {
+                headers: {
+                    'x-dune-api-key': DUNE_API_KEY
+                }
+            });
+            state = executeStatusUrlResponse.data.state;
+            logger.debug(executeStatusUrlResponse.data);
+            totalRowCount = executeStatusUrlResponse.data?.result_metadata?.total_row_count;
+            logger.debug(`Status: ${state} - ${totalRowCount} rows`);
+            await new Promise(resolve => setTimeout(resolve, 1000));
+        }
+
+        logger.info(`Execution finished - fetching results`);
+
+        const url = `${DUNE_API_BASE_URL}/v1/execution/${executionId}/results`;
+        const rows = [];
+        const limit = 1000;
+        let offset = 0;
+        logger.debug(`Fetching results for execution ${executionId}`);
+        while (offset * limit < totalRowCount) {
+            // fetch results in chunks
+            logger.debug(`Fetching results offset ${offset} limit ${limit}`);
+            const response = await axios.get(`${url}?limit=${limit}&offset=${offset}`, {
+                headers: {
+                    'x-dune-api-key': DUNE_API_KEY
+                }
+            });
+            rows.push(...response.data.result.rows);
+            offset += limit;
+        }
+
+        logger.info(`Fetched ${rows.length} rows`);
+
+        return rows;
+    }
+
+
+    async getLatestResult(queryId: string, blocknumber?: number): Promise<any> {
+        logger.info('Getting latest result for query ' + queryId);
+
+        const url = `${DUNE_API_BASE_URL}/v1/query/${queryId}/results`;
+        const rows = [];
+        const limit = 1000;
+        let offset = 0;
+        let totalRowCount = 0;
+        do {
+            // fetch results in chunks
+            logger.debug(`Fetching results offset ${offset} limit ${limit}`);
+            const response = await axios.get(`${url}?limit=${limit}&offset=${offset}`, {
+                headers: {
+                    'x-dune-api-key': DUNE_API_KEY
+                }
+            });
+            console.log(response.data);
+            if (totalRowCount === 0) {
+                totalRowCount = response.data.result.metadata.total_row_count;
+            }
+            rows.push(...response.data.result.rows);
+            offset += limit;
+        } while (totalRowCount > 0 && offset * limit < totalRowCount)
+
+        logger.info(`Fetched ${rows.length} rows`);
+
+        return rows;
+    }
+}

src/instance_processor.ts

@@ -0,0 +1,89 @@
+import { PrismaClient } from "@prisma/client";
+import { Instance } from "./types/instance";
+import { DecodedLogEntry } from "./types/logdata";
+import { logger } from "./logger";
+import { IInstanceService__factory } from "./generated/contracts/gif";
+
+export default class InstanceProcessor {
+    private prisma: PrismaClient;
+
+    constructor(prisma: PrismaClient) {
+        this.prisma = prisma;
+    }
+
+    async persistInstances(instances: Instance[]): Promise<void> {
+        for (const instance of instances) {
+            await this.prisma.instance.upsert({
+                where: { nftId: instance.nftId as bigint },
+                update: {
+                    instanceAddress: instance.instanceAddress,
+                    modified_blockNumber: instance.modified.blockNumber,
+                    modified_txHash: instance.modified.txHash,
+                    modified_from: instance.modified.from
+                },
+                create: {
+                    nftId: instance.nftId as bigint,
+                    instanceAddress: instance.instanceAddress,
+                    created_blockNumber: instance.created.blockNumber,
+                    created_txHash: instance.created.txHash,
+                    created_from: instance.created.from,
+                    modified_blockNumber: instance.modified.blockNumber,
+                    modified_txHash: instance.modified.txHash,
+                    modified_from: instance.modified.from
+                }
+            });
+        }
+    }
+
+    async processInstanceServiceEvents(instanceEvents: Array<DecodedLogEntry>): Promise<Array<Instance>> {
+        return instanceEvents.map(event => {
+            logger.info(`Processing instance service event ${event.tx_hash} - ${event.event_name} - ${event.data}`);
+            const data = this.decodeIInstanceServiceEvent(event);
+            if (data === null || data === undefined) {
+                logger.error(`Failed to decode event ${event.tx_hash} - ${event.event_name} - ${event.data}`);
+                return null as unknown as Instance;
+            }
+            if (data.name !== 'LogInstanceServiceInstanceCreated') {
+                return null as unknown as Instance;
+            }
+
+            logger.debug(`args: ${JSON.stringify(data.args)}`);
+            const nftId = data.args[0] as BigInt;
+            const instanceAddress = data.args[1] as string;
+            return {
+                nftId,
+                instanceAddress,
+                created: {
+                    blockNumber: event.block_number,
+                    txHash: event.tx_hash,
+                    from: event.tx_from
+                },
+                modified: {
+                    blockNumber: event.block_number,
+                    txHash: event.tx_hash,
+                    from: event.tx_from
+                }
+            } as Instance;
+        }).filter(event => event !== null);
+    }
+
+
+
+    decodeIInstanceServiceEvent(event: DecodedLogEntry) {
+        const topic0 = event.topic0;
+        let topic1 = event.topic1;
+        if (topic1 === null || topic1 === undefined || topic1 === '') {
+            topic1 = '0x';
+        }
+        let topic2 = event.topic2;
+        if (topic2 === null || topic2 === undefined || topic2 === '') {
+            topic2 = '0x';
+        }
+        let topic3 = event.topic3;
+        if (topic3 === null || topic3 === undefined || topic3 === '') {
+            topic3 = '0x';
+        }
+        return IInstanceService__factory.createInterface().parseLog({ topics: [topic0, topic1, topic2, topic3], data: event.data });
+    }
+
+}