feat(store-sync,store-indexer): consolidate sync logic, add syncToSqlite

.changeset/smooth-elephants-wave.md

@@ -0,0 +1,9 @@
+---
+"@latticexyz/dev-tools": patch
+"@latticexyz/store-indexer": minor
+"@latticexyz/store-sync": minor
+---
+
+Store sync logic is now consolidated into a `createStoreSync` function exported from `@latticexyz/store-sync`. This simplifies each storage sync strategy to just a simple wrapper around the storage adapter. You can now sync to RECS with `syncToRecs` or SQLite with `syncToSqlite` and PostgreSQL support coming soon.
+
+There are no breaking changes if you were just using `syncToRecs` from `@latticexyz/store-sync` or running the `sqlite-indexer` binary from `@latticexyz/store-indexer`.

e2e/packages/client-vanilla/src/mud/setupNetwork.ts

@@ -18,8 +18,6 @@ export async function setupNetwork() {
     pollingInterval: 1000,
   } as const satisfies ClientConfig;
 
-  console.log("client options", clientOptions);
-
   const publicClient = createPublicClient(clientOptions);
 
   const burnerAccount = createBurnerAccount(networkConfig.privateKey as Hex);

e2e/packages/sync-test/setup/startIndexer.ts

@@ -11,13 +11,17 @@ export function startIndexer(
 ) {
   let resolve: () => void;
   let reject: (reason?: string) => void;
+  const doneSyncing = new Promise<void>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
 
   console.log(chalk.magenta("[indexer]:"), "start syncing");
 
   const proc = execa("pnpm", ["start"], {
     cwd: path.join(__dirname, "..", "..", "..", "..", "packages", "store-indexer"),
     env: {
-      DEBUG: "mud:store-indexer",
+      DEBUG: "mud:*",
       PORT: port.toString(),
       CHAIN_ID: "31337",
       RPC_HTTP_URL: rpcUrl,
@@ -32,31 +36,23 @@ export function startIndexer(
     reject(errorMessage);
   });
 
-  proc.stdout?.on("data", (data) => {
-    const dataString = data.toString();
-    const errors = extractLineContaining("ERROR", dataString).join("\n");
+  function onLog(data: string) {
+    const errors = extractLineContaining("ERROR", data).join("\n");
     if (errors) {
-      console.log(chalk.magenta("[indexer error]:", errors));
-      reject(errors);
-    }
-    console.log(chalk.magentaBright("[indexer]:", dataString));
-  });
-
-  proc.stderr?.on("data", (data) => {
-    const dataString = data.toString();
-    const modeErrors = extractLineContaining("ERROR", dataString).join("\n");
-    if (modeErrors) {
-      const errorMessage = chalk.magenta("[indexer error]:", modeErrors);
+      const errorMessage = chalk.magenta("[indexer error]:", errors);
       console.log(errorMessage);
       reportError(errorMessage);
-      reject(modeErrors);
+      reject(errors);
     }
     if (data.toString().includes("all caught up")) {
       console.log(chalk.magenta("[indexer]:"), "done syncing");
       resolve();
     }
-    console.log(chalk.magentaBright("[indexer ingress]:", dataString));
-  });
+    console.log(chalk.magentaBright("[indexer]:", data));
+  }
+
+  proc.stdout?.on("data", (data) => onLog(data.toString()));
+  proc.stderr?.on("data", (data) => onLog(data.toString()));
 
   function cleanUp() {
     // attempt to clean up sqlite file
@@ -75,10 +71,7 @@ export function startIndexer(
 
   return {
     url: `http://127.0.0.1:${port}`,
-    doneSyncing: new Promise<void>((res, rej) => {
-      resolve = res;
-      reject = rej;
-    }),
+    doneSyncing,
     process: proc,
     kill: () =>
       new Promise<void>((resolve) => {

examples/minimal/packages/client-phaser/src/ui/LoadingScreen/LoadingScreen.tsx

@@ -4,7 +4,8 @@ import { LoadingBar } from "./LoadingBar";
 import { BootScreen } from "./BootScreen";
 import { useComponentValue } from "@latticexyz/react";
 import { useMUD } from "../../store";
-import { SyncStep, singletonEntity } from "@latticexyz/store-sync/recs";
+import { singletonEntity } from "@latticexyz/store-sync/recs";
+import { SyncStep } from "@latticexyz/store-sync";
 
 export const LoadingScreen = () => {
   const {

packages/dev-tools/src/events/EventIcon.tsx

@@ -1,21 +1,22 @@
 import { assertExhaustive } from "@latticexyz/common/utils";
-import { StoreEventsAbiItem } from "@latticexyz/store";
+import { StoreConfig } from "@latticexyz/store";
+import { StorageOperation } from "@latticexyz/store-sync";
 
 type Props = {
-  eventName: StoreEventsAbiItem["name"];
+  type: StorageOperation<StoreConfig>["type"];
 };
 
-export function EventIcon({ eventName }: Props) {
-  switch (eventName) {
-    case "StoreSetRecord":
+export function EventIcon({ type }: Props) {
+  switch (type) {
+    case "SetRecord":
       return <span className="text-green-500 font-bold">=</span>;
-    case "StoreSetField":
+    case "SetField":
       return <span className="text-cyan-500 font-bold">+</span>;
-    case "StoreDeleteRecord":
+    case "DeleteRecord":
       return <span className="text-red-500 font-bold">-</span>;
-    case "StoreEphemeralRecord":
-      return <span className="text-violet-400 font-bold">~</span>;
+    // case "EphemeralRecord":
+    //   return <span className="text-violet-400 font-bold">~</span>;
     default:
-      return assertExhaustive(eventName, `Unexpected event name: ${eventName}`);
+      return assertExhaustive(type, `Unexpected storage operation type: ${type}`);
   }
 }

packages/dev-tools/src/events/StorageOperationsTable.tsx

@@ -23,16 +23,23 @@ export function StorageOperationsTable({ operations }: Props) {
       </thead>
       <tbody className="font-mono text-xs">
         {operations.map((operation) => (
-          <tr key={`${operation.log.transactionHash}:${operation.log.transactionIndex}`} className="hover:bg-blue-800">
+          <tr
+            key={
+              operation.log
+                ? `${operation.log.blockHash}:${operation.log.logIndex}`
+                : `${operation.namespace}:${operation.name}:${serialize(operation.key)}`
+            }
+            className="hover:bg-blue-800"
+          >
             <td className="px-1 whitespace-nowrap overflow-hidden text-ellipsis text-white/40">
-              {operation.log.blockNumber.toString()}
+              {operation.log?.blockNumber.toString()}
             </td>
             <td className="px-1 whitespace-nowrap overflow-hidden text-ellipsis">
               {operation.namespace}:{operation.name}
             </td>
             <td className="px-1 whitespace-nowrap overflow-hidden text-ellipsis">{serialize(operation.key)}</td>
             <td className="px-1 whitespace-nowrap">
-              <EventIcon eventName={operation.log.eventName} />
+              <EventIcon type={operation.type} />
             </td>
             <td className="px-1 whitespace-nowrap overflow-hidden text-ellipsis">
               {operation.type === "SetRecord" ? serialize(operation.value) : null}

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

@@ -7,13 +7,14 @@ import Database from "better-sqlite3";
 import { createPublicClient, fallback, webSocket, http, Transport } from "viem";
 import { createHTTPServer } from "@trpc/server/adapters/standalone";
 import { createAppRouter } from "@latticexyz/store-sync/trpc-indexer";
-import { chainState, schemaVersion } from "@latticexyz/store-sync/sqlite";
-import { createIndexer } from "../src/sqlite/createIndexer";
+import { chainState, schemaVersion, syncToSqlite } from "@latticexyz/store-sync/sqlite";
 import { createStorageAdapter } from "../src/sqlite/createStorageAdapter";
 import type { Chain } from "viem/chains";
 import * as mudChains from "@latticexyz/common/chains";
 import * as chains from "viem/chains";
 import { isNotNull } from "@latticexyz/common/utils";
+import { combineLatest, filter, first } from "rxjs";
+import { debug } from "../src/debug";
 
 const possibleChains = Object.values({ ...mudChains, ...chains }) as Chain[];
 
@@ -89,13 +90,24 @@ try {
   // ignore errors, this is optional
 }
 
-await createIndexer({
+const { latestBlockNumber$, blockStorageOperations$ } = await syncToSqlite({
   database,
   publicClient,
   startBlock,
   maxBlockRange: env.MAX_BLOCK_RANGE,
 });
 
+combineLatest([latestBlockNumber$, blockStorageOperations$])
+  .pipe(
+    filter(
+      ([latestBlockNumber, { blockNumber: lastBlockNumberProcessed }]) => latestBlockNumber === lastBlockNumberProcessed
+    ),
+    first()
+  )
+  .subscribe(() => {
+    console.log("all caught up");
+  });
+
 const server = createHTTPServer({
   middleware: cors(),
   router: createAppRouter(),

packages/store-sync/src/SyncStep.ts

@@ -0,0 +1,6 @@
+export enum SyncStep {
+  INITIALIZE = "initialize",
+  SNAPSHOT = "snapshot",
+  RPC = "rpc",
+  LIVE = "live",
+}

packages/store-sync/src/blockLogsToStorage.test.ts

@@ -120,6 +120,7 @@ describe("blockLogsToStorage", () => {
               "blockNumber": 5448n,
               "operations": [
                 {
+                  "address": "0x5FbDB2315678afecb367f032d93F642f64180aa3",
                   "fieldName": "amount",
                   "fieldValue": 8,
                   "key": {
@@ -173,6 +174,7 @@ describe("blockLogsToStorage", () => {
         "blockNumber": 5448n,
         "operations": [
           {
+            "address": "0x5FbDB2315678afecb367f032d93F642f64180aa3",
             "fieldName": "amount",
             "fieldValue": 8,
             "key": {

packages/store-sync/src/blockLogsToStorage.ts

@@ -202,6 +202,7 @@ export function blockLogsToStorage<TConfig extends StoreConfig = StoreConfig>({
           //       they'll eventually be turned into "events", but unclear if that should translate to client storage operations
           return {
             log,
+            address: getAddress(log.address),
             type: "SetRecord",
             ...tableId,
             key,
@@ -217,6 +218,7 @@ export function blockLogsToStorage<TConfig extends StoreConfig = StoreConfig>({
           >[typeof fieldName];
           return {
             log,
+            address: getAddress(log.address),
             type: "SetField",
             ...tableId,
             key,
@@ -228,6 +230,7 @@ export function blockLogsToStorage<TConfig extends StoreConfig = StoreConfig>({
         if (log.eventName === "StoreDeleteRecord") {
           return {
             log,
+            address: getAddress(log.address),
             type: "DeleteRecord",
             ...tableId,
             key,