mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-20 22:16:26 +00:00
feat(plugins): isolated per-plugin runtime + capability RPC (M1)
Every plugin now runs in its own forked child process with a scrubbed env (no JWT_SECRET, no db path, nothing inherited). It talks to TREK only over a JSON-RPC channel, and the host's capability router registers ONLY the methods a plugin's granted permissions unlock — so an ungranted call is unreachable, not merely refused. The plugin's own data lives in a separate sqlite file it can never open directly; core reads (trips/users) go through membership-checked, column-projected host methods; ws broadcasts are force-namespaced. - protocol/envelope: the wire types + method→permission map (pure, shared by host and the isolated child) - host/rpc-host: the capability router = the enforcement point (dispatch, BAD_PARAMS / PERMISSION_DENIED / RESOURCE_FORBIDDEN / UNKNOWN_METHOD) - host/plugin-data: the per-plugin sqlite file (db:own), guarded against ATTACH/PRAGMA escape, idempotent migrations - host/create-rpc-host: wires the router to the real db/websocket (host-only) - runtime/plugin-sdk + plugin-host-entry: the child bootstrap + definePlugin ctx; turns each ctx call into an RPC, never imports a privileged module - supervisor: spawn on activate, heartbeat/reap, crash backoff + auto-disable, graceful shutdown — a plugin crash/OOM/hang can never reach the Nest loop - paths: code/data layout, dist-vs-tsx child entry resolution Nothing is wired into activation yet (that's the next slice); exercised by unit tests for the router/sdk/data and an integration test that forks a real child.
This commit is contained in:
mauriceboe
@@ -0,0 +1,37 @@
|
||||
import { db, canAccessTrip } from '../../../db/database';
|
||||
import { broadcast, broadcastToUser } from '../../../websocket';
|
||||
import { PluginDataDb } from './plugin-data.service';
|
||||
import { PluginRpcHost } from './rpc-host';
|
||||
|
||||
/**
|
||||
* Wires a plugin's capability host to the REAL privileged modules (#plugins,
|
||||
* M1). This is the ONLY plugin file that imports db/websocket — it runs in the
|
||||
* host (parent), never in the child. Broadcasts are force-namespaced to
|
||||
* `plugin:{id}:{event}` so a plugin can't forge a core event.
|
||||
*/
|
||||
|
||||
const dataDbs = new Map<string, PluginDataDb>();
|
||||
|
||||
export function getPluginDataDb(id: string): PluginDataDb {
|
||||
let d = dataDbs.get(id);
|
||||
if (!d) {
|
||||
d = new PluginDataDb(id);
|
||||
dataDbs.set(id, d);
|
||||
}
|
||||
return d;
|
||||
}
|
||||
|
||||
export function closePluginDataDb(id: string): void {
|
||||
dataDbs.get(id)?.close();
|
||||
dataDbs.delete(id);
|
||||
}
|
||||
|
||||
export function createRealRpcHost(id: string, granted: ReadonlySet<string>): PluginRpcHost {
|
||||
return new PluginRpcHost(id, granted, {
|
||||
data: getPluginDataDb(id),
|
||||
db,
|
||||
canAccessTrip: (tripId, userId) => canAccessTrip(tripId, userId),
|
||||
broadcastToTrip: (tripId, event, payload) => broadcast(tripId, `plugin:${id}:${event}`, payload),
|
||||
broadcastToUser: (userId, payload) => broadcastToUser(userId, { type: `plugin:${id}`, ...payload }),
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
import fs from 'node:fs';
|
||||
import Database from 'better-sqlite3';
|
||||
import type { Database as Db } from 'better-sqlite3';
|
||||
import { pluginDataDir, pluginDbFile } from '../paths';
|
||||
|
||||
/**
|
||||
* A plugin's own sqlite database (#plugins, db:own). The HOST owns the handle;
|
||||
* the plugin child never gets a path or a connection — it can only reach this
|
||||
* through RPC (db.exec / db.query / db.migrate). Because it is a SEPARATE FILE,
|
||||
* containment is a filesystem fact: the plugin physically cannot read trek.db,
|
||||
* and we don't have to police table-name prefixes in its SQL.
|
||||
*
|
||||
* A thin guard still rejects statements that would let a plugin escape its file
|
||||
* (ATTACH another db, VACUUM INTO elsewhere, PRAGMA fiddling) or DoS via
|
||||
* oversize SQL.
|
||||
*/
|
||||
|
||||
const MAX_SQL_LENGTH = 100_000;
|
||||
const FORBIDDEN = /\b(ATTACH|DETACH|VACUUM|PRAGMA)\b/i;
|
||||
|
||||
export class PluginDataDb {
|
||||
private db: Db;
|
||||
|
||||
constructor(pluginId: string) {
|
||||
fs.mkdirSync(pluginDataDir(pluginId), { recursive: true });
|
||||
this.db = new Database(pluginDbFile(pluginId));
|
||||
this.db.pragma('journal_mode = WAL');
|
||||
this.db.pragma('foreign_keys = ON');
|
||||
// Track applied migrations so db.migrate is idempotent per (plugin, id).
|
||||
this.db.exec(
|
||||
`CREATE TABLE IF NOT EXISTS _plugin_migrations (id TEXT PRIMARY KEY, applied_at INTEGER)`,
|
||||
);
|
||||
}
|
||||
|
||||
private guard(sql: string): void {
|
||||
if (typeof sql !== 'string') throw new Error('sql must be a string');
|
||||
if (sql.length > MAX_SQL_LENGTH) throw new Error('sql too long');
|
||||
if (FORBIDDEN.test(sql)) throw new Error('statement type not allowed for plugin databases');
|
||||
}
|
||||
|
||||
/** Read query — returns all rows. Single statement only (better-sqlite3 prepare). */
|
||||
query(sql: string, args: unknown[] = []): unknown[] {
|
||||
this.guard(sql);
|
||||
return this.db.prepare(sql).all(...(args as never[]));
|
||||
}
|
||||
|
||||
/** Write statement(s). exec() allows multiple statements (e.g. a small setup script). */
|
||||
exec(sql: string, args: unknown[] = []): { changes: number } {
|
||||
this.guard(sql);
|
||||
if (args.length > 0) {
|
||||
const info = this.db.prepare(sql).run(...(args as never[]));
|
||||
return { changes: info.changes };
|
||||
}
|
||||
this.db.exec(sql);
|
||||
return { changes: 0 };
|
||||
}
|
||||
|
||||
/** Run a migration once, keyed by id. Re-running with the same id is a no-op. */
|
||||
migrate(id: string, sql: string): { applied: boolean } {
|
||||
this.guard(sql);
|
||||
const seen = this.db.prepare('SELECT 1 FROM _plugin_migrations WHERE id = ?').get(id);
|
||||
if (seen) return { applied: false };
|
||||
this.db.transaction(() => {
|
||||
this.db.exec(sql);
|
||||
this.db.prepare('INSERT INTO _plugin_migrations (id, applied_at) VALUES (?, ?)').run(id, Date.now());
|
||||
})();
|
||||
return { applied: true };
|
||||
}
|
||||
|
||||
close(): void {
|
||||
try {
|
||||
this.db.close();
|
||||
} catch {
|
||||
/* already closed */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Delete a plugin's data directory (uninstall "delete data"). */
|
||||
export function removePluginData(pluginId: string): void {
|
||||
fs.rmSync(pluginDataDir(pluginId), { recursive: true, force: true });
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
import {
|
||||
KNOWN_METHODS,
|
||||
type KnownMethod,
|
||||
type RpcError,
|
||||
type RpcRequest,
|
||||
type RpcResponse,
|
||||
} from '../protocol/envelope';
|
||||
import type { PluginDataDb } from './plugin-data.service';
|
||||
|
||||
/**
|
||||
* The per-plugin capability router (#plugins, M1) — the ENFORCEMENT POINT.
|
||||
*
|
||||
* Built from the plugin's GRANTED permission set. Only the methods a permission
|
||||
* unlocks are registered; an ungranted method is simply never in the map, so the
|
||||
* plugin cannot "call it anyway" — there is no shared object, only messages, and
|
||||
* the host is the sole holder of the trek.db handle and the broadcast fns.
|
||||
*
|
||||
* Runs in the HOST (parent) process.
|
||||
*/
|
||||
|
||||
/** Thrown by a handler when the acting user may not touch the requested resource. */
|
||||
export class ForbiddenResource extends Error {}
|
||||
|
||||
interface CoreDb {
|
||||
prepare(sql: string): {
|
||||
all(...args: unknown[]): unknown[];
|
||||
get(...args: unknown[]): unknown;
|
||||
};
|
||||
}
|
||||
|
||||
export interface HostDeps {
|
||||
/** The plugin's own sqlite (db:own). */
|
||||
data: PluginDataDb;
|
||||
/** Read-only handle to the core trek.db, used ONLY through the typed readers here. */
|
||||
db: CoreDb;
|
||||
/** Returns the trip row if the user may access it, else undefined. */
|
||||
canAccessTrip(tripId: number, userId: number): unknown;
|
||||
/** Namespaced trip broadcast (host forces the plugin:{id}:{event} event type). */
|
||||
broadcastToTrip(tripId: number, eventType: string, payload: Record<string, unknown>): void;
|
||||
/** Namespaced per-user broadcast. */
|
||||
broadcastToUser(userId: number, payload: Record<string, unknown>): void;
|
||||
}
|
||||
|
||||
type Handler = (params: Record<string, unknown>) => unknown;
|
||||
|
||||
const num = (v: unknown, name: string): number => {
|
||||
const n = typeof v === 'string' ? Number(v) : v;
|
||||
if (typeof n !== 'number' || !Number.isFinite(n)) throw new BadParams(`${name} must be a number`);
|
||||
return n;
|
||||
};
|
||||
const str = (v: unknown, name: string): string => {
|
||||
if (typeof v !== 'string') throw new BadParams(`${name} must be a string`);
|
||||
return v;
|
||||
};
|
||||
class BadParams extends Error {}
|
||||
|
||||
export class PluginRpcHost {
|
||||
private methods = new Map<string, Handler>();
|
||||
|
||||
constructor(
|
||||
private readonly pluginId: string,
|
||||
granted: ReadonlySet<string>,
|
||||
private readonly deps: HostDeps,
|
||||
) {
|
||||
const has = (p: string) => granted.has(p);
|
||||
|
||||
if (has('db:own')) {
|
||||
this.methods.set('db.query', (p) => deps.data.query(str(p.sql, 'sql'), asArgs(p.args)));
|
||||
this.methods.set('db.exec', (p) => deps.data.exec(str(p.sql, 'sql'), asArgs(p.args)));
|
||||
this.methods.set('db.migrate', (p) => deps.data.migrate(str(p.id, 'id'), str(p.sql, 'sql')));
|
||||
}
|
||||
|
||||
if (has('db:read:trips')) {
|
||||
this.methods.set('trips.getById', (p) => this.tripRead(p, () => deps.canAccessTrip(num(p.tripId, 'tripId'), num(p.asUserId, 'asUserId'))));
|
||||
this.methods.set('trips.getPlaces', (p) =>
|
||||
this.tripRead(p, () => deps.db.prepare('SELECT * FROM places WHERE trip_id = ? ORDER BY day_id, position').all(num(p.tripId, 'tripId'))),
|
||||
);
|
||||
this.methods.set('trips.getReservations', (p) =>
|
||||
this.tripRead(p, () => deps.db.prepare('SELECT * FROM reservations WHERE trip_id = ? ORDER BY reservation_time').all(num(p.tripId, 'tripId'))),
|
||||
);
|
||||
}
|
||||
|
||||
if (has('db:read:users')) {
|
||||
this.methods.set('users.getById', (p) =>
|
||||
deps.db
|
||||
.prepare('SELECT id, username, display_name, avatar FROM users WHERE id = ?')
|
||||
.get(num(p.id, 'id')),
|
||||
);
|
||||
}
|
||||
|
||||
if (has('ws:broadcast:trip')) {
|
||||
this.methods.set('ws.broadcastToTrip', (p) => {
|
||||
deps.broadcastToTrip(num(p.tripId, 'tripId'), str(p.event, 'event'), asPayload(p.data));
|
||||
return { ok: true };
|
||||
});
|
||||
}
|
||||
if (has('ws:broadcast:user')) {
|
||||
this.methods.set('ws.broadcastToUser', (p) => {
|
||||
deps.broadcastToUser(num(p.userId, 'userId'), { event: str(p.event, 'event'), ...asPayload(p.data) });
|
||||
return { ok: true };
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** Membership-check every trip read against the acting user; forbid otherwise. */
|
||||
private tripRead(p: Record<string, unknown>, read: () => unknown): unknown {
|
||||
const tripId = num(p.tripId, 'tripId');
|
||||
const asUserId = num(p.asUserId, 'asUserId');
|
||||
if (!this.deps.canAccessTrip(tripId, asUserId)) {
|
||||
throw new ForbiddenResource(`no access to trip ${tripId}`);
|
||||
}
|
||||
return read();
|
||||
}
|
||||
|
||||
async dispatch(req: RpcRequest): Promise<RpcResponse | RpcError> {
|
||||
const handler = this.methods.get(req.method);
|
||||
if (!handler) {
|
||||
const known = (KNOWN_METHODS as readonly string[]).includes(req.method as KnownMethod);
|
||||
return this.err(
|
||||
req.id,
|
||||
known ? 'PERMISSION_DENIED' : 'UNKNOWN_METHOD',
|
||||
known
|
||||
? `${req.method} requires a permission "${this.pluginId}" was not granted`
|
||||
: `unknown method ${req.method}`,
|
||||
);
|
||||
}
|
||||
const params = (req.params ?? {}) as Record<string, unknown>;
|
||||
try {
|
||||
const result = await handler(params);
|
||||
return { k: 'res', id: req.id, ok: true, result };
|
||||
} catch (e) {
|
||||
if (e instanceof BadParams) return this.err(req.id, 'BAD_PARAMS', e.message);
|
||||
if (e instanceof ForbiddenResource) return this.err(req.id, 'RESOURCE_FORBIDDEN', e.message);
|
||||
return this.err(req.id, 'HOST_ERROR', e instanceof Error ? e.message : 'internal error');
|
||||
}
|
||||
}
|
||||
|
||||
private err(id: string, code: RpcError['error']['code'], message: string): RpcError {
|
||||
return { k: 'res', id, ok: false, error: { code, message } };
|
||||
}
|
||||
|
||||
/** Release host-held resources (the plugin's own db handle) on terminal stop. */
|
||||
dispose(): void {
|
||||
try {
|
||||
this.deps.data.close();
|
||||
} catch {
|
||||
/* already closed */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function asArgs(v: unknown): unknown[] {
|
||||
if (v == null) return [];
|
||||
if (Array.isArray(v)) return v;
|
||||
throw new BadParams('args must be an array');
|
||||
}
|
||||
function asPayload(v: unknown): Record<string, unknown> {
|
||||
return v && typeof v === 'object' ? (v as Record<string, unknown>) : { value: v };
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
|
||||
/**
|
||||
* Filesystem layout for the plugin system (#plugins). Code and data are two
|
||||
* separate trees so the child can be given a read-only view of its code and a
|
||||
* read-write view of only its own data (M1 sets the env; the container runtime
|
||||
* in v2 enforces the mounts).
|
||||
*
|
||||
* Defaults sit under the persisted data dir (server/data), overridable by env
|
||||
* so a Docker deployment can point them at dedicated volumes.
|
||||
*/
|
||||
|
||||
const DATA_ROOT = path.resolve(__dirname, '../../../data');
|
||||
|
||||
// Read lazily so a deployment (or a test) can point these at dedicated volumes
|
||||
// via env without import-order surprises.
|
||||
export function pluginsCodeRoot(): string {
|
||||
return process.env.TREK_PLUGINS_DIR || path.join(DATA_ROOT, 'plugins');
|
||||
}
|
||||
export function pluginsDataRoot(): string {
|
||||
return process.env.TREK_PLUGINS_DATA_DIR || path.join(DATA_ROOT, 'plugins-data');
|
||||
}
|
||||
|
||||
/** A plugin's installed code directory (contains trek-plugin.json + server/index.js). */
|
||||
export function pluginCodeDir(id: string): string {
|
||||
return path.join(pluginsCodeRoot(), id);
|
||||
}
|
||||
|
||||
/** A plugin's writable data directory (its own sqlite file + any blobs). */
|
||||
export function pluginDataDir(id: string): string {
|
||||
return path.join(pluginsDataRoot(), id);
|
||||
}
|
||||
|
||||
/** The plugin's own sqlite file — opened by the HOST, reached by the plugin only via RPC. */
|
||||
export function pluginDbFile(id: string): string {
|
||||
return path.join(pluginDataDir(id), 'plugin.db');
|
||||
}
|
||||
|
||||
/**
|
||||
* The child bootstrap entry + the execArgv to fork it with.
|
||||
*
|
||||
* Prod/dev run the tsc output, so the sibling `runtime/plugin-host-entry.js`
|
||||
* exists and forks as plain node. Under vitest the code runs from `src` as TS,
|
||||
* so no `.js` sibling exists — fall back to the `.ts` source loaded via tsx (a
|
||||
* prod dependency). This keeps the fork path identical in tests and production.
|
||||
*/
|
||||
export function resolveChildEntry(): { entry: string; execArgv: string[]; forkCwd?: string } {
|
||||
const js = path.join(__dirname, 'runtime', 'plugin-host-entry.js');
|
||||
const ts = path.join(__dirname, 'runtime', 'plugin-host-entry.ts');
|
||||
if (!fs.existsSync(js) && fs.existsSync(ts)) {
|
||||
// tsx (dev/test only) is resolved relative to the child's cwd, so the child
|
||||
// must run from a dir where `tsx` is on the node_modules chain (the server
|
||||
// root) — NOT the plugin dir. The plugin itself is still loaded by absolute
|
||||
// path via createRequire, so this doesn't weaken the prod isolation, where
|
||||
// the .js branch below keeps cwd at the plugin dir.
|
||||
return { entry: ts, execArgv: ['--import', 'tsx'], forkCwd: process.cwd() };
|
||||
}
|
||||
return { entry: js, execArgv: ['--max-old-space-size=192'] };
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
/**
|
||||
* The host <-> plugin JSON-RPC wire protocol (#plugins, M1).
|
||||
*
|
||||
* PURE TYPES ONLY — this file must never import anything with runtime side
|
||||
* effects. It is loaded by BOTH the privileged host (parent process) and the
|
||||
* isolated plugin child, and the child must not transitively pull in db, config
|
||||
* or the websocket server. Keep it dependency-free.
|
||||
*/
|
||||
|
||||
export type PluginErrCode =
|
||||
| 'PERMISSION_DENIED' // a real method the plugin was not granted
|
||||
| 'UNKNOWN_METHOD' // not a method the host exposes at all
|
||||
| 'BAD_PARAMS' // params failed validation
|
||||
| 'RESOURCE_FORBIDDEN' // granted, but the acting user can't touch this resource
|
||||
| 'TIMEOUT'
|
||||
| 'PLUGIN_ERROR'
|
||||
| 'HOST_ERROR';
|
||||
|
||||
export interface RpcRequest {
|
||||
k: 'req';
|
||||
id: string;
|
||||
method: string;
|
||||
params: unknown;
|
||||
}
|
||||
export interface RpcResponse {
|
||||
k: 'res';
|
||||
id: string;
|
||||
ok: true;
|
||||
result: unknown;
|
||||
}
|
||||
export interface RpcError {
|
||||
k: 'res';
|
||||
id: string;
|
||||
ok: false;
|
||||
error: { code: PluginErrCode; message: string };
|
||||
}
|
||||
export interface RpcEvent {
|
||||
k: 'evt';
|
||||
topic: string;
|
||||
data: unknown;
|
||||
}
|
||||
export type Envelope = RpcRequest | RpcResponse | RpcError | RpcEvent;
|
||||
|
||||
/**
|
||||
* Every method the host CAN expose. The capability router registers only the
|
||||
* subset a plugin was granted; anything here but ungranted resolves to
|
||||
* PERMISSION_DENIED, anything not here at all resolves to UNKNOWN_METHOD.
|
||||
*/
|
||||
export const KNOWN_METHODS = [
|
||||
'db.exec',
|
||||
'db.query',
|
||||
'db.migrate',
|
||||
'trips.getById',
|
||||
'trips.getPlaces',
|
||||
'trips.getReservations',
|
||||
'users.getById',
|
||||
'ws.broadcastToTrip',
|
||||
'ws.broadcastToUser',
|
||||
] as const;
|
||||
export type KnownMethod = (typeof KNOWN_METHODS)[number];
|
||||
|
||||
/** Which permission unlocks which method(s). The single source for the router. */
|
||||
export const METHOD_PERMISSION: Record<KnownMethod, string> = {
|
||||
'db.exec': 'db:own',
|
||||
'db.query': 'db:own',
|
||||
'db.migrate': 'db:own',
|
||||
'trips.getById': 'db:read:trips',
|
||||
'trips.getPlaces': 'db:read:trips',
|
||||
'trips.getReservations': 'db:read:trips',
|
||||
'users.getById': 'db:read:users',
|
||||
'ws.broadcastToTrip': 'ws:broadcast:trip',
|
||||
'ws.broadcastToUser': 'ws:broadcast:user',
|
||||
};
|
||||
|
||||
/** All permission strings the host understands (unknown ones are rejected at activation). */
|
||||
export const KNOWN_PERMISSIONS = [
|
||||
'db:own',
|
||||
'db:read:trips',
|
||||
'db:read:users',
|
||||
'ws:broadcast:trip',
|
||||
'ws:broadcast:user',
|
||||
'hook:photo-provider',
|
||||
'hook:calendar-source',
|
||||
'http:outbound',
|
||||
] as const;
|
||||
|
||||
export function isKnownPermission(p: string): boolean {
|
||||
return (KNOWN_PERMISSIONS as readonly string[]).includes(p) || p.startsWith('http:outbound:');
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
/**
|
||||
* The isolated plugin child bootstrap (#plugins, M1).
|
||||
*
|
||||
* Runs as a forked node process (`dist/nest/plugins/runtime/plugin-host-entry.js`)
|
||||
* with a scrubbed env — NO JWT_SECRET, NO db path, NO inherited process.env. It
|
||||
* loads the plugin's own code and turns every ctx call into an RPC message to
|
||||
* the parent, which is the only side holding real capabilities.
|
||||
*
|
||||
* MUST NOT import any privileged server module (db, config, websocket). Its only
|
||||
* imports are the pure protocol + SDK.
|
||||
*/
|
||||
|
||||
import path from 'node:path';
|
||||
import { createRequire } from 'node:module';
|
||||
import { createPluginContext, type ChildTransport, type PluginContext, type PluginDefinition } from './plugin-sdk';
|
||||
import type { Envelope, RpcError } from '../protocol/envelope';
|
||||
|
||||
const pluginId = process.argv[2] || process.env.TREK_PLUGIN_ID || 'unknown';
|
||||
const pluginDir = process.argv[3] || '';
|
||||
|
||||
const pending = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
|
||||
let seq = 0;
|
||||
|
||||
function send(msg: Envelope): void {
|
||||
process.send?.(msg);
|
||||
}
|
||||
|
||||
const transport: ChildTransport = {
|
||||
rpc(method, params) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const id = `${++seq}`;
|
||||
pending.set(id, { resolve, reject });
|
||||
send({ k: 'req', id, method, params });
|
||||
});
|
||||
},
|
||||
emit(topic, data) {
|
||||
send({ k: 'evt', topic, data });
|
||||
},
|
||||
};
|
||||
|
||||
let def: PluginDefinition | null = null;
|
||||
let ctx: PluginContext | null = null;
|
||||
|
||||
async function boot(config: Record<string, unknown>): Promise<void> {
|
||||
try {
|
||||
// createRequire works whether this bootstrap runs as CJS (prod dist) or ESM
|
||||
// (tsx in tests), so `require` being undefined in ESM never bites us.
|
||||
const entry = path.join(pluginDir, 'server', 'index.js');
|
||||
const requirePlugin = createRequire(entry);
|
||||
const mod = requirePlugin(entry);
|
||||
def = mod && mod.default ? (mod.default as PluginDefinition) : (mod as PluginDefinition);
|
||||
ctx = createPluginContext(pluginId, config, transport);
|
||||
if (typeof def.onLoad === 'function') await def.onLoad(ctx);
|
||||
send({ k: 'evt', topic: 'loaded', data: {} });
|
||||
// An immediate first heartbeat confirms liveness without waiting a full interval.
|
||||
send({ k: 'evt', topic: 'heartbeat', data: { rss: process.memoryUsage().rss } });
|
||||
} catch (e) {
|
||||
send({ k: 'evt', topic: 'load-error', data: { message: errMsg(e), stack: errStack(e) } });
|
||||
}
|
||||
}
|
||||
|
||||
async function shutdown(): Promise<void> {
|
||||
try {
|
||||
if (def && typeof def.onUnload === 'function' && ctx) await def.onUnload(ctx);
|
||||
} catch {
|
||||
/* best effort */
|
||||
}
|
||||
clearInterval(heartbeat);
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
process.on('message', (raw: unknown) => {
|
||||
const msg = raw as Envelope;
|
||||
if (!msg || typeof msg !== 'object') return;
|
||||
if (msg.k === 'res') {
|
||||
const p = pending.get(msg.id);
|
||||
if (!p) return;
|
||||
pending.delete(msg.id);
|
||||
if (msg.ok) {
|
||||
p.resolve(msg.result);
|
||||
} else {
|
||||
const em = (msg as RpcError).error;
|
||||
p.reject(new Error(`${em.code}: ${em.message}`));
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (msg.k === 'evt') {
|
||||
if (msg.topic === 'init') void boot(((msg.data as { config?: Record<string, unknown> })?.config) ?? {});
|
||||
else if (msg.topic === 'shutdown') void shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
// Ask the host for the init payload (instance config), then wait for it.
|
||||
send({ k: 'evt', topic: 'hello', data: {} });
|
||||
|
||||
// Liveness — unref so it never keeps the process alive on its own.
|
||||
const heartbeat = setInterval(() => {
|
||||
send({ k: 'evt', topic: 'heartbeat', data: { rss: process.memoryUsage().rss } });
|
||||
}, 5000);
|
||||
heartbeat.unref?.();
|
||||
|
||||
// A plugin that throws asynchronously must not take the host down — it only
|
||||
// crashes THIS child, which the supervisor detects and restarts/disables.
|
||||
process.on('uncaughtException', (e) => {
|
||||
send({ k: 'evt', topic: 'load-error', data: { message: errMsg(e), stack: errStack(e) } });
|
||||
process.exit(1);
|
||||
});
|
||||
process.on('unhandledRejection', (e) => {
|
||||
send({ k: 'evt', topic: 'load-error', data: { message: errMsg(e), stack: errStack(e) } });
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
function errMsg(e: unknown): string {
|
||||
return e instanceof Error ? e.message : String(e);
|
||||
}
|
||||
function errStack(e: unknown): string | undefined {
|
||||
return e instanceof Error ? e.stack : undefined;
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
/**
|
||||
* The plugin-author-facing SDK surface (#plugins, M1) — the minimal in-repo
|
||||
* version. The published `@trek/plugin-sdk` (M6) will re-export these types; for
|
||||
* now the runtime ships its own copy so the child has zero external deps.
|
||||
*
|
||||
* PURE — no server imports. This runs inside the isolated child. Every ctx
|
||||
* method is plumbing that turns a call into an RPC message to the host; the
|
||||
* child holds no db handle, no secrets, no network by default.
|
||||
*/
|
||||
|
||||
export interface PluginContext {
|
||||
readonly id: string;
|
||||
readonly config: Readonly<Record<string, unknown>>;
|
||||
db: {
|
||||
query<T = unknown>(sql: string, ...args: unknown[]): Promise<T[]>;
|
||||
exec(sql: string, ...args: unknown[]): Promise<{ changes: number }>;
|
||||
migrate(id: string, sql: string): Promise<{ applied: boolean }>;
|
||||
};
|
||||
trips: {
|
||||
getById(tripId: number, asUserId: number): Promise<unknown>;
|
||||
getPlaces(tripId: number, asUserId: number): Promise<unknown[]>;
|
||||
getReservations(tripId: number, asUserId: number): Promise<unknown[]>;
|
||||
};
|
||||
users: {
|
||||
getById(id: number): Promise<unknown>;
|
||||
};
|
||||
ws: {
|
||||
broadcastToTrip(tripId: number, event: string, data: Record<string, unknown>): Promise<void>;
|
||||
broadcastToUser(userId: number, event: string, data: Record<string, unknown>): Promise<void>;
|
||||
};
|
||||
log: {
|
||||
info(msg: string, meta?: Record<string, unknown>): void;
|
||||
warn(msg: string, meta?: Record<string, unknown>): void;
|
||||
error(msg: string, meta?: Record<string, unknown>): void;
|
||||
};
|
||||
}
|
||||
|
||||
export interface PluginRequest {
|
||||
method: string;
|
||||
path: string;
|
||||
query: Record<string, unknown>;
|
||||
body: unknown;
|
||||
user: { id: number; username: string; isAdmin: boolean } | null;
|
||||
}
|
||||
export interface PluginResponse {
|
||||
status: number;
|
||||
headers?: Record<string, string>;
|
||||
body?: unknown;
|
||||
}
|
||||
export interface PluginRoute {
|
||||
method: 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH';
|
||||
path: string;
|
||||
auth?: boolean;
|
||||
handler(req: PluginRequest, ctx: PluginContext): Promise<PluginResponse>;
|
||||
}
|
||||
export interface PluginJob {
|
||||
id: string;
|
||||
schedule: string;
|
||||
handler(ctx: PluginContext): Promise<void>;
|
||||
}
|
||||
export interface PluginDefinition {
|
||||
onLoad?(ctx: PluginContext): Promise<void> | void;
|
||||
onUnload?(ctx: PluginContext): Promise<void> | void;
|
||||
routes?: PluginRoute[];
|
||||
jobs?: PluginJob[];
|
||||
// hooks (photo/calendar) land in M2
|
||||
}
|
||||
|
||||
/** Identity helper: gives authors types + a stable shape. A plain object works too. */
|
||||
export function definePlugin(def: PluginDefinition): PluginDefinition {
|
||||
return def;
|
||||
}
|
||||
|
||||
/** Transport the child entry wires to process.send / message correlation. */
|
||||
export interface ChildTransport {
|
||||
rpc(method: string, params: Record<string, unknown>): Promise<unknown>;
|
||||
emit(topic: string, data: unknown): void;
|
||||
}
|
||||
|
||||
/** Build the ctx the plugin's handlers receive — every method is an RPC call. */
|
||||
export function createPluginContext(
|
||||
id: string,
|
||||
config: Record<string, unknown>,
|
||||
t: ChildTransport,
|
||||
): PluginContext {
|
||||
return {
|
||||
id,
|
||||
config: Object.freeze({ ...config }),
|
||||
db: {
|
||||
query: (sql, ...args) => t.rpc('db.query', { sql, args }) as Promise<never[]>,
|
||||
exec: (sql, ...args) => t.rpc('db.exec', { sql, args }) as Promise<{ changes: number }>,
|
||||
migrate: (mid, sql) => t.rpc('db.migrate', { id: mid, sql }) as Promise<{ applied: boolean }>,
|
||||
},
|
||||
trips: {
|
||||
getById: (tripId, asUserId) => t.rpc('trips.getById', { tripId, asUserId }),
|
||||
getPlaces: (tripId, asUserId) => t.rpc('trips.getPlaces', { tripId, asUserId }) as Promise<unknown[]>,
|
||||
getReservations: (tripId, asUserId) => t.rpc('trips.getReservations', { tripId, asUserId }) as Promise<unknown[]>,
|
||||
},
|
||||
users: {
|
||||
getById: (uid) => t.rpc('users.getById', { id: uid }),
|
||||
},
|
||||
ws: {
|
||||
broadcastToTrip: async (tripId, event, data) => {
|
||||
await t.rpc('ws.broadcastToTrip', { tripId, event, data });
|
||||
},
|
||||
broadcastToUser: async (userId, event, data) => {
|
||||
await t.rpc('ws.broadcastToUser', { userId, event, data });
|
||||
},
|
||||
},
|
||||
log: {
|
||||
info: (msg, meta) => t.emit('log', { level: 'info', msg, meta }),
|
||||
warn: (msg, meta) => t.emit('log', { level: 'warn', msg, meta }),
|
||||
error: (msg, meta) => t.emit('log', { level: 'error', msg, meta }),
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,245 @@
|
||||
import { fork, type ChildProcess } from 'node:child_process';
|
||||
import { resolveChildEntry, pluginCodeDir } from '../paths';
|
||||
import type { Envelope, RpcRequest } from '../protocol/envelope';
|
||||
import type { PluginRpcHost } from '../host/rpc-host';
|
||||
|
||||
/**
|
||||
* Owns the lifecycle of every running plugin child (#plugins, M1): spawn on
|
||||
* activate, route RPC between the child and its capability host, watch
|
||||
* heartbeats, and restart-with-backoff / auto-disable on crashes. A child dying
|
||||
* — segfault, throw, OOM, infinite loop — only ever kills the child; the Nest
|
||||
* event loop never hiccups. That is what finally makes "a plugin can't crash
|
||||
* TREK" true.
|
||||
*/
|
||||
|
||||
export type PluginStatus = 'starting' | 'active' | 'error' | 'stopped';
|
||||
|
||||
export interface SupervisorHooks {
|
||||
onStatus?(id: string, status: PluginStatus, error?: string): void;
|
||||
onLog?(id: string, level: string, msg: string, meta?: unknown): void;
|
||||
/** Any non-lifecycle event the child emits (e.g. a plugin's own signals). */
|
||||
onEvent?(id: string, topic: string, data: unknown): void;
|
||||
}
|
||||
|
||||
interface Supervised {
|
||||
id: string;
|
||||
granted: ReadonlySet<string>;
|
||||
config: Record<string, unknown>;
|
||||
rpcHost: PluginRpcHost;
|
||||
child: ChildProcess | null;
|
||||
status: PluginStatus;
|
||||
crashes: number[]; // crash timestamps (ms)
|
||||
lastBeat: number;
|
||||
activation?: { resolve: () => void; reject: (e: Error) => void };
|
||||
}
|
||||
|
||||
export interface SupervisorTuning {
|
||||
heartbeatTimeoutMs?: number;
|
||||
crashWindowMs?: number;
|
||||
crashLimit?: number;
|
||||
backoffCapMs?: number;
|
||||
killGraceMs?: number;
|
||||
}
|
||||
|
||||
const DEFAULTS: Required<SupervisorTuning> = {
|
||||
heartbeatTimeoutMs: 20_000, // 3–4 missed 5s beats
|
||||
crashWindowMs: 5 * 60_000,
|
||||
crashLimit: 5,
|
||||
backoffCapMs: 30_000,
|
||||
killGraceMs: 3000,
|
||||
};
|
||||
|
||||
export class PluginSupervisor {
|
||||
private running = new Map<string, Supervised>();
|
||||
private sweep: ReturnType<typeof setInterval> | null = null;
|
||||
private readonly tuning: Required<SupervisorTuning>;
|
||||
|
||||
constructor(
|
||||
private readonly createRpcHost: (id: string, granted: ReadonlySet<string>) => PluginRpcHost,
|
||||
private readonly hooks: SupervisorHooks = {},
|
||||
tuning: SupervisorTuning = {},
|
||||
) {
|
||||
this.tuning = { ...DEFAULTS, ...tuning };
|
||||
}
|
||||
|
||||
/** Spawn a plugin and resolve once it reports `loaded` (or reject on load error). */
|
||||
activate(id: string, granted: ReadonlySet<string>, config: Record<string, unknown> = {}): Promise<void> {
|
||||
if (this.running.has(id)) return Promise.resolve();
|
||||
const sup: Supervised = {
|
||||
id,
|
||||
granted,
|
||||
config,
|
||||
rpcHost: this.createRpcHost(id, granted),
|
||||
child: null,
|
||||
status: 'starting',
|
||||
crashes: [],
|
||||
lastBeat: Date.now(),
|
||||
};
|
||||
this.running.set(id, sup);
|
||||
this.ensureSweep();
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
sup.activation = { resolve, reject };
|
||||
this.spawn(sup);
|
||||
});
|
||||
}
|
||||
|
||||
/** Stop a plugin: ask it to unload, then kill. Idempotent. */
|
||||
async disable(id: string): Promise<void> {
|
||||
const sup = this.running.get(id);
|
||||
if (!sup) return;
|
||||
this.running.delete(id);
|
||||
this.setStatus(sup, 'stopped');
|
||||
await this.kill(sup);
|
||||
sup.rpcHost.dispose();
|
||||
}
|
||||
|
||||
isActive(id: string): boolean {
|
||||
return this.running.get(id)?.status === 'active';
|
||||
}
|
||||
statusOf(id: string): PluginStatus | null {
|
||||
return this.running.get(id)?.status ?? null;
|
||||
}
|
||||
|
||||
async shutdownAll(): Promise<void> {
|
||||
if (this.sweep) clearInterval(this.sweep);
|
||||
this.sweep = null;
|
||||
const all = [...this.running.values()];
|
||||
await Promise.all(all.map((s) => this.kill(s)));
|
||||
for (const s of all) s.rpcHost.dispose();
|
||||
this.running.clear();
|
||||
}
|
||||
|
||||
// ── internals ──────────────────────────────────────────────────────────────
|
||||
|
||||
private spawn(sup: Supervised): void {
|
||||
const { entry, execArgv, forkCwd } = resolveChildEntry();
|
||||
const child = fork(entry, [sup.id, pluginCodeDir(sup.id)], {
|
||||
cwd: forkCwd ?? pluginCodeDir(sup.id),
|
||||
execArgv,
|
||||
stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
|
||||
// Whitelist env — nothing inherited. No JWT_SECRET, no DB creds, no PATH-leaked secrets.
|
||||
env: {
|
||||
NODE_ENV: process.env.NODE_ENV ?? 'production',
|
||||
TZ: process.env.TZ ?? '',
|
||||
PATH: process.env.PATH ?? '',
|
||||
TREK_PLUGIN_ID: sup.id,
|
||||
},
|
||||
});
|
||||
sup.child = child;
|
||||
sup.lastBeat = Date.now();
|
||||
|
||||
child.on('message', (raw: unknown) => this.onMessage(sup, raw as Envelope));
|
||||
child.on('exit', (code, signal) => this.onExit(sup, code, signal));
|
||||
child.on('error', (e) => this.hooks.onLog?.(sup.id, 'error', `child error: ${e.message}`));
|
||||
child.stdout?.on('data', (b) => this.hooks.onLog?.(sup.id, 'info', String(b).trimEnd()));
|
||||
child.stderr?.on('data', (b) => this.hooks.onLog?.(sup.id, 'error', String(b).trimEnd()));
|
||||
}
|
||||
|
||||
private async onMessage(sup: Supervised, msg: Envelope): Promise<void> {
|
||||
if (!msg || typeof msg !== 'object') return;
|
||||
|
||||
if (msg.k === 'req') {
|
||||
// A ctx.* call from the plugin — dispatch through its capability host.
|
||||
const res = await sup.rpcHost.dispatch(msg as RpcRequest);
|
||||
sup.child?.send(res);
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg.k === 'evt') {
|
||||
switch (msg.topic) {
|
||||
case 'hello':
|
||||
sup.child?.send({ k: 'evt', topic: 'init', data: { config: sup.config } } satisfies Envelope);
|
||||
break;
|
||||
case 'heartbeat':
|
||||
sup.lastBeat = Date.now();
|
||||
break;
|
||||
case 'loaded':
|
||||
sup.lastBeat = Date.now();
|
||||
this.setStatus(sup, 'active');
|
||||
sup.activation?.resolve();
|
||||
sup.activation = undefined;
|
||||
break;
|
||||
case 'load-error': {
|
||||
const message = (msg.data as { message?: string })?.message || 'plugin load failed';
|
||||
this.setStatus(sup, 'error', message);
|
||||
sup.activation?.reject(new Error(message));
|
||||
sup.activation = undefined;
|
||||
await this.kill(sup);
|
||||
sup.rpcHost.dispose();
|
||||
break;
|
||||
}
|
||||
case 'log': {
|
||||
const d = msg.data as { level?: string; msg?: string; meta?: unknown };
|
||||
this.hooks.onLog?.(sup.id, d.level || 'info', d.msg || '', d.meta);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
this.hooks.onEvent?.(sup.id, msg.topic, msg.data);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private onExit(sup: Supervised, code: number | null, signal: string | null): void {
|
||||
// Reject any pending ctx calls implicitly by killing the child; the child is gone.
|
||||
sup.child = null;
|
||||
// A clean stop we asked for isn't a crash.
|
||||
if (sup.status === 'stopped' || sup.status === 'error') return;
|
||||
if (!this.running.has(sup.id)) return;
|
||||
|
||||
sup.crashes.push(Date.now());
|
||||
const recent = sup.crashes.filter((t) => t > Date.now() - this.tuning.crashWindowMs);
|
||||
if (recent.length >= this.tuning.crashLimit) {
|
||||
this.setStatus(sup, 'error', `auto-disabled after ${recent.length} crashes`);
|
||||
sup.activation?.reject(new Error('plugin crashed repeatedly'));
|
||||
sup.activation = undefined;
|
||||
sup.rpcHost.dispose();
|
||||
return;
|
||||
}
|
||||
const delay = Math.min(this.tuning.backoffCapMs, 1000 * 2 ** (recent.length - 1));
|
||||
this.hooks.onLog?.(sup.id, 'warn', `crashed (code=${code} sig=${signal}); restarting in ${delay}ms`);
|
||||
this.setStatus(sup, 'starting');
|
||||
const timer = setTimeout(() => {
|
||||
if (this.running.has(sup.id)) this.spawn(sup);
|
||||
}, delay);
|
||||
timer.unref?.();
|
||||
}
|
||||
|
||||
private ensureSweep(): void {
|
||||
if (this.sweep) return;
|
||||
this.sweep = setInterval(() => this.reapStale(), 5000);
|
||||
this.sweep.unref?.();
|
||||
}
|
||||
|
||||
/** Kill any active plugin that has stopped sending heartbeats (drives the crash path). */
|
||||
reapStale(now = Date.now()): void {
|
||||
for (const sup of this.running.values()) {
|
||||
if (sup.status === 'active' && now - sup.lastBeat > this.tuning.heartbeatTimeoutMs) {
|
||||
this.hooks.onLog?.(sup.id, 'warn', 'missed heartbeats; killing');
|
||||
sup.child?.kill('SIGKILL');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async kill(sup: Supervised): Promise<void> {
|
||||
const child = sup.child;
|
||||
if (!child) return;
|
||||
sup.child = null;
|
||||
child.send?.({ k: 'evt', topic: 'shutdown', data: {} } satisfies Envelope);
|
||||
await new Promise<void>((resolve) => {
|
||||
const t = setTimeout(() => {
|
||||
child.kill('SIGKILL');
|
||||
resolve();
|
||||
}, this.tuning.killGraceMs);
|
||||
t.unref?.();
|
||||
child.once('exit', () => {
|
||||
clearTimeout(t);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private setStatus(sup: Supervised, status: PluginStatus, error?: string): void {
|
||||
sup.status = status;
|
||||
this.hooks.onStatus?.(sup.id, status, error);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user