indexer: add drizzle sink

examples/indexer/package.json

@@ -23,6 +23,8 @@
     "citty": "^0.1.6",
     "consola": "^3.2.3",
     "csv-stringify": "^6.5.0",
+    "drizzle-orm": "^0.33.0",
+    "postgres": "^3.4.4",
     "viem": "^2.12.4"
   },
   "devDependencies": {

examples/indexer/src/main.ts

@@ -6,6 +6,10 @@ import consola from "consola";
 import { createIndexerConfig } from "./indexer";
 
 import "./instrumentation";
+import { drizzle as drizzleSink } from "@apibara/indexer/sinks/drizzle";
+import { pgTable, serial, text, varchar } from "drizzle-orm/pg-core";
+import { drizzle } from "drizzle-orm/postgres-js";
+import postgres from "postgres";
 
 const command = defineCommand({
   meta: {
@@ -33,7 +37,24 @@ const command = defineCommand({
       indexer.options.streamUrl,
     );
 
-    await run(client, indexer);
+    /** TEMPORARY EXAMPLE OF DRIZZLE SINK - WILL BE REMOVED */
+    const users = pgTable("users", {
+      id: serial("id").primaryKey(),
+      fullName: text("full_name"),
+      phone: varchar("phone", { length: 256 }),
+    });
+
+    const pgClient = postgres("your_connection_string");
+    const db = drizzle(pgClient);
+
+    const sink = drizzleSink({ database: db, table: users });
+
+    // Demo of how we can infer type for the transform function
+    async function _transform(): typeof sink.$inferTransform {
+      return [{ id: 1, fullName: "John Doe", phone: "1234567890" }];
+    }
+
+    await run(client, indexer, sink);
   },
 });
 

packages/indexer/build.config.ts

@@ -5,6 +5,7 @@ export default defineBuildConfig({
     "./src/index.ts",
     "./src/sinks/sqlite.ts",
     "./src/sinks/csv.ts",
+    "./src/sinks/drizzle.ts",
     "./src/testing/index.ts",
   ],
   clean: true,

packages/indexer/package.json

@@ -9,6 +9,7 @@
     "./plugins": "./src/plugins/index.ts",
     "./sinks/sqlite": "./src/sinks/sqlite.ts",
     "./sinks/csv": "./src/sinks/csv.ts",
+    "./sinks/drizzle": "./src/sinks/drizzle.ts",
     "./testing": "./src/testing/index.ts"
   },
   "publishConfig": {
@@ -34,6 +35,12 @@
         "require": "./dist/sinks/csv.cjs",
         "default": "./dist/sinks/csv.mjs"
       },
+      "./sinks/drizzle": {
+        "types": "./dist/sinks/drizzle.d.ts",
+        "import": "./dist/sinks/drizzle.mjs",
+        "require": "./dist/sinks/drizzle.cjs",
+        "default": "./dist/sinks/drizzle.mjs"
+      },
       "./testing": {
         "types": "./dist/testing/index.d.ts",
         "import": "./dist/testing/index.mjs",
@@ -56,6 +63,7 @@
     "@types/node": "^20.14.0",
     "better-sqlite3": "^11.1.2",
     "csv-stringify": "^6.5.0",
+    "drizzle-orm": "^0.33.0",
     "unbuild": "^2.0.0",
     "vitest": "^1.6.0"
   },
@@ -71,6 +79,7 @@
   "peerDependencies": {
     "better-sqlite3": "^11.1.2",
     "csv-stringify": "^6.5.0",
+    "drizzle-orm": "^0.33.0",
     "vitest": "^1.6.0"
   }
 }

packages/indexer/src/sinks/drizzle.ts

@@ -0,0 +1,81 @@
+import type {
+  ExtractTablesWithRelations,
+  TablesRelationalConfig,
+} from "drizzle-orm";
+import type {
+  PgDatabase,
+  PgInsertValue,
+  PgQueryResultHKT,
+  PgTable,
+  PgTableWithColumns,
+  TableConfig,
+} from "drizzle-orm/pg-core";
+import { Sink, type SinkWriteArgs } from "../sink";
+
+export type DrizzleSinkOptions<
+  TTableConfig extends TableConfig,
+  TQueryResult extends PgQueryResultHKT,
+  TFullSchema extends Record<string, unknown> = Record<string, never>,
+  TSchema extends
+    TablesRelationalConfig = ExtractTablesWithRelations<TFullSchema>,
+> = {
+  /**
+   * Database instance of drizzle-orm
+   */
+  database: PgDatabase<TQueryResult, TFullSchema, TSchema>;
+  /**
+   * The table where data will be inserted.
+   */
+  table: PgTableWithColumns<TTableConfig>;
+};
+
+export class DrizzleSink<
+  TTableConfig extends TableConfig,
+  TQueryResult extends PgQueryResultHKT,
+  TFullSchema extends Record<string, unknown> = Record<string, never>,
+  TSchema extends
+    TablesRelationalConfig = ExtractTablesWithRelations<TFullSchema>,
+> extends Sink {
+  private _table: PgTableWithColumns<TTableConfig>;
+  private _db: PgDatabase<TQueryResult, TFullSchema, TSchema>;
+
+  public $inferTransform: Promise<PgInsertValue<PgTable<TTableConfig>>[]> =
+    Promise.resolve([]);
+
+  constructor(
+    options: DrizzleSinkOptions<
+      TTableConfig,
+      TQueryResult,
+      TFullSchema,
+      TSchema
+    >,
+  ) {
+    super();
+    const { database, table } = options;
+    this._table = table;
+    this._db = database;
+  }
+
+  async write({ data, endCursor, finality }: SinkWriteArgs) {
+    await this.callHook("write", { data });
+
+    await this._db
+      .insert(this._table)
+      // biome-ignore lint/suspicious/noExplicitAny: <explanation>
+      .values(data as any[]);
+
+    await this.callHook("flush", { endCursor, finality });
+  }
+}
+
+export const drizzle = <
+  TTableConfig extends TableConfig,
+  TQueryResult extends PgQueryResultHKT,
+  TFullSchema extends Record<string, unknown> = Record<string, never>,
+  TSchema extends
+    TablesRelationalConfig = ExtractTablesWithRelations<TFullSchema>,
+>(
+  args: DrizzleSinkOptions<TTableConfig, TQueryResult, TFullSchema, TSchema>,
+) => {
+  return new DrizzleSink(args);
+};