diff --git a/packages/core/src/plugin/provider.ts b/packages/core/src/plugin/provider.ts index 18ef4e9310d..25c9b146272 100644 --- a/packages/core/src/plugin/provider.ts +++ b/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, diff --git a/packages/core/src/plugin/provider/digitalocean.ts b/packages/core/src/plugin/provider/digitalocean.ts new file mode 100644 index 00000000000..e2da1af4ab3 --- /dev/null +++ b/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((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, + ), + ) +} diff --git a/packages/core/test/plugin/provider-digitalocean.test.ts b/packages/core/test/plugin/provider-digitalocean.test.ts new file mode 100644 index 00000000000..3323fa34bb5 --- /dev/null +++ b/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 }) + }), + ) + }) +})