fix(v2): keep refresh dedup alive across the host persistence gap

.changeset/fix-v2-refresh-dedup.md

@@ -0,0 +1,5 @@
+---
+"opencode-supabase": patch
+---
+
+Fix V2 OAuth refresh dedup dropping its entry too early and keying on the raw broker URL (#90, #91). Keep a successful refresh cached under the consumed old refresh token until the process ends: the OpenCode host persists the rotated credential only after `refresh()` resolves, so a second resolve in that window must converge on the cached result instead of re-sending the single-use token (broker 401). Failed refreshes still delete the entry so waiters are released and a retry reaches the broker. Dedup keys now strip the trailing slash with the same normalization `makeBrokerRequest()` applies, so `https://host/broker` and `https://host/broker/` aliases no longer send the credential twice.

src/server/v2.ts

@@ -9,7 +9,7 @@ import type { Credential } from "@opencode/schema/credential";
 import type { Skill } from "@opencode/schema/skill";
 import { Tool } from "@opencode/schema/tool";
 import { Effect } from "effect";
-import { refreshTokenThroughBroker } from "../shared/broker.ts";
+import { normalizeBrokerBaseUrl, refreshTokenThroughBroker } from "../shared/broker.ts";
 import { readSupabaseConfig } from "../shared/cfg.ts";
 import { createSupabaseLogger } from "../shared/log.ts";
 import type { SupabaseTokenResponse } from "../shared/types.ts";
@@ -26,6 +26,8 @@ type Definition = {
 };
 
 const fromPromise = <A>(run: () => Promise<A>) => Effect.tryPromise({ try: run, catch: (error) => error });
+// ponytail: success entries stay for process lifetime (one per rotation per account);
+// add an LRU cap only if that ever matters.
 const refreshes = new Map<string, Promise<SupabaseTokenResponse>>();
 
 export function setupV2Server(ctx: Context) {
@@ -71,12 +73,18 @@ export function setupV2Server(ctx: Context) {
         refresh: (credential) => fromPromise(async () => {
           const config = readSupabaseConfig(options);
           // Concurrent tools/Locations must not rotate the same refresh token twice.
-          const key = `${config.brokerBaseUrl}:${credential.refresh}`;
+          const key = `${normalizeBrokerBaseUrl(config.brokerBaseUrl)}:${credential.refresh}`;
           let pending = refreshes.get(key);
           if (!pending) {
             pending = refreshTokenThroughBroker(
               { baseUrl: config.brokerBaseUrl }, { refresh_token: credential.refresh }, undefined, logger,
-            ).finally(() => refreshes.delete(key));
+            ).catch((error) => {
+              // Success stays cached: the host persists the rotated credential only after
+              // refresh() resolves, so late resolves with the still-stored old (single-use)
+              // token must converge on the cached result instead of re-sending it.
+              refreshes.delete(key);
+              throw error;
+            });
             refreshes.set(key, pending);
           }
           const tokens = await pending;

src/shared/broker.ts

@@ -91,14 +91,18 @@ function normalizeTokenResponse(payload: unknown): SupabaseTokenResponse {
   };
 }
 
+export function normalizeBrokerBaseUrl(baseUrl: string): string {
+  return baseUrl.replace(/\/$/, "");
+}
+
 async function makeBrokerRequest(
   config: BrokerConfig,
   endpoint: string,
   body: unknown,
   fetchImpl: FetchLike,
   logger?: SupabaseLogger,
 ): Promise<SupabaseTokenResponse> {
-  const url = `${config.baseUrl.replace(/\/$/, "")}${endpoint}`;
+  const url = `${normalizeBrokerBaseUrl(config.baseUrl)}${endpoint}`;
 
   await logger?.debug("supabase broker request started", {
     endpoint,

test/v2-server.test.ts

@@ -23,7 +23,7 @@ afterEach(async () => {
   await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
 });
 
-async function setup() {
+async function setup(optionOverrides: Record<string, unknown> = {}) {
   const directory = await mkdtemp(join(tmpdir(), "supabase-v2-test-"));
   roots.push(directory);
   const scope = await Effect.runPromise(Scope.make());
@@ -34,7 +34,7 @@ async function setup() {
   let account = { type: "oauth", methodID: "supabase", access: "host-A", refresh: "refresh-A", expires: Date.now() + 3600000 } as Credential.OAuth;
   let activeCalls = 0;
   const context = {
-    options: { skills: false, brokerBaseUrl: "https://example.com/broker" },
+    options: { skills: false, brokerBaseUrl: "https://example.com/broker", ...optionOverrides },
     location: { directory, project: { directory } },
     integration: {
       transform: (f: (editor: IntegrationEditor) => void) => Effect.sync(() => f({
@@ -125,6 +125,78 @@ describe("V2 server", () => {
     expect(results.map((r) => r.access)).toEqual(["A-access", "A-access", "B-access"]);
   });
 
+  test("a second resolve in the host persistence gap converges on the completed refresh", async () => {
+    const instance = await setup();
+    const requests: string[] = [];
+    globalThis.fetch = (async (_url, init) => {
+      const token = JSON.parse(String(init?.body)).refresh_token;
+      requests.push(token);
+      if (requests.filter((t) => t === token).length > 1) {
+        return Response.json({ error: { code: "unauthorized", message: "refresh token already used" } }, { status: 401 });
+      }
+      return Response.json({ access_token: "gap-new-access", refresh_token: "gap-new-refresh", expires_in: 3600, token_type: "bearer" });
+    }) as typeof fetch;
+    const old = { type: "oauth", methodID: "supabase", access: "gap-old-access", refresh: "gap-A", expires: 0 } as Credential.OAuth;
+    const first = await Effect.runPromise(instance.method.refresh(old));
+    // The first refresh fully resolved: the broker response has landed, and the host
+    // still has the old token stored (nothing persisted it). The gap is maximal here.
+    const second = await Effect.runPromise(instance.method.refresh({ ...old }));
+    const a = first;
+    const b = second;
+    expect(requests).toEqual(["gap-A"]);
+    expect(a.access).toBe("gap-new-access");
+    expect(b.access).toBe("gap-new-access");
+    expect(b.refresh).toBe("gap-new-refresh");
+  });
+
+  test("failed refreshes release waiters, drop the pending entry, and retry hits the broker again", async () => {
+    const instance = await setup();
+    let calls = 0;
+    globalThis.fetch = (async (_url, _init) => {
+      calls++;
+      return Response.json({ error: { code: "unauthorized", message: "invalid" } }, { status: 401 });
+    }) as typeof fetch;
+    const cred = { type: "oauth", methodID: "supabase", access: "x", refresh: "fail-A", expires: 0 } as Credential.OAuth;
+    const settled = await Promise.allSettled([
+      Effect.runPromise(instance.method.refresh(cred)),
+      Effect.runPromise(instance.method.refresh({ ...cred })),
+    ]);
+    expect(settled.map((s) => s.status)).toEqual(["rejected", "rejected"]);
+    expect(calls).toBe(1);
+    globalThis.fetch = (async (_url, _init) =>
+      Response.json({ access_token: "retry-access", refresh_token: "retry-refresh", expires_in: 3600, token_type: "bearer" })) as typeof fetch;
+    const retried = await Effect.runPromise(instance.method.refresh(cred));
+    expect(retried.access).toBe("retry-access");
+    expect(calls).toBe(1);
+  });
+
+  test("trailing-slash broker aliases share one refresh; distinct brokers and tokens stay isolated", async () => {
+    const plain = await setup();
+    const slashed = await setup({ brokerBaseUrl: "https://example.com/broker/" });
+    const other = await setup({ brokerBaseUrl: "https://other.example.com/broker" });
+    const path2 = await setup({ brokerBaseUrl: "https://example.com/broker2" });
+    const requests: string[] = [];
+    globalThis.fetch = (async (url, init) => {
+      const token = JSON.parse(String(init?.body)).refresh_token;
+      requests.push(`${String(url)}|${token}`);
+      await new Promise((resolve) => setTimeout(resolve, 10));
+      return Response.json({ access_token: `${token}-access`, refresh_token: `${token}-next`, expires_in: 3600, token_type: "bearer" });
+    }) as typeof fetch;
+    const cred = { type: "oauth", methodID: "supabase", access: "expired", refresh: "alias-A", expires: 0 } as Credential.OAuth;
+    const results = await Promise.all([
+      Effect.runPromise(plain.method.refresh(cred)),
+      Effect.runPromise(slashed.method.refresh({ ...cred })),
+      Effect.runPromise(other.method.refresh({ ...cred, refresh: "alias-B" })),
+      Effect.runPromise(path2.method.refresh({ ...cred })),
+    ]);
+    expect(requests).toEqual([
+      "https://example.com/broker/refresh|alias-A",
+      "https://other.example.com/broker/refresh|alias-B",
+      "https://example.com/broker2/refresh|alias-A",
+    ]);
+    expect(results.map((r) => r.access)).toEqual(["alias-A-access", "alias-A-access", "alias-B-access", "alias-A-access"]);
+  });
+
   test("closing one location leaves another location's OAuth pending; attempt scope cancels only its flow", async () => {
     const a = await setup();
     const b = await setup();