|
| 1 | +import { |
| 2 | + catchError, |
| 3 | + combineLatest, |
| 4 | + concatMap, |
| 5 | + from, |
| 6 | + map, |
| 7 | + mergeMap, |
| 8 | + Observable, |
| 9 | + of, |
| 10 | + Subject, |
| 11 | + tap, |
| 12 | + throwError, |
| 13 | + switchMap, |
| 14 | + merge, |
| 15 | + filter, |
| 16 | + startWith, |
| 17 | +} from "rxjs"; |
| 18 | +import { StorageAdapterBlock, StoreEventsLog, SyncFilter } from "./common"; |
| 19 | +import { watchLogs } from "./watchLogs"; |
| 20 | +import { Hex } from "viem"; |
| 21 | +import { fromEventSource } from "./fromEventSource"; |
| 22 | +import { isLogsApiResponse } from "./indexer-client/isLogsApiResponse"; |
| 23 | +import { toStorageAdapterBlock } from "./indexer-client/toStorageAdapterBlock"; |
| 24 | +import { fetchAndStoreLogs } from "./fetchAndStoreLogs"; |
| 25 | +import { storeEventsAbi } from "@latticexyz/store"; |
| 26 | +import { bigIntMax, isDefined } from "@latticexyz/common/utils"; |
| 27 | +import { getRpcClient, GetRpcClientOptions } from "@latticexyz/block-logs-stream"; |
| 28 | +import { debug } from "./debug"; |
| 29 | + |
| 30 | +type PreconfirmedBlockStreamOptions = GetRpcClientOptions & { |
| 31 | + fromBlock: bigint; |
| 32 | + preconfirmedLogsUrl: string; |
| 33 | + indexerUrl?: string; |
| 34 | + chainId: number; |
| 35 | + address?: Hex; |
| 36 | + filters: SyncFilter[]; |
| 37 | + latestBlockNumber$: Observable<bigint>; |
| 38 | + maxBlockRange?: bigint; |
| 39 | +}; |
| 40 | + |
| 41 | +export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptions): Observable<StorageAdapterBlock> { |
| 42 | + const recreatePreconfirmedStream$ = new Subject<void>(); |
| 43 | + const recreateLatestStream$ = new Subject<void>(); |
| 44 | + |
| 45 | + let restartBlockNumber = opts.fromBlock; |
| 46 | + let initialCatchUpBlockNumber: bigint | undefined = undefined; |
| 47 | + getRpcClient(opts) |
| 48 | + .request({ method: "eth_blockNumber" }) |
| 49 | + .then((blockNumber) => { |
| 50 | + console.log("initial catch up block number", BigInt(blockNumber)); |
| 51 | + initialCatchUpBlockNumber = BigInt(blockNumber); |
| 52 | + }); |
| 53 | + |
| 54 | + const latestBlock$ = recreateLatestStream$.pipe( |
| 55 | + startWith(undefined), |
| 56 | + switchMap(() => |
| 57 | + createLatestBlockStream({ ...opts, fromBlock: restartBlockNumber }).pipe( |
| 58 | + catchError((e) => { |
| 59 | + debug("Error in latest block stream, recreating", e); |
| 60 | + recreateLatestStream$.next(); |
| 61 | + return throwError(() => e); |
| 62 | + }), |
| 63 | + ), |
| 64 | + ), |
| 65 | + ); |
| 66 | + |
| 67 | + let processedBlockLogs: { [blockNumber: string]: { [logIndex: number]: boolean } } = {}; |
| 68 | + let preconfirmedLogsState: "initializing" | "initialized" | "waiting" = "waiting"; |
| 69 | + const preconfirmedLogs$ = recreatePreconfirmedStream$.pipe( |
| 70 | + tap(() => { |
| 71 | + debug("initializing preconfirmed logs stream"); |
| 72 | + preconfirmedLogsState = "initializing"; |
| 73 | + processedBlockLogs = {}; |
| 74 | + }), |
| 75 | + switchMap(() => |
| 76 | + watchLogs({ |
| 77 | + ...opts, |
| 78 | + url: opts.preconfirmedLogsUrl, |
| 79 | + fromBlock: restartBlockNumber, |
| 80 | + }).logs$.pipe( |
| 81 | + catchError((e) => { |
| 82 | + debug("Error in preconfirmed logs stream, recreating", e); |
| 83 | + recreatePreconfirmedStream$.next(); |
| 84 | + return throwError(() => e); |
| 85 | + }), |
| 86 | + ), |
| 87 | + ), |
| 88 | + tap((block) => { |
| 89 | + debug("preconfirmed block", block.blockNumber, "with", block.logs.length, "logs"); |
| 90 | + preconfirmedLogsState = "initialized"; |
| 91 | + restartBlockNumber = block.blockNumber; |
| 92 | + const seenLogs = (processedBlockLogs[String(block.blockNumber)] ??= {}); |
| 93 | + block.logs.forEach((log) => { |
| 94 | + seenLogs[log.logIndex!] = true; |
| 95 | + }); |
| 96 | + debug("got preconfirmed block", block.blockNumber, "with", block.logs.length, "logs"); |
| 97 | + }), |
| 98 | + ); |
| 99 | + |
| 100 | + const missingLogs$ = latestBlock$.pipe( |
| 101 | + map((block) => { |
| 102 | + const missingBlock = processedBlockLogs[String(block.blockNumber)] == null; |
| 103 | + const seenLogs = processedBlockLogs[String(block.blockNumber)] ?? {}; |
| 104 | + const missingLogs = block.logs.filter((log) => !seenLogs[log.logIndex!]); |
| 105 | + delete processedBlockLogs[String(block.blockNumber)]; |
| 106 | + restartBlockNumber = block.blockNumber + 1n; |
| 107 | + |
| 108 | + debug( |
| 109 | + "got latest block", |
| 110 | + block.blockNumber, |
| 111 | + "with", |
| 112 | + block.logs.length, |
| 113 | + "logs (", |
| 114 | + missingBlock ? "missing block," : "block seen,", |
| 115 | + `${missingLogs.length} new logs`, |
| 116 | + ")", |
| 117 | + ); |
| 118 | + |
| 119 | + if (preconfirmedLogsState === "waiting") { |
| 120 | + // Once the initial catch up block is reached, we can start the preconfirmed logs stream |
| 121 | + if ( |
| 122 | + initialCatchUpBlockNumber != null && // initial catch up block fetched |
| 123 | + block.blockNumber >= initialCatchUpBlockNumber // initial catch up block reached |
| 124 | + ) { |
| 125 | + debug("initial catch up block number", initialCatchUpBlockNumber, "reached, creating preconfirmed stream"); |
| 126 | + recreatePreconfirmedStream$.next(); |
| 127 | + } |
| 128 | + // While the preconfirmed logs stream is waiting, pass the block through |
| 129 | + return block; |
| 130 | + } |
| 131 | + |
| 132 | + // While the preconfirmed logs stream is initializing, don't recreate it and pass the block through |
| 133 | + if (preconfirmedLogsState === "initializing") { |
| 134 | + debug("preconfirmed logs stream is initializing, not recreating"); |
| 135 | + return block; |
| 136 | + } |
| 137 | + |
| 138 | + // If the preconfirmed logs stream is initialized but there are missing logs, recreate it and pass the block through. |
| 139 | + // Pass all logs from this block, not just the missing ones, to make sure they appear in the right order. |
| 140 | + if (preconfirmedLogsState === "initialized" && (missingLogs.length > 0 || missingBlock)) { |
| 141 | + debug("missing logs found in latest block", block.blockNumber, "recreating preconfirmed stream", { |
| 142 | + missingLogs: missingLogs.length, |
| 143 | + missingBlock, |
| 144 | + }); |
| 145 | + recreatePreconfirmedStream$.next(); |
| 146 | + return block; |
| 147 | + } |
| 148 | + |
| 149 | + debug("no missing logs found in latest block", block.blockNumber, "not recreating preconfirmed stream"); |
| 150 | + return; |
| 151 | + }), |
| 152 | + filter(isDefined), |
| 153 | + ); |
| 154 | + |
| 155 | + return merge(preconfirmedLogs$, missingLogs$); |
| 156 | +} |
| 157 | + |
| 158 | +// TODO: refactor to reduce duplication with indexer/rpc stream in `createStoreSync.ts` |
| 159 | +function createLatestBlockStream({ |
| 160 | + fromBlock, |
| 161 | + indexerUrl, |
| 162 | + chainId, |
| 163 | + address, |
| 164 | + filters, |
| 165 | + latestBlockNumber$, |
| 166 | + maxBlockRange, |
| 167 | + ...opts |
| 168 | +}: PreconfirmedBlockStreamOptions): Observable<StorageAdapterBlock> { |
| 169 | + const indexerBlocks$ = indexerUrl |
| 170 | + ? of(indexerUrl).pipe( |
| 171 | + mergeMap((indexerUrl) => { |
| 172 | + const url = new URL( |
| 173 | + `api/logs-live?${new URLSearchParams({ |
| 174 | + input: JSON.stringify({ chainId, address, filters }), |
| 175 | + block_num: fromBlock.toString(), |
| 176 | + include_tx_hash: "true", |
| 177 | + })}`, |
| 178 | + indexerUrl, |
| 179 | + ); |
| 180 | + return fromEventSource<string>(url); |
| 181 | + }), |
| 182 | + map((messageEvent) => { |
| 183 | + const data = JSON.parse(messageEvent.data); |
| 184 | + if (!isLogsApiResponse(data)) { |
| 185 | + throw new Error("Received unexpected from indexer:" + messageEvent.data); |
| 186 | + } |
| 187 | + return toStorageAdapterBlock(data); |
| 188 | + }), |
| 189 | + ) |
| 190 | + : throwError(() => new Error("No indexer URL provided")); |
| 191 | + |
| 192 | + let lastBlockNumberProcessed = 0n; |
| 193 | + const ethRpcBlocks$ = combineLatest([of(fromBlock), latestBlockNumber$]).pipe( |
| 194 | + map(([startBlock, endBlock]) => ({ startBlock, endBlock })), |
| 195 | + concatMap((range) => { |
| 196 | + const storedBlocks = fetchAndStoreLogs({ |
| 197 | + ...opts, |
| 198 | + address, |
| 199 | + events: storeEventsAbi, |
| 200 | + maxBlockRange, |
| 201 | + fromBlock: lastBlockNumberProcessed |
| 202 | + ? bigIntMax(range.startBlock, lastBlockNumberProcessed + 1n) |
| 203 | + : range.startBlock, |
| 204 | + toBlock: range.endBlock, |
| 205 | + logFilter: filters.length |
| 206 | + ? (log: StoreEventsLog): boolean => |
| 207 | + filters.some( |
| 208 | + (filter) => |
| 209 | + filter.tableId === log.args.tableId && |
| 210 | + (filter.key0 == null || filter.key0 === log.args.keyTuple[0]) && |
| 211 | + (filter.key1 == null || filter.key1 === log.args.keyTuple[1]), |
| 212 | + ) |
| 213 | + : undefined, |
| 214 | + storageAdapter: () => Promise.resolve(), |
| 215 | + }); |
| 216 | + return from(storedBlocks); |
| 217 | + }), |
| 218 | + tap((block) => { |
| 219 | + lastBlockNumberProcessed = block.blockNumber; |
| 220 | + }), |
| 221 | + ); |
| 222 | + |
| 223 | + const latestBlock$ = indexerBlocks$.pipe( |
| 224 | + catchError((error) => { |
| 225 | + debug("failed to stream logs from indexer:", error.message); |
| 226 | + debug("falling back to streaming logs from ETH RPC"); |
| 227 | + return ethRpcBlocks$; |
| 228 | + }), |
| 229 | + ); |
| 230 | + |
| 231 | + return latestBlock$; |
| 232 | +} |
0 commit comments