From 0b997e52bf51170db8a54df7ada6326280c46f86 Mon Sep 17 00:00:00 2001
From: saltbo <saltbo@foxmail.com>
Date: Mon, 3 Aug 2026 05:33:03 -0400
Subject: [PATCH] fix(quota): resume duplicate cloud webhooks

---
 server/adapters/repos/cloud-store.test.ts | 36 ++++++++++++++++----
 server/adapters/repos/cloud-store.ts      | 40 ++++++++++-------------
 2 files changed, 48 insertions(+), 28 deletions(-)

diff --git a/server/adapters/repos/cloud-store.test.ts b/server/adapters/repos/cloud-store.test.ts
index cccffea0..84df5f14 100644
--- a/server/adapters/repos/cloud-store.test.ts
+++ b/server/adapters/repos/cloud-store.test.ts
@@ -40,6 +40,16 @@ function createAsyncDb(
           if (table === orgQuotaEntitlements) state.entitlementInserts += Array.isArray(values) ? values.length : 1
         }
         if (table === orgQuotaEntitlements) return { onConflictDoUpdate: async () => apply() }
+        if (table === webhookEvents) {
+          return {
+            onConflictDoNothing: () => ({
+              returning: async () => {
+                apply()
+                return [{ id: values.id }]
+              },
+            }),
+          }
+        }
         return Promise.resolve(apply())
       },
     }),
@@ -75,9 +85,13 @@ function createAsyncDb(
 function createFailingBeginDb() {
   return {
     insert: () => ({
-      values: async () => {
-        throw new Error('insert failed')
-      },
+      values: () => ({
+        onConflictDoNothing: () => ({
+          returning: async () => {
+            throw new Error('insert failed')
+          },
+        }),
+      }),
     }),
   } as unknown as Database
 }
@@ -100,11 +114,13 @@ function createUniqueConflictDb(existing: { id: string; payloadHash: string; sta
       return Promise.all(queries)
     },
     insert: (table: unknown) => ({
-      values: async (values: Record<string, unknown>) => {
-        if (table === webhookEvents) throw new Error('UNIQUE constraint failed: webhook_events.source, event_id')
+      values: (values: Record<string, unknown>) => {
+        if (table === webhookEvents) {
+          return { onConflictDoNothing: () => ({ returning: async () => [] }) }
+        }
         if (table === auditEvents) state.audits += 1
         if (table === orgQuotaEntitlements) state.entitlementInserts += Array.isArray(values) ? values.length : 1
-        return values
+        return Promise.resolve(values)
       },
     }),
     select: () => ({
@@ -163,6 +179,14 @@ function createSyncDb(
           return {
             run: apply,
             onConflictDoUpdate: () => ({ run: apply }),
+            onConflictDoNothing: () => ({
+              returning: () => ({
+                all: () => {
+                  apply()
+                  return [{ id: values.id }]
+                },
+              }),
+            }),
           }
         },
       }
diff --git a/server/adapters/repos/cloud-store.ts b/server/adapters/repos/cloud-store.ts
index 8ed2a68d..2e0d5643 100644
--- a/server/adapters/repos/cloud-store.ts
+++ b/server/adapters/repos/cloud-store.ts
@@ -273,22 +273,24 @@ async function beginWebhookEvent(
   payloadHash: string,
 ): Promise<{ id: string; duplicate: boolean }> {
   const id = nanoid()
-  try {
-    await db.insert(webhookEvents).values({
-      id,
-      source: 'cloud',
-      eventId: event.eventId,
-      eventType: event.eventType,
-      payloadHash,
-      rawPayload,
-      status: 'processing',
-      createdAt: new Date(),
-    })
-    return { id, duplicate: false }
-  } catch (error) {
-    if (isUniqueConflict(error)) return resumeWebhookEvent(db, event, rawPayload, payloadHash)
-    throw error
-  }
+  const inserted = await executeRows(
+    db
+      .insert(webhookEvents)
+      .values({
+        id,
+        source: 'cloud',
+        eventId: event.eventId,
+        eventType: event.eventType,
+        payloadHash,
+        rawPayload,
+        status: 'processing',
+        createdAt: new Date(),
+      })
+      .onConflictDoNothing({ target: [webhookEvents.source, webhookEvents.eventId] })
+      .returning({ id: webhookEvents.id }),
+  )
+  if (inserted[0]) return { id: inserted[0].id, duplicate: false }
+  return resumeWebhookEvent(db, event, rawPayload, payloadHash)
 }
 
 async function resumeWebhookEvent(
@@ -333,12 +335,6 @@ function parseOrgType(metadata: string | null): string {
   return (JSON.parse(metadata) as { type?: string }).type ?? 'unknown'
 }
 
-function isUniqueConflict(error: unknown): boolean {
-  if (!(error instanceof Error)) return false
-  const message = error.message.toLowerCase()
-  return message.includes('unique') || message.includes('constraint failed')
-}
-
 export function createCloudStoreRepo(db: Database): CloudStoreRepo {
   return {
     getAccessibleTargets: (userId) => getAccessibleTargets(db, userId),