1import type { EvmClient } from '../client/EvmClient';
2import type { FetchOptions, Patch, SubscriptionGraph } from './types';
3
4async function fetchNodePatches(client: EvmClient, userId: number, fromIndex: number): Promise<Patch | null> {
5  const [compressedActions, indexes, stringIndexes] = await client.archivistStorage.loadActionsSince(userId, fromIndex);
6  if (!compressedActions || compressedActions.length === 0) return null;
7
8  return {
9    userId,
10    fromIndex,
11    toIndex: fromIndex + compressedActions.length,
12    raw: { compressedActions, indexes, stringIndexes },
13  };
14}
15
16/**
17 * Fetches new on-chain patches for every node discovered by walkGraph, since
18 * each node's cursor position (or from the start, for a full sync).
19 *
20 * rootUserId is always present in `graph.nodes` (walkGraph seeds the walk with
21 * it), so its own patches are always included here too — spec O5 ("the user
22 * should always be in sync with himself"). Precedence between the root's own
23 * patch and one reached via the subscription graph, where both touch the same
24 * logical record, is a merge rule for the calling app's patch-application
25 * layer, not something enforced here — dedup in this package is purely by
26 * (userId, index range), never by content.
27 */
28export async function fetchPatches(
29  client: EvmClient,
30  graph: SubscriptionGraph,
31  cursor?: Map<number, number>,
32  opts?: FetchOptions
33): Promise<Patch[]> {
34  const concurrency = opts?.concurrency ?? 4;
35  const userIds = Array.from(graph.nodes.keys());
36  const patches: Patch[] = [];
37
38  for (let i = 0; i < userIds.length; i += concurrency) {
39    const slice = userIds.slice(i, i + concurrency);
40    const results = await Promise.all(
41      slice.map((userId) =>
42        fetchNodePatches(client, userId, cursor?.get(userId) ?? 0).catch((e) => {
43          console.warn(`[subscription/patch-resolver] failed to fetch patches for userId ${userId}`, e);
44          return null;
45        })
46      )
47    );
48    for (const patch of results) {
49      if (patch) patches.push(patch);
50    }
51  }
52
53  return patches;
54}
55