feat: handle pending blocks

change/@apibara-indexer-c30e94ee-538a-48f6-b0b1-66d2783ae188.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "indexer: do not call factory for pending blocks",
+  "packageName": "@apibara/indexer",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

change/@apibara-plugin-drizzle-5e0243be-f427-4e3b-b06e-f3e7bec2af3b.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "plugin-drizzle: handle pending blocks",
+  "packageName": "@apibara/plugin-drizzle",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

change/@apibara-plugin-mongo-5e9a70b3-021c-462d-9865-f886dc09b8ca.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "plugin-mongo: handle pending blocks",
+  "packageName": "@apibara/plugin-mongo",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

change/@apibara-plugin-sqlite-e6c5f565-8996-4a19-857e-54031f098166.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "plugin-sqlite: handle pending blocks",
+  "packageName": "@apibara/plugin-sqlite",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

packages/indexer/src/indexer.ts

@@ -292,7 +292,7 @@ export async function run<TFilter, TBlock>(
               let block: TBlock | null;
 
               // when factory mode
-              if (isFactoryMode) {
+              if (isFactoryMode && finality !== "pending") {
                 assert(indexer.options.factory !== undefined);
 
                 const [factoryBlock, mainBlock] = blocks;

packages/plugin-drizzle/src/index.ts

@@ -128,6 +128,7 @@ export function drizzleStorage<
     let tableNames: string[] = [];
     let indexerId = "";
     const alwaysReindex = process.env["APIBARA_ALWAYS_REINDEX"] === "true";
+    let prevFinality: DataFinality | undefined;
 
     try {
       tableNames = Object.values((schema as TSchema) ?? db._.schema ?? {}).map(
@@ -299,7 +300,8 @@ export function drizzleStorage<
     indexer.hooks.hook("handler:middleware", async ({ use }) => {
       use(async (context, next) => {
         try {
-          const { endCursor, finality } = context as {
+          const { endCursor, finality, cursor } = context as {
+            cursor: Cursor;
             endCursor: Cursor;
             finality: DataFinality;
           };
@@ -315,6 +317,11 @@ export function drizzleStorage<
               TSchema
             >;
 
+            if (prevFinality === "pending") {
+              // invalidate if previous block's finality was "pending"
+              await invalidate(tx, cursor, idColumn, indexerId);
+            }
+
             if (finality !== "finalized") {
               await registerTriggers(
                 tx,
@@ -328,13 +335,15 @@ export function drizzleStorage<
             await next();
             delete context[DRIZZLE_PROPERTY];
 
-            if (enablePersistence) {
+            if (enablePersistence && finality !== "pending") {
               await persistState({
                 tx,
                 endCursor,
                 indexerId,
               });
             }
+
+            prevFinality = finality;
           });
 
           if (finality !== "finalized") {

packages/plugin-drizzle/tests/helper.ts

@@ -6,7 +6,7 @@ import { DrizzleStorageError } from "../src/utils";
 export const testTable = pgTable("test", {
   id: serial("id").primaryKey(),
   blockNumber: integer("block_number").notNull(),
-  key: text("key"),
+  key: text("key").unique(),
   count: integer("count"),
   data: text("data"),
   createdAt: timestamp("created_at"),
@@ -41,7 +41,7 @@ export async function migratePgliteDb(db: PgLiteDb) {
         CREATE TABLE IF NOT EXISTS test (
           id SERIAL PRIMARY KEY,
           block_number INTEGER NOT NULL,
-          key TEXT,
+          key TEXT UNIQUE,
           count INTEGER,
           data TEXT,
           created_at TIMESTAMP