k1nq/TREK
1
0
Fork 0
mirror of https://github.com/mauriceboe/TREK.git synced 2026-10-23 15:36:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(plugins): async DB-touching methods; transactions through UnitOfWork

Browse Source
This commit is contained in:
jubnl
2026-09-22 00:00:53 +02:00
parent e413308227
commit e28dc6de28
68 changed files with 743 additions and 706 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/nest/plugins/host/plugin-audit.ts
+7 -6
View File
@@ -98,7 +98,7 @@ function envInt(name: string, def: number): number {
/** Keep only the newest `MAX_AUDIT_ROWS` rows for a plugin. Called amortised from
* appendAudit; exported for tests. No-op when disabled or under the cap. */
export function pruneAudit(db: AuditDb, pluginId: string, keep = MAX_AUDIT_ROWS): void {
export async function pruneAudit(db: AuditDb, pluginId: string, keep = MAX_AUDIT_ROWS): Promise<void> {
if (keep <= 0) return;
db.prepare(
`DELETE FROM plugin_capability_audit WHERE plugin_id = ? AND id NOT IN
@@ -106,8 +106,9 @@ export function pruneAudit(db: AuditDb, pluginId: string, keep = MAX_AUDIT_ROWS)
).run(pluginId, pluginId, keep);
}
/** Append one entry to the per-plugin hash chain. Synchronous (better-sqlite3). */
export function appendAudit(db: AuditDb, e: AuditEntry): void {
/** Append one entry to the per-plugin hash chain. The statements themselves are
* synchronous (better-sqlite3); the method is async for the ORM seam. */
export async function appendAudit(db: AuditDb, e: AuditEntry): Promise<void> {
const prev =
(db.prepare('SELECT hash FROM plugin_capability_audit WHERE plugin_id = ? ORDER BY id DESC LIMIT 1').get(e.pluginId) as
| { hash: string }
@@ -120,7 +121,7 @@ export function appendAudit(db: AuditDb, e: AuditEntry): void {
).run(e.pluginId, e.actingUserId ?? null, e.method, e.resource ?? null, e.code, ts, prev || null, hash);
// Amortised retention: prune roughly every PRUNE_EVERY appends per plugin.
const n = (appendsSincePrune.get(e.pluginId) ?? 0) + 1;
if (n >= PRUNE_EVERY) { appendsSincePrune.set(e.pluginId, 0); pruneAudit(db, e.pluginId); }
if (n >= PRUNE_EVERY) { appendsSincePrune.set(e.pluginId, 0); await pruneAudit(db, e.pluginId); }
else appendsSincePrune.set(e.pluginId, n);
}
@@ -128,7 +129,7 @@ export function appendAudit(db: AuditDb, e: AuditEntry): void {
* "what have plugins done in my name?" view. This is what legitimizes the broad
* read grants: the user, not just the admin, can see every plugin action bound to
* them. Joined with the plugin name for display; capped. */
export function readAuditForUser(db: AuditDb, userId: number, limit = 200): unknown[] {
export async function readAuditForUser(db: AuditDb, userId: number, limit = 200): Promise<unknown[]> {
return db
.prepare(
`SELECT a.ts, a.plugin_id, p.name AS plugin_name, a.method, a.resource, a.code
@@ -139,7 +140,7 @@ export function readAuditForUser(db: AuditDb, userId: number, limit = 200): unkn
}
/** Read the most recent audit rows for a plugin (admin view). */
export function readAudit(db: AuditDb, pluginId: string, limit = 200): unknown[] {
export async function readAudit(db: AuditDb, pluginId: string, limit = 200): Promise<unknown[]> {
return db
.prepare('SELECT ts, acting_user_id, method, resource, code FROM plugin_capability_audit WHERE plugin_id = ? ORDER BY id DESC LIMIT ?')
.all(pluginId, limit);
server/src/nest/plugins/host/plugin-guards.service.ts
+1 -1
View File
@@ -31,7 +31,7 @@ export class PluginGuards {
* so a plugin cannot read another user's trips by naming their id. A userless
* context (a job, or onLoad) has no acting user and is refused.
*/
tripRead<T>(params: Record<string, unknown>, ctx: PluginRpcContext, read: (userId: number) => T): T {
async tripRead<T>(params: Record<string, unknown>, ctx: PluginRpcContext, read: (userId: number) => T): Promise<T> {
const tripId = num(params.tripId, 'tripId');
if (ctx.actingUserId === undefined) {
throw new ForbiddenResource('trip reads require an authenticated user context');
server/src/nest/plugins/host/plugin-host-state.ts
+3 -3
View File
@@ -44,7 +44,7 @@ export function closePluginDataDb(id: string): void {
// nothing persisted or phoned home.
const budgets = new Map<string, DailyBudget>();
export function budgetFor(id: string, conn: Database.Database): DailyBudget {
export async function budgetFor(id: string, conn: Database.Database): Promise<DailyBudget> {
let b = budgets.get(id);
if (!b) {
const now = Date.now();
@@ -64,6 +64,6 @@ export function budgetFor(id: string, conn: Database.Database): DailyBudget {
}
/** Today's broker usage for one plugin (admin view). Seeds the counter if unseen. */
export function pluginBudgetUsage(id: string, conn: Database.Database): ReturnType<DailyBudget['used']> {
return budgetFor(id, conn).used(Date.now());
export async function pluginBudgetUsage(id: string, conn: Database.Database): Promise<ReturnType<DailyBudget['used']>> {
return (await budgetFor(id, conn)).used(Date.now());
}
server/src/nest/plugins/host/plugin-rpc-host.factory.ts
+1 -1
View File
@@ -9,7 +9,7 @@ import { getPluginDataDb } from './plugin-host-state';
/** Routes inter-plugin calls/events; supplied by PluginRuntimeService (owns the supervisor). */
export interface PluginCallRouter {
callPlugin(callerId: string, targetId: string, fn: string, args: unknown, actingUserId: number | undefined): Promise<unknown>;
emitPluginEvent(sourceId: string, event: string, payload: unknown): void;
emitPluginEvent(sourceId: string, event: string, payload: unknown): Promise<void>;
}
/**
server/src/nest/plugins/host/rpc-host.ts
+3 -3
View File
@@ -46,9 +46,9 @@ export interface HostDeps {
* the dependency edge + the target's `provides` allowlist, forwards the acting user. */
callPlugin(targetId: string, fn: string, args: unknown, actingUserId: number | undefined): Promise<unknown>;
/** Publish an event from this host's plugin to its subscribed dependents. */
emitPluginEvent(event: string, payload: unknown): void;
emitPluginEvent(event: string, payload: unknown): Promise<void>;
/** Optional sink for the capability audit log (host-side, hash-chained). */
audit?(entry: { pluginId: string; actingUserId?: number; method: string; resource: string | null; code: string }): void;
audit?(entry: { pluginId: string; actingUserId?: number; method: string; resource: string | null; code: string }): Promise<void>;
}
type Handler = (params: Record<string, unknown>, actingUserId: number | undefined) => unknown;
@@ -98,7 +98,7 @@ export class PluginRpcHost {
// Audit the core-data / broadcast surface (incl. denials) at the boundary.
if (this.deps.audit && isAuditable(req.method)) {
try {
this.deps.audit({
await this.deps.audit({
pluginId: this.pluginId,
actingUserId,
method: req.method,
server/src/nest/plugins/host/rpc-kit/types.ts
+1 -1
View File
@@ -34,7 +34,7 @@ export interface PluginRpcContext {
/** What a plugin may do to its declared dependencies. */
export interface PluginRpcPeers {
call(targetId: string, fn: string, args: unknown, actingUserId: number | undefined): Promise<unknown>;
emit(event: string, payload: unknown): void;
emit(event: string, payload: unknown): Promise<void>;
}
export type PluginRpcHandler = (params: Record<string, unknown>, ctx: PluginRpcContext) => unknown;
server/src/nest/plugins/host/rpc/db.rpc.ts
+4 -4
View File
@@ -48,8 +48,8 @@ export class DbRpc {
/** Same deal: the router checks the emitter's `emits` allowlist. */
@PluginOpenMethod('events.emit')
emitEvent(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
ctx.plugins.emit(str(params.event, 'event'), params.payload);
async emitEvent(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
await ctx.plugins.emit(str(params.event, 'event'), params.payload);
return { ok: true };
}
@@ -60,8 +60,8 @@ export class DbRpc {
* plugin falls back to ctx.config.
*/
@PluginOpenMethod('settings.get')
getSetting(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
async getSetting(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
if (ctx.actingUserId === undefined) return { value: undefined };
return { value: this.settings.readOne(ctx.pluginId, ctx.actingUserId, str(params.key, 'key')) };
return { value: await this.settings.readOne(ctx.pluginId, ctx.actingUserId, str(params.key, 'key')) };
}
}
server/src/nest/plugins/host/rpc/host-surface.rpc.ts
+11 -11
View File
@@ -45,19 +45,19 @@ export class HostSurfaceRpc {
) {}
@PluginMethod('users.getById', { permission: 'db:read:users' })
getUser(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
async getUser(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
// Scoped to people the acting user can actually see (themselves, or someone they
// share a trip with), so a plugin cannot enumerate every account by looping ids.
const id = num(params.id, 'id');
if (ctx.actingUserId === undefined) throw new ForbiddenResource('user reads require an authenticated user context');
if (id !== ctx.actingUserId && !this.sharesATrip(ctx.actingUserId, id)) {
if (id !== ctx.actingUserId && !(await this.sharesATrip(ctx.actingUserId, id))) {
throw new ForbiddenResource(`no access to user ${id}`);
}
return this.db.prepare('SELECT id, username, display_name, avatar FROM users WHERE id = ?').get(id);
}
@PluginMethod('ws.broadcastToTrip', { permission: 'ws:broadcast:trip' })
broadcastToTrip(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
async broadcastToTrip(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
// The TARGET is gated like a read: only a trip room the acting user belongs to.
// Namespacing the event type alone does not cross the membership boundary.
const tripId = num(params.tripId, 'tripId');
@@ -105,7 +105,7 @@ export class HostSurfaceRpc {
throw new ForbiddenResource('the acting user is not a member of that trip');
}
const link = this.safeLink(input.link);
if (!budgetFor(ctx.pluginId, this.db.connection).take('notify', Date.now())) {
if (!(await budgetFor(ctx.pluginId, this.db.connection)).take('notify', Date.now())) {
throw new BadParams('daily notification budget exhausted (resets at UTC midnight)');
}
await this.notifications.send({
@@ -126,7 +126,7 @@ export class HostSurfaceRpc {
const prompt = typeof params.prompt === 'string' ? params.prompt : '';
if (prompt.trim() === '') throw new BadParams('prompt is required');
if (prompt.length > AI_TEXT_MAX) throw new BadParams(`prompt exceeds the ${AI_TEXT_MAX}-char cap`);
this.takeAiBudget(ctx);
await this.takeAiBudget(ctx);
const system = typeof params.system === 'string' ? params.system.slice(0, 4000) : undefined;
const results = await this.runModel(config, {
prompt: system || 'You are a helpful assistant. Reply with a JSON object of the form {"text": "..."} whose "text" field holds your answer.',
@@ -150,7 +150,7 @@ export class HostSurfaceRpc {
if (typeof params.jsonSchema !== 'object' || params.jsonSchema === null) {
throw new BadParams('jsonSchema (an object) is required');
}
this.takeAiBudget(ctx);
await this.takeAiBudget(ctx);
const hint = typeof params.prompt === 'string' ? params.prompt.slice(0, 4000) : '';
const results = await this.runModel(config, {
prompt: hint || 'Extract structured data from the text into the given JSON schema.',
@@ -173,7 +173,7 @@ export class HostSurfaceRpc {
}
@PluginMethod('scheduler.set', { permission: 'jobs:run' })
schedulerSet(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
async schedulerSet(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
const name = str(params.name, 'name');
const dueAt = num(params.dueAt, 'dueAt');
const everyMs = params.everyMs != null ? num(params.everyMs, 'everyMs') : undefined;
@@ -200,7 +200,7 @@ export class HostSurfaceRpc {
}
@PluginMethod('scheduler.cancel', { permission: 'jobs:run' })
schedulerCancel(params: Record<string, unknown>, ctx: PluginRpcContext): unknown {
async schedulerCancel(params: Record<string, unknown>, ctx: PluginRpcContext): Promise<unknown> {
const r = this.db
.prepare('DELETE FROM plugin_scheduled_tasks WHERE plugin_id = ? AND name = ?')
.run(ctx.pluginId, str(params.name, 'name'));
@@ -208,7 +208,7 @@ export class HostSurfaceRpc {
}
/** Two users share a trip when both are owner-or-member of the same one. */
private sharesATrip(actingUserId: number, targetUserId: number): boolean {
private async sharesATrip(actingUserId: number, targetUserId: number): Promise<boolean> {
return !!this.db
.prepare(
`SELECT 1 FROM trips t
@@ -245,8 +245,8 @@ export class HostSurfaceRpc {
}
}
private takeAiBudget(ctx: PluginRpcContext): void {
if (!budgetFor(ctx.pluginId, this.db.connection).take('ai', Date.now())) {
private async takeAiBudget(ctx: PluginRpcContext): Promise<void> {
if (!(await budgetFor(ctx.pluginId, this.db.connection)).take('ai', Date.now())) {
throw new BadParams('daily AI budget exhausted (resets at UTC midnight)');
}
}
server/src/nest/plugins/host/rpc/meta.rpc.ts
+2 -2
View File
@@ -129,7 +129,7 @@ export class MetaRpc {
}
const entityId = num(params.entityId, 'entityId');
if (ctx.actingUserId === undefined) throw new ForbiddenResource('metadata requires an authenticated user context');
const tripId = this.entityTrip(entityType, entityId);
const tripId = await this.entityTrip(entityType, entityId);
if (tripId === undefined || !this.db.canAccessTrip(tripId, ctx.actingUserId)) {
throw new ForbiddenResource(`no access to ${entityType} ${entityId}`);
}
@@ -139,7 +139,7 @@ export class MetaRpc {
return { entityType, entityId };
}
private entityTrip(entityType: string, entityId: number): number | undefined {
private async entityTrip(entityType: string, entityId: number): Promise<number | undefined> {
if (entityType === 'trip') {
return (this.db.prepare('SELECT id FROM trips WHERE id = ?').get(entityId) as { id: number } | undefined)?.id;
}