feat: give plugins their own tables, migrations and sync entities

This commit is contained in:
LukeGus committed 2026-09-22 15:49:54 -05:00
1 parent 5796186a1c
commit c496a093ca
47 files changed
+32595 -290

No files matched your search

+77 -5
View File
@@ -31,6 +31,8 @@ import { getActor, runAsActor } from "./actor.js";
import { DisposableBag } from "./disposables.js";
import type { PluginManifest } from "@termix/plugin-sdk/manifest";
import type { PluginContext, PluginModule } from "@termix/plugin-sdk/backend";
import type { PluginTableDefinition } from "@termix/plugin-sdk/db";
import * as syncRegistry from "./sync-registry.js";
export type { PluginModule };
@@ -38,6 +40,20 @@ export type { PluginModule };
const MAX_KV_KEY_LENGTH = 128;
const MAX_KV_VALUE_BYTES = 256 * 1024;
/**
* How many keys one plugin may hold.
*
* The value cap alone bounds a single row, not the table: a plugin writing
* unique keys in a loop could still fill the database. Configurable because
* the right ceiling depends on the install, not on the plugin.
*/
const DEFAULT_MAX_KV_KEYS = 10_000;
function maxKvKeys(env: NodeJS.ProcessEnv = process.env): number {
const raw = Number(env.PLUGIN_MAX_KV_KEYS);
return Number.isFinite(raw) && raw > 0 ? raw : DEFAULT_MAX_KV_KEYS;
}
export interface PluginHandle {
module: PluginModule;
bag: DisposableBag;
@@ -169,11 +185,21 @@ export function createPluginContext(
}
const { createCurrentPluginStorageRepository } =
await import("../database/repositories/factory.js");
await createCurrentPluginStorageRepository().set(
pluginId,
key,
serialized,
);
const repository = createCurrentPluginStorageRepository();
// Only a new key can grow the table, so an overwrite is never blocked
// by a full one. Checked before the write, not after.
const existing = await repository.get(pluginId, key);
if (existing === null) {
const limit = maxKvKeys();
if ((await repository.countKeys(pluginId)) >= limit) {
throw new Error(
`Plugin "${pluginId}" has reached the ${limit} key limit for ctx.kv`,
);
}
}
await repository.set(pluginId, key, serialized);
},
{ action: "kv_set" },
);
@@ -200,6 +226,39 @@ export function createPluginContext(
{ action: "kv_list" },
);
const dbDefine = guarded(
manifest,
"db:own",
async (definition: PluginTableDefinition) => {
const { registerTable } = await import("./data.js");
return registerTable(pluginId, definition);
},
{ action: "db_define", details: () => "table definition registered" },
);
const dbClient = guarded(
manifest,
"db:own",
async () => {
const { getDb } = await import("../database/db/index.js");
return getDb();
},
{ action: "db_client" },
);
const dbRefs = guarded(
manifest,
"db:own",
async () => {
// Read-only by convention, not by engine: a plugin holding these can
// write through them. The capability and the audit line are the record
// that it did. See the header of this file.
const schema = await import("../database/db/schema.js");
return { users: schema.users, hosts: schema.hosts };
},
{ action: "db_refs" },
);
return {
pluginId,
manifest,
@@ -237,6 +296,19 @@ export function createPluginContext(
},
},
db: {
define: (definition) => dbDefine(definition) as never,
client: () => dbClient() as never,
refs: () => dbRefs() as never,
},
sync: {
registerEntity: (entity) => {
const dispose = syncRegistry.registerEntity(pluginId, entity);
handle.bag.add(dispose, `sync entity ${entity.type}`);
},
},
kv: {
get: (key) => kvGet(key),
set: (key, value) => kvSet(key, value),
+177
View File
@@ -0,0 +1,177 @@
/**
* The live database side of plugin-owned data.
*
* Holds the table definitions each plugin has registered, builds the runner
* that applies its migrations against the real database, and drops everything
* again on uninstall. The pure logic lives in migrations.ts so it can be
* tested without a database; this file is the part that needs one.
*/
import { sql } from "drizzle-orm";
import { pluginLogger } from "../utils/logger.js";
import { resolveDatabaseDialect } from "../database/db/dialect.js";
import type { DatabaseDialect } from "../database/db/dialect.js";
import type { PluginTableDefinition } from "@termix/plugin-sdk/db";
import { prefixedTableName } from "@termix/plugin-sdk/db";
import { buildTable, dropTableSql } from "./table-builder.js";
import {
applyPluginMigrations,
readMigrations,
tableExistsSql,
type MigrationRunner,
} from "./migrations.js";
/** Definitions registered by a plugin, in the order it declared them. */
const registered = new Map<string, PluginTableDefinition[]>();
/** Built table objects, keyed by plugin id then definition name. */
const built = new Map<string, Map<string, unknown>>();
export function registerTable(
pluginId: string,
definition: PluginTableDefinition,
): unknown {
const definitions = registered.get(pluginId) ?? [];
const existing = definitions.find((entry) => entry.name === definition.name);
if (existing) {
// Re-registering on re-activation is normal: the module is cached, so
// activate runs again against the same definitions.
return getTable(pluginId, definition.name);
}
definitions.push(definition);
registered.set(pluginId, definitions);
const table = buildTable(pluginId, definition);
const tables = built.get(pluginId) ?? new Map<string, unknown>();
tables.set(definition.name, table);
built.set(pluginId, tables);
return table;
}
export function getTable(pluginId: string, name: string): unknown {
return built.get(pluginId)?.get(name);
}
export function listTables(pluginId: string): PluginTableDefinition[] {
return [...(registered.get(pluginId) ?? [])];
}
/** Forgets a plugin's definitions. Used by the tests and by uninstall. */
export function forgetTables(pluginId: string): void {
registered.delete(pluginId);
built.delete(pluginId);
}
export function resetPluginData(): void {
registered.clear();
built.clear();
}
async function execute(statements: string[]): Promise<void> {
const { getDb } = await import("../database/db/index.js");
const db = getDb();
for (const statement of statements) {
await db.run(sql.raw(statement));
}
}
/**
* Builds the runner that applies migrations against the live database.
*
* SQLite gets a transaction; the client-server engines do not, because MySQL
* commits implicitly on DDL so a "transaction" there would only be a promise
* the engine does not keep. A half-applied migration is recorded statement by
* statement either way, so a retry resumes rather than repeating.
*/
export async function createMigrationRunner(
dialect: DatabaseDialect = resolveDatabaseDialect(),
): Promise<MigrationRunner> {
const { createCurrentPluginMigrationRepository } =
await import("../database/repositories/factory.js");
const repository = createCurrentPluginMigrationRepository();
return {
dialect,
listApplied: async (pluginId) => {
const rows = await repository.listByPlugin(pluginId);
return rows.map((row) => ({
migrationId: row.migrationId,
checksum: row.checksum,
}));
},
execute,
record: async (pluginId, migration) => {
await repository.record(pluginId, migration.id, migration.checksum);
},
};
}
/**
* Applies a plugin's migrations. Throws so the loader can fail the plugin.
*
* Returns early when the plugin ships none, before anything asks for a
* database handle: most plugins own no tables, and activating one should not
* require a connection it never uses.
*/
export async function migratePlugin(
pluginId: string,
pluginDir: string,
dialect: DatabaseDialect = resolveDatabaseDialect(),
): Promise<string[]> {
if (readMigrations(pluginDir, dialect).length === 0) return [];
const runner = await createMigrationRunner(dialect);
return applyPluginMigrations(pluginId, pluginDir, runner);
}
/** Whether a table exists, asked in the active dialect's own catalog. */
export async function tableExists(
table: string,
dialect: DatabaseDialect = resolveDatabaseDialect(),
): Promise<boolean> {
const { getDb } = await import("../database/db/index.js");
const rows = (await getDb().all(tableExistsSql(dialect, table))) as unknown[];
return rows.length > 0;
}
/**
* Removes everything a plugin owns.
*
* Tables go in reverse registration order so a table referenced by a later one
* is dropped last. Disabling a plugin never calls this: data outlives being
* turned off, and only an explicit uninstall throws it away.
*/
export async function removePluginData(
pluginId: string,
dialect: DatabaseDialect = resolveDatabaseDialect(),
): Promise<{ tables: string[]; kvKeys: number; migrations: number }> {
const definitions = listTables(pluginId);
const dropped: string[] = [];
for (const definition of [...definitions].reverse()) {
await execute([dropTableSql(dialect, pluginId, definition)]);
dropped.push(prefixedTableName(pluginId, definition.name));
}
const {
createCurrentPluginStorageRepository,
createCurrentPluginMigrationRepository,
createCurrentPluginPermissionGrantRepository,
} = await import("../database/repositories/factory.js");
const kvKeys =
await createCurrentPluginStorageRepository().deleteByPlugin(pluginId);
const migrations =
await createCurrentPluginMigrationRepository().deleteByPlugin(pluginId);
await createCurrentPluginPermissionGrantRepository().deleteByPlugin(pluginId);
forgetTables(pluginId);
pluginLogger.info(
`Removed data for plugin ${pluginId}: ${dropped.length} table(s), ${kvKeys} kv key(s)`,
{ operation: "plugin_remove_data" },
);
return { tables: dropped, kvKeys, migrations };
}
+7
View File
@@ -338,6 +338,13 @@ export class PluginLoader {
);
}
// Before activate, not inside it: a plugin's first line may query its
// own tables, and a migration that fails should stop it starting rather
// than leave it half-running against a schema that is not there. A
// throw here lands in the catch below, which fails this plugin only.
const { migratePlugin } = await import("./data.js");
await migratePlugin(plugin.id, plugin.dir);
const handle = createPluginHandle(plugin.id, { activate, deactivate });
const ctx = createPluginContext(plugin.manifest, handle);
+316
View File
@@ -0,0 +1,316 @@
/**
* Applies a plugin's migrations before it activates.
*
* A plugin ships plain SQL per dialect in migrations/<dialect>/NNNN_name.sql.
* Not drizzle-kit: core's own SQLite schema is hand-written DDL and only the
* client-server engines use drizzle migrations, so a plugin that used
* drizzle-kit would still need a hand-rolled SQLite path. One runner for all
* three keeps a plugin's tables identical everywhere.
*
* Failures are contained. An edited migration that already ran blocks that
* plugin; a migration that throws marks it failed. Core keeps booting either
* way, because one plugin's bad SQL is not a reason to take the server down.
*/
import crypto from "node:crypto";
import fs from "node:fs";
import path from "node:path";
import { sql } from "drizzle-orm";
import { pluginLogger } from "../utils/logger.js";
import type { DatabaseDialect } from "../database/db/dialect.js";
import { tablePrefix, LEGACY_TABLE_OWNERS } from "@termix/plugin-sdk/db";
export { LEGACY_TABLE_OWNERS };
/** Dialect directory names, matching DatabaseDialect exactly. */
const DIALECTS: readonly DatabaseDialect[] = ["sqlite", "postgres", "mysql"];
const FILE_PATTERN = /^(\d{4})_([a-z0-9_]+)\.sql$/;
export interface PluginMigration {
/** The file's stem, e.g. "0001_init". Stable across dialects. */
id: string;
sequence: number;
file: string;
sql: string;
checksum: string;
}
/** Raised when an already-applied migration no longer matches its checksum. */
export class PluginMigrationChecksumError extends Error {
readonly code = "EPLUGINMIGRATIONCHECKSUM";
constructor(pluginId: string, migrationId: string) {
super(
`Plugin "${pluginId}" migration "${migrationId}" has changed since it was applied. ` +
`An applied migration is immutable: add a new one instead of editing this file.`,
);
this.name = "PluginMigrationChecksumError";
}
}
export function checksum(contents: string): string {
// Normalised so a line-ending change between platforms is not a mismatch.
const normalised = contents.replace(/\r\n/g, "\n").trimEnd();
return crypto.createHash("sha256").update(normalised).digest("hex");
}
export function migrationsDir(
pluginDir: string,
dialect: DatabaseDialect,
): string {
return path.join(pluginDir, "migrations", dialect);
}
/** Reads the migrations a plugin ships for one dialect, in sequence order. */
export function readMigrations(
pluginDir: string,
dialect: DatabaseDialect,
): PluginMigration[] {
const dir = migrationsDir(pluginDir, dialect);
if (!fs.existsSync(dir)) return [];
const migrations: PluginMigration[] = [];
for (const file of fs.readdirSync(dir).sort()) {
const match = FILE_PATTERN.exec(file);
if (!match) {
if (file.endsWith(".sql")) {
throw new Error(
`Migration "${file}" must be named NNNN_name.sql, lower snake_case`,
);
}
continue;
}
const contents = fs.readFileSync(path.join(dir, file), "utf8");
migrations.push({
id: file.replace(/\.sql$/, ""),
sequence: Number(match[1]),
file,
sql: contents,
checksum: checksum(contents),
});
}
migrations.sort((a, b) => a.sequence - b.sequence);
return migrations;
}
/**
* Splits a migration file into statements.
*
* Deliberately simple: statements end at a semicolon that is not inside a
* string literal or a comment. A plugin needing more than that should put the
* logic in its own code, not in DDL.
*/
export function splitStatements(source: string): string[] {
const statements: string[] = [];
let current = "";
let quote: string | null = null;
let lineComment = false;
let blockComment = false;
for (let i = 0; i < source.length; i++) {
const char = source[i];
const next = source[i + 1];
if (lineComment) {
current += char;
if (char === "\n") lineComment = false;
continue;
}
if (blockComment) {
current += char;
if (char === "*" && next === "/") {
current += next;
i++;
blockComment = false;
}
continue;
}
if (quote) {
current += char;
// '' inside a quoted string is an escaped quote, not a terminator.
if (char === quote) {
if (next === quote) {
current += next;
i++;
} else {
quote = null;
}
}
continue;
}
if (char === "-" && next === "-") {
lineComment = true;
current += char;
continue;
}
if (char === "/" && next === "*") {
blockComment = true;
current += char;
continue;
}
if (char === "'" || char === '"' || char === "`") {
quote = char;
current += char;
continue;
}
if (char === ";") {
if (current.trim()) statements.push(current.trim());
current = "";
continue;
}
current += char;
}
if (current.trim()) statements.push(current.trim());
return statements;
}
/**
* Tables a migration may create or alter.
*
* A plugin owns the p_<id>_ namespace and nothing else. Enforced here as well
* as in the CLI, because a plugin installed from a tarball never ran the CLI.
*/
const CREATE_OR_ALTER =
/\b(?:CREATE\s+(?:TEMP\s+|TEMPORARY\s+)?TABLE(?:\s+IF\s+NOT\s+EXISTS)?|ALTER\s+TABLE|DROP\s+TABLE(?:\s+IF\s+EXISTS)?|TRUNCATE\s+TABLE)\s+([`"']?)([a-zA-Z0-9_]+)\1/gi;
/**
* Checks that every table a migration touches belongs to this plugin.
*
* A legacy table it is allowed to adopt is permitted too, because adoption
* renames it into the plugin's namespace in the same migration.
*/
export function assertOwnedTables(
pluginId: string,
statements: string[],
): void {
const prefix = tablePrefix(pluginId);
for (const statement of statements) {
CREATE_OR_ALTER.lastIndex = 0;
let match: RegExpExecArray | null;
while ((match = CREATE_OR_ALTER.exec(statement)) !== null) {
const table = match[2];
if (table.startsWith(prefix)) continue;
if (LEGACY_TABLE_OWNERS[table] === pluginId) continue;
const owner = LEGACY_TABLE_OWNERS[table];
throw new Error(
owner
? `Plugin "${pluginId}" may not touch "${table}": it belongs to the "${owner}" plugin.`
: `Plugin "${pluginId}" may only create or alter tables prefixed "${prefix}", not "${table}".`,
);
}
}
}
export interface AppliedMigration {
migrationId: string;
checksum: string;
}
export interface MigrationRunner {
dialect: DatabaseDialect;
listApplied: (pluginId: string) => Promise<AppliedMigration[]>;
execute: (statements: string[]) => Promise<void>;
record: (pluginId: string, migration: PluginMigration) => Promise<void>;
/** Runs fn inside a transaction where the dialect supports DDL in one. */
transaction?: (fn: () => Promise<void>) => Promise<void>;
}
/**
* Brings one plugin's tables up to date.
*
* Returns the ids applied, so a caller can log what changed. Throws on a
* checksum mismatch or a failing statement; the loader turns either into a
* failed plugin without disturbing the rest of the server.
*/
export async function applyPluginMigrations(
pluginId: string,
pluginDir: string,
runner: MigrationRunner,
): Promise<string[]> {
const migrations = readMigrations(pluginDir, runner.dialect);
if (migrations.length === 0) return [];
const applied = new Map(
(await runner.listApplied(pluginId)).map((row) => [
row.migrationId,
row.checksum,
]),
);
const pending: PluginMigration[] = [];
for (const migration of migrations) {
const previous = applied.get(migration.id);
if (previous === undefined) {
pending.push(migration);
continue;
}
if (previous !== migration.checksum) {
throw new PluginMigrationChecksumError(pluginId, migration.id);
}
}
if (pending.length === 0) return [];
const run = async () => {
for (const migration of pending) {
const statements = splitStatements(migration.sql);
assertOwnedTables(pluginId, statements);
await runner.execute(statements);
await runner.record(pluginId, migration);
}
};
if (runner.transaction) {
await runner.transaction(run);
} else {
await run();
}
pluginLogger.info(
`Applied ${pending.length} migration(s) for plugin ${pluginId}: ${pending
.map((m) => m.id)
.join(", ")}`,
{ operation: "plugin_migrate" },
);
return pending.map((migration) => migration.id);
}
/** The dialect directories a plugin is expected to ship. */
export function expectedDialects(): readonly DatabaseDialect[] {
return DIALECTS;
}
/**
* Renames a legacy core table into a plugin's namespace, or creates it fresh.
*
* Adoption is how a feature keeps its data when it moves out of core. The
* table is renamed rather than copied so nothing is duplicated and nothing has
* to be migrated row by row.
*/
export function adoptLegacyTableSql(
dialect: DatabaseDialect,
legacyName: string,
newName: string,
): string {
if (dialect === "mysql") {
return `RENAME TABLE \`${legacyName}\` TO \`${newName}\`;`;
}
return `ALTER TABLE "${legacyName}" RENAME TO "${newName}";`;
}
/** Whether a table exists, asked in the dialect's own catalog. */
export function tableExistsSql(dialect: DatabaseDialect, table: string) {
if (dialect === "sqlite") {
return sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = ${table}`;
}
if (dialect === "postgres") {
return sql`SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = ${table}`;
}
return sql`SELECT table_name FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = ${table}`;
}
+100
View File
@@ -0,0 +1,100 @@
/**
* Which entities remote sync carries.
*
* Core registers its ten at boot and a plugin registers its own through
* ctx.sync, so the list is built rather than hardcoded in four places. The
* wire names are unchanged: existing tombstones and rows on the other side of
* a pair match on those strings, so renaming one would orphan data.
*
* Order is dependency order, not declaration order. A row referencing another
* entity is resolved by that entity's syncId, so the target has to exist on
* the far side first. That is why every registration carries an explicit
* order rather than relying on registration sequence.
*/
import { pluginLogger } from "../utils/logger.js";
import type { SyncEntityRegistration } from "@termix/plugin-sdk/backend";
export interface RegisteredSyncEntity extends SyncEntityRegistration {
/** Which plugin registered it, or "core". */
owner: string;
order: number;
userColumn: string;
readOnlyFields: readonly string[];
encryptedFields: readonly string[];
}
const entities = new Map<string, RegisteredSyncEntity>();
/** Core's own entities register under this owner. */
export const CORE_OWNER = "core";
export function registerEntity(
owner: string,
entity: SyncEntityRegistration,
): () => void {
if (!entity.type) {
throw new Error("A sync entity needs a type");
}
const existing = entities.get(entity.type);
if (existing && existing.owner !== owner) {
throw new Error(
`Sync entity "${entity.type}" is already registered by "${existing.owner}"`,
);
}
entities.set(entity.type, {
...entity,
owner,
order: entity.order ?? 100,
userColumn: entity.userColumn ?? "userId",
readOnlyFields: entity.readOnlyFields ?? [],
encryptedFields: entity.encryptedFields ?? [],
});
return () => {
const current = entities.get(entity.type);
// Identity-checked: a plugin that restarted must not revoke the
// registration its own restart installed.
if (current && current.owner === owner) entities.delete(entity.type);
};
}
export function getEntity(type: string): RegisteredSyncEntity | undefined {
return entities.get(type);
}
export function hasEntity(type: string): boolean {
return entities.has(type);
}
/** Every entity, in dependency order. */
export function listEntities(): RegisteredSyncEntity[] {
return [...entities.values()].sort((a, b) => {
if (a.order !== b.order) return a.order - b.order;
return a.type.localeCompare(b.type);
});
}
/** The ordered wire names, which is what the Electron client syncs by. */
export function listEntityTypes(): string[] {
return listEntities().map((entity) => entity.type);
}
export function unregisterByOwner(owner: string): void {
for (const [type, entity] of [...entities]) {
if (entity.owner === owner) entities.delete(type);
}
}
export function resetSyncRegistry(): void {
entities.clear();
}
/** Logged once at boot so a missing entity is visible without a debugger. */
export function logRegisteredEntities(): void {
pluginLogger.info(`Sync entities: ${listEntityTypes().join(", ")}`, {
operation: "sync_registry",
});
}
+104
View File
@@ -0,0 +1,104 @@
/**
* Turns a plugin's table definition into the Drizzle table it is queried
* through.
*
* One table object for every dialect. The sqlite-core definitions already
* encode correctly on each engine at query time, so the repositories use them
* everywhere; only DDL genuinely differs, and that lives in the SDK's ddl
* entry so the CLI and the server share one mapping.
*/
import {
sqliteTable,
text as sqliteText,
integer as sqliteInteger,
index as sqliteIndex,
uniqueIndex as sqliteUniqueIndex,
type AnySQLiteColumn,
} from "drizzle-orm/sqlite-core";
import { sql } from "drizzle-orm";
import { prefixedTableName } from "@termix/plugin-sdk/db";
import type { PluginTableDefinition } from "@termix/plugin-sdk/db";
import { columnName } from "@termix/plugin-sdk/ddl";
type ColumnChain = {
notNull: () => unknown;
primaryKey: () => unknown;
unique: () => unknown;
default: (value: unknown) => unknown;
};
/**
* Builds the queryable table object.
*
* Always sqlite-core, on every dialect. drizzle's three table classes share no
* base class, and the query-builder surface a plugin uses is identical across
* them - the same deliberate approximation repositories/database-context.ts
* documents for core.
*/
export function buildTable(
pluginId: string,
definition: PluginTableDefinition,
): unknown {
const physical = prefixedTableName(pluginId, definition.name);
const columns: Record<string, unknown> = {};
for (const [property, column] of Object.entries(definition.columns)) {
const name = columnName(property, column);
let built: unknown;
switch (column.type) {
case "id":
built = sqliteInteger(name).primaryKey({ autoIncrement: true });
break;
case "integer":
case "bigint":
case "refHost":
built = sqliteInteger(name);
break;
case "boolean":
built = sqliteInteger(name, { mode: "boolean" });
break;
default:
built = sqliteText(name);
break;
}
if (column.type !== "id") {
if (column.primaryKey) built = (built as ColumnChain).primaryKey();
if (column.notNull) built = (built as ColumnChain).notNull();
if (column.unique) built = (built as ColumnChain).unique();
if (column.defaultNow) {
built = (built as ColumnChain).default(sql`CURRENT_TIMESTAMP`);
} else if (column.defaultValue !== undefined) {
built = (built as ColumnChain).default(column.defaultValue);
}
}
columns[property] = built;
}
return sqliteTable(physical, columns as never, (table) =>
definition.indexes.map((entry) => {
const make = entry.unique ? sqliteUniqueIndex : sqliteIndex;
const target = table as unknown as Record<string, AnySQLiteColumn>;
return make(entry.name).on(
...(entry.columns.map((c) => target[c]) as [
AnySQLiteColumn,
...AnySQLiteColumn[],
]),
);
}),
);
}
// The DDL emitters live in the SDK, because termix-plugin writes the same SQL
// this server applies. Re-exported here so core keeps one import site.
export {
createTableSql,
dropTableSql,
addColumnSql,
keyedColumns,
columnName,
REFERENCEABLE,
} from "@termix/plugin-sdk/ddl";