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
22 changes: 4 additions & 18 deletions src/datasourcePool.js
Original file line number Diff line number Diff line change
@@ -1,30 +1,16 @@
import { configuredPool } from './datasourcePoolCache.js';

export function getActivePool(service, source) {
// Re-read persisted status before consulting the cache, including for stale callers.
source = source && service.database.getDataSource(source.id);
if (!source) {
throw new Error('data source not found');
}
if (Number(source.status) !== 1) {
throw Object.assign(new Error('data source is disabled'), {
code: 'DATASOURCE_DISABLED', statusCode: 403,
});
}
if (!service.pools.has(source.code)) {
service.pools.set(source.code, service.createPool(source));
}
return service.pools.get(source.code);
return configuredPool(service, source);
}

export async function getActivePoolAsync(service, source) {
source = source && (service.database.dataSourceRepository?.getDataSourceAsync
? await service.database.dataSourceRepository.getDataSourceAsync(source.id)
: service.database.getDataSource(source.id));
if (!source) throw new Error('data source not found');
if (Number(source.status) !== 1) {
throw Object.assign(new Error('data source is disabled'), { code: 'DATASOURCE_DISABLED', statusCode: 403 });
}
if (!service.pools.has(source.code)) service.pools.set(source.code, service.createPool(source));
return service.pools.get(source.code);
return configuredPool(service, source);
}

export function fallbackUnlessDisabled(error, fallback) {
Expand Down
31 changes: 31 additions & 0 deletions src/datasourcePoolCache.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
import { isDeepStrictEqual } from 'node:util';

// Bind each pool to the persisted connection settings used to create it.
// Weak keys keep retired pools and their credential snapshots collectible.
const configurations = new WeakMap();

function connectionConfiguration(source) {
return [source.id, source.dbType, source.host, source.port, source.databaseName,
source.username, source.encryptedPassword, source.options];
}

export function configuredPool(service, source) {
if (!source) throw new Error('data source not found');
if (Number(source.status) !== 1) {
throw Object.assign(new Error('data source is disabled'), {
code: 'DATASOURCE_DISABLED', statusCode: 403,
});
}
const configuration = connectionConfiguration(source);
let pool = service.pools.get(source.code);
if (pool && !isDeepStrictEqual(configurations.get(pool), configuration)) {
service.closePool(source.code);
pool = null;
}
if (!pool) {
pool = service.createPool(source);
configurations.set(pool, structuredClone(configuration));
service.pools.set(source.code, pool);
}
return pool;
}
40 changes: 40 additions & 0 deletions tests/datasourceLifecycle.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import assert from 'node:assert/strict';
import test from 'node:test';
import { PlatformDatabase } from '../src/database.js';
import { BusinessDatasetService } from '../src/businessDatasets.js';
import { getActivePoolAsync } from '../src/datasourcePool.js';

const payload = {
code: 'source', name: 'Source', dbType: 'POSTGRESQL', host: 'old.example',
Expand All @@ -19,6 +20,45 @@ function fixture(t) {
return { database, service, closed };
}

for (const [mode, getPool] of [
['sync', (service, source) => service.getPool(source)],
['async', (service, source) => {
// The in-memory fixture has separate native/async connections; expose its
// real persisted rows through the asynchronous repository contract.
service.database.dataSourceRepository.getDataSourceAsync = async id => service.database.getDataSource(id);
return getActivePoolAsync(service, source);
}],
]) {
for (const [field, value] of Object.entries({
dbType: 'MYSQL', host: 'new.example', port: 5433, databaseName: 'other',
username: 'other', encryptedPassword: 'new-ciphertext', options: { ssl: false },
})) {
test(`${mode} pool refreshes after another writer changes datasource ${field}`, async t => {
const { database, service, closed } = fixture(t);
service.createPool = source => ({ source, end: async () => { closed.push(source.host); } });
const original = database.saveDataSource(payload);
const oldPool = await getPool(service, original);
database.saveDataSource({ ...payload, [field]: value }, original.id);
const refreshed = await getPool(service, original);
assert.notEqual(refreshed, oldPool);
assert.deepEqual(refreshed.source[field], value);
assert.deepEqual(closed, ['old.example']);
assert.equal(await getPool(service, original), refreshed);
assert.equal(service.pools.size, 1);
});
}

test(`${mode} pool survives non-connection edits from another writer`, async t => {
const { database, service, closed } = fixture(t);
const original = database.saveDataSource(payload);
const pool = await getPool(service, original);
database.updateDataSourceTest(original.id, { ok: true, message: 'passed' });
database.saveDataSource({ ...payload, name: 'Renamed', options: { ca: 'test-ca', ssl: true } }, original.id);
assert.equal(await getPool(service, original), pool);
assert.deepEqual(closed, []);
});
}

test('changing a datasource code releases its original connection pool', t => {
const { service, closed } = fixture(t);
const original = service.saveDataSource(payload);
Expand Down
Loading