fix(coexist): keep a chunked history sync draining across continuations

Also threads the workspace id through the WhatsApp phone-number listing, keeps
provider URLs in persisted error messages, and matches public routes by segment.

Author: viethungpv91

apps/builder/__tests__/proxy-public-routes.test.ts

@@ -21,8 +21,14 @@ describe("isPublicRoute", () => {
     expect(isPublicRoute("/channels/create")).toBe(false)
   })
 
-  test("the /t/ prefix keeps its trailing slash so /templates stays private", () => {
+  test("matching is by segment, so no entry opens a longer first segment", () => {
+    // "/t" vs "/templates" is the case that already bit us; the same bare
+    // `startsWith` would have opened "/rpcadmin" or "/storage-exports" the
+    // day either route appeared.
     expect(isPublicRoute("/t/abc")).toBe(true)
     expect(isPublicRoute("/templates")).toBe(false)
+    expect(isPublicRoute("/rpcadmin")).toBe(false)
+    expect(isPublicRoute("/apikeys")).toBe(false)
+    expect(isPublicRoute("/authorized-apps")).toBe(false)
   })
 })

apps/builder/__tests__/rpc-router-auth-surface.test.ts

@@ -0,0 +1,109 @@
+import { readdirSync, readFileSync } from "node:fs"
+import { join, relative } from "node:path"
+import { describe, expect, test } from "vitest"
+
+/**
+ * `/rpc` is a public route in the proxy (see `lib/public-routes.ts`), so the
+ * only thing standing between an anonymous request and a feature handler is
+ * the auth middleware each procedure carries. Nothing type-checks that: a
+ * procedure built on a raw oRPC base compiles, mounts, and answers without a
+ * session.
+ *
+ * These tests pin the structural reason that cannot happen — `@/orpc` exports
+ * no unauthenticated base, and no feature api module imports one from
+ * anywhere else.
+ */
+
+const BUILDER_SRC = join(import.meta.dirname, "..", "src")
+
+/** Modules that may hold a raw base, with the reason they are not procedures. */
+const RAW_BASE_ALLOWLIST = new Set([
+  // A middleware, not a procedure: it is chained onto the connect routes,
+  // which are themselves built on `authorizedAPI`.
+  "features/channel-connect/api/audit-context.ts",
+])
+
+const AUTHENTICATED_BASE =
+  /\b(authorizedAPI|workspaceTokenAuthAPIForScope|channelApiTokenAPI)\b/
+
+const PROCEDURE_HANDLER = /\.handler\(/
+
+const EXPORTED_CONST = /^export const (\w+)/gm
+
+const RAW_BASE_IMPORT =
+  /import\s*\{[^}]*\b(?:base|os|implement)\b[^}]*\}\s*from\s*["'](?:@orpc\/server|@\/middlewares\/context)["']/
+
+/** Every `.ts` under a `features/<name>/api/` directory, recursively. */
+const apiModules = (): string[] => {
+  const featuresDir = join(BUILDER_SRC, "features")
+  const collect = (dir: string): string[] =>
+    readdirSync(dir, { withFileTypes: true, recursive: true })
+      .filter((entry) => entry.isFile() && entry.name.endsWith(".ts"))
+      .map((entry) => join(entry.parentPath, entry.name))
+
+  return readdirSync(featuresDir, { withFileTypes: true })
+    .filter((entry) => entry.isDirectory())
+    .flatMap((feature) => {
+      const apiDir = join(featuresDir, feature.name, "api")
+      try {
+        return collect(apiDir)
+      } catch {
+        return []
+      }
+    })
+}
+
+describe("/rpc router auth surface", () => {
+  test("@/orpc exports only authenticated procedure bases", () => {
+    const source = readFileSync(join(BUILDER_SRC, "orpc.ts"), "utf8")
+    const exported = [...source.matchAll(EXPORTED_CONST)].map(
+      (match) => match[1],
+    )
+
+    // Adding an export here means adding a way to mount a procedure. If it is
+    // not one of these three, it must carry its own auth middleware — and this
+    // test is where that decision gets recorded.
+    expect(exported.sort()).toEqual([
+      "authorizedAPI",
+      "channelApiTokenAPI",
+      "workspaceTokenAuthAPIForScope",
+    ])
+  })
+
+  test("no feature api module imports a raw oRPC base", () => {
+    const offenders: string[] = []
+
+    for (const file of apiModules()) {
+      const relativePath = relative(BUILDER_SRC, file)
+      if (RAW_BASE_ALLOWLIST.has(relativePath)) {
+        continue
+      }
+      if (RAW_BASE_IMPORT.test(readFileSync(file, "utf8"))) {
+        offenders.push(relativePath)
+      }
+    }
+
+    expect(offenders).toEqual([])
+  })
+
+  test("every feature api module builds on one of the authenticated bases", () => {
+    const files = apiModules()
+    expect(files.length).toBeGreaterThan(50)
+
+    // Helper modules that define no procedure at all are not a surface.
+    const procedureModules = files
+      .map((file) => ({
+        path: relative(BUILDER_SRC, file),
+        source: readFileSync(file, "utf8"),
+      }))
+      .filter(({ source }) => PROCEDURE_HANDLER.test(source))
+    // Guards the filter itself: an empty set would make the check vacuous.
+    expect(procedureModules.length).toBeGreaterThan(50)
+
+    const unauthenticated = procedureModules
+      .filter(({ source }) => !AUTHENTICATED_BASE.test(source))
+      .map(({ path }) => path)
+
+    expect(unauthenticated).toEqual([])
+  })
+})

apps/builder/__tests__/whatsapp-connect-card.test.tsx

@@ -182,9 +182,13 @@ describe("WhatsappCreate connect card", () => {
       clickButtonByText("actions.continue")
       await flush()
 
+      // `workspaceId` is load-bearing: the procedure hands it to
+      // `resolvePlatformOwnerId`, so dropping it resolves the platform-global
+      // WhatsApp credential instead of the reseller's for a sub-account.
       expect(listWhatsappPhoneNumbersInternalAPI).toHaveBeenCalledWith({
         wabaId: "waba-1",
         accessToken: "token-1",
+        workspaceId: "ws-1",
       })
       expect(radios()).toHaveLength(1)
 

apps/builder/src/features/integration-whatsapp/components/whatsapp-connect-sections.tsx

@@ -31,8 +31,15 @@ export const SWITCH_FIELD_CLASS =
  * The manual path's phone-number lookup: the numbers Meta returned for the
  * WABA id + token typed into the form, cleared again whenever the operator
  * leaves manual mode so a stale list can never be submitted.
+ *
+ * `workspaceId` is what lets `resolvePlatformOwnerId` reach the reseller's
+ * WhatsApp credential; without it the listing falls back to the acting user
+ * and a sub-account resolves the platform-global credential instead.
  */
-function useManualPhoneNumbers(isManualConnect: boolean) {
+function useManualPhoneNumbers(
+  isManualConnect: boolean,
+  workspaceId?: string | null,
+) {
   const t = useTranslations()
   const { getValues, setValue } = useFormContext()
   const [phoneNumbers, setPhoneNumbers] = useState<WhatsappPhoneNumber[]>([])
@@ -65,6 +72,7 @@ function useManualPhoneNumbers(isManualConnect: boolean) {
             {
               wabaId: formData.wabaId ?? "",
               accessToken: formData.accessToken ?? "",
+              workspaceId: workspaceId ?? undefined,
             },
           )
 
@@ -77,7 +85,7 @@ function useManualPhoneNumbers(isManualConnect: boolean) {
         await clientErrorHandler(error)
       }
     })
-  }, [getValues, t])
+  }, [getValues, t, workspaceId])
 
   return { phoneNumbers, isLoading, listPhoneNumbers }
 }
@@ -206,6 +214,7 @@ export function PhoneNumberSelectionSection({
 
 type ManualConnectSectionProps = {
   watchManualConnect: boolean
+  workspaceId?: string | null
 }
 
 /** WABA id + access token, until they yield a phone-number list. */
@@ -280,10 +289,13 @@ function ManualPhoneNumberStep({
 
 export function ManualConnectSection({
   watchManualConnect,
+  workspaceId,
 }: ManualConnectSectionProps) {
   const t = useTranslations()
-  const { phoneNumbers, isLoading, listPhoneNumbers } =
-    useManualPhoneNumbers(watchManualConnect)
+  const { phoneNumbers, isLoading, listPhoneNumbers } = useManualPhoneNumbers(
+    watchManualConnect,
+    workspaceId,
+  )
 
   return (
     <>

apps/builder/src/features/integration-whatsapp/components/whatsapp-create.tsx

@@ -116,7 +116,12 @@ export default function WhatsappCreate({
     }
 
     if (watchManualConnect) {
-      return <ManualConnectSection watchManualConnect={watchManualConnect} />
+      return (
+        <ManualConnectSection
+          watchManualConnect={watchManualConnect}
+          workspaceId={workspaceId}
+        />
+      )
     }
 
     return (

apps/builder/src/lib/public-routes.ts

@@ -26,20 +26,22 @@ export const PUBLIC_ROUTES = [
   "/booking",
   "/portal/redeem",
   "/webchat",
-  // Trailing slash is deliberate: `isPublicRoute` below is a bare
-  // unanchored `startsWith`, so "/t" (no slash) would also match
-  // "/templates" and make the authenticated template list world-readable.
   "/t/",
 ]
 
 /**
- * Whether the middleware lets a request through without a session. A prefix
- * added to `PUBLIC_ROUTES` silently opens every path under it, so the list is
- * pinned by a test.
+ * Whether the middleware lets a request through without a session.
+ *
+ * Matching is by path SEGMENT, never by bare `startsWith`: a plain prefix test
+ * opens far more than the entry names — "/t" would also match "/templates",
+ * and "/rpc" would match a future "/rpcadmin". A trailing slash on an entry is
+ * therefore cosmetic here, and an entry still opens everything nested under it,
+ * so the list is pinned by a test.
  */
 export function isPublicRoute(pathname: string) {
   for (const route of PUBLIC_ROUTES) {
-    if (pathname.startsWith(route)) {
+    const base = route.endsWith("/") ? route.slice(0, -1) : route
+    if (pathname === base || pathname.startsWith(`${base}/`)) {
       return true
     }
   }

apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts

@@ -781,4 +781,68 @@ describe("coexistWhatsappFlush — run lifecycle", () => {
     ]
     expect(bulkArgs?.batch).toEqual([])
   })
+
+  // ── the chunk chain must hand the run back before queueing the next one ──
+  // `claimRun` refuses a `running` run whose heartbeat is under 10 minutes
+  // old. A continuation queued while this worker still held the claim
+  // therefore lost its own claim and abandoned, leaving the chain to the
+  // scheduler's 1-hour stale sweep — one chunk per hour, then `failed`.
+
+  /** Makes the post-drain tail re-check find a late row, forcing a continuation. */
+  const stageLateTailRow = () => {
+    harness.lastStagedChain?.limit
+      .mockReset()
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([{ id: "row-late" }])
+      .mockResolvedValue([])
+  }
+
+  it("releases the claim before queueing the continuation, so the next chunk can claim it", async () => {
+    wireSelect(runRow(), [])
+    stageLateTailRow()
+
+    await coexistWhatsappFlush({ runId, phoneNumberId })
+
+    const released = setPayloads().find(
+      (payload) => payload.status === "init" && "lastHeartbeatAt" in payload,
+    )
+    expect(released?.status).toBe("init")
+    expect(released?.lastHeartbeatAt).toBeInstanceOf(Date)
+    // The run must be `init` BEFORE the job exists, or the continuation can
+    // start against a still-`running` row and abandon.
+    expect(Math.min(...mockRunWrite.mock.invocationCallOrder)).toBeLessThan(
+      mockQueueAdd.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY,
+    )
+    expect(mockQueueAdd).toHaveBeenCalledWith(
+      "coexistWhatsappFlush",
+      expect.objectContaining({
+        data: { runId, phoneNumberId },
+      }),
+      expect.anything(),
+    )
+    // Still a guarded write: a worker that lost the run cannot release it.
+    expect(mockRunWriteGuards()).toContainEqual(
+      expect.objectContaining({ status: "running" }),
+    )
+  })
+
+  it("queues no continuation when the release finds the run already reclaimed", async () => {
+    // Write 0 is the drain's ownership heartbeat (still ours); write 1 is the
+    // release, which a reclaim by another worker makes match no rows.
+    wireUpdate((callIndex) => (callIndex === 0 ? 1 : 0))
+    wireSelect(runRow(), [])
+    stageLateTailRow()
+
+    await coexistWhatsappFlush({ runId, phoneNumberId })
+
+    expect(mockQueueAdd).not.toHaveBeenCalled()
+    expect(mockWarn).toHaveBeenCalledWith(
+      expect.objectContaining({
+        runId,
+        phoneNumberId,
+        reason: "release before continuation matched no rows",
+      }),
+      expect.stringContaining("no longer claimed"),
+    )
+  })
 })

apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts

@@ -402,13 +402,45 @@ const resolveFinalStatus = async (
 }
 
 /**
- * Hot-chains the next chunk. On enqueue failure the run is handed back to the
- * scheduler rather than left `running` with nobody driving it.
+ * Hands the run back before the continuation is queued.
+ *
+ * `claimRun` refuses a run that is `running` with a heartbeat under 10 minutes
+ * old — that is what stops two workers driving one run. A continuation
+ * enqueued while this worker still holds the claim therefore loses its own
+ * claim and abandons, so the chunk chain has to release ownership first: back
+ * to `init` with a fresh heartbeat, counters and pending patches untouched,
+ * exactly as `resetForRetry` hands a run back after a transient error. The
+ * refreshed `updatedAt` also keeps `pickDueRuns` (which only considers `init`
+ * runs idle for 10s) from racing a second job in alongside the continuation.
+ *
+ * @returns false when the claim was already lost, in which case no
+ * continuation is queued — whoever holds the run now is driving it.
+ */
+const releaseForContinuation = async (
+  context: FlushContext,
+): Promise<boolean> => {
+  const written = await coexistService.updateProgress({
+    runId: context.runId,
+    expect: context.guard,
+    fields: { status: "init", lastHeartbeatAt: new Date() },
+  })
+
+  return written > 0
+}
+
+/**
+ * Hot-chains the next chunk. On enqueue failure the run is left `init` for the
+ * scheduler rather than `running` with nobody driving it.
  */
 const enqueueContinuation = async (
   context: FlushContext,
   state: FlushState,
 ): Promise<void> => {
+  if (!(await releaseForContinuation(context))) {
+    abandon(context, "release before continuation matched no rows")
+    return
+  }
+
   try {
     await integrationQueue.add(
       IntegrationJobAction.coexistWhatsappFlush,
@@ -436,15 +468,13 @@ const enqueueContinuation = async (
       "[coexist] WhatsApp flush chunk done — continuation enqueued",
     )
   } catch (error) {
+    // The run is already `init` with a fresh heartbeat, so the scheduler's
+    // next pass drives it — nothing left to write here, and the claim this
+    // worker held is gone.
     logger.error(
       { error, runId: context.runId },
       "[coexist] WhatsApp continuation enqueue failed — fallback to scheduler",
     )
-    await coexistService.resetForRetry({
-      runId: context.runId,
-      currentError: state.currentError ?? "continuation enqueue failed",
-      expect: context.guard,
-    })
   }
 }