feat: add plugin worker protocol, bootstrap and plugin_storage table

Worker-side ctx proxy plus the message envelope it talks over, and the
per-plugin key/value table behind ctx.storage with migrations for all
three dialects. Rows cascade with the plugin so uninstalling leaves
nothing behind.
This commit is contained in:
LukeGus committed 2026-09-21 07:19:01 -05:00
1 parent a1805cb7ae
commit 248698f62d
19 files changed
+30035

No files matched your search

+11
View File
@@ -0,0 +1,11 @@
CREATE TABLE `plugin_storage` (
`id` int AUTO_INCREMENT NOT NULL,
`plugin_id` varchar(255) NOT NULL,
`key` varchar(255) NOT NULL,
`value` text NOT NULL,
`updated_at` varchar(255) NOT NULL DEFAULT (CURRENT_TIMESTAMP),
CONSTRAINT `plugin_storage_id` PRIMARY KEY(`id`),
CONSTRAINT `idx_plugin_storage_plugin_key` UNIQUE(`plugin_id`,`key`)
);
--> statement-breakpoint
ALTER TABLE `plugin_storage` ADD CONSTRAINT `plugin_storage_plugin_id_plugins_id_fk` FOREIGN KEY (`plugin_id`) REFERENCES `plugins`(`id`) ON DELETE cascade ON UPDATE no action;
File diff suppressed because it is too large. Load diff
+7
View File
@@ -169,6 +169,13 @@
"when": 1789751455875,
"tag": "0023_even_beast",
"breakpoints": true
},
{
"idx": 24,
"version": "5",
"when": 1789991404402,
"tag": "0024_thin_colleen_wing",
"breakpoints": true
}
]
}
@@ -0,0 +1,10 @@
CREATE TABLE "plugin_storage" (
"id" serial PRIMARY KEY NOT NULL,
"plugin_id" varchar(255) NOT NULL,
"key" varchar(255) NOT NULL,
"value" text NOT NULL,
"updated_at" varchar(255) DEFAULT CURRENT_TIMESTAMP NOT NULL
);
--> statement-breakpoint
ALTER TABLE "plugin_storage" ADD CONSTRAINT "plugin_storage_plugin_id_plugins_id_fk" FOREIGN KEY ("plugin_id") REFERENCES "public"."plugins"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
CREATE UNIQUE INDEX "idx_plugin_storage_plugin_key" ON "plugin_storage" USING btree ("plugin_id","key");
File diff suppressed because it is too large. Load diff
+7
View File
@@ -176,6 +176,13 @@
"when": 1789751455415,
"tag": "0024_flimsy_forge",
"breakpoints": true
},
{
"idx": 25,
"version": "7",
"when": 1789991403847,
"tag": "0025_purple_squirrel_girl",
"breakpoints": true
}
]
}
+10
View File
@@ -0,0 +1,10 @@
CREATE TABLE `plugin_storage` (
`id` integer PRIMARY KEY AUTOINCREMENT NOT NULL,
`plugin_id` text NOT NULL,
`key` text NOT NULL,
`value` text NOT NULL,
`updated_at` text DEFAULT CURRENT_TIMESTAMP NOT NULL,
FOREIGN KEY (`plugin_id`) REFERENCES `plugins`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
CREATE UNIQUE INDEX `idx_plugin_storage_plugin_key` ON `plugin_storage` (`plugin_id`,`key`);
File diff suppressed because it is too large. Load diff
+7
View File
@@ -148,6 +148,13 @@
"when": 1789751454932,
"tag": "0020_moaning_fat_cobra",
"breakpoints": true
},
{
"idx": 21,
"version": "6",
"when": 1789991403283,
"tag": "0021_dashing_lester",
"breakpoints": true
}
]
}
+10
View File
@@ -728,6 +728,16 @@ async function initializeCompleteDatabase(): Promise<void> {
UNIQUE (plugin_id, registry_id)
);
CREATE TABLE IF NOT EXISTS plugin_storage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plugin_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (plugin_id, key),
FOREIGN KEY (plugin_id) REFERENCES plugins (id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS api_keys (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
+23
View File
@@ -2279,4 +2279,27 @@ export const pluginInstallCounts = mysqlTable(
],
);
/**
* Per-plugin key/value state, written through ctx.storage. Rows are always
* scoped to the calling plugin by the broker, so a plugin cannot name another
* plugin's scope. Cascades with the plugin so uninstalling leaves nothing.
*/
export const pluginStorage = mysqlTable(
"plugin_storage",
{
id: int("id").autoincrement().primaryKey(),
pluginId: varchar("plugin_id", { length: 255 })
.notNull()
.references(() => plugins.id, { onDelete: "cascade" }),
key: varchar("key", { length: 255 }).notNull(),
value: text("value").notNull(),
updatedAt: varchar("updated_at", { length: 255 })
.notNull()
.default(sql`(CURRENT_TIMESTAMP)`),
},
(table) => [
uniqueIndex("idx_plugin_storage_plugin_key").on(table.pluginId, table.key),
],
);
// --- plugins end ---
+23
View File
@@ -2280,4 +2280,27 @@ export const pluginInstallCounts = pgTable(
],
);
/**
* Per-plugin key/value state, written through ctx.storage. Rows are always
* scoped to the calling plugin by the broker, so a plugin cannot name another
* plugin's scope. Cascades with the plugin so uninstalling leaves nothing.
*/
export const pluginStorage = pgTable(
"plugin_storage",
{
id: serial("id").primaryKey(),
pluginId: varchar("plugin_id", { length: 255 })
.notNull()
.references(() => plugins.id, { onDelete: "cascade" }),
key: varchar("key", { length: 255 }).notNull(),
value: text("value").notNull(),
updatedAt: varchar("updated_at", { length: 255 })
.notNull()
.default(sql`CURRENT_TIMESTAMP`),
},
(table) => [
uniqueIndex("idx_plugin_storage_plugin_key").on(table.pluginId, table.key),
],
);
// --- plugins end ---
+23
View File
@@ -2278,4 +2278,27 @@ export const pluginInstallCounts = sqliteTable(
],
);
/**
* Per-plugin key/value state, written through ctx.storage. Rows are always
* scoped to the calling plugin by the broker, so a plugin cannot name another
* plugin's scope. Cascades with the plugin so uninstalling leaves nothing.
*/
export const pluginStorage = sqliteTable(
"plugin_storage",
{
id: integer("id").primaryKey({ autoIncrement: true }),
pluginId: text("plugin_id")
.notNull()
.references(() => plugins.id, { onDelete: "cascade" }),
key: text("key").notNull(),
value: text("value").notNull(),
updatedAt: text("updated_at")
.notNull()
.default(sql`CURRENT_TIMESTAMP`),
},
(table) => [
uniqueIndex("idx_plugin_storage_plugin_key").on(table.pluginId, table.key),
],
);
// --- plugins end ---
@@ -33,6 +33,7 @@ import { NetworkTopologyRepository } from "./network-topology-repository.js";
import { OpenTabRepository } from "./open-tab-repository.js";
import { OpksshTokenRepository } from "./opkssh-token-repository.js";
import { PluginRepository } from "./plugin-repository.js";
import { PluginStorageRepository } from "./plugin-storage-repository.js";
import { PluginPermissionGrantRepository } from "./plugin-permission-grant-repository.js";
import { PluginRegistryRepository } from "./plugin-registry-repository.js";
import { PluginInstallCountRepository } from "./plugin-install-count-repository.js";
@@ -385,6 +386,13 @@ export function createCurrentPluginPermissionGrantRepository(): PluginPermission
);
}
export function createCurrentPluginStorageRepository(): PluginStorageRepository {
return new PluginStorageRepository(
createCurrentRepositoryContext(),
createCurrentRepositoryWriteHook("plugin_storage_repository_write"),
);
}
export function createCurrentPluginRegistryRepository(): PluginRegistryRepository {
return new PluginRegistryRepository(
createCurrentRepositoryContext(),
@@ -0,0 +1,93 @@
import { and, eq } from "drizzle-orm";
import { pluginStorage } from "../db/schema.js";
import type { DatabaseContext } from "./database-context.js";
import { rowsAffected } from "./mutation-result.js";
import { insertReturning, updateReturning } from "./returning.js";
export type PluginStorageRecord = typeof pluginStorage.$inferSelect;
export class PluginStorageRepository {
constructor(
private readonly context: DatabaseContext,
private readonly onWrite?: () => void | Promise<void>,
) {}
async get(pluginId: string, key: string): Promise<string | null> {
const rows = await this.context.drizzle
.select()
.from(pluginStorage)
.where(
and(eq(pluginStorage.pluginId, pluginId), eq(pluginStorage.key, key)),
)
.limit(1);
return rows[0]?.value ?? null;
}
async listKeys(pluginId: string): Promise<string[]> {
const rows = await this.context.drizzle
.select()
.from(pluginStorage)
.where(eq(pluginStorage.pluginId, pluginId));
return rows.map((row) => row.key);
}
async set(
pluginId: string,
key: string,
value: string,
now = new Date().toISOString(),
): Promise<void> {
const existing = await this.context.drizzle
.select()
.from(pluginStorage)
.where(
and(eq(pluginStorage.pluginId, pluginId), eq(pluginStorage.key, key)),
)
.limit(1);
if (existing[0]) {
await updateReturning(
this.context,
pluginStorage,
{ value, updatedAt: now },
and(eq(pluginStorage.pluginId, pluginId), eq(pluginStorage.key, key))!,
);
} else {
await insertReturning(this.context, pluginStorage, {
pluginId,
key,
value,
updatedAt: now,
});
}
await this.afterWrite();
}
async delete(pluginId: string, key: string): Promise<boolean> {
const result = await this.context.drizzle
.delete(pluginStorage)
.where(
and(eq(pluginStorage.pluginId, pluginId), eq(pluginStorage.key, key)),
);
const affected = rowsAffected(result) > 0;
if (affected) await this.afterWrite();
return affected;
}
async deleteByPlugin(pluginId: string): Promise<number> {
const result = await this.context.drizzle
.delete(pluginStorage)
.where(eq(pluginStorage.pluginId, pluginId));
if (rowsAffected(result) > 0) await this.afterWrite();
return rowsAffected(result);
}
private async afterWrite(): Promise<void> {
await this.onWrite?.();
}
}
+22
View File
@@ -0,0 +1,22 @@
import path from "node:path";
/**
* Where unpacked plugins live. Resolved per call rather than captured at module
* scope: tests set DATA_DIR in beforeEach, and a module-scope constant would
* freeze whatever the first import saw.
*/
export function getPluginsDir(): string {
return path.join(process.env.DATA_DIR || "./db/data", "plugins");
}
export function getPluginDir(pluginId: string): string {
return path.join(getPluginsDir(), pluginId);
}
export function getPluginBackendEntry(pluginDir: string): string {
return path.join(pluginDir, "backend", "index.mjs");
}
export function getPluginManifestPath(pluginDir: string): string {
return path.join(pluginDir, "manifest.json");
}
+64
View File
@@ -0,0 +1,64 @@
/**
* Wire format between a plugin worker and the main-thread broker.
*
* Everything crossing this boundary must be structured-cloneable. That is not
* just a serialisation detail: it is the reason a plugin cannot be handed a db
* handle, an ssh2 Client or a live socket by accident. Live objects stay in the
* main thread and the worker refers to them by opaque handle strings.
*/
export interface PluginRequest {
/** Correlates a reply to its request. Worker-generated, monotonic. */
id: number;
/** Dotted ctx path, e.g. "hosts.get" or "storage.set". */
method: string;
args: unknown[];
}
export interface PluginResponseOk {
id: number;
ok: true;
value: unknown;
}
export interface PluginResponseErr {
id: number;
ok: false;
error: { message: string; code?: string };
}
export type PluginResponse = PluginResponseOk | PluginResponseErr;
/** Main thread -> worker, pushed rather than requested. */
export interface PluginPush {
kind: "event" | "http" | "timer";
/** Event topic, route key, or timer id depending on kind. */
key: string;
payload: unknown;
/** Set when the push expects a reply (an HTTP route invocation). */
replyTo?: number;
}
/** First message the worker receives, before any plugin code runs. */
export interface PluginBootstrapData {
pluginId: string;
pluginDir: string;
entryPath: string;
manifest: unknown;
}
export function isPluginRequest(value: unknown): value is PluginRequest {
if (!value || typeof value !== "object") return false;
const v = value as Record<string, unknown>;
return (
typeof v.id === "number" &&
typeof v.method === "string" &&
Array.isArray(v.args)
);
}
export function isPluginResponse(value: unknown): value is PluginResponse {
if (!value || typeof value !== "object") return false;
const v = value as Record<string, unknown>;
return typeof v.id === "number" && typeof v.ok === "boolean";
}
+243
View File
@@ -0,0 +1,243 @@
/**
* The worker thread's entry point. Runs before any plugin code.
*
* Builds the ctx object, imports the plugin's backend/index.mjs, and calls its
* exported activate(ctx). The plugin gets ctx and nothing else from us -- no
* db, no fs, no net handle is ever passed in. See loader.ts for an honest
* account of what that does and does not guarantee.
*/
import { parentPort, workerData } from "node:worker_threads";
import type {
PluginBootstrapData,
PluginPush,
PluginResponse,
} from "./protocol.js";
import { isPluginResponse } from "./protocol.js";
const port = parentPort;
if (!port) {
throw new Error("Plugin worker started without a parent port");
}
const boot = workerData as PluginBootstrapData;
let nextRequestId = 1;
const pending = new Map<
number,
{ resolve: (value: unknown) => void; reject: (error: Error) => void }
>();
/** Handlers the plugin registered, keyed the same way the broker pushes them. */
const httpRoutes = new Map<string, (req: unknown) => unknown>();
const eventListeners = new Map<string, Set<(payload: unknown) => void>>();
const timers = new Map<string, () => void>();
function call(method: string, ...args: unknown[]): Promise<unknown> {
const id = nextRequestId++;
return new Promise((resolve, reject) => {
pending.set(id, { resolve, reject });
port!.postMessage({ id, method, args });
});
}
function handleResponse(message: PluginResponse): void {
const waiter = pending.get(message.id);
if (!waiter) return;
pending.delete(message.id);
if (message.ok === true) {
waiter.resolve(message.value);
return;
}
const failure = message.error;
const error = new Error(failure.message);
if (failure.code) {
(error as Error & { code?: string }).code = failure.code;
}
waiter.reject(error);
}
async function handlePush(push: PluginPush): Promise<void> {
if (push.kind === "event") {
const listeners = eventListeners.get(push.key);
if (!listeners) return;
for (const listener of listeners) {
try {
listener(push.payload);
} catch {
// A throwing listener must not take down the worker.
}
}
return;
}
if (push.kind === "timer") {
const handler = timers.get(push.key);
if (handler) {
try {
handler();
} catch {
// same
}
}
return;
}
if (push.kind === "http") {
const handler = httpRoutes.get(push.key);
if (push.replyTo === undefined) return;
if (!handler) {
port!.postMessage({
id: push.replyTo,
ok: false,
error: { message: `No handler registered for route ${push.key}` },
});
return;
}
try {
const value = await handler(push.payload);
port!.postMessage({ id: push.replyTo, ok: true, value });
} catch (error) {
port!.postMessage({
id: push.replyTo,
ok: false,
error: {
message: error instanceof Error ? error.message : String(error),
},
});
}
}
}
port.on("message", (message: unknown) => {
if (isPluginResponse(message)) {
handleResponse(message);
return;
}
const push = message as PluginPush;
if (push && typeof push.kind === "string") {
void handlePush(push);
}
});
function notImplemented(name: string) {
return () =>
Promise.reject(
new Error(
`ctx.${name} is not implemented in plugin SDK version 1. It is reserved and will be added in a later version.`,
),
);
}
const ctx = {
pluginId: boot.pluginId,
manifest: boot.manifest,
log: {
debug: (message: string, context?: unknown) =>
call("log.debug", message, context),
info: (message: string, context?: unknown) =>
call("log.info", message, context),
warn: (message: string, context?: unknown) =>
call("log.warn", message, context),
error: (message: string, context?: unknown) =>
call("log.error", message, context),
},
audit: (action: string, details?: unknown) => call("audit", action, details),
hosts: {
list: () => call("hosts.list"),
get: (hostId: number) => call("hosts.get", hostId),
},
ssh: {
connect: (hostId: number) => call("ssh.connect", hostId),
exec: (handle: string, command: string, timeoutMs?: number) =>
call("ssh.exec", handle, command, timeoutMs),
close: (handle: string) => call("ssh.close", handle),
},
storage: {
get: (key: string) => call("storage.get", key),
set: (key: string, value: unknown) => call("storage.set", key, value),
delete: (key: string) => call("storage.delete", key),
list: () => call("storage.list"),
},
http: {
route: (
method: string,
routePath: string,
handler: (req: unknown) => unknown,
) => {
const key = `${method.toUpperCase()} ${routePath}`;
httpRoutes.set(key, handler);
return call("http.route", method.toUpperCase(), routePath);
},
},
events: {
emit: (topic: string, payload: unknown) =>
call("events.emit", topic, payload),
on: async (topic: string, listener: (payload: unknown) => void) => {
let listeners = eventListeners.get(topic);
if (!listeners) {
listeners = new Set();
eventListeners.set(topic, listeners);
await call("events.subscribe", topic);
}
listeners.add(listener);
return () => {
listeners!.delete(listener);
};
},
},
schedule: {
every: async (intervalMs: number, handler: () => void) => {
const id = (await call("schedule.every", intervalMs)) as string;
timers.set(id, handler);
return id;
},
cancel: async (id: string) => {
timers.delete(id);
await call("schedule.cancel", id);
},
},
secrets: {
get: notImplemented("secrets.get"),
set: notImplemented("secrets.set"),
},
notify: notImplemented("notify"),
fetch: notImplemented("fetch"),
};
async function start(): Promise<void> {
const module = await import(boot.entryPath);
const activate = module.activate ?? module.default?.activate;
if (typeof activate !== "function") {
throw new Error(
`Plugin ${boot.pluginId} backend entry does not export an activate(ctx) function`,
);
}
await activate(ctx);
port!.postMessage({ id: 0, ok: true, value: "activated" });
}
start().catch((error) => {
port!.postMessage({
id: 0,
ok: false,
error: {
message: error instanceof Error ? error.message : String(error),
},
});
});
@@ -0,0 +1,96 @@
import { afterEach, describe, expect, it } from "vitest";
import { TestSqliteDatabase } from "./test-support.js";
import { PluginStorageRepository } from "../../../database/repositories/plugin-storage-repository.js";
describe("PluginStorageRepository", () => {
let adapter: TestSqliteDatabase | null = null;
afterEach(async () => {
if (adapter) {
await adapter.close();
adapter = null;
}
});
async function createRepository(
onWrite?: () => void | Promise<void>,
): Promise<PluginStorageRepository> {
adapter = new TestSqliteDatabase();
const context = await adapter.connect();
await adapter.exec(`
INSERT INTO plugins (id, name, version, manifest_json)
VALUES ('plugin-a', 'Plugin A', '1.0.0', '{}');
INSERT INTO plugins (id, name, version, manifest_json)
VALUES ('plugin-b', 'Plugin B', '1.0.0', '{}');
`);
return new PluginStorageRepository(context, onWrite);
}
it("round-trips a value", async () => {
const repo = await createRepository();
expect(await repo.get("plugin-a", "missing")).toBeNull();
await repo.set("plugin-a", "token", '{"v":1}');
expect(await repo.get("plugin-a", "token")).toBe('{"v":1}');
});
it("overwrites rather than duplicating an existing key", async () => {
const repo = await createRepository();
await repo.set("plugin-a", "counter", "1");
await repo.set("plugin-a", "counter", "2");
expect(await repo.get("plugin-a", "counter")).toBe("2");
expect(await repo.listKeys("plugin-a")).toEqual(["counter"]);
});
it("keeps each plugin's keys separate", async () => {
const repo = await createRepository();
await repo.set("plugin-a", "shared-key", "from-a");
await repo.set("plugin-b", "shared-key", "from-b");
expect(await repo.get("plugin-a", "shared-key")).toBe("from-a");
expect(await repo.get("plugin-b", "shared-key")).toBe("from-b");
});
it("deletes a key and reports whether a row changed", async () => {
let writes = 0;
const repo = await createRepository(() => {
writes += 1;
});
await repo.set("plugin-a", "k", "v");
expect(writes).toBe(1);
expect(await repo.delete("plugin-a", "nope")).toBe(false);
expect(writes).toBe(1);
expect(await repo.delete("plugin-a", "k")).toBe(true);
expect(writes).toBe(2);
expect(await repo.get("plugin-a", "k")).toBeNull();
});
it("deletes everything belonging to one plugin", async () => {
const repo = await createRepository();
await repo.set("plugin-a", "one", "1");
await repo.set("plugin-a", "two", "2");
await repo.set("plugin-b", "keep", "3");
expect(await repo.deleteByPlugin("plugin-a")).toBe(2);
expect(await repo.listKeys("plugin-a")).toEqual([]);
expect(await repo.listKeys("plugin-b")).toEqual(["keep"]);
});
it("drops rows when the owning plugin is removed", async () => {
const repo = await createRepository();
await repo.set("plugin-a", "k", "v");
await adapter!.exec(`DELETE FROM plugins WHERE id = 'plugin-a';`);
expect(await repo.listKeys("plugin-a")).toEqual([]);
});
});