feat(plugin): add durable storage API (#43525)

2026-08-19 17:51:58 -04:00
parent 6adb98c266
commit 98a9d864e6
16 changed files with 273 additions and 4 deletions
@@ -1,6 +1,6 @@
 export * as KV from "./kv.js"

-import { eq } from "drizzle-orm"
+import { and, asc, eq, gt, gte, lt } from "drizzle-orm"
 import { Context, Effect, Layer, Schema } from "effect"
 import { Database } from "./database/database.js"
 import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
@@ -8,10 +8,27 @@ import { KVTable } from "./kv/sql.js"

 export type Value = Schema.Json

+export interface Entry {
+  readonly key: string
+  readonly value: Value
+}
+
+export interface ScanOptions {
+  readonly prefix: string
+  readonly after?: string
+  readonly limit?: number
+}
+
+export interface ScanResult {
+  readonly entries: readonly Entry[]
+  readonly next?: string
+}
+
 export interface Interface {
  readonly get: (key: string) => Effect.Effect<Value | undefined>
  readonly set: (key: string, value: Value) => Effect.Effect<void>
  readonly remove: (key: string) => Effect.Effect<void>
+  readonly scan: (options: ScanOptions) => Effect.Effect<ScanResult>
 }

 export class Service extends Context.Service<Service, Interface>()("@opencode/KV") {}
@@ -40,8 +57,36 @@ const layer = Layer.effect(
      remove: Effect.fn("KV.remove")(function* (key) {
        yield* db.delete(KVTable).where(eq(KVTable.key, key)).run().pipe(Effect.orDie)
      }),
+      scan: Effect.fn("KV.scan")(function* (options) {
+        const limit = Number.isNaN(options.limit) ? 100 : Math.min(Math.max(Math.floor(options.limit ?? 100), 1), 1000)
+        const end = prefixEnd(options.prefix)
+        const rows = yield* db
+          .select({ key: KVTable.key, value: KVTable.value })
+          .from(KVTable)
+          .where(
+            and(
+              options.prefix === "" ? undefined : gte(KVTable.key, options.prefix),
+              end === undefined ? undefined : lt(KVTable.key, end),
+              options.after === undefined ? undefined : gt(KVTable.key, options.after),
+            ),
+          )
+          .orderBy(asc(KVTable.key))
+          .limit(limit + 1)
+          .all()
+          .pipe(Effect.orDie)
+        const entries = rows.slice(0, limit)
+        if (rows.length <= limit) return { entries }
+        return { entries, next: entries[entries.length - 1].key }
+      }),
    })
  }),
 )

+function prefixEnd(prefix: string) {
+  const points = Array.from(prefix)
+  const index = points.findLastIndex((value) => value.codePointAt(0)! < 0x10ffff)
+  if (index < 0) return undefined
+  return `${points.slice(0, index).join("")}${String.fromCodePoint(points[index].codePointAt(0)! + 1)}`
+}
+
 export const node = makeGlobalNode({ service: Service, layer, deps: [Database.node] })
@@ -11,6 +11,7 @@ import { Catalog } from "./catalog.js"
 import { Command } from "./command.js"
 import { Bus } from "./bus.js"
 import { Integration } from "./integration.js"
+import { KV } from "./kv.js"
 import { MCP } from "./mcp/index.js"
 import { Location } from "./location.js"
 import { PluginHost } from "./plugin/host.js"
@@ -41,16 +42,18 @@ const layer = Layer.effect(
  Service,
  Effect.gen(function* () {
    const bus = yield* Bus.Service
+    const kv = yield* KV.Service
    const scope = yield* Scope.make()
    const active = new Map<Plugin.ID, { readonly plugin: Versioned; readonly scope: Scope.Closeable }>()
    const lock = Semaphore.makeUnsafe(1)
    let inventory: Plugin.Info[] = []
    let host: Parameters<import("@opencode-ai/plugin/effect/plugin").Plugin["effect"]>[0]
-
    const load = Effect.fnUntraced(function* (plugin: Versioned) {
      const child = yield* Scope.fork(scope)
      const inherit = yield* State.inherit()
-      const loaded = yield* Effect.suspend(() => plugin.effect(host)).pipe(
+      const loaded = yield* Effect.suspend(() =>
+        plugin.effect({ ...host, storage: PluginHost.storage(kv, plugin.id) }),
+      ).pipe(
        inherit,
        Effect.updateContext((context: Context.Context<never>) =>
          Context.make(Scope.Scope, child).pipe(
@@ -189,6 +192,7 @@ export const node = makeLocationNode({
    Catalog.node,
    Command.node,
    Integration.node,
+    KV.node,
    MCP.node,
    Location.node,
    Reference.node,
@@ -14,6 +14,7 @@ import { Command } from "../command.js"
 import { Credential } from "../credential.js"
 import { Bus } from "../bus.js"
 import { Integration } from "../integration.js"
+import { KV } from "../kv.js"
 import { Location } from "../location.js"
 import { Model } from "../model.js"
 import { MCP } from "../mcp/index.js"
@@ -28,7 +29,10 @@ import { WebSearch } from "../websearch.js"
 import { PluginHooks } from "./hooks.js"

 const mutable = <T>(value: T) => value as DeepMutable<T>
-export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../plugin.js").Interface) {
+export const make = Effect.fn("PluginHost.make")(function* (
+  plugin: import("../plugin.js").Interface,
+  pluginID: string = "test",
+) {
  const app = yield* App.Metadata
  const agents = yield* Agent.Service
  const aisdk = yield* AISDK.Service
@@ -36,6 +40,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
  const commands = yield* Command.Service
  const bus = yield* Bus.Service
  const integration = yield* Integration.Service
+  const kv = yield* KV.Service
  const mcp = yield* MCP.Service
  const location = yield* Location.Service
  const reference = yield* Reference.Service
@@ -340,6 +345,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
          })
        }),
    },
+    storage: storage(kv, pluginID),
    shell: {
      hook: (name, callback) => hooks.register("shell", name, callback),
    },
@@ -406,6 +412,35 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
  } satisfies Plugin.Context
 })

+export function storage(kv: KV.Interface, pluginID: string): Plugin.Context["storage"] {
+  const namespace = `plugin:${pluginID
+    .split("")
+    .map((value) => value.charCodeAt(0).toString(16).padStart(4, "0"))
+    .join("")}:`
+  return {
+    get: (key) => kv.get(namespace + key),
+    set: (key, value) => kv.set(namespace + key, value),
+    remove: (key) => kv.remove(namespace + key),
+    scan: (options) =>
+      kv
+        .scan({
+          prefix: namespace + options.prefix,
+          after: options.after === undefined ? undefined : namespace + options.after,
+          limit: options.limit,
+        })
+        .pipe(
+          Effect.map((result) => {
+            const entries = result.entries.map((entry) => ({
+              key: entry.key.slice(namespace.length),
+              value: entry.value,
+            }))
+            if (result.next === undefined) return { entries }
+            return { entries, next: result.next.slice(namespace.length) }
+          }),
+        ),
+  }
+}
+
 function methodImplementation(input: IntegrationMethodRegistration): Integration.Implementation {
  if ("authorize" in input) {
    const refresh = input.refresh
@@ -18,8 +18,64 @@ describe("KV", () => {
      yield* kv.set("wellknown:sources", ["https://example.com", "https://example.org"])
      expect(yield* kv.get("wellknown:sources")).toEqual(["https://example.com", "https://example.org"])

+      yield* kv.remove("wellknown:sources")
      yield* kv.remove("wellknown:sources")
      expect(yield* kv.get("wellknown:sources")).toBeUndefined()
    }),
  )
+
+  it.effect("scans prefixes in deterministic pages", () =>
+    Effect.gen(function* () {
+      const kv = yield* KV.Service
+      const prefix = "scan:%_:/雪/"
+      yield* Effect.forEach(
+        [
+          [`${prefix}beta`, { order: 2 }],
+          [`${prefix}alpha`, { order: 1 }],
+          [`${prefix}éclair`, { order: 3 }],
+          ["scan:other", { order: 0 }],
+        ] as const,
+        ([key, value]) => kv.set(key, value),
+        { discard: true },
+      )
+
+      const first = yield* kv.scan({ prefix, limit: 2 })
+      expect(first).toEqual({
+        entries: [
+          { key: `${prefix}alpha`, value: { order: 1 } },
+          { key: `${prefix}beta`, value: { order: 2 } },
+        ],
+        next: `${prefix}beta`,
+      })
+      expect(yield* kv.scan({ prefix, after: first.next, limit: 2 })).toEqual({
+        entries: [{ key: `${prefix}éclair`, value: { order: 3 } }],
+      })
+      expect(yield* kv.scan({ prefix: `${prefix}%_` })).toEqual({ entries: [] })
+    }),
+  )
+
+  it.effect("defaults, normalizes, and caps scan limits", () =>
+    Effect.gen(function* () {
+      const kv = yield* KV.Service
+      const prefix = "scan:limits/"
+      yield* Effect.forEach(
+        Array.from({ length: 1001 }, (_, index) => `${prefix}${index.toString().padStart(4, "0")}`),
+        (key) => kv.set(key, key),
+        { discard: true },
+      )
+
+      const defaultPage = yield* kv.scan({ prefix })
+      expect(defaultPage.entries).toHaveLength(100)
+      expect(defaultPage.next).toBe(`${prefix}0099`)
+
+      const cappedPage = yield* kv.scan({ prefix, limit: 10_000 })
+      expect(cappedPage.entries).toHaveLength(1000)
+      expect(cappedPage.next).toBe(`${prefix}0999`)
+
+      expect((yield* kv.scan({ prefix, limit: 2.9 })).entries).toHaveLength(2)
+      expect((yield* kv.scan({ prefix, limit: 0 })).entries).toHaveLength(1)
+      expect((yield* kv.scan({ prefix, limit: -10 })).entries).toHaveLength(1)
+      expect((yield* kv.scan({ prefix, limit: Number.NaN })).entries).toHaveLength(100)
+    }),
+  )
 })
@@ -338,6 +338,54 @@ describe("Plugin", () => {
    }),
  )

+  it.effect("provides isolated durable storage for each plugin ID", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const storage = new Map<string, EffectPlugin.Context["storage"]>()
+      yield* plugins.activate(
+        ["a", "a:b", "雪"].map((id) => ({
+          id,
+          version: "1",
+          effect: (context: EffectPlugin.Context) => Effect.sync(() => storage.set(id, context.storage)),
+        })),
+      )
+      const first = storage.get("a")
+      const second = storage.get("a:b")
+      const unicode = storage.get("雪")
+      if (!first || !second || !unicode) return yield* Effect.die("plugin storage was not activated")
+
+      yield* first.set("b:c", { plugin: "a" })
+      yield* second.set("c", { plugin: "a:b" })
+      yield* unicode.set("c", { plugin: "雪" })
+      expect(yield* first.get("b:c")).toEqual({ plugin: "a" })
+      expect(yield* second.get("c")).toEqual({ plugin: "a:b" })
+      expect(yield* unicode.get("c")).toEqual({ plugin: "雪" })
+      expect(yield* first.get("c")).toBeUndefined()
+
+      const prefix = "%_:/雪/"
+      yield* first.set(`${prefix}beta`, [2])
+      yield* first.set(`${prefix}alpha`, [1])
+      const firstPage = yield* first.scan({ prefix, limit: 1 })
+      expect(firstPage).toEqual({ entries: [{ key: `${prefix}alpha`, value: [1] }], next: `${prefix}alpha` })
+      expect(yield* first.scan({ prefix, after: firstPage.next, limit: 1 })).toEqual({
+        entries: [{ key: `${prefix}beta`, value: [2] }],
+      })
+      expect(yield* first.scan({ prefix: `${prefix}%_` })).toEqual({ entries: [] })
+      expect(yield* first.scan({ prefix: "" })).toEqual({
+        entries: [
+          { key: `${prefix}alpha`, value: [1] },
+          { key: `${prefix}beta`, value: [2] },
+          { key: "b:c", value: { plugin: "a" } },
+        ],
+      })
+
+      yield* first.remove("b:c")
+      yield* first.remove("b:c")
+      expect(yield* first.get("b:c")).toBeUndefined()
+      return undefined
+    }),
+  )
+
  it.effect("registers location tools through the plugin context", () =>
    Effect.gen(function* () {
      const plugins = yield* Plugin.Service
@@ -11,6 +11,7 @@ import { FileSystem } from "@opencode-ai/core/filesystem"
 import { FSUtil } from "@opencode-ai/util/fs-util"
 import { Form } from "@opencode-ai/core/form"
 import { Integration } from "@opencode-ai/core/integration"
+import { KV } from "@opencode-ai/core/kv"
 import { Location } from "@opencode-ai/core/location"
 import { MCP } from "@opencode-ai/core/mcp/index"
 import { Npm } from "@opencode-ai/util/npm"
@@ -52,6 +53,7 @@ export const PluginTestLayer = LayerNode.compile(
    Catalog.node,
    Command.node,
    Integration.node,
+    KV.node,
    MCP.node,
    PluginRuntime.node,
    PluginHooks.node,
@@ -94,6 +94,12 @@ export function host(overrides: Overrides = {}): Plugin.Context {
      transform: () => Effect.die("unused skill.transform"),
      reload: () => Effect.die("unused skill.reload"),
    },
+    storage: overrides.storage ?? {
+      get: () => Effect.die("unused storage.get"),
+      set: () => Effect.die("unused storage.set"),
+      remove: () => Effect.die("unused storage.remove"),
+      scan: () => Effect.die("unused storage.scan"),
+    },
    shell: overrides.shell ?? {
      hook: () => Effect.die("unused shell.hook"),
    },
@@ -27,6 +27,32 @@ import { host as testHost } from "./host"
 const it = testEffect(PluginTestLayer)

 describe("fromPromise", () => {
+  it.effect("adapts plugin storage methods", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const adapted = PluginPromise.fromPromise(
+        define({
+          id: "promise-storage",
+          setup: async (ctx) => {
+            expect(await ctx.storage.get("missing")).toBeUndefined()
+            await ctx.storage.set("items/b", { order: 2 })
+            await ctx.storage.set("items/a", { order: 1 })
+            expect(await ctx.storage.get("items/a")).toEqual({ order: 1 })
+            expect(await ctx.storage.scan({ prefix: "items/", limit: 1 })).toEqual({
+              entries: [{ key: "items/a", value: { order: 1 } }],
+              next: "items/a",
+            })
+            await ctx.storage.remove("items/a")
+            await ctx.storage.remove("items/a")
+            expect(await ctx.storage.get("items/a")).toBeUndefined()
+          },
+        }),
+      )
+
+      yield* plugins.activate([{ ...adapted, version: "1" }])
+    }),
+  )
+
  it.effect("adapts session creation through the protocol schema", () =>
    Effect.gen(function* () {
      let seen: unknown
@@ -1,4 +1,5 @@
 export * as Plugin from "./plugin.js"
+export type { StorageEntry, StorageScanOptions, StorageScanResult } from "../storage.js"

 export { Agent } from "@opencode-ai/schema/agent"
 export { Command } from "@opencode-ai/schema/command"
@@ -13,6 +13,7 @@ import type { ReferenceDomain } from "./reference.js"
 import type { SessionDomain } from "./session.js"
 import type { ShellDomain } from "./shell.js"
 import type { SkillDomain } from "./skill.js"
+import type { StorageDomain } from "./storage.js"
 import type { ToolDomain } from "./tool.js"
 import type { WebSearchDomain } from "./websearch.js"

@@ -31,6 +32,7 @@ export interface Context {
  readonly session: SessionDomain
  readonly shell: ShellDomain
  readonly skill: SkillDomain
+  readonly storage: StorageDomain
  readonly tool: ToolDomain
  readonly websearch: WebSearchDomain
 }
@@ -0,0 +1,9 @@
+import type { Effect, Schema } from "effect"
+import type { StorageScanOptions, StorageScanResult } from "../storage.js"
+
+export interface StorageDomain {
+  readonly get: (key: string) => Effect.Effect<Schema.Json | undefined>
+  readonly set: (key: string, value: Schema.Json) => Effect.Effect<void>
+  readonly remove: (key: string) => Effect.Effect<void>
+  readonly scan: (options: StorageScanOptions) => Effect.Effect<StorageScanResult>
+}
@@ -261,6 +261,12 @@ export function fromPromise(plugin: Plugin) {
            transform: transform(host.skill),
            reload: () => run(host.skill.reload()),
          },
+          storage: {
+            get: (key) => run(host.storage.get(key)),
+            set: (key, value) => run(host.storage.set(key, value)),
+            remove: (key) => run(host.storage.remove(key)),
+            scan: (options) => run(host.storage.scan(options)),
+          },
          tool: {
            transform: (callback) =>
              register(
@@ -1,4 +1,5 @@
 export type { PluginOptions } from "../options.js"
+export type { StorageEntry, StorageScanOptions, StorageScanResult } from "../storage.js"
 export * as Plugin from "./plugin.js"

 export { Agent } from "@opencode-ai/schema/agent"
@@ -12,6 +12,7 @@ import type { ReferenceDomain } from "./reference.js"
 import type { SessionDomain } from "./session.js"
 import type { ShellDomain } from "./shell.js"
 import type { SkillDomain } from "./skill.js"
+import type { StorageDomain } from "./storage.js"
 import type { ToolDomain } from "./tool.js"
 import type { WebSearchDomain } from "./websearch.js"

@@ -30,6 +31,7 @@ export interface Context {
  readonly session: SessionDomain
  readonly shell: ShellDomain
  readonly skill: SkillDomain
+  readonly storage: StorageDomain
  readonly tool: ToolDomain
  readonly websearch: WebSearchDomain
 }
@@ -0,0 +1,9 @@
+import type { Schema } from "effect"
+import type { StorageScanOptions, StorageScanResult } from "../storage.js"
+
+export interface StorageDomain {
+  readonly get: (key: string) => Promise<Schema.Json | undefined>
+  readonly set: (key: string, value: Schema.Json) => Promise<void>
+  readonly remove: (key: string) => Promise<void>
+  readonly scan: (options: StorageScanOptions) => Promise<StorageScanResult>
+}
@@ -0,0 +1,17 @@
+import type { Schema } from "effect"
+
+export interface StorageEntry {
+  readonly key: string
+  readonly value: Schema.Json
+}
+
+export interface StorageScanOptions {
+  readonly prefix: string
+  readonly after?: string
+  readonly limit?: number
+}
+
+export interface StorageScanResult {
+  readonly entries: readonly StorageEntry[]
+  readonly next?: string
+}