feat: apply logs from receipt

packages/block-logs-stream/src/groupLogsByBlockNumber.ts

@@ -5,7 +5,7 @@ type PartialLog = { blockNumber: bigint; logIndex: number };
 
 export type GroupLogsByBlockNumberResult<TLog extends PartialLog> = {
   blockNumber: TLog["blockNumber"];
-  logs: TLog[];
+  logs: readonly TLog[];
 }[];
 
 /**

packages/store-sync/src/createStoreSync.ts

@@ -1,5 +1,5 @@
 import { storeEventsAbi } from "@latticexyz/store";
-import { GetTransactionReceiptErrorType, Hex } from "viem";
+import { GetTransactionReceiptErrorType, Hex, parseEventLogs } from "viem";
 import {
   StorageAdapter,
   StorageAdapterBlock,
@@ -10,7 +10,7 @@ import {
   internalTableIds,
   WaitForTransactionResult,
 } from "./common";
-import { createBlockStream } from "@latticexyz/block-logs-stream";
+import { createBlockStream, groupLogsByBlockNumber } from "@latticexyz/block-logs-stream";
 import {
   filter,
   map,
@@ -199,6 +199,19 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
   let startBlock: bigint | null = null;
   let endBlock: bigint | null = null;
   let lastBlockNumberProcessed: bigint | null = null;
+  let optimisticLogs: readonly StoreEventsLog[] = [];
+
+  async function applyOptimisticLogs(): Promise<readonly StorageAdapterBlock[]> {
+    const blocks = groupLogsByBlockNumber(optimisticLogs).filter(
+      (block) =>
+        block.logs.length > 0 && (lastBlockNumberProcessed == null || block.blockNumber > lastBlockNumberProcessed),
+    );
+    for (const block of blocks) {
+      await storageAdapter(block);
+    }
+    optimisticLogs = blocks.flatMap((block) => block.logs);
+    return blocks;
+  }
 
   const storedBlock$ = combineLatest([startBlock$, latestBlockNumber$]).pipe(
     map(([startBlock, endBlock]) => ({ startBlock, endBlock })),
@@ -219,9 +232,12 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
         storageAdapter,
         logFilter,
       });
-
       return from(storedBlocks);
     }),
+    concatMap(async (block) => {
+      await applyOptimisticLogs();
+      return block;
+    }),
     tap(({ blockNumber, logs }) => {
       debug("stored", logs.length, "logs for block", blockNumber);
       lastBlockNumberProcessed = blockNumber;
@@ -267,6 +283,14 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
 
   // TODO: move to its own file so we can test it, have its own debug instance, etc.
   async function waitForTransaction(tx: Hex): Promise<WaitForTransactionResult> {
+    const receipt = await publicClient.getTransactionReceipt({ hash: tx });
+
+    const logs = parseEventLogs({ abi: storeEventsAbi, logs: receipt.logs });
+    optimisticLogs = [...optimisticLogs, ...logs];
+    await applyOptimisticLogs();
+
+    return receipt;
+
     debug("waiting for tx", tx);
 
     // This currently blocks for async call on each block processed

packages/store-sync/src/fetchAndStoreLogs.ts

@@ -15,6 +15,7 @@ export async function* fetchAndStoreLogs({
   for await (const { logs, toBlock } of fetchLogs(fetchLogsOptions)) {
     const blocks = groupLogsByBlockNumber(logFilter ? logs.filter(logFilter) : logs, toBlock);
     for (const block of blocks) {
+      console.log("applying state update", block.blockNumber);
       await storageAdapter(block);
       yield block;
     }