event processor

libs/services/src/event-processor/event.processor.ts

@@ -0,0 +1,148 @@
+import axios from 'axios';
+
+export class EventProcessor {
+ private options: EventProcessorOptions = new EventProcessorOptions();
+
+ async start(options: EventProcessorOptions) {
+ this.options = options;
+
+ if (!options.emitterAddresses && !options.eventIdentifiers) {
+ throw new Error(`No emitter address or identifier set. Could not resolve all the events without filters.`);
+ }
+
+ await this.startProcess(options);
+ }
+
+ private async startProcess(options: EventProcessorOptions) {
+ const lastTimestampFunc = options.getLastProcessedTimestamp;
+ if (!lastTimestampFunc) {
+ throw new Error(`Undefined function for getting the last processed timestamp`);
+ }
+
+ const maxHeight = await lastTimestampFunc();
+ if (!maxHeight) {
+ throw new Error(`Cannot get the last processed timestamp via the provided getLastProcessedTimestamp()`);
+ }
+
+ await this.callElasticsearchEvents();
+ }
+
+ async callElasticsearchEvents() {
+ const url = `${this.options.elasticUrl}/events/_search?scroll=${this.options.scrollTimeout}`;
+ const timestampFunc = this.options.getLastProcessedTimestamp;
+ if (!timestampFunc) {
+ return;
+ }
+
+ const elasticQuery = this.generateElasticsearchQuery(await timestampFunc() ?? -1);
+ const result = await axios.post(url, elasticQuery);
+
+ if (!result.data || !result.data.hits || !result.data.hits || !result.data.hits.hits) {
+ return;
+ }
+
+ const elasticEvents = result.data.hits.hits;
+ const events = elasticEvents.map((e: { _source: any; }) => e._source);
+ await this.handleElasticEvents(events);
+
+ const scrollId = result.data._scroll_id;
+ while (true) {
+ const scrollResult = await axios.post(`${this.options.elasticUrl}/_search/scroll`,
+ {
+ scroll_id: scrollId,
+ });
+
+ const scrollDocuments = scrollResult.data.hits.hits;
+ if (scrollDocuments.length === 0) {
+ break;
+ }
+
+ const scrollEvents = scrollDocuments.map((e: { _source: any; }) => e._source);
+ await this.handleElasticEvents(scrollEvents);
+ }
+ }
+
+ async handleElasticEvents(events: EventSource[]) {
+ if (events.length === 0) {
+ return;
+ }
+
+ const lastTimestamp = events[events.length - 1].timestamp ?? 0;
+
+ const onEventsFunc = this.options.onEventsReceived;
+ if (onEventsFunc) {
+ await onEventsFunc(lastTimestamp, events);
+ }
+
+ const setLatestProcessedTimestampFunc = this.options.setLastProcessedTimestamp;
+ if (setLatestProcessedTimestampFunc) {
+ await setLatestProcessedTimestampFunc(lastTimestamp);
+ }
+ }
+
+ generateElasticsearchQuery(timestamp: number) {
+ return {
+ size: this.options.pageSize,
+ query: {
+ bool: {
+ must: [
+ {
+ terms: {
+ identifier: this.options.eventIdentifiers, // Query by identifiers
+ },
+ },
+ {
+ terms: {
+ address: this.options.emitterAddresses, // Query by addresses
+ },
+ },
+ {
+ range: {
+ timestamp: {
+ gt: `${timestamp}`,
+ },
+ },
+ },
+ ],
+ },
+ },
+ sort: [
+ {
+ timestamp: {
+ order: 'asc', // Sorting by timestamp in ascending order
+ },
+ },
+ ],
+ };
+ }
+}
+
+export class EventProcessorOptions {
+ elasticUrl?: string;
+ emitterAddresses?: string[];
+ eventIdentifiers?: string[];
+ pageSize: number = 10000;
+ scrollTimeout: string = "1m";
+ onEventsReceived?: (highestTimestamp: number, events: EventSource[]) => Promise<void>;
+ getLastProcessedTimestamp?: () => Promise<number | undefined>;
+ setLastProcessedTimestamp?: (nonce: number) => Promise<void>;
+ timeout?: number | undefined;
+
+ constructor(options: Partial<EventProcessorOptions> = {}) {
+ Object.assign(this, options);
+ }
+}
+
+export class EventSource {
+ originalTxHash?: string;
+ logAddress?: string;
+ identifier?: string;
+ address?: string;
+ topics?: string[];
+ shardID?: number;
+ additionalData?: string[];
+ txOrder?: number;
+ txHash?: string;
+ order?: number;
+ timestamp?: number;
+}

libs/services/src/event-processor/test.ts

@@ -0,0 +1,19 @@
+import { EventProcessor, EventProcessorOptions } from './event.processor';
+
+const eventsProcessor = new EventProcessor();
+
+void eventsProcessor.start(new EventProcessorOptions({
+ elasticUrl: 'https://index.multiversx.com',
+ eventIdentifiers: ['ESDTTransfer'],
+ emitterAddresses: ['erd1z08z0svvqs84dnrh5hrm47agcxmch79fslmupzgqcgfdtpp3slwqlna25a'],
+ pageSize: 3,
+ getLastProcessedTimestamp: async () => {
+ await Promise.resolve();
+ return 1726700632;
+ },
+ onEventsReceived: async (highestTimestamp, events) => {
+ await Promise.resolve();
+ console.log(`onEventsReceived -> highestTimestamp: ${highestTimestamp}`);
+ console.log(`onEventsReceived -> events: ${JSON.stringify(events)}`);
+ },
+}));