feat: 添加 G5 独立写入功能
- 新增 G5 数据库连接配置与可关闭的写入开关 - 在现有 legacy/G4 写入成功路径后，追加独立的 G5 写入流程 - G5 使用与 G4 相同的数据结构映射，但不写入 guid，由数据库自生成 int4 guid - room_status 新增 G5 独立 upsert 写入路径，并保留旧表与 G5 表的独立开关 - 新增 G5 写入统计与启动摘要输出 - 更新 StatsCounters 和 StatsReporter 以支持 G5 统计 - 增加测试覆盖，确保 G5 写入逻辑与 room_status 的独立执行 - 新增 G5 相关数据库表结构 SQL 文件
This commit is contained in:
@@ -22,6 +22,7 @@ const buildBasePayload = () => ({
|
||||
|
||||
function makeDualResult({ legacyEnabled = true, legacySuccess = true, legacyConn = false,
|
||||
g4HotEnabled = false, g4HotSuccess = true, g4HotConn = false,
|
||||
g5Enabled = false, g5Success = true, g5Conn = false,
|
||||
insertedCount = 1, failedRecords = [] } = {}) {
|
||||
return {
|
||||
legacy: {
|
||||
@@ -34,6 +35,11 @@ function makeDualResult({ legacyEnabled = true, legacySuccess = true, legacyConn
|
||||
failedRecords: g4HotSuccess ? [] : failedRecords, error: g4HotSuccess ? null : new Error('g4hot fail'),
|
||||
isConnectionError: g4HotConn, batchError: g4HotSuccess ? null : new Error('g4hot fail'),
|
||||
},
|
||||
g5: {
|
||||
enabled: g5Enabled, success: g5Success, insertedCount: g5Success ? insertedCount : 0,
|
||||
failedRecords: g5Success ? [] : failedRecords, error: g5Success ? null : new Error('g5 fail'),
|
||||
isConnectionError: g5Conn, batchError: g5Success ? null : new Error('g5 fail'),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -45,6 +51,7 @@ function buildMockDb(overrides = {}) {
|
||||
roomStatusEnabled: true,
|
||||
legacyTable: 'heartbeat.heartbeat_events',
|
||||
g4HotTable: 'heartbeat.heartbeat_events_g4_hot',
|
||||
roomStatusTable: 'room_status.room_status_moment',
|
||||
...overrides.config,
|
||||
},
|
||||
insertHeartbeatEventsDual: overrides.insertHeartbeatEventsDual ?? (async () => makeDualResult()),
|
||||
@@ -55,11 +62,40 @@ function buildMockDb(overrides = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
function buildProcessor(dbOverrides = {}, processorConfig = {}) {
|
||||
function buildMockG5Db(overrides = {}) {
|
||||
return {
|
||||
config: {
|
||||
enabled: true,
|
||||
g5HeartbeatEnabled: true,
|
||||
g5Table: 'heartbeat.heartbeat_events_g5',
|
||||
roomStatusEnabled: true,
|
||||
roomStatusTable: 'room_status.room_status_moment_g5',
|
||||
...overrides.config,
|
||||
},
|
||||
insertHeartbeatEventsG5: overrides.insertHeartbeatEventsG5 ?? (async () => ({
|
||||
enabled: true,
|
||||
success: true,
|
||||
insertedCount: 1,
|
||||
failedRecords: [],
|
||||
error: null,
|
||||
isConnectionError: false,
|
||||
batchError: null,
|
||||
})),
|
||||
upsertRoomStatusG5: overrides.upsertRoomStatusG5 ?? (async () => ({ rowCount: 1 })),
|
||||
_isDbConnectionError: overrides._isDbConnectionError ?? (() => false),
|
||||
};
|
||||
}
|
||||
|
||||
function buildProcessor(dbOverrides = {}, processorConfig = {}, g5Overrides = null) {
|
||||
const db = buildMockDb(dbOverrides);
|
||||
const deps = {};
|
||||
if (g5Overrides) {
|
||||
deps.g5DatabaseManager = buildMockG5Db(g5Overrides);
|
||||
}
|
||||
return new HeartbeatProcessor(
|
||||
{ batchSize: 1, batchTimeout: 1000, ...processorConfig },
|
||||
db
|
||||
db,
|
||||
deps
|
||||
);
|
||||
}
|
||||
|
||||
@@ -218,6 +254,112 @@ describe('Dual-write: room_status 始终执行', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('RoomStatus dual-write', () => {
|
||||
it('should write old and g5 room_status independently', async () => {
|
||||
let oldCalled = false;
|
||||
let g5Called = false;
|
||||
const processor = buildProcessor(
|
||||
{
|
||||
config: { legacyHeartbeatEnabled: true, g4HotHeartbeatEnabled: false, roomStatusEnabled: true },
|
||||
insertHeartbeatEventsDual: async () => makeDualResult({ legacyEnabled: true, g4HotEnabled: false }),
|
||||
upsertRoomStatus: async () => { oldCalled = true; return { rowCount: 1 }; },
|
||||
},
|
||||
{},
|
||||
{
|
||||
config: { roomStatusEnabled: true },
|
||||
upsertRoomStatusG5: async () => { g5Called = true; return { rowCount: 1 }; },
|
||||
}
|
||||
);
|
||||
|
||||
const msg = { value: Buffer.from(JSON.stringify(buildBasePayload()), 'utf8') };
|
||||
await processor.processMessage(msg);
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
assert.equal(oldCalled, true);
|
||||
assert.equal(g5Called, true);
|
||||
});
|
||||
|
||||
it('should allow old room_status off while g5 room_status stays on', async () => {
|
||||
let oldCalled = false;
|
||||
let g5Called = false;
|
||||
const processor = buildProcessor(
|
||||
{
|
||||
config: { legacyHeartbeatEnabled: true, g4HotHeartbeatEnabled: false, roomStatusEnabled: false },
|
||||
insertHeartbeatEventsDual: async () => makeDualResult({ legacyEnabled: true, g4HotEnabled: false }),
|
||||
upsertRoomStatus: async () => { oldCalled = true; return { rowCount: 1 }; },
|
||||
},
|
||||
{},
|
||||
{
|
||||
config: { roomStatusEnabled: true },
|
||||
upsertRoomStatusG5: async () => { g5Called = true; return { rowCount: 1 }; },
|
||||
}
|
||||
);
|
||||
|
||||
const msg = { value: Buffer.from(JSON.stringify(buildBasePayload()), 'utf8') };
|
||||
await processor.processMessage(msg);
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
assert.equal(oldCalled, false);
|
||||
assert.equal(g5Called, true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('G5-write: 独立写库', () => {
|
||||
it('should write to g5 independently after legacy/g4 succeed', async () => {
|
||||
let g5Captured = null;
|
||||
const processor = buildProcessor(
|
||||
{
|
||||
config: { legacyHeartbeatEnabled: true, g4HotHeartbeatEnabled: true },
|
||||
insertHeartbeatEventsDual: async () => makeDualResult({ legacyEnabled: true, g4HotEnabled: true }),
|
||||
},
|
||||
{},
|
||||
{
|
||||
insertHeartbeatEventsG5: async (events) => {
|
||||
g5Captured = events;
|
||||
return {
|
||||
enabled: true,
|
||||
success: true,
|
||||
insertedCount: events.length,
|
||||
failedRecords: [],
|
||||
error: null,
|
||||
isConnectionError: false,
|
||||
batchError: null,
|
||||
};
|
||||
},
|
||||
}
|
||||
);
|
||||
|
||||
const msg = { value: Buffer.from(JSON.stringify(buildBasePayload()), 'utf8') };
|
||||
const res = await processor.processMessage(msg);
|
||||
assert.deepEqual(res, { insertedCount: 1 });
|
||||
assert.ok(Array.isArray(g5Captured));
|
||||
assert.equal(g5Captured.length, 1);
|
||||
});
|
||||
|
||||
it('should not block main flow when g5 write fails', async () => {
|
||||
const processor = buildProcessor(
|
||||
{
|
||||
config: { legacyHeartbeatEnabled: true, g4HotHeartbeatEnabled: false },
|
||||
insertHeartbeatEventsDual: async () => makeDualResult({ legacyEnabled: true, g4HotEnabled: false }),
|
||||
},
|
||||
{},
|
||||
{
|
||||
insertHeartbeatEventsG5: async () => ({
|
||||
enabled: true,
|
||||
success: false,
|
||||
insertedCount: 0,
|
||||
failedRecords: [{ error: new Error('g5 fail'), record: buildBasePayload() }],
|
||||
error: new Error('g5 fail'),
|
||||
isConnectionError: false,
|
||||
batchError: new Error('g5 fail'),
|
||||
}),
|
||||
}
|
||||
);
|
||||
|
||||
const msg = { value: Buffer.from(JSON.stringify(buildBasePayload()), 'utf8') };
|
||||
const res = await processor.processMessage(msg);
|
||||
assert.deepEqual(res, { insertedCount: 1 });
|
||||
});
|
||||
});
|
||||
|
||||
describe('Dual-write: 暂停消费策略', () => {
|
||||
it('should NOT pause when both detail writes disabled but room_status enabled', async () => {
|
||||
let paused = false;
|
||||
@@ -371,6 +513,48 @@ describe('DatabaseManager: _g4HotToRowValues', () => {
|
||||
const guid = values[cols.indexOf('guid')];
|
||||
assert.equal(guid, 'a0b1c2d3e4f56789abcdef0123456789');
|
||||
});
|
||||
|
||||
it('omits guid column and value for g5 rows', () => {
|
||||
const dm = new DatabaseManager({ host: 'x', port: 5432, user: 'x', password: 'x', database: 'x' });
|
||||
const cols = dm._getG5Columns();
|
||||
const values = dm._g5ToRowValues(buildBasePayload());
|
||||
assert.equal(cols.includes('guid'), false);
|
||||
assert.equal(values.length, cols.length);
|
||||
});
|
||||
|
||||
it('writes g5 base array columns as null', () => {
|
||||
const dm = new DatabaseManager({ host: 'x', port: 5432, user: 'x', password: 'x', database: 'x' });
|
||||
const cols = dm._getG5Columns();
|
||||
const values = dm._g5ToRowValues({
|
||||
...buildBasePayload(),
|
||||
service_mask: 7,
|
||||
elec_address: ['e1', 'e2'],
|
||||
air_address: ['ac1', 'ac2'],
|
||||
voltage: [220.5, 221.5],
|
||||
ampere: [1.1, 1.2],
|
||||
power: [100, 200],
|
||||
phase: ['A', 'B'],
|
||||
energy: [10, 20],
|
||||
sum_energy: [100, 200],
|
||||
state: [1, 0],
|
||||
model: [2, 3],
|
||||
speed: [1, 2],
|
||||
set_temp: [24, 25],
|
||||
now_temp: [26, 27],
|
||||
solenoid_valve: [1, 0],
|
||||
extra: { source: 'test' },
|
||||
});
|
||||
|
||||
for (const column of ['service_mask', 'elec_address', 'air_address', 'voltage', 'ampere', 'power', 'phase', 'energy', 'sum_energy', 'state', 'model', 'speed', 'set_temp', 'now_temp', 'solenoid_valve', 'extra']) {
|
||||
assert.equal(values[cols.indexOf(column)], null);
|
||||
}
|
||||
|
||||
assert.equal(values[cols.indexOf('svc_01')], true);
|
||||
assert.equal(values[cols.indexOf('svc_02')], true);
|
||||
assert.equal(values[cols.indexOf('svc_03')], true);
|
||||
assert.equal(values[cols.indexOf('air_address_1')], 'ac1');
|
||||
assert.equal(values[cols.indexOf('elec_address_1')], 'e1');
|
||||
});
|
||||
});
|
||||
|
||||
describe('DatabaseManager: _formatPgCol', () => {
|
||||
@@ -382,6 +566,37 @@ describe('DatabaseManager: _formatPgCol', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('DatabaseManager: room_status upsert SQL', () => {
|
||||
it('does not update ts_ms in old room_status DO UPDATE SET', () => {
|
||||
const dm = new DatabaseManager({ host: 'x', port: 5432, user: 'x', password: 'x', database: 'x', roomStatusTable: 'room_status.room_status_moment' });
|
||||
const built = dm._buildRoomStatusUpsertQuery([buildBasePayload()], {
|
||||
tableName: 'room_status.room_status_moment',
|
||||
conflictColumns: ['hotel_id', 'room_id', 'device_id'],
|
||||
includeGuid: true,
|
||||
autoCreatePartitions: true,
|
||||
tableRef: 'room_status.room_status_moment',
|
||||
logPrefix: 'upsertRoomStatus',
|
||||
});
|
||||
assert.match(built.sql, /ON CONFLICT \(hotel_id, room_id, device_id\)/);
|
||||
assert.doesNotMatch(built.sql, /ts_ms = EXCLUDED\.ts_ms/);
|
||||
});
|
||||
|
||||
it('uses hotel_id and room_id as conflict key for g5 room_status', () => {
|
||||
const dm = new DatabaseManager({ host: 'x', port: 5432, user: 'x', password: 'x', database: 'x', roomStatusTable: 'room_status.room_status_moment_g5' });
|
||||
const built = dm._buildRoomStatusUpsertQuery([buildBasePayload()], {
|
||||
tableName: 'room_status.room_status_moment_g5',
|
||||
conflictColumns: ['hotel_id', 'room_id'],
|
||||
includeGuid: false,
|
||||
autoCreatePartitions: false,
|
||||
tableRef: 'room_status.room_status_moment_g5',
|
||||
logPrefix: 'upsertRoomStatusG5',
|
||||
});
|
||||
assert.match(built.sql, /ON CONFLICT \(hotel_id, room_id\)/);
|
||||
assert.doesNotMatch(built.sql, /ts_ms = EXCLUDED\.ts_ms/);
|
||||
assert.equal(/guid/.test(built.sql), false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('DatabaseManager: insertHeartbeatEventsDual', () => {
|
||||
it('returns empty results when both targets disabled', async () => {
|
||||
const dm = new DatabaseManager({
|
||||
|
||||
Reference in New Issue
Block a user