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; batches(): Promise; vehicles(batchIds: string[]): Promise>; } 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) {} private async read(sql: string, args: unknown[] = []): Promise { try { const [rows] = await this.db.query(sql, args); return rows; } catch (error) { if ((error as { code?: string }).code === 'ER_NO_SUCH_TABLE') throw new HistoryStorageUnavailable('历史快照存储尚未部署'); throw error; } } async settings(): Promise { 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 { 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> { const result = new Map(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 { 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 { // 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 { 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("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(`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(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('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; } }