diff --git a/apps/server/package.json b/apps/server/package.json index 0dacbbe90720c..a06e2346d534c 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -77,6 +77,7 @@ "@effect/vitest": "catalog:", "@opencode-ai/sdk": "^1.3.15", "@t3tools/contracts": "workspace:*", + "@t3tools/plugin-sdk": "workspace:*", "@t3tools/provider-testing": "workspace:*", "@t3tools/secret-provider-core": "workspace:*", "@t3tools/secret-provider-definitions": "workspace:*", diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 8044740a7e2ee..d43b5a5c2f734 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -68,6 +68,16 @@ describe("RPC authorization scopes", () => { } }); + it("keeps enabling, reloading and configuring plugins, which runs their code, behind settings write", () => { + for (const method of [ + WS_METHODS.pluginsSetEnabled, + WS_METHODS.pluginsReload, + WS_METHODS.pluginsSettingsUpdate, + ]) { + expect(requiredScopeForRpcMethod(method)).toBe(AuthSettingsWriteScope); + } + }); + it("rejects unknown RPC method names", () => { for (const method of ["server.notRegistered", "toString", "constructor"]) { expect(() => requiredScopeForRpcMethod(method)).toThrow( diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 6049121b41a5d..0fb91a610c9e1 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -205,6 +205,13 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.deviceDetail]: AuthOrchestrationReadScope, [WS_METHODS.deviceAction]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeDeviceState]: AuthOrchestrationReadScope, + // `plugins.invoke` also needs the operate scope for methods a plugin registered as writes; + // PluginHost checks that against the plugin's own declaration. Channels are reads. + [WS_METHODS.pluginsList]: AuthOrchestrationReadScope, + [WS_METHODS.pluginsSubscribeStates]: AuthOrchestrationReadScope, + [WS_METHODS.pluginsSettingsGet]: AuthOrchestrationReadScope, + [WS_METHODS.pluginsInvoke]: AuthOrchestrationReadScope, + [WS_METHODS.pluginsSubscribe]: AuthOrchestrationReadScope, [WS_METHODS.subscribeServerConfig]: AuthOrchestrationReadScope, [WS_METHODS.subscribeServerLifecycle]: AuthOrchestrationReadScope, [WS_METHODS.subscribeAuthAccess]: AuthAccessReadScope, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 72d7f0c2de7ef..b2b2d6956398d 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -243,6 +243,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, threadForceStop: true, environmentThemes: true, + plugins: true, usageLimitSources: true, providerWorkspaceSnapshotsOnDemand: true, usagePriceOverrides: true, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 1e3388e4136ad..76ebfeee70cab 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -87,6 +87,14 @@ const RPC_AGGREGATES = { [WS_METHODS.serverRefreshUsageRates]: "server", [WS_METHODS.serverSignalProcess]: "server", [WS_METHODS.serverReportClientActivity]: "server", + [WS_METHODS.pluginsList]: "plugins", + [WS_METHODS.pluginsSubscribeStates]: "plugins", + [WS_METHODS.pluginsSetEnabled]: "plugins", + [WS_METHODS.pluginsReload]: "plugins", + [WS_METHODS.pluginsSettingsGet]: "plugins", + [WS_METHODS.pluginsSettingsUpdate]: "plugins", + [WS_METHODS.pluginsInvoke]: "plugins", + [WS_METHODS.pluginsSubscribe]: "plugins", [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks", [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks", [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks", diff --git a/apps/server/src/plugins/PluginApi.ts b/apps/server/src/plugins/PluginApi.ts new file mode 100644 index 0000000000000..5457845244c68 --- /dev/null +++ b/apps/server/src/plugins/PluginApi.ts @@ -0,0 +1,149 @@ +import { PluginRpcName, type PluginId } from "@t3tools/contracts"; +import type { PluginApi, PluginRpcAccess, PluginRpcHandler } from "@t3tools/plugin-sdk/server"; +import * as Cause from "effect/Cause"; +import type * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import type * as PluginSettingsStore from "./PluginSettingsStore.ts"; +import type * as PluginStorage from "./PluginStorage.ts"; + +const isJson = Schema.is(Schema.Json); +const isPluginRpcName = Schema.is(PluginRpcName); +const isPluginRpcAccess = Schema.is(Schema.Literals(["read", "write"])); + +export interface PluginPublication { + readonly channel: string; + readonly value: Schema.Json; +} + +/** Everything one activation registered. The host routes calls through it and tears it down. */ +export interface PluginActivation { + readonly methods: Map< + string, + { readonly handler: PluginRpcHandler; readonly access: PluginRpcAccess } + >; + readonly cleanups: Array<() => void | Promise>; + /** Aborted on deactivate; handler signals and settings iterators follow it. */ + readonly controller: AbortController; + readonly publications: PubSub.PubSub; +} + +export const makePluginActivation = ( + publications: PubSub.PubSub, +): PluginActivation => ({ + methods: new Map(), + cleanups: [], + controller: new AbortController(), + publications, +}); + +const assertActive = (activation: PluginActivation) => { + if (activation.controller.signal.aborted) { + throw new Error("This plugin activation has ended."); + } +}; + +/** + * Builds the object a plugin's `activate` receives. Effect stays on this side: every call runs + * with the host's services and settles as a plain Promise that rejects with an `Error`. + */ +export const makePluginApi = (input: { + readonly id: PluginId; + readonly rootDir: string; + readonly settingsSchema: Schema.JsonObject | undefined; + readonly activation: PluginActivation; + readonly context: Context.Context; + readonly settings: PluginSettingsStore.PluginSettingsStore["Service"]; + readonly storage: PluginStorage.PluginStorage["Service"]; +}): PluginApi => { + const { id, activation, context } = input; + const runExit = Effect.runPromiseExitWith(context); + const call = (effect: Effect.Effect): Promise => + runExit(effect).then((exit) => { + if (Exit.isSuccess(exit)) return exit.value; + const error = Cause.squash(exit.cause); + throw new Error(error instanceof Error ? error.message : "Plugin API call failed."); + }); + const log = + (level: "Debug" | "Info" | "Warn" | "Error") => + (message: string, fields?: Schema.JsonObject) => { + Effect.runForkWith(context)( + Effect.logWithLevel(level)(message, fields ?? {}).pipe( + Effect.annotateLogs({ pluginId: id }), + ), + ); + }; + + return { + version: 1, + id, + rootDir: input.rootDir, + log: { debug: log("Debug"), info: log("Info"), warn: log("Warn"), error: log("Error") }, + settings: { + get: () => call(input.settings.get(id, input.settingsSchema)), + changes: () => + Stream.toAsyncIterableWith( + input.settings.changes(id).pipe( + Stream.interruptWhen( + Effect.callback((resume) => { + const onAbort = () => resume(Effect.void); + if (activation.controller.signal.aborted) return onAbort(); + activation.controller.signal.addEventListener("abort", onAbort, { once: true }); + return Effect.sync(() => + activation.controller.signal.removeEventListener("abort", onAbort), + ); + }), + ), + ), + context, + ), + }, + storage: { + get: (key) => call(input.storage.get(id, key)).then(Option.getOrUndefined), + set: (key, value) => { + if (!isJson(value)) return Promise.reject(new TypeError("Storage values must be JSON.")); + return call(input.storage.set(id, key, value)); + }, + delete: (key) => call(input.storage.remove(id, key)), + list: (prefix) => call(input.storage.list(id, prefix)), + }, + rpc: { + register: (method, handler, options) => { + assertActive(activation); + if (!isPluginRpcName(method)) + throw new TypeError(`"${method}" is not a valid method name.`); + // Plugins are untyped JavaScript, and a misspelled access level must not open a write. + if (!isPluginRpcAccess(options.access)) { + throw new TypeError(`Method "${method}" needs access "read" or "write".`); + } + if (activation.methods.has(method)) { + throw new Error(`Method "${method}" is already registered.`); + } + activation.methods.set(method, { handler, access: options.access }); + }, + publish: (channel, value) => { + assertActive(activation); + if (!isPluginRpcName(channel)) { + throw new TypeError(`"${channel}" is not a valid channel name.`); + } + if (!isJson(value)) throw new TypeError("Published values must be JSON."); + PubSub.publishUnsafe(activation.publications, { channel, value }); + }, + }, + // Something created after deactivation has nothing left to wait for, so it cleans up now. + onDeactivate: (cleanup) => { + if (activation.controller.signal.aborted) { + void Promise.resolve() + .then(cleanup) + .catch(() => undefined); + return; + } + activation.cleanups.push(cleanup); + }, + }; +}; diff --git a/apps/server/src/plugins/PluginHost.test.ts b/apps/server/src/plugins/PluginHost.test.ts new file mode 100644 index 0000000000000..dc9ff9336536c --- /dev/null +++ b/apps/server/src/plugins/PluginHost.test.ts @@ -0,0 +1,379 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, + PluginId, + type PluginState, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Stream from "effect/Stream"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as PluginHost from "./PluginHost.ts"; +import * as PluginSettingsStore from "./PluginSettingsStore.ts"; +import * as PluginStorage from "./PluginStorage.ts"; + +const fixturesDir = new URL("../../../../packages/plugin-sdk/fixtures/", import.meta.url).pathname; +const hello = PluginId.make("hello-plugin"); +const broken = PluginId.make("broken-plugin"); +const operate = [AuthOrchestrationReadScope, AuthOrchestrationOperateScope] as const; + +const layerConfig = () => + Layer.fresh(ServerConfig.layerTest(process.cwd(), { prefix: "t3code-plugin-host-test-" })); + +const layerHost = () => + Layer.fresh( + PluginHost.layer.pipe( + Layer.provideMerge(Layer.mergeAll(PluginSettingsStore.layer, PluginStorage.layer)), + Layer.provide(ServerSecretStore.layer), + ), + ); + +const writeFile = Effect.fn(function* (filePath: string, contents: string) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* fs.makeDirectory(path.dirname(filePath), { recursive: true }); + yield* fs.writeFileString(filePath, contents); +}); + +const writePluginsConfig = Effect.fn(function* (config: { + readonly enabled?: Record; + readonly devPaths?: ReadonlyArray; +}) { + const { stateDir } = yield* ServerConfig.ServerConfig; + yield* writeFile(`${stateDir}/plugins.json`, JSON.stringify(config)); +}); + +/** A plugin written for one test, under `/plugins/`. */ +const writeInstalledPlugin = Effect.fn(function* (id: string, serverSource: string) { + const { stateDir } = yield* ServerConfig.ServerConfig; + const rootDir = `${stateDir}/plugins/${id}`; + yield* writeFile( + `${rootDir}/t3-plugin.json`, + JSON.stringify({ + id, + name: id, + version: "0.1.0", + engines: { t3: ">=0.1.0" }, + server: "./server.cjs", + }), + ); + yield* writeFile(`${rootDir}/server.cjs`, serverSource); + return rootDir; +}); + +const stateOf = (states: ReadonlyArray, id: string) => + states.find((state) => state.id === id); + +/** Resolves with the first list where every listed plugin has its expected status. */ +const waitForStatuses = ( + host: PluginHost.PluginHost["Service"], + expected: Readonly>, +) => + host.streamStates.pipe( + Stream.filter((states) => + Object.entries(expected).every(([id, status]) => stateOf(states, id)?.status === status), + ), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + +it.layer(NodeServices.layer)("PluginHost", (it) => { + it.effect("activates an enabled plugin and serves its methods, settings and storage", () => + Effect.gen(function* () { + yield* writePluginsConfig({ + enabled: { [hello]: true }, + devPaths: [`${fixturesDir}hello-plugin`], + }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const states = yield* waitForStatuses(host, { [hello]: "active" }); + assert.include(stateOf(states, hello), { source: "dev", hasServer: true, enabled: true }); + + const ping = (input: string) => + host.invoke({ pluginId: hello, method: "ping", input, scopes: operate }); + assert.deepStrictEqual(yield* ping("a"), { pong: "a", greeting: "hello" }); + const changed = yield* host + .subscribe({ pluginId: hello, channel: "settings" }) + .pipe(Stream.runHead, Effect.forkChild({ startImmediately: true })); + yield* host.updateSettings(hello, { greeting: "hey" }); + assert.deepStrictEqual(yield* Fiber.join(changed), Option.some({ greeting: "hey" })); + assert.deepStrictEqual(yield* ping("b"), { pong: "b", greeting: "hey" }); + + const counted = yield* host + .subscribe({ pluginId: hello, channel: "counted" }) + .pipe(Stream.runHead, Effect.forkChild({ startImmediately: true })); + assert.strictEqual( + yield* host.invoke({ + pluginId: hello, + method: "count", + input: undefined, + scopes: operate, + }), + 1, + ); + assert.deepStrictEqual(yield* Fiber.join(counted), Option.some(1)); + + const denied = yield* host + .invoke({ + pluginId: hello, + method: "count", + input: undefined, + scopes: [AuthOrchestrationReadScope], + }) + .pipe(Effect.flip); + assert.strictEqual(denied._tag, "PluginAccessDeniedError"); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("marks a plugin that throws in activate as failed and keeps the others running", () => + Effect.gen(function* () { + yield* writePluginsConfig({ + enabled: { [hello]: true, [broken]: true }, + devPaths: [`${fixturesDir}broken-plugin`, `${fixturesDir}hello-plugin`], + }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const states = yield* waitForStatuses(host, { [broken]: "failed", [hello]: "active" }); + assert.deepStrictEqual(stateOf(states, broken)?.failure, { + stage: "activate", + message: "broken-plugin refuses to start", + }); + const missing = yield* host + .invoke({ pluginId: broken, method: "ping", input: undefined, scopes: operate }) + .pipe(Effect.flip); + assert.strictEqual(missing._tag, "PluginNotActiveError"); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("runs cleanup on disable and stops serving the plugin", () => + Effect.gen(function* () { + yield* writePluginsConfig({ + enabled: { [hello]: true }, + devPaths: [`${fixturesDir}hello-plugin`], + }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const storage = yield* PluginStorage.PluginStorage; + yield* waitForStatuses(host, { [hello]: "active" }); + + const disabled = yield* host.setEnabled(hello, false); + assert.include(disabled, { status: "disabled", enabled: false }); + assert.deepStrictEqual(yield* storage.get(hello, "deactivated"), Option.some(true)); + const error = yield* host + .invoke({ pluginId: hello, method: "ping", input: undefined, scopes: operate }) + .pipe(Effect.flip); + assert.strictEqual(error._tag, "PluginNotActiveError"); + + const { stateDir } = yield* ServerConfig.ServerConfig; + const saved = yield* FileSystem.FileSystem.use((fs) => + fs.readFileString(`${stateDir}/plugins.json`), + ); + assert.deepStrictEqual(JSON.parse(saved).enabled, { [hello]: false }); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("reload picks up a changed manifest and changed code", () => + Effect.gen(function* () { + const rootDir = yield* writeInstalledPlugin( + "reloading", + `module.exports = { activate(api) { api.rpc.register("which", () => "v1", { access: "read" }); } };`, + ); + yield* writePluginsConfig({ enabled: { reloading: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const id = PluginId.make("reloading"); + yield* waitForStatuses(host, { reloading: "active" }); + const which = host.invoke({ + pluginId: id, + method: "which", + input: undefined, + scopes: operate, + }); + assert.strictEqual(yield* which, "v1"); + + yield* writeFile( + `${rootDir}/t3-plugin.json`, + JSON.stringify({ + id: "reloading", + name: "Reloaded", + version: "0.2.0", + engines: { t3: ">=0.1.0" }, + server: "./server.cjs", + }), + ); + yield* writeFile( + `${rootDir}/server.cjs`, + `module.exports = { activate(api) { api.rpc.register("which", () => "v2", { access: "read" }); } };`, + ); + const reloaded = yield* host.reload(id); + assert.include(reloaded, { status: "active", version: "0.2.0", name: "Reloaded" }); + assert.strictEqual(yield* which, "v2"); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("fails a plugin that registers a method with an unknown access level", () => + Effect.gen(function* () { + yield* writeInstalledPlugin( + "typo", + `module.exports = { activate(api) { api.rpc.register("go", () => 1, { access: "Write" }); } };`, + ); + yield* writePluginsConfig({ enabled: { typo: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const states = yield* waitForStatuses(host, { typo: "failed" }); + assert.deepStrictEqual(stateOf(states, "typo")?.failure, { + stage: "activate", + message: 'Method "go" needs access "read" or "write".', + }); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("reload stops a plugin whose folder was removed", () => + Effect.gen(function* () { + const stays = yield* writeInstalledPlugin("stays", `module.exports = { activate() {} };`); + const goes = yield* writeInstalledPlugin( + "goes", + `const marker = require("node:path").join(__dirname, "..", "..", "goes-cleaned"); +module.exports = { activate(api) { api.onDeactivate(() => require("node:fs").writeFileSync(marker, "")); } };`, + ); + yield* writePluginsConfig({ enabled: { stays: true, goes: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const fs = yield* FileSystem.FileSystem; + yield* waitForStatuses(host, { stays: "active", goes: "active" }); + yield* fs.remove(goes, { recursive: true }); + + yield* host.reload(PluginId.make("stays")); + + assert.isTrue(yield* fs.exists(`${stays}/../../goes-cleaned`)); + assert.isUndefined(stateOf(yield* host.list, "goes")); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("reload runs new code from a symlinked plugin folder", () => + Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const realDir = `${stateDir}/checkout/linked`; + const plugin = (version: string) => + `module.exports = { activate(api) { api.rpc.register("which", () => "${version}", { access: "read" }); } };`; + yield* writeFile( + `${realDir}/t3-plugin.json`, + JSON.stringify({ + id: "linked", + name: "Linked", + version: "0.1.0", + engines: { t3: ">=0.1.0" }, + server: "./server.cjs", + }), + ); + yield* writeFile(`${realDir}/server.cjs`, plugin("v1")); + yield* fs.makeDirectory(`${stateDir}/plugins`, { recursive: true }); + yield* fs.symlink(realDir, `${stateDir}/plugins/linked`); + yield* writePluginsConfig({ enabled: { linked: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const id = PluginId.make("linked"); + const which = host.invoke({ + pluginId: id, + method: "which", + input: undefined, + scopes: operate, + }); + yield* waitForStatuses(host, { linked: "active" }); + assert.strictEqual(yield* which, "v1"); + + yield* writeFile(`${realDir}/server.cjs`, plugin("v2")); + yield* host.reload(id); + assert.strictEqual(yield* which, "v2"); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("finishes disabling a plugin when the caller gives up during its cleanup", () => + Effect.gen(function* () { + const started = Promise.withResolvers(); + const gate = Promise.withResolvers(); + Reflect.set(globalThis, "__t3PluginHostCleanupStarted", started.resolve); + Reflect.set(globalThis, "__t3PluginHostCleanupGate", gate.promise); + yield* writeInstalledPlugin( + "stubborn", + `module.exports = { activate(api) { + api.rpc.register("ping", () => "pong", { access: "read" }); + api.onDeactivate(() => { globalThis.__t3PluginHostCleanupStarted(); return globalThis.__t3PluginHostCleanupGate; }); +} };`, + ); + yield* writePluginsConfig({ enabled: { stubborn: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const id = PluginId.make("stubborn"); + yield* waitForStatuses(host, { stubborn: "active" }); + + const disabling = yield* host + .setEnabled(id, false) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => started.promise); + yield* Fiber.interrupt(disabling); + gate.resolve(); + + yield* waitForStatuses(host, { stubborn: "disabled" }); + const error = yield* host + .invoke({ pluginId: id, method: "ping", input: undefined, scopes: operate }) + .pipe(Effect.flip); + assert.strictEqual(error._tag, "PluginNotActiveError"); + }).pipe(Effect.provide(layerHost())); + Reflect.deleteProperty(globalThis, "__t3PluginHostCleanupStarted"); + Reflect.deleteProperty(globalThis, "__t3PluginHostCleanupGate"); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("does not load a disabled plugin at startup", () => + Effect.gen(function* () { + const rootDir = yield* writeInstalledPlugin( + "dormant", + `require("node:fs").writeFileSync(__dirname + "/loaded", ""); +module.exports = { activate() {} };`, + ); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + const states = yield* waitForStatuses(host, { dormant: "disabled" }); + assert.include(stateOf(states, "dormant"), { enabled: false, hasServer: true }); + const fs = yield* FileSystem.FileSystem; + assert.isFalse(yield* fs.exists(`${rootDir}/loaded`)); + }).pipe(Effect.provide(layerHost())); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("finishes starting before a slow plugin finishes activating", () => + Effect.gen(function* () { + const gate = Promise.withResolvers(); + Reflect.set(globalThis, "__t3PluginHostTestGate", gate.promise); + yield* writeInstalledPlugin( + "slow", + `module.exports = { activate() { return globalThis.__t3PluginHostTestGate; } };`, + ); + yield* writePluginsConfig({ enabled: { slow: true } }); + yield* Effect.gen(function* () { + const host = yield* PluginHost.PluginHost; + yield* waitForStatuses(host, { slow: "loading" }); + gate.resolve(); + yield* waitForStatuses(host, { slow: "active" }); + }).pipe(Effect.provide(layerHost())); + Reflect.deleteProperty(globalThis, "__t3PluginHostTestGate"); + }).pipe(Effect.provide(layerConfig())), + ); +}); diff --git a/apps/server/src/plugins/PluginHost.ts b/apps/server/src/plugins/PluginHost.ts new file mode 100644 index 0000000000000..aece9a0e2809b --- /dev/null +++ b/apps/server/src/plugins/PluginHost.ts @@ -0,0 +1,690 @@ +// @effect-diagnostics nodeBuiltinImport:off - plugins load through Node's CommonJS loader, which +// the FileSystem service does not wrap. +import * as NodeModule from "node:module"; + +import { + AuthEnvironmentScope, + AuthOrchestrationOperateScope, + type PluginFailure, + PluginId, + type PluginManifest, + type PluginState, + type PluginStatus, +} from "@t3tools/contracts"; +import type { PluginDefinition } from "@t3tools/plugin-sdk/server"; +import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; +import { fromJsonStringPretty } from "@t3tools/shared/schemaJson"; +import { satisfiesSemverRange } from "@t3tools/shared/semver"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FiberMap from "effect/FiberMap"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; + +import packageJson from "../../package.json" with { type: "json" }; +import * as ServerConfig from "../config.ts"; +import { forkParked } from "../serverActivation.ts"; +import { + makePluginActivation, + makePluginApi, + type PluginActivation, + type PluginPublication, +} from "./PluginApi.ts"; +import { discoverPlugins, type DiscoveredPlugin } from "./pluginDiscovery.ts"; +import * as PluginSettingsStore from "./PluginSettingsStore.ts"; +import * as PluginStorage from "./PluginStorage.ts"; + +const MAX_FAILURE_MESSAGE_CHARS = 500; +// Plugin code that never settles must not keep a plugin loading, or Disable and Reload waiting, +// forever. The cleanup deadline covers all of a plugin's cleanups together. +const ACTIVATE_DEADLINE = Duration.seconds(30); +const CLEANUP_DEADLINE = Duration.seconds(5); +const PUBLICATION_BUFFER = 64; + +export class PluginNotFoundError extends Schema.TaggedError()( + "PluginNotFoundError", + { pluginId: PluginId }, +) { + override get message(): string { + return `Plugin ${this.pluginId} is not installed.`; + } +} + +export class PluginNotActiveError extends Schema.TaggedError()( + "PluginNotActiveError", + { pluginId: PluginId, status: Schema.String }, +) { + override get message(): string { + return `Plugin ${this.pluginId} is ${this.status}.`; + } +} + +export class PluginMethodNotFoundError extends Schema.TaggedError()( + "PluginMethodNotFoundError", + { pluginId: PluginId, method: Schema.String }, +) { + override get message(): string { + return `Plugin ${this.pluginId} has no method ${this.method}.`; + } +} + +export class PluginAccessDeniedError extends Schema.TaggedError()( + "PluginAccessDeniedError", + { pluginId: PluginId, method: Schema.String, requiredScope: AuthEnvironmentScope }, +) { + override get message(): string { + return `Calling ${this.method} on plugin ${this.pluginId} needs the ${this.requiredScope} scope.`; + } +} +export const isPluginAccessDeniedError = Schema.is(PluginAccessDeniedError); + +// `reason` is the plugin's own error text, bounded. Plugins are trusted code, and their UI needs +// to show why its own call failed. +export class PluginHandlerError extends Schema.TaggedError()( + "PluginHandlerError", + { pluginId: PluginId, method: Schema.String, reason: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Plugin ${this.pluginId} failed in ${this.method}: ${this.reason}`; + } +} + +/** Plugin code threw or rejected while loading, activating, or cleaning up. */ +class PluginCodeError extends Schema.TaggedError()("PluginCodeError", { + cause: Schema.Defect(), +}) { + override get message(): string { + return errorText(this.cause); + } +} + +export class PluginConfigError extends Schema.TaggedError()( + "PluginConfigError", + { path: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not update ${this.path}. If it was edited by hand, fix or delete it.`; + } +} + +export class PluginHost extends Context.Service< + PluginHost, + { + readonly list: Effect.Effect>; + /** The current list, then the whole list after every change. */ + readonly streamStates: Stream.Stream>; + /** Persists the flag, then waits for activation or deactivation to finish. */ + readonly setEnabled: ( + pluginId: PluginId, + enabled: boolean, + ) => Effect.Effect; + /** Rescans the plugin folders and restarts this plugin with fresh code. */ + readonly reload: ( + pluginId: PluginId, + ) => Effect.Effect; + readonly getSettings: ( + pluginId: PluginId, + ) => Effect.Effect< + Schema.JsonObject, + PluginNotFoundError | PluginNotActiveError | PluginSettingsError + >; + readonly updateSettings: ( + pluginId: PluginId, + patch: Schema.JsonObject, + ) => Effect.Effect< + Schema.JsonObject, + PluginNotFoundError | PluginNotActiveError | PluginSettingsError + >; + /** `scopes` are the caller's; methods the plugin registered as writes need the operate scope. */ + readonly invoke: (input: { + readonly pluginId: PluginId; + readonly method: string; + readonly input: Schema.Json | undefined; + readonly scopes: ReadonlyArray; + }) => Effect.Effect< + Schema.Json, + | PluginNotFoundError + | PluginNotActiveError + | PluginMethodNotFoundError + | PluginAccessDeniedError + | PluginHandlerError + >; + /** Values the plugin publishes on `channel` until it deactivates. */ + readonly subscribe: (input: { + readonly pluginId: PluginId; + readonly channel: string; + }) => Stream.Stream; + } +>()("t3/plugins/PluginHost") {} + +type PluginSettingsError = + | PluginSettingsStore.PluginSettingsPersistenceError + | PluginSettingsStore.PluginSettingsSchemaError + | PluginSettingsStore.PluginSettingsValidationError; + +// Unknown keys survive a rewrite, so a newer build's additions outlive a toggle by an older one. +const PluginsConfigFile = Schema.StructWithRest( + Schema.Struct({ + enabled: Schema.optionalKey(Schema.Record(Schema.String, Schema.Boolean)), + devPaths: Schema.optionalKey(Schema.Array(Schema.String)), + }), + [Schema.Record(Schema.String, Schema.Json)], +); +type PluginsConfigFile = typeof PluginsConfigFile.Type; +const decodeConfig = Schema.decodeUnknownEffect(Schema.fromJsonString(PluginsConfigFile)); +const encodeConfig = Schema.encodeEffect(fromJsonStringPretty(PluginsConfigFile)); +const isJson = Schema.is(Schema.Json); + +interface PluginRecord { + readonly plugin: DiscoveredPlugin; + readonly status: PluginStatus; + readonly failure?: PluginFailure; + readonly activation?: PluginActivation; +} + +const errorText = (cause: unknown) => { + let text = "Unknown error"; + try { + text = + cause instanceof Error && typeof cause.message === "string" ? cause.message : String(cause); + } catch { + // Plugin code can throw values whose conversion to a string throws again. + } + return text.slice(0, MAX_FAILURE_MESSAGE_CHARS); +}; + +const manifestOf = (plugin: DiscoveredPlugin): PluginManifest | undefined => + "manifest" in plugin ? plugin.manifest : undefined; + +const toState = (record: PluginRecord, enabled: boolean): PluginState => { + const manifest = manifestOf(record.plugin); + return { + id: record.plugin.id, + name: manifest?.name ?? record.plugin.id, + ...(manifest === undefined + ? {} + : { + version: manifest.version, + ...(manifest.description === undefined ? {} : { description: manifest.description }), + ...(manifest.settingsSchema === undefined + ? {} + : { settingsSchema: manifest.settingsSchema }), + ...(manifest.contributes === undefined ? {} : { contributes: manifest.contributes }), + }), + permissions: manifest?.permissions ?? [], + source: record.plugin.source, + enabled, + status: record.status, + ...(record.failure === undefined ? {} : { failure: record.failure }), + hasServer: manifest?.server !== undefined, + hasWeb: manifest?.web !== undefined, + }; +}; + +// One loader for the packaged single executable and plain Node: a require created from an +// absolute path loads files from disk in both, and plugin code resolves its own imports. +const requireFromDisk = NodeModule.createRequire(process.execPath); + +const isPluginDefinition = (value: unknown): value is PluginDefinition => + typeof value === "object" && + value !== null && + "activate" in value && + typeof value.activate === "function"; + +const make = Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const settingsStore = yield* PluginSettingsStore.PluginSettingsStore; + const storage = yield* PluginStorage.PluginStorage; + const context = yield* Effect.context(); + + const pluginsDir = path.join(config.stateDir, "plugins"); + const configPath = path.join(config.stateDir, "plugins.json"); + + const records = yield* Ref.make>([]); + const pluginsConfig = yield* Ref.make({}); + const changes = yield* PubSub.sliding>(1); + const lifecycle = yield* Semaphore.make(1); + const publishLock = yield* Semaphore.make(1); + const activations = yield* FiberMap.make(); + const hostScope = yield* Effect.scope; + + const currentStates = Effect.gen(function* () { + const enabled = (yield* Ref.get(pluginsConfig)).enabled ?? {}; + return (yield* Ref.get(records)).map((record) => + toState(record, enabled[record.plugin.id] === true), + ); + }); + // Reading and publishing under one lock, so a slower fiber cannot publish an older list last. + const publishStates = publishLock.withPermits(1)( + currentStates.pipe( + Effect.flatMap((states) => PubSub.publish(changes, states)), + Effect.asVoid, + ), + ); + const findRecord = (pluginId: string) => + Ref.get(records).pipe(Effect.map((all) => all.find((record) => record.plugin.id === pluginId))); + const stateOf = (pluginId: PluginId) => + currentStates.pipe( + Effect.flatMap((states) => { + const state = states.find((candidate) => candidate.id === pluginId); + return state === undefined + ? Effect.fail(new PluginNotFoundError({ pluginId })) + : Effect.succeed(state); + }), + ); + // Records are replaced by id; a shadowed duplicate keeps its own failure entry. + const setRecord = (pluginId: PluginId, update: Omit) => + Ref.update(records, (all) => { + const index = all.findIndex((record) => record.plugin.id === pluginId); + if (index === -1) return all; + const next = [...all]; + next[index] = { plugin: all[index]!.plugin, ...update }; + return next; + }).pipe(Effect.andThen(publishStates)); + + const readConfig = Effect.gen(function* () { + const exists = yield* fs.exists(configPath); + if (!exists) { + const emptyConfig: PluginsConfigFile = {}; + return emptyConfig; + } + const raw = yield* fs.readFileString(configPath); + return yield* decodeConfig(raw); + }).pipe(Effect.mapError((cause) => new PluginConfigError({ path: configPath, cause }))); + + const writeEnabled = Effect.fnUntraced(function* (pluginId: PluginId, enabled: boolean) { + const current = yield* readConfig; + const next: PluginsConfigFile = { + ...current, + enabled: { ...current.enabled, [pluginId]: enabled }, + }; + const contents = yield* encodeConfig(next).pipe( + Effect.mapError((cause) => new PluginConfigError({ path: configPath, cause })), + ); + yield* writeFileStringAtomically({ filePath: configPath, contents: `${contents}\n` }).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + Effect.mapError((cause) => new PluginConfigError({ path: configPath, cause })), + ); + yield* Ref.set(pluginsConfig, next); + }); + + const runCleanup = (pluginId: PluginId, cleanup: () => void | Promise) => + Effect.tryPromise({ + try: () => Promise.resolve().then(cleanup), + catch: (cause) => new PluginCodeError({ cause }), + }).pipe( + Effect.catchTags({ + PluginCodeError: (error) => + Effect.logWarning("plugin cleanup failed", { pluginId, reason: error.message }), + }), + ); + + // Aborts first so handlers and iterators stop, then runs cleanups newest first. Teardown also + // runs as an interrupt finalizer, so the deadline gets an interruptible region to cut. + const teardown = Effect.fnUntraced(function* (pluginId: PluginId, activation: PluginActivation) { + activation.controller.abort(); + const timedOut = yield* Effect.forEach( + activation.cleanups.toReversed(), + (cleanup) => runCleanup(pluginId, cleanup), + { discard: true }, + ).pipe(Effect.interruptible, Effect.timeoutOption(CLEANUP_DEADLINE), Effect.map(Option.isNone)); + if (timedOut) yield* Effect.logWarning("plugin cleanup timed out", { pluginId }); + yield* PubSub.shutdown(activation.publications); + }); + + const loadDefinition = (plugin: DiscoveredPlugin, entry: string) => + Effect.try({ + try: () => { + const entryPath = path.join(plugin.rootDir, entry); + // Drop every cached module under the plugin so reload runs current code. Node still + // holds the old instances until the server restarts. + const prefix = `${plugin.rootDir}${path.sep}`; + for (const cached of Object.keys(requireFromDisk.cache)) { + if (cached.startsWith(prefix)) Reflect.deleteProperty(requireFromDisk.cache, cached); + } + const exported: unknown = requireFromDisk(entryPath); + // Bundlers that compile `export default` to CommonJS put the definition on `default`. + const definition = + typeof exported === "object" && exported !== null && "default" in exported + ? exported.default + : exported; + if (!isPluginDefinition(definition)) { + throw new Error("The server entry does not export a plugin with an activate function."); + } + return definition; + }, + catch: (cause) => new PluginCodeError({ cause }), + }); + + const activatePlugin = Effect.fnUntraced(function* (plugin: DiscoveredPlugin) { + const manifest = manifestOf(plugin); + if (manifest === undefined) return; + if (!satisfiesSemverRange(packageJson.version, manifest.engines.t3)) { + return yield* setRecord(plugin.id, { + status: "failed", + failure: { + stage: "compatibility", + message: `Needs T3 Code ${manifest.engines.t3}; this server is ${packageJson.version}.`, + }, + }); + } + if (manifest.server === undefined) { + return yield* setRecord(plugin.id, { status: "active" }); + } + yield* setRecord(plugin.id, { status: "loading" }); + const definition = yield* Effect.exit(loadDefinition(plugin, manifest.server)); + if (definition._tag === "Failure") { + return yield* setRecord(plugin.id, { + status: "failed", + failure: { stage: "load", message: errorText(Cause.squash(definition.cause)) }, + }); + } + const activation = makePluginActivation( + yield* PubSub.sliding(PUBLICATION_BUFFER), + ); + const api = makePluginApi({ + id: plugin.id, + rootDir: plugin.rootDir, + settingsSchema: manifest.settingsSchema, + activation, + context, + settings: settingsStore, + storage, + }); + const activated = yield* Effect.tryPromise({ + try: () => Promise.resolve().then(() => definition.value.activate(api)), + catch: (cause) => new PluginCodeError({ cause }), + }).pipe( + Effect.timeoutOption(ACTIVATE_DEADLINE), + Effect.onInterrupt(() => teardown(plugin.id, activation)), + Effect.exit, + ); + if (activated._tag === "Failure" || Option.isNone(activated.value)) { + yield* teardown(plugin.id, activation); + return yield* setRecord(plugin.id, { + status: "failed", + failure: { + stage: "activate", + message: + activated._tag === "Failure" + ? errorText(Cause.squash(activated.cause)) + : `activate did not finish within ${Duration.toSeconds(ACTIVATE_DEADLINE)} seconds.`, + }, + }); + } + yield* setRecord(plugin.id, { status: "active", activation }); + yield* Effect.logInfo("plugin activated", { pluginId: plugin.id }); + }); + + /** + * Interrupts a pending activation, then tears down a finished one. The record stops routing + * calls before cleanups run; the caller sets the plugin's next status. + */ + const stopPlugin = Effect.fnUntraced(function* (pluginId: PluginId) { + yield* FiberMap.remove(activations, pluginId); + const record = yield* findRecord(pluginId); + if (record?.activation === undefined) return; + yield* setRecord(pluginId, { status: "disabled" }); + yield* teardown(pluginId, record.activation); + }); + + const startPlugin = (plugin: DiscoveredPlugin) => + FiberMap.run(activations, plugin.id, activatePlugin(plugin)); + + const initialRecord = (plugin: DiscoveredPlugin, enabled: boolean): PluginRecord => + "failure" in plugin + ? { plugin, status: "failed", failure: plugin.failure } + : { plugin, status: enabled ? "loading" : "disabled" }; + + // Startup tolerates an unreadable plugins.json; an explicit reload reports it instead. + const discover = Effect.fnUntraced(function* (lenient: boolean) { + const emptyConfig: PluginsConfigFile = {}; + const current = lenient + ? yield* readConfig.pipe( + Effect.catchTags({ + PluginConfigError: (error) => + Effect.logWarning("ignoring unreadable plugins.json", { path: error.path }).pipe( + Effect.as(emptyConfig), + ), + }), + ) + : yield* readConfig; + yield* Ref.set(pluginsConfig, current); + const plugins = yield* discoverPlugins({ pluginsDir, devPaths: current.devPaths ?? [] }).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); + return { plugins, enabled: current.enabled ?? {} }; + }); + + const startup = lifecycle.withPermits(1)( + Effect.gen(function* () { + const { plugins, enabled } = yield* discover(true); + yield* Ref.set( + records, + plugins.map((plugin) => initialRecord(plugin, enabled[plugin.id] === true)), + ); + yield* publishStates; + // Each plugin activates in its own fiber, so a slow one delays nobody else. + for (const plugin of plugins) { + if (enabled[plugin.id] === true && "manifest" in plugin) yield* startPlugin(plugin); + } + }), + ); + + // Lifecycle changes run in the host's scope, so a caller that disconnects only stops waiting + // and never leaves a plugin half torn down. + const runLifecycle = (body: Effect.Effect) => + lifecycle.withPermits(1)(body).pipe(Effect.forkIn(hostScope), Effect.flatMap(Fiber.join)); + + const awaitActivation = (pluginId: PluginId) => + FiberMap.get(activations, pluginId).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.void, + onSome: (fiber) => Effect.asVoid(Fiber.await(fiber)), + }), + ), + ); + + const setEnabled: PluginHost["Service"]["setEnabled"] = Effect.fn("PluginHost.setEnabled")( + function* (pluginId, enabled) { + yield* runLifecycle( + Effect.gen(function* () { + const record = yield* findRecord(pluginId); + if (record === undefined) return yield* new PluginNotFoundError({ pluginId }); + yield* writeEnabled(pluginId, enabled); + if ("failure" in record.plugin) return yield* publishStates; + // Enabling a plugin that is running or starting keeps it; a failed one is retried. + if (enabled && (record.status === "active" || record.status === "loading")) { + return yield* publishStates; + } + yield* stopPlugin(pluginId); + if (enabled) { + yield* startPlugin(record.plugin); + } else { + yield* setRecord(pluginId, { status: "disabled" }); + } + }), + ); + yield* awaitActivation(pluginId); + return yield* stateOf(pluginId); + }, + ); + + const reload: PluginHost["Service"]["reload"] = Effect.fn("PluginHost.reload")( + function* (pluginId) { + yield* runLifecycle( + Effect.gen(function* () { + const { plugins, enabled } = yield* discover(false); + const fresh = plugins.find((plugin) => plugin.id === pluginId); + yield* stopPlugin(pluginId); + // Plugins whose folders are gone are stopped too, or they would keep running unlisted. + const vanished = (yield* Ref.get(records)).filter( + (record) => !plugins.some((plugin) => plugin.id === record.plugin.id), + ); + for (const record of vanished) yield* stopPlugin(record.plugin.id); + const known = yield* Ref.get(records); + const kept = known.filter( + (record) => + record.plugin.id !== pluginId && plugins.some((p) => p.id === record.plugin.id), + ); + const added = plugins.filter( + (plugin) => + plugin.id === pluginId || !known.some((record) => record.plugin.id === plugin.id), + ); + yield* Ref.set(records, [ + ...kept, + ...added.map((plugin) => initialRecord(plugin, enabled[plugin.id] === true)), + ]); + yield* publishStates; + if (fresh === undefined) return yield* new PluginNotFoundError({ pluginId }); + for (const plugin of added) { + if (enabled[plugin.id] === true && "manifest" in plugin) yield* startPlugin(plugin); + } + }), + ); + yield* awaitActivation(pluginId); + return yield* stateOf(pluginId); + }, + ); + + const manifestFor = Effect.fnUntraced(function* (pluginId: PluginId) { + const record = yield* findRecord(pluginId); + if (record === undefined) return yield* new PluginNotFoundError({ pluginId }); + const manifest = manifestOf(record.plugin); + if (manifest === undefined) { + return yield* new PluginNotActiveError({ pluginId, status: record.status }); + } + return manifest; + }); + + const activeRecord = Effect.fnUntraced(function* (pluginId: PluginId) { + const record = yield* findRecord(pluginId); + if (record === undefined) return yield* new PluginNotFoundError({ pluginId }); + if (record.status !== "active" || record.activation === undefined) { + return yield* new PluginNotActiveError({ pluginId, status: record.status }); + } + return record.activation; + }); + + yield* forkParked(startup); + yield* Effect.addFinalizer(() => + Ref.get(records).pipe( + Effect.flatMap((all) => + Effect.forEach(all, (record) => + record.activation === undefined + ? Effect.void + : teardown(record.plugin.id, record.activation), + ), + ), + ), + ); + + return PluginHost.of({ + list: currentStates, + streamStates: Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + const snapshot = yield* currentStates; + return Stream.concat(Stream.make(snapshot), Stream.fromSubscription(subscription)); + }), + ), + setEnabled, + reload, + getSettings: Effect.fn("PluginHost.getSettings")(function* (pluginId) { + const manifest = yield* manifestFor(pluginId); + return yield* settingsStore.getForClient(pluginId, manifest.settingsSchema); + }), + updateSettings: Effect.fn("PluginHost.updateSettings")(function* (pluginId, patch) { + const manifest = yield* manifestFor(pluginId); + return yield* settingsStore.update(pluginId, manifest.settingsSchema, patch); + }), + invoke: Effect.fn("PluginHost.invoke")(function* (request) { + yield* Effect.annotateCurrentSpan({ + "plugin.id": request.pluginId, + "plugin.method": request.method, + }); + const activation = yield* activeRecord(request.pluginId); + const registered = activation.methods.get(request.method); + if (registered === undefined) { + return yield* new PluginMethodNotFoundError({ + pluginId: request.pluginId, + method: request.method, + }); + } + if ( + registered.access === "write" && + !request.scopes.includes(AuthOrchestrationOperateScope) + ) { + return yield* new PluginAccessDeniedError({ + pluginId: request.pluginId, + method: request.method, + requiredScope: AuthOrchestrationOperateScope, + }); + } + const handlerError = (cause: unknown) => + new PluginHandlerError({ + pluginId: request.pluginId, + method: request.method, + reason: errorText(cause), + cause, + }); + const deactivated = activation.controller.signal; + const output = yield* Effect.raceFirst( + Effect.tryPromise({ + try: (signal) => + Promise.resolve().then(() => + registered.handler(request.input, { + signal: AbortSignal.any([signal, deactivated]), + }), + ), + catch: handlerError, + }), + // A handler that ignores its abort signal must not hold the call open after disable. + Effect.callback((resume) => { + const onAbort = () => + resume( + Effect.fail( + new PluginNotActiveError({ pluginId: request.pluginId, status: "disabled" }), + ), + ); + if (deactivated.aborted) return onAbort(); + deactivated.addEventListener("abort", onAbort, { once: true }); + return Effect.sync(() => deactivated.removeEventListener("abort", onAbort)); + }), + ); + if (output === undefined) return null; + if (!isJson(output)) { + return yield* handlerError(new TypeError("The method returned a value that is not JSON.")); + } + return output; + }), + subscribe: (request) => + Stream.unwrap( + activeRecord(request.pluginId).pipe( + Effect.map((activation) => + Stream.fromPubSub(activation.publications).pipe( + Stream.filter((publication) => publication.channel === request.channel), + Stream.map((publication) => publication.value), + ), + ), + ), + ), + }); +}); + +export const layer = Layer.effect(PluginHost, make); diff --git a/apps/server/src/plugins/PluginSettingsStore.test.ts b/apps/server/src/plugins/PluginSettingsStore.test.ts new file mode 100644 index 0000000000000..f9546057b9436 --- /dev/null +++ b/apps/server/src/plugins/PluginSettingsStore.test.ts @@ -0,0 +1,126 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { PluginId } from "@t3tools/contracts"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import { SECRET_REDACTED } from "../settingsSecrets.ts"; +import * as PluginSettingsStore from "./PluginSettingsStore.ts"; + +const hello = PluginId.make("hello"); + +const settingsSchema: Schema.JsonObject = { + type: "object", + properties: { + count: { type: "integer", minimum: 0 }, + name: { type: "string" }, + token: { type: "string", format: "t3-secret" }, + }, + additionalProperties: false, +}; + +const layerTest = () => + Layer.mergeAll(PluginSettingsStore.layer, ServerSettings.layer).pipe( + Layer.provideMerge(ServerSecretStore.layer), + Layer.provideMerge(Layer.fresh(SqlitePersistence.layerMemory)), + Layer.provideMerge( + Layer.fresh( + ServerConfig.layerTest(process.cwd(), { prefix: "t3code-plugin-settings-test-" }), + ), + ), + ); + +const readSettingsFile = Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + return yield* fileSystem.readFileString( + path.join(stateDir, "plugin-data", hello, "settings.json"), + ); +}); + +it.layer(NodeServices.layer)("PluginSettingsStore", (it) => { + it.effect("rejects values outside the schema without writing them", () => + Effect.gen(function* () { + const store = yield* PluginSettingsStore.PluginSettingsStore; + yield* store.update(hello, settingsSchema, { count: 1 }); + + const error = yield* Effect.flip( + store.update(hello, settingsSchema, { count: -1, extra: true }), + ); + assert.strictEqual(error._tag, "PluginSettingsValidationError"); + assert.include(error.message, 'at ["count"]: Expected a value greater than or equal to 0'); + assert.include(error.message, 'at ["extra"]: Expected no excess property'); + assert.deepStrictEqual(yield* store.get(hello, settingsSchema), { count: 1 }); + }).pipe(Effect.provide(layerTest())), + ); + + it.effect("keeps secrets out of settings.json and redacts them for clients", () => + Effect.gen(function* () { + const store = yield* PluginSettingsStore.PluginSettingsStore; + const { secretsDir } = yield* ServerConfig.ServerConfig; + + assert.deepStrictEqual( + yield* store.update(hello, settingsSchema, { name: "a", token: "s3cret" }), + { name: "a", token: SECRET_REDACTED }, + ); + assert.deepStrictEqual(yield* store.get(hello, settingsSchema), { + name: "a", + token: "s3cret", + }); + assert.deepStrictEqual(JSON.parse(yield* readSettingsFile), { name: "a" }); + + yield* store.update(hello, settingsSchema, { name: "b", token: SECRET_REDACTED }); + assert.deepStrictEqual(yield* store.get(hello, settingsSchema), { + name: "b", + token: "s3cret", + }); + + yield* store.update(hello, settingsSchema, { token: null }); + assert.deepStrictEqual(yield* store.getForClient(hello, settingsSchema), { name: "b" }); + const secretNames = yield* FileSystem.FileSystem.use((fileSystem) => + fileSystem.readDirectory(secretsDir), + ); + assert.isFalse(secretNames.some((name) => name.startsWith(`plugin-${hello}-`))); + }).pipe(Effect.provide(layerTest())), + ); + + it.effect("keeps plugin settings when the server saves its own settings", () => + Effect.gen(function* () { + const store = yield* PluginSettingsStore.PluginSettingsStore; + const serverSettings = yield* ServerSettings.ServerSettingsService; + yield* store.update(hello, settingsSchema, { count: 2, token: "s3cret" }); + + yield* serverSettings.updateSettings({ addProjectBaseDirectory: "~/next" }); + + assert.deepStrictEqual(yield* store.get(hello, settingsSchema), { + count: 2, + token: "s3cret", + }); + }).pipe(Effect.provide(layerTest())), + ); + + it.effect("emits the plugin view after an update of that plugin", () => + Effect.gen(function* () { + const store = yield* PluginSettingsStore.PluginSettingsStore; + const next = yield* store + .changes(hello) + .pipe(Stream.runHead, Effect.forkChild({ startImmediately: true })); + + yield* store.update(PluginId.make("other"), undefined, { count: 9 }); + yield* store.update(hello, settingsSchema, { count: 3, token: "s3cret" }); + + assert.deepStrictEqual(yield* Fiber.join(next), Option.some({ count: 3, token: "s3cret" })); + }).pipe(Effect.provide(layerTest())), + ); +}); diff --git a/apps/server/src/plugins/PluginSettingsStore.ts b/apps/server/src/plugins/PluginSettingsStore.ts new file mode 100644 index 0000000000000..aec67dd0b4003 --- /dev/null +++ b/apps/server/src/plugins/PluginSettingsStore.ts @@ -0,0 +1,337 @@ +import { PluginId } from "@t3tools/contracts"; +import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; +import { fromJsonStringPretty } from "@t3tools/shared/schemaJson"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as JsonSchema from "effect/JsonSchema"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Predicate from "effect/Predicate"; +import * as PubSub from "effect/PubSub"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import * as SchemaIssue from "effect/SchemaIssue"; +import * as SchemaRepresentation from "effect/SchemaRepresentation"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import { SECRET_REDACTED } from "../settingsSecrets.ts"; + +// The Settings page shows the summary, so it stays readable when a patch breaks many fields. +const MAX_VALIDATION_SUMMARY_LENGTH = 1000; + +const persistenceOperations = { + read: "read the settings", + write: "save the settings", + readSecret: "read a secret setting", + writeSecret: "save a secret setting", +} as const; + +export class PluginSettingsPersistenceError extends Schema.TaggedError()( + "PluginSettingsPersistenceError", + { + pluginId: PluginId, + operation: Schema.Literals(["read", "write", "readSecret", "writeSecret"]), + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to ${persistenceOperations[this.operation]} of plugin ${this.pluginId}.`; + } +} + +export class PluginSettingsSchemaError extends Schema.TaggedError()( + "PluginSettingsSchemaError", + { + pluginId: PluginId, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Plugin ${this.pluginId} has a settings schema T3 Code cannot validate.`; + } +} + +export class PluginSettingsValidationError extends Schema.TaggedError()( + "PluginSettingsValidationError", + { + pluginId: PluginId, + summary: Schema.String, + }, +) { + override get message(): string { + return `Invalid settings for plugin ${this.pluginId}: ${this.summary}`; + } +} + +export class PluginSettingsStore extends Context.Service< + PluginSettingsStore, + { + /** Stored settings with secret values, for the plugin. JSON Schema defaults are not applied. */ + readonly get: ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + ) => Effect.Effect; + /** Stored settings for clients. A secret that has a value reads as `SECRET_REDACTED`. */ + readonly getForClient: ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + ) => Effect.Effect; + /** + * Merges `patch` over the stored settings at the top level, where `null` removes a key and + * `SECRET_REDACTED` keeps a secret, validates the result, saves it and returns the client view. + */ + readonly update: ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + patch: Schema.JsonObject, + ) => Effect.Effect< + Schema.JsonObject, + PluginSettingsPersistenceError | PluginSettingsSchemaError | PluginSettingsValidationError + >; + /** The plugin view after each successful `update` of that plugin. */ + readonly changes: (pluginId: PluginId) => Stream.Stream; + } +>()("t3/plugins/PluginSettingsStore") {} + +interface StoredSettings { + readonly values: Schema.JsonObject; + readonly secrets: Readonly>; +} + +interface SettingsChange { + readonly pluginId: PluginId; + readonly settings: Schema.JsonObject; +} + +const SettingsJson = fromJsonStringPretty(Schema.JsonObject); +const decodeSettingsJson = Schema.decodeUnknownEffect(SettingsJson); +const encodeSettingsJson = Schema.encodeEffect(SettingsJson); +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + +// Top-level string fields with `format: "t3-secret"` live in ServerSecretStore, never settings.json. +const secretFields = (settingsSchema: Schema.JsonObject | undefined): ReadonlySet => { + const properties = settingsSchema?.properties; + if (!Predicate.isObject(properties)) return new Set(); + return new Set( + Object.entries(properties).flatMap(([field, property]) => + Predicate.isObject(property) && property.type === "string" && property.format === "t3-secret" + ? [field] + : [], + ), + ); +}; + +// Secret names become file names, so the field is base64url encoded. +const secretName = (pluginId: PluginId, field: string) => + `plugin-${pluginId}-${Buffer.from(field, "utf8").toString("base64url")}`; + +const formatPathSegment = (segment: PropertyKey | { readonly key: PropertyKey }) => { + const key = Predicate.isObject(segment) ? segment.key : segment; + return typeof key === "number" ? `[${key}]` : `["${String(key)}"]`; +}; + +// Issues never carry input values because decoding does not set `reportInput`. +const summarizeIssue = (issue: SchemaIssue.Issue) => { + const summary = SchemaIssue.makeFormatterStandardSchemaV1()(issue) + .issues.map(({ path, message }) => + path === undefined || path.length === 0 + ? message + : `at ${path.map(formatPathSegment).join("")}: ${message}`, + ) + .join("; "); + return summary.length > MAX_VALIDATION_SUMMARY_LENGTH + ? `${summary.slice(0, MAX_VALIDATION_SUMMARY_LENGTH - 1)}…` + : summary; +}; + +const compileSettingsSchema = (settingsSchema: Schema.JsonObject) => + Result.try(() => + Schema.decodeUnknownResult( + // Imported JSON Schemas need no decoding services, which `Schema.Top` cannot express. + Schema.make>( + SchemaRepresentation.fromJsonSchemaDocument( + JsonSchema.fromSchemaDraft2020_12(settingsSchema), + // Plugins are trusted code, so their patterns are applied rather than rejected. + { patterns: "apply" }, + ).ast, + ), + { onExcessProperty: "error", errors: "all" }, + ), + ); + +const make = Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const secretStore = yield* ServerSecretStore.ServerSecretStore; + const writeSemaphore = yield* Semaphore.make(1); + // Sliding, so a subscriber that stops pulling drops old changes instead of blocking `update`. + const changesPubSub = yield* PubSub.sliding(64); + // Keyed by schema identity: a reloaded plugin brings a new manifest object. + const validators = new WeakMap>(); + + const settingsPath = (pluginId: PluginId) => + path.join(stateDir, "plugin-data", pluginId, "settings.json"); + + const validatorFor = (pluginId: PluginId, settingsSchema: Schema.JsonObject) => { + let validator = validators.get(settingsSchema); + if (validator === undefined) { + validator = compileSettingsSchema(settingsSchema); + validators.set(settingsSchema, validator); + } + return Result.isSuccess(validator) + ? Effect.succeed(validator.success) + : Effect.fail(new PluginSettingsSchemaError({ pluginId, cause: validator.failure })); + }; + + // Shares the span of the public operation that reads. + const load = Effect.fnUntraced( + function* (pluginId: PluginId, fields: ReadonlySet) { + const raw = yield* fileSystem.readFileString(settingsPath(pluginId)).pipe( + Effect.asSome, + Effect.catchIf( + (error) => error.reason._tag === "NotFound", + () => Effect.succeedNone, + ), + ); + const stored = Option.isSome(raw) ? yield* decodeSettingsJson(raw.value) : {}; + const values = Object.fromEntries(Object.entries(stored).filter(([key]) => !fields.has(key))); + const secrets = yield* Effect.forEach( + fields, + (field) => + secretStore.get(secretName(pluginId, field)).pipe( + Effect.mapError( + (cause) => + new PluginSettingsPersistenceError({ pluginId, operation: "readSecret", cause }), + ), + Effect.map(Option.map((bytes) => [field, textDecoder.decode(bytes)] as const)), + ), + { concurrency: "unbounded" }, + ); + return { + values, + secrets: Object.fromEntries(secrets.flatMap((secret) => Option.toArray(secret))), + } satisfies StoredSettings; + }, + (effect, pluginId) => + effect.pipe( + Effect.catchTags({ + PlatformError: (cause) => + Effect.fail(new PluginSettingsPersistenceError({ pluginId, operation: "read", cause })), + SchemaError: (cause) => + Effect.fail(new PluginSettingsPersistenceError({ pluginId, operation: "read", cause })), + }), + ), + ); + + const clientView = ({ values, secrets }: StoredSettings): Schema.JsonObject => ({ + ...values, + ...Object.fromEntries(Object.keys(secrets).map((field) => [field, SECRET_REDACTED])), + }); + + const get = Effect.fn("PluginSettingsStore.get")(function* ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + ) { + const { values, secrets } = yield* load(pluginId, secretFields(settingsSchema)); + return { ...values, ...secrets } satisfies Schema.JsonObject; + }); + + const getForClient = Effect.fn("PluginSettingsStore.getForClient")(function* ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + ) { + return clientView(yield* load(pluginId, secretFields(settingsSchema))); + }); + + const update = Effect.fn("PluginSettingsStore.update")(function* ( + pluginId: PluginId, + settingsSchema: Schema.JsonObject | undefined, + patch: Schema.JsonObject, + ) { + const fields = secretFields(settingsSchema); + const validate = + settingsSchema === undefined ? undefined : yield* validatorFor(pluginId, settingsSchema); + return yield* writeSemaphore.withPermits(1)( + Effect.gen(function* () { + const current = yield* load(pluginId, fields); + const merged: Record = { ...current.values, ...current.secrets }; + for (const [key, value] of Object.entries(patch)) { + if (fields.has(key) && value === SECRET_REDACTED) continue; + if (value === null) delete merged[key]; + else merged[key] = value; + } + + // Secrets are validated with their real values so `required` and type checks hold. + const validation = validate?.(merged); + if (validation !== undefined && Result.isFailure(validation)) { + return yield* new PluginSettingsValidationError({ + pluginId, + summary: summarizeIssue(validation.failure.issue), + }); + } + + const next: StoredSettings = { + values: Object.fromEntries(Object.entries(merged).filter(([key]) => !fields.has(key))), + secrets: Object.fromEntries( + Object.entries(merged).filter( + (entry): entry is [string, string] => + fields.has(entry[0]) && typeof entry[1] === "string", + ), + ), + }; + yield* Effect.forEach( + fields, + (field) => { + const value = next.secrets[field]; + if (value === current.secrets[field]) return Effect.void; + const name = secretName(pluginId, field); + return ( + value === undefined + ? secretStore.remove(name) + : secretStore.set(name, textEncoder.encode(value)) + ).pipe( + Effect.mapError( + (cause) => + new PluginSettingsPersistenceError({ pluginId, operation: "writeSecret", cause }), + ), + ); + }, + { discard: true }, + ); + yield* encodeSettingsJson(next.values).pipe( + Effect.flatMap((contents) => + writeFileStringAtomically({ filePath: settingsPath(pluginId), contents }), + ), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.mapError( + (cause) => new PluginSettingsPersistenceError({ pluginId, operation: "write", cause }), + ), + ); + + yield* PubSub.publish(changesPubSub, { + pluginId, + settings: { ...next.values, ...next.secrets }, + }); + return clientView(next); + }), + ); + }); + + const changes = (pluginId: PluginId) => + Stream.fromPubSub(changesPubSub).pipe( + Stream.filter((change) => change.pluginId === pluginId), + Stream.map((change) => change.settings), + ); + + return PluginSettingsStore.of({ get, getForClient, update, changes }); +}); + +export const layer = Layer.effect(PluginSettingsStore, make); diff --git a/apps/server/src/plugins/PluginStorage.test.ts b/apps/server/src/plugins/PluginStorage.test.ts new file mode 100644 index 0000000000000..022fa4fa13d5f --- /dev/null +++ b/apps/server/src/plugins/PluginStorage.test.ts @@ -0,0 +1,110 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { PluginId } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as ServerConfig from "../config.ts"; +import * as PluginStorage from "./PluginStorage.ts"; + +const hello = PluginId.make("hello"); +const other = PluginId.make("other"); + +const layerConfig = () => + Layer.fresh(ServerConfig.layerTest(process.cwd(), { prefix: "t3code-plugin-storage-test-" })); + +// A fresh layer per call, so closing it closes its SQLite connection. +const withStorage = ( + use: (storage: PluginStorage.PluginStorage["Service"]) => Effect.Effect, +) => PluginStorage.PluginStorage.use(use).pipe(Effect.provide(Layer.fresh(PluginStorage.layer))); + +it.layer(NodeServices.layer)("PluginStorage", (it) => { + it.effect("creates its own plugins.sqlite on first use", () => + Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const filename = path.join(stateDir, "plugins.sqlite"); + + yield* withStorage((storage) => + Effect.gen(function* () { + assert.isFalse(yield* fileSystem.exists(filename)); + yield* storage.set(hello, "greeting", { text: "hi" }); + }), + ); + + const rows = yield* SqlClient.SqlClient.use( + (sql) => sql<{ readonly key: string }>`SELECT key FROM plugin_storage`, + ).pipe(Effect.provide(NodeSqliteClient.layer({ filename, readonly: true }))); + assert.deepStrictEqual( + rows.map((row) => row.key), + ["greeting"], + ); + assert.isFalse(yield* fileSystem.exists(path.join(stateDir, "statev2.sqlite"))); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("round-trips values and keeps them after the layer closes", () => + Effect.gen(function* () { + yield* withStorage((storage) => + Effect.gen(function* () { + yield* storage.set(hello, "b", [1, "two", null]); + yield* storage.set(hello, "a", { nested: { ok: true } }); + yield* storage.set(hello, "gone", 1); + yield* storage.set(other, "a", "other plugin"); + yield* storage.set(hello, "a", { nested: { ok: false } }); + yield* storage.remove(hello, "gone"); + + assert.deepStrictEqual(yield* storage.list(hello), ["a", "b"]); + assert.deepStrictEqual(yield* storage.get(hello, "gone"), Option.none()); + }), + ); + + yield* withStorage((storage) => + Effect.gen(function* () { + assert.deepStrictEqual( + yield* storage.get(hello, "a"), + Option.some({ nested: { ok: false } }), + ); + assert.deepStrictEqual(yield* storage.get(hello, "b"), Option.some([1, "two", null])); + assert.deepStrictEqual(yield* storage.get(other, "a"), Option.some("other plugin")); + }), + ); + }).pipe(Effect.provide(layerConfig())), + ); + + it.effect("matches list prefixes literally", () => + withStorage((storage) => + Effect.gen(function* () { + for (const key of ["a_b", "axb", "A_b", "a%c", "abc"]) { + yield* storage.set(hello, key, true); + } + yield* storage.set(other, "a_other", true); + + assert.deepStrictEqual(yield* storage.list(hello, "a_"), ["a_b"]); + assert.deepStrictEqual(yield* storage.list(hello, "a%"), ["a%c"]); + }), + ).pipe(Effect.provide(layerConfig())), + ); + + it.effect("rejects oversize values and keys", () => + withStorage((storage) => + Effect.gen(function* () { + const tooLarge = yield* Effect.flip(storage.set(hello, "big", "x".repeat(1024 * 1024))); + assert.strictEqual(tooLarge._tag, "PluginStorageValueTooLargeError"); + + const emptyKey = yield* Effect.flip(storage.set(hello, "", 1)); + assert.strictEqual(emptyKey._tag, "PluginStorageKeyError"); + const longKey = yield* Effect.flip(storage.get(hello, "k".repeat(513))); + assert.strictEqual(longKey._tag, "PluginStorageKeyError"); + + assert.deepStrictEqual(yield* storage.list(hello), []); + }), + ).pipe(Effect.provide(layerConfig())), + ); +}); diff --git a/apps/server/src/plugins/PluginStorage.ts b/apps/server/src/plugins/PluginStorage.ts new file mode 100644 index 0000000000000..e5bb5b498df29 --- /dev/null +++ b/apps/server/src/plugins/PluginStorage.ts @@ -0,0 +1,237 @@ +import { PluginId } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as ServerConfig from "../config.ts"; + +const MAX_KEY_LENGTH = 512; +const MAX_VALUE_BYTES = 1024 * 1024; + +export class PluginStoragePersistenceError extends Schema.TaggedError()( + "PluginStoragePersistenceError", + { + pluginId: PluginId, + operation: Schema.Literals(["open", "get", "set", "remove", "list"]), + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Plugin storage ${this.operation} failed for plugin ${this.pluginId}.`; + } +} + +export class PluginStorageKeyError extends Schema.TaggedError()( + "PluginStorageKeyError", + { + pluginId: PluginId, + length: Schema.Number, + }, +) { + override get message(): string { + return `Plugin storage keys must be 1 to ${MAX_KEY_LENGTH} characters long, got ${this.length}.`; + } +} + +export class PluginStorageInvalidValueError extends Schema.TaggedError()( + "PluginStorageInvalidValueError", + { + pluginId: PluginId, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return "Plugin storage values must be JSON."; + } +} + +export class PluginStorageValueTooLargeError extends Schema.TaggedError()( + "PluginStorageValueTooLargeError", + { + pluginId: PluginId, + bytes: Schema.Number, + }, +) { + override get message(): string { + return `Plugin storage values must be at most ${MAX_VALUE_BYTES} bytes of JSON, got ${this.bytes}.`; + } +} + +export class PluginStorage extends Context.Service< + PluginStorage, + { + readonly get: ( + pluginId: PluginId, + key: string, + ) => Effect.Effect< + Option.Option, + PluginStoragePersistenceError | PluginStorageKeyError + >; + readonly set: ( + pluginId: PluginId, + key: string, + value: Schema.Json, + ) => Effect.Effect< + void, + | PluginStoragePersistenceError + | PluginStorageKeyError + | PluginStorageInvalidValueError + | PluginStorageValueTooLargeError + >; + readonly remove: ( + pluginId: PluginId, + key: string, + ) => Effect.Effect; + /** Keys that start with `prefix`, sorted. */ + readonly list: ( + pluginId: PluginId, + prefix?: string, + ) => Effect.Effect, PluginStoragePersistenceError>; + } +>()("t3/plugins/PluginStorage") {} + +const ValueJson = Schema.fromJsonString(Schema.Json); +const encodeValue = Schema.encodeUnknownEffect(ValueJson); +const decodeValue = Schema.decodeUnknownEffect(ValueJson); + +const checkKey = (pluginId: PluginId, key: string) => + key.length >= 1 && key.length <= MAX_KEY_LENGTH + ? Effect.void + : Effect.fail(new PluginStorageKeyError({ pluginId, length: key.length })); + +const make = Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const path = yield* Path.Path; + const layerScope = yield* Effect.scope; + const filename = path.join(stateDir, "plugins.sqlite"); + + // Opened on first use so a server without plugins never creates the file. A failed open is + // retried on the next call; a successful one stays open until the layer closes. + const sqlClient = yield* Effect.cachedWithTTL( + Effect.gen(function* () { + const scope = yield* Scope.fork(layerScope); + return yield* Layer.buildWithScope(NodeSqliteClient.layer({ filename }), scope).pipe( + Effect.map((context) => Context.get(context, SqlClient.SqlClient)), + Effect.tap( + (sql) => sql` + CREATE TABLE IF NOT EXISTS plugin_storage ( + plugin_id TEXT NOT NULL, + key TEXT NOT NULL, + value_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (plugin_id, key) + ) + `, + ), + Effect.onError((cause) => Scope.close(scope, Exit.failCause(cause))), + ); + }), + (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero), + ); + + const open = (pluginId: PluginId) => + sqlClient.pipe( + Effect.mapError( + (cause) => new PluginStoragePersistenceError({ pluginId, operation: "open", cause }), + ), + ); + + const get = Effect.fn("PluginStorage.get")( + function* (pluginId: PluginId, key: string) { + yield* checkKey(pluginId, key); + const sql = yield* open(pluginId); + const rows = yield* sql<{ readonly value_json: string }>` + SELECT value_json FROM plugin_storage WHERE plugin_id = ${pluginId} AND key = ${key} + `; + const row = rows[0]; + return row === undefined ? Option.none() : Option.some(yield* decodeValue(row.value_json)); + }, + (effect, pluginId) => + effect.pipe( + Effect.catchTags({ + SqlError: (cause) => + Effect.fail(new PluginStoragePersistenceError({ pluginId, operation: "get", cause })), + SchemaError: (cause) => + Effect.fail(new PluginStoragePersistenceError({ pluginId, operation: "get", cause })), + }), + ), + ); + + const set = Effect.fn("PluginStorage.set")( + function* (pluginId: PluginId, key: string, value: Schema.Json) { + yield* checkKey(pluginId, key); + const valueJson = yield* encodeValue(value).pipe( + Effect.mapError((cause) => new PluginStorageInvalidValueError({ pluginId, cause })), + ); + const bytes = Buffer.byteLength(valueJson, "utf8"); + if (bytes > MAX_VALUE_BYTES) { + return yield* new PluginStorageValueTooLargeError({ pluginId, bytes }); + } + const sql = yield* open(pluginId); + const updatedAt = DateTime.formatIso(yield* DateTime.now); + yield* sql` + INSERT INTO plugin_storage (plugin_id, key, value_json, updated_at) + VALUES (${pluginId}, ${key}, ${valueJson}, ${updatedAt}) + ON CONFLICT (plugin_id, key) + DO UPDATE SET value_json = excluded.value_json, updated_at = excluded.updated_at + `; + }, + (effect, pluginId) => + effect.pipe( + Effect.catchTags({ + SqlError: (cause) => + Effect.fail(new PluginStoragePersistenceError({ pluginId, operation: "set", cause })), + }), + ), + ); + + const remove = Effect.fn("PluginStorage.remove")( + function* (pluginId: PluginId, key: string) { + yield* checkKey(pluginId, key); + const sql = yield* open(pluginId); + yield* sql`DELETE FROM plugin_storage WHERE plugin_id = ${pluginId} AND key = ${key}`; + }, + (effect, pluginId) => + effect.pipe( + Effect.catchTags({ + SqlError: (cause) => + Effect.fail( + new PluginStoragePersistenceError({ pluginId, operation: "remove", cause }), + ), + }), + ), + ); + + const list = Effect.fn("PluginStorage.list")( + function* (pluginId: PluginId, prefix = "") { + const sql = yield* open(pluginId); + // Not LIKE: it treats `%` and `_` as wildcards and ignores ASCII case. + const rows = yield* sql<{ readonly key: string }>` + SELECT key FROM plugin_storage + WHERE plugin_id = ${pluginId} AND substr(key, 1, length(${prefix})) = ${prefix} + ORDER BY key + `; + return rows.map((row) => row.key); + }, + (effect, pluginId) => + effect.pipe( + Effect.catchTags({ + SqlError: (cause) => + Effect.fail(new PluginStoragePersistenceError({ pluginId, operation: "list", cause })), + }), + ), + ); + + return PluginStorage.of({ get, set, remove, list }); +}); + +export const layer = Layer.effect(PluginStorage, make); diff --git a/apps/server/src/plugins/pluginDiscovery.test.ts b/apps/server/src/plugins/pluginDiscovery.test.ts new file mode 100644 index 0000000000000..de9c55acbf1f3 --- /dev/null +++ b/apps/server/src/plugins/pluginDiscovery.test.ts @@ -0,0 +1,57 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; + +import { discoverPlugins } from "./pluginDiscovery.ts"; + +const manifest = (id: string) => + JSON.stringify({ id, name: id, version: "1.0.0", engines: { t3: ">=0.1.0" } }); + +it.layer(NodeServices.layer)("discoverPlugins", (it) => { + it.effect("reports valid, missing, invalid, mismatched and duplicate plugins", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-plugin-discovery-" }); + const pluginsDir = path.join(root, "plugins"); + const write = Effect.fn(function* (dir: string, contents: string | undefined) { + yield* fs.makeDirectory(dir, { recursive: true }); + if (contents !== undefined) { + yield* fs.writeFileString(path.join(dir, "t3-plugin.json"), contents); + } + }); + yield* write(path.join(pluginsDir, "valid"), manifest("valid")); + yield* write(path.join(pluginsDir, "empty"), undefined); + yield* write( + path.join(pluginsDir, "invalid"), + JSON.stringify({ ...JSON.parse(manifest("invalid")), version: "1" }), + ); + yield* write(path.join(pluginsDir, "renamed"), manifest("other-name")); + yield* write(path.join(pluginsDir, "Not_An_Id"), manifest("ignored")); + const devCopy = path.join(root, "dev", "valid"); + yield* write(devCopy, manifest("valid")); + const devOnly = path.join(root, "dev", "workbench"); + yield* write(devOnly, manifest("dev-only")); + + const plugins = yield* discoverPlugins({ pluginsDir, devPaths: [devCopy, devOnly] }); + const summary = plugins.map((plugin) => ({ + id: plugin.id, + source: plugin.source, + stage: "failure" in plugin ? plugin.failure.stage : "ok", + })); + + assert.deepStrictEqual(summary, [ + { id: "empty", source: "installed", stage: "discovery" }, + { id: "invalid", source: "installed", stage: "manifest" }, + { id: "renamed", source: "installed", stage: "manifest" }, + { id: "valid", source: "installed", stage: "ok" }, + { id: "valid", source: "dev", stage: "discovery" }, + { id: "dev-only", source: "dev", stage: "ok" }, + ]); + const invalid = plugins.find((plugin) => plugin.id === "invalid"); + assert.include(invalid && "failure" in invalid ? invalid.failure.message : "", "version"); + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/plugins/pluginDiscovery.ts b/apps/server/src/plugins/pluginDiscovery.ts new file mode 100644 index 0000000000000..9e48f8a8968de --- /dev/null +++ b/apps/server/src/plugins/pluginDiscovery.ts @@ -0,0 +1,180 @@ +import { + PluginId, + PluginManifest, + type PluginFailure, + type PluginSource, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +const PLUGIN_MANIFEST_FILE = "t3-plugin.json"; + +// Discovery runs at startup and on every reload, so a plugins folder full of unrelated +// directories or an oversized manifest must not stall the server. +const MAX_PLUGIN_DIRECTORIES = 64; +const MAX_MANIFEST_BYTES = 64 * 1024; +const MAX_FAILURE_MESSAGE_CHARS = 500; + +const decodeManifest = Schema.decodeUnknownExit(Schema.fromJsonString(PluginManifest)); +const decodeLooseId = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Struct({ id: PluginId })), +); +const isPluginId = Schema.is(PluginId); + +const causeMessage = (cause: Cause.Cause) => { + const error = Cause.squash(cause); + return error instanceof Error ? error.message : String(error); +}; + +interface PluginLocation { + readonly id: PluginId; + readonly rootDir: string; + readonly source: PluginSource; +} + +/** A plugin directory, with its manifest when it could be read and decoded. */ +export type DiscoveredPlugin = PluginLocation & + ({ readonly manifest: PluginManifest } | { readonly failure: PluginFailure }); + +const failed = (location: PluginLocation, failure: PluginFailure): DiscoveredPlugin => ({ + ...location, + failure: { ...failure, message: failure.message.slice(0, MAX_FAILURE_MESSAGE_CHARS) }, +}); + +/** + * Lists installed plugins (`//t3-plugin.json`) and dev plugins (`devPaths`). + * An installed plugin wins over a dev plugin with the same id, which is reported as failed. + */ +export const discoverPlugins = Effect.fn("PluginDiscovery.discover")(function* (input: { + readonly pluginsDir: string; + readonly devPaths: ReadonlyArray; +}) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + const readManifest = Effect.fnUntraced(function* (rootDir: string) { + const manifestPath = path.join(rootDir, PLUGIN_MANIFEST_FILE); + const info = yield* fs.stat(manifestPath).pipe(Effect.option); + if (info._tag === "None" || info.value.type !== "File") { + return { _tag: "Missing" } as const; + } + if (Number(info.value.size) > MAX_MANIFEST_BYTES) { + return { _tag: "TooLarge" } as const; + } + const raw = yield* fs.readFileString(manifestPath).pipe(Effect.option); + return raw._tag === "None" + ? ({ _tag: "Missing" } as const) + : ({ _tag: "Read", raw: raw.value } as const); + }); + + // Node caches modules by real path, so reload can only find a plugin's modules by its real root. + const inspect = Effect.fnUntraced(function* ( + location: string, + source: PluginSource, + fallbackId: PluginId | undefined, + ) { + const rootDir = yield* fs.realPath(location).pipe(Effect.orElseSucceed(() => location)); + const manifest = yield* readManifest(rootDir); + if (manifest._tag !== "Read") { + if (fallbackId === undefined) return undefined; + return failed( + { id: fallbackId, rootDir, source }, + { + stage: "discovery", + message: + manifest._tag === "Missing" + ? `No ${PLUGIN_MANIFEST_FILE} in the plugin folder.` + : `${PLUGIN_MANIFEST_FILE} is larger than ${MAX_MANIFEST_BYTES} bytes.`, + }, + ); + } + const decoded = decodeManifest(manifest.raw); + if (decoded._tag === "Success") { + return { id: decoded.value.id, rootDir, source, manifest: decoded.value } as const; + } + const looseId = decodeLooseId(manifest.raw); + const id = looseId._tag === "Some" ? looseId.value.id : fallbackId; + if (id === undefined) return undefined; + return failed( + { id, rootDir, source }, + { + stage: "manifest", + message: `Invalid ${PLUGIN_MANIFEST_FILE}: ${causeMessage(decoded.cause)}`, + }, + ); + }); + + const plugins: Array = []; + const seen = new Set(); + let examined = 0; + const underLimit = Effect.fnUntraced(function* (location: string) { + examined += 1; + if (examined <= MAX_PLUGIN_DIRECTORIES) return true; + yield* Effect.logWarning("ignoring plugin folders past the limit", { + path: location, + limit: MAX_PLUGIN_DIRECTORIES, + }); + return false; + }); + + const entries = yield* fs + .readDirectory(input.pluginsDir) + .pipe(Effect.orElseSucceed((): Array => [])); + for (const entry of entries.toSorted()) { + if (entry.startsWith(".")) continue; + const rootDir = path.join(input.pluginsDir, entry); + const info = yield* fs.stat(rootDir).pipe(Effect.option); + if (info._tag === "None" || info.value.type !== "Directory") continue; + if (!isPluginId(entry)) { + yield* Effect.logWarning("ignoring plugin folder whose name is not a plugin id", { + path: rootDir, + }); + continue; + } + if (!(yield* underLimit(input.pluginsDir))) break; + const plugin = yield* inspect(rootDir, "installed", entry); + if (plugin === undefined) continue; + seen.add(entry); + plugins.push( + "manifest" in plugin && plugin.manifest.id !== entry + ? failed( + { id: entry, rootDir, source: "installed" }, + { + stage: "manifest", + message: `The manifest id "${plugin.manifest.id}" does not match the folder name "${entry}".`, + }, + ) + : plugin, + ); + } + + for (const devPath of input.devPaths) { + const rootDir = path.resolve(devPath); + if (!(yield* underLimit(rootDir))) break; + const basename = path.basename(rootDir); + const plugin = yield* inspect(rootDir, "dev", isPluginId(basename) ? basename : undefined); + if (plugin === undefined) { + yield* Effect.logWarning("ignoring dev plugin path without a usable plugin id", { + path: rootDir, + }); + continue; + } + if (seen.has(plugin.id)) { + plugins.push( + failed(plugin, { + stage: "discovery", + message: + "Another plugin with this id is already loaded from the plugins folder or an earlier dev path.", + }), + ); + continue; + } + seen.add(plugin.id); + plugins.push(plugin); + } + + return plugins; +}); diff --git a/apps/server/src/plugins/pluginRpcHandlers.ts b/apps/server/src/plugins/pluginRpcHandlers.ts new file mode 100644 index 0000000000000..30c2601faf5ce --- /dev/null +++ b/apps/server/src/plugins/pluginRpcHandlers.ts @@ -0,0 +1,66 @@ +import { + type AuthEnvironmentScope, + PluginError, + type PluginId, + type PluginInvokeInput, + type PluginListInput, + type PluginReloadInput, + type PluginSetEnabledInput, + type PluginSettingsGetInput, + type PluginSettingsUpdateInput, + type PluginSubscribeInput, + WS_METHODS, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; + +import { rpcAuthorizationError } from "../auth/RpcAuthorization.ts"; +import * as PluginHost from "./PluginHost.ts"; + +const toRpcError = (pluginId: PluginId) => (error: { readonly message: string }) => + PluginHost.isPluginAccessDeniedError(error) + ? rpcAuthorizationError(error.requiredScope) + : new PluginError({ message: error.message, pluginId }); + +/** The `plugins.*` WebSocket handlers. `scopes` are the connected session's. */ +export const makePluginRpcHandlers = ( + pluginHost: PluginHost.PluginHost["Service"], + scopes: ReadonlyArray, +) => ({ + [WS_METHODS.pluginsList]: (_input: PluginListInput) => + pluginHost.list.pipe(Effect.map((plugins) => ({ plugins }))), + [WS_METHODS.pluginsSubscribeStates]: (_input: PluginListInput) => + pluginHost.streamStates.pipe(Stream.map((plugins) => ({ plugins }))), + [WS_METHODS.pluginsSetEnabled]: (input: PluginSetEnabledInput) => + pluginHost.setEnabled(input.pluginId, input.enabled).pipe( + Effect.map((plugin) => ({ plugin })), + Effect.mapError(toRpcError(input.pluginId)), + ), + [WS_METHODS.pluginsReload]: (input: PluginReloadInput) => + pluginHost.reload(input.pluginId).pipe( + Effect.map((plugin) => ({ plugin })), + Effect.mapError(toRpcError(input.pluginId)), + ), + [WS_METHODS.pluginsSettingsGet]: (input: PluginSettingsGetInput) => + pluginHost.getSettings(input.pluginId).pipe( + Effect.map((settings) => ({ pluginId: input.pluginId, settings })), + Effect.mapError(toRpcError(input.pluginId)), + ), + [WS_METHODS.pluginsSettingsUpdate]: (input: PluginSettingsUpdateInput) => + pluginHost.updateSettings(input.pluginId, input.patch).pipe( + Effect.map((settings) => ({ pluginId: input.pluginId, settings })), + Effect.mapError(toRpcError(input.pluginId)), + ), + [WS_METHODS.pluginsInvoke]: (input: PluginInvokeInput) => + pluginHost + .invoke({ pluginId: input.pluginId, method: input.method, input: input.input, scopes }) + .pipe( + Effect.map((output) => ({ output })), + Effect.mapError(toRpcError(input.pluginId)), + ), + [WS_METHODS.pluginsSubscribe]: (input: PluginSubscribeInput) => + pluginHost.subscribe(input).pipe( + Stream.map((value) => ({ value })), + Stream.mapError(toRpcError(input.pluginId)), + ), +}); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 452addfb4a667..f059acdb931d6 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -88,6 +88,9 @@ import * as PreviewBrowser from "./preview/PreviewBrowser.ts"; import * as ProcessRunner from "./processRunner.ts"; import * as GitManager from "./git/GitManager.ts"; import * as EnvironmentTheme from "./environmentTheme.ts"; +import * as PluginHost from "./plugins/PluginHost.ts"; +import * as PluginSettingsStore from "./plugins/PluginSettingsStore.ts"; +import * as PluginStorage from "./plugins/PluginStorage.ts"; import * as Keybindings from "./keybindings.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerOwnership from "./serverOwnership.ts"; @@ -681,6 +684,11 @@ const layerRuntimeCoreDependencies = layerRuntimeCoreDependenciesBase.pipe( Layer.provideMerge(Layer.mergeAll(NativeAppIconResolver.layer, layerProjectFaviconResolver)), Layer.provideMerge(layerRepositoryIdentityResolver), Layer.provideMerge(CodexManagedTokenBroker.layer), + Layer.provideMerge( + PluginHost.layer.pipe( + Layer.provide(Layer.mergeAll(PluginSettingsStore.layer, PluginStorage.layer)), + ), + ), Layer.provideMerge(layerServerEnvironment), Layer.provideMerge(layerAuth), Layer.provideMerge(ServerSecretStore.layer), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 641845f322ea6..c10d7ef9965cb 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -79,6 +79,7 @@ import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolve import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; +import * as PluginHost from "./plugins/PluginHost.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; import * as GuestShareService from "./guest/GuestShareService.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; @@ -529,6 +530,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(ResourceTelemetry.ResourceTelemetry)({}), Layer.mock(ReviewService.ReviewService)({}), Layer.mock(ScheduledTasks.ScheduledTaskService)({}), + Layer.mock(PluginHost.PluginHost)({}), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(SecretResolver.SecretResolver)({}), Layer.mock(GuestShareService.GuestShareService)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4781fcf78dde6..f7b7390cbcacd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -121,6 +121,8 @@ import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts" import * as DocumentText from "./provider/DocumentText.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; +import * as PluginHost from "./plugins/PluginHost.ts"; +import { makePluginRpcHandlers } from "./plugins/pluginRpcHandlers.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; import * as GuestShareService from "./guest/GuestShareService.ts"; import { @@ -1273,6 +1275,7 @@ export const layerWsRpc = ( const commandOutputHub = yield* CommandOutputHub.CommandOutputHub; const providerSessionManager = yield* ProviderSessionManager.ProviderSessionManagerV2; const scheduledTasks = yield* ScheduledTasks.ScheduledTaskService; + const pluginHost = yield* PluginHost.PluginHost; const secretRequests = yield* SecretRequests.SecretRequests; const guestShares = yield* GuestShareService.GuestShareService; const pullRequests = yield* PullRequestService.PullRequestService; @@ -1867,6 +1870,7 @@ export const layerWsRpc = ( }); const handlers = ServerWsRpcGroup.of({ + ...makePluginRpcHandlers(pluginHost, currentSession.scopes), [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) => Effect.annotateCurrentSpan({ "orchestration_v2.command_id": command.commandId, diff --git a/docs/internals/plugins.md b/docs/internals/plugins.md new file mode 100644 index 0000000000000..77f9cd7503528 --- /dev/null +++ b/docs/internals/plugins.md @@ -0,0 +1,70 @@ +# Plugins + +A plugin is a directory with a `t3-plugin.json` manifest +([schema](../../packages/contracts/src/plugins.ts)), an optional CommonJS server entry, and an +optional ESM web bundle, typed with [`@t3tools/plugin-sdk`](../../packages/plugin-sdk). The server +side is [`PluginHost`](../../apps/server/src/plugins/PluginHost.ts). + +## Trust and failure + +Plugins are code the user installed on purpose. Server entries run in the server process and web +bundles run as React code in the page, with no sandbox; manifest `permissions` are shown, not +enforced. Do not shape the API around a future sandbox. The host isolates failure instead: a plugin +that throws or hangs while loading or activating is marked failed, one that throws in a handler fails +that call, hung cleanups are abandoned after a deadline, and none of them stops the server. Enable, +disable and reload run in the host's scope, so a client that disconnects mid-change never leaves a +plugin half torn down. A newly found +plugin starts disabled; the user enables it. + +## The plugin API is plain JavaScript + +The server bundle inlines `effect`, so plugins never share the host's Effect runtime or `Schema` +instances. They get Promises, `AsyncIterable` streams and `AbortSignal` cancellation, never Effect. + +## Server entries are CommonJS + +The packaged server is a Node single executable, which cannot `import()` a file from disk. A +`require` made with `module.createRequire(process.execPath)` loads files in both the executable and +plain Node, so the host uses that one path, and the manifest accepts only `.cjs` entries. Reload +drops the plugin's entries from `require.cache`; Node keeps the old module instances in memory until +the server restarts. + +## State lives beside T3's + +Enabled flags and dev paths go in `/plugins.json`, settings in +`/plugin-data//settings.json`, storage in `/plugins.sqlite`, opened on first +use. Not in `settings.json`, whose writer re-encodes its decoded struct and drops unknown keys. Not a +migration in `statev2.sqlite`, whose numbered list refuses to start when a fork migration id +collides with an upstream one. Settings stay out of `plugins//` so replacing a plugin's files +cannot lose them and a dev plugin's settings never look like an installed plugin. + +## Ids and manifests + +Ids are lowercase because they name directories on case-insensitive filesystems, and have no +underscores so the `plg__` prefix on plugin MCP tool names belongs to one plugin. Entry paths are +checked lexically; code that reads or serves them must still resolve symlinks. `contributes` rejects +unknown keys, so a plugin written for a newer T3 fails with a reason instead of losing one silently. +Settings are validated by Effect's JSON Schema 2020-12 importer; a schema it cannot import rejects +saves with the importer's reason. + +## Clients + +Plugin state reaches clients through `plugins.subscribeStates`, not the server config stream: a +settings page that opted into the shared stream would need a second subscription that resends the +whole config. Desktop and hosted web authenticate `/api` with bearer tokens, which a bare +`import(url)` cannot send, so web bundles are fetched with the authenticated fetch and imported from +a `blob:` URL; the desktop CSP allows `blob:` scripts. Bundles reach React and the UI kit through +`globalThis.__t3PluginHost`, not an import map. Mobile release builds cannot load code at all, so +mobile renders settings from the manifest's JSON Schema and declarative contributions only. + +Where T3 has a closed union and a `switch` (right panel kinds, settings paths, keybinding commands, +drivers, MCP toolkits), plugins add to a registry that built-ins register into first. Where T3 +already has a server-supplied open variant (provider slash commands, dynamic tool items, themes), +plugins use it. Hot paths such as timeline and sidebar rows read a module-level `Map` and re-render +on one list-level version, never a per-row subscription. + +## Nothing installed costs nothing + +Plugins activate after the server is listening, each in its own fiber. Clients show plugin UI only +when the environment advertises `plugins`. Plugin MCP tools are registered only while a plugin is +installed, because every agent turn pays for the advertised tool list. diff --git a/knip.jsonc b/knip.jsonc index c2093737fd942..a7fcfe5e6aa3e 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -140,6 +140,12 @@ // These workspace packages are private to this monorepo. "includeEntryExports": true, }, + "packages/plugin-sdk": { + // Plugin authors outside this repository import these exports. + "includeEntryExports": false, + // The plugin host loads fixture entries from their manifests. + "entry": ["fixtures/*/*.cjs"], + }, "packages/effect-acp": { "includeEntryExports": true, "entry": ["test/fixtures/*.ts"], diff --git a/package.json b/package.json index c78b879301b9e..e80dc1015c9f3 100644 --- a/package.json +++ b/package.json @@ -30,7 +30,7 @@ "report:renders": "node scripts/render-metrics.ts", "lint": "vp lint --report-unused-disable-directives", "knip": "knip --preprocessor ./scripts/knip-schemas.ts", - "knip:check": "knip --include files,dependencies --no-config-hints && knip --workspace apps/server --workspace apps/desktop --workspace apps/web --workspace packages/client-runtime --workspace packages/contracts --workspace packages/effect-acp --workspace packages/effect-codex-app-server --workspace packages/shared --workspace packages/ssh --workspace packages/tailscale --workspace packages/provider-core --workspace packages/provider-acp --workspace packages/provider-acp-registry --workspace packages/provider-testing --workspace packages/provider-pi --workspace packages/provider-muse --workspace packages/provider-opencode --workspace packages/provider-cursor --workspace packages/provider-fx --workspace packages/provider-grok --workspace packages/secret-provider-core --workspace packages/secret-provider-definitions --workspace packages/secret-provider-onepassword --workspace packages/source-control-core --workspace packages/source-control-definitions --workspace packages/source-control-azure-devops --workspace packages/source-control-bitbucket --workspace packages/source-control-forgejo --workspace packages/source-control-gitcafe --workspace packages/source-control-github --workspace packages/source-control-gitlab --workspace packages/source-control-testing --exports --preprocessor ./scripts/knip-schemas.ts --no-config-hints", + "knip:check": "knip --include files,dependencies --no-config-hints && knip --workspace apps/server --workspace apps/desktop --workspace apps/web --workspace packages/client-runtime --workspace packages/contracts --workspace packages/effect-acp --workspace packages/effect-codex-app-server --workspace packages/shared --workspace packages/ssh --workspace packages/tailscale --workspace packages/plugin-sdk --workspace packages/provider-core --workspace packages/provider-acp --workspace packages/provider-acp-registry --workspace packages/provider-testing --workspace packages/provider-pi --workspace packages/provider-muse --workspace packages/provider-opencode --workspace packages/provider-cursor --workspace packages/provider-fx --workspace packages/provider-grok --workspace packages/secret-provider-core --workspace packages/secret-provider-definitions --workspace packages/secret-provider-onepassword --workspace packages/source-control-core --workspace packages/source-control-definitions --workspace packages/source-control-azure-devops --workspace packages/source-control-bitbucket --workspace packages/source-control-forgejo --workspace packages/source-control-gitcafe --workspace packages/source-control-github --workspace packages/source-control-gitlab --workspace packages/source-control-testing --exports --preprocessor ./scripts/knip-schemas.ts --no-config-hints", "knip:production": "knip --production --preprocessor ./scripts/knip-schemas.ts", "lint:mobile": "node scripts/mobile-native-static-check.ts", "mobile:install": "node scripts/mobile-native-client.ts ensure", diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index d3139f2eeff24..f83fabb9ad04a 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -57,6 +57,10 @@ export const CLIENT_GUARDED_RPC_SCOPES = { [WS_METHODS.scheduledTasksUpsert]: AuthOrchestrationOperateScope, [WS_METHODS.scheduledTasksSetEnabled]: AuthOrchestrationOperateScope, [WS_METHODS.scheduledTasksDelete]: AuthOrchestrationOperateScope, + // Enabling a plugin runs its code in the server, like pointing a provider at a binary. + [WS_METHODS.pluginsSetEnabled]: AuthSettingsWriteScope, + [WS_METHODS.pluginsReload]: AuthSettingsWriteScope, + [WS_METHODS.pluginsSettingsUpdate]: AuthSettingsWriteScope, [WS_METHODS.scheduledTasksRunNow]: AuthOrchestrationOperateScope, [WS_METHODS.scheduledTasksRotateWebhookToken]: AuthOrchestrationOperateScope, [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: AuthOrchestrationOperateScope, diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 9d60bec43fe5c..a6e2afb23a278 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -105,6 +105,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ threadForceStop: Schema.optionalKey(Schema.Boolean), /** Every protocol-2 server sends this; older mobile builds still read it. */ environmentThemes: Schema.optionalKey(Schema.Boolean), + /** Serves the `plugins.*` methods. Clients hide every plugin surface without it. */ + plugins: Schema.optionalKey(Schema.Boolean), usageLimitSources: Schema.optionalKey(Schema.Boolean), /** Honors `workspaceSnapshotsOnDemand`; older servers send every snapshot and drop them in the stream. */ providerWorkspaceSnapshotsOnDemand: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index f00976d243b37..9138cdbb11982 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -62,6 +62,7 @@ export * from "./previewAutomation.ts"; export * from "./previewCapture.ts"; export * from "./usage.ts"; export * from "./scheduledTask.ts"; +export * from "./plugins.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/plugins.test.ts b/packages/contracts/src/plugins.test.ts new file mode 100644 index 0000000000000..aa72156f3a749 --- /dev/null +++ b/packages/contracts/src/plugins.test.ts @@ -0,0 +1,46 @@ +import * as Schema from "effect/Schema"; +import { describe, expect, it } from "vite-plus/test"; + +import { PluginManifest } from "./plugins.ts"; + +const decodeManifest = Schema.decodeUnknownSync(PluginManifest); + +const manifest = { + id: "hello-plugin", + name: "Hello", + version: "0.1.0", + engines: { t3: ">=0.1.6" }, + server: "./dist/server.cjs", + web: "dist/web.js", + settingsSchema: { type: "object", properties: { greeting: { type: "string" } } }, + permissions: ["Reads thread titles"], + contributes: {}, +}; + +describe("PluginManifest", () => { + it("decodes a valid manifest", () => { + expect(decodeManifest(manifest)).toEqual(manifest); + expect(decodeManifest({ ...manifest, version: "1.0.0-beta.2+build.5" }).version).toBe( + "1.0.0-beta.2+build.5", + ); + expect(decodeManifest({ ...manifest, engines: { t3: "^0.1 || >=1.0.0 <2" } }).engines.t3).toBe( + "^0.1 || >=1.0.0 <2", + ); + }); + + it.each([ + ["a missing id", { id: undefined }], + ["an id with uppercase letters", { id: "Hello" }], + ["an id with an underscore", { id: "hello_plugin" }], + ["a version that is not semver", { version: "1.0" }], + ["an engines range the host cannot evaluate", { engines: { t3: "~0.1.6" } }], + ["an unknown contribution kind", { contributes: { panels: [] } }], + ["a server entry that escapes the plugin directory", { server: "../other/server.cjs" }], + ["an absolute server entry", { server: "/usr/lib/server.cjs" }], + ["a Windows server entry", { server: "C:\\plugins\\server.cjs" }], + ["a server entry that is not CommonJS", { server: "dist/server.js" }], + ["a web entry that escapes the plugin directory", { web: "dist/../../web.js" }], + ])("rejects %s", (_name, override) => { + expect(() => decodeManifest({ ...manifest, ...override })).toThrow(); + }); +}); diff --git a/packages/contracts/src/plugins.ts b/packages/contracts/src/plugins.ts new file mode 100644 index 0000000000000..b383961900432 --- /dev/null +++ b/packages/contracts/src/plugins.ts @@ -0,0 +1,197 @@ +import * as Schema from "effect/Schema"; + +import { ForwardCompatibleArray, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +// Lowercase because the id names the plugin's directory on case-insensitive filesystems. No +// underscores, so the `plg__` prefix on plugin MCP tool names belongs to exactly one plugin. +export const PluginId = TrimmedNonEmptyString.check( + Schema.isMaxLength(32), + Schema.isPattern(/^[a-z][a-z0-9-]*$/), +).pipe(Schema.brand("PluginId")); +export type PluginId = typeof PluginId.Type; + +// https://semver.org/#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string +const SEMVER_PATTERN = + /^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$/; + +const PluginVersion = TrimmedNonEmptyString.check( + Schema.isMaxLength(64), + Schema.isPattern(SEMVER_PATTERN, { expected: "a semver version such as 1.2.3" }), +); + +// The syntax `satisfiesSemverRange` in packages/shared/src/semver.ts evaluates. Anything else, +// such as `~1.2`, would never match, so it fails here with a reason instead. +const ENGINE_COMPARATOR = String.raw`(?:\^|>=|>|<=|<|=)?v?\d+(?:\.\d+){0,2}`; +const ENGINE_GROUP = String.raw`${ENGINE_COMPARATOR}(?:\s+${ENGINE_COMPARATOR})*`; +const PluginEngineRange = TrimmedNonEmptyString.check( + Schema.isMaxLength(128), + Schema.isPattern(new RegExp(String.raw`^${ENGINE_GROUP}(?:\s*\|\|\s*${ENGINE_GROUP})*$`), { + expected: "a version range using ^, >=, >, <=, <, = and ||, such as >=0.1.6", + }), +); + +// Lexical check only. Code that reads or serves the file still resolves symlinks. +const isInsidePluginDirectory = Schema.makeFilter( + (path: string) => + !path.startsWith("/") && + !path.includes("\\") && + !/^[a-zA-Z]:/.test(path) && + path.split("/").every((segment) => segment !== ".."), + { expected: "a relative path inside the plugin directory" }, +); +const PluginEntryPath = TrimmedNonEmptyString.check( + Schema.isMaxLength(256), + isInsidePluginDirectory, +); + +// The packaged server is a Node single executable that can only `require` plugin code, and a +// `.js` file under a `"type": "module"` package.json would load as ESM. +const PluginServerEntryPath = PluginEntryPath.check( + Schema.isPattern(/\.cjs$/, { expected: "a CommonJS file ending in .cjs" }), +); + +// Each phase that designs a declarative contribution kind adds its key here. Unknown keys fail +// decoding, so a plugin written for a newer T3 says why instead of silently losing a contribution. +export const PluginContributions = Schema.StructWithRest(Schema.Struct({}), [ + Schema.Record(Schema.String, Schema.Never.annotate({ message: "Unknown contribution kind" })), +]); +export type PluginContributions = typeof PluginContributions.Type; + +/** `t3-plugin.json` at the root of a plugin directory. */ +export const PluginManifest = Schema.Struct({ + id: PluginId, + name: TrimmedNonEmptyString.check(Schema.isMaxLength(80)), + version: PluginVersion, + description: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(500))), + engines: Schema.Struct({ t3: PluginEngineRange }), + server: Schema.optionalKey(PluginServerEntryPath), + web: Schema.optionalKey(PluginEntryPath), + /** JSON Schema for the plugin's settings. Fields with `format: "t3-secret"` are write-only. */ + settingsSchema: Schema.optionalKey(Schema.JsonObject), + /** Shown to the user before enabling. Not enforced: plugins are trusted code. */ + permissions: Schema.optionalKey( + Schema.Array(TrimmedNonEmptyString.check(Schema.isMaxLength(200))).check( + Schema.isMaxLength(32), + ), + ), + contributes: Schema.optionalKey(PluginContributions), +}); +export type PluginManifest = typeof PluginManifest.Type; + +export const PluginSource = Schema.Literals(["installed", "dev"]); +export type PluginSource = typeof PluginSource.Type; + +export const PluginStatus = Schema.Literals(["active", "disabled", "failed", "loading"]); +export type PluginStatus = typeof PluginStatus.Type; + +export const PluginFailureStage = Schema.Literals([ + "discovery", + "manifest", + "compatibility", + "load", + "activate", +]); +export type PluginFailureStage = typeof PluginFailureStage.Type; + +export const PluginFailure = Schema.Struct({ + stage: PluginFailureStage, + message: Schema.String, +}); +export type PluginFailure = typeof PluginFailure.Type; + +/** What the server knows about one discovered plugin, including ones whose manifest failed. */ +export const PluginState = Schema.Struct({ + id: PluginId, + name: TrimmedNonEmptyString, + /** Absent when the manifest could not be decoded. */ + version: Schema.optionalKey(Schema.String), + description: Schema.optionalKey(Schema.String), + permissions: Schema.Array(Schema.String), + source: PluginSource, + enabled: Schema.Boolean, + status: PluginStatus, + failure: Schema.optionalKey(PluginFailure), + hasServer: Schema.Boolean, + hasWeb: Schema.Boolean, + settingsSchema: Schema.optionalKey(Schema.JsonObject), + contributes: Schema.optionalKey(PluginContributions), +}); +export type PluginState = typeof PluginState.Type; + +/** Name of a method or channel a plugin registered for `plugins.invoke` and `plugins.subscribe`. */ +export const PluginRpcName = TrimmedNonEmptyString.check( + Schema.isMaxLength(64), + Schema.isPattern(/^[a-zA-Z][a-zA-Z0-9._-]*$/), +); +export type PluginRpcName = typeof PluginRpcName.Type; + +export const PluginListInput = Schema.Struct({}); +export type PluginListInput = typeof PluginListInput.Type; + +export const PluginListResult = Schema.Struct({ + plugins: ForwardCompatibleArray(PluginState), +}); +export type PluginListResult = typeof PluginListResult.Type; + +export const PluginSetEnabledInput = Schema.Struct({ + pluginId: PluginId, + enabled: Schema.Boolean, +}); +export type PluginSetEnabledInput = typeof PluginSetEnabledInput.Type; + +export const PluginReloadInput = Schema.Struct({ + pluginId: PluginId, +}); +export type PluginReloadInput = typeof PluginReloadInput.Type; + +/** Result of `plugins.setEnabled` and `plugins.reload`. */ +export const PluginMutationResult = Schema.Struct({ + plugin: PluginState, +}); +export type PluginMutationResult = typeof PluginMutationResult.Type; + +export const PluginSettingsGetInput = Schema.Struct({ + pluginId: PluginId, +}); +export type PluginSettingsGetInput = typeof PluginSettingsGetInput.Type; + +export const PluginSettingsUpdateInput = Schema.Struct({ + pluginId: PluginId, + patch: Schema.JsonObject, +}); +export type PluginSettingsUpdateInput = typeof PluginSettingsUpdateInput.Type; + +/** Secret fields read back as the server's redaction marker, never their value. */ +export const PluginSettingsResult = Schema.Struct({ + pluginId: PluginId, + settings: Schema.JsonObject, +}); +export type PluginSettingsResult = typeof PluginSettingsResult.Type; + +export const PluginInvokeInput = Schema.Struct({ + pluginId: PluginId, + method: PluginRpcName, + input: Schema.optionalKey(Schema.Json), +}); +export type PluginInvokeInput = typeof PluginInvokeInput.Type; + +export const PluginInvokeResult = Schema.Struct({ + output: Schema.Json, +}); +export type PluginInvokeResult = typeof PluginInvokeResult.Type; + +export const PluginSubscribeInput = Schema.Struct({ + pluginId: PluginId, + channel: PluginRpcName, +}); +export type PluginSubscribeInput = typeof PluginSubscribeInput.Type; + +export const PluginSubscriptionEvent = Schema.Struct({ + value: Schema.Json, +}); +export type PluginSubscriptionEvent = typeof PluginSubscriptionEvent.Type; + +export class PluginError extends Schema.TaggedError()("PluginError", { + message: Schema.String, + pluginId: Schema.optionalKey(PluginId), +}) {} diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 37ea5866ae733..dc68b3db0a992 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -375,6 +375,21 @@ import { ScheduledTaskUpsertInput, ScheduledTaskMutationResult, } from "./scheduledTask.ts"; +import { + PluginError, + PluginInvokeInput, + PluginInvokeResult, + PluginListInput, + PluginListResult, + PluginMutationResult, + PluginReloadInput, + PluginSetEnabledInput, + PluginSettingsGetInput, + PluginSettingsResult, + PluginSettingsUpdateInput, + PluginSubscribeInput, + PluginSubscriptionEvent, +} from "./plugins.ts"; import { SecretRequestAnswerInput, SecretRequestError } from "./secretRequest.ts"; import { ProjectCloneActionInput, @@ -548,6 +563,15 @@ export const WS_METHODS = { scheduledTasksListWebhookDeliveries: "scheduledTasks.listWebhookDeliveries", scheduledTasksGetWebhookDelivery: "scheduledTasks.getWebhookDelivery", + pluginsList: "plugins.list", + pluginsSubscribeStates: "plugins.subscribeStates", + pluginsSetEnabled: "plugins.setEnabled", + pluginsReload: "plugins.reload", + pluginsSettingsGet: "plugins.settings.get", + pluginsSettingsUpdate: "plugins.settings.update", + pluginsInvoke: "plugins.invoke", + pluginsSubscribe: "plugins.subscribe", + cloudGetRelayClientStatus: "cloud.getRelayClientStatus", cloudInstallRelayClient: "cloud.installRelayClient", @@ -1950,6 +1974,59 @@ const WsScheduledTasksGetWebhookDeliveryRpc = Rpc.make( }, ); +const PluginRpcError = Schema.Union([PluginError, EnvironmentAuthorizationError]); + +const WsPluginsListRpc = Rpc.make(WS_METHODS.pluginsList, { + payload: PluginListInput, + success: PluginListResult, + error: PluginRpcError, +}); + +/** The current list, then the whole list again after every change. */ +const WsPluginsSubscribeStatesRpc = Rpc.make(WS_METHODS.pluginsSubscribeStates, { + payload: PluginListInput, + success: PluginListResult, + error: PluginRpcError, + stream: true, +}); + +const WsPluginsSetEnabledRpc = Rpc.make(WS_METHODS.pluginsSetEnabled, { + payload: PluginSetEnabledInput, + success: PluginMutationResult, + error: PluginRpcError, +}); + +const WsPluginsReloadRpc = Rpc.make(WS_METHODS.pluginsReload, { + payload: PluginReloadInput, + success: PluginMutationResult, + error: PluginRpcError, +}); + +const WsPluginsSettingsGetRpc = Rpc.make(WS_METHODS.pluginsSettingsGet, { + payload: PluginSettingsGetInput, + success: PluginSettingsResult, + error: PluginRpcError, +}); + +const WsPluginsSettingsUpdateRpc = Rpc.make(WS_METHODS.pluginsSettingsUpdate, { + payload: PluginSettingsUpdateInput, + success: PluginSettingsResult, + error: PluginRpcError, +}); + +const WsPluginsInvokeRpc = Rpc.make(WS_METHODS.pluginsInvoke, { + payload: PluginInvokeInput, + success: PluginInvokeResult, + error: PluginRpcError, +}); + +const WsPluginsSubscribeRpc = Rpc.make(WS_METHODS.pluginsSubscribe, { + payload: PluginSubscribeInput, + success: PluginSubscriptionEvent, + error: PluginRpcError, + stream: true, +}); + const WsSubscribeAuthAccessRpc = Rpc.make(WS_METHODS.subscribeAuthAccess, { payload: Schema.Struct({}), success: AuthAccessStreamEvent, @@ -2035,6 +2112,14 @@ export const WsRpcGroup = RpcGroup.make( WsSecretsAnswerRequestRpc, WsScheduledTasksListWebhookDeliveriesRpc, WsScheduledTasksGetWebhookDeliveryRpc, + WsPluginsListRpc, + WsPluginsSubscribeStatesRpc, + WsPluginsSetEnabledRpc, + WsPluginsReloadRpc, + WsPluginsSettingsGetRpc, + WsPluginsSettingsUpdateRpc, + WsPluginsInvokeRpc, + WsPluginsSubscribeRpc, WsServerReportClientActivityRpc, WsCloudGetRelayClientStatusRpc, WsCloudInstallRelayClientRpc, diff --git a/packages/plugin-sdk/fixtures/broken-plugin/server.cjs b/packages/plugin-sdk/fixtures/broken-plugin/server.cjs new file mode 100644 index 0000000000000..3c152c0154f1b --- /dev/null +++ b/packages/plugin-sdk/fixtures/broken-plugin/server.cjs @@ -0,0 +1,6 @@ +/** @type {import("@t3tools/plugin-sdk/server").PluginDefinition} */ +module.exports = { + activate() { + throw new Error("broken-plugin refuses to start"); + }, +}; diff --git a/packages/plugin-sdk/fixtures/broken-plugin/t3-plugin.json b/packages/plugin-sdk/fixtures/broken-plugin/t3-plugin.json new file mode 100644 index 0000000000000..2ad4ed7d34299 --- /dev/null +++ b/packages/plugin-sdk/fixtures/broken-plugin/t3-plugin.json @@ -0,0 +1,8 @@ +{ + "id": "broken-plugin", + "name": "Broken", + "version": "0.1.0", + "description": "Test fixture whose activate throws.", + "engines": { "t3": ">=0.1.0" }, + "server": "./server.cjs" +} diff --git a/packages/plugin-sdk/fixtures/hello-plugin/server.cjs b/packages/plugin-sdk/fixtures/hello-plugin/server.cjs new file mode 100644 index 0000000000000..f087c319cea49 --- /dev/null +++ b/packages/plugin-sdk/fixtures/hello-plugin/server.cjs @@ -0,0 +1,36 @@ +// Plain CommonJS with no SDK import, so the plugin host loads it as is. The type comes from the SDK. + +/** @type {import("@t3tools/plugin-sdk/server").PluginDefinition} */ +module.exports = { + activate(api) { + api.log.info("hello-plugin activated"); + api.rpc.register( + "ping", + async (input) => { + const settings = await api.settings.get(); + const greeting = typeof settings.greeting === "string" ? settings.greeting : "hello"; + return { pong: input ?? null, greeting }; + }, + { access: "read" }, + ); + api.rpc.register( + "count", + async () => { + const current = await api.storage.get("count"); + const next = (typeof current === "number" ? current : 0) + 1; + await api.storage.set("count", next); + api.rpc.publish("counted", next); + return next; + }, + { access: "write" }, + ); + api.onDeactivate(() => api.storage.set("deactivated", true)); + void (async () => { + // Only the greeting: settings include secret values, and any reader can subscribe. + for await (const settings of api.settings.changes()) { + const greeting = typeof settings.greeting === "string" ? settings.greeting : null; + api.rpc.publish("settings", { greeting }); + } + })(); + }, +}; diff --git a/packages/plugin-sdk/fixtures/hello-plugin/t3-plugin.json b/packages/plugin-sdk/fixtures/hello-plugin/t3-plugin.json new file mode 100644 index 0000000000000..dc8391fc6d822 --- /dev/null +++ b/packages/plugin-sdk/fixtures/hello-plugin/t3-plugin.json @@ -0,0 +1,16 @@ +{ + "id": "hello-plugin", + "name": "Hello", + "version": "0.1.0", + "description": "Test fixture that answers ping with its greeting setting and counts calls in storage.", + "engines": { "t3": ">=0.1.0" }, + "server": "./server.cjs", + "settingsSchema": { + "type": "object", + "properties": { + "greeting": { "type": "string", "maxLength": 40 }, + "token": { "type": "string", "format": "t3-secret" } + }, + "additionalProperties": false + } +} diff --git a/packages/plugin-sdk/package.json b/packages/plugin-sdk/package.json new file mode 100644 index 0000000000000..94e4c8db87cd0 --- /dev/null +++ b/packages/plugin-sdk/package.json @@ -0,0 +1,22 @@ +{ + "name": "@t3tools/plugin-sdk", + "private": true, + "type": "module", + "exports": { + "./server": { + "types": "./src/server.ts", + "import": "./src/server.ts" + }, + "./web": { + "types": "./src/web.ts", + "import": "./src/web.ts" + }, + "./shared": { + "types": "./src/shared.ts", + "import": "./src/shared.ts" + } + }, + "scripts": { + "typecheck": "tsc --noEmit" + } +} diff --git a/packages/plugin-sdk/src/server.ts b/packages/plugin-sdk/src/server.ts new file mode 100644 index 0000000000000..9af0eb27fe0c9 --- /dev/null +++ b/packages/plugin-sdk/src/server.ts @@ -0,0 +1,59 @@ +import type { JsonObject, JsonValue } from "./shared.ts"; + +/** `read` methods need the caller's `orchestration:read` scope, `write` methods `orchestration:operate`. */ +export type PluginRpcAccess = "read" | "write"; + +export interface PluginRpcContext { + /** Aborts when the caller goes away or the plugin deactivates. */ + readonly signal: AbortSignal; +} + +export type PluginRpcHandler = ( + input: JsonValue | undefined, + context: PluginRpcContext, +) => JsonValue | Promise; + +export interface PluginLogger { + readonly debug: (message: string, fields?: JsonObject) => void; + readonly info: (message: string, fields?: JsonObject) => void; + readonly warn: (message: string, fields?: JsonObject) => void; + readonly error: (message: string, fields?: JsonObject) => void; +} + +/** What `activate` receives. Everything is scoped to one activation and released on deactivate. */ +export interface PluginApi { + readonly version: 1; + readonly id: string; + readonly rootDir: string; + readonly log: PluginLogger; + readonly settings: { + readonly get: () => Promise; + /** Yields the settings, secret values included, after each change until deactivation. */ + readonly changes: () => AsyncIterable; + }; + readonly storage: { + readonly get: (key: string) => Promise; + readonly set: (key: string, value: JsonValue) => Promise; + readonly delete: (key: string) => Promise; + readonly list: (prefix?: string) => Promise>; + }; + readonly rpc: { + /** Answers `plugins.invoke` calls for `method` from clients and agents. */ + readonly register: ( + method: string, + handler: PluginRpcHandler, + options: { readonly access: PluginRpcAccess }, + ) => void; + /** Sends `value` to every `plugins.subscribe` caller on `channel`. Any session that can read this environment may subscribe, so never publish secrets. */ + readonly publish: (channel: string, value: JsonValue) => void; + }; + /** Runs when the plugin is disabled or reloaded, or the server stops; at once if that already happened. */ + readonly onDeactivate: (cleanup: () => void | Promise) => void; +} + +export interface PluginDefinition { + readonly activate: (api: PluginApi) => void | Promise; +} + +/** Types a server plugin. The CommonJS entry sets `module.exports` to the result. */ +export const definePlugin = (definition: PluginDefinition): PluginDefinition => definition; diff --git a/packages/plugin-sdk/src/shared.ts b/packages/plugin-sdk/src/shared.ts new file mode 100644 index 0000000000000..df08bd4af8224 --- /dev/null +++ b/packages/plugin-sdk/src/shared.ts @@ -0,0 +1,5 @@ +export type JsonValue = null | boolean | number | string | ReadonlyArray | JsonObject; + +export interface JsonObject { + readonly [key: string]: JsonValue; +} diff --git a/packages/plugin-sdk/src/web.ts b/packages/plugin-sdk/src/web.ts new file mode 100644 index 0000000000000..c607bd8893c9e --- /dev/null +++ b/packages/plugin-sdk/src/web.ts @@ -0,0 +1,12 @@ +/** What a web plugin's `activate` receives. */ +export interface WebPluginApi { + readonly version: 1; + readonly id: string; +} + +export interface WebPluginDefinition { + readonly activate: (api: WebPluginApi) => void | Promise; +} + +/** Types a web plugin. The ESM bundle's default export is the result. */ +export const defineWebPlugin = (definition: WebPluginDefinition): WebPluginDefinition => definition; diff --git a/packages/plugin-sdk/tsconfig.json b/packages/plugin-sdk/tsconfig.json new file mode 100644 index 0000000000000..87cd934c40bc1 --- /dev/null +++ b/packages/plugin-sdk/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "types": [], + "allowJs": true, + "checkJs": true + }, + "include": ["src", "fixtures"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fb4789ac81439..99238163d3cd3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -956,6 +956,9 @@ importers: '@t3tools/contracts': specifier: workspace:* version: link:../../packages/contracts + '@t3tools/plugin-sdk': + specifier: workspace:* + version: link:../../packages/plugin-sdk '@t3tools/provider-testing': specifier: workspace:* version: link:../../packages/provider-testing @@ -1590,6 +1593,8 @@ importers: specifier: 1.4.0 version: 1.4.0(@astrojs/compiler-rs@0.5.1)(@tsc-rs/typescript@0.8.0)(@types/node@24.12.4)(@vitest/browser-playwright@5.0.5)(@vitest/ui@5.0.5)(@volar/typescript@2.4.28(@tsc-rs/typescript@0.8.0))(bufferutil@4.1.0)(esbuild@0.28.2)(jiti@2.7.0)(jsdom@30.1.0(@noble/hashes@1.8.0))(playwright@1.60.0)(terser@5.48.0)(unrun@0.2.39)(utf-8-validate@6.0.6)(yaml@2.9.0) + packages/plugin-sdk: {} + packages/provider-acp: dependencies: '@t3tools/contracts':