Merge daily asset snapshots and bump version to 1.1.22
ci/woodpecker/push/woodpecker Pipeline was canceled
ci/woodpecker/push/woodpecker Pipeline was canceled
This commit is contained in:
+22
-12
@@ -14,7 +14,7 @@
|
|||||||
|
|
||||||
- 复用资产仓储 `MAIN_SQL`、`transformRow`、`isOperatingVehicle`、`isInventoryVehicle`。
|
- 复用资产仓储 `MAIN_SQL`、`transformRow`、`isOperatingVehicle`、`isInventoryVehicle`。
|
||||||
- 总数为未删除、未退出运营的车辆;运营为租赁/自营,库存为可运营/待运营;异动、待交车不另加。
|
- 总数为未删除、未退出运营的车辆;运营为租赁/自营,库存为可运营/待运营;异动、待交车不另加。
|
||||||
- 月末最后一天 `Asia/Shanghai` 17:00 为计划时点;只允许在 17:00:00(含)到 17:01:00(不含)内开始读取。
|
- 每天 `Asia/Shanghai` 23:55 为计划时点,归属当天;只允许在 23:55:00(含)到 23:56:00(不含)内开始读取。每天独立保存完整逐车状态;月度展示月末当天、年度展示12月31日快照,不累加每日存量。
|
||||||
- `captured_at` 是程序记录的实际开始读取时间,`scheduled_for` 是计划时点;两者不宣称完全相同。
|
- `captured_at` 是程序记录的实际开始读取时间,`scheduled_for` 是计划时点;两者不宣称完全相同。
|
||||||
- 一致性读取可在窗口内开始后继续执行;持久化失败最多重试三次,始终使用同一份已取得数据,不重新读晚到的台账。
|
- 一致性读取可在窗口内开始后继续执行;持久化失败最多重试三次,始终使用同一份已取得数据,不重新读晚到的台账。
|
||||||
- 当月在截止前显示实时点;年末前当年年度点显示实时点。截止后只能显示归档、归档中、失败或缺失。12 月快照直接用作年度年末点。
|
- 当月在截止前显示实时点;年末前当年年度点显示实时点。截止后只能显示归档、归档中、失败或缺失。12 月快照直接用作年度年末点。
|
||||||
@@ -33,9 +33,9 @@
|
|||||||
|
|
||||||
## 存储
|
## 存储
|
||||||
|
|
||||||
`bi_asset_history_settings` 保存首次实际启用时间。重复启用保留首次时间。
|
`bi_asset_history_settings.enabled_at` 保留旧采集首次启用时间;`daily_enabled_at` 保存每日采集首次实际启用时间。重复启用保留首次时间,升级前的日期不会被标记为漏采。
|
||||||
|
|
||||||
`bi_asset_history_batches` 保存年月、口径版本、计划及实际时间、来源、状态、失败码、全量校验计数及导入追溯信息。年月与口径版本唯一。正式归档为 `COMPLETE`,失败为 `FAILED`,错过为 `MISSED`。
|
`bi_asset_history_batches` 保存期间键、口径版本、计划及实际时间、来源、状态、失败码、全量校验计数及导入追溯信息。`period_key` 与口径版本唯一;新快照使用 YYYY-MM-DD,旧月度 YYYY-MM 键和明细原样保留。月度查询仅选月末日及旧月度记录;同月优先新日快照,但失败或缺失不会遮蔽旧 COMPLETE 归档。正式归档为 `COMPLETE`,失败为 `FAILED`,错过为 `MISSED`。
|
||||||
|
|
||||||
`bi_asset_history_vehicles` 保存车辆字符串 ID、原始运营/车辆状态、归档分类、公司、部门、负责人 ID 和冻结车辆 JSON。批次内车辆唯一;不以当前车辆表为外键。JSON 保留车牌、VIN、型号、品牌、业务归属等明细信息,排除无关的异动工作详情和待交车附件信息。
|
`bi_asset_history_vehicles` 保存车辆字符串 ID、原始运营/车辆状态、归档分类、公司、部门、负责人 ID 和冻结车辆 JSON。批次内车辆唯一;不以当前车辆表为外键。JSON 保留车牌、VIN、型号、品牌、业务归属等明细信息,排除无关的异动工作详情和待交车附件信息。
|
||||||
|
|
||||||
@@ -73,27 +73,37 @@ node --import tsx scripts/migrate-asset-history.ts --apply
|
|||||||
|
|
||||||
如部分建表失败,修正原因后可重复执行;脚本使用 `CREATE TABLE IF NOT EXISTS`。新建空表不会伪造以前历史。
|
如部分建表失败,修正原因后可重复执行;脚本使用 `CREATE TABLE IF NOT EXISTS`。新建空表不会伪造以前历史。
|
||||||
|
|
||||||
授权启用后,将服务进程环境变量设为 `ASSET_HISTORY_CAPTURE_ENABLED=1` 并按现有流程重启服务。调度器首次成功连接时记录 `enabled_at`,启用前的时点不会追补。如果在月末 17:00 之后才首次启用,即使仍处于该分钟,也从下一个月末开始采集。
|
每日采集升级(2026-10-30):先停止旧版采集服务,执行基础建表迁移(已有表不变),再审阅并执行一次增量迁移:
|
||||||
|
|
||||||
按本需求确认日期 2026-10-19,如果在 2026-10-31 17:00 前成功启用,首个计划时点为北京时间 **2026-10-31 17:00**;实际计划以数据库首次启用时间和服务时间计算,不硬编码。
|
```sh
|
||||||
|
node --import tsx scripts/migrate-asset-history.ts --daily
|
||||||
|
node --import tsx scripts/migrate-asset-history.ts --daily --apply
|
||||||
|
```
|
||||||
|
|
||||||
保持服务在计划时点在线。停止期间错过的 eligible 月份会在重启时记为 MISSED;已 FAILED 的记录保留诊断标记。记录到启用前没有归档,不能用重启时的实时数回填。日志前缀为 `[asset-history]`,包含年月及 captured/already-complete/locked/failed/missed 结果,不记录客户明细或连接凭据。
|
增量迁移把 `period_month CHAR(7)` 改为 `period_key CHAR(10)`,原唯一索引继续约束期间键和版本;新增可空 `daily_enabled_at`。这是一次性 ALTER,不能重复执行。应在目标库完成迁移后启动新版服务并设置采集开关;旧版程序不能直接使用新字段名。迁移不启动采集,也不修改旧归档的计划/实际时间。当前页面仍提供月度和年度浏览,普通日期快照保存在数据库中。
|
||||||
|
|
||||||
建议仅让受控服务账号写快照表;源表仍只需读权限。多实例可启用同一数据库下的同一采集配置,命名锁和数据库唯一约束共同保证幂等。启用开关为进程级配置,`enabled_at` 是历史启用记录,不能仅根据该记录判断某个进程现在是否运行调度器。
|
|
||||||
|
授权启用后,将服务进程环境变量设为 `ASSET_HISTORY_CAPTURE_ENABLED=1` 并按现有流程重启服务。调度器首次成功连接时记录 `daily_enabled_at`,启用前的时点不会追补。如果在当天 23:55 之后才首次启用,即使仍处于该分钟,也从次日开始采集。
|
||||||
|
|
||||||
|
按每日采集需求确认日期 2026-10-30,如果在当天 23:55 前成功启用,首个计划时点为北京时间 **2026-10-30 23:55**;实际计划以每日采集首次启用时间和服务时间计算,不硬编码。
|
||||||
|
|
||||||
|
保持服务在计划时点在线。停止期间错过的 eligible 日期会在重启时记为 MISSED;已 FAILED 的记录保留诊断标记。记录到启用前没有归档,不能用重启时的实时数回填。日志前缀为 `[asset-history]`,包含日期及 captured/already-complete/locked/failed/missed 结果,不记录客户明细或连接凭据。
|
||||||
|
|
||||||
|
建议仅让受控服务账号写快照表;源表仍只需读权限。多实例可启用同一数据库下的同一采集配置,命名锁和数据库唯一约束共同保证幂等。启用开关为进程级配置,`daily_enabled_at` 是历史启用记录,不能仅根据该记录判断某个进程现在是否运行调度器。
|
||||||
|
|
||||||
管理员可用以下只读 SQL 检查,不通过普通用户接口暴露全量计数:
|
管理员可用以下只读 SQL 检查,不通过普通用户接口暴露全量计数:
|
||||||
|
|
||||||
```sql
|
```sql
|
||||||
SELECT enabled_at FROM bi_asset_history_settings WHERE id=1;
|
SELECT enabled_at FROM bi_asset_history_settings WHERE id=1;
|
||||||
SELECT period_month, scheduled_for, captured_at, completed_at,
|
SELECT period_key, scheduled_for, captured_at, completed_at,
|
||||||
definition_version, source, status, failure_code,
|
definition_version, source, status, failure_code,
|
||||||
total_count, operating_count, inventory_count
|
total_count, operating_count, inventory_count
|
||||||
FROM bi_asset_history_batches ORDER BY period_month DESC LIMIT 12;
|
FROM bi_asset_history_batches ORDER BY period_key DESC LIMIT 31;
|
||||||
SELECT b.period_month, b.total_count, COUNT(v.vehicle_id) AS detail_count
|
SELECT b.period_key, b.total_count, COUNT(v.vehicle_id) AS detail_count
|
||||||
FROM bi_asset_history_batches b
|
FROM bi_asset_history_batches b
|
||||||
LEFT JOIN bi_asset_history_vehicles v ON v.batch_id=b.id
|
LEFT JOIN bi_asset_history_vehicles v ON v.batch_id=b.id
|
||||||
WHERE b.status='COMPLETE'
|
WHERE b.status='COMPLETE'
|
||||||
GROUP BY b.id, b.period_month, b.total_count;
|
GROUP BY b.id, b.period_key, b.total_count;
|
||||||
```
|
```
|
||||||
|
|
||||||
## 回退
|
## 回退
|
||||||
@@ -108,7 +118,7 @@ GROUP BY b.id, b.period_month, b.total_count;
|
|||||||
|
|
||||||
提供仅表头模板 `docs/asset-history-import-template.csv`。使用 UTF-8 CSV,每行一辆车,同一文件只允许一个年月、一个口径版本和一个统计时点。
|
提供仅表头模板 `docs/asset-history-import-template.csv`。使用 UTF-8 CSV,每行一辆车,同一文件只允许一个年月、一个口径版本和一个统计时点。
|
||||||
|
|
||||||
必填字段:`period_month`、`scheduled_for`、`captured_at`、`definition_version`、`source_reference`、`vehicle_id`、`operation_status`。空缺车牌/VIN使用空值,不造车辆 ID。ID 均按文本读取。`scheduled_for` 必须是指定年月最后一天北京时间 17:00,`captured_at` 必须来自可信历史来源,不能把导入执行时间当作历史读取时间。
|
必填字段:`period_month`、`scheduled_for`、`captured_at`、`definition_version`、`source_reference`、`vehicle_id`、`operation_status`。空缺车牌/VIN使用空值,不造车辆 ID。ID 均按文本读取。旧版月度导入契约的 `scheduled_for` 是指定年月最后一天北京时间 17:00(保留旧格式,不能作为每日自动采集规则),`captured_at` 必须来自可信历史来源,不能把导入执行时间当作历史读取时间。
|
||||||
|
|
||||||
其余字段见模板。部门、负责人、主体缺失须显式留空,不能用今日台账补齐;部门/个人权限因此可能看不到未归属的运营车辆,这是数据缺失而非自动扩大授权的理由。展示型号、品牌、原始车辆状态按历史来源保存;历史状态映射复用现有分类函数,不接受退出运营车辆或未知运营状态作为有效存量。归档分类由导入校验程序计算,不直接信任文件中的分类列。
|
其余字段见模板。部门、负责人、主体缺失须显式留空,不能用今日台账补齐;部门/个人权限因此可能看不到未归属的运营车辆,这是数据缺失而非自动扩大授权的理由。展示型号、品牌、原始车辆状态按历史来源保存;历史状态映射复用现有分类函数,不接受退出运营车辆或未知运营状态作为有效存量。归档分类由导入校验程序计算,不直接信任文件中的分类列。
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,4 @@
|
|||||||
|
-- Run once after 20261019_asset_history.sql; stop old capture workers before migration.
|
||||||
|
-- Existing YYYY-MM keys and archived vehicle rows remain unchanged.
|
||||||
|
ALTER TABLE bi_asset_history_batches CHANGE COLUMN period_month period_key CHAR(10) NOT NULL;
|
||||||
|
ALTER TABLE bi_asset_history_settings ADD COLUMN daily_enabled_at DATETIME(3) NULL;
|
||||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
|||||||
{
|
{
|
||||||
"name": "ln-bi",
|
"name": "ln-bi",
|
||||||
"version": "1.1.21",
|
"version": "1.1.22",
|
||||||
"lockfileVersion": 3,
|
"lockfileVersion": 3,
|
||||||
"requires": true,
|
"requires": true,
|
||||||
"packages": {
|
"packages": {
|
||||||
"": {
|
"": {
|
||||||
"name": "ln-bi",
|
"name": "ln-bi",
|
||||||
"version": "1.1.21",
|
"version": "1.1.22",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"@amap/amap-jsapi-loader": "^1.0.1",
|
"@amap/amap-jsapi-loader": "^1.0.1",
|
||||||
"@hono/node-server": "^1.13.0",
|
"@hono/node-server": "^1.13.0",
|
||||||
|
|||||||
+1
-1
@@ -1,7 +1,7 @@
|
|||||||
{
|
{
|
||||||
"name": "ln-bi",
|
"name": "ln-bi",
|
||||||
"private": true,
|
"private": true,
|
||||||
"version": "1.1.21",
|
"version": "1.1.22",
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"dev": "concurrently -n server,client -c blue,green \"npm run dev:server\" \"npm run dev:client\"",
|
"dev": "concurrently -n server,client -c blue,green \"npm run dev:server\" \"npm run dev:client\"",
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import { readFile } from 'node:fs/promises';
|
import { readFile } from 'node:fs/promises';
|
||||||
const sql = await readFile(new URL('../migrations/20261019_asset_history.sql', import.meta.url), 'utf8');
|
const migration = process.argv.includes('--daily') ? '20261030_asset_history_daily.sql' : '20261019_asset_history.sql';
|
||||||
|
const sql = await readFile(new URL(`../migrations/${migration}`, import.meta.url), 'utf8');
|
||||||
if (!process.argv.includes('--apply')) {
|
if (!process.argv.includes('--apply')) {
|
||||||
console.log(sql);
|
console.log(sql);
|
||||||
console.log('Review only. Pass --apply only after approving the configured database migration.');
|
console.log('Review only. Pass --apply only after approving the configured database migration.');
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ export function AssetHistorySection({ subject, onSubjects }: { subject: string |
|
|||||||
<div className="asset-history-toolbar">
|
<div className="asset-history-toolbar">
|
||||||
{mode === 'monthly' && <HistorySelect label="历史年份" value={String(year)} options={(data?.years || [year]).map(y => ({ value: String(y), label: `${y}年` }))} onChange={value => { setYear(Number(value)); setDetails(null); }} />}
|
{mode === 'monthly' && <HistorySelect label="历史年份" value={String(year)} options={(data?.years || [year]).map(y => ({ value: String(y), label: `${y}年` }))} onChange={value => { setYear(Number(value)); setDetails(null); }} />}
|
||||||
{data && <HistorySelect label="统计期间" value={selected || ''} onChange={value => { setSelected(value); setDetails(null); }} options={data.points.map(p => ({ value: p.period, label: p.label, detail: p.state === 'RECONSTRUCTED' ? '估算' : p.state === 'LIVE' ? '实时' : p.state === 'ARCHIVED' ? '归档' : p.state === 'FUTURE' ? '未来' : p.state === 'MISSING' ? '缺失' : HISTORY_STATE_LABELS[p.state] }))} />}
|
{data && <HistorySelect label="统计期间" value={selected || ''} onChange={value => { setSelected(value); setDetails(null); }} options={data.points.map(p => ({ value: p.period, label: p.label, detail: p.state === 'RECONSTRUCTED' ? '估算' : p.state === 'LIVE' ? '实时' : p.state === 'ARCHIVED' ? '归档' : p.state === 'FUTURE' ? '未来' : p.state === 'MISSING' ? '缺失' : HISTORY_STATE_LABELS[p.state] }))} />}
|
||||||
<span className="asset-history-cutoff" title="每月最后一天北京时间17:00采集;年度复用12月快照">月末 17:00</span>
|
<span className="asset-history-cutoff" title="每天北京时间23:55采集;月度取月末、年度取年末快照">每日 23:55</span>
|
||||||
<button type="button" className="asset-history-refresh" aria-label="刷新资产历史趋势" disabled={loading} onClick={() => setRetry(x => x + 1)}><RefreshCw size={15} /></button>
|
<button type="button" className="asset-history-refresh" aria-label="刷新资产历史趋势" disabled={loading} onClick={() => setRetry(x => x + 1)}><RefreshCw size={15} /></button>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
@@ -1,17 +1,17 @@
|
|||||||
import test from 'node:test';
|
import test from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
import { execFileSync } from 'node:child_process';
|
import { execFileSync } from 'node:child_process';
|
||||||
import { monthCutoff, nextCapture, shanghaiMonth, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
import { dayCutoff, shiftDay, monthCutoff, nextCapture, shanghaiMonth, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
||||||
import { counts, freezeRows, makePoint, freezeLive } from './model.js';
|
import { counts, freezeRows, makePoint, freezeLive } from './model.js';
|
||||||
import { testBatch, testRow, testVehicle } from './fixtures.js';
|
import { testBatch, testRow, testVehicle } from './fixtures.js';
|
||||||
|
|
||||||
test('Shanghai month-end cutoff covers leap years, short/long months and cross-year boundaries', () => {
|
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']]) {
|
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(shanghaiTimestamp(monthCutoff(month)), `${expected}T23:55:00.000+08:00`);
|
||||||
assert.equal(monthCutoff(month).toISOString(), `${expected}T09:00:00.000Z`);
|
assert.equal(monthCutoff(month).toISOString(), `${expected}T15:55:00.000Z`);
|
||||||
}
|
}
|
||||||
assert.equal(shanghaiMonth(new Date('2026-12-31T16:00:00Z')), '2027-01');
|
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.equal(nextCapture(new Date('2026-12-31T15:55:00Z')), '2027-01-01T23:55:00.000+08:00');
|
||||||
assert.throws(() => monthCutoff('2026-13'));
|
assert.throws(() => monthCutoff('2026-13'));
|
||||||
});
|
});
|
||||||
test('host TZ cannot change cutoff or Shanghai calendar', () => {
|
test('host TZ cannot change cutoff or Shanghai calendar', () => {
|
||||||
@@ -32,11 +32,25 @@ test('source rows preserve precise IDs and raw states; pending and abnormal do n
|
|||||||
});
|
});
|
||||||
test('cutoff switches live to capturing/missing without fallback; a zero snapshot is archived', () => {
|
test('cutoff switches live to capturing/missing without fallback; a zero snapshot is archived', () => {
|
||||||
const live = freezeLive([testVehicle('1')]);
|
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-31T15:54:59Z'), undefined, [], live).state, 'LIVE');
|
||||||
assert.equal(makePoint('2026-10', new Date('2026-10-31T09:00:00Z'), undefined, [], live).state, 'CAPTURING');
|
assert.equal(makePoint('2026-10', new Date('2026-10-31T15:55:00Z'), undefined, [], live).state, 'CAPTURING');
|
||||||
const missing = makePoint('2026-10', new Date('2026-10-31T09:01:00Z'), undefined, [], live);
|
const missing = makePoint('2026-10', new Date('2026-10-31T15:56:00Z'), undefined, [], live);
|
||||||
assert.equal(missing.state, 'MISSING'); assert.equal(missing.counts, null);
|
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);
|
const archived = makePoint('2026-10', new Date('2026-10-31T15:55:20Z'), testBatch('2026-10'), [], live);
|
||||||
assert.equal(archived.state, 'ARCHIVED'); assert.deepEqual(archived.counts, { total:0,operating:0,inventory:0 });
|
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');
|
assert.equal(makePoint('2026-11', new Date('2026-10-19T00:00Z')).state, 'FUTURE');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('daily calendar handles leap days, year changes and next capture within a month', () => {
|
||||||
|
assert.equal(shiftDay('2024-02-28',1),'2024-02-29');
|
||||||
|
assert.equal(shiftDay('2026-02-28',1),'2026-03-01');
|
||||||
|
assert.equal(shiftDay('2026-12-31',1),'2027-01-01');
|
||||||
|
assert.equal(nextCapture(new Date('2026-10-30T23:54:59+08:00')),'2026-10-30T23:55:00.000+08:00');
|
||||||
|
assert.equal(nextCapture(new Date('2026-10-30T23:55:00+08:00')),'2026-10-31T23:55:00.000+08:00');
|
||||||
|
assert.throws(()=>dayCutoff('2026-02-29'));
|
||||||
|
});
|
||||||
|
test('legacy monthly archive keeps its original scheduled and captured times', () => {
|
||||||
|
const batch={...testBatch('2026-09'),scheduledFor:'2026-09-30T17:00:00+08:00',capturedAt:'2026-09-30T17:00:02+08:00'};
|
||||||
|
const point=makePoint('2026-09',new Date('2026-10-30T00:00:00+08:00'),batch,[]);
|
||||||
|
assert.equal(point.state,'ARCHIVED');assert.equal(point.scheduledFor,batch.scheduledFor);assert.equal(point.asOf,batch.capturedAt);
|
||||||
|
});
|
||||||
|
|||||||
@@ -47,7 +47,7 @@ export function counts(vehicles: Vehicle[]): HistoryCounts {
|
|||||||
return { total: vehicles.length, operating: vehicles.filter(isOperatingVehicle).length, inventory: vehicles.filter(isInventoryVehicle).length };
|
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 {
|
export function makePoint(month: string, now: Date, batch?: HistoryBatch, data?: HistoryVehicle[], live?: HistoryVehicle[], annual = false): HistoryPoint {
|
||||||
const cutoff = monthCutoff(month);
|
const cutoff = batch ? new Date(batch.scheduledFor) : monthCutoff(month);
|
||||||
let state: HistoryPoint['state'] = 'MISSING';
|
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';
|
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 === 'COMPLETE') state = 'ARCHIVED';
|
||||||
@@ -58,7 +58,7 @@ export function makePoint(month: string, now: Date, batch?: HistoryBatch, data?:
|
|||||||
return {
|
return {
|
||||||
period: annual ? month.slice(0, 4) : month,
|
period: annual ? month.slice(0, 4) : month,
|
||||||
label: annual ? `${month.slice(0, 4)}年` : `${Number(month.slice(5))}月`, month,
|
label: annual ? `${month.slice(0, 4)}年` : `${Number(month.slice(5))}月`, month,
|
||||||
scheduledFor: shanghaiTimestamp(cutoff),
|
scheduledFor: batch?.scheduledFor || shanghaiTimestamp(cutoff),
|
||||||
asOf: state === 'LIVE' ? shanghaiTimestamp(now) : state === 'ARCHIVED' ? batch!.capturedAt : null,
|
asOf: state === 'LIVE' ? shanghaiTimestamp(now) : state === 'ARCHIVED' ? batch!.capturedAt : null,
|
||||||
state, source: state === 'LIVE' ? 'LIVE' : state === 'ARCHIVED' ? batch!.source : null,
|
state, source: state === 'LIVE' ? 'LIVE' : state === 'ARCHIVED' ? batch!.source : null,
|
||||||
batchId: state === 'ARCHIVED' ? batch!.id : null, counts: pointCounts,
|
batchId: state === 'ARCHIVED' ? batch!.id : null, counts: pointCounts,
|
||||||
|
|||||||
@@ -1,42 +1,51 @@
|
|||||||
import test from 'node:test';
|
import test from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
import { HistoryScheduler, type HistoryCaptureStore } from './scheduler.js';
|
import { HistoryScheduler, type HistoryCaptureStore } from './scheduler.js';
|
||||||
import { testBatch } from './fixtures.js';
|
|
||||||
function fixture(enabledAt='2026-10-19T13:00:00+08:00') {
|
function fixture(enabledAt='2026-10-30T13:00:00+08:00') {
|
||||||
const captures:string[]=[],missed:string[]=[],failed:string[]=[];
|
const captures:string[]=[],missed:string[]=[],failed:string[]=[];
|
||||||
let activates=0;
|
let activates=0;
|
||||||
const store:HistoryCaptureStore={
|
const store:HistoryCaptureStore={
|
||||||
activate:async()=>{activates++;return{enabledAt};},batches:async()=>[],
|
activate:async()=>{activates++;return{enabledAt};},dailyBatches:async()=>[],
|
||||||
capture:async(month)=>{captures.push(month);return'captured';},
|
capture:async(month)=>{captures.push(month);return'captured';},
|
||||||
recordOutcome:async(month,status)=>{(status==='FAILED'?failed:missed).push(month);},
|
recordOutcome:async(month,status)=>{(status==='FAILED'?failed:missed).push(month);},
|
||||||
};
|
};
|
||||||
return {store,captures,missed,failed,activates:()=>activates};
|
return {store,captures,missed,failed,activates:()=>activates};
|
||||||
}
|
}
|
||||||
test('scheduler captures only in month-end 17:00 minute and activation is persisted once',async()=>{
|
test('scheduler captures only in daily 23:55 minute and activation is persisted once',async()=>{
|
||||||
const f=fixture();let date=new Date('2026-10-31T16:59:59+08:00');
|
const f=fixture();let date=new Date('2026-10-31T23:54:59+08:00');
|
||||||
const scheduler=new HistoryScheduler(f.store,()=>date,()=>{});
|
const scheduler=new HistoryScheduler(f.store,()=>date,()=>{});
|
||||||
await scheduler.tick();assert.deepEqual(f.captures,[]);
|
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);
|
date=new Date('2026-10-31T23:55:00+08:00');await scheduler.tick();assert.deepEqual(f.captures,['2026-10-31']);assert.equal(f.activates(),1);
|
||||||
});
|
});
|
||||||
test('restart records missed eligible months without backfilling old live data or pre-activation periods',async()=>{
|
test('restart records missed eligible days 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'),()=>{});
|
const f=fixture();const scheduler=new HistoryScheduler(f.store,()=>new Date('2026-11-02T08:00:00+08:00'),()=>{});
|
||||||
await scheduler.tick();assert.deepEqual(f.captures,[]);assert.deepEqual(f.missed,['2026-10','2026-11','2026-12']);
|
await scheduler.tick();assert.deepEqual(f.captures,[]);assert.deepEqual(f.missed,['2026-10-30','2026-10-31','2026-11-01']);
|
||||||
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();
|
const late=fixture('2026-10-31T23:55:10+08:00');await new HistoryScheduler(late.store,()=>new Date('2026-10-31T23:55:20+08:00'),()=>{}).tick();
|
||||||
assert.deepEqual(late.captures,[]);assert.deepEqual(late.missed,[]);
|
assert.deepEqual(late.captures,[]);assert.deepEqual(late.missed,[]);
|
||||||
});
|
});
|
||||||
test('completed December skips duplicate annual capture; failures stay marked without late retries',async()=>{
|
test('completed days skip capture and existing failures are preserved after the window',async()=>{
|
||||||
const f=fixture();f.store.batches=async()=>[testBatch('2026-10')];
|
const f=fixture();f.store.dailyBatches=async()=>[{day:'2026-10-30',status:'COMPLETE'}];
|
||||||
f.store.capture=async()=>{throw new Error('source unavailable');};
|
f.store.capture=async()=>{throw new Error('source unavailable');};
|
||||||
await new HistoryScheduler(f.store,()=>new Date('2026-12-31T17:00:20+08:00'),()=>{}).tick();
|
await new HistoryScheduler(f.store,()=>new Date('2026-10-31T23:55:20+08:00'),()=>{}).tick();
|
||||||
assert.deepEqual(f.missed,['2026-11']);assert.deepEqual(f.failed,['2026-12']);
|
assert.deepEqual(f.missed,[]);assert.deepEqual(f.failed,['2026-10-31']);
|
||||||
const done=fixture('2026-12-01T00:00+08:00');done.store.batches=async()=>[testBatch('2026-12')];
|
f.store.dailyBatches=async()=>[{day:'2026-10-30',status:'COMPLETE'},{day:'2026-10-31',status:'FAILED'}];
|
||||||
await new HistoryScheduler(done.store,()=>new Date('2026-12-31T17:00:20+08:00'),()=>{}).tick();assert.deepEqual(done.captures,[]);
|
await new HistoryScheduler(f.store,()=>new Date('2026-10-31T23:56:00+08:00'),()=>{}).tick();
|
||||||
|
assert.deepEqual(f.failed,['2026-10-31']);assert.deepEqual(f.missed,[]);
|
||||||
});
|
});
|
||||||
test('concurrent ticks and repeated start do not register duplicate jobs',async()=>{
|
test('concurrent ticks and repeated start do not register duplicate jobs',async()=>{
|
||||||
const f=fixture();let release!:()=>void;
|
const f=fixture();let release!:()=>void;
|
||||||
f.store.capture=async()=>{f.captures.push('2026-10');await new Promise<void>(r=>{release=r;});return'captured';};
|
f.store.capture=async()=>{f.captures.push('2026-10-31');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 scheduler=new HistoryScheduler(f.store,()=>new Date('2026-10-31T23:55:20+08:00'),()=>{});
|
||||||
const stop=scheduler.start();scheduler.start();
|
const stop=scheduler.start();scheduler.start();
|
||||||
await new Promise<void>(r=>setImmediate(r));await scheduler.tick();assert.equal(f.captures.length,1);
|
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));
|
stop();release();await new Promise<void>(r=>setImmediate(r));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('consecutive dates in one month and across month boundary capture independently',async()=>{
|
||||||
|
const f=fixture();
|
||||||
|
for(const day of ['2026-10-30','2026-10-31','2026-11-01']) {
|
||||||
|
await new HistoryScheduler(f.store,()=>new Date(`${day}T23:55:00+08:00`),()=>{}).tick();
|
||||||
|
}
|
||||||
|
assert.deepEqual(f.captures,['2026-10-30','2026-10-31','2026-11-01']);
|
||||||
|
});
|
||||||
|
|||||||
@@ -1,12 +1,12 @@
|
|||||||
import { CAPTURE_WINDOW_MS, monthCutoff, shanghaiMonth, shiftMonth } from '../../../shared/assets/history.js';
|
import { CAPTURE_WINDOW_MS, dayCutoff, shanghaiDay, shiftDay } from '../../../shared/assets/history.js';
|
||||||
import type { CaptureResult } from './store.js';
|
import type { CaptureResult } from './store.js';
|
||||||
import { HistoryDataError, type HistoryBatch, type HistorySettings } from './model.js';
|
import { HistoryDataError, type HistoryBatch, type HistorySettings } from './model.js';
|
||||||
|
|
||||||
export interface HistoryCaptureStore {
|
export interface HistoryCaptureStore {
|
||||||
activate(now: Date): Promise<HistorySettings>;
|
activate(now: Date): Promise<HistorySettings>;
|
||||||
batches(): Promise<HistoryBatch[]>;
|
dailyBatches(): Promise<Array<{ day: string; status: HistoryBatch['status'] }>>;
|
||||||
capture(month: string, clock: () => Date): Promise<CaptureResult>;
|
capture(day: string, clock: () => Date): Promise<CaptureResult>;
|
||||||
recordOutcome(month: string, status: 'MISSED' | 'FAILED', code: string): Promise<void>;
|
recordOutcome(day: string, status: 'MISSED' | 'FAILED', code: string): Promise<void>;
|
||||||
}
|
}
|
||||||
export class HistoryScheduler {
|
export class HistoryScheduler {
|
||||||
private timer: ReturnType<typeof setTimeout> | undefined;
|
private timer: ReturnType<typeof setTimeout> | undefined;
|
||||||
@@ -24,20 +24,20 @@ export class HistoryScheduler {
|
|||||||
const now = this.clock();
|
const now = this.clock();
|
||||||
this.settings ||= await this.store.activate(now);
|
this.settings ||= await this.store.activate(now);
|
||||||
const enabled = new Date(this.settings.enabledAt);
|
const enabled = new Date(this.settings.enabledAt);
|
||||||
const existing = new Map((await this.store.batches()).map(b => [b.month, b]));
|
const existing = new Map((await this.store.dailyBatches()).map(b => [b.day, b]));
|
||||||
for (let month = shanghaiMonth(enabled); month <= shanghaiMonth(now); month = shiftMonth(month, 1)) {
|
for (let day = shanghaiDay(enabled); day <= shanghaiDay(now); day = shiftDay(day, 1)) {
|
||||||
const cutoff = monthCutoff(month);
|
const cutoff = dayCutoff(day);
|
||||||
if (cutoff < enabled || cutoff > now || existing.get(month)?.status === 'COMPLETE') continue;
|
if (cutoff < enabled || cutoff > now || existing.get(day)?.status === 'COMPLETE') continue;
|
||||||
if (now.getTime() >= cutoff.getTime() + CAPTURE_WINDOW_MS) {
|
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}`); }
|
if (!existing.has(day)) { await this.store.recordOutcome(day, 'MISSED', 'CAPTURE_WINDOW_MISSED'); this.log(`[asset-history] missed ${day}`); }
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
try { const result = await this.store.capture(month, this.clock); this.log(`[asset-history] ${month}: ${result}`); }
|
try { const result = await this.store.capture(day, this.clock); this.log(`[asset-history] ${day}: ${result}`); }
|
||||||
catch (error) {
|
catch (error) {
|
||||||
const databaseCode = (error as { code?: unknown })?.code;
|
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';
|
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);
|
await this.store.recordOutcome(day, 'FAILED', code);
|
||||||
this.log(`[asset-history] failed ${month} (${code}); no later live data will be substituted`);
|
this.log(`[asset-history] failed ${day} (${code}); no later live data will be substituted`);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} finally { this.ticking = false; }
|
} finally { this.ticking = false; }
|
||||||
@@ -50,7 +50,7 @@ export class HistoryScheduler {
|
|||||||
try { await this.tick(); } catch { this.log('[asset-history] scheduler unavailable; check migration and database access'); }
|
try { await this.tick(); } catch { this.log('[asset-history] scheduler unavailable; check migration and database access'); }
|
||||||
if (!this.running) return;
|
if (!this.running) return;
|
||||||
const now = this.clock();
|
const now = this.clock();
|
||||||
const untilCutoff = monthCutoff(shanghaiMonth(now)).getTime() - now.getTime();
|
const untilCutoff = dayCutoff(shanghaiDay(now)).getTime() - now.getTime();
|
||||||
this.timer = setTimeout(run, untilCutoff > 0 ? Math.min(untilCutoff, 60_000) : 15_000);
|
this.timer = setTimeout(run, untilCutoff > 0 ? Math.min(untilCutoff, 60_000) : 15_000);
|
||||||
this.timer.unref?.();
|
this.timer.unref?.();
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -84,7 +84,7 @@ test('missing storage degrades to live without any DDL; genuine database errors
|
|||||||
});
|
});
|
||||||
test('once cutoff passes live values cannot replace missing history, even when capture was not enabled', async () => {
|
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 {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 service=new HistoryService(store,async()=>{liveCalls++;return[leased];},()=>new Date('2026-10-31T23:55:30+08:00'));
|
||||||
const result=await service.trend(user(),{});assert.equal(result.points[9].state,'MISSING');assert.equal(result.points[9].counts,null);
|
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);
|
assert.equal(liveCalls,0);await assert.rejects(()=>service.details(user(),{month:'2026-10'}),HistoryPeriodUnavailable);
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ class Database {
|
|||||||
locked=1; existing: Array<{id:string;status:string}>=[];
|
locked=1; existing: Array<{id:string;status:string}>=[];
|
||||||
committed:string[]=[]; staged:string[]=[];
|
committed:string[]=[]; staged:string[]=[];
|
||||||
failDetails=0; failReads=false; failCommitAck=false; beforeLock?:()=>void; afterSource?:()=>void; onReadTransaction?:()=>void;
|
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');
|
released=false; clock=()=>new Date('2026-10-31T23:55:00.500+08:00');
|
||||||
async query(sql:string,_args?:unknown[]) {
|
async query(sql:string,_args?:unknown[]) {
|
||||||
this.commands.push(sql);
|
this.commands.push(sql);
|
||||||
if(sql.includes('GET_LOCK')){this.beforeLock?.();return[[{acquired:this.locked}]];}
|
if(sql.includes('GET_LOCK')){this.beforeLock?.();return[[{acquired:this.locked}]];}
|
||||||
@@ -38,7 +38,7 @@ class Database {
|
|||||||
store(){return new MysqlHistoryStore(this as unknown as Pool);}
|
store(){return new MysqlHistoryStore(this as unknown as Pool);}
|
||||||
}
|
}
|
||||||
test('capture uses uncached shared SQL and one consistent read; detail/header publish atomically',async()=>{
|
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');
|
const db=new Database();assert.equal(await db.store().capture('2026-10-31',db.clock),'captured');
|
||||||
assert.deepEqual(db.committed,['batch','details','complete']);
|
assert.deepEqual(db.committed,['batch','details','complete']);
|
||||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
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('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY')<db.commands.indexOf(MAIN_SQL));
|
||||||
@@ -46,52 +46,52 @@ test('capture uses uncached shared SQL and one consistent read; detail/header pu
|
|||||||
assert.equal(db.released,true);assert.ok(db.commands.at(-1)!.includes('RELEASE_LOCK'));
|
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()=>{
|
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');
|
const db=new Database();db.failDetails=1;let now=new Date('2026-10-31T23:55:30+08:00');
|
||||||
db.afterSource=()=>{now=new Date('2026-10-31T18:00:00+08:00');};const clock=()=>now;
|
db.afterSource=()=>{now=new Date('2026-10-31T23:59:00+08:00');};const clock=()=>now;
|
||||||
assert.equal(await db.store().capture('2026-10',clock),'captured');
|
assert.equal(await db.store().capture('2026-10-31',clock),'captured');
|
||||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||||
assert.deepEqual(db.committed,['batch','details','complete']);
|
assert.deepEqual(db.committed,['batch','details','complete']);
|
||||||
assert.equal(db.commands.filter(s=>s==='BEGIN').length,2);
|
assert.equal(db.commands.filter(s=>s==='BEGIN').length,2);
|
||||||
});
|
});
|
||||||
test('lost commit acknowledgement does not overwrite or duplicate an already-published batch',async()=>{
|
test('lost commit acknowledgement does not overwrite or duplicate an already-published batch',async()=>{
|
||||||
const db=new Database();db.failCommitAck=true;
|
const db=new Database();db.failCommitAck=true;
|
||||||
assert.equal(await db.store().capture('2026-10',db.clock),'already-complete');
|
assert.equal(await db.store().capture('2026-10-31',db.clock),'already-complete');
|
||||||
assert.deepEqual(db.committed,['batch','details','complete']);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===MAIN_SQL).length,1);
|
||||||
});
|
});
|
||||||
test('source is not read if transaction establishment crosses the capture deadline',async()=>{
|
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');
|
const db=new Database();let now=new Date('2026-10-31T23:55:59.900+08:00');
|
||||||
db.onReadTransaction=()=>{now=new Date('2026-10-31T17:01:00+08:00');};
|
db.onReadTransaction=()=>{now=new Date('2026-10-31T23:56:00+08:00');};
|
||||||
assert.equal(await db.store().capture('2026-10',()=>now),'outside-window');
|
assert.equal(await db.store().capture('2026-10-31',()=>now),'outside-window');
|
||||||
assert.equal(db.commands.includes(MAIN_SQL),false);assert.deepEqual(db.committed,[]);
|
assert.equal(db.commands.includes(MAIN_SQL),false);assert.deepEqual(db.committed,[]);
|
||||||
});
|
});
|
||||||
test('persistent detail-write failure publishes neither partial header nor rows',async()=>{
|
test('persistent detail-write failure publishes neither partial header nor rows',async()=>{
|
||||||
const db=new Database();db.failDetails=5;
|
const db=new Database();db.failDetails=5;
|
||||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/write failed/);
|
await assert.rejects(()=>db.store().capture('2026-10-31',db.clock),/write failed/);
|
||||||
assert.deepEqual(db.committed,[]);assert.deepEqual(db.staged,[]);assert.equal(db.released,true);
|
assert.deepEqual(db.committed,[]);assert.deepEqual(db.staged,[]);assert.equal(db.released,true);
|
||||||
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
assert.equal(db.commands.filter(s=>s===MAIN_SQL).length,1);
|
||||||
});
|
});
|
||||||
test('source failure and invalid rows cannot publish snapshots',async()=>{
|
test('source failure and invalid rows cannot publish snapshots',async()=>{
|
||||||
const db=new Database();db.failReads=true;
|
const db=new Database();db.failReads=true;
|
||||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/read failed/);assert.deepEqual(db.committed,[]);
|
await assert.rejects(()=>db.store().capture('2026-10-31',db.clock),/read failed/);assert.deepEqual(db.committed,[]);
|
||||||
db.failReads=false;db.rows=[testRow('duplicate'),testRow('duplicate')];
|
db.failReads=false;db.rows=[testRow('duplicate'),testRow('duplicate')];
|
||||||
await assert.rejects(()=>db.store().capture('2026-10',db.clock),/重复/);assert.deepEqual(db.committed,[]);
|
await assert.rejects(()=>db.store().capture('2026-10-31',db.clock),/重复/);assert.deepEqual(db.committed,[]);
|
||||||
});
|
});
|
||||||
test('completed period, concurrent lock and expired window cannot issue a source read',async()=>{
|
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']){
|
for(const scenario of ['complete','lock','expired','delayed-lock','late-activation']){
|
||||||
const db=new Database();let clock=db.clock;
|
const db=new Database();let clock=db.clock;
|
||||||
if(scenario==='complete')db.existing=[{id:'existing',status:'COMPLETE'}];
|
if(scenario==='complete')db.existing=[{id:'existing',status:'COMPLETE'}];
|
||||||
if(scenario==='lock')db.locked=0;
|
if(scenario==='lock')db.locked=0;
|
||||||
if(scenario==='expired')clock=()=>new Date('2026-10-31T17:01:00+08:00');
|
if(scenario==='expired')clock=()=>new Date('2026-10-31T23:56:00+08:00');
|
||||||
if(scenario==='delayed-lock')db.beforeLock=()=>{clock=()=>new Date('2026-10-31T17:01:00+08:00');};
|
if(scenario==='delayed-lock')db.beforeLock=()=>{clock=()=>new Date('2026-10-31T23:56:00+08:00');};
|
||||||
if(scenario==='late-activation')db.settings='2026-10-31T17:00:01+08:00';
|
if(scenario==='late-activation')db.settings='2026-10-31T23:55:01+08:00';
|
||||||
const result=await db.store().capture('2026-10',()=>clock());
|
const result=await db.store().capture('2026-10-31',()=>clock());
|
||||||
assert.ok(['already-complete','locked','outside-window'].includes(result));
|
assert.ok(['already-complete','locked','outside-window'].includes(result));
|
||||||
assert.equal(db.commands.includes(MAIN_SQL),false,scenario);assert.deepEqual(db.committed,[]);
|
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()=>{
|
test('a legitimate empty source commits a zero-count complete batch without detail inserts',async()=>{
|
||||||
const db=new Database();db.rows=[];
|
const db=new Database();db.rows=[];
|
||||||
assert.equal(await db.store().capture('2026-10',db.clock),'captured');assert.deepEqual(db.committed,['batch','complete']);
|
assert.equal(await db.store().capture('2026-10-31',db.clock),'captured');assert.deepEqual(db.committed,['batch','complete']);
|
||||||
});
|
});
|
||||||
test('missing storage reads never issue schema writes and report storage-unavailable',async()=>{
|
test('missing storage reads never issue schema writes and report storage-unavailable',async()=>{
|
||||||
const calls:string[]=[];
|
const calls:string[]=[];
|
||||||
@@ -101,3 +101,12 @@ test('missing storage reads never issue schema writes and report storage-unavail
|
|||||||
await assert.rejects(()=>store.batches(),HistoryStorageUnavailable);
|
await assert.rejects(()=>store.batches(),HistoryStorageUnavailable);
|
||||||
assert.ok(calls.every(sql=>sql.startsWith('SELECT')));
|
assert.ok(calls.every(sql=>sql.startsWith('SELECT')));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('monthly reader selects month-end keys and retains completed legacy archives when a daily capture failed',async()=>{
|
||||||
|
let query='';
|
||||||
|
const row=(id:string,month:string,status:string)=>({id,month,status,source:'AUTO',total_count:0,operating_count:0,inventory_count:0});
|
||||||
|
const db={query:async(sql:string)=>{query=sql;return[[row('old-sep','2026-09','COMPLETE'),row('new-sep','2026-09','FAILED'),row('old-oct','2026-10','COMPLETE'),row('new-oct','2026-10','COMPLETE')]];}};
|
||||||
|
const batches=await new MysqlHistoryStore(db as unknown as Pool).batches();
|
||||||
|
assert.deepEqual(batches.map(b=>b.id),['old-sep','new-oct']);
|
||||||
|
assert.match(query,/LAST_DAY/);assert.match(query,/CHAR_LENGTH\(period_key\)=7/);
|
||||||
|
});
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import { randomUUID } from 'node:crypto';
|
import { randomUUID } from 'node:crypto';
|
||||||
import type { Pool, PoolConnection, RowDataPacket } from 'mysql2/promise';
|
import type { Pool, PoolConnection, RowDataPacket } from 'mysql2/promise';
|
||||||
import { CAPTURE_WINDOW_MS, HISTORY_VERSION, monthCutoff, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
import { CAPTURE_WINDOW_MS, HISTORY_VERSION, dayCutoff, shanghaiTimestamp } from '../../../shared/assets/history.js';
|
||||||
import type { VehicleRow } from '../../types.js';
|
import type { VehicleRow } from '../../types.js';
|
||||||
import { MAIN_SQL } from '../vehicles/repository.js';
|
import { MAIN_SQL } from '../vehicles/repository.js';
|
||||||
import { counts, freezeRows, HistoryStorageUnavailable, type HistoryBatch, type HistorySettings, type HistoryVehicle } from './model.js';
|
import { counts, freezeRows, HistoryStorageUnavailable, type HistoryBatch, type HistorySettings, type HistoryVehicle } from './model.js';
|
||||||
@@ -26,16 +26,21 @@ export class MysqlHistoryStore implements HistoryReader {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
async settings(): Promise<HistorySettings | null> {
|
async settings(): Promise<HistorySettings | null> {
|
||||||
const rows = await this.read(`SELECT ${dateColumn('enabled_at')} AS enabledAt FROM bi_asset_history_settings WHERE id=1`);
|
const rows = await this.read(`SELECT ${dateColumn('daily_enabled_at')} AS enabledAt FROM bi_asset_history_settings WHERE id=1`);
|
||||||
return rows[0] ? { enabledAt: rows[0].enabledAt } : null;
|
return rows[0]?.enabledAt ? { enabledAt: rows[0].enabledAt } : null;
|
||||||
}
|
}
|
||||||
async batches(): Promise<HistoryBatch[]> {
|
async batches(): Promise<HistoryBatch[]> {
|
||||||
const rows = await this.read(`SELECT id, period_month AS month, status, source,
|
const rows = await this.read(`SELECT id, LEFT(period_key,7) AS month, status, source,
|
||||||
total_count,operating_count,inventory_count,
|
total_count,operating_count,inventory_count,
|
||||||
${dateColumn('scheduled_for')} AS scheduledFor, ${dateColumn('captured_at')} AS capturedAt
|
${dateColumn('scheduled_for')} AS scheduledFor, ${dateColumn('captured_at')} AS capturedAt
|
||||||
FROM bi_asset_history_batches WHERE definition_version=? ORDER BY period_month`, [HISTORY_VERSION]);
|
FROM bi_asset_history_batches WHERE definition_version=? AND (CHAR_LENGTH(period_key)=7 OR period_key=DATE_FORMAT(LAST_DAY(CONCAT(LEFT(period_key,7),'-01')), '%Y-%m-%d')) ORDER BY period_key`, [HISTORY_VERSION]);
|
||||||
return rows.map(row => ({ id: row.id, month: row.month, status: row.status, source: row.source, scheduledFor: row.scheduledFor, capturedAt: row.capturedAt,
|
const batches = 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 } }));
|
counts: { total: row.total_count, operating: row.operating_count, inventory: row.inventory_count } }));
|
||||||
|
const byMonth = new Map<string, HistoryBatch>();
|
||||||
|
for (const batch of batches) {
|
||||||
|
if (batch.status === 'COMPLETE' || byMonth.get(batch.month)?.status !== 'COMPLETE') byMonth.set(batch.month, batch);
|
||||||
|
}
|
||||||
|
return [...byMonth.values()];
|
||||||
}
|
}
|
||||||
async vehicles(batchIds: string[]): Promise<Map<string, HistoryVehicle[]>> {
|
async vehicles(batchIds: string[]): Promise<Map<string, HistoryVehicle[]>> {
|
||||||
const result = new Map<string, HistoryVehicle[]>(batchIds.map(id => [id, []]));
|
const result = new Map<string, HistoryVehicle[]>(batchIds.map(id => [id, []]));
|
||||||
@@ -49,24 +54,29 @@ export class MysqlHistoryStore implements HistoryReader {
|
|||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
async activate(now: Date): Promise<HistorySettings> {
|
async activate(now: Date): Promise<HistorySettings> {
|
||||||
await this.db.query('INSERT IGNORE INTO bi_asset_history_settings (id,enabled_at) VALUES (1,?)', [sqlTime(now)]);
|
await this.db.query(`INSERT INTO bi_asset_history_settings (id,enabled_at,daily_enabled_at) VALUES (1,?,?)
|
||||||
|
ON DUPLICATE KEY UPDATE daily_enabled_at=COALESCE(daily_enabled_at,VALUES(daily_enabled_at))`, [sqlTime(now), sqlTime(now)]);
|
||||||
return (await this.settings())!;
|
return (await this.settings())!;
|
||||||
}
|
}
|
||||||
async recordOutcome(month: string, status: 'FAILED' | 'MISSED', code: string): Promise<void> {
|
async dailyBatches(): Promise<Array<{ day: string; status: HistoryBatch['status'] }>> {
|
||||||
|
const rows = await this.read('SELECT period_key AS day,status FROM bi_asset_history_batches WHERE definition_version=? AND CHAR_LENGTH(period_key)=10', [HISTORY_VERSION]);
|
||||||
|
return rows.map(row => ({ day: row.day, status: row.status }));
|
||||||
|
}
|
||||||
|
async recordOutcome(day: string, status: 'FAILED' | 'MISSED', code: string): Promise<void> {
|
||||||
// Never downgrade a completed snapshot, or replace a useful failure with MISSED.
|
// Never downgrade a completed snapshot, or replace a useful failure with MISSED.
|
||||||
await this.db.query(`INSERT INTO bi_asset_history_batches
|
await this.db.query(`INSERT INTO bi_asset_history_batches
|
||||||
(id,period_month,definition_version,scheduled_for,source,status,failure_code)
|
(id,period_key,definition_version,scheduled_for,source,status,failure_code)
|
||||||
VALUES (?,?,?,?,'AUTO',?,?) ON DUPLICATE KEY UPDATE
|
VALUES (?,?,?,?,'AUTO',?,?) ON DUPLICATE KEY UPDATE
|
||||||
failure_code=IF(status='COMPLETE' OR (status='FAILED' AND VALUES(status)='MISSED'),failure_code,VALUES(failure_code)),
|
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))`,
|
status=IF(status='COMPLETE' OR (status='FAILED' AND VALUES(status)='MISSED'),status,VALUES(status))`,
|
||||||
[randomUUID(), month, HISTORY_VERSION, sqlTime(monthCutoff(month)), status, code]);
|
[randomUUID(), day, HISTORY_VERSION, sqlTime(dayCutoff(day)), status, code]);
|
||||||
}
|
}
|
||||||
async capture(month: string, clock: () => Date = () => new Date()): Promise<CaptureResult> {
|
async capture(day: string, clock: () => Date = () => new Date()): Promise<CaptureResult> {
|
||||||
const cutoff = monthCutoff(month).getTime();
|
const cutoff = dayCutoff(day).getTime();
|
||||||
const inWindow = () => clock().getTime() >= cutoff && clock().getTime() < cutoff + CAPTURE_WINDOW_MS;
|
const inWindow = () => clock().getTime() >= cutoff && clock().getTime() < cutoff + CAPTURE_WINDOW_MS;
|
||||||
if (!inWindow()) return 'outside-window';
|
if (!inWindow()) return 'outside-window';
|
||||||
const connection = await this.db.getConnection();
|
const connection = await this.db.getConnection();
|
||||||
const lock = `ln-bi:asset:${HISTORY_VERSION}:${month}`;
|
const lock = `ln-bi:asset:${HISTORY_VERSION}:${day}`;
|
||||||
let locked = false;
|
let locked = false;
|
||||||
try {
|
try {
|
||||||
const [locks] = await connection.query<RowDataPacket[]>("SELECT GET_LOCK(CONCAT(DATABASE(),':',?),0) AS acquired", [lock]);
|
const [locks] = await connection.query<RowDataPacket[]>("SELECT GET_LOCK(CONCAT(DATABASE(),':',?),0) AS acquired", [lock]);
|
||||||
@@ -74,7 +84,7 @@ export class MysqlHistoryStore implements HistoryReader {
|
|||||||
if (!locked) return 'locked';
|
if (!locked) return 'locked';
|
||||||
const settings = await this.settings();
|
const settings = await this.settings();
|
||||||
if (!settings || new Date(settings.enabledAt).getTime() > cutoff || !inWindow()) return 'outside-window';
|
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]);
|
const [existing] = await connection.query<RowDataPacket[]>(`SELECT id,status FROM bi_asset_history_batches WHERE period_key=? AND definition_version=?`, [day, HISTORY_VERSION]);
|
||||||
if (existing[0]?.status === 'COMPLETE') return 'already-complete';
|
if (existing[0]?.status === 'COMPLETE') return 'already-complete';
|
||||||
await connection.query("SET time_zone='+08:00'");
|
await connection.query("SET time_zone='+08:00'");
|
||||||
await connection.query('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
await connection.query('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
||||||
@@ -90,7 +100,7 @@ export class MysqlHistoryStore implements HistoryReader {
|
|||||||
// Retries below keep this exact frozen dataset. Never reread a later live ledger.
|
// Retries below keep this exact frozen dataset. Never reread a later live ledger.
|
||||||
const id = existing[0]?.id || randomUUID();
|
const id = existing[0]?.id || randomUUID();
|
||||||
for (let attempt = 0; ; attempt++) {
|
for (let attempt = 0; ; attempt++) {
|
||||||
try { return await this.publish(connection, id, month, capturedAt, frozen, clock); }
|
try { return await this.publish(connection, id, day, capturedAt, frozen, clock); }
|
||||||
catch (error) { await connection.rollback(); if (attempt >= 2) throw error; }
|
catch (error) { await connection.rollback(); if (attempt >= 2) throw error; }
|
||||||
}
|
}
|
||||||
} finally {
|
} finally {
|
||||||
@@ -98,19 +108,19 @@ export class MysqlHistoryStore implements HistoryReader {
|
|||||||
finally { connection.release(); }
|
finally { connection.release(); }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
private async publish(connection: PoolConnection, id: string, month: string, capturedAt: Date, vehicles: HistoryVehicle[], clock: () => Date) {
|
private async publish(connection: PoolConnection, id: string, day: string, capturedAt: Date, vehicles: HistoryVehicle[], clock: () => Date) {
|
||||||
const totals = counts(vehicles);
|
const totals = counts(vehicles);
|
||||||
await connection.beginTransaction();
|
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]);
|
const [existing] = await connection.query<RowDataPacket[]>('SELECT id,status FROM bi_asset_history_batches WHERE period_key=? AND definition_version=? FOR UPDATE', [day, HISTORY_VERSION]);
|
||||||
// Also handles a committed transaction whose acknowledgement was lost.
|
// 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]?.status === 'COMPLETE') { await connection.commit(); return 'already-complete' as const; }
|
||||||
if (existing[0]) id = existing[0].id;
|
if (existing[0]) id = existing[0].id;
|
||||||
// Under the advisory lock, only FAILED/MISSED placeholders may be promoted.
|
// Under the advisory lock, only FAILED/MISSED placeholders may be promoted.
|
||||||
await connection.query(`INSERT INTO bi_asset_history_batches
|
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)
|
(id,period_key,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),
|
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)`,
|
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]);
|
[id, day, HISTORY_VERSION, sqlTime(dayCutoff(day)), sqlTime(capturedAt), totals.total, totals.operating, totals.inventory]);
|
||||||
for (let offset = 0; offset < vehicles.length; offset += 100) {
|
for (let offset = 0; offset < vehicles.length; offset += 100) {
|
||||||
const chunk = vehicles.slice(offset, offset + 100);
|
const chunk = vehicles.slice(offset, offset + 100);
|
||||||
await connection.query(`INSERT INTO bi_asset_history_vehicles
|
await connection.query(`INSERT INTO bi_asset_history_vehicles
|
||||||
|
|||||||
@@ -57,15 +57,24 @@ export function shanghaiTimestamp(now: Date): string { return new Date(now.getTi
|
|||||||
export function monthCutoff(month: string): Date {
|
export function monthCutoff(month: string): Date {
|
||||||
if (!/^\d{4}-(0[1-9]|1[0-2])$/.test(month)) throw new Error('月份格式无效');
|
if (!/^\d{4}-(0[1-9]|1[0-2])$/.test(month)) throw new Error('月份格式无效');
|
||||||
const [year, m] = month.split('-').map(Number);
|
const [year, m] = month.split('-').map(Number);
|
||||||
return new Date(Date.UTC(year, m, 0, 9)); // 17:00 Asia/Shanghai, on the last day.
|
return new Date(Date.UTC(year, m, 0, 15, 55)); // 23:55 Asia/Shanghai, on the last day.
|
||||||
}
|
}
|
||||||
export function shiftMonth(month: string, offset: number): string {
|
export function shiftMonth(month: string, offset: number): string {
|
||||||
const [year, m] = month.split('-').map(Number);
|
const [year, m] = month.split('-').map(Number);
|
||||||
return new Date(Date.UTC(year, m - 1 + offset, 1)).toISOString().slice(0, 7);
|
return new Date(Date.UTC(year, m - 1 + offset, 1)).toISOString().slice(0, 7);
|
||||||
}
|
}
|
||||||
|
export function shanghaiDay(now: Date): string { return shanghaiTimestamp(now).slice(0, 10); }
|
||||||
|
export function dayCutoff(day: string): Date {
|
||||||
|
const date = new Date(`${day}T23:55:00+08:00`);
|
||||||
|
if (!/^\d{4}-\d{2}-\d{2}$/.test(day) || !Number.isFinite(date.getTime()) || shanghaiDay(date) !== day) throw new Error('日期格式无效');
|
||||||
|
return date;
|
||||||
|
}
|
||||||
|
export function shiftDay(day: string, offset: number): string {
|
||||||
|
return shanghaiDay(new Date(dayCutoff(day).getTime() + offset * 86_400_000));
|
||||||
|
}
|
||||||
export function nextCapture(now: Date): string {
|
export function nextCapture(now: Date): string {
|
||||||
const month = shanghaiMonth(now);
|
const day = shanghaiDay(now);
|
||||||
return shanghaiTimestamp(monthCutoff(monthCutoff(month).getTime() > now.getTime() ? month : shiftMonth(month, 1)));
|
return shanghaiTimestamp(dayCutoff(dayCutoff(day) > now ? day : shiftDay(day, 1)));
|
||||||
}
|
}
|
||||||
export function historyTime(value: string | null): string {
|
export function historyTime(value: string | null): string {
|
||||||
if (!value) return '暂无时间';
|
if (!value) return '暂无时间';
|
||||||
|
|||||||
Reference in New Issue
Block a user