feat: observe BI data source health

docs/bi-refactor-roadmap.md

@@ -172,7 +172,7 @@ SUM(fee) / NULLIF(SUM(kwh), 0)
 - 轮换已经进入版本历史的氢能数据库凭据。
 - 已提供无敏感值的 `.env.example`；测试、生产值由各环境密钥管理。
 - 已拆分进程存活检查与无敏感值的配置就绪检查。
-- 增加数据源连通性、最后成功时间和错误率监控。
+- 已增加基于真实业务请求的数据源最后结果、最后成功时间和滚动失败率监控。
 
 ## 9. 每批验收门槛
 

src/server/auth/middleware.ts

@@ -31,7 +31,11 @@ export async function authMiddleware(c: Context, next: Next) {
   }
 
   // 跳过不需要认证的路径
-  if (path.startsWith('/api/health') || path.startsWith('/api/auth/')) {
+  if (
+    path === '/api/health'
+    || path === '/api/health/readiness'
+    || path.startsWith('/api/auth/')
+  ) {
     return next();
   }
 

src/server/index.ts

@@ -40,6 +40,8 @@ app.route('/api/vehicle-heatmap', vehicleHeatmapRouter);
 app.route('/api/hydrogen-heatmap', hydrogenHeatmapRouter);
 app.route('/api/analytics', analyticsRouter);
 
+app.all('/api/*', (c) => c.json({ error: 'API route not found' }, 404));
+
 // Serve static files in production
 app.use('/*', serveStatic({ root: './dist' }));
 app.use('/*', serveStatic({ root: './dist', path: 'index.html' }));

src/server/routes/analytics/index.test.ts

@@ -50,3 +50,13 @@ test('rejects unpublished metric domains', async () => {
 
   assert.equal(response.status, 400);
 });
+
+test('returns observed data-source outcomes without sensitive details', async () => {
+  const response = await app.request('/data-sources');
+  const payload = await response.json();
+
+  assert.equal(response.status, 200);
+  assert.equal(payload.semantics, 'observed-requests');
+  assert.ok(Array.isArray(payload.sources));
+  assert.equal(JSON.stringify(payload).includes('password'), false);
+});

src/server/routes/analytics/index.ts

@@ -5,6 +5,7 @@ import {
   listMetricDefinitions,
   type MetricDomain,
 } from '../../../shared/analytics/catalog.js';
+import { getDataSourceTelemetry } from '../../source-telemetry.js';
 
 const app = new Hono();
 
@@ -26,4 +27,6 @@ app.get('/metrics', (c) => {
   });
 });
 
+app.get('/data-sources', (c) => c.json(getDataSourceTelemetry()));
+
 export default app;

src/server/routes/energy/cache.ts

@@ -1,3 +1,5 @@
+import { observeDataSource, type DataSourceId } from '../../source-telemetry.js';
+
 /**
  * SWR 缓存：始终返回热数据，后台定时刷新。
  *
@@ -70,24 +72,28 @@ async function runRefresh(key: string) {
 
 export interface CachedOpts {
   force?: boolean;
+  source?: DataSourceId;
 }
 
 export async function cached<T>(key: string, loader: () => Promise<T>, opts: CachedOpts = {}): Promise<T> {
+  const effectiveLoader = opts.source
+    ? () => observeDataSource(opts.source!, loader)
+    : loader;
   const now = Date.now();
   const hit = cache.get(key) as Entry<T> | undefined;
   if (hit) {
     hit.lastAccess = now;
-    hit.loader = loader;
+    hit.loader = effectiveLoader;
   }
 
   // 强制刷新：等待 loader 完成
   if (opts.force) {
     const ongoing = inflight.get(key) as Promise<T> | undefined;
     if (ongoing) return ongoing;
-    const p = loader()
+    const p = effectiveLoader()
       .then(value => {
         const t = Date.now();
-        const next: Entry<T> = { value, freshAt: t, expiresAt: t + TTL_MS, loader, lastAccess: t };
+        const next: Entry<T> = { value, freshAt: t, expiresAt: t + TTL_MS, loader: effectiveLoader, lastAccess: t };
         cache.set(key, next);
         scheduleRefresh(key, next);
         return value;
@@ -112,10 +118,10 @@ export async function cached<T>(key: string, loader: () => Promise<T>, opts: Cac
   const ongoing = inflight.get(key) as Promise<T> | undefined;
   if (ongoing) return ongoing;
 
-  const p = loader()
+  const p = effectiveLoader()
     .then(value => {
       const t = Date.now();
-      const entry: Entry<T> = { value, freshAt: t, expiresAt: t + TTL_MS, loader, lastAccess: t };
+      const entry: Entry<T> = { value, freshAt: t, expiresAt: t + TTL_MS, loader: effectiveLoader, lastAccess: t };
       cache.set(key, entry);
       scheduleRefresh(key, entry);
       return value;

src/server/routes/energy/index.ts

@@ -389,7 +389,7 @@ app.get('/hydrogen/overview', async (c) => {
   }));
 
   return { kpi, top5, regions, monthly, customers, stations, availableYears, year };
-  }, { force });
+  }, { force, source: 'hydrogenDatabase' });
   return c.json(data);
 });
 
@@ -530,7 +530,7 @@ app.get('/hydrogen/daily', async (c) => {
   // 按日期降序返回
   const result = ascDays.slice().sort((a, b) => b.date.localeCompare(a.date));
   return result;
-  }, { force });
+  }, { force, source: 'hydrogenDatabase' });
   return c.json(data);
 });
 
@@ -615,7 +615,7 @@ app.get('/electric/overview', async (c) => {
     trend: trendArr,
     latestRecordAt,
   };
-  }, { force });
+  }, { force, source: 'electricDatabase' });
   return c.json(data);
 });
 
@@ -688,7 +688,7 @@ app.get('/electric/orders', async (c) => {
         totalFee: Math.round((Number(row.totalFee) || 0) * 100) / 100,
       })),
     };
-  });
+  }, { source: 'electricDatabase' });
 
   return c.json(data);
 });
@@ -779,7 +779,7 @@ app.get('/electric/monthly', async (c) => {
     });
 
   return months;
-  }, { force });
+  }, { force, source: 'electricDatabase' });
   return c.json(data);
 });
 
@@ -862,7 +862,7 @@ app.get('/etc/overview', async (c) => {
         latestTransactionTime: record.latestTransactionTime ? String(record.latestTransactionTime) : null,
       },
     } as const;
-  }, { force });
+  }, { force, source: 'etcDatabase' });
 
   return c.json(data);
 });

src/server/routes/mileage/oneos-api.ts

@@ -4,6 +4,7 @@ import {
   normalizeOneOsProtocol,
   type OneOsProtocol,
 } from './source-policy.js';
+import { observeDataSource } from '../../source-telemetry.js';
 
 dotenv.config();
 
@@ -313,7 +314,7 @@ export async function fetchOneOsDailyMileage(
   } else {
     let pending = inflight.get(cacheKey);
     if (!pending) {
-      pending = requestDate(date, protocolPriority).then(result => {
+      pending = observeDataSource('oneOsMileage', () => requestDate(date, protocolPriority)).then(result => {
         const ttl = date === shanghaiDate() ? CURRENT_DAY_TTL_MS : HISTORICAL_TTL_MS;
         cache.delete(cacheKey);
         if (ttl > 0) {
@@ -372,12 +373,12 @@ export async function fetchOneOsMileageDates(
     ].join(':');
     let pending = rangeInflight.get(key);
     if (!pending) {
-      pending = requestRange(
+      pending = observeDataSource('oneOsMileage', () => requestRange(
-        startDate,
+          startDate,
-        endDate,
+          endDate,
-        selectedPlates,
+          selectedPlates,
-        protocolPriority,
+          protocolPriority,
-      ).finally(() => rangeInflight.delete(key));
+        )).finally(() => rangeInflight.delete(key));
       rangeInflight.set(key, pending);
     }
     const rowsByDate = await pending;

src/server/source-telemetry.test.ts

@@ -0,0 +1,58 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  clearDataSourceTelemetry,
+  getDataSourceTelemetry,
+  observeDataSource,
+  recordDataSourceOutcome,
+} from './source-telemetry.js';
+
+test('reports unobserved sources without treating them as failures', () => {
+  clearDataSourceTelemetry();
+  const report = getDataSourceTelemetry(new Date('2026-08-07T08:00:00Z'));
+  assert.ok(report.sources.every(source => source.state === 'unobserved'));
+  assert.ok(report.sources.every(source => source.failureRate === null));
+});
+
+test('publishes rolling outcomes and last success/failure timestamps', () => {
+  clearDataSourceTelemetry();
+  recordDataSourceOutcome('oneOsMileage', true, 1_000);
+  recordDataSourceOutcome('oneOsMileage', false, 2_000);
+  const source = getDataSourceTelemetry().sources.find(item => item.source === 'oneOsMileage');
+  assert.deepEqual(source && {
+    state: source.state,
+    attempts: source.attempts,
+    failures: source.failures,
+    failureRate: source.failureRate,
+    lastSuccessAt: source.lastSuccessAt,
+    lastFailureAt: source.lastFailureAt,
+  }, {
+    state: 'failing',
+    attempts: 2,
+    failures: 1,
+    failureRate: 0.5,
+    lastSuccessAt: '1970-01-01T00:00:01.000Z',
+    lastFailureAt: '1970-01-01T00:00:02.000Z',
+  });
+});
+
+test('observes final operation outcomes without retaining error details', async () => {
+  clearDataSourceTelemetry();
+  await assert.rejects(observeDataSource('hydrogenDatabase', async () => {
+    throw new Error('sensitive connection detail');
+  }));
+  const serialized = JSON.stringify(getDataSourceTelemetry());
+  assert.equal(serialized.includes('sensitive connection detail'), false);
+});
+
+test('keeps only the latest 100 outcomes', () => {
+  clearDataSourceTelemetry();
+  recordDataSourceOutcome('etcDatabase', false, 1);
+  for (let index = 0; index < 100; index += 1) {
+    recordDataSourceOutcome('etcDatabase', true, index + 2);
+  }
+  const source = getDataSourceTelemetry().sources.find(item => item.source === 'etcDatabase');
+  assert.equal(source?.attempts, 100);
+  assert.equal(source?.failures, 0);
+  assert.equal(source?.failureRate, 0);
+});

src/server/source-telemetry.ts

@@ -0,0 +1,78 @@
+export const DATA_SOURCE_IDS = [
+  'oneOsMileage',
+  'hydrogenDatabase',
+  'electricDatabase',
+  'etcDatabase',
+] as const;
+
+export type DataSourceId = typeof DATA_SOURCE_IDS[number];
+
+interface Outcome {
+  success: boolean;
+  at: number;
+}
+
+const MAX_OUTCOMES = 100;
+const outcomes = new Map<DataSourceId, Outcome[]>();
+const startedAt = new Date().toISOString();
+
+function lastMatching(entries: Outcome[], success: boolean): Outcome | undefined {
+  for (let index = entries.length - 1; index >= 0; index -= 1) {
+    if (entries[index].success === success) return entries[index];
+  }
+  return undefined;
+}
+
+export function recordDataSourceOutcome(
+  source: DataSourceId,
+  success: boolean,
+  at = Date.now(),
+): void {
+  const entries = outcomes.get(source) || [];
+  entries.push({ success, at });
+  if (entries.length > MAX_OUTCOMES) entries.splice(0, entries.length - MAX_OUTCOMES);
+  outcomes.set(source, entries);
+}
+
+export async function observeDataSource<T>(
+  source: DataSourceId,
+  operation: () => Promise<T>,
+): Promise<T> {
+  try {
+    const result = await operation();
+    recordDataSourceOutcome(source, true);
+    return result;
+  } catch (error) {
+    recordDataSourceOutcome(source, false);
+    throw error;
+  }
+}
+
+export function getDataSourceTelemetry(now = new Date()) {
+  return {
+    semantics: 'observed-requests' as const,
+    windowSize: MAX_OUTCOMES,
+    processStartedAt: startedAt,
+    checkedAt: now.toISOString(),
+    sources: DATA_SOURCE_IDS.map(source => {
+      const entries = outcomes.get(source) || [];
+      const failures = entries.filter(entry => !entry.success).length;
+      const last = entries.at(-1);
+      const lastSuccess = lastMatching(entries, true);
+      const lastFailure = lastMatching(entries, false);
+      return {
+        source,
+        state: !last ? 'unobserved' as const : last.success ? 'available' as const : 'failing' as const,
+        attempts: entries.length,
+        failures,
+        failureRate: entries.length > 0 ? Math.round(failures / entries.length * 10_000) / 10_000 : null,
+        lastSuccessAt: lastSuccess ? new Date(lastSuccess.at).toISOString() : null,
+        lastFailureAt: lastFailure ? new Date(lastFailure.at).toISOString() : null,
+      };
+    }),
+  };
+}
+
+export function clearDataSourceTelemetry(): void {
+  outcomes.clear();
+}