fix: on conflict criteria is ambiguous and cannot be used for upsert (#…

…2197)

## Why is this change needed?

Shuttle defines the messages table with two unique constraints, one for
the hash, another for the triple of hash, fid and type. Postgres does
not allow multiple ON CONFLICT criteria, and ambiguous conflicts are
essentially surfaced as uniqueness violation failures in spite of ON
CONFLICT criteria given. This change alters the merge to instead attempt
insert, doing nothing on any conflict, then resolving with an update if
necessary and the original conflict resolution criteria is satisfied.

Addresses Issue #2167 

## 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.


<!-- start pr-codex -->

---

## PR-Codex overview
This PR updates the conflict resolution logic in `MessageProcessor` to
fix ambiguity issues during upsert operations.

### Detailed summary
- Updated conflict resolution logic to use `doNothing()` method instead
of complex conditions
- Improved handling of conflict criteria for upsert operations

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

<!-- end pr-codex -->

.changeset/tidy-apricots-unite.md

@@ -0,0 +1,5 @@
+---
+"@farcaster/shuttle": patch
+---
+
+fix: on conflict criteria is ambiguous and cannot be used for upsert

packages/shuttle/src/shuttle/messageProcessor.ts

@@ -47,7 +47,7 @@ export class MessageProcessor {
     }
 
     // @ts-ignore
-    const result = await trx
+    let result = await trx
       .insertInto("messages")
       .values({
         fid: message.data.fid,
@@ -62,28 +62,35 @@ export class MessageProcessor {
         ...opData,
       })
       .returning(["id"])
-      .onConflict((oc) =>
-        oc
-          .columns(["hash", "fid", "type"])
-          // In case the signer was changed, make sure to always update it
-          .doUpdateSet({
-            signatureScheme: message.signatureScheme,
-            signer: message.signer,
-            raw: Message.encode(message).finish(),
-            ...opData,
-          })
-          .where(({ eb, or }) =>
-            or([
-              eb("excluded.deletedAt", "is not", null).and("messages.deletedAt", "is", null),
-              eb("excluded.deletedAt", "is", null).and("messages.deletedAt", "is not", null),
-              eb("excluded.prunedAt", "is not", null).and("messages.prunedAt", "is", null),
-              eb("excluded.prunedAt", "is", null).and("messages.prunedAt", "is not", null),
-              eb("excluded.revokedAt", "is not", null).and("messages.revokedAt", "is", null),
-              eb("excluded.revokedAt", "is", null).and("messages.revokedAt", "is not", null),
-            ]),
-          ),
-      )
+      .onConflict((oc) => oc.doNothing())
       .executeTakeFirst();
+
+    if (!result) {
+      const data = message.data;
+      result = await trx
+        .updateTable("messages")
+        .set({
+          signatureScheme: message.signatureScheme,
+          signer: message.signer,
+          raw: Message.encode(message).finish(),
+          ...opData,
+        })
+        .returning(["id"])
+        .where((eb) =>
+          eb.and([
+            eb("hash", "=", message.hash),
+            eb("fid", "=", data.fid),
+            eb("type", "=", data.type),
+            eb.or([
+              eb("deletedAt", !opData.deletedAt ? "is not" : "is", null),
+              eb("prunedAt", !opData.prunedAt ? "is not" : "is", null),
+              eb("revokedAt", !opData.revokedAt ? "is not" : "is", null),
+            ]),
+          ]),
+        )
+        .executeTakeFirst();
+    }
+
     return !!result;
   }