get records and make a plan

Author: frolic

packages/cli/src/commands/mirror.ts

@@ -1,9 +1,12 @@
 import type { CommandModule, InferredOptionTypes } from "yargs";
-import { createClient, getAddress, http } from "viem";
+import { createClient, getAddress, http, isHex } from "viem";
 import chalk from "chalk";
 import { getChainId } from "viem/actions";
 import { defaultChains } from "../defaultChains";
 import { mirror } from "../mirror/mirror";
+import { MUDError } from "@latticexyz/common/errors";
+import { kmsKeyToAccount } from "@latticexyz/common/kms";
+import { privateKeyToAccount } from "viem/accounts";
 
 const options = {
   rpcBatch: {
@@ -19,6 +22,10 @@ const options = {
     desc: "Source world address to mirror from.",
     required: true,
   },
+  fromBlock: {
+    type: "number",
+    desc: "Block number of source world deploy.",
+  },
   fromRpc: {
     type: "string",
     desc: "RPC URL of source chain to mirror from.",
@@ -55,11 +62,36 @@ const commandModule: CommandModule<Options, Options> = {
     });
     const fromChainId = await getChainId(fromClient);
     const fromIndexer = opts.fromIndexer ?? defaultChains.find((chain) => chain.id === fromChainId)?.indexerUrl;
+    if (!fromIndexer) {
+      throw new MUDError(`No \`--fromIndexer\` provided or indexer URL configured for chain ${fromChainId}.`);
+    }
+
+    const account = await (async () => {
+      if (opts.kms) {
+        const keyId = process.env.AWS_KMS_KEY_ID;
+        if (!keyId) {
+          throw new MUDError(
+            "Missing `AWS_KMS_KEY_ID` environment variable. This is required when using with `--kms` option.",
+          );
+        }
+
+        return await kmsKeyToAccount({ keyId });
+      } else {
+        const privateKey = process.env.PRIVATE_KEY;
+        if (!isHex(privateKey)) {
+          throw new MUDError(
+            `Missing or invalid \`PRIVATE_KEY\` environment variable. To use the default Anvil private key, run\n\n  echo "PRIVATE_KEY=0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" > .env\n`,
+          );
+        }
+        return privateKeyToAccount(privateKey);
+      }
+    })();
 
     const toClient = createClient({
       transport: http(opts.toRpc, {
         batch: opts.rpcBatch ? { batchSize: 100, wait: 1000 } : undefined,
       }),
+      account,
     });
     const toChainId = await getChainId(toClient);
 
@@ -71,11 +103,17 @@ const commandModule: CommandModule<Options, Options> = {
       ),
     );
 
-    // TODO: load up account from KMS or private key
-
     await mirror({
-      from: { world: fromWorld, client: fromClient, indexer: fromIndexer },
-      to: { client: toClient },
+      rootDir: process.cwd(),
+      from: {
+        world: fromWorld,
+        block: opts.fromBlock != null ? BigInt(opts.fromBlock) : undefined,
+        client: fromClient,
+        indexer: fromIndexer,
+      },
+      to: {
+        client: toClient,
+      },
     });
   },
 };

packages/cli/src/mirror/debug.ts

@@ -0,0 +1,10 @@
+import { debug as parentDebug } from "../debug";
+
+export const debug = parentDebug.extend("mirro");
+export const error = parentDebug.extend("mirro");
+
+// Pipe debug output to stdout instead of stderr
+debug.log = console.debug.bind(console);
+
+// Pipe error output to stderr
+error.log = console.error.bind(console);

packages/cli/src/mirror/mirror.ts

@@ -1,30 +1,93 @@
-import { Address, Client } from "viem";
+import { createWriteStream } from "fs";
+import { createGzip } from "zlib";
+import { pipeline } from "stream/promises";
+import { Account, Address, Chain, Client, Transport } from "viem";
+import { getWorldDeploy } from "../deploy/getWorldDeploy";
+import { getChainId } from "viem/actions";
+import { getTables } from "../deploy/getTables";
+import { resourceToLabel } from "@latticexyz/common";
+import { getRecordsAsLogs } from "@latticexyz/store-sync";
+import pRetry from "p-retry";
+import { Table } from "@latticexyz/config";
+import path from "path";
 
-// TODO: decide if preserving world address is important (chain config has to change anyway)
+// TODO: attempt to create world the same way as it was originally created, thus preserving world address
+// TODO: set up table to track migrated records with original metadata (block number/timestamp) and for lazy migrations
 
-export async function mirror(opts: {
+export async function mirror({
+  rootDir,
+  from,
+  to,
+}: {
+  rootDir: string;
   from: {
+    block?: bigint;
     world: Address;
     client: Client;
-    indexer?: string;
+    indexer: string;
   };
   to: {
-    client: Client;
+    client: Client<Transport, Chain | undefined, Account>;
   };
 }) {
   // TODO: check for world balance, warn
-  //
-  // TODO: get world salt, factory, and deployer address
-  // TODO: prepare world deploy, make sure resulting world addresses will match
   // TODO: deploy world
   //
-  // TODO: set up a table to track migrated records? would make this idempotent and enable lazy mirroring
-  //
+
   // TODO: fetch data from indexer
   // TODO: check each system for state/balance, warn
   //
   // TODO: set records for each table
   //
   // TODO: deploy each system via original bytecode
   // TODO: update system addresses as necessary (should this be done as part of setting records?)
+  //
+  const fromChainId = await getChainId(from.client);
+  const toChainId = await getChainId(to.client);
+
+  const plan = createPlan(path.join(rootDir, "mud-mirror-plan.json.gz"));
+  plan.add("mirror", { from: { chainId: fromChainId, world: from.world }, to: { chainId: toChainId } });
+
+  const worldDeploy = await getWorldDeploy(from.client, from.world, from.block);
+  const tables = await getTables({
+    client: from.client,
+    worldDeploy,
+    indexerUrl: from.indexer,
+    chainId: fromChainId,
+  });
+
+  for (const table of tables) {
+    const logs = await pRetry(() =>
+      getRecordsAsLogs<Table>({
+        worldAddress: from.world,
+        table: table as never,
+        client: from.client,
+        indexerUrl: from.indexer,
+        chainId: fromChainId,
+      }),
+    );
+    console.log("got", logs.length, "logs for", resourceToLabel(table));
+    for (const log of logs) {
+      plan.add("setRecord", log.args);
+    }
+  }
+
+  await plan.end();
+}
+
+function createPlan(filename: string) {
+  const gzip = createGzip();
+  const fileStream = createWriteStream(filename);
+  const output = pipeline(gzip, fileStream);
+  gzip.write("[\n");
+  return {
+    add(step: string, data: any) {
+      gzip.write(JSON.stringify({ step, data }) + ",\n");
+      return this;
+    },
+    async end() {
+      gzip.end("]\n");
+      await output;
+    },
+  };
 }

packages/store-sync/src/exports/internal.ts

@@ -1,5 +1,10 @@
+export * from "../getRecords";
+
 // SQL
-export * from "../sql";
+export * from "../sql/common";
+export * from "../sql/fetchRecords";
+export * from "../sql/selectFrom";
+export * from "../sql/getSnapshot";
 
 // Stash
 export * from "../stash/common";

packages/store-sync/src/getRecords.ts

@@ -43,46 +43,48 @@ type GetRecordsResult<table extends Table = Table> = {
 export async function getRecords<table extends Table>(
   options: GetRecordsOptions<table>,
 ): Promise<GetRecordsResult<table>> {
-  async function getLogs(): Promise<readonly StorageAdapterLog[]> {
-    if (options.indexerUrl && options.chainId) {
-      debug("fetching records for", options.table.label, "via indexer from", options.indexerUrl);
-      const logs = await getSnapshot({
-        chainId: options.chainId,
-        address: options.worldAddress,
-        indexerUrl: options.indexerUrl,
-        filters: [{ tableId: options.table.tableId }],
-      });
-      // By default, the indexer includes the `store.Tables` table as part of the snapshot.
-      // Once we change this default, we can remove the filter here.
-      // See https://github.com/latticexyz/mud/issues/3386.
-      return logs?.logs.filter((log) => log.args.tableId === options.table.tableId) ?? [];
-    } else {
-      const client =
-        options.client ??
-        createClient({
-          transport: http(options.rpcUrl),
-        });
-      debug("fetching records for", options.table.label, "via RPC from", client.transport.url);
-      const blockLogs = await fetchBlockLogs({
-        fromBlock: options.fromBlock ?? 0n,
-        toBlock: options.toBlock ?? (await getAction(client, getBlockNumber, "getBlockNumber")({})),
-        maxBlockRange: 100_000n,
-        async getLogs({ fromBlock, toBlock }) {
-          return getStoreLogs(client, {
-            address: options.worldAddress,
-            fromBlock,
-            toBlock,
-            tableId: options.table.tableId,
-          });
-        },
-      });
-      return flattenStoreLogs(blockLogs.flatMap((block) => block.logs));
-    }
-  }
-
-  const logs = await getLogs();
-  const records = logs.map((log) => logToRecord({ log: log as LogToRecordArgs<table>["log"], table: options.table }));
+  const logs = await getRecordsAsLogs(options);
+  const records = logs.map((log) => logToRecord({ log: log as never, table: options.table }));
   const blockNumber = logs.length > 0 ? logs[logs.length - 1].blockNumber ?? 0n : 0n;
   debug("found", records.length, "records for table", options.table.label, "at block", blockNumber);
   return { records, blockNumber };
 }
+
+export async function getRecordsAsLogs<table extends Table>(
+  options: GetRecordsOptions<table>,
+): Promise<readonly Extract<StorageAdapterLog, { eventName: "Store_SetRecord" }>[]> {
+  if (options.indexerUrl && options.chainId) {
+    debug("fetching records for", options.table.label, "via indexer from", options.indexerUrl);
+    const snapshot = await getSnapshot({
+      chainId: options.chainId,
+      address: options.worldAddress,
+      indexerUrl: options.indexerUrl,
+      filters: [{ tableId: options.table.tableId }],
+    });
+    // By default, the indexer includes the `store.Tables` table as part of the snapshot.
+    // Once we change this default, we can remove the filter here.
+    // See https://github.com/latticexyz/mud/issues/3386.
+    return (snapshot?.logs.filter((log) => log.args.tableId === options.table.tableId) ?? []) as never;
+  }
+
+  const client =
+    options.client ??
+    createClient({
+      transport: http(options.rpcUrl),
+    });
+  debug("fetching records for", options.table.label, "via RPC from", client.transport.url);
+  const blockLogs = await fetchBlockLogs({
+    fromBlock: options.fromBlock ?? 0n,
+    toBlock: options.toBlock ?? (await getAction(client, getBlockNumber, "getBlockNumber")({})),
+    maxBlockRange: 100_000n,
+    async getLogs({ fromBlock, toBlock }) {
+      return getStoreLogs(client, {
+        address: options.worldAddress,
+        fromBlock,
+        toBlock,
+        tableId: options.table.tableId,
+      });
+    },
+  });
+  return flattenStoreLogs(blockLogs.flatMap((block) => block.logs));
+}

packages/store-sync/src/sql/getSnapshot.ts

@@ -49,7 +49,7 @@ export async function getSnapshot({
           });
 
       if (logsFilters && logsFilters.length === 0) {
-        return undefined;
+        return;
       }
 
       return getSnapshotLogs({