feat(core): add DigitalOcean OAuth and router discovery (#47137)

packages/core/src/plugin/provider.ts

@@ -6,6 +6,7 @@ import { CerebrasPlugin } from "./provider/cerebras.js"
 import { CloudflareAIGatewayPlugin } from "./provider/cloudflare-ai-gateway.js"
 import { CloudflareWorkersAIPlugin } from "./provider/cloudflare-workers-ai.js"
 import { CoherePlugin } from "./provider/cohere.js"
+import { DigitalOceanPlugin } from "./provider/digitalocean.js"
 import { DynamicProviderPlugin } from "./provider/dynamic.js"
 import { GatewayPlugin } from "./provider/gateway.js"
 import { GithubCopilotPlugin } from "./provider/github-copilot.js"
@@ -41,6 +42,7 @@ export const ProviderPlugins: PluginInternal.InternalPlugin[] = [
   CloudflareAIGatewayPlugin,
   CloudflareWorkersAIPlugin,
   CoherePlugin,
+  DigitalOceanPlugin,
   GatewayPlugin,
   GithubCopilotPlugin,
   GitLabPlugin,

packages/core/src/plugin/provider/digitalocean.ts

@@ -0,0 +1,199 @@
+import type { IntegrationOAuthMethodRegistration } from "@opencode-ai/plugin/effect/integration"
+import { define } from "@opencode-ai/plugin/effect/plugin"
+import { Clock, Deferred, Effect, Option, Schedule, Schema, Semaphore, Stream } from "effect"
+import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
+import type { Server } from "node:http"
+import { App } from "../../app.js"
+import { Bus } from "../../bus.js"
+import { Credential } from "../../credential.js"
+import { Integration } from "../../integration.js"
+import { Model } from "../../model.js"
+import { OauthCallbackPage } from "../../oauth/page.js"
+import { Provider } from "../../provider.js"
+import type { PluginInternal } from "../internal.js"
+
+const providerID = Provider.ID.make("digitalocean")
+const integrationID = Integration.ID.make("digitalocean")
+const methodID = Integration.MethodID.make("browser")
+
+const clientID = "b1a6c5158156caac821fd1b30253ca8acb52454a48fa744420e41889cb589f82"
+const authorizeEndpoint = "https://cloud.digitalocean.com/v1/oauth/authorize"
+const routersEndpoint = "https://api.digitalocean.com/v2/gen-ai/models/routers"
+const callbackPort = 1456
+const callbackPath = "/auth/callback"
+const tokenPath = "/auth/token"
+const RoutersResponse = Schema.Struct({
+  model_routers: Schema.Array(Schema.Struct({ name: Schema.NonEmptyString })),
+})
+const decodeCallback = Schema.decodeUnknownOption(
+  Schema.fromJsonString(
+    Schema.Struct({
+      access_token: Schema.optional(Schema.String),
+      expires_in: Schema.optional(Schema.String),
+      state: Schema.optional(Schema.String),
+      error: Schema.optional(Schema.String),
+      error_description: Schema.optional(Schema.String),
+    }),
+  ),
+)
+
+const oauth = {
+  integrationID,
+  method: {
+    id: methodID,
+    type: "oauth",
+    label: "Login with DigitalOcean",
+  },
+  authorize: () =>
+    Effect.gen(function* () {
+      const state = Buffer.from(crypto.getRandomValues(new Uint8Array(32))).toString("base64url")
+      const token = yield* Deferred.make<{ access: string; expiresIn: number }, Error>()
+      // Lazy so runtimes without a loopback listener (workerd) never evaluate node:http.
+      const { createServer } = yield* Effect.promise(() => import("node:http"))
+      const server = createServer((request, response) => {
+        const url = new URL(request.url ?? "/", "http://localhost")
+        if (request.method === "GET" && url.pathname === callbackPath) {
+          response
+            .writeHead(200, { "Content-Type": "text/html" })
+            .end(OauthCallbackPage.bootstrap({ tokenPath, provider: "DigitalOcean" }))
+          return
+        }
+        if (request.method === "POST" && url.pathname === tokenPath) {
+          const chunks: Buffer[] = []
+          request.on("data", (chunk: Buffer) => chunks.push(chunk))
+          request.on("end", () => {
+            const body = Option.getOrUndefined(decodeCallback(Buffer.concat(chunks).toString("utf8")))
+            const error = body?.error_description || body?.error
+            const access = body?.access_token
+            if (error || !access || body?.state !== state) {
+              const message = error || (access ? "Invalid OAuth state" : "Missing access token")
+              Effect.runFork(Deferred.fail(token, new Error(message)))
+              response.writeHead(400, { "Content-Type": "application/json" }).end(JSON.stringify({ error: message }))
+              return
+            }
+            const expires = Number.parseInt(body.expires_in ?? "0", 10)
+            Effect.runFork(
+              Deferred.succeed(token, {
+                access,
+                expiresIn: Number.isFinite(expires) && expires > 0 ? expires : 60 * 60 * 24 * 30,
+              }),
+            )
+            response.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ ok: true }))
+          })
+          return
+        }
+        response.writeHead(404).end("Not found")
+      })
+      const port = yield* listen(server)
+      yield* Effect.addFinalizer(() => Effect.sync(() => server.close()))
+      const redirect = `http://localhost:${port}${callbackPath}`
+      return {
+        mode: "auto" as const,
+        url: `${authorizeEndpoint}?${new URLSearchParams({
+          response_type: "token",
+          client_id: clientID,
+          redirect_uri: redirect,
+          scope: "genai:read inference:query",
+          state,
+        })}`,
+        instructions: "Complete authorization in your browser. This window will close automatically.",
+        callback: Effect.gen(function* () {
+          const payload = yield* Deferred.await(token)
+          return Credential.OAuth.make({
+            type: "oauth",
+            methodID,
+            access: payload.access,
+            refresh: "",
+            expires: (yield* Clock.currentTimeMillis) + payload.expiresIn * 1000,
+          })
+        }),
+      }
+    }),
+} satisfies IntegrationOAuthMethodRegistration
+
+export const DigitalOceanPlugin = define({
+  id: "opencode.provider.digitalocean",
+  effect: Effect.fn(function* (ctx) {
+    const bus = yield* Bus.Service
+    const http = HttpClient.filterStatusOk(yield* HttpClient.HttpClient)
+    const loading = Semaphore.makeUnsafe(1)
+    const loaded: { access?: string; routers: (typeof RoutersResponse.Type)["model_routers"] } = { routers: [] }
+
+    const load = Effect.fn("DigitalOceanPlugin.load")(function* () {
+      const connection = yield* ctx.integration.connection.active(integrationID)
+      const credential = connection
+        ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined))
+        : undefined
+      if (credential?.type !== "oauth" || credential.methodID !== methodID) {
+        loaded.access = undefined
+        loaded.routers = []
+        return
+      }
+      if (loaded.access !== credential.access) {
+        loaded.access = credential.access
+        loaded.routers = []
+      }
+      if (credential.expires !== 0 && credential.expires <= (yield* Clock.currentTimeMillis)) return
+      const result = yield* http
+        .execute(
+          HttpClientRequest.get(routersEndpoint).pipe(
+            HttpClientRequest.bearerToken(credential.access),
+            HttpClientRequest.acceptJson,
+            HttpClientRequest.setHeader("User-Agent", App.useragent(ctx.app)),
+          ),
+        )
+        .pipe(
+          Effect.flatMap(HttpClientResponse.schemaBodyJson(RoutersResponse)),
+          Effect.timeout("10 seconds"),
+          Effect.catch((cause) =>
+            Effect.logWarning("failed to sync DigitalOcean routers", { cause }).pipe(Effect.as(undefined)),
+          ),
+        )
+      if (result) loaded.routers = result.model_routers
+    })
+
+    yield* ctx.integration.transform((draft) => {
+      draft.method.update(oauth)
+    })
+    yield* ctx.catalog.transform((evt) => {
+      if (!evt.provider.get(providerID)) return
+      for (const router of loaded.routers) {
+        const id = `router:${router.name}`
+        if (evt.model.get(providerID, id)) continue
+        evt.model.update(providerID, id, (draft) => {
+          draft.name = router.name
+          draft.family = Model.Family.make("digitalocean-inference-routers")
+          draft.capabilities = { tools: true, input: ["text"], output: ["text"] }
+          draft.limit = { context: 128_000, output: 8_192 }
+        })
+      }
+    })
+    const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload())))
+    yield* bus.subscribe(Credential.Event.Switched).pipe(
+      Stream.filter((event) => event.data.integrationID === integrationID),
+      Stream.runForEach(refresh),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    // Retain the last successful inventory for this credential through transient failures.
+    yield* refresh().pipe(Effect.repeat(Schedule.spaced("5 minutes")), Effect.forkScoped)
+  }),
+} satisfies PluginInternal.InternalPlugin)
+
+function listen(server: Server) {
+  return Effect.callback<number, Error>((resume) => {
+    const onError = (error: Error) => resume(Effect.fail(error))
+    server.once("error", onError)
+    server.listen(callbackPort, "localhost", () => {
+      server.off("error", onError)
+      resume(Effect.succeed(callbackPort))
+    })
+  }).pipe(
+    Effect.mapError((cause) =>
+      "code" in cause && cause.code === "EADDRINUSE"
+        ? new Error(
+            `DigitalOcean login needs local port ${callbackPort}, but it is already in use. Stop the process using that port and try again.`,
+          )
+        : cause,
+    ),
+  )
+}

packages/core/test/plugin/provider-digitalocean.test.ts

@@ -0,0 +1,295 @@
+import { describe, expect, test } from "bun:test"
+import { Clock, Effect, Schedule } from "effect"
+import { TestClock } from "effect/testing"
+import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
+import { Catalog } from "@opencode-ai/core/catalog"
+import { Credential } from "@opencode-ai/core/credential"
+import { Integration } from "@opencode-ai/core/integration"
+import { Model } from "@opencode-ai/core/model"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { PluginHost } from "@opencode-ai/core/plugin/host"
+import { Provider } from "@opencode-ai/core/provider"
+import { ProviderPlugins } from "@opencode-ai/core/plugin/provider"
+import { DigitalOceanPlugin } from "@opencode-ai/core/plugin/provider/digitalocean"
+import { drain } from "../lib/clock"
+import { testEffect } from "../lib/effect"
+import { PluginTestLayer } from "./fixture"
+
+const it = testEffect(PluginTestLayer)
+const providerID = Provider.ID.make("digitalocean")
+const integrationID = Integration.ID.make("digitalocean")
+const methodID = Integration.MethodID.make("browser")
+
+const addPlugin = Effect.fn(function* () {
+  const plugin = yield* Plugin.Service
+  const host = yield* PluginHost.make(plugin)
+  yield* DigitalOceanPlugin.effect(host)
+})
+
+const oauthCredential = Effect.fn(function* (access = "do-token", expires = 0) {
+  const credentials = yield* Credential.Service
+  return yield* credentials.create({
+    integrationID,
+    value: Credential.OAuth.make({ type: "oauth", methodID, access, refresh: "", expires }),
+  })
+})
+
+const discovery = Effect.gen(function* () {
+  const catalog = yield* Catalog.Service
+  yield* catalog.transform((draft) => {
+    draft.provider.update(providerID, (provider) => {
+      provider.package = Provider.aisdk("@ai-sdk/openai-compatible")
+      provider.settings = { baseURL: "https://inference.do-ai.run/v1" }
+    })
+    draft.model.update(providerID, Model.ID.make("snapshot-model"), () => {})
+    draft.model.update(providerID, Model.ID.make("router:configured"), (model) => {
+      model.name = "Configured router"
+      model.limit.context = 256_000
+    })
+    draft.model.update(Provider.ID.openai, Model.ID.make("router:alpha"), () => {})
+  })
+  const remote: { status: number; body: unknown; requests: HttpClientRequest.HttpClientRequest[] } = {
+    status: 200,
+    body: { model_routers: [{ name: "alpha" }, { name: "configured" }] },
+    requests: [],
+  }
+  const http = HttpClient.make((request) =>
+    Effect.sync(() => {
+      remote.requests.push(request)
+      return HttpClientResponse.fromWeb(request, Response.json(remote.body, { status: remote.status }))
+    }),
+  )
+  return { catalog, remote, install: addPlugin().pipe(Effect.provideService(HttpClient.HttpClient, http)) }
+})
+
+const status = Effect.fn(function* (attempt: Integration.Attempt) {
+  const integrations = yield* Integration.Service
+  return yield* integrations.oauth
+    .status({ integrationID, attemptID: attempt.attemptID })
+    .pipe(
+      Effect.repeat({ until: (value) => value.status !== "pending", schedule: Schedule.spaced("10 millis") }),
+      Effect.timeout("3 seconds"),
+    )
+})
+
+describe("DigitalOceanPlugin", () => {
+  test("is registered alongside the other provider plugins", () => {
+    expect(ProviderPlugins.map((item) => item.id)).toContain("opencode.provider.digitalocean")
+  })
+
+  it.effect("registers browser OAuth alongside generic key and environment methods", () =>
+    Effect.gen(function* () {
+      const integrations = yield* Integration.Service
+      yield* integrations.transform((draft) => {
+        draft.method.update({ integrationID, method: { type: "key" } })
+        draft.method.update({ integrationID, method: { type: "env", names: ["DIGITALOCEAN_ACCESS_TOKEN"] } })
+      })
+      yield* addPlugin()
+      expect((yield* integrations.get(integrationID))?.methods).toEqual([
+        { type: "key" },
+        { type: "env", names: ["DIGITALOCEAN_ACCESS_TOKEN"] },
+        { id: methodID, type: "oauth", label: "Login with DigitalOcean" },
+      ])
+    }),
+  )
+
+  it.effect("refreshes routers, retains the latest success on errors, and accepts an empty inventory", () =>
+    Effect.gen(function* () {
+      const fixture = yield* discovery
+      yield* oauthCredential()
+      yield* fixture.install
+      yield* drain
+
+      expect(fixture.remote.requests).toHaveLength(1)
+      expect(fixture.remote.requests[0]).toMatchObject({
+        method: "GET",
+        url: "https://api.digitalocean.com/v2/gen-ai/models/routers",
+        headers: { authorization: "Bearer do-token", accept: "application/json", "user-agent": expect.any(String) },
+      })
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:alpha"))).toMatchObject({
+        name: "alpha",
+        family: "digitalocean-inference-routers",
+        package: "aisdk:@ai-sdk/openai-compatible",
+        settings: { baseURL: "https://inference.do-ai.run/v1" },
+        capabilities: { tools: true, input: ["text"], output: ["text"] },
+        limit: { context: 128_000, output: 8_192 },
+      })
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:configured"))).toMatchObject({
+        name: "Configured router",
+        limit: { context: 256_000 },
+      })
+
+      fixture.remote.body = { model_routers: [{ name: "beta" }] }
+      yield* TestClock.adjust("4 minutes")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(1)
+      yield* TestClock.adjust("1 minute")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(2)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:alpha"))).toBeUndefined()
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeDefined()
+
+      fixture.remote.status = 503
+      yield* TestClock.adjust("5 minutes")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(3)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeDefined()
+
+      fixture.remote.status = 200
+      fixture.remote.body = { model_routers: [{ name: 123 }] }
+      yield* TestClock.adjust("5 minutes")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(4)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeDefined()
+
+      fixture.remote.body = { model_routers: [] }
+      yield* TestClock.adjust("5 minutes")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(5)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeUndefined()
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("snapshot-model"))).toBeDefined()
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:configured"))).toBeDefined()
+      expect(yield* fixture.catalog.model.get(Provider.ID.openai, Model.ID.make("router:alpha"))).toBeDefined()
+    }),
+  )
+
+  it.effect("loads on credential changes and clears the previous account's routers", () =>
+    Effect.gen(function* () {
+      const fixture = yield* discovery
+      const credentials = yield* Credential.Service
+      yield* fixture.install
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(0)
+
+      yield* oauthCredential()
+      yield* drain
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:alpha"))).toBeDefined()
+
+      fixture.remote.status = 503
+      const second = yield* oauthCredential("other-account")
+      yield* drain
+      expect(fixture.remote.requests.at(-1)?.headers.authorization).toBe("Bearer other-account")
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:alpha"))).toBeUndefined()
+
+      fixture.remote.status = 200
+      fixture.remote.body = { model_routers: [{ name: "beta" }] }
+      yield* TestClock.adjust("5 minutes")
+      yield* drain
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeDefined()
+
+      const count = fixture.remote.requests.length
+      yield* credentials.create({ integrationID, value: Credential.Key.make({ type: "key", key: "do-key" }) })
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(count)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeUndefined()
+
+      yield* credentials.activate(second.id)
+      yield* drain
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeDefined()
+      yield* credentials.remove(second.id)
+      yield* drain
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:beta"))).toBeUndefined()
+    }),
+  )
+
+  it.effect("stops discovery when the token expires while retaining its last successful inventory", () =>
+    Effect.gen(function* () {
+      const fixture = yield* discovery
+      yield* oauthCredential("do-token", (yield* Clock.currentTimeMillis) + 60_000)
+      yield* fixture.install
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(1)
+      yield* TestClock.adjust("5 minutes")
+      yield* drain
+      expect(fixture.remote.requests).toHaveLength(1)
+      expect(yield* fixture.catalog.model.get(providerID, Model.ID.make("router:alpha"))).toBeDefined()
+    }),
+  )
+
+  it.live("completes browser OAuth independently of router discovery and releases the listener", () =>
+    Effect.gen(function* () {
+      const fixture = yield* discovery
+      fixture.remote.status = 503
+      yield* fixture.install
+      const integrations = yield* Integration.Service
+      const credentials = yield* Credential.Service
+      const attempt = yield* integrations.oauth.connect({ integrationID, methodID })
+      const url = new URL(attempt.url)
+      expect(attempt.mode).toBe("auto")
+      expect(url.origin + url.pathname).toBe("https://cloud.digitalocean.com/v1/oauth/authorize")
+      expect(url.searchParams.get("response_type")).toBe("token")
+      expect(url.searchParams.get("client_id")).toBe("b1a6c5158156caac821fd1b30253ca8acb52454a48fa744420e41889cb589f82")
+      expect(url.searchParams.get("scope")).toBe("genai:read inference:query")
+      expect(url.searchParams.get("redirect_uri")).toBe("http://localhost:1456/auth/callback")
+      expect(url.searchParams.get("state")).toBeTruthy()
+      const page = yield* Effect.promise(() =>
+        fetch("http://localhost:1456/auth/callback", { headers: { Connection: "close" } }).then((res) => res.text()),
+      )
+      expect(page).toContain("/auth/token")
+      const now = Date.now()
+      const response = yield* Effect.promise(() =>
+        fetch("http://localhost:1456/auth/token", {
+          method: "POST",
+          headers: { "Content-Type": "application/json", Connection: "close" },
+          body: JSON.stringify({ access_token: "do-token", expires_in: "3600", state: url.searchParams.get("state") }),
+        }),
+      )
+      expect(response.status).toBe(200)
+      expect(yield* status(attempt)).toMatchObject({ status: "complete" })
+      const saved = (yield* credentials.list(integrationID))[0]?.value
+      expect(saved).toMatchObject({ type: "oauth", methodID, access: "do-token", refresh: "" })
+      expect(saved?.metadata).toBeUndefined()
+      if (saved?.type !== "oauth") throw new Error("Expected OAuth credential")
+      expect(saved.expires).toBeGreaterThanOrEqual(now + 3_600_000)
+      expect(saved.expires).toBeLessThanOrEqual(Date.now() + 3_600_000)
+
+      const next = yield* integrations.oauth.connect({ integrationID, methodID })
+      expect(new URL(next.url).searchParams.get("state")).not.toBe(url.searchParams.get("state"))
+      yield* integrations.oauth.cancel({ integrationID, attemptID: next.attemptID })
+      const retry = yield* integrations.oauth.connect({ integrationID, methodID })
+      yield* integrations.oauth.cancel({ integrationID, attemptID: retry.attemptID })
+    }),
+  )
+
+  const invalid = [
+    {
+      name: "mismatched state",
+      body: JSON.stringify({ access_token: "token", state: "wrong" }),
+      message: "Invalid OAuth state",
+    },
+    { name: "missing token", body: JSON.stringify({}), message: "Missing access token" },
+    { name: "malformed JSON", body: "not-json", message: "Missing access token" },
+    {
+      name: "provider denial",
+      body: JSON.stringify({ error: "access_denied", error_description: "Denied" }),
+      message: "Denied",
+    },
+    {
+      name: "provider denial without a description",
+      body: JSON.stringify({ error: "access_denied", error_description: "" }),
+      message: "access_denied",
+    },
+  ]
+  invalid.forEach((fixture) => {
+    it.live(`rejects ${fixture.name} and releases the listener`, () =>
+      Effect.gen(function* () {
+        yield* addPlugin()
+        const integrations = yield* Integration.Service
+        const credentials = yield* Credential.Service
+        const attempt = yield* integrations.oauth.connect({ integrationID, methodID })
+        const response = yield* Effect.promise(() =>
+          fetch("http://localhost:1456/auth/token", {
+            method: "POST",
+            headers: { "Content-Type": "application/json", Connection: "close" },
+            body: fixture.body,
+          }),
+        )
+        expect(response.status).toBe(400)
+        expect(yield* status(attempt)).toMatchObject({ status: "failed", message: fixture.message })
+        expect(yield* credentials.list(integrationID)).toEqual([])
+        const next = yield* integrations.oauth.connect({ integrationID, methodID })
+        yield* integrations.oauth.cancel({ integrationID, attemptID: next.attemptID })
+      }),
+    )
+  })
+})