feat(store-sync): add support for live sync from indexer

.changeset/great-dragons-sit.md

@@ -0,0 +1,5 @@
+---
+"@latticexyz/store-sync": patch
+---
+
+Added support for streaming logs from the indexer.

packages/store-sync/src/createStoreSync.ts

@@ -27,11 +27,13 @@ import {
   combineLatest,
   scan,
   mergeMap,
+  throwError,
 } from "rxjs";
 import { debug as parentDebug } from "./debug";
 import { SyncStep } from "./SyncStep";
 import { bigIntMax, chunk, isDefined, waitForIdle } from "@latticexyz/common/utils";
 import { getSnapshot } from "./getSnapshot";
+import { fromEventSource } from "./fromEventSource";
 import { fetchAndStoreLogs } from "./fetchAndStoreLogs";
 
 const debug = parentDebug.extend("createStoreSync");
@@ -61,7 +63,7 @@ export async function createStoreSync({
   maxBlockRange,
   initialState,
   initialBlockLogs,
-  indexerUrl,
+  indexerUrl: indexerUrlInput,
 }: CreateStoreSyncOptions): Promise<SyncResult> {
   const filters: SyncFilter[] =
     initialFilters.length || tableIds.length
@@ -78,9 +80,17 @@ export async function createStoreSync({
         )
     : undefined;
 
-  const initialBlockLogs$ = defer(async (): Promise<StorageAdapterBlock | undefined> => {
-    const chainId = publicClient.chain?.id ?? (await publicClient.getChainId());
+  const indexerUrl =
+    indexerUrlInput !== false
+      ? indexerUrlInput ??
+        (publicClient.chain && "indexerUrl" in publicClient.chain && typeof publicClient.chain.indexerUrl === "string"
+          ? publicClient.chain.indexerUrl
+          : undefined)
+      : undefined;
+
+  const chainId = publicClient.chain?.id ?? (await publicClient.getChainId());
 
+  const initialBlockLogs$ = defer(async (): Promise<StorageAdapterBlock | undefined> => {
     onProgress?.({
       step: SyncStep.SNAPSHOT,
       percentage: 0,
@@ -95,15 +105,7 @@ export async function createStoreSync({
       filters,
       initialState,
       initialBlockLogs,
-      indexerUrl:
-        indexerUrl !== false
-          ? indexerUrl ??
-            (publicClient.chain &&
-            "indexerUrl" in publicClient.chain &&
-            typeof publicClient.chain.indexerUrl === "string"
-              ? publicClient.chain.indexerUrl
-              : undefined)
-          : undefined,
+      indexerUrl,
     });
 
     onProgress?.({
@@ -199,7 +201,31 @@ export async function createStoreSync({
   let endBlock: bigint | null = null;
   let lastBlockNumberProcessed: bigint | null = null;
 
-  const storedBlock$ = combineLatest([startBlock$, latestBlockNumber$]).pipe(
+  const storedIndexerLogs$ = indexerUrl
+    ? startBlock$.pipe(
+        mergeMap((startBlock) => {
+          const url = new URL(
+            `api/logs-live?${new URLSearchParams({
+              input: JSON.stringify({ chainId, address, filters }),
+              block_num: startBlock.toString(),
+              include_tx_hash: "true",
+            })}`,
+            indexerUrl,
+          );
+          return fromEventSource<string>(url);
+        }),
+        map((messageEvent) => {
+          const parsedBlock = JSON.parse(messageEvent.data);
+          return { ...parsedBlock, blockNumber: BigInt(parsedBlock.blockNumber) } as StorageAdapterBlock;
+        }),
+        concatMap(async (block) => {
+          await storageAdapter(block);
+          return block;
+        }),
+      )
+    : throwError(() => new Error("No indexer URL provided"));
+
+  const storedRpcLogs$ = combineLatest([startBlock$, latestBlockNumber$]).pipe(
     map(([startBlock, endBlock]) => ({ startBlock, endBlock })),
     tap((range) => {
       startBlock = range.startBlock;
@@ -215,13 +241,21 @@ export async function createStoreSync({
           ? bigIntMax(range.startBlock, lastBlockNumberProcessed + 1n)
           : range.startBlock,
         toBlock: range.endBlock,
-        storageAdapter,
         logFilter,
+        storageAdapter,
       });
 
       return from(storedBlocks);
     }),
-    tap(({ blockNumber, logs }) => {
+  );
+
+  const storedBlock$ = storedIndexerLogs$.pipe(
+    catchError((e) => {
+      debug("failed to stream logs from indexer:", e.message);
+      debug("falling back to streaming logs from RPC");
+      return storedRpcLogs$;
+    }),
+    tap(async ({ logs, blockNumber }) => {
       debug("stored", logs.length, "logs for block", blockNumber);
       lastBlockNumberProcessed = blockNumber;
 

packages/store-sync/src/fromEventSource.ts

@@ -0,0 +1,10 @@
+import { Observable } from "rxjs";
+
+export function fromEventSource<T>(url: string | URL): Observable<MessageEvent<T>> {
+  return new Observable<MessageEvent>((subscriber) => {
+    const eventSource = new EventSource(url);
+    eventSource.onmessage = (ev): void => subscriber.next(ev);
+    eventSource.onerror = (): void => subscriber.error(new Error("Event source closed: " + url));
+    return () => eventSource.close();
+  });
+}

packages/store-sync/src/index.ts

@@ -7,3 +7,4 @@ export * from "./logToTable";
 export * from "./tablesWithRecordsToLogs";
 export * from "./tableToLog";
 export * from "./recordToLog";
+export * from "./logToRecord";