WIP indexer SSE prototype

packages/protocol-parser/src/encodeLengths.test.ts

@@ -2,6 +2,12 @@ import { describe, expect, it } from "vitest";
 import { encodeLengths } from "./encodeLengths";
 
 describe("encodeLengths", () => {
+  it("can encode empty tuple", () => {
+    expect(encodeLengths([])).toMatchInlineSnapshot(
+      '"0x0000000000000000000000000000000000000000000000000000000000000000"'
+    );
+  });
+
   it("can encode bool key tuple", () => {
     expect(encodeLengths(["0x1234", "0x12345678"])).toMatchInlineSnapshot(
       '"0x0000000000000000000000000000000000000004000000000200000000000006"'

packages/protocol-parser/src/encodeValueArgs.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest";
+import { encodeValueArgs } from "./encodeValueArgs";
+import { stringToHex } from "viem";
+
+describe("encodeValueArgs", () => {
+  it("can encode record value to hex", () => {
+    const valueSchema = {
+      entityId: "bytes32",
+      exists: "bool",
+      playerName: "string",
+      badges: "uint256[]",
+    } as const;
+
+    const result = encodeValueArgs(valueSchema, {
+      entityId: stringToHex("hello", { size: 32 }),
+      exists: true,
+      playerName: "henry",
+      badges: [42n],
+    });
+
+    expect(result).toMatchInlineSnapshot(`
+      {
+        "dynamicData": "0x68656e7279000000000000000000000000000000000000000000000000000000000000002a",
+        "encodedLengths": "0x0000000000000000000000000000000000000020000000000500000000000025",
+        "staticData": "0x68656c6c6f00000000000000000000000000000000000000000000000000000001",
+      }
+    `);
+  });
+
+  it("encodes record when key order of value and valueSchema do not match", () => {
+    const valueSchema = {
+      entityId: "bytes32",
+      playerName: "string",
+      exists: "bool",
+      badges: "uint256[]",
+    } as const;
+
+    const result = encodeValueArgs(valueSchema, {
+      exists: true,
+      playerName: "henry",
+      entityId: stringToHex("hello", { size: 32 }),
+      badges: [42n],
+    });
+
+    expect(result).toMatchInlineSnapshot(`
+      {
+        "dynamicData": "0x68656e7279000000000000000000000000000000000000000000000000000000000000002a",
+        "encodedLengths": "0x0000000000000000000000000000000000000020000000000500000000000025",
+        "staticData": "0x68656c6c6f00000000000000000000000000000000000000000000000000000001",
+      }
+    `);
+  });
+});

packages/protocol-parser/src/encodeValueArgs.ts

@@ -1,4 +1,11 @@
-import { StaticPrimitiveType, DynamicPrimitiveType, isStaticAbiType, isDynamicAbiType } from "@latticexyz/schema-type";
+import {
+  StaticPrimitiveType,
+  DynamicPrimitiveType,
+  isStaticAbiType,
+  isDynamicAbiType,
+  StaticAbiType,
+  DynamicAbiType,
+} from "@latticexyz/schema-type";
 import { concatHex } from "viem";
 import { encodeField } from "./encodeField";
 import { SchemaToPrimitives, ValueArgs, ValueSchema } from "./common";
@@ -8,15 +15,17 @@ export function encodeValueArgs<TSchema extends ValueSchema>(
   valueSchema: TSchema,
   value: SchemaToPrimitives<TSchema>
 ): ValueArgs {
-  const staticFields = Object.values(valueSchema).filter(isStaticAbiType);
-  const dynamicFields = Object.values(valueSchema).filter(isDynamicAbiType);
+  const valueSchemaEntries = Object.entries(valueSchema);
+  const staticFields = valueSchemaEntries.filter(([, type]) => isStaticAbiType(type)) as [string, StaticAbiType][];
+  const dynamicFields = valueSchemaEntries.filter(([, type]) => isDynamicAbiType(type)) as [string, DynamicAbiType][];
+  // TODO: validate <=5 dynamic fields
+  // TODO: validate <=28 total fields
 
-  const values = Object.values(value);
-  const staticValues = values.slice(0, staticFields.length) as readonly StaticPrimitiveType[];
-  const dynamicValues = values.slice(staticFields.length) as readonly DynamicPrimitiveType[];
+  const encodedStaticValues = staticFields.map(([name, type]) => encodeField(type, value[name] as StaticPrimitiveType));
+  const encodedDynamicValues = dynamicFields.map(([name, type]) =>
+    encodeField(type, value[name] as DynamicPrimitiveType)
+  );
 
-  const encodedStaticValues = staticValues.map((value, i) => encodeField(staticFields[i], value));
-  const encodedDynamicValues = dynamicValues.map((value, i) => encodeField(dynamicFields[i], value));
   const encodedLengths = encodeLengths(encodedDynamicValues);
 
   return {

packages/store-indexer/bin/sqlite-indexer.ts

@@ -14,6 +14,7 @@ import { createQueryAdapter } from "../src/sqlite/createQueryAdapter";
 import { isDefined } from "@latticexyz/common/utils";
 import { combineLatest, filter, first } from "rxjs";
 import { parseEnv } from "./parseEnv";
+import { streamHandler } from "./stream";
 
 const env = parseEnv(
   z.object({
@@ -99,5 +100,7 @@ server.register(fastifyTRPCPlugin<AppRouter>, {
   },
 });
 
+server.get("/stream/findAll", streamHandler);
+
 await server.listen({ host: env.HOST, port: env.PORT });
 console.log(`indexer server listening on http://${env.HOST}:${env.PORT}`);

packages/store-indexer/bin/stream.ts

@@ -0,0 +1,87 @@
+import { FastifyReply, FastifyRequest } from "fastify";
+import { parseEnv } from "./parseEnv";
+import { z } from "zod";
+import { drizzle } from "drizzle-orm/better-sqlite3";
+import Database from "better-sqlite3";
+import { buildTable, getTables } from "@latticexyz/store-sync/sqlite";
+import { BaseSQLiteDatabase } from "drizzle-orm/sqlite-core";
+import { hexToString, padHex } from "viem";
+import { decodeDynamicField, encodeValue } from "@latticexyz/protocol-parser";
+import { eq } from "drizzle-orm";
+
+const env = parseEnv(
+  z.object({
+    SQLITE_FILENAME: z.string().default("indexer.db"),
+  })
+);
+
+const database = drizzle(new Database(env.SQLITE_FILENAME)) as BaseSQLiteDatabase<"sync", any>;
+
+// \n and \r are event-terminating characters in SSE spec, so we need to escape them, along with the escape character itself.
+const charsToEscape = ["\x27", "\n", "\r"];
+function escapeForSSE(value: string): string {
+  return value.replaceAll(/[\x27\n\r]/g, (char) => `\0${charsToEscape.indexOf(char)}`);
+}
+// TODO: unescape
+
+export async function streamHandler(req: FastifyRequest, res: FastifyReply): Promise<void> {
+  try {
+    res.raw.writeHead(200, {
+      "Content-Type": "text/event-stream",
+      Connection: "keep-alive",
+      "Cache-Control": "no-cache",
+    });
+
+    const tables = getTables(database);
+    for (const table of tables) {
+      const sqlTable = buildTable(table);
+      const records = database
+        .select({
+          blockNumber: sqlTable.__lastUpdatedBlockNumber,
+          encodedKey: sqlTable.__key,
+          staticData: sqlTable.__staticData,
+          encodedLengths: sqlTable.__encodedLengths,
+          dynamicData: sqlTable.__dynamicData,
+        })
+        .from(sqlTable)
+        .where(eq(sqlTable.__isDeleted, false))
+        .all();
+
+      for (const record of records) {
+        const encodedEvent = encodeValue(
+          {
+            blockNumber: "uint256",
+            address: "address",
+            event: "string",
+            tableId: "bytes32",
+            keyTuple: "bytes32[]",
+            staticData: "bytes",
+            encodedLengths: "bytes32",
+            dynamicData: "bytes",
+          },
+          {
+            blockNumber: record.blockNumber,
+            address: table.address,
+            event: "Store_SetRecord",
+            tableId: table.tableId,
+            keyTuple: decodeDynamicField("bytes32[]", record.encodedKey),
+            staticData: record.staticData ?? "0x",
+            encodedLengths: record.encodedLengths ?? padHex("0x", { size: 32 }),
+            dynamicData: record.dynamicData ?? "0x",
+          }
+        );
+
+        res.raw.write(`data: ${escapeForSSE(hexToString(encodedEvent))}\n`);
+        res.raw.write(`\n`);
+      }
+    }
+
+    res.raw.write(":end\n\n");
+    console.log("end");
+
+    res.raw.end();
+  } catch (error) {
+    console.error(error);
+    throw error;
+  }
+}

packages/store-indexer/package.json

@@ -25,20 +25,19 @@
     "dev": "tsup --watch",
     "lint": "eslint .",
     "start:postgres": "tsx bin/postgres-indexer",
-    "start:postgres:local": "DATABASE_URL=postgres://127.0.0.1/postgres CHAIN_ID=31337 pnpm start:postgres",
-    "start:postgres:testnet": "DATABASE_URL=postgres://127.0.0.1/postgres CHAIN_ID=4242 START_BLOCK=19037160 pnpm start:postgres",
-    "start:postgres:testnet2": "DATABASE_URL=postgres://127.0.0.1/postgres CHAIN_ID=4243 pnpm start:postgres",
+    "start:postgres:local": "DEBUG=mud:store-sync:createStoreSync DATABASE_URL=postgres://127.0.0.1/postgres RPC_HTTP_URL=http://127.0.0.1:8545 pnpm start:postgres",
+    "start:postgres:testnet": "DEBUG=mud:store-sync:createStoreSync DATABASE_URL=postgres://127.0.0.1/postgres RPC_HTTP_URL=https://follower.testnet-chain.linfra.xyz pnpm start:postgres",
     "start:sqlite": "tsx bin/sqlite-indexer",
-    "start:sqlite:local": "SQLITE_FILENAME=anvil.db CHAIN_ID=31337 pnpm start:sqlite",
-    "start:sqlite:testnet": "SQLITE_FILENAME=testnet.db CHAIN_ID=4242 START_BLOCK=19037160 pnpm start:sqlite",
-    "start:sqlite:testnet2": "SQLITE_FILENAME=testnet2.db CHAIN_ID=4243 pnpm start:sqlite",
+    "start:sqlite:local": "DEBUG=mud:store-sync:createStoreSync SQLITE_FILENAME=anvil.db RPC_HTTP_URL=http://127.0.0.1:8545 pnpm start:sqlite",
+    "start:sqlite:testnet": "DEBUG=mud:store-sync:createStoreSync SQLITE_FILENAME=testnet.db RPC_HTTP_URL=https://follower.testnet-chain.linfra.xyz pnpm start:sqlite",
     "test": "tsc --noEmit --skipLibCheck",
     "test:ci": "pnpm run test"
   },
   "dependencies": {
     "@fastify/cors": "^8.3.0",
     "@latticexyz/block-logs-stream": "workspace:*",
     "@latticexyz/common": "workspace:*",
+    "@latticexyz/protocol-parser": "workspace:*",
     "@latticexyz/store": "workspace:*",
     "@latticexyz/store-sync": "workspace:*",
     "@trpc/client": "10.34.0",