fix(core): dedupe credential refreshes

packages/core/src/integration.ts

@@ -1,6 +1,7 @@
 export * as Integration from "./integration"
 
 import {
+  Cache,
   Cause,
   Clock,
   Context,
@@ -310,6 +311,25 @@ export const locationLayer = Layer.effect(
     const authorize = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
       effect.pipe(Effect.mapError((cause) => new AuthorizationError({ cause })))
 
+    const refreshes = yield* Cache.make<Credential.ID, Credential.Value | undefined, AuthorizationError>({
+      capacity: Number.POSITIVE_INFINITY,
+      timeToLive: Duration.zero,
+      lookup: Effect.fnUntraced(function* (credentialID) {
+        const credential = yield* credentials.get(credentialID)
+        if (!credential || credential.value.type === "key") return credential?.value
+        const implementation = state
+          .get()
+          .integrations.get(credential.integrationID)
+          ?.implementations.get(credential.value.methodID)
+        if (!implementation?.refresh) return credential.value
+        const now = yield* Clock.currentTimeMillis
+        if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value
+        const value = yield* authorize(implementation.refresh(credential.value))
+        yield* credentials.update(credentialID, { value })
+        return value
+      }),
+    })
+
     const close = (attemptScope: Scope.Closeable) =>
       Scope.close(attemptScope, Exit.void).pipe(Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
 
@@ -396,9 +416,7 @@ export const locationLayer = Layer.effect(
           if (!implementation?.refresh) return credential.value
           const now = yield* Clock.currentTimeMillis
           if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value
-          const value = yield* authorize(implementation.refresh(credential.value))
-          yield* credentials.update(credential.id, { value })
-          return value
+          return yield* Cache.get(refreshes, credential.id)
         }),
         key: Effect.fn("Integration.connection.key")(function* (input) {
           const method = state

packages/core/test/integration.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { Duration, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect"
+import { Deferred, Duration, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect"
 import * as TestClock from "effect/testing/TestClock"
 import { Integration } from "@opencode-ai/core/integration"
 import { Credential } from "@opencode-ai/core/credential"
@@ -346,4 +346,116 @@ describe("Integration", () => {
         }),
     )
   })
+
+  it.effect("shares concurrent OAuth credential refreshes", () =>
+    Effect.gen(function* () {
+      const integrations = yield* Integration.Service
+      const credentials = yield* Credential.Service
+      const integrationID = Integration.ID.make("openai")
+      const methodID = Integration.MethodID.make("chatgpt")
+      const started = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      let refreshes = 0
+      const value = Credential.OAuth.make({
+        type: "oauth",
+        methodID,
+        access: "refreshed",
+        refresh: "refresh-2",
+        expires: Duration.toMillis(Duration.hours(1)),
+      })
+      yield* integrations.transform((editor) =>
+        editor.method.update({
+          integrationID,
+          method: { id: methodID, type: "oauth", label: "ChatGPT" },
+          authorize: () => Effect.die("unexpected authorization"),
+          refresh: () =>
+            Effect.sync(() => refreshes++).pipe(
+              Effect.andThen(Deferred.succeed(started, undefined)),
+              Effect.andThen(Deferred.await(release)),
+              Effect.as(value),
+            ),
+        }),
+      )
+      const credential = yield* credentials.create({
+        integrationID,
+        value: Credential.OAuth.make({
+          type: "oauth",
+          methodID,
+          access: "expired",
+          refresh: "refresh-1",
+          expires: 0,
+        }),
+      })
+      const connection = { type: "credential" as const, id: credential.id, label: credential.label }
+
+      const first = yield* integrations.connection.resolve(connection).pipe(Effect.forkChild)
+      yield* Deferred.await(started)
+      const second = yield* integrations.connection.resolve(connection).pipe(Effect.forkChild)
+      yield* Effect.yieldNow
+      expect(refreshes).toBe(1)
+
+      yield* Deferred.succeed(release, undefined)
+      expect(yield* Effect.all([Fiber.join(first), Fiber.join(second)], { concurrency: "unbounded" })).toEqual([
+        value,
+        value,
+      ])
+      expect(refreshes).toBe(1)
+      expect((yield* credentials.get(credential.id))?.value).toEqual(value)
+    }),
+  )
+
+  it.effect("shares concurrent refresh failures and retries later", () =>
+    Effect.gen(function* () {
+      const integrations = yield* Integration.Service
+      const credentials = yield* Credential.Service
+      const integrationID = Integration.ID.make("openai")
+      const methodID = Integration.MethodID.make("chatgpt")
+      const started = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      const failure = new Error("refresh failed")
+      let refreshes = 0
+      yield* integrations.transform((editor) =>
+        editor.method.update({
+          integrationID,
+          method: { id: methodID, type: "oauth", label: "ChatGPT" },
+          authorize: () => Effect.die("unexpected authorization"),
+          refresh: () =>
+            Effect.sync(() => refreshes++).pipe(
+              Effect.andThen(Deferred.succeed(started, undefined)),
+              Effect.andThen(Deferred.await(release)),
+              Effect.andThen(Effect.fail(failure)),
+            ),
+        }),
+      )
+      const credential = yield* credentials.create({
+        integrationID,
+        value: Credential.OAuth.make({
+          type: "oauth",
+          methodID,
+          access: "expired",
+          refresh: "refresh",
+          expires: 0,
+        }),
+      })
+      const connection = { type: "credential" as const, id: credential.id, label: credential.label }
+
+      const first = yield* integrations.connection.resolve(connection).pipe(Effect.flip, Effect.forkChild)
+      yield* Deferred.await(started)
+      const second = yield* integrations.connection.resolve(connection).pipe(Effect.flip, Effect.forkChild)
+      yield* Effect.yieldNow
+      expect(refreshes).toBe(1)
+
+      yield* Deferred.succeed(release, undefined)
+      const results = yield* Effect.all([Fiber.join(first), Fiber.join(second)], { concurrency: "unbounded" })
+      expect(results).toEqual([
+        new Integration.AuthorizationError({ cause: failure }),
+        new Integration.AuthorizationError({ cause: failure }),
+      ])
+
+      expect(yield* integrations.connection.resolve(connection).pipe(Effect.flip)).toEqual(
+        new Integration.AuthorizationError({ cause: failure }),
+      )
+      expect(refreshes).toBe(2)
+    }),
+  )
 })