feat(assets): add verified asset history trends and compact overview
This commit is contained in:
shishengliang
@@ -0,0 +1,32 @@
|
||||
// Synthetic fixtures imported only by tests and the explicitly isolated QA harness.
|
||||
import type { Vehicle, VehicleRow } from '../../types.js';
|
||||
import { counts, freezeLive, type HistoryBatch, type HistorySettings, type HistoryVehicle } from './model.js';
|
||||
import { monthCutoff, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
||||
import type { HistoryReader } from './store.js';
|
||||
|
||||
export function testVehicle(id: string, overrides: Partial<Vehicle> = {}): Vehicle {
|
||||
return { id, plateNumber: `测试-${id}`, vin: `TESTVIN${id}`, type: '18T', model: '测试18T厢货', color: '白色',
|
||||
location: '嘉兴', region: '嘉兴', province: '浙江省', city: '嘉兴市', status: 'Operating', operationStatus: '1', vehicleStatus: '5',
|
||||
ownership: 'Self', rentCompany: '', contractNo: '测试合同', customerName: '测试客户公司', orgName: '业务公司',
|
||||
departmentName: '部门甲', subjectOrg: '公司甲', projectName: '测试项目', customerManager: '负责人甲', managerId: 'manager-a', brandLabel: '测试品牌', ...overrides };
|
||||
}
|
||||
export function testRow(id: string, operation = '1'): VehicleRow {
|
||||
return { id, 车牌号: `测试-${id}`, vin: `TESTVIN${id}`, 车辆租赁状态: '5', 车辆租赁状态Label: operation,
|
||||
主体: '公司甲', 合同归属部门: '部门甲', 经理ID: 'manager-a', 车辆型号Label: '18吨厢式货车' } as VehicleRow;
|
||||
}
|
||||
export function testBatch(month: string, vehicles: Vehicle[] = [], overrides: Partial<HistoryBatch> = {}): HistoryBatch {
|
||||
return { id: `batch-${month}`, month, status: 'COMPLETE', source: 'AUTO', scheduledFor: shanghaiTimestamp(monthCutoff(month)),
|
||||
capturedAt: shanghaiTimestamp(monthCutoff(month)), counts: counts(vehicles), ...overrides };
|
||||
}
|
||||
export class MemoryHistoryReader implements HistoryReader {
|
||||
settingsValue: HistorySettings | null = { enabledAt: '2026-08-01T00:00:00+08:00' };
|
||||
catalog: HistoryBatch[] = [];
|
||||
rows = new Map<string, HistoryVehicle[]>();
|
||||
calls: string[] = [];
|
||||
async settings() { this.calls.push('settings'); return this.settingsValue; }
|
||||
async batches() { this.calls.push('batches'); return this.catalog; }
|
||||
async vehicles(ids: string[]) { this.calls.push('vehicles'); return new Map(ids.map(id => [id, this.rows.get(id) || []])); }
|
||||
add(month: string, vehicles: Vehicle[], overrides: Partial<HistoryBatch> = {}) {
|
||||
const batch = testBatch(month, vehicles, overrides); this.catalog.push(batch); this.rows.set(batch.id, freezeLive(vehicles)); return batch;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import { Hono } from 'hono';
|
||||
import type { AuthUser } from '../../auth/types.js';
|
||||
import pool from '../../db.js';
|
||||
import { vehicleRepository } from '../vehicles/repository.js';
|
||||
import { MysqlHistoryStore } from './store.js';
|
||||
import { HistoryService, HistoryInputError, HistoryPeriodUnavailable } from './service.js';
|
||||
|
||||
export const historyStore = new MysqlHistoryStore(pool);
|
||||
// Published routes expose live data and verified snapshots only.
|
||||
// The retired handover reconstruction remains an offline audit artifact.
|
||||
export const historyService = new HistoryService(historyStore, () => vehicleRepository.getVehicles());
|
||||
export function createHistoryRouter(service: Pick<HistoryService, 'trend' | 'details' | 'livePoint'> = historyService) {
|
||||
const router = new Hono<{ Variables: { user: AuthUser } }>();
|
||||
router.use('*', async (c, next) => {
|
||||
c.header('Cache-Control', 'private, no-store');
|
||||
if (!c.get('user')) return c.json({ error: 'Unauthorized' }, 401);
|
||||
await next();
|
||||
});
|
||||
router.get('/trend', async c => c.json(await service.trend(c.get('user'), c.req.query())));
|
||||
router.get('/live', async c => c.json(await service.livePoint(c.get('user'), c.req.query())));
|
||||
router.get('/details', async c => c.json(await service.details(c.get('user'), c.req.query())));
|
||||
router.onError((error, c) => {
|
||||
if (error instanceof HistoryInputError) return c.json({ error: error.message }, 400);
|
||||
if (error instanceof HistoryPeriodUnavailable) return c.json({ error: error.message }, 409);
|
||||
console.error('[asset-history] read failed');
|
||||
return c.json({ error: '资产历史数据暂不可用，请稍后重试' }, 503);
|
||||
});
|
||||
return router;
|
||||
}
|
||||
export default createHistoryRouter();
|
||||
@@ -0,0 +1,42 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { execFileSync } from 'node:child_process';
|
||||
import { monthCutoff, nextCapture, shanghaiMonth, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
||||
import { counts, freezeRows, makePoint, freezeLive } from './model.js';
|
||||
import { testBatch, testRow, testVehicle } from './fixtures.js';
|
||||
|
||||
test('Shanghai month-end cutoff covers leap years, short/long months and cross-year boundaries', () => {
|
||||
for (const [month, expected] of [['2024-02','2024-02-29'], ['2026-02','2026-02-28'], ['2026-04','2026-04-30'], ['2026-10','2026-10-31'], ['2026-12','2026-12-31']]) {
|
||||
assert.equal(shanghaiTimestamp(monthCutoff(month)), `${expected}T17:00:00.000+08:00`);
|
||||
assert.equal(monthCutoff(month).toISOString(), `${expected}T09:00:00.000Z`);
|
||||
}
|
||||
assert.equal(shanghaiMonth(new Date('2026-12-31T16:00:00Z')), '2027-01');
|
||||
assert.equal(nextCapture(new Date('2026-12-31T09:00:00Z')), '2027-01-31T17:00:00.000+08:00');
|
||||
assert.throws(() => monthCutoff('2026-13'));
|
||||
});
|
||||
test('host TZ cannot change cutoff or Shanghai calendar', () => {
|
||||
const script = "import {monthCutoff,shanghaiMonth} from './src/shared/assets/history.ts'; console.log(monthCutoff('2024-02').toISOString(),shanghaiMonth(new Date('2026-12-31T16:01:00Z')))";
|
||||
const run = (TZ: string) => execFileSync(process.execPath, ['--import', 'tsx', '--input-type=module', '-e', script], { env: { ...process.env, TZ }, encoding: 'utf8' }).trim();
|
||||
assert.equal(run('UTC'), run('America/Los_Angeles'));
|
||||
});
|
||||
test('source rows preserve precise IDs and raw states; pending and abnormal do not add to inventory', () => {
|
||||
const rows = [testRow('9223372036854775000'), testRow('self','2'), { ...testRow('pending','3'), 车辆租赁状态: '4' }, { ...testRow('abnormal','4'), 车辆租赁状态: '14' }];
|
||||
const frozen = freezeRows(rows);
|
||||
assert.deepEqual(counts(frozen), { total:4, operating:2, inventory:2 });
|
||||
assert.equal(frozen[0].id, '9223372036854775000');
|
||||
assert.equal(frozen[2].vehicleStatus, '4');
|
||||
assert.equal(frozen[3].vehicleStatus, '14');
|
||||
assert.equal('abnormalMove' in frozen[0], false);
|
||||
assert.throws(() => freezeRows([testRow('1'),testRow('1')]), /重复/);
|
||||
for (const state of ['5','99','']) assert.throws(() => freezeRows([testRow('bad',state)]), /未分类/);
|
||||
});
|
||||
test('cutoff switches live to capturing/missing without fallback; a zero snapshot is archived', () => {
|
||||
const live = freezeLive([testVehicle('1')]);
|
||||
assert.equal(makePoint('2026-10', new Date('2026-10-31T08:59:59Z'), undefined, [], live).state, 'LIVE');
|
||||
assert.equal(makePoint('2026-10', new Date('2026-10-31T09:00:00Z'), undefined, [], live).state, 'CAPTURING');
|
||||
const missing = makePoint('2026-10', new Date('2026-10-31T09:01:00Z'), undefined, [], live);
|
||||
assert.equal(missing.state, 'MISSING'); assert.equal(missing.counts, null);
|
||||
const archived = makePoint('2026-10', new Date('2026-10-31T09:00:20Z'), testBatch('2026-10'), [], live);
|
||||
assert.equal(archived.state, 'ARCHIVED'); assert.deepEqual(archived.counts, { total:0,operating:0,inventory:0 });
|
||||
assert.equal(makePoint('2026-11', new Date('2026-10-19T00:00Z')).state, 'FUTURE');
|
||||
});
|
||||
@@ -0,0 +1,84 @@
|
||||
import { filterAssetVehiclesByPermission, maskCustomerNames } from '../../auth/permissions.js';
|
||||
import type { AuthUser } from '../../auth/types.js';
|
||||
import type { Vehicle, VehicleRow } from '../../types.js';
|
||||
import { isOperatingVehicle, isInventoryVehicle, transformRow } from '../vehicles/model.js';
|
||||
import { CAPTURE_WINDOW_MS, monthCutoff, shanghaiTimestamp, type HistoryCounts, type HistoryPoint } from '../../../shared/assets/history.js';
|
||||
|
||||
export interface HistoryVehicle extends Vehicle {
|
||||
vehicleStatus: string | null;
|
||||
disposition: 'operating' | 'inventory';
|
||||
}
|
||||
export interface HistoryBatch {
|
||||
id: string; month: string; status: 'COMPLETE' | 'FAILED' | 'MISSED';
|
||||
source: 'AUTO' | 'IMPORT'; scheduledFor: string; capturedAt: string | null;
|
||||
counts: HistoryCounts;
|
||||
}
|
||||
export interface HistorySettings { enabledAt: string }
|
||||
export class HistoryDataError extends Error {}
|
||||
export class HistoryStorageUnavailable extends Error {}
|
||||
|
||||
export function validateVehicles(vehicles: Vehicle[]): void {
|
||||
const ids = new Set<string>();
|
||||
for (const vehicle of vehicles) {
|
||||
if (typeof vehicle.id !== 'string' || !vehicle.id || ids.has(vehicle.id)) throw new HistoryDataError('车辆 ID 缺失或重复，无法生成可信快照');
|
||||
ids.add(vehicle.id);
|
||||
if (!isOperatingVehicle(vehicle) && !isInventoryVehicle(vehicle)) throw new HistoryDataError('存在未分类运营状态，无法生成可信快照');
|
||||
}
|
||||
}
|
||||
export function freezeRows(rows: VehicleRow[]): HistoryVehicle[] {
|
||||
const vehicles = rows.map(row => {
|
||||
const vehicle = transformRow(row);
|
||||
return { ...vehicle, vehicleStatus: row.车辆租赁状态,
|
||||
disposition: isOperatingVehicle(vehicle) ? 'operating' as const : 'inventory' as const };
|
||||
});
|
||||
validateVehicles(vehicles);
|
||||
// Only freeze fields used by asset history. Movement work/customer attachments are unnecessary.
|
||||
return vehicles.map(({ abnormalMove: _movement, pendingDelivery: _pending, ...vehicle }) => vehicle);
|
||||
}
|
||||
export function freezeLive(vehicles: Vehicle[]): HistoryVehicle[] {
|
||||
validateVehicles(vehicles);
|
||||
return vehicles.map(({ abnormalMove: _movement, pendingDelivery: _pending, ...v }) => ({ ...v,
|
||||
vehicleStatus: v.vehicleStatus || null, disposition: isOperatingVehicle(v) ? 'operating' : 'inventory' }));
|
||||
}
|
||||
export function authorizedVehicles(vehicles: HistoryVehicle[], user: AuthUser, subject?: string): HistoryVehicle[] {
|
||||
return maskCustomerNames(filterAssetVehiclesByPermission(vehicles, user).filter(v => !subject || v.subjectOrg === subject));
|
||||
}
|
||||
export function counts(vehicles: Vehicle[]): HistoryCounts {
|
||||
return { total: vehicles.length, operating: vehicles.filter(isOperatingVehicle).length, inventory: vehicles.filter(isInventoryVehicle).length };
|
||||
}
|
||||
export function makePoint(month: string, now: Date, batch?: HistoryBatch, data?: HistoryVehicle[], live?: HistoryVehicle[], annual = false): HistoryPoint {
|
||||
const cutoff = monthCutoff(month);
|
||||
let state: HistoryPoint['state'] = 'MISSING';
|
||||
if (cutoff > now) state = month.slice(0, 4) === shanghaiTimestamp(now).slice(0, 4) && (annual || month === shanghaiTimestamp(now).slice(0, 7)) ? 'LIVE' : 'FUTURE';
|
||||
else if (batch?.status === 'COMPLETE') state = 'ARCHIVED';
|
||||
else if (batch?.status === 'FAILED') state = 'FAILED';
|
||||
else if (now.getTime() < cutoff.getTime() + CAPTURE_WINDOW_MS && batch?.status !== 'MISSED') state = 'CAPTURING';
|
||||
const available = state === 'ARCHIVED' || state === 'LIVE';
|
||||
const pointCounts = state === 'ARCHIVED' ? counts(data || []) : state === 'LIVE' ? counts(live || []) : null;
|
||||
return {
|
||||
period: annual ? month.slice(0, 4) : month,
|
||||
label: annual ? `${month.slice(0, 4)}年` : `${Number(month.slice(5))}月`, month,
|
||||
scheduledFor: shanghaiTimestamp(cutoff),
|
||||
asOf: state === 'LIVE' ? shanghaiTimestamp(now) : state === 'ARCHIVED' ? batch!.capturedAt : null,
|
||||
state, source: state === 'LIVE' ? 'LIVE' : state === 'ARCHIVED' ? batch!.source : null,
|
||||
batchId: state === 'ARCHIVED' ? batch!.id : null, counts: pointCounts,
|
||||
difference: null, comparisonLabel: '暂无可比数据', detailAvailable: available,
|
||||
};
|
||||
}
|
||||
export function comparePoint(point: HistoryPoint, previous: HistoryPoint | undefined, annual: boolean): HistoryPoint {
|
||||
if (point.reconstruction || previous?.reconstruction) {
|
||||
const comparable = point.reconstruction && previous?.reconstruction && point.reconstruction.workbookSha256 === previous.reconstruction.workbookSha256;
|
||||
return { ...point, difference: null,
|
||||
...(point.reconstruction ? { reconstruction: { ...point.reconstruction, difference: comparable ? {
|
||||
operating: point.reconstruction.operating - previous!.reconstruction!.operating,
|
||||
inventory: point.reconstruction.inventory - previous!.reconstruction!.inventory,
|
||||
} : null } } : {}),
|
||||
comparisonLabel: comparable ? `推算值较${annual ? '上年年末' : '上月末'}` : '推算与实时或正式快照口径不同，暂无可比数据' };
|
||||
}
|
||||
const label = annual ? '上年末' : '上月末';
|
||||
return { ...point, difference: point.counts && previous?.counts ? {
|
||||
total: point.counts.total - previous.counts.total,
|
||||
inventory: point.counts.inventory - previous.counts.inventory,
|
||||
operating: point.counts.operating - previous.counts.operating,
|
||||
} : null, comparisonLabel: point.counts && previous?.counts ? `${point.state === 'LIVE' ? '截至当前' : ''}较${label}` : '暂无可比数据' };
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtemp, readFile, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import type { AuthUser } from '../../auth/types.js';
|
||||
import { MemoryHistoryReader, testVehicle } from './fixtures.js';
|
||||
import { HistoryService, HistoryPeriodUnavailable } from './service.js';
|
||||
import { FileReconstructionReader, publishReconstruction, reconstructionPoint, validateReconstruction, type ReconstructionData } from './reconstruction.js';
|
||||
|
||||
const user = (permissionLevel: AuthUser['permissionLevel'] = 'full'): AuthUser => ({ userId:'1',userName:'测试',loginName:'test',depName:'甲',depCode:'a',permissionLevel,roles:[] });
|
||||
const now = () => new Date('2026-10-19T14:38:33+08:00');
|
||||
const data = (): ReconstructionData => ({ schemaVersion:1,method:'HANDOVER_REVIEW_V1',workbookSha256:'a'.repeat(64),sourceAsOf:'2026-10-19T14:38:33+08:00',sourceReference:'测试核对表.xlsx',
|
||||
periods:[{month:'2026-08',operating:1,inventory:0,unclassified:2,currentlyExited:1,vehicles:[{id:'90071992547409931',plateNumber:'历史车牌',vin:'OLDVIN',disposition:'operating',basis:'交车 D:1，2026-08-01 10:00:00'}]},
|
||||
{month:'2026-09',operating:0,inventory:1,unclassified:2,currentlyExited:1,vehicles:[{id:'90071992547409931',plateNumber:'历史车牌',vin:'OLDVIN',disposition:'inventory',basis:'还车 R:2，2026-09-01 10:00:00'}]}] });
|
||||
function setup(value = data()) {
|
||||
const store = new MemoryHistoryReader(); let reads = 0;
|
||||
const service = new HistoryService(store,async()=>[testVehicle('today')],now,{read:async()=>{reads++;return value;}});
|
||||
return { store,service,reads:()=>reads };
|
||||
}
|
||||
test('reconstruction preserves unknown total, source date and string IDs without claiming a captured snapshot', async()=>{
|
||||
const {service}=setup(); const trend=await service.trend(user(),{year:'2026'});
|
||||
const p=trend.points[8];
|
||||
assert.equal(p.state,'RECONSTRUCTED');assert.equal(p.counts,null);assert.equal(p.asOf,null);
|
||||
assert.equal(p.reconstruction!.operating,0);assert.equal(p.reconstruction!.inventory,1);
|
||||
assert.deepEqual(p.reconstruction!.difference,{operating:-1,inventory:1});
|
||||
assert.equal(trend.points[7].reconstruction!.difference,null);
|
||||
assert.equal(trend.points[9].state,'LIVE');assert.equal(trend.points[9].difference,null);
|
||||
assert.equal(trend.points[10].state,'FUTURE');
|
||||
const detail=await service.details(user(),{month:'2026-09',batchId:p.batchId!});
|
||||
assert.equal(detail.counts,null);assert.equal(detail.items[0].id,'90071992547409931');
|
||||
assert.equal(detail.items[0].plateNumber,'历史车牌');assert.equal(detail.items[0].subjectOrg,null);
|
||||
assert.equal(detail.items[0].operationStatus,null);assert.equal(detail.items[0].customerName,null);
|
||||
assert.match(detail.items[0].reconstructionBasis!,/R:2/);
|
||||
});
|
||||
test('company/department/personal scopes cannot load or query unowned reconstruction rows', async()=>{
|
||||
for(const [level,params] of [['department',{}],['personal',{}],['full',{subject:'任意公司'}]] as const){
|
||||
const {service,reads}=setup();const identity=user(level);
|
||||
const trend=await service.trend(identity,{year:'2026',...params});assert.equal(trend.points[8].state,'MISSING');
|
||||
await assert.rejects(()=>service.details(identity,{month:'2026-09',batchId:`rec-${'a'.repeat(12)}-2026-09`,...params}),HistoryPeriodUnavailable);
|
||||
assert.equal(reads(),0);
|
||||
}
|
||||
});
|
||||
test('formal archives take precedence and forged or stale reconstruction batch IDs are refused',async()=>{
|
||||
const {service,store}=setup();store.add('2026-09',[testVehicle('formal')]);
|
||||
const trend=await service.trend(user(),{});assert.equal(trend.points[8].state,'ARCHIVED');assert.equal(trend.points[8].reconstruction,undefined);
|
||||
await assert.rejects(()=>service.details(user(),{month:'2026-09',batchId:`rec-${'a'.repeat(12)}-2026-09`}),HistoryPeriodUnavailable);
|
||||
await assert.rejects(()=>service.details(user(),{month:'2026-08',batchId:'rec-forged'}),HistoryPeriodUnavailable);
|
||||
assert.equal((await service.details(user(),{month:'2026-09'})).items[0].id,'formal');
|
||||
});
|
||||
test('reconstructed category/search/pagination counts agree with the selected period; missing months stay absent',async()=>{
|
||||
const value=data();value.periods[1].vehicles=Array.from({length:61},(_,i)=>({id:String(i+1),plateNumber:`历史${i}`,vin:`OLD${i}`,disposition:'inventory',basis:'还车 R:1'}));value.periods[1].inventory=61;
|
||||
const {service}=setup(value);
|
||||
const all=await service.details(user(),{month:'2026-09',page:'2',pageSize:'30'});assert.equal(all.total,61);assert.equal(all.items.length,30);
|
||||
assert.equal((await service.details(user(),{month:'2026-09',category:'operating'})).total,0);
|
||||
assert.equal((await service.details(user(),{month:'2026-09',search:'OLD60'})).total,1);
|
||||
assert.equal((await service.trend(user(),{})).points[6].state,'MISSING');
|
||||
});
|
||||
test('annual estimates reuse December only; no sum, current-month replacement, or cross-source difference',async()=>{
|
||||
const value=data();value.periods.push({...value.periods[0],month:'2025-12'});
|
||||
const {service}=setup(value);const result=await service.trend(user(),{mode:'annual'});
|
||||
assert.equal(result.points.find(p=>p.period==='2025')!.reconstruction!.operating,1);
|
||||
assert.equal(result.points.find(p=>p.period==='2026')!.state,'LIVE');assert.equal(result.points.at(-1)!.difference,null);
|
||||
assert.equal(reconstructionPoint(value,'2026-10',false,now()),null);
|
||||
});
|
||||
test('schema rejects duplicates, unsafe IDs, invented counts, current/future months and missing provenance',()=>{
|
||||
const mutate=[(d:ReconstructionData)=>{d.periods.push(d.periods[0]);},(d:ReconstructionData)=>{d.periods[0].vehicles[0].id=1 as never;},
|
||||
(d:ReconstructionData)=>{d.periods[0].operating=99;},(d:ReconstructionData)=>{d.periods[0].month='2026-10';},
|
||||
(d:ReconstructionData)=>{d.workbookSha256='';},(d:ReconstructionData)=>{d.periods[0].vehicles.push(d.periods[0].vehicles[0]);d.periods[0].operating=2;},
|
||||
(d:ReconstructionData)=>{d.periods[0].unclassified=-1;}];
|
||||
for(const change of mutate){const d=data();change(d);assert.throws(()=>validateReconstruction(d),/无效/);}
|
||||
const d=data();Object.assign(d.periods[0].vehicles[0],{subjectOrg:'今日主体',customerName:'不应导入的客户'});
|
||||
assert.equal(JSON.stringify(validateReconstruction(d)).includes('今日主体'),false);
|
||||
const zero=data();zero.periods[0]={month:'2026-08',operating:0,inventory:0,unclassified:3,currentlyExited:0,vehicles:[]};
|
||||
assert.equal(reconstructionPoint(validateReconstruction(zero),'2026-08',false,now())!.reconstruction!.operating,0);
|
||||
});
|
||||
test('explicit local publication is atomic, idempotent and immutable; reader has no database dependencies',async()=>{
|
||||
const dir=await mkdtemp(join(tmpdir(),'asset-reconstruction-'));
|
||||
try{
|
||||
const path=join(dir,'history.json');const value=data();
|
||||
const results=await Promise.all([publishReconstruction(path,value),publishReconstruction(path,value)]);
|
||||
assert.deepEqual(results.sort(),['created','unchanged']);
|
||||
const before=await readFile(path,'utf8');const different=data();different.workbookSha256='b'.repeat(64);
|
||||
await assert.rejects(()=>publishReconstruction(path,different),/禁止自动覆盖/);assert.equal(await readFile(path,'utf8'),before);
|
||||
assert.deepEqual(await new FileReconstructionReader(path).read(),validateReconstruction(value));
|
||||
}finally{await rm(dir,{recursive:true,force:true});}
|
||||
});
|
||||
@@ -0,0 +1,86 @@
|
||||
import { readFile, writeFile, link, unlink, mkdir } from 'node:fs/promises';
|
||||
import { dirname } from 'node:path';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { monthCutoff, shanghaiTimestamp, type HistoryDetailItem, type HistoryPoint } from '../../../shared/assets/history.js';
|
||||
|
||||
export interface ReconstructionPeriod {
|
||||
month: string; operating: number; inventory: number; unclassified: number; currentlyExited: number;
|
||||
vehicles: { id: string; plateNumber: string; vin: string; disposition: 'operating' | 'inventory'; basis: string }[];
|
||||
}
|
||||
export interface ReconstructionData {
|
||||
schemaVersion: 1; method: 'HANDOVER_REVIEW_V1'; workbookSha256: string;
|
||||
sourceAsOf: string; sourceReference: string; periods: ReconstructionPeriod[];
|
||||
}
|
||||
export function validateReconstruction(input: unknown): ReconstructionData {
|
||||
const fail = (): never => { throw new Error('历史回溯文件无效，未加载任何回填数据'); };
|
||||
if (!input || typeof input !== 'object') return fail();
|
||||
const data = input as ReconstructionData;
|
||||
if (data.schemaVersion !== 1 || data.method !== 'HANDOVER_REVIEW_V1' || !/^[a-f0-9]{64}$/.test(data.workbookSha256) ||
|
||||
typeof data.sourceReference !== 'string' || !data.sourceReference.trim() || data.sourceReference.length > 200 ||
|
||||
!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?\+08:00$/.test(data.sourceAsOf) || !Number.isFinite(Date.parse(data.sourceAsOf)) ||
|
||||
!Array.isArray(data.periods) || data.periods.length > 1200) return fail();
|
||||
const months = new Set<string>();
|
||||
for (const period of data.periods) {
|
||||
if (!period || !/^20\d{2}-(0[1-9]|1[0-2])$/.test(period.month) || months.has(period.month) || monthCutoff(period.month).getTime() >= Date.parse(data.sourceAsOf) ||
|
||||
![period.operating, period.inventory, period.unclassified, period.currentlyExited].every(v => Number.isSafeInteger(v) && v >= 0) ||
|
||||
period.currentlyExited > period.operating + period.inventory || !Array.isArray(period.vehicles) || period.vehicles.length !== period.operating + period.inventory) return fail();
|
||||
months.add(period.month);
|
||||
const ids = new Set<string>(); let operating = 0;
|
||||
for (const v of period.vehicles) {
|
||||
if (!v || typeof v.id !== 'string' || !/^[1-9]\d{0,63}$/.test(v.id) || ids.has(v.id) || !['operating', 'inventory'].includes(v.disposition) ||
|
||||
typeof v.plateNumber !== 'string' || v.plateNumber.length > 100 || typeof v.vin !== 'string' || v.vin.length > 100 ||
|
||||
typeof v.basis !== 'string' || !v.basis.trim() || v.basis.length > 1000) return fail();
|
||||
ids.add(v.id); if (v.disposition === 'operating') operating++;
|
||||
}
|
||||
if (operating !== period.operating) return fail();
|
||||
}
|
||||
// Return only reviewed fields. Ignore any attempted current company/customer/role injection.
|
||||
return { schemaVersion: 1, method: data.method, workbookSha256: data.workbookSha256, sourceAsOf: data.sourceAsOf, sourceReference: data.sourceReference,
|
||||
periods: data.periods.map(p => ({ month: p.month, operating: p.operating, inventory: p.inventory, unclassified: p.unclassified, currentlyExited: p.currentlyExited,
|
||||
vehicles: p.vehicles.map(v => ({ id: v.id, plateNumber: v.plateNumber, vin: v.vin, disposition: v.disposition, basis: v.basis })) })) };
|
||||
}
|
||||
|
||||
export interface ReconstructionReader { read(): Promise<ReconstructionData> }
|
||||
/** Opt-in, immutable local file. Reading never creates storage or writes a database. */
|
||||
export class FileReconstructionReader implements ReconstructionReader {
|
||||
private loaded?: Promise<ReconstructionData>;
|
||||
constructor(private readonly path: string) {}
|
||||
read() {
|
||||
this.loaded ||= readFile(this.path, 'utf8').then(text => {
|
||||
if (text.length > 50_000_000) throw new Error('历史回溯文件超出限制');
|
||||
return validateReconstruction(JSON.parse(text));
|
||||
});
|
||||
return this.loaded;
|
||||
}
|
||||
}
|
||||
|
||||
/** Explicit local publication only: create atomically; never replace a different revision. */
|
||||
export async function publishReconstruction(path: string, input: unknown): Promise<'created' | 'unchanged'> {
|
||||
const text = JSON.stringify(validateReconstruction(input));
|
||||
await mkdir(dirname(path), { recursive: true });
|
||||
const temporary = `${path}.${randomUUID()}.tmp`;
|
||||
await writeFile(temporary, text, { flag: 'wx', mode: 0o600 });
|
||||
try {
|
||||
try { await link(temporary, path); return 'created'; }
|
||||
catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error;
|
||||
if (await readFile(path, 'utf8') === text) return 'unchanged';
|
||||
throw new Error('目标已有不同版本的回溯数据，禁止自动覆盖');
|
||||
}
|
||||
} finally { await unlink(temporary); }
|
||||
}
|
||||
|
||||
export function reconstructionPoint(data: ReconstructionData, month: string, annual: boolean, now: Date): HistoryPoint | null {
|
||||
const period = data.periods.find(p => p.month === month);
|
||||
if (!period || monthCutoff(month) >= now) return null;
|
||||
return { period: annual ? month.slice(0, 4) : month, label: annual ? `${month.slice(0, 4)}年` : `${Number(month.slice(5))}月`, month,
|
||||
scheduledFor: shanghaiTimestamp(monthCutoff(month)), asOf: null, state: 'RECONSTRUCTED', source: 'RECONSTRUCTED',
|
||||
batchId: `rec-${data.workbookSha256.slice(0, 12)}-${month}`, counts: null, difference: null, comparisonLabel: '暂无可比数据', detailAvailable: true,
|
||||
reconstruction: { operating: period.operating, inventory: period.inventory, unclassified: period.unclassified,
|
||||
currentlyExited: period.currentlyExited, sourceAsOf: data.sourceAsOf, sourceReference: data.sourceReference, workbookSha256: data.workbookSha256, difference: null } };
|
||||
}
|
||||
export function reconstructionItems(period: ReconstructionPeriod): HistoryDetailItem[] {
|
||||
return period.vehicles.map(v => ({ id: v.id, plateNumber: v.plateNumber, vin: v.vin, disposition: v.disposition,
|
||||
type: '', model: '', operationStatus: null, vehicleStatus: null, subjectOrg: null, departmentName: null,
|
||||
customerManager: null, customerName: null, contractNo: null, reconstructionBasis: v.basis }));
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
import { historyStore } from './index.js';
|
||||
import { HistoryScheduler } from './scheduler.js';
|
||||
|
||||
let scheduler: HistoryScheduler | undefined;
|
||||
export function startAssetHistoryScheduler(): void {
|
||||
if (process.env.ASSET_HISTORY_CAPTURE_ENABLED !== '1' || scheduler) return;
|
||||
scheduler = new HistoryScheduler(historyStore);
|
||||
scheduler.start();
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { HistoryScheduler, type HistoryCaptureStore } from './scheduler.js';
|
||||
import { testBatch } from './fixtures.js';
|
||||
function fixture(enabledAt='2026-10-19T13:00:00+08:00') {
|
||||
const captures:string[]=[],missed:string[]=[],failed:string[]=[];
|
||||
let activates=0;
|
||||
const store:HistoryCaptureStore={
|
||||
activate:async()=>{activates++;return{enabledAt};},batches:async()=>[],
|
||||
capture:async(month)=>{captures.push(month);return'captured';},
|
||||
recordOutcome:async(month,status)=>{(status==='FAILED'?failed:missed).push(month);},
|
||||
};
|
||||
return {store,captures,missed,failed,activates:()=>activates};
|
||||
}
|
||||
test('scheduler captures only in month-end 17:00 minute and activation is persisted once',async()=>{
|
||||
const f=fixture();let date=new Date('2026-10-31T16:59:59+08:00');
|
||||
const scheduler=new HistoryScheduler(f.store,()=>date,()=>{});
|
||||
await scheduler.tick();assert.deepEqual(f.captures,[]);
|
||||
date=new Date('2026-10-31T17:00:00+08:00');await scheduler.tick();assert.deepEqual(f.captures,['2026-10']);assert.equal(f.activates(),1);
|
||||
});
|
||||
test('restart records missed eligible months without backfilling old live data or pre-activation periods',async()=>{
|
||||
const f=fixture();const scheduler=new HistoryScheduler(f.store,()=>new Date('2027-01-15T08:00:00+08:00'),()=>{});
|
||||
await scheduler.tick();assert.deepEqual(f.captures,[]);assert.deepEqual(f.missed,['2026-10','2026-11','2026-12']);
|
||||
const late=fixture('2026-10-31T17:00:10+08:00');await new HistoryScheduler(late.store,()=>new Date('2026-10-31T17:00:20+08:00'),()=>{}).tick();
|
||||
assert.deepEqual(late.captures,[]);assert.deepEqual(late.missed,[]);
|
||||
});
|
||||
test('completed December skips duplicate annual capture; failures stay marked without late retries',async()=>{
|
||||
const f=fixture();f.store.batches=async()=>[testBatch('2026-10')];
|
||||
f.store.capture=async()=>{throw new Error('source unavailable');};
|
||||
await new HistoryScheduler(f.store,()=>new Date('2026-12-31T17:00:20+08:00'),()=>{}).tick();
|
||||
assert.deepEqual(f.missed,['2026-11']);assert.deepEqual(f.failed,['2026-12']);
|
||||
const done=fixture('2026-12-01T00:00+08:00');done.store.batches=async()=>[testBatch('2026-12')];
|
||||
await new HistoryScheduler(done.store,()=>new Date('2026-12-31T17:00:20+08:00'),()=>{}).tick();assert.deepEqual(done.captures,[]);
|
||||
});
|
||||
test('concurrent ticks and repeated start do not register duplicate jobs',async()=>{
|
||||
const f=fixture();let release!:()=>void;
|
||||
f.store.capture=async()=>{f.captures.push('2026-10');await new Promise<void>(r=>{release=r;});return'captured';};
|
||||
const scheduler=new HistoryScheduler(f.store,()=>new Date('2026-10-31T17:00:20+08:00'),()=>{});
|
||||
const stop=scheduler.start();scheduler.start();
|
||||
await new Promise<void>(r=>setImmediate(r));await scheduler.tick();assert.equal(f.captures.length,1);
|
||||
stop();release();await new Promise<void>(r=>setImmediate(r));
|
||||
});
|
||||
@@ -0,0 +1,61 @@
|
||||
import { CAPTURE_WINDOW_MS, monthCutoff, shanghaiMonth, shiftMonth } from '../../../shared/assets/history.js';
|
||||
import type { CaptureResult } from './store.js';
|
||||
import { HistoryDataError, type HistoryBatch, type HistorySettings } from './model.js';
|
||||
|
||||
export interface HistoryCaptureStore {
|
||||
activate(now: Date): Promise<HistorySettings>;
|
||||
batches(): Promise<HistoryBatch[]>;
|
||||
capture(month: string, clock: () => Date): Promise<CaptureResult>;
|
||||
recordOutcome(month: string, status: 'MISSED' | 'FAILED', code: string): Promise<void>;
|
||||
}
|
||||
export class HistoryScheduler {
|
||||
private timer: ReturnType<typeof setTimeout> | undefined;
|
||||
private running = false;
|
||||
private ticking = false;
|
||||
private settings: HistorySettings | null = null;
|
||||
constructor(private readonly store: HistoryCaptureStore, private readonly clock: () => Date = () => new Date(),
|
||||
private readonly log: (message: string) => void = console.log) {}
|
||||
|
||||
/** Exported for deterministic testing and explicit operational checks; never mounted as HTTP write. */
|
||||
async tick(): Promise<void> {
|
||||
if (this.ticking) return;
|
||||
this.ticking = true;
|
||||
try {
|
||||
const now = this.clock();
|
||||
this.settings ||= await this.store.activate(now);
|
||||
const enabled = new Date(this.settings.enabledAt);
|
||||
const existing = new Map((await this.store.batches()).map(b => [b.month, b]));
|
||||
for (let month = shanghaiMonth(enabled); month <= shanghaiMonth(now); month = shiftMonth(month, 1)) {
|
||||
const cutoff = monthCutoff(month);
|
||||
if (cutoff < enabled || cutoff > now || existing.get(month)?.status === 'COMPLETE') continue;
|
||||
if (now.getTime() >= cutoff.getTime() + CAPTURE_WINDOW_MS) {
|
||||
if (!existing.has(month)) { await this.store.recordOutcome(month, 'MISSED', 'CAPTURE_WINDOW_MISSED'); this.log(`[asset-history] missed ${month}`); }
|
||||
continue;
|
||||
}
|
||||
try { const result = await this.store.capture(month, this.clock); this.log(`[asset-history] ${month}: ${result}`); }
|
||||
catch (error) {
|
||||
const databaseCode = (error as { code?: unknown })?.code;
|
||||
const code = error instanceof HistoryDataError ? 'INVALID_VEHICLE_DATA' : typeof databaseCode === 'string' && /^[A-Z0-9_]{1,48}$/.test(databaseCode) ? databaseCode : 'CAPTURE_FAILED';
|
||||
await this.store.recordOutcome(month, 'FAILED', code);
|
||||
this.log(`[asset-history] failed ${month} (${code}); no later live data will be substituted`);
|
||||
}
|
||||
}
|
||||
} finally { this.ticking = false; }
|
||||
}
|
||||
start(): () => void {
|
||||
if (this.running) return () => this.stop();
|
||||
this.running = true;
|
||||
const run = async () => {
|
||||
if (!this.running) return;
|
||||
try { await this.tick(); } catch { this.log('[asset-history] scheduler unavailable; check migration and database access'); }
|
||||
if (!this.running) return;
|
||||
const now = this.clock();
|
||||
const untilCutoff = monthCutoff(shanghaiMonth(now)).getTime() - now.getTime();
|
||||
this.timer = setTimeout(run, untilCutoff > 0 ? Math.min(untilCutoff, 60_000) : 15_000);
|
||||
this.timer.unref?.();
|
||||
};
|
||||
void run();
|
||||
return () => this.stop();
|
||||
}
|
||||
stop(): void { this.running = false; clearTimeout(this.timer); }
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { Hono } from 'hono';
|
||||
import type { AuthUser } from '../../auth/types.js';
|
||||
import { createPreviewMiddleware } from '../../auth/preview/routes.js';
|
||||
import { createHistoryRouter } from './index.js';
|
||||
import { HistoryService, HistoryInputError, HistoryPeriodUnavailable } from './service.js';
|
||||
import { HistoryStorageUnavailable } from './model.js';
|
||||
import { MemoryHistoryReader, testVehicle } from './fixtures.js';
|
||||
|
||||
const user = (permissionLevel: AuthUser['permissionLevel'] = 'full'): AuthUser => ({ userId:'manager-a',loginName:'a',userName:'甲',depCode:'a',depName:'部门甲',permissionLevel,roles:[] });
|
||||
const now = () => new Date('2026-10-19T13:00:00+08:00');
|
||||
const inventory = testVehicle('stock',{ operationStatus:'3',status:'Abnormal',departmentName:'部门乙',managerId:'manager-b',subjectOrg:'历史公司' });
|
||||
const leased = testVehicle('own');
|
||||
const foreign = testVehicle('foreign',{ managerId:'manager-b',departmentName:'部门乙' });
|
||||
function setup() {
|
||||
const store = new MemoryHistoryReader(); store.add('2026-09',[leased,foreign,inventory]);
|
||||
const live = [testVehicle('today'), testVehicle('today-stock',{ operationStatus:'4',status:'Pending' })];
|
||||
return { store, live, service:new HistoryService(store,async()=>live,now) };
|
||||
}
|
||||
test('monthly holes/future remain null; current values are real; comparisons never skip missing periods', async () => {
|
||||
const {store,service}=setup(); store.add('2026-07',[leased]);
|
||||
const result=await service.trend(user(),{year:'2026'});
|
||||
assert.equal(result.points.length,12);
|
||||
assert.deepEqual(result.points[8].counts,{total:3,operating:2,inventory:1});
|
||||
assert.equal(result.points[8].difference,null); // August is missing, July cannot substitute.
|
||||
assert.deepEqual(result.points[9].difference,{total:-1,operating:-1,inventory:0});
|
||||
assert.equal(result.points[9].state,'LIVE'); assert.equal(result.points[0].counts,null);
|
||||
assert.equal(result.points[10].state,'FUTURE'); assert.equal(result.points[10].counts,null);
|
||||
assert.ok(result.subjects.includes('历史公司'));
|
||||
});
|
||||
test('annual counts use only December snapshots, not a sum of months', async () => {
|
||||
const {store,service}=setup(); store.add('2025-12',[leased]); store.add('2025-11',[leased,foreign,inventory]);
|
||||
const result=await service.trend(user(),{mode:'annual'});
|
||||
assert.deepEqual(result.points.find(p=>p.period==='2025')!.counts,{total:1,operating:1,inventory:0});
|
||||
assert.equal(result.points.at(-1)!.state,'LIVE');
|
||||
assert.deepEqual(result.points.at(-1)!.difference,{total:1,operating:0,inventory:1});
|
||||
});
|
||||
test('full/department/personal and manager aliases share snapshot/live public inventory policy', async () => {
|
||||
const {service}=setup();
|
||||
for (const level of ['full','department','personal'] as const) {
|
||||
const result=await service.trend(user(level),{});
|
||||
const september=result.points[8];
|
||||
assert.equal(september.counts!.total,level==='full'?3:2);
|
||||
const detail=await service.details(user(level),{month:'2026-09'});
|
||||
assert.equal(detail.total,september.counts!.total); assert.deepEqual(detail.counts,september.counts);
|
||||
assert.ok(detail.items.some(v=>v.id==='stock'));
|
||||
assert.equal(detail.items.some(v=>v.id==='foreign'),level==='full');
|
||||
assert.ok(!JSON.stringify(detail).includes('测试客户公司'));
|
||||
}
|
||||
const alias={...user('personal'),userId:'new-id',managerIds:['manager-a']};
|
||||
assert.equal((await service.details(alias,{month:'2026-09'})).total,2);
|
||||
});
|
||||
test('company filter intersects permission; foreign batch IDs cannot bypass month or identity', async () => {
|
||||
const {service}=setup();
|
||||
const trend=await service.trend(user('personal'),{subject:'历史公司'});
|
||||
assert.deepEqual(trend.points[8].counts,{total:1,operating:0,inventory:1});
|
||||
assert.equal((await service.details(user('personal'),{month:'2026-09',subject:'公司甲'})).total,1);
|
||||
assert.equal((await service.details(user('personal'),{month:'2026-09',subject:'任意公司'})).total,0);
|
||||
await assert.rejects(()=>service.details(user(),{month:'2026-09',batchId:'other-batch'}),HistoryPeriodUnavailable);
|
||||
});
|
||||
test('historical vehicles survive current deletion, reassignment and exit without any live join', async () => {
|
||||
const {store}=setup(); let calls=0;
|
||||
const service=new HistoryService(store,async()=>{calls++;throw new Error('live must not be read');},now);
|
||||
const result=await service.details(user('personal'),{month:'2026-09'});
|
||||
assert.deepEqual(result.items.map(v=>v.id).sort(),['own','stock']);
|
||||
assert.equal(calls,0);
|
||||
assert.equal((await service.trend(user(),{year:'2025'})).points.every(p=>p.state==='MISSING'),true);
|
||||
assert.equal(calls,0);
|
||||
});
|
||||
test('pagination/search/category use authorized snapshot only, keeping unsearched counts intact', async () => {
|
||||
const {store,service}=setup();store.add('2026-08',Array.from({length:65},(_,i)=>testVehicle(String(i))));
|
||||
const detail=await service.details(user(),{month:'2026-08',page:'2',pageSize:'30'});
|
||||
assert.equal(detail.total,65);assert.equal(detail.items.length,30);assert.equal(detail.counts!.total,65);
|
||||
const filtered=await service.details(user(),{month:'2026-09',category:'inventory',search:'TESTVINstock'});
|
||||
assert.equal(filtered.total,1);assert.equal(filtered.counts!.total,3);
|
||||
});
|
||||
test('missing storage degrades to live without any DDL; genuine database errors are not hidden', async () => {
|
||||
const {store,live}=setup();store.settings=async()=>{throw new HistoryStorageUnavailable();};store.batches=async()=>{throw new HistoryStorageUnavailable();};
|
||||
const service=new HistoryService(store,async()=>live,now);
|
||||
const result=await service.trend(user(),{});assert.equal(result.storageReady,false);assert.equal(result.points[9].counts!.total,2);assert.equal(result.points[8].counts,null);
|
||||
await assert.rejects(()=>service.details(user(),{month:'2026-09'}),HistoryPeriodUnavailable);
|
||||
store.batches=async()=>{throw new Error('database offline');};await assert.rejects(()=>service.trend(user(),{}),/offline/);
|
||||
});
|
||||
test('once cutoff passes live values cannot replace missing history, even when capture was not enabled', async () => {
|
||||
const {store}=setup();store.settingsValue=null;let liveCalls=0;
|
||||
const service=new HistoryService(store,async()=>{liveCalls++;return[leased];},()=>new Date('2026-10-31T17:00:30+08:00'));
|
||||
const result=await service.trend(user(),{});assert.equal(result.points[9].state,'MISSING');assert.equal(result.points[9].counts,null);
|
||||
assert.equal(liveCalls,0);await assert.rejects(()=>service.details(user(),{month:'2026-10'}),HistoryPeriodUnavailable);
|
||||
});
|
||||
test('snapshot count integrity failure never appears as a legitimate zero', async () => {
|
||||
const {store,service}=setup();store.rows.clear();await assert.rejects(()=>service.trend(user(),{}),/integrity/);
|
||||
});
|
||||
test('invalid parameters fail before querying details or live; valid archived zero retains detail access', async () => {
|
||||
const {store,service}=setup();store.add('2026-08',[]);
|
||||
assert.equal((await service.details(user(),{month:'2026-08'})).total,0);
|
||||
for(const params of [{year:'NaN'},{year:'2027'},{mode:'weekly'}] as Record<string,string>[]) await assert.rejects(()=>service.trend(user(),params),HistoryInputError);
|
||||
for(const params of [{month:'2026-13'},{month:'2026-09',page:'0'},{month:'2026-09',pageSize:'101'},{month:'2026-09',category:'hacked'}] as Record<string,string>[]) await assert.rejects(()=>service.details(user(),params),HistoryInputError);
|
||||
});
|
||||
test('HTTP router authenticates, validates and exposes only read routes with private no-store', async () => {
|
||||
const {service}=setup();const router=createHistoryRouter(service);
|
||||
assert.equal((await router.request('/trend')).status,401);
|
||||
const app=new Hono<{Variables:{user:AuthUser}}>();app.use('*',async(c,next)=>{c.set('user',user('personal'));await next();});app.route('/',router);
|
||||
const response=await app.request('/trend?year=2026');assert.equal(response.status,200);assert.match(response.headers.get('Cache-Control')!,/private.*no-store/);
|
||||
assert.equal((await app.request('/details?month=2026-08')).status,409);
|
||||
assert.equal((await app.request('/trend?year=x')).status,400);
|
||||
assert.equal((await app.request('/details?month=2026-09&batchId=forged')).status,409);
|
||||
assert.equal((await app.request('/capture',{method:'POST'})).status,404);
|
||||
});
|
||||
test('permission-preview identity applies to history GET; capture/import writes remain blocked', async () => {
|
||||
const {service}=setup();const audit:string[]=[];
|
||||
const preview={current:async()=>({user:user('personal'),sessionId:'test-session'}),audit:async(event:{event:string})=>{audit.push(event.event);}};
|
||||
const app=new Hono<{Variables:{user:AuthUser}}>();app.use('*',async(c,next)=>{c.set('user',user());await next();});
|
||||
app.use('*',createPreviewMiddleware(preview as never));app.route('/api/asset-history',createHistoryRouter(service));
|
||||
const response=await app.request('/api/asset-history/details?month=2026-09',{headers:{'X-BI-Preview':'isolated-test'}});
|
||||
assert.equal(response.status,200);assert.equal((await response.json()).total,2);
|
||||
assert.equal((await app.request('/api/asset-history/import',{method:'POST',headers:{'X-BI-Preview':'isolated-test'}})).status,403);
|
||||
assert.deepEqual(audit,['read','blocked-write']);
|
||||
});
|
||||
@@ -0,0 +1,126 @@
|
||||
import type { AuthUser } from '../../auth/types.js';
|
||||
import type { Vehicle } from '../../types.js';
|
||||
import { isOperatingVehicle } from '../vehicles/model.js';
|
||||
import { HISTORY_STATE_LABELS, nextCapture, shanghaiMonth, shiftMonth, monthCutoff, type HistoryMode, type HistoryTrend, type HistoryCategory, type HistoryDetails } from '../../../shared/assets/history.js';
|
||||
import { authorizedVehicles, comparePoint, counts, freezeLive, HistoryStorageUnavailable, makePoint, validateVehicles, type HistoryBatch, type HistorySettings, type HistoryVehicle } from './model.js';
|
||||
import type { HistoryReader } from './store.js';
|
||||
import { reconstructionItems, reconstructionPoint, type ReconstructionReader } from './reconstruction.js';
|
||||
|
||||
export class HistoryInputError extends Error {}
|
||||
export class HistoryPeriodUnavailable extends Error {}
|
||||
export function parseHistoryQuery(params: Record<string, string>, now: Date) {
|
||||
const mode = params.mode || 'monthly';
|
||||
const year = Number(params.year || shanghaiMonth(now).slice(0, 4));
|
||||
if (!['monthly', 'annual'].includes(mode) || !Number.isInteger(year) || year < 2000 || year > Number(shanghaiMonth(now).slice(0, 4))) throw new HistoryInputError('年份或统计方式无效');
|
||||
if ((params.subject || '').length > 500) throw new HistoryInputError('公司筛选无效');
|
||||
return { mode: mode as HistoryMode, year, subject: params.subject?.trim() || undefined };
|
||||
}
|
||||
export class HistoryService {
|
||||
constructor(private readonly store: HistoryReader, private readonly live: () => Promise<Vehicle[]>, private readonly clock: () => Date = () => new Date(), private readonly reconstruction?: ReconstructionReader) {}
|
||||
|
||||
private async reconstructionData(user: AuthUser, subject?: string) {
|
||||
// These reviewed records have no trustworthy historical owner/dept/company.
|
||||
// Never grant their visibility from today's vehicle IDs or company values.
|
||||
return this.reconstruction && user.permissionLevel === 'full' && !subject ? this.reconstruction.read() : null;
|
||||
}
|
||||
|
||||
async livePoint(user: AuthUser, params: Record<string, string>) {
|
||||
const now = this.clock();
|
||||
const query = parseHistoryQuery(params, now);
|
||||
const month = query.mode === 'annual' ? `${shanghaiMonth(now).slice(0, 4)}-12` : shanghaiMonth(now);
|
||||
const rows = now < monthCutoff(month) ? authorizedVehicles(freezeLive(await this.live()), user, query.subject) : [];
|
||||
return makePoint(month, now, undefined, [], rows, query.mode === 'annual');
|
||||
}
|
||||
|
||||
private async archives(months: string[], catalog?: HistoryBatch[]) {
|
||||
let settings: HistorySettings | null = null;
|
||||
let batches: HistoryBatch[] = [];
|
||||
try {
|
||||
[settings, batches] = await Promise.all([this.store.settings(), catalog ? Promise.resolve(catalog) : this.store.batches()]);
|
||||
const selected = batches.filter(b => b.status === 'COMPLETE' && months.includes(b.month));
|
||||
const vehicles = await this.store.vehicles(selected.map(b => b.id));
|
||||
for (const batch of selected) {
|
||||
const rows = vehicles.get(batch.id) || [];
|
||||
validateVehicles(rows);
|
||||
const totals = counts(rows);
|
||||
if (!batch.capturedAt || Object.entries(totals).some(([key, value]) => value !== batch.counts[key as keyof typeof totals]) || rows.some(v => v.disposition !== (isOperatingVehicle(v) ? 'operating' : 'inventory'))) throw new Error('Snapshot integrity mismatch');
|
||||
}
|
||||
return { settings, batches, vehicles, ready: true };
|
||||
} catch (error) {
|
||||
if (!(error instanceof HistoryStorageUnavailable)) throw error;
|
||||
return { settings: null, batches: [], vehicles: new Map<string, HistoryVehicle[]>(), ready: false };
|
||||
}
|
||||
}
|
||||
async trend(user: AuthUser, params: Record<string, string>): Promise<HistoryTrend> {
|
||||
const now = this.clock();
|
||||
const query = parseHistoryQuery(params, now);
|
||||
const currentYear = Number(shanghaiMonth(now).slice(0, 4));
|
||||
const reconstruction = await this.reconstructionData(user, query.subject);
|
||||
// Metadata is small (one batch/month); only relevant detail batches are loaded.
|
||||
let catalog: HistoryBatch[] = [];
|
||||
try { catalog = await this.store.batches(); } catch (e) { if (!(e instanceof HistoryStorageUnavailable)) throw e; }
|
||||
const firstYear = Math.min(currentYear - 4, query.year, ...catalog.map(b => Number(b.month.slice(0, 4))), ...(reconstruction?.periods.map(p => Number(p.month.slice(0, 4))) || []));
|
||||
const years = Array.from({ length: currentYear - firstYear + 1 }, (_, i) => currentYear - i);
|
||||
const months = query.mode === 'monthly'
|
||||
? Array.from({ length: 12 }, (_, i) => `${query.year}-${String(i + 1).padStart(2, '0')}`)
|
||||
: [...years].reverse().map(y => `${y}-12`);
|
||||
const previous = query.mode === 'monthly' ? shiftMonth(months[0], -1) : `${firstYear - 1}-12`;
|
||||
const range = [previous, ...months];
|
||||
const hasLive = months.some(m => monthCutoff(m) > now && (query.mode === 'annual' ? m.startsWith(String(currentYear)) : m === shanghaiMonth(now)));
|
||||
const [archive, rawLive] = await Promise.all([this.archives(range, catalog), hasLive ? this.live() : Promise.resolve([])]);
|
||||
const live = freezeLive(rawLive);
|
||||
const subjects = new Set(authorizedVehicles(live, user).map(v => v.subjectOrg).filter((s): s is string => !!s));
|
||||
const byMonth = new Map(archive.batches.map(batch => [batch.month, batch]));
|
||||
for (const rows of archive.vehicles.values()) for (const v of authorizedVehicles(rows, user)) if (v.subjectOrg) subjects.add(v.subjectOrg);
|
||||
const points = range.map(month => {
|
||||
const batch = byMonth.get(month);
|
||||
const rows = batch?.status === 'COMPLETE' ? archive.vehicles.get(batch.id)! : [];
|
||||
const point = makePoint(month, now, batch, authorizedVehicles(rows, user, query.subject), authorizedVehicles(live, user, query.subject), query.mode === 'annual');
|
||||
if (point.state === 'CAPTURING' && (!archive.settings || new Date(archive.settings.enabledAt) > monthCutoff(month))) point.state = 'MISSING';
|
||||
if (point.state === 'MISSING' && reconstruction) return reconstructionPoint(reconstruction, month, query.mode === 'annual', now) || point;
|
||||
return point;
|
||||
});
|
||||
return { mode: query.mode, year: query.year, points: points.slice(1).map((p, i) => comparePoint(p, points[i], query.mode === 'annual')),
|
||||
years, subjects: [...subjects].sort((a, b) => a.localeCompare(b, 'zh-CN')), storageReady: archive.ready,
|
||||
enabledAt: archive.settings?.enabledAt || null, nextCaptureAt: archive.settings ? nextCapture(now) : null,
|
||||
warning: [!archive.ready ? '正式历史快照存储尚未部署。' : !archive.settings ? '自动采集尚未启用。' : null,
|
||||
this.reconstruction ? reconstruction ? '历史空白已按核对表回填业务推算值；历史总数待核实，不属于月末正式快照。回溯数据仅支持全量权限及全部公司。'
|
||||
: '回溯估算缺少可靠历史主体和权限归属，仅全量权限在全部公司下可见；当前筛选保留正式快照或空白。' : '未采集历史期间留空。'].filter(Boolean).join(' ') || null };
|
||||
}
|
||||
async details(user: AuthUser, params: Record<string, string>): Promise<HistoryDetails> {
|
||||
const now = this.clock();
|
||||
const month = params.month || '';
|
||||
if (!/^(20\d{2})-(0[1-9]|1[0-2])$/.test(month)) throw new HistoryInputError('月份无效');
|
||||
parseHistoryQuery({ ...params, year: month.slice(0, 4) }, now);
|
||||
const category = params.category || 'all';
|
||||
const page = Number(params.page || '1'), pageSize = Number(params.pageSize || '50');
|
||||
if (!['all', 'operating', 'inventory'].includes(category) || !Number.isSafeInteger(page) || page < 1 || page > 100000 || !Number.isSafeInteger(pageSize) || pageSize < 1 || pageSize > 100 || (params.search || '').length > 100 || (params.batchId || '').length > 36) throw new HistoryInputError('明细筛选或分页参数无效');
|
||||
const archive = await this.archives([month]);
|
||||
const batch = archive.batches.find(b => b.month === month);
|
||||
const isCurrent = month === shanghaiMonth(now) && now < monthCutoff(month);
|
||||
if (!isCurrent && makePoint(month, now, batch).state === 'MISSING') {
|
||||
const reconstruction = await this.reconstructionData(user, params.subject?.trim());
|
||||
const point = reconstruction && reconstructionPoint(reconstruction, month, false, now);
|
||||
if (point) {
|
||||
if (params.batchId && params.batchId !== point.batchId) throw new HistoryPeriodUnavailable('回溯版本已变化，请刷新趋势');
|
||||
const period = reconstruction!.periods.find(p => p.month === month)!;
|
||||
const search = (params.search || '').trim().toLocaleLowerCase();
|
||||
const items = reconstructionItems(period).filter(v => (category === 'all' || v.disposition === category) && (!search || `${v.plateNumber} ${v.vin}`.toLocaleLowerCase().includes(search)))
|
||||
.sort((a, b) => a.plateNumber.localeCompare(b.plateNumber, 'zh-CN') || a.id.localeCompare(b.id));
|
||||
return { point, counts: null, total: items.length, page, pageSize, items: items.slice((page - 1) * pageSize, page * pageSize) };
|
||||
}
|
||||
}
|
||||
if (params.batchId && (!batch || params.batchId !== batch.id || batch.status !== 'COMPLETE')) throw new HistoryPeriodUnavailable('所选快照不可用，请刷新趋势');
|
||||
const rows = isCurrent ? freezeLive(await this.live()) : batch?.status === 'COMPLETE' ? archive.vehicles.get(batch.id)! : [];
|
||||
const scoped = authorizedVehicles(rows, user, params.subject?.trim());
|
||||
const point = makePoint(month, now, batch, scoped, scoped);
|
||||
if (!point.detailAvailable) throw new HistoryPeriodUnavailable(HISTORY_STATE_LABELS[point.state]);
|
||||
const search = (params.search || '').trim().toLocaleLowerCase();
|
||||
const filtered = scoped.filter(v => (category === 'all' || v.disposition === category as HistoryCategory) && (!search || `${v.plateNumber} ${v.vin}`.toLocaleLowerCase().includes(search)))
|
||||
.sort((a, b) => a.plateNumber.localeCompare(b.plateNumber, 'zh-CN') || a.id.localeCompare(b.id));
|
||||
return { point, counts: counts(scoped), total: filtered.length, page, pageSize,
|
||||
items: filtered.slice((page - 1) * pageSize, page * pageSize).map(v => ({ id: v.id, plateNumber: v.plateNumber, vin: v.vin, type: v.type, model: v.model,
|
||||
disposition: v.disposition, operationStatus: v.operationStatus, vehicleStatus: v.vehicleStatus, subjectOrg: v.subjectOrg,
|
||||
departmentName: v.departmentName, customerManager: v.customerManager, customerName: v.customerName, contractNo: v.contractNo })) };
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import type { Pool } from 'mysql2/promise';
|
||||
import { MysqlHistoryStore } from './store.js';
|
||||
import { MAIN_SQL } from '../vehicles/repository.js';
|
||||
import { HistoryStorageUnavailable } from './model.js';
|
||||
import { testRow } from './fixtures.js';
|
||||
|
||||
/** Isolated transactional database double: failed transactions discard ALL staged writes. */
|
||||
class Database {
|
||||
commands:string[]=[];
|
||||
rows=[testRow('9223372036854775000'),testRow('stock','3')];
|
||||
settings='2026-10-19T13:00:00+08:00';
|
||||
locked=1; existing: Array<{id:string;status:string}>=[];
|
||||
committed:string[]=[]; staged:string[]=[];
|
||||
failDetails=0; failReads=false; failCommitAck=false; beforeLock?:()=>void; afterSource?:()=>void; onReadTransaction?:()=>void;
|
||||
released=false; clock=()=>new Date('2026-10-31T17:00:00.500+08:00');
|
||||
async query(sql:string,_args?:unknown[]) {
|
||||
this.commands.push(sql);
|
||||
if(sql.includes('GET_LOCK')){this.beforeLock?.();return[[{acquired:this.locked}]];}
|
||||
if(sql.includes('RELEASE_LOCK'))return[[{released:1}]];
|
||||
if(sql.includes('FROM bi_asset_history_settings'))return[[{enabledAt:this.settings}]];
|
||||
if(sql.includes('SELECT id,status'))return[this.existing];
|
||||
if(sql.startsWith('START TRANSACTION'))this.onReadTransaction?.();
|
||||
if(sql===MAIN_SQL){if(this.failReads)throw new Error('read failed');this.afterSource?.();return[this.rows];}
|
||||
if(sql.startsWith('INSERT INTO bi_asset_history_vehicles')){
|
||||
this.staged.push('details');if(this.failDetails-->0)throw new Error('write failed');
|
||||
}
|
||||
if(sql.startsWith('INSERT INTO bi_asset_history_batches'))this.staged.push('batch');
|
||||
if(sql.startsWith('UPDATE bi_asset_history_batches'))this.staged.push('complete');
|
||||
return[[]];
|
||||
}
|
||||
async getConnection(){return this;}
|
||||
async beginTransaction(){this.commands.push('BEGIN');this.staged=[];}
|
||||
async rollback(){this.commands.push('ROLLBACK');this.staged=[];}
|
||||
async commit(){this.commands.push('COMMIT');this.committed.push(...this.staged);this.staged=[];if(this.failCommitAck){this.failCommitAck=false;this.existing=[{id:'committed',status:'COMPLETE'}];throw new Error('commit acknowledgement lost');}}
|
||||
release(){this.released=true;}
|
||||
store(){return new MysqlHistoryStore(this as unknown as Pool);}
|
||||
}
|
||||
test('capture uses uncached shared SQL and one consistent read; detail/header publish atomically',async()=>{
|
||||
const db=new Database();assert.equal(await db.store().capture('2026-10',db.clock),'captured');
|
||||
assert.deepEqual(db.committed,['batch','details','complete']);
|
||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||
assert.ok(db.commands.indexOf('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY')<db.commands.indexOf(MAIN_SQL));
|
||||
assert.ok(db.commands.indexOf(MAIN_SQL)<db.commands.indexOf('ROLLBACK'));
|
||||
assert.equal(db.released,true);assert.ok(db.commands.at(-1)!.includes('RELEASE_LOCK'));
|
||||
});
|
||||
test('write retry preserves the exact captured rows and time, even after the minute expires',async()=>{
|
||||
const db=new Database();db.failDetails=1;let now=new Date('2026-10-31T17:00:30+08:00');
|
||||
db.afterSource=()=>{now=new Date('2026-10-31T18:00:00+08:00');};const clock=()=>now;
|
||||
assert.equal(await db.store().capture('2026-10',clock),'captured');
|
||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||
assert.deepEqual(db.committed,['batch','details','complete']);
|
||||
assert.equal(db.commands.filter(s=>s==='BEGIN').length,2);
|
||||
});
|
||||
test('lost commit acknowledgement does not overwrite or duplicate an already-published batch',async()=>{
|
||||
const db=new Database();db.failCommitAck=true;
|
||||
assert.equal(await db.store().capture('2026-10',db.clock),'already-complete');
|
||||
assert.deepEqual(db.committed,['batch','details','complete']);assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||
});
|
||||
test('source is not read if transaction establishment crosses the capture deadline',async()=>{
|
||||
const db=new Database();let now=new Date('2026-10-31T17:00:59.900+08:00');
|
||||
db.onReadTransaction=()=>{now=new Date('2026-10-31T17:01:00+08:00');};
|
||||
assert.equal(await db.store().capture('2026-10',()=>now),'outside-window');
|
||||
assert.equal(db.commands.includes(MAIN_SQL),false);assert.deepEqual(db.committed,[]);
|
||||
});
|
||||
test('persistent detail-write failure publishes neither partial header nor rows',async()=>{
|
||||
const db=new Database();db.failDetails=5;
|
||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/write failed/);
|
||||
assert.deepEqual(db.committed,[]);assert.deepEqual(db.staged,[]);assert.equal(db.released,true);
|
||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||
});
|
||||
test('source failure and invalid rows cannot publish snapshots',async()=>{
|
||||
const db=new Database();db.failReads=true;
|
||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/read failed/);assert.deepEqual(db.committed,[]);
|
||||
db.failReads=false;db.rows=[testRow('duplicate'),testRow('duplicate')];
|
||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/重复/);assert.deepEqual(db.committed,[]);
|
||||
});
|
||||
test('completed period, concurrent lock and expired window cannot issue a source read',async()=>{
|
||||
for(const scenario of ['complete','lock','expired','delayed-lock','late-activation']){
|
||||
const db=new Database();let clock=db.clock;
|
||||
if(scenario==='complete')db.existing=[{id:'existing',status:'COMPLETE'}];
|
||||
if(scenario==='lock')db.locked=0;
|
||||
if(scenario==='expired')clock=()=>new Date('2026-10-31T17:01:00+08:00');
|
||||
if(scenario==='delayed-lock')db.beforeLock=()=>{clock=()=>new Date('2026-10-31T17:01:00+08:00');};
|
||||
if(scenario==='late-activation')db.settings='2026-10-31T17:00:01+08:00';
|
||||
const result=await db.store().capture('2026-10',()=>clock());
|
||||
assert.ok(['already-complete','locked','outside-window'].includes(result));
|
||||
assert.equal(db.commands.includes(MAIN_SQL),false,scenario);assert.deepEqual(db.committed,[]);
|
||||
}
|
||||
});
|
||||
test('a legitimate empty source commits a zero-count complete batch without detail inserts',async()=>{
|
||||
const db=new Database();db.rows=[];
|
||||
assert.equal(await db.store().capture('2026-10',db.clock),'captured');assert.deepEqual(db.committed,['batch','complete']);
|
||||
});
|
||||
test('missing storage reads never issue schema writes and report storage-unavailable',async()=>{
|
||||
const calls:string[]=[];
|
||||
const db={query:async(sql:string)=>{calls.push(sql);throw{code:'ER_NO_SUCH_TABLE'};}};
|
||||
const store=new MysqlHistoryStore(db as unknown as Pool);
|
||||
await assert.rejects(()=>store.settings(),HistoryStorageUnavailable);
|
||||
await assert.rejects(()=>store.batches(),HistoryStorageUnavailable);
|
||||
assert.ok(calls.every(sql=>sql.startsWith('SELECT')));
|
||||
});
|
||||
@@ -0,0 +1,125 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import type { Pool, PoolConnection, RowDataPacket } from 'mysql2/promise';
|
||||
import { CAPTURE_WINDOW_MS, HISTORY_VERSION, monthCutoff, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
||||
import type { VehicleRow } from '../../types.js';
|
||||
import { MAIN_SQL } from '../vehicles/repository.js';
|
||||
import { counts, freezeRows, HistoryStorageUnavailable, type HistoryBatch, type HistorySettings, type HistoryVehicle } from './model.js';
|
||||
|
||||
export interface HistoryReader {
|
||||
settings(): Promise<HistorySettings | null>;
|
||||
batches(): Promise<HistoryBatch[]>;
|
||||
vehicles(batchIds: string[]): Promise<Map<string, HistoryVehicle[]>>;
|
||||
}
|
||||
export type CaptureResult = 'captured' | 'already-complete' | 'locked' | 'outside-window';
|
||||
const sqlTime = (date: Date) => shanghaiTimestamp(date).slice(0, 23).replace('T', ' ');
|
||||
const dateColumn = (name: string) => `DATE_FORMAT(${name}, '%Y-%m-%dT%H:%i:%s.%f+08:00')`;
|
||||
|
||||
/** No schema writes in constructor or any read. All lifecycle writes are explicit. */
|
||||
export class MysqlHistoryStore implements HistoryReader {
|
||||
constructor(private readonly db: Pick<Pool, 'query' | 'getConnection'>) {}
|
||||
|
||||
private async read(sql: string, args: unknown[] = []): Promise<RowDataPacket[]> {
|
||||
try { const [rows] = await this.db.query<RowDataPacket[]>(sql, args); return rows; }
|
||||
catch (error) {
|
||||
if ((error as { code?: string }).code === 'ER_NO_SUCH_TABLE') throw new HistoryStorageUnavailable('历史快照存储尚未部署');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
async settings(): Promise<HistorySettings | null> {
|
||||
const rows = await this.read(`SELECT ${dateColumn('enabled_at')} AS enabledAt FROM bi_asset_history_settings WHERE id=1`);
|
||||
return rows[0] ? { enabledAt: rows[0].enabledAt } : null;
|
||||
}
|
||||
async batches(): Promise<HistoryBatch[]> {
|
||||
const rows = await this.read(`SELECT id, period_month AS month, status, source,
|
||||
total_count,operating_count,inventory_count,
|
||||
${dateColumn('scheduled_for')} AS scheduledFor, ${dateColumn('captured_at')} AS capturedAt
|
||||
FROM bi_asset_history_batches WHERE definition_version=? ORDER BY period_month`, [HISTORY_VERSION]);
|
||||
return rows.map(row => ({ id: row.id, month: row.month, status: row.status, source: row.source, scheduledFor: row.scheduledFor, capturedAt: row.capturedAt,
|
||||
counts: { total: row.total_count, operating: row.operating_count, inventory: row.inventory_count } }));
|
||||
}
|
||||
async vehicles(batchIds: string[]): Promise<Map<string, HistoryVehicle[]>> {
|
||||
const result = new Map<string, HistoryVehicle[]>(batchIds.map(id => [id, []]));
|
||||
if (!batchIds.length) return result;
|
||||
const rows = await this.read(`SELECT batch_id, frozen_vehicle FROM bi_asset_history_vehicles WHERE batch_id IN (${batchIds.map(() => '?').join(',')}) ORDER BY vehicle_id`, batchIds);
|
||||
for (const row of rows) {
|
||||
const value = typeof row.frozen_vehicle === 'string' ? JSON.parse(row.frozen_vehicle) : row.frozen_vehicle;
|
||||
if (!value || typeof value.id !== 'string') throw new Error('Invalid archived vehicle');
|
||||
result.get(row.batch_id)!.push(value as HistoryVehicle);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
async activate(now: Date): Promise<HistorySettings> {
|
||||
await this.db.query('INSERT IGNORE INTO bi_asset_history_settings (id,enabled_at) VALUES (1,?)', [sqlTime(now)]);
|
||||
return (await this.settings())!;
|
||||
}
|
||||
async recordOutcome(month: string, status: 'FAILED' | 'MISSED', code: string): Promise<void> {
|
||||
// Never downgrade a completed snapshot, or replace a useful failure with MISSED.
|
||||
await this.db.query(`INSERT INTO bi_asset_history_batches
|
||||
(id,period_month,definition_version,scheduled_for,source,status,failure_code)
|
||||
VALUES (?,?,?,?,'AUTO',?,?) ON DUPLICATE KEY UPDATE
|
||||
failure_code=IF(status='COMPLETE' OR (status='FAILED' AND VALUES(status)='MISSED'),failure_code,VALUES(failure_code)),
|
||||
status=IF(status='COMPLETE' OR (status='FAILED' AND VALUES(status)='MISSED'),status,VALUES(status))`,
|
||||
[randomUUID(), month, HISTORY_VERSION, sqlTime(monthCutoff(month)), status, code]);
|
||||
}
|
||||
async capture(month: string, clock: () => Date = () => new Date()): Promise<CaptureResult> {
|
||||
const cutoff = monthCutoff(month).getTime();
|
||||
const inWindow = () => clock().getTime() >= cutoff && clock().getTime() < cutoff + CAPTURE_WINDOW_MS;
|
||||
if (!inWindow()) return 'outside-window';
|
||||
const connection = await this.db.getConnection();
|
||||
const lock = `ln-bi:asset:${HISTORY_VERSION}:${month}`;
|
||||
let locked = false;
|
||||
try {
|
||||
const [locks] = await connection.query<RowDataPacket[]>("SELECT GET_LOCK(CONCAT(DATABASE(),':',?),0) AS acquired", [lock]);
|
||||
locked = Number(locks[0]?.acquired) === 1;
|
||||
if (!locked) return 'locked';
|
||||
const settings = await this.settings();
|
||||
if (!settings || new Date(settings.enabledAt).getTime() > cutoff || !inWindow()) return 'outside-window';
|
||||
const [existing] = await connection.query<RowDataPacket[]>(`SELECT id,status FROM bi_asset_history_batches WHERE period_month=? AND definition_version=?`, [month, HISTORY_VERSION]);
|
||||
if (existing[0]?.status === 'COMPLETE') return 'already-complete';
|
||||
await connection.query("SET time_zone='+08:00'");
|
||||
await connection.query('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
||||
if (!inWindow()) return 'outside-window';
|
||||
await connection.query('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY');
|
||||
const capturedAt = clock();
|
||||
let frozen: HistoryVehicle[];
|
||||
try {
|
||||
if (!inWindow()) return 'outside-window';
|
||||
const [rows] = await connection.query<RowDataPacket[]>(MAIN_SQL);
|
||||
frozen = freezeRows(rows as VehicleRow[]);
|
||||
} finally { await connection.rollback(); }
|
||||
// Retries below keep this exact frozen dataset. Never reread a later live ledger.
|
||||
const id = existing[0]?.id || randomUUID();
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
try { return await this.publish(connection, id, month, capturedAt, frozen, clock); }
|
||||
catch (error) { await connection.rollback(); if (attempt >= 2) throw error; }
|
||||
}
|
||||
} finally {
|
||||
try { if (locked) await connection.query("SELECT RELEASE_LOCK(CONCAT(DATABASE(),':',?))", [lock]); }
|
||||
finally { connection.release(); }
|
||||
}
|
||||
}
|
||||
private async publish(connection: PoolConnection, id: string, month: string, capturedAt: Date, vehicles: HistoryVehicle[], clock: () => Date) {
|
||||
const totals = counts(vehicles);
|
||||
await connection.beginTransaction();
|
||||
const [existing] = await connection.query<RowDataPacket[]>('SELECT id,status FROM bi_asset_history_batches WHERE period_month=? AND definition_version=? FOR UPDATE', [month, HISTORY_VERSION]);
|
||||
// Also handles a committed transaction whose acknowledgement was lost.
|
||||
if (existing[0]?.status === 'COMPLETE') { await connection.commit(); return 'already-complete' as const; }
|
||||
if (existing[0]) id = existing[0].id;
|
||||
// Under the advisory lock, only FAILED/MISSED placeholders may be promoted.
|
||||
await connection.query(`INSERT INTO bi_asset_history_batches
|
||||
(id,period_month,definition_version,scheduled_for,captured_at,source,status,total_count,operating_count,inventory_count)
|
||||
VALUES (?,?,?,?,?,'AUTO','FAILED',?,?,?) ON DUPLICATE KEY UPDATE captured_at=VALUES(captured_at),
|
||||
total_count=VALUES(total_count),operating_count=VALUES(operating_count),inventory_count=VALUES(inventory_count)`,
|
||||
[id, month, HISTORY_VERSION, sqlTime(monthCutoff(month)), sqlTime(capturedAt), totals.total, totals.operating, totals.inventory]);
|
||||
for (let offset = 0; offset < vehicles.length; offset += 100) {
|
||||
const chunk = vehicles.slice(offset, offset + 100);
|
||||
await connection.query(`INSERT INTO bi_asset_history_vehicles
|
||||
(batch_id,vehicle_id,operation_status,vehicle_status,disposition,subject_org,department_name,manager_id,frozen_vehicle)
|
||||
VALUES ${chunk.map(() => '(?,?,?,?,?,?,?,?,CAST(? AS JSON))').join(',')}`, chunk.flatMap(v =>
|
||||
[id, v.id, v.operationStatus, v.vehicleStatus, v.disposition, v.subjectOrg, v.departmentName, v.managerId, JSON.stringify(v)]));
|
||||
}
|
||||
await connection.query("UPDATE bi_asset_history_batches SET status='COMPLETE',completed_at=?,failure_code=NULL WHERE id=?", [sqlTime(clock()), id]);
|
||||
await connection.commit();
|
||||
return 'captured' as const;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user