indexer: make sink hookable to ensure ordering

packages/indexer/package.json

@@ -58,7 +58,6 @@
     "@apibara/protocol": "workspace:*",
     "@opentelemetry/api": "^1.9.0",
     "consola": "^3.2.3",
-    "eventemitter3": "^5.0.1",
     "hookable": "^5.5.3",
     "klona": "^2.0.6",
     "nice-grpc": "^2.1.8",

packages/indexer/src/indexer.ts

@@ -120,10 +120,10 @@ export async function run<TFilter, TBlock, TRet>(
 
     const sink = sinkArg ?? defaultSink();
 
-    sink.on("write", async ({ data }) => {
+    sink.hook("write", async ({ data }) => {
       await indexer.hooks.callHook("sink:write", { data });
     });
-    sink.on("flush", async () => {
+    sink.hook("flush", async () => {
       await indexer.hooks.callHook("sink:flush");
     });
 

packages/indexer/src/sink.ts

@@ -1,5 +1,5 @@
 import type { Cursor, DataFinality } from "@apibara/protocol";
-import { EventEmitter } from "eventemitter3";
+import { Hookable } from "hookable";
 
 export interface SinkEvents<TData> {
   write({ data }: { data: TData[] }): void;
@@ -13,7 +13,7 @@ export type SinkWriteArgs<TData> = {
   finality: DataFinality;
 };
 
-export abstract class Sink<TData> extends EventEmitter<SinkEvents<TData>> {
+export abstract class Sink<TData> extends Hookable<SinkEvents<TData>> {
   abstract write({
     data,
     cursor,
@@ -24,8 +24,8 @@ export abstract class Sink<TData> extends EventEmitter<SinkEvents<TData>> {
 
 export class DefaultSink<TData = unknown> extends Sink<TData> {
   async write({ data }: SinkWriteArgs<TData>) {
-    this.emit("write", { data });
-    this.emit("flush");
+    await this.callHook("write", { data });
+    await this.callHook("flush");
   }
 }
 

packages/indexer/src/sinks/csv.ts

@@ -34,13 +34,13 @@ export class CsvSink<
   }
 
   async write({ data, endCursor }: SinkWriteArgs<TData>) {
-    this.emit("write", { data });
+    await this.callHook("write", { data });
     // adds a "_cursor" property if "cursorColumn" is not specified by user
     data = this.processCursorColumn(data, endCursor);
     // Insert the data into csv
     await this.insertToCSV(data);
 
-    this.emit("flush");
+    await this.callHook("flush");
   }
 
   private async insertToCSV(data: TData[]) {

packages/indexer/src/sinks/sqlite.ts

@@ -35,12 +35,12 @@ export class SqliteSink<
   }
 
   async write({ data, endCursor }: SinkWriteArgs<TData>) {
-    this.emit("write", { data });
+    await this.callHook("write", { data });
 
     data = this.processCursorColumn(data, endCursor);
     await this.insertJsonArray(data);
 
-    this.emit("flush");
+    await this.callHook("flush");
   }
 
   private async insertJsonArray(data: TData[]) {

packages/indexer/src/testing/vcr.ts

@@ -5,9 +5,9 @@ export class VcrSink<TData> extends Sink<TData> {
   public result: VcrReplayResult<TData>["outputs"] = [];
 
   async write({ data, endCursor }: SinkWriteArgs<TData>) {
-    this.emit("write", { data });
+    await this.callHook("write", { data });
     this.result.push({ data, endCursor });
-    this.emit("flush");
+    await this.callHook("flush");
   }
 }