feat: Support onchain events and fnames in sync trie (farcasterxyz#1417)

* feat: Support onchain events and fnames in sync trie

* changeset

* fix flaky test

.changeset/cold-snakes-join.md

@@ -0,0 +1,6 @@
+---
+"@farcaster/core": patch
+"@farcaster/hubble": patch
+---
+
+feat: support onchain events and fnames in sync trie

apps/hubble/src/eth/fnameRegistryEventsProvider.test.ts

@@ -2,6 +2,7 @@ import {
   FNameRegistryClientInterface,
   FNameRegistryEventsProvider,
   FNameTransfer,
+  FNameTransferRequest,
 } from "./fnameRegistryEventsProvider.js";
 import { jestRocksDB } from "../storage/db/jestUtils.js";
 import Engine from "../storage/engine/index.js";
@@ -11,6 +12,7 @@ import { getUserNameProof } from "../storage/db/nameRegistryEvent.js";
 import { utf8ToBytes } from "@noble/curves/abstract/utils";
 import { generatePrivateKey, privateKeyToAccount, PrivateKeyAccount, Address } from "viem/accounts";
 import { bytesToHex } from "viem";
+import { HubState } from "@farcaster/hub-nodejs";
 
 class MockFnameRegistryClient implements FNameRegistryClientInterface {
   private transfersToReturn: FNameTransfer[][] = [];
@@ -57,12 +59,12 @@ class MockFnameRegistryClient implements FNameRegistryClientInterface {
     this.timesToThrow = 1;
   }
 
-  async getTransfers(since: 0): Promise<FNameTransfer[]> {
+  async getTransfers(params: FNameTransferRequest): Promise<FNameTransfer[]> {
     if (this.timesToThrow > 0) {
       this.timesToThrow--;
       throw new Error("connection failed");
     }
-    expect(since).toBeGreaterThanOrEqual(this.minimumSince);
+    expect(params.fromId).toBeGreaterThanOrEqual(this.minimumSince);
     const transfers = this.transfersToReturn.shift();
     if (!transfers) {
       return Promise.resolve([]);
@@ -146,7 +148,7 @@ describe("fnameRegistryEventsProvider", () => {
     });
 
     it("fetches events from where it left off", async () => {
-      await hub.putHubState({ lastFnameProof: transferEvents[0]?.id ?? 0, lastEthBlock: 0, lastL2Block: 0 });
+      await hub.putHubState(HubState.create({ lastFnameProof: transferEvents[0]?.id ?? 0 }));
       mockFnameRegistryClient.setMinimumSince(transferEvents[0]?.id ?? 0);
       await provider.start();
       expect(await getUserNameProof(db, utf8ToBytes("test2"))).toBeTruthy();

apps/hubble/src/eth/fnameRegistryEventsProvider.ts

@@ -29,8 +29,13 @@ export type FNameTransfer = {
   to: number;
 };
 
+export type FNameTransferRequest = {
+  fromId?: number;
+  name?: string;
+};
+
 export interface FNameRegistryClientInterface {
-  getTransfers(fromId: number): Promise<FNameTransfer[]>;
+  getTransfers(params: FNameTransferRequest): Promise<FNameTransfer[]>;
   getSigner(): Promise<string>;
 }
 
@@ -40,8 +45,9 @@ export class FNameRegistryClient implements FNameRegistryClientInterface {
     this.url = url;
   }
 
-  public async getTransfers(fromId = 0): Promise<FNameTransfer[]> {
-    const response = await axios.get(`${this.url}/transfers?from_id=${fromId}`, {
+  public async getTransfers(params: FNameTransferRequest): Promise<FNameTransfer[]> {
+    const response = await axios.get(`${this.url}/transfers`, {
+      params,
       timeout: DEFAULT_READ_TIMEOUT_IN_MS,
     });
     return response.data.transfers;
@@ -114,7 +120,7 @@ export class FNameRegistryEventsProvider {
     }
 
     this.lastTransferId = fromId;
-    let transfers = await this.safeGetTransfers(fromId);
+    let transfers = await this.safeGetTransfers({ fromId });
     let transfersCount = 0;
     while (transfers.length > 0 && !this.shouldStop) {
       transfersCount += transfers.length;
@@ -124,7 +130,7 @@ export class FNameRegistryEventsProvider {
         break;
       }
       this.lastTransferId = lastTransfer.id;
-      transfers = await this.safeGetTransfers(this.lastTransferId);
+      transfers = await this.safeGetTransfers({ fromId: this.lastTransferId });
     }
     log.info(`Fetched ${transfersCount} fname events upto ${this.lastTransferId}`);
     const result = await this.hub.getHubState();
@@ -136,11 +142,18 @@ export class FNameRegistryEventsProvider {
     }
   }
 
-  private async safeGetTransfers(fromId: number) {
+  public async retryTransferByName(name: Uint8Array) {
+    const nameStr = Buffer.from(name).toString("utf-8");
+    const transfers = await this.safeGetTransfers({ name: nameStr });
+    await this.mergeTransfers(transfers);
+    log.info(`Retried ${transfers.length} fname events for ${nameStr}`);
+  }
+
+  private async safeGetTransfers(params: FNameTransferRequest) {
     try {
-      return await this.client.getTransfers(fromId);
+      return await this.client.getTransfers(params);
     } catch (err) {
-      log.error(err, `Failed to get transfers from ${fromId}`);
+      log.error(err, `Failed to get transfers ${params}`);
       return [];
     }
   }

apps/hubble/src/hubble.ts

@@ -347,7 +347,13 @@ export class Hub implements HubInterface {
     this.engine = new Engine(this.rocksDB, options.network, eventHandler, mainnetClient);
 
     const profileSync = options.profileSync ?? false;
-    this.syncEngine = new SyncEngine(this, this.rocksDB, this.l2RegistryProvider, profileSync);
+    this.syncEngine = new SyncEngine(
+      this,
+      this.rocksDB,
+      this.l2RegistryProvider,
+      this.fNameRegistryEventsProvider,
+      profileSync,
+    );
 
     // If profileSync is true, exit after sync is complete
     if (profileSync) {
@@ -848,7 +854,7 @@ export class Hub implements HubInterface {
     const result = await ResultAsync.fromPromise(getHubState(this.rocksDB), (e) => e as HubError);
     if (result.isErr() && result.error.errCode === "not_found") {
       log.info("hub state not found, resetting state");
-      const hubState = HubState.create({ lastEthBlock: 0, lastFnameProof: 0 });
+      const hubState = HubState.create({ lastEthBlock: 0, lastFnameProof: 0, syncEvents: false });
       await putHubState(this.rocksDB, hubState);
       return ok(hubState);
     }