diff --git a/src/businessDatasets.js b/src/businessDatasets.js index 2263911..f1967a9 100644 --- a/src/businessDatasets.js +++ b/src/businessDatasets.js @@ -1,4 +1,5 @@ import { DatasourceCrypto } from './datasourceCrypto.js'; +import { fallbackUnlessDisabled, getActivePool } from './datasourcePool.js'; import { getDialect, normalizeDataSource } from './dialects/index.js'; import { classifyColumnType as normalizeDataType } from './dialects/common.js'; import { @@ -175,10 +176,7 @@ export class BusinessDatasetService { } getPool(source) { - if (!this.pools.has(source.code)) { - this.pools.set(source.code, this.createPool(source)); - } - return this.pools.get(source.code); + return getActivePool(this, source); } closePool(code) { @@ -357,11 +355,14 @@ export class BusinessDatasetService { try { sample = await this.sampleForProfiling(dataset, fields, sampleSize); } catch (error) { - degraded = `未能读取样本数据,已按字段名与类型降级识别:${error.message}`; + degraded = fallbackUnlessDisabled( + error, + `未能读取样本数据,已按字段名与类型降级识别:${error.message}`, + ); } const ranges = degraded ? {} - : await this.listFieldRanges(dataset, fields).catch(() => ({})); + : await this.listFieldRanges(dataset, fields).catch(error => fallbackUnlessDisabled(error, {})); return { ...buildDatasetProfile({ datasetId: dataset.id, diff --git a/src/database.js b/src/database.js index 5b4c217..a7a490b 100644 --- a/src/database.js +++ b/src/database.js @@ -2198,7 +2198,7 @@ export class PlatformDatabase { ) END AS canQuery FROM business_datasets d JOIN data_sources s ON s.id = d.datasource_id - WHERE (? = 1 OR d.status = 1) + WHERE (? = 1 OR (d.status = 1 AND s.status = 1)) ORDER BY d.id `).all( userId ? Number(userId) : null, diff --git a/src/datasourcePool.js b/src/datasourcePool.js new file mode 100644 index 0000000..1dbfcdb --- /dev/null +++ b/src/datasourcePool.js @@ -0,0 +1,23 @@ +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); +} + +export function fallbackUnlessDisabled(error, fallback) { + if (error.code === 'DATASOURCE_DISABLED') { + throw error; + } + return fallback; +} diff --git a/tests/disabledDatasource.test.js b/tests/disabledDatasource.test.js new file mode 100644 index 0000000..ccf74d6 --- /dev/null +++ b/tests/disabledDatasource.test.js @@ -0,0 +1,130 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { BusinessDatasetService } from '../src/businessDatasets.js'; + +function fixture(t, status = 0) { + const database = new PlatformDatabase(':memory:'); + const service = new BusinessDatasetService({ database, crypto: {}, config: {} }); + const source = database.saveDataSource({ + code: 'guarded', name: 'Guarded', dbType: 'MYSQL', host: 'never-connect.invalid', + port: 3306, databaseName: 'analytics', username: 'reader', status, + }); + const dataset = database.saveBusinessDataset({ + code: 'sales', name: 'Sales', datasourceId: source.id, + schemaName: 'analytics', primaryTable: 'sales', status: 1, + }); + database.replaceDatasetFields(dataset.id, [{ + fieldName: 'amount', displayName: 'Amount', dataType: 'decimal', + semanticType: 'NUMBER', role: 'METRIC', aggregator: 'SUM', enabled: true, + }]); + const calls = { created: 0, queried: 0, ended: 0 }; + service.createPool = () => { + calls.created += 1; + return { + query: async () => { + calls.queried += 1; + return [[{ amount: 42 }], [{ name: 'amount' }]]; + }, + end: async () => { calls.ended += 1; }, + }; + }; + t.after(async () => { await service.close(); database.close(); }); + const execute = () => service.executeQuery({ + datasetId: dataset.id, request: { metrics: ['amount'] }, + }); + const setStatus = value => database.saveDataSource({ ...source, status: value }, source.id); + return { database, service, source, dataset, calls, execute, setStatus }; +} + +function disabled(error) { + assert.equal(error.statusCode, 403); + assert.match(error.message, /data source.*disabled/i); + return true; +} + +test('disabled datasource rejects business queries before creating a connection', async t => { + const { execute, calls } = fixture(t); + await assert.rejects(execute, disabled); + assert.equal(calls.created, 0); + assert.equal(calls.queried, 0); +}); + +test('disabling a cached datasource denies the next query and re-enabling restores it', async t => { + const { service, source, calls, execute, setStatus } = fixture(t, 1); + assert.deepEqual((await execute()).rows, [{ amount: 42 }]); + setStatus(0); + await assert.rejects(execute, disabled); + assert.throws(() => service.getPool(source), disabled); + assert.equal(calls.created, 1); + assert.equal(calls.queried, 1); + setStatus(1); + assert.deepEqual((await execute()).rows, [{ amount: 42 }]); + assert.equal(calls.queried, 2); +}); + +for (const [name, operation] of [ + ['schemas', f => f.service.listDatabases(f.source.id)], + ['tables', f => f.service.listTables(f.source.id, 'analytics')], + ['columns', f => f.service.listColumns(f.source.id, 'analytics', 'sales')], + ['samples', f => f.service.sampleDataset(f.dataset.id)], + ['distinct values', f => f.service.listDistinctFieldValues(f.dataset.id, 'amount')], + ['profiling', f => f.service.profileDataset(f.dataset.id)], +]) { + test(`disabled datasource denies ${name} without connecting`, async t => { + const f = fixture(t); + await assert.rejects(() => operation(f), disabled); + assert.equal(f.calls.created, 0); + assert.equal(f.calls.queried, 0); + }); +} + +test('normal dataset discovery hides disabled sources while admin discovery retains them', t => { + const { database, service, source, dataset, setStatus } = fixture(t); + const user = database.getUserByUsername('east_manager'); + database.grantDatasetAccess(user.id, dataset.id, true); + assert.deepEqual(service.listDatasets(), []); + assert.deepEqual(service.listDatasets({ userId: user.id }), []); + assert.equal(service.listDatasets({ includeDisabled: true })[0].id, dataset.id); + assert.equal(service.listDataSources().find(item => item.id === source.id).status, 0); + setStatus(1); + assert.equal(service.listDatasets({ userId: user.id })[0].id, dataset.id); +}); + +test('admin connection tests remain available for disabled sources', async t => { + const { service, source, calls } = fixture(t); + const result = await service.testDataSource(source.id); + assert.equal(result.lastTestOk, true); + assert.equal(result.status, 0); + assert.deepEqual(calls, { created: 1, queried: 1, ended: 1 }); +}); + +test('profiling rejects disablement during sampling instead of hiding the range-query denial', async t => { + const { database, service, dataset, calls, setStatus } = fixture(t, 1); + database.replaceDatasetFields(dataset.id, [{ + fieldName: 'ordered_at', displayName: 'Order date', dataType: 'date', + semanticType: 'DATE', role: 'TIME', enabled: true, + }]); + const createPool = service.createPool; + let releaseSample; + let samplingStarted; + const started = new Promise(resolve => { samplingStarted = resolve; }); + service.createPool = source => { + const pool = createPool(source); + return { + ...pool, + query: async (...args) => { + const waiting = new Promise(resolve => { releaseSample = resolve; }); + samplingStarted(); + await waiting; + return pool.query(...args); + }, + }; + }; + const profiling = service.profileDataset(dataset.id); + await started; + setStatus(0); + releaseSample(); + await assert.rejects(profiling, disabled); + assert.equal(calls.queried, 1); +});