feat: add support for backfilling lend storage messages (#2655)

## Why is this change needed?

Add support for backfilling lend storage messages from snapchain to
clients.

## Merge Checklist

_Choose all relevant options below by adding an `x` now or at any time
before submitting for review_

- [x] PR title adheres to the [conventional
commits](https://www.conventionalcommits.org/en/v1.0.0/) standard
- [x] PR has a
[changeset](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#35-adding-changesets)
- [x] PR has been tagged with a change label(s) (i.e. documentation,
feature, bugfix, or chore)
- [ ] PR includes
[documentation](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#32-writing-docs)
if necessary.

<!-- start pr-codex -->

---

## PR-Codex overview
This PR focuses on updating the `@farcaster/shuttle` package to version
`0.9.5`, adding support for backfilling lend storage messages, and
introducing new methods related to lend storage message retrieval.

### Detailed summary
- Updated version in `package.json` files to `0.9.5`.
- Added support for backfilling lend storage messages in `CHANGELOG.md`.
- Introduced `getAllLendStorageMessagesByFid` method in
`shuttle.integration.test.ts`.
- Included `LEND_STORAGE` message type in message reconciliation logic
in `messageReconciliation.ts`.
- Created `getAllLendStorageMessagesByFidInBatchesOf` for batch
processing of lend storage messages.

> ✨ Ask PR-Codex anything about this PR by commenting with `/codex {your
question}`

<!-- end pr-codex -->

packages/shuttle/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @farcaster/hub-shuttle
 
+## 0.9.5
+
+### Patch Changes
+
+- 872c2c78: feat: add support for backfilling lend storage messages
+
 ## 0.9.4
 
 ### Patch Changes

packages/shuttle/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@farcaster/shuttle",
-  "version": "0.9.4",
+  "version": "0.9.5",
   "main": "./dist/index.js",
   "module": "./dist/index.mjs",
   "types": "./dist/index.d.ts",

packages/shuttle/src/example-app/package.json

@@ -4,7 +4,7 @@
   "main": "./app.ts",
   "license": "MIT",
   "dependencies": {
-    "@farcaster/shuttle": "^0.9.3",
+    "@farcaster/shuttle": "^0.9.5",
     "@figma/hot-shots": "^9.0.0-figma.1",
     "commander": "^11.0.0",
     "ioredis": "^5.3.2",

packages/shuttle/src/shuttle.integration.test.ts

@@ -621,6 +621,18 @@ describe("shuttle", () => {
           }),
         );
       },
+      getAllLendStorageMessagesByFid: async (
+        _request: FidRequest,
+        _metadata: Metadata,
+        _options: Partial<CallOptions>,
+      ) => {
+        return ok(
+          MessagesResponse.create({
+            messages: [],
+            nextPageToken: undefined,
+          }),
+        );
+      },
     };
 
     // Only include 2 of the 3 messages in the time window

packages/shuttle/src/shuttle/messageReconciliation.ts

@@ -47,13 +47,13 @@ export class MessageReconciliation {
     startTimestamp?: number,
     stopTimestamp?: number,
   ) {
-    // Don't reconcile storage lends
     for (const type of [
       MessageType.CAST_ADD,
       MessageType.REACTION_ADD,
       MessageType.LINK_ADD,
       MessageType.VERIFICATION_ADD_ETH_ADDRESS,
       MessageType.USER_DATA_ADD,
+      MessageType.LEND_STORAGE,
     ]) {
       this.log.debug(`Reconciling messages for FID ${fid} of type ${type}`);
       await this.reconcileMessagesOfTypeForFid(fid, type, onHubMessage, onDbMessage, startTimestamp, stopTimestamp);
@@ -149,6 +149,9 @@ export class MessageReconciliation {
       case MessageType.USER_DATA_ADD:
         fn = this.getAllUserDataMessagesByFidInBatchesOf;
         break;
+      case MessageType.LEND_STORAGE:
+        fn = this.getAllLendStorageMessagesByFidInBatchesOf;
+        break;
       default:
         throw `Unknown message type ${type}`;
     }
@@ -177,6 +180,10 @@ export class MessageReconciliation {
     return await this.client.getAllUserDataMessagesByFid(request);
   }
 
+  private async getAllLendStorageMessagesByFid(request: FidTimestampRequest) {
+    return await this.client.getAllLendStorageMessagesByFid(request);
+  }
+
   private async *getAllCastMessagesByFidInBatchesOf(
     fid: number,
     pageSize: number,
@@ -314,6 +321,33 @@ export class MessageReconciliation {
     }
   }
 
+  private async *getAllLendStorageMessagesByFidInBatchesOf(
+    fid: number,
+    pageSize: number,
+    startTimestamp?: number,
+    stopTimestamp?: number,
+  ) {
+    let result = await this.getAllLendStorageMessagesByFid({ pageSize, fid, startTimestamp, stopTimestamp });
+    for (;;) {
+      if (result.isErr()) {
+        throw new Error(`Unable to get all lend storage messages for FID ${fid}: ${result.error?.message}`);
+      }
+
+      const { messages, nextPageToken: pageToken } = result.value;
+
+      yield messages;
+
+      if (!pageToken?.length) break;
+      result = await this.getAllLendStorageMessagesByFid({
+        pageSize,
+        pageToken,
+        fid,
+        startTimestamp,
+        stopTimestamp,
+      });
+    }
+  }
+
   private async allActiveDbMessagesOfTypeForFid(
     fid: number,
     type: MessageType,