fix(datasources): block queries through disabled sources

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,

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,

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;
+}

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);
+});