feat: support compact state type for links in message reconciliation (#2074)

## Motivation

Currently message reconciliation skips over LinkCompactState messages
when fetching from hubs, and depending on the logic of the caller of
that reconciler, it may make the caller mark the message as deleted (and
subsequently perform deletions on the follows).

## Change Summary

This change alters the hub message fetch to also call this RPC when
reconciling links. Important: users of this version will have to have
the corresponding RPC from #2068.

## 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)
- [x] PR includes
[documentation](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#32-writing-docs)
if necessary.
- [x] All [commits have been
signed](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#22-signing-commits)

## Additional Context

If this is a relatively large or complex change, provide more details
here that will help reviewers

<!-- start pr-codex -->

---

## PR-Codex overview
This PR focuses on enhancing link reconciliation in the `shuttle`
package by calling the compact state RPC.

### Detailed summary
- Added a new method to retrieve link compact state messages by FID
- Updated imports in integration tests
- Implemented a test for reconciling messages and flagging incorrectly
deleted messages

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

<!-- end pr-codex -->

Author: CassOnMars

.changeset/proud-crabs-teach.md

@@ -0,0 +1,5 @@
+---
+"@farcaster/shuttle": minor
+---
+
+Link reconciliation now calls the compact state rpc as well

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

@@ -1,7 +1,19 @@
 import { migrateToLatest } from "./example-app/db";
 import { log } from "./log";
 import { sql } from "kysely";
-import { Factories, HubEvent, HubEventType, Message } from "@farcaster/hub-nodejs";
+import {
+  CallOptions,
+  Factories,
+  FidRequest,
+  HubEvent,
+  HubEventType,
+  HubRpcClient,
+  LinkCompactStateBody,
+  Message,
+  MessageType,
+  MessagesResponse,
+  Metadata,
+} from "@farcaster/hub-nodejs";
 import {
   RedisClient,
   HubSubscriber,
@@ -11,7 +23,9 @@ import {
   StoreMessageOperation,
   HubEventProcessor,
   MessageState,
+  MessageReconciliation,
 } from "./shuttle";
+import { ok } from "neverthrow";
 
 let db: DB;
 let subscriber: FakeHubSubscriber;
@@ -193,6 +207,99 @@ describe("shuttle", () => {
     expect(removeMessageInDb.deletedAt).toBeNull();
   });
 
+  test("reconciler flags incorrectly deleted messages", async () => {
+    const addMessage = await Factories.LinkAddMessage.create({}, { transient: { signer } });
+    const targetFid = addMessage.data.linkBody.targetFid;
+    expect(targetFid).toBeDefined();
+    const compactMessage = await Factories.LinkCompactStateMessage.create(
+      {
+        data: {
+          fid: addMessage.data.fid,
+          linkCompactStateBody: {
+            type: addMessage.data.linkBody.type,
+            targetFids: [targetFid ?? 0],
+          },
+        },
+      },
+      { transient: { signer } },
+    );
+
+    await subscriber.processHubEvent(
+      HubEvent.create({ id: 1, type: HubEventType.MERGE_MESSAGE, mergeMessageBody: { message: addMessage } }),
+    );
+    await subscriber.processHubEvent(
+      HubEvent.create({
+        id: 2,
+        type: HubEventType.MERGE_MESSAGE,
+        mergeMessageBody: { message: compactMessage },
+      }),
+    );
+
+    // set compact message to deleted:
+    await db.updateTable("messages").where("hash", "=", compactMessage.hash).set({ deletedAt: new Date() }).execute();
+
+    // It's a hack, but mockito is not handling this well:
+    const mockRPCClient = {
+      getAllLinkMessagesByFid: async (_request: FidRequest, _metadata: Metadata, _options: Partial<CallOptions>) => {
+        return ok(
+          MessagesResponse.create({
+            messages: [addMessage],
+            nextPageToken: undefined,
+          }),
+        );
+      },
+      getLinkCompactStateMessageByFid: async (
+        _request: FidRequest,
+        _metadata: Metadata,
+        _options: Partial<CallOptions>,
+      ) => {
+        return ok(
+          MessagesResponse.create({
+            messages: [compactMessage],
+            nextPageToken: undefined,
+          }),
+        );
+      },
+    };
+
+    const reconciler = new MessageReconciliation(mockRPCClient as unknown as HubRpcClient, db, log);
+    const messagesOnHub: Message[] = [];
+    const missingFromHub: boolean[] = [];
+    const prunedInDb: boolean[] = [];
+    const revokedInDb: boolean[] = [];
+    const messagesInDb: {
+      hash: Uint8Array;
+      prunedAt: Date | null;
+      revokedAt: Date | null;
+      fid: number;
+      type: MessageType;
+      raw: Uint8Array;
+      signer: Uint8Array;
+    }[] = [];
+    const missingFromDb: boolean[] = [];
+    const a = await reconciler.reconcileMessagesOfTypeForFid(
+      addMessage.data.fid,
+      MessageType.LINK_ADD,
+      async (msg, missing, pruned, revoked) => {
+        messagesOnHub.push(msg);
+        missingFromHub.push(missing);
+        prunedInDb.push(pruned);
+        revokedInDb.push(revoked);
+      },
+      async (dbMsg, missing) => {
+        messagesInDb.push(dbMsg);
+        missingFromDb.push(missing);
+      },
+    );
+
+    expect(messagesOnHub.length).toBe(2);
+    expect(messagesInDb.length).toBe(1);
+    expect(missingFromHub).toMatchObject([false, false]);
+    expect(prunedInDb).toMatchObject([false, false]);
+    expect(revokedInDb).toMatchObject([false, false]);
+    expect(missingFromDb).toMatchObject([false]);
+  });
+
   test("marks messages as pruned", async () => {
     const addMessage = await Factories.ReactionAddMessage.create({}, { transient: { signer } });
     subscriber.addMessageCallback((msg, operation, state, isNew, wasMissed) => {
@@ -272,7 +379,7 @@ describe("shuttle", () => {
         .select(["hash", "body"])
         .where("hash", "=", message.hash)
         .executeTakeFirstOrThrow();
-      expect(res.body.targetFids).toEqual([1, 2, 3]);
+      expect((res.body as LinkCompactStateBody).targetFids).toEqual([1, 2, 3]);
     });
   });
 });

packages/shuttle/src/shuttle/messageReconciliation.ts

@@ -174,6 +174,20 @@ export class MessageReconciliation {
       if (!pageToken?.length) break;
       result = await this.client.getAllLinkMessagesByFid({ pageSize, pageToken, fid });
     }
+
+    let deltaResult = await this.client.getLinkCompactStateMessageByFid({ fid, pageSize });
+    for (;;) {
+      if (deltaResult.isErr()) {
+        throw new Error(`Unable to get all link compact results for FID ${fid}: ${deltaResult.error?.message}`);
+      }
+
+      const { messages, nextPageToken: pageToken } = deltaResult.value;
+
+      yield messages;
+
+      if (!pageToken?.length) break;
+      deltaResult = await this.client.getLinkCompactStateMessageByFid({ pageSize, pageToken, fid });
+    }
   }
 
   private async *getAllVerificationMessagesByFidInBatchesOf(fid: number, pageSize: number) {