Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -229,3 +229,62 @@ describe('DatasourceAdminServicePlugin: persistence + bound count', () => {
await expect(service.removeDatasource('reporting')).rejects.toThrow(/1 object\(s\)/);
});
});

describe('DatasourceAdminServicePlugin: runtime datasource durability', () => {
/** In-memory `sys_metadata` engine shared across two boots (a "restart"). */
function fakeSysMetadataEngine() {
const rows: Array<Record<string, unknown>> = [];
return {
rows,
registerDriver() {},
registerDatasourceDef() {},
getDriverByName() { return undefined; },
findOne: async (_o: string, q: { where?: Record<string, unknown> }) => {
const w = q.where ?? {};
return rows.find((r) => Object.entries(w).every(([k, v]) => r[k] === v));
},
find: async (_o: string, q: { where?: Record<string, unknown> }) => {
const w = q.where ?? {};
return rows.filter((r) => Object.entries(w).every(([k, v]) => r[k] === v));
},
insert: async (_o: string, row: Record<string, unknown>) => { rows.push({ ...row }); },
update: async (_o: string, row: Record<string, unknown>, opts: { where: Record<string, unknown> }) => {
const i = rows.findIndex((r) => r.id === opts.where.id);
if (i >= 0) rows[i] = { ...rows[i], ...row };
},
delete: async (_o: string, opts: { where: Record<string, unknown> }) => {
const i = rows.findIndex((r) => r.id === opts.where.id);
if (i >= 0) rows.splice(i, 1);
},
};
}

it('persists a UI-created datasource to sys_metadata and restores it after a restart', async () => {
const data = fakeSysMetadataEngine();

// Boot #1: create a runtime sqlite datasource (no secret needed).
const b1 = await boot({ services: { data } });
await b1.service.createDatasource({ name: 'demo_ext', driver: 'sqlite', config: { filename: '/tmp/x.db' } });
// It is durably written to sys_metadata (not just the in-memory registry).
expect(data.rows.filter((r) => r.type === 'datasource' && r.name === 'demo_ext')).toHaveLength(1);

// Boot #2 = "restart": fresh in-memory registry, SAME sys_metadata engine.
const b2 = await boot({ services: { data } });
// Before restore, the fresh registry is empty.
expect(await b2.service.listDatasources()).toHaveLength(0);
// start() restores runtime rows from sys_metadata into the registry.
await b2.plugin.start(b2.ctx);
const after = await b2.service.listDatasources();
expect(after.map((d) => d.name)).toContain('demo_ext');
expect(after.find((d) => d.name === 'demo_ext')?.origin).toBe('runtime');
});

it('removes the durable sys_metadata row when a datasource is deleted', async () => {
const data = fakeSysMetadataEngine();
const b = await boot({ services: { data } });
await b.service.createDatasource({ name: 'gone', driver: 'sqlite', config: { filename: '/tmp/y.db' } });
expect(data.rows.some((r) => r.name === 'gone')).toBe(true);
await b.service.removeDatasource('gone');
expect(data.rows.some((r) => r.name === 'gone')).toBe(false);
});
});
116 changes: 113 additions & 3 deletions packages/services/service-datasource/src/datasource-admin-plugin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,82 @@ interface DataEngineLike {
registerDriver?: (driver: unknown, isDefault?: boolean) => void;
registerDatasourceDef?: (def: { name: string; schemaMode?: string; external?: { allowWrites?: boolean } }) => void;
getDriverByName?: (name: string) => unknown;
// sys_metadata CRUD used to persist runtime datasource records durably (same
// table runtime objects use). Optional — absent on lightweight kernels, in
// which case persistence degrades to in-memory (pre-existing behavior).
findOne?: (object: string, query: { where?: Record<string, unknown> }) => Promise<Record<string, unknown> | undefined | null>;
find?: (object: string, query: { where?: Record<string, unknown> }) => Promise<Record<string, unknown>[]>;
insert?: (object: string, row: Record<string, unknown>) => Promise<unknown>;
update?: (object: string, row: Record<string, unknown>, opts: { where: Record<string, unknown> }) => Promise<unknown>;
delete?: (object: string, opts: { where: Record<string, unknown> }) => Promise<unknown>;
}

/**
* Durable persistence for runtime datasource records via the `sys_metadata`
* table — the same store runtime objects use (the protocol writes objects there
* directly). `MetadataManager.register()` alone is in-memory unless a writable
* `datasource:` loader is wired, which standalone `serve` does not do; so a
* UI-created datasource vanished on restart. These helpers persist on write and
* the plugin restores them into the registry on boot before rehydrating pools.
* Credential cleartext is never stored — only the opaque `external.credentialsRef`.
*/
const DS_META_TYPE = 'datasource';
const SYS_METADATA = 'sys_metadata';

function newMetaId(): string {
return typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function'
? crypto.randomUUID()
: `meta_${Date.now()}_${Math.random().toString(36).slice(2)}`;
}

async function persistDatasourceRow(engine: DataEngineLike | undefined, record: { name: string }): Promise<void> {
if (!engine?.insert || !engine.findOne) return; // no durable store — in-memory only
const now = new Date().toISOString();
const existing = await engine.findOne(SYS_METADATA, {
where: { type: DS_META_TYPE, name: record.name, state: 'active' },
});
if (existing) {
await engine.update?.(
SYS_METADATA,
{ metadata: JSON.stringify(record), updated_at: now, version: ((existing.version as number) || 0) + 1, state: 'active' },
{ where: { id: existing.id } },
);
} else {
await engine.insert(SYS_METADATA, {
id: newMetaId(),
name: record.name,
type: DS_META_TYPE,
scope: 'platform',
metadata: JSON.stringify(record),
state: 'active',
version: 1,
created_at: now,
updated_at: now,
});
}
}

async function deleteDatasourceRow(engine: DataEngineLike | undefined, name: string): Promise<void> {
if (!engine?.findOne) return;
const existing = await engine.findOne(SYS_METADATA, { where: { type: DS_META_TYPE, name, state: 'active' } });
if (!existing) return;
if (engine.delete) await engine.delete(SYS_METADATA, { where: { id: existing.id } });
else await engine.update?.(SYS_METADATA, { state: 'inactive' }, { where: { id: existing.id } });
}

async function loadDatasourceRows(engine: DataEngineLike | undefined): Promise<Array<Record<string, unknown>>> {
if (!engine?.find) return [];
const rows = await engine.find(SYS_METADATA, { where: { type: DS_META_TYPE, state: 'active' } });
const out: Array<Record<string, unknown>> = [];
for (const r of rows ?? []) {
const raw = (r as { metadata?: unknown }).metadata;
try {
out.push(typeof raw === 'string' ? JSON.parse(raw) : (raw as Record<string, unknown>));
} catch {
/* skip corrupt row */
}
}
return out;
}

/**
Expand Down Expand Up @@ -147,7 +223,10 @@ export class DatasourceAdminServicePlugin implements Plugin {
if (!metadata?.register) {
throw new Error('Metadata service is unavailable; cannot persist datasource.');
}
// In-memory registry (immediate visibility) + durable sys_metadata row
// (survives restart; restored on boot by restoreRuntimeDatasources).
await metadata.register('datasource', record.name, record);
await persistDatasourceRow(engineOf(), record);
},

deleteDatasourceRecord: async (name) => {
Expand All @@ -156,6 +235,7 @@ export class DatasourceAdminServicePlugin implements Plugin {
throw new Error('Metadata service is unavailable; cannot remove datasource.');
}
await metadata.unregister('datasource', name);
await deleteDatasourceRow(engineOf(), name);
},

writeSecret: async (input, hint) => {
Expand Down Expand Up @@ -265,13 +345,43 @@ export class DatasourceAdminServicePlugin implements Plugin {
}

async start(ctx: PluginContext): Promise<void> {
// Rebuild live connection pools for persisted runtime datasources before
// announcing readiness — a node restart otherwise leaves UI-created
// datasources with a record but no open pool until the next write.
// Restore UI-created (runtime) datasources from the durable sys_metadata
// store back into the in-memory registry, THEN rebuild their live pools.
// `register()` is in-memory only in standalone serve (no writable
// `datasource:` loader), so without this a node restart drops every
// UI-created datasource. Code-defined datasources come from the artifact and
// are unaffected.
await this.restoreRuntimeDatasources(ctx);
await this.rehydratePools();
if (this.service) await ctx.trigger('datasource-admin:ready', this.service);
}

/** Reload persisted runtime datasource rows (sys_metadata) into the registry. */
private async restoreRuntimeDatasources(ctx: PluginContext): Promise<void> {
const engine = safeGetService<DataEngineLike>(ctx, 'data');
const metadata = safeGetService<MetadataServiceLike>(ctx, 'metadata');
if (!engine?.find || !metadata?.register) return;
let rows: Array<Record<string, unknown>>;
try {
rows = await loadDatasourceRows(engine);
} catch (err) {
this.options.logger?.warn?.('datasource restore: reading sys_metadata failed', err);
return;
}
let restored = 0;
for (const rec of rows) {
const name = (rec as { name?: string }).name;
if (!name) continue;
try {
await metadata.register('datasource', name, rec);
restored += 1;
} catch (err) {
this.options.logger?.warn?.(`datasource restore: register '${name}' failed`, err);
}
}
if (restored > 0) this.options.logger?.info?.(`datasource: restored ${restored} runtime record(s) from sys_metadata`);
}

/**
* Boot-time rehydration: list persisted runtime datasources and re-register
* each one's connection pool (driver build → connect → registerDriver),
Expand Down
Loading