chore: keep go branch go-only
This commit is contained in:
lingniu
@@ -1,600 +0,0 @@
|
||||
# GB32960 Body Parser — Per-Block Exception Isolation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** `Gb32960BodyParser` 主循环对单信息块的解析异常实现**单块隔离**：一块解析失败不再整帧放弃，改为把失败块兜成 `InfoBlock.Raw` 并继续解析后续信息块。
|
||||
|
||||
**Architecture:** 在主循环每次 `parser.parse(body)` 外面套 try/catch（仅捕获 `DecodeException` / `BufferUnderflowException` / `IndexOutOfBoundsException`，放行 RuntimeException 以免掩盖 parser bug）。失败时：
|
||||
- 若 `fixedLen ≥ 0` 且剩余字节 ≥ fixedLen → 回滚 reader → 按 fixedLen 截取 Raw → position 前进 fixedLen → `continue` 循环
|
||||
- 否则（变长 parser 或剩余不足）→ 剩余字节全部兜成 Raw → `break` 循环（变长块无法安全找下一块边界）
|
||||
|
||||
通过 `Gb32960Properties.Parse.lenientBlockFailure`（默认 true）控制开关，可以回退到严格模式。**不改 `InfoBlock.Raw` record 结构**（避免 downstream 兼容风险——`Gb32960EventMapper` 按 `findBlock(Class)` 类型查找，新增 Raw 不影响其行为）。
|
||||
|
||||
**Tech Stack:** Java 25, JUnit 5, AssertJ, Spring Boot ConfigurationProperties, Maven。仅改动 `protocol-gb32960` 模块。
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
- **Modify** `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960Properties.java` — 新增 `Parse` 内嵌类 + `parse` 字段/getter/setter
|
||||
- **Modify** `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParser.java` — 主循环 try/catch + recovery 逻辑 + `lenientBlockFailure` 字段
|
||||
- **Modify** `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java` — 把配置穿给 BodyParser
|
||||
- **Create** `protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParserIsolationTest.java` — 3 个隔离场景 + 1 个严格模式回退场景
|
||||
- **Modify** `bootstrap-all/src/main/resources/application.yml` — 添加 `parse:` 注释段示例
|
||||
- **Modify** `CHANGELOG.md` — 追加本次变更条目
|
||||
|
||||
---
|
||||
|
||||
## Task 1: 基线验证 —— 现有测试全绿
|
||||
|
||||
**Files:** 无
|
||||
|
||||
- [ ] **Step 1.1: 跑 protocol-gb32960 模块测试**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 test -q`
|
||||
Expected: BUILD SUCCESS，所有现有测试通过（`Gb32960DecoderGoldenTest`、`Gb32960FullBlocksTest`、`Gb32960DecoderTest`、`GuangdongFcEndToEndTest`、parser/* 各 Block Parser 单测、profile/* 选择器单测）。
|
||||
|
||||
如果基线红了，**先停下来修复基线**再进入 Task 2。
|
||||
|
||||
---
|
||||
|
||||
## Task 2: 添加 Parse 配置子节点
|
||||
|
||||
**Files:**
|
||||
- Modify: `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960Properties.java`
|
||||
|
||||
- [ ] **Step 2.1: 在 `Gb32960Properties` 类体中（在 `Auth` 字段后、`Tls` 字段前）新增 `parse` 字段**
|
||||
|
||||
位置参考：现在 L24 `private Auth auth = new Auth();` 下面一行，在 L27 `private Tls tls = new Tls();` 之前插入：
|
||||
|
||||
```java
|
||||
/**
|
||||
* 报文解析行为配置。
|
||||
*
|
||||
* <p>默认 {@code lenientBlockFailure=true}：单个信息块解析异常时兜成
|
||||
* {@link com.lingniu.ingest.protocol.gb32960.model.InfoBlock.Raw} 继续解析，
|
||||
* 不再整帧放弃。关闭后恢复旧行为（任意异常直接抛 {@code DecodeException}
|
||||
* 放弃整帧），仅在灰度回滚时使用。
|
||||
*/
|
||||
private Parse parse = new Parse();
|
||||
```
|
||||
|
||||
- [ ] **Step 2.2: 在类底部（`Tls` 静态类之后、`VendorExtension` 静态类之前）添加 `Parse` 静态内嵌类**
|
||||
|
||||
```java
|
||||
/**
|
||||
* 报文解析容错策略。
|
||||
*/
|
||||
public static class Parse {
|
||||
/**
|
||||
* 单块解析异常时是否兜底为 {@link com.lingniu.ingest.protocol.gb32960.model.InfoBlock.Raw}
|
||||
* 继续解析。默认开启。
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code true}（默认）：parser 抛 DecodeException / BufferUnderflowException /
|
||||
* IndexOutOfBoundsException 时，固定长度块按 fixedLen 截取 Raw 后 continue；
|
||||
* 变长块或剩余字节不足时，剩余全部兜成 Raw 后 break 循环。
|
||||
* <li>{@code false}：保留旧行为，任意异常直接抛出 DecodeException 放弃整帧。
|
||||
* </ul>
|
||||
*/
|
||||
private boolean lenientBlockFailure = true;
|
||||
|
||||
public boolean isLenientBlockFailure() { return lenientBlockFailure; }
|
||||
public void setLenientBlockFailure(boolean lenientBlockFailure) {
|
||||
this.lenientBlockFailure = lenientBlockFailure;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2.3: 在属性 getter/setter 区域（现有 `getTls` 之后）补充 `getParse` / `setParse`**
|
||||
|
||||
```java
|
||||
public Parse getParse() { return parse; }
|
||||
public void setParse(Parse parse) { this.parse = parse; }
|
||||
```
|
||||
|
||||
- [ ] **Step 2.4: 编译验证**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 compile -q`
|
||||
Expected: BUILD SUCCESS。
|
||||
|
||||
- [ ] **Step 2.5: 提交**
|
||||
|
||||
```bash
|
||||
git add protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960Properties.java
|
||||
git commit -m "$(cat <<'EOF'
|
||||
config(gb32960): add parse.lenientBlockFailure toggle
|
||||
|
||||
新增 lingniu.ingest.gb32960.parse.lenientBlockFailure 配置开关（默认 true），
|
||||
为 Gb32960BodyParser 的单块异常隔离行为做回退开关。实现在后续 commit。
|
||||
|
||||
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||||
EOF
|
||||
)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 3: RED —— 写三个失败测试覆盖隔离场景
|
||||
|
||||
**Files:**
|
||||
- Create: `protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParserIsolationTest.java`
|
||||
|
||||
- [ ] **Step 3.1: 创建测试文件**
|
||||
|
||||
完整内容：
|
||||
|
||||
```java
|
||||
package com.lingniu.ingest.protocol.gb32960.codec;
|
||||
|
||||
import com.lingniu.ingest.api.spi.DecodeException;
|
||||
import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.AlarmV2016BlockParser;
|
||||
import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.PositionV2016BlockParser;
|
||||
import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.VehicleV2016BlockParser;
|
||||
import com.lingniu.ingest.protocol.gb32960.model.InfoBlock;
|
||||
import com.lingniu.ingest.protocol.gb32960.model.InfoBlockType;
|
||||
import com.lingniu.ingest.protocol.gb32960.model.ProtocolVersion;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* 验证 {@link Gb32960BodyParser} 单块异常隔离行为。
|
||||
*
|
||||
* <p>三个隔离场景：
|
||||
* <ol>
|
||||
* <li>固定长度块 parser 抛异常 → 失败块兜 Raw（按 fixedLen 截取）+ 后续块继续解析
|
||||
* <li>固定长度块剩余字节不足（截断帧） → 兜 Raw(剩余) + break 循环
|
||||
* <li>变长块 parser 抛异常 → 兜 Raw(从失败块起剩余全部) + break 循环
|
||||
* </ol>
|
||||
*
|
||||
* <p>外加一个严格模式回退测试：{@code lenientBlockFailure=false} 时任意异常应抛 DecodeException。
|
||||
*/
|
||||
class Gb32960BodyParserIsolationTest {
|
||||
|
||||
/** 构造一个"看起来合法但中途 parser 失败"用来注入失败的 Position parser。 */
|
||||
private static final InfoBlockParser EXPLODING_FIXED_LEN_POSITION = new InfoBlockParser() {
|
||||
@Override public ProtocolVersion version() { return ProtocolVersion.V2016; }
|
||||
@Override public int typeCode() { return 0x05; }
|
||||
@Override public int fixedLength() { return 9; }
|
||||
@Override public InfoBlock parse(ByteBuffer buffer) {
|
||||
// 模拟 parser 读了几字节后发现不合法就抛
|
||||
buffer.get(); buffer.get();
|
||||
throw new DecodeException("simulated parser failure in Position");
|
||||
}
|
||||
};
|
||||
|
||||
/** 变长块（fixedLength=-1），parse 时消费若干字节后抛。模拟 Alarm/Voltage 类列表读越界。 */
|
||||
private static final InfoBlockParser EXPLODING_VAR_LEN_ALARM = new InfoBlockParser() {
|
||||
@Override public ProtocolVersion version() { return ProtocolVersion.V2016; }
|
||||
@Override public int typeCode() { return 0x07; }
|
||||
@Override public int fixedLength() { return -1; }
|
||||
@Override public InfoBlock parse(ByteBuffer buffer) {
|
||||
// 假装读了个长度字段就爆
|
||||
buffer.get();
|
||||
throw new DecodeException("simulated parser failure in Alarm list-length read");
|
||||
}
|
||||
};
|
||||
|
||||
@Test
|
||||
void fixedLengthBlockFailure_isIsolated_subsequentBlocksStillParsed() {
|
||||
// 帧布局：Vehicle(0x01, 20B) + Position(0x05, 9B 但 parser 爆) + Engine(0x04) 不构造，
|
||||
// 简化为 Vehicle + FailingPosition + Vehicle 再次，验证"Position 之后还能继续"。
|
||||
InfoBlockParserRegistry registry = new InfoBlockParserRegistry(List.of(
|
||||
new VehicleV2016BlockParser(),
|
||||
EXPLODING_FIXED_LEN_POSITION));
|
||||
|
||||
Gb32960BodyParser parser = new Gb32960BodyParser(registry);
|
||||
|
||||
ByteArrayOutputStream os = new ByteArrayOutputStream();
|
||||
writeValidVehicle(os); // 1+20 = 21B
|
||||
writePositionTypeAnd9ByteBody(os); // 1+9 = 10B (parser 会爆)
|
||||
writeValidVehicle(os); // 1+20 = 21B
|
||||
ByteBuffer body = ByteBuffer.wrap(os.toByteArray());
|
||||
|
||||
var result = parser.parse(ProtocolVersion.V2016, body);
|
||||
|
||||
assertThat(result.blocks()).hasSize(3);
|
||||
assertThat(result.blocks().get(0)).isInstanceOf(InfoBlock.Gb32960V2016.Vehicle.class);
|
||||
assertThat(result.blocks().get(1)).isInstanceOfSatisfying(InfoBlock.Raw.class, raw -> {
|
||||
assertThat(raw.typeCode()).isEqualTo(0x05);
|
||||
assertThat(raw.type()).isEqualTo(InfoBlockType.RAW);
|
||||
assertThat(raw.bytes()).hasSize(9); // 按 fixedLen 截取
|
||||
});
|
||||
assertThat(result.blocks().get(2)).isInstanceOf(InfoBlock.Gb32960V2016.Vehicle.class);
|
||||
assertThat(body.hasRemaining()).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
void truncatedFixedLengthBlock_isWrappedAsRaw_loopTerminates() {
|
||||
// Vehicle(21B) + Position(type 0x05)但 body 只给 3B —— 不足 9B
|
||||
InfoBlockParserRegistry registry = new InfoBlockParserRegistry(List.of(
|
||||
new VehicleV2016BlockParser(),
|
||||
new PositionV2016BlockParser()));
|
||||
|
||||
Gb32960BodyParser parser = new Gb32960BodyParser(registry);
|
||||
|
||||
ByteArrayOutputStream os = new ByteArrayOutputStream();
|
||||
writeValidVehicle(os);
|
||||
os.write(0x05); // Position typeCode
|
||||
os.write(0); os.write(0); os.write(0); // 只 3B，远远不够 9B
|
||||
ByteBuffer body = ByteBuffer.wrap(os.toByteArray());
|
||||
|
||||
var result = parser.parse(ProtocolVersion.V2016, body);
|
||||
|
||||
assertThat(result.blocks()).hasSize(2);
|
||||
assertThat(result.blocks().get(0)).isInstanceOf(InfoBlock.Gb32960V2016.Vehicle.class);
|
||||
assertThat(result.blocks().get(1)).isInstanceOfSatisfying(InfoBlock.Raw.class, raw -> {
|
||||
assertThat(raw.typeCode()).isEqualTo(0x05);
|
||||
assertThat(raw.bytes()).hasSize(3); // 剩余字节全兜成 Raw
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void variableLengthBlockFailure_swallowsRemainderAsRaw_loopBreaks() {
|
||||
// Vehicle(21B) + FailingAlarm(0x07)后面还有一个 Vehicle，但因 Alarm 变长无法安全跳过，
|
||||
// 整段 Alarm 起的剩余字节都被兜成 Raw 后 break。
|
||||
InfoBlockParserRegistry registry = new InfoBlockParserRegistry(List.of(
|
||||
new VehicleV2016BlockParser(),
|
||||
EXPLODING_VAR_LEN_ALARM));
|
||||
|
||||
Gb32960BodyParser parser = new Gb32960BodyParser(registry);
|
||||
|
||||
ByteArrayOutputStream os = new ByteArrayOutputStream();
|
||||
writeValidVehicle(os); // 21B
|
||||
os.write(0x07); // Alarm typeCode
|
||||
for (int i = 0; i < 10; i++) os.write(0xAA); // 10B 的 alarm body（parser 爆）
|
||||
writeValidVehicle(os); // 21B（应当不再被解析）
|
||||
ByteBuffer body = ByteBuffer.wrap(os.toByteArray());
|
||||
|
||||
var result = parser.parse(ProtocolVersion.V2016, body);
|
||||
|
||||
assertThat(result.blocks()).hasSize(2);
|
||||
assertThat(result.blocks().get(0)).isInstanceOf(InfoBlock.Gb32960V2016.Vehicle.class);
|
||||
assertThat(result.blocks().get(1)).isInstanceOfSatisfying(InfoBlock.Raw.class, raw -> {
|
||||
assertThat(raw.typeCode()).isEqualTo(0x07);
|
||||
// 剩余 10B alarm body + 1+20=21B 后续 Vehicle = 31B
|
||||
assertThat(raw.bytes()).hasSize(31);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void strictMode_throwsOnAnyBlockFailure() {
|
||||
InfoBlockParserRegistry registry = new InfoBlockParserRegistry(List.of(
|
||||
new VehicleV2016BlockParser(),
|
||||
EXPLODING_FIXED_LEN_POSITION));
|
||||
|
||||
Gb32960BodyParser parser = new Gb32960BodyParser(registry);
|
||||
parser.setLenientBlockFailure(false);
|
||||
|
||||
ByteArrayOutputStream os = new ByteArrayOutputStream();
|
||||
writeValidVehicle(os);
|
||||
writePositionTypeAnd9ByteBody(os);
|
||||
ByteBuffer body = ByteBuffer.wrap(os.toByteArray());
|
||||
|
||||
assertThatThrownBy(() -> parser.parse(ProtocolVersion.V2016, body))
|
||||
.isInstanceOf(DecodeException.class);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------
|
||||
// helpers
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
/** 写一个完整的合法 V2016 Vehicle 块（typeCode + 20B body）。 */
|
||||
private static void writeValidVehicle(ByteArrayOutputStream os) {
|
||||
os.write(0x01); // typeCode Vehicle
|
||||
os.write(0x01); // vehicleState=1
|
||||
os.write(0x01); // chargingState=1
|
||||
os.write(0x01); // runningMode=1
|
||||
os.write(0); os.write(0); // speed=0
|
||||
os.write(0); os.write(0); os.write(0); os.write(0); // mileage=0
|
||||
os.write(0); os.write(0); // totalVoltage=0
|
||||
os.write(0); os.write(0); // totalCurrent=0（offset 1000，原值0）
|
||||
os.write(50); // soc=50%
|
||||
os.write(0x01); // dcdc
|
||||
os.write(0); // gear
|
||||
os.write(0); os.write(0); // insulation
|
||||
os.write(0); // accelerator
|
||||
os.write(0); // brake
|
||||
}
|
||||
|
||||
/** 写 Position typeCode(0x05) + 9 字节 body（内容不重要，parser 替换成 EXPLODING 的）。 */
|
||||
private static void writePositionTypeAnd9ByteBody(ByteArrayOutputStream os) {
|
||||
os.write(0x05);
|
||||
for (int i = 0; i < 9; i++) os.write(0);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3.2: 跑测试验证 RED**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 test -Dtest=Gb32960BodyParserIsolationTest -q`
|
||||
Expected:
|
||||
- `fixedLengthBlockFailure_isIsolated_subsequentBlocksStillParsed` FAIL（当前会抛 DecodeException）
|
||||
- `truncatedFixedLengthBlock_isWrappedAsRaw_loopTerminates` FAIL（当前 L132 会抛 DecodeException）
|
||||
- `variableLengthBlockFailure_swallowsRemainderAsRaw_loopBreaks` FAIL（异常冒出整帧失败）
|
||||
- `strictMode_throwsOnAnyBlockFailure` FAIL（没有 `setLenientBlockFailure` 方法，编译就红）
|
||||
|
||||
**验证点**：编译失败（strictMode test 里 `setLenientBlockFailure` 尚未存在）是**预期** RED 信号——下一步实现。
|
||||
|
||||
---
|
||||
|
||||
## Task 4: GREEN —— 实现单块异常隔离
|
||||
|
||||
**Files:**
|
||||
- Modify: `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParser.java`
|
||||
|
||||
- [ ] **Step 4.1: 在类字段区新增 `lenientBlockFailure` 字段 + setter**
|
||||
|
||||
在 `private final VendorExtensionSelector selector;`（现 L54）下面插入：
|
||||
|
||||
```java
|
||||
/**
|
||||
* 单块解析异常时是否兜底为 Raw 后继续。由
|
||||
* {@link com.lingniu.ingest.protocol.gb32960.config.Gb32960Properties.Parse#isLenientBlockFailure()}
|
||||
* 注入；默认 true。
|
||||
*/
|
||||
private boolean lenientBlockFailure = true;
|
||||
|
||||
public void setLenientBlockFailure(boolean lenientBlockFailure) {
|
||||
this.lenientBlockFailure = lenientBlockFailure;
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4.2: 改造主循环：替换 L130~L150 的整个 "parse + 长度校验" 段**
|
||||
|
||||
**完整替换块**：以下代码替换原来从 `int fixedLen = parser.fixedLength();`（L130）开始到 `blocks.add(block);`（L150）结束的整块。
|
||||
|
||||
```java
|
||||
int fixedLen = parser.fixedLength();
|
||||
int posBefore = body.position();
|
||||
|
||||
// 旧行为：fixedLen 预检不足 → 直接抛。新行为（lenient 模式）：走统一恢复路径。
|
||||
if (fixedLen >= 0 && body.remaining() < fixedLen) {
|
||||
if (!lenientBlockFailure) {
|
||||
throw new DecodeException(
|
||||
"info block 0x" + Integer.toHexString(typeCode)
|
||||
+ " needs " + fixedLen + " bytes but got " + body.remaining());
|
||||
}
|
||||
// 剩余字节数不足 fixedLen —— 无法按块截取，整尾兜 Raw + break
|
||||
int remaining = body.remaining();
|
||||
byte[] tail = new byte[remaining];
|
||||
body.get(tail);
|
||||
log.warn("[gb32960] truncated block typeCode=0x{} declaredFixedLen={} remaining={} — wrapping tail as Raw",
|
||||
Integer.toHexString(typeCode), fixedLen, remaining);
|
||||
blocks.add(new InfoBlock.Raw(typeCode, InfoBlockType.RAW, tail));
|
||||
break;
|
||||
}
|
||||
|
||||
InfoBlock block;
|
||||
try {
|
||||
block = parser.parse(body);
|
||||
} catch (DecodeException | java.nio.BufferUnderflowException | IndexOutOfBoundsException e) {
|
||||
if (!lenientBlockFailure) {
|
||||
if (e instanceof DecodeException de) throw de;
|
||||
throw new DecodeException(
|
||||
"parser " + parser.getClass().getSimpleName() + " failed: " + e.getMessage(), e);
|
||||
}
|
||||
// 恢复路径：先回滚 position 到 parse 入口
|
||||
body.position(posBefore);
|
||||
if (fixedLen >= 0) {
|
||||
// 固定长度块：按 fixedLen 截取 → 兜 Raw → continue 循环
|
||||
byte[] corrupt = new byte[fixedLen];
|
||||
body.get(corrupt);
|
||||
log.warn("[gb32960] block parse failed typeCode=0x{} parser={} fixedLen={} — isolated as Raw, continuing",
|
||||
Integer.toHexString(typeCode), parser.getClass().getSimpleName(), fixedLen, e);
|
||||
blocks.add(new InfoBlock.Raw(typeCode, InfoBlockType.RAW, corrupt));
|
||||
continue;
|
||||
} else {
|
||||
// 变长块：无法安全找下一块边界 → 剩余整段兜 Raw + break
|
||||
int remaining = body.remaining();
|
||||
byte[] tail = new byte[remaining];
|
||||
body.get(tail);
|
||||
log.warn("[gb32960] variable-length block parse failed typeCode=0x{} parser={} remaining={} — wrapping remainder as Raw, stopping loop",
|
||||
Integer.toHexString(typeCode), parser.getClass().getSimpleName(), remaining, e);
|
||||
blocks.add(new InfoBlock.Raw(typeCode, InfoBlockType.RAW, tail));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
int consumed = body.position() - posBefore;
|
||||
if (fixedLen >= 0 && consumed != fixedLen) {
|
||||
// 这是 parser 契约违反（声明 fixedLen=X 但实际读了 Y），保持抛异常以暴露 parser bug。
|
||||
throw new DecodeException(
|
||||
"parser " + parser.getClass().getSimpleName()
|
||||
+ " consumed " + consumed
|
||||
+ " bytes but declared " + fixedLen);
|
||||
}
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("[gb32960] block typeCode=0x{} parser={} pos {}→{} consumed={} declaredFixed={}",
|
||||
Integer.toHexString(typeCode), parser.getClass().getSimpleName(),
|
||||
typeStartPos, body.position(), consumed, fixedLen);
|
||||
}
|
||||
blocks.add(block);
|
||||
```
|
||||
|
||||
- [ ] **Step 4.3: 运行隔离测试验证 GREEN**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 test -Dtest=Gb32960BodyParserIsolationTest -q`
|
||||
Expected: 4 个测试全部 PASS。
|
||||
|
||||
- [ ] **Step 4.4: 运行全模块测试检查回归**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 test -q`
|
||||
Expected: BUILD SUCCESS，所有原有测试（含 Golden、FullBlocks、GuangdongFcEndToEnd）依旧通过。
|
||||
|
||||
如果有原有测试红了：**停下来诊断**。最可能的原因是某个现有测试之前靠"抛异常"行为验证错误帧的，需要手动判断这是测试问题还是实现问题。
|
||||
|
||||
- [ ] **Step 4.5: 提交**
|
||||
|
||||
```bash
|
||||
git add protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParser.java \
|
||||
protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/codec/Gb32960BodyParserIsolationTest.java
|
||||
git commit -m "$(cat <<'EOF'
|
||||
feat(gb32960): isolate per-block parse failures in body parser
|
||||
|
||||
主循环改造：单信息块 parser 抛 DecodeException / BufferUnderflowException /
|
||||
IndexOutOfBoundsException 时不再放弃整帧。
|
||||
|
||||
- 固定长度块（fixedLen≥0）：回滚 reader，按 fixedLen 截取字节兜成 InfoBlock.Raw，
|
||||
position 前进 fixedLen，continue 循环继续解析后续块
|
||||
- 变长块（fixedLen=-1）或剩余不足 fixedLen：剩余字节全部兜成 Raw 后 break
|
||||
- parser 契约违反（consumed != declared fixedLen）仍抛异常，以暴露 parser bug
|
||||
- 通过 lingniu.ingest.gb32960.parse.lenientBlockFailure=false 可回退严格模式
|
||||
- 新增 Gb32960BodyParserIsolationTest 覆盖 3 种失败场景 + 1 严格模式
|
||||
|
||||
不改 InfoBlock.Raw record 结构；下游 Gb32960EventMapper 按 findBlock(Class) 类型
|
||||
查找，新增 Raw 不影响其行为。
|
||||
|
||||
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||||
EOF
|
||||
)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 5: 把配置注入 BodyParser（AutoConfiguration 连线）
|
||||
|
||||
**Files:**
|
||||
- Modify: `protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java`
|
||||
|
||||
- [ ] **Step 5.1: 定位 `Gb32960BodyParser` bean 定义**
|
||||
|
||||
先查看：`grep -n "Gb32960BodyParser" protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java`
|
||||
|
||||
根据返回的行号，在构造 `Gb32960BodyParser` 的 @Bean 方法里，构造完成后立即调用：
|
||||
|
||||
```java
|
||||
Gb32960BodyParser parser = new Gb32960BodyParser(profileRegistry, selector);
|
||||
parser.setLenientBlockFailure(properties.getParse().isLenientBlockFailure());
|
||||
return parser;
|
||||
```
|
||||
|
||||
（如果目前的构造返回一行表达式，改成先赋给 local 变量再设 flag 再 return。）
|
||||
|
||||
- [ ] **Step 5.2: 编译并运行全模块测试**
|
||||
|
||||
Run: `mvn -pl protocol-gb32960 test -q`
|
||||
Expected: BUILD SUCCESS。
|
||||
|
||||
- [ ] **Step 5.3: 提交**
|
||||
|
||||
```bash
|
||||
git add protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java
|
||||
git commit -m "$(cat <<'EOF'
|
||||
config(gb32960): wire parse.lenientBlockFailure into BodyParser bean
|
||||
|
||||
Gb32960AutoConfiguration 在创建 Gb32960BodyParser bean 后，将
|
||||
Gb32960Properties.Parse.lenientBlockFailure 通过 setter 注入，让运行期配置生效。
|
||||
|
||||
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||||
EOF
|
||||
)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 6: bootstrap-all 配置文档化
|
||||
|
||||
**Files:**
|
||||
- Modify: `bootstrap-all/src/main/resources/application.yml`
|
||||
|
||||
- [ ] **Step 6.1: 在 `lingniu.ingest.gb32960` 节点的 `vendor-extensions` 段之后、`jt808` 之前，插入 parse 配置示例**
|
||||
|
||||
参考位置：现有 application.yml L57 末尾（`vendor-extensions` list 结束）。在 L58 `jt808:` 之前加入：
|
||||
|
||||
```yaml
|
||||
# 报文解析容错。默认启用单块异常隔离：某个信息块解析失败时兜成 Raw 后继续，
|
||||
# 不再整帧丢弃。仅在需要严格失败语义（灰度回滚或抓虫）时设为 false。
|
||||
parse:
|
||||
lenient-block-failure: true
|
||||
```
|
||||
|
||||
- [ ] **Step 6.2: 启动一次 bootstrap-all 的 dry-run 编译**
|
||||
|
||||
Run: `mvn -pl bootstrap-all compile -q`
|
||||
Expected: BUILD SUCCESS（只是配置段注释变化，不会影响编译）。
|
||||
|
||||
- [ ] **Step 6.3: 提交**
|
||||
|
||||
```bash
|
||||
git add bootstrap-all/src/main/resources/application.yml
|
||||
git commit -m "$(cat <<'EOF'
|
||||
config(gb32960): document parse.lenient-block-failure in application.yml
|
||||
|
||||
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||||
EOF
|
||||
)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 7: 全仓库回归测试
|
||||
|
||||
**Files:** 无（只是验证）
|
||||
|
||||
- [ ] **Step 7.1: 跑完整项目测试**
|
||||
|
||||
Run: `mvn test -q`
|
||||
Expected: BUILD SUCCESS。重点关注：
|
||||
- `protocol-gb32960` 模块全绿（Golden、FullBlocks、Isolation、GuangdongFcEndToEnd、Decoder、Mapper、所有 parser 单测、profile）
|
||||
- 其它模块（`ingest-core`、`sink-kafka` 等）不受影响（无改动应自动通过）
|
||||
|
||||
若有红，停下来诊断。
|
||||
|
||||
---
|
||||
|
||||
## Task 8: 更新 CHANGELOG
|
||||
|
||||
**Files:**
|
||||
- Modify: `CHANGELOG.md`
|
||||
|
||||
- [ ] **Step 8.1: 在 CHANGELOG.md 文件顶部（在 `## [0.1.0] — 2026-04-15` 标题之前）新增一个 `## [Unreleased]` 段落；如已存在则追加**
|
||||
|
||||
新增段落内容（如已存在 Unreleased 段则在其 `### Added` / `### Changed` 内部追加）：
|
||||
|
||||
```markdown
|
||||
## [Unreleased]
|
||||
|
||||
### Changed —— GB/T 32960 Body Parser 单块异常隔离
|
||||
- `Gb32960BodyParser` 主循环对单信息块的 parser 异常不再放弃整帧：
|
||||
- 固定长度块（`fixedLength ≥ 0`）：回滚 reader → 按 fixedLen 截取字节兜成
|
||||
`InfoBlock.Raw` → 继续解析后续块；
|
||||
- 变长块或剩余字节不足：剩余字节全部兜成 `Raw` 后终止循环；
|
||||
- `parser` 契约违反（声明 fixedLen=X 但实际读了 Y）仍抛 `DecodeException`，
|
||||
以暴露 parser bug，不被静默吞掉；
|
||||
- 只捕获 `DecodeException` / `BufferUnderflowException` / `IndexOutOfBoundsException`，
|
||||
`RuntimeException` 继续向上抛，保留 bug 可见性。
|
||||
- 新增配置 `lingniu.ingest.gb32960.parse.lenient-block-failure`（默认 `true`）；
|
||||
回退严格模式用 `false`。
|
||||
- `InfoBlock.Raw` record 结构**未变**，下游 `Gb32960EventMapper` 按 `findBlock(Class)`
|
||||
类型匹配，新增 Raw 不影响业务事件映射。
|
||||
- 覆盖测试：`Gb32960BodyParserIsolationTest` —— 3 种隔离场景 + 严格模式回退。
|
||||
```
|
||||
|
||||
- [ ] **Step 8.2: 提交**
|
||||
|
||||
```bash
|
||||
git add CHANGELOG.md
|
||||
git commit -m "$(cat <<'EOF'
|
||||
docs(changelog): record gb32960 body parser block isolation
|
||||
|
||||
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||||
EOF
|
||||
)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Self-Review Checklist
|
||||
|
||||
- [x] **Spec coverage**: 原讨论的 A1~A7 全部覆盖——A1 状态机（Task 4 Step 4.2）、A2 三类错误策略（同）、A3 收窄异常种类（同，只 catch 三种）、A4 日志分类（warn 带 parser 名/typeCode/原因）、A5 配置开关（Task 2）、A6 合成帧测试（Task 3 三场景 + 严格回退）、A7 下游兼容（不改 Raw record，`findBlock` 类型匹配不受影响，plan 中已说明）。
|
||||
- [x] **Placeholder scan**: 无 TBD / TODO / "实现上面的" 等占位；每个 Step 含实际代码 or 命令。
|
||||
- [x] **Type consistency**: `lenientBlockFailure` 在 Properties / BodyParser 字段 / setter / 测试中拼写一致。`InfoBlockType.RAW` 枚举值依赖当前存在（已核对 InfoBlock.Raw record）。
|
||||
- [x] **测试命名一致**：`Gb32960BodyParserIsolationTest` 在 Task 3 创建、Task 4/7 引用一致。
|
||||
- [x] **未引入新枚举/proto 变更**：InfoBlock.Raw 沿用现有 `(int typeCode, InfoBlockType type, byte[] bytes)` 构造。
|
||||
@@ -1,817 +0,0 @@
|
||||
> **Superseded:** This 2026-06-23 DuckDB hot-store plan is historical context.
|
||||
> Use `docs/target-architecture.md` and
|
||||
> `docs/superpowers/specs/2026-06-29-vehicle-ingest-redesign.md` for the
|
||||
> current production architecture: TDengine stores hot history, `sink-archive`
|
||||
> owns raw bytes, and `event-file-store` / DuckDB are no longer part of the
|
||||
> current build surface.
|
||||
|
||||
# GB32960 History DuckDB Hot Store Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Replace the current high-write-risk Parquet rewrite history path with a DuckDB hot history store that supports fast `vin + time` queries and full RAW frame replay.
|
||||
|
||||
**Architecture:** Add a new `DuckDbHotEventFileStore` implementation behind the existing `EventFileStore` interface, backed by append/upsert DuckDB tables instead of rewriting per-vehicle Parquet files. Keep the current HTTP history API and RAW archive reader intact, then switch `vehicle-history-app` to the hot DuckDB store by configuration.
|
||||
|
||||
**Tech Stack:** Java 26, Spring Boot, DuckDB JDBC, JUnit 5, AssertJ, existing Kafka `VehicleEnvelope`, existing `ArchiveStore`.
|
||||
|
||||
---
|
||||
|
||||
## Scope
|
||||
|
||||
This plan implements Phase 1 from `docs/superpowers/specs/2026-06-23-gb32960-production-readiness-design.md`.
|
||||
|
||||
Included:
|
||||
|
||||
- DuckDB hot event table for `EventFileRecord`.
|
||||
- Idempotent writes by `event_id`.
|
||||
- Query by protocol/date/VIN/eventType/eventTime/order/limit.
|
||||
- Lookup by `rawArchiveUri`.
|
||||
- Spring configuration to select the hot store.
|
||||
- Tests that prove history endpoints still replay RAW frames.
|
||||
- Local verification using the existing `vehicle-history-app`.
|
||||
|
||||
Not included in this plan:
|
||||
|
||||
- MySQL daily statistics.
|
||||
- VIN-to-platform local mapping.
|
||||
- Full telemetry point columnar table.
|
||||
- Alarm timeline MySQL tables.
|
||||
- Cold Parquet export.
|
||||
|
||||
Those will be separate plans after this foundation lands.
|
||||
|
||||
## Files
|
||||
|
||||
- Create: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
|
||||
- Create: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
|
||||
- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java`
|
||||
- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java`
|
||||
- Modify: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java`
|
||||
- Modify: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
|
||||
- Modify: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
|
||||
- Modify: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java`
|
||||
|
||||
## Task 1: Add Hot Store Configuration
|
||||
|
||||
**Files:**
|
||||
- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java`
|
||||
- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java`
|
||||
- Test: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java`
|
||||
|
||||
- [ ] **Step 1: Write the failing auto-configuration test**
|
||||
|
||||
Add a test that sets `lingniu.ingest.event-file-store.storage=duckdb-hot` and expects the bean class to be `DuckDbHotEventFileStore`.
|
||||
|
||||
```java
|
||||
@Test
|
||||
void createsDuckDbHotStoreWhenStorageIsDuckDbHot() {
|
||||
contextRunner
|
||||
.withPropertyValues(
|
||||
"lingniu.ingest.event-file-store.enabled=true",
|
||||
"lingniu.ingest.event-file-store.storage=duckdb-hot",
|
||||
"lingniu.ingest.event-file-store.path=" + tempDir.resolve("history"))
|
||||
.run(context -> assertThat(context)
|
||||
.hasSingleBean(EventFileStore.class)
|
||||
.getBean(EventFileStore.class)
|
||||
.isInstanceOf(DuckDbHotEventFileStore.class));
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the new test and verify it fails**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store -Dtest=EventFileStoreAutoConfigurationTest#createsDuckDbHotStoreWhenStorageIsDuckDbHot test
|
||||
```
|
||||
|
||||
Expected: compilation failure because `DuckDbHotEventFileStore` and `storage` property do not exist.
|
||||
|
||||
- [ ] **Step 3: Add the `storage` property**
|
||||
|
||||
Add to `EventFileStoreProperties`:
|
||||
|
||||
```java
|
||||
/**
|
||||
* Storage backend. `duckdb-hot` is the production backend; `parquet-sidecar`
|
||||
* keeps the previous implementation available for compatibility tests.
|
||||
*/
|
||||
private String storage = "duckdb-hot";
|
||||
|
||||
public String getStorage() {
|
||||
return storage;
|
||||
}
|
||||
|
||||
public void setStorage(String storage) {
|
||||
this.storage = storage;
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Switch auto-configuration by storage mode**
|
||||
|
||||
Change `eventFileStore(...)` to:
|
||||
|
||||
```java
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public EventFileStore eventFileStore(EventFileStoreProperties properties,
|
||||
ObjectProvider<ObjectMapper> objectMapper) {
|
||||
ObjectMapper mapper = mapper(objectMapper);
|
||||
Path root = Path.of(properties.getPath());
|
||||
ZoneId zoneId = ZoneId.of(properties.getZoneId());
|
||||
String storage = properties.getStorage() == null ? "" : properties.getStorage().trim();
|
||||
return switch (storage) {
|
||||
case "", "duckdb-hot" -> new DuckDbHotEventFileStore(root, zoneId, mapper);
|
||||
case "parquet-sidecar" -> new DuckDbParquetEventFileStore(root, zoneId, mapper);
|
||||
default -> throw new IllegalStateException(
|
||||
"unsupported event-file-store storage: " + properties.getStorage());
|
||||
};
|
||||
}
|
||||
```
|
||||
|
||||
Import `DuckDbHotEventFileStore`.
|
||||
|
||||
- [ ] **Step 5: Run configuration tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store -Dtest=EventFileStoreAutoConfigurationTest test
|
||||
```
|
||||
|
||||
Expected: tests pass after the store class exists in Task 2.
|
||||
|
||||
## Task 2: Implement DuckDB Hot Store Schema and Writes
|
||||
|
||||
**Files:**
|
||||
- Create: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
|
||||
- Create: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
|
||||
|
||||
- [ ] **Step 1: Write failing append/query/idempotency tests**
|
||||
|
||||
Create `DuckDbHotEventFileStoreTest`:
|
||||
|
||||
```java
|
||||
class DuckDbHotEventFileStoreTest {
|
||||
|
||||
@TempDir
|
||||
Path tempDir;
|
||||
|
||||
@Test
|
||||
void appendsRecordsAndQueriesByVinTypeAndTime() throws Exception {
|
||||
EventFileStore store = store();
|
||||
store.append(rawRecord("raw-1", "VIN001", "2026-06-23T01:00:00Z"));
|
||||
store.append(rawRecord("raw-2", "VIN002", "2026-06-23T01:00:01Z"));
|
||||
store.append(rawRecord("raw-3", "VIN001", "2026-06-23T01:00:02Z"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-23"),
|
||||
LocalDate.parse("2026-06-23"),
|
||||
Instant.parse("2026-06-23T01:00:00Z"),
|
||||
Instant.parse("2026-06-23T01:00:03Z"),
|
||||
EventFileQuery.Order.DESC,
|
||||
2,
|
||||
"VIN001",
|
||||
"RAW_ARCHIVE");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("raw-3", "raw-1");
|
||||
}
|
||||
|
||||
@Test
|
||||
void appendAllIsIdempotentByEventId() throws Exception {
|
||||
EventFileStore store = store();
|
||||
EventFileRecord original = rawRecord("same-id", "VIN001", "2026-06-23T01:00:00Z");
|
||||
EventFileRecord replacement = new EventFileRecord(
|
||||
"same-id",
|
||||
ProtocolId.GB32960,
|
||||
"RAW_ARCHIVE",
|
||||
"VIN001",
|
||||
Instant.parse("2026-06-23T01:00:05Z"),
|
||||
Instant.parse("2026-06-23T01:00:06Z"),
|
||||
"archive://replacement.bin",
|
||||
Map.of("source", "replacement"),
|
||||
"{\"replacement\":true}");
|
||||
|
||||
store.appendAll(List.of(original, replacement));
|
||||
|
||||
assertThat(store.query(new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-23"),
|
||||
LocalDate.parse("2026-06-23"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10,
|
||||
"VIN001",
|
||||
"RAW_ARCHIVE")))
|
||||
.singleElement()
|
||||
.satisfies(record -> {
|
||||
assertThat(record.eventId()).isEqualTo("same-id");
|
||||
assertThat(record.rawArchiveUri()).isEqualTo("archive://replacement.bin");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void findsRecordByRawArchiveUri() throws Exception {
|
||||
EventFileStore store = store();
|
||||
EventFileRecord record = rawRecord("raw-uri", "VIN001", "2026-06-23T01:00:00Z");
|
||||
store.append(record);
|
||||
|
||||
EventFileRecord found = store.findByRawArchiveUri(record.rawArchiveUri());
|
||||
|
||||
assertThat(found).isNotNull();
|
||||
assertThat(found.eventId()).isEqualTo("raw-uri");
|
||||
}
|
||||
|
||||
private EventFileStore store() {
|
||||
return new DuckDbHotEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), new ObjectMapper());
|
||||
}
|
||||
|
||||
private static EventFileRecord rawRecord(String id, String vin, String eventTime) {
|
||||
return new EventFileRecord(
|
||||
id,
|
||||
ProtocolId.GB32960,
|
||||
"RAW_ARCHIVE",
|
||||
vin,
|
||||
Instant.parse(eventTime),
|
||||
Instant.parse(eventTime).plusMillis(100),
|
||||
"archive://" + id + ".bin",
|
||||
Map.of("platformAccount", "Hyundai", "command", "REALTIME_REPORT"),
|
||||
"{\"eventId\":\"" + id + "\"}");
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the test and verify it fails**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
|
||||
```
|
||||
|
||||
Expected: compilation failure because `DuckDbHotEventFileStore` does not exist.
|
||||
|
||||
- [ ] **Step 3: Create the hot store class**
|
||||
|
||||
Create `DuckDbHotEventFileStore` with this structure:
|
||||
|
||||
```java
|
||||
public final class DuckDbHotEventFileStore implements EventFileStore {
|
||||
|
||||
private static final TypeReference<Map<String, String>> STRING_MAP = new TypeReference<>() {};
|
||||
|
||||
private final Path root;
|
||||
private final Path dbPath;
|
||||
private final ZoneId partitionZone;
|
||||
private final ObjectMapper objectMapper;
|
||||
private volatile boolean initialized;
|
||||
|
||||
public DuckDbHotEventFileStore(Path root, ZoneId partitionZone) {
|
||||
this(root, partitionZone, new ObjectMapper());
|
||||
}
|
||||
|
||||
public DuckDbHotEventFileStore(Path root, ZoneId partitionZone, ObjectMapper objectMapper) {
|
||||
if (root == null) {
|
||||
throw new IllegalArgumentException("root must not be null");
|
||||
}
|
||||
this.root = root.toAbsolutePath();
|
||||
this.dbPath = this.root.resolve("events.duckdb");
|
||||
this.partitionZone = partitionZone == null ? ZoneId.of("Asia/Shanghai") : partitionZone;
|
||||
this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void appendAll(List<EventFileRecord> records) throws IOException {
|
||||
if (records == null || records.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
ensureInitialized();
|
||||
try (Connection connection = DriverManager.getConnection(jdbcUrl())) {
|
||||
connection.setAutoCommit(false);
|
||||
try {
|
||||
upsertRecords(connection, records);
|
||||
connection.commit();
|
||||
} catch (SQLException | IOException e) {
|
||||
connection.rollback();
|
||||
throw e;
|
||||
} finally {
|
||||
connection.setAutoCommit(true);
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("write duckdb hot event store failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<EventFileRecord> query(EventFileQuery query) throws IOException {
|
||||
ensureInitialized();
|
||||
// Implement in Task 3.
|
||||
return List.of();
|
||||
}
|
||||
|
||||
@Override
|
||||
public EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
|
||||
ensureInitialized();
|
||||
// Implement in Task 3.
|
||||
return null;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Add schema initialization**
|
||||
|
||||
Add:
|
||||
|
||||
```java
|
||||
private void ensureInitialized() throws IOException {
|
||||
if (initialized) {
|
||||
return;
|
||||
}
|
||||
synchronized (this) {
|
||||
if (initialized) {
|
||||
return;
|
||||
}
|
||||
Files.createDirectories(root);
|
||||
try (Connection connection = DriverManager.getConnection(jdbcUrl());
|
||||
Statement statement = connection.createStatement()) {
|
||||
statement.execute("""
|
||||
CREATE TABLE IF NOT EXISTS event_records (
|
||||
event_id VARCHAR PRIMARY KEY,
|
||||
protocol VARCHAR NOT NULL,
|
||||
event_type VARCHAR NOT NULL,
|
||||
vin VARCHAR NOT NULL,
|
||||
event_time_ms BIGINT NOT NULL,
|
||||
ingest_time_ms BIGINT NOT NULL,
|
||||
partition_date DATE NOT NULL,
|
||||
raw_archive_uri VARCHAR NOT NULL,
|
||||
metadata_json VARCHAR NOT NULL,
|
||||
payload_json VARCHAR NOT NULL
|
||||
)
|
||||
""");
|
||||
statement.execute("""
|
||||
CREATE INDEX IF NOT EXISTS event_records_protocol_date_time_idx
|
||||
ON event_records(protocol, partition_date, event_time_ms)
|
||||
""");
|
||||
statement.execute("""
|
||||
CREATE INDEX IF NOT EXISTS event_records_vin_date_time_idx
|
||||
ON event_records(protocol, vin, partition_date, event_time_ms)
|
||||
""");
|
||||
statement.execute("""
|
||||
CREATE INDEX IF NOT EXISTS event_records_vin_type_date_time_idx
|
||||
ON event_records(protocol, vin, event_type, partition_date, event_time_ms)
|
||||
""");
|
||||
statement.execute("""
|
||||
CREATE INDEX IF NOT EXISTS event_records_raw_archive_uri_idx
|
||||
ON event_records(raw_archive_uri)
|
||||
""");
|
||||
initialized = true;
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("initialize duckdb hot event store failed", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Add idempotent batch upsert**
|
||||
|
||||
Use DuckDB `INSERT OR REPLACE` inside one transaction:
|
||||
|
||||
```java
|
||||
private void upsertRecords(Connection connection, List<EventFileRecord> records)
|
||||
throws SQLException, IOException {
|
||||
try (PreparedStatement ps = connection.prepareStatement("""
|
||||
INSERT OR REPLACE INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""")) {
|
||||
for (EventFileRecord record : records) {
|
||||
ps.setString(1, record.eventId());
|
||||
ps.setString(2, record.protocol().name());
|
||||
ps.setString(3, record.eventType());
|
||||
ps.setString(4, record.vin());
|
||||
ps.setLong(5, record.eventTime().toEpochMilli());
|
||||
ps.setLong(6, record.ingestTime().toEpochMilli());
|
||||
ps.setString(7, LocalDate.ofInstant(record.eventTime(), partitionZone).toString());
|
||||
ps.setString(8, record.rawArchiveUri());
|
||||
ps.setString(9, objectMapper.writeValueAsString(record.metadata()));
|
||||
ps.setString(10, record.payloadJson());
|
||||
ps.addBatch();
|
||||
}
|
||||
ps.executeBatch();
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 6: Run the hot store tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
|
||||
```
|
||||
|
||||
Expected: query tests still fail until Task 3 implements reads; append initialization should compile.
|
||||
|
||||
## Task 3: Implement Hot Store Queries and Raw URI Lookup
|
||||
|
||||
**Files:**
|
||||
- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
|
||||
- Test: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
|
||||
|
||||
- [ ] **Step 1: Implement `query(EventFileQuery)`**
|
||||
|
||||
Use prepared statements for every external value:
|
||||
|
||||
```java
|
||||
@Override
|
||||
public List<EventFileRecord> query(EventFileQuery query) throws IOException {
|
||||
ensureInitialized();
|
||||
String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC";
|
||||
StringBuilder where = new StringBuilder("""
|
||||
WHERE protocol = ?
|
||||
AND partition_date BETWEEN CAST(? AS DATE) AND CAST(? AS DATE)
|
||||
""");
|
||||
if (query.vin() != null) {
|
||||
where.append(" AND vin = ?\n");
|
||||
}
|
||||
if (query.eventType() != null) {
|
||||
where.append(" AND event_type = ?\n");
|
||||
}
|
||||
if (query.eventTimeFrom() != null) {
|
||||
where.append(" AND event_time_ms >= ?\n");
|
||||
}
|
||||
if (query.eventTimeTo() != null) {
|
||||
where.append(" AND event_time_ms <= ?\n");
|
||||
}
|
||||
String sql = """
|
||||
SELECT event_id, protocol, event_type, vin, event_time_ms, ingest_time_ms,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM event_records
|
||||
%s
|
||||
ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s
|
||||
LIMIT ?
|
||||
""".formatted(where, order, order, order);
|
||||
try (Connection connection = DriverManager.getConnection(jdbcUrl());
|
||||
PreparedStatement ps = connection.prepareStatement(sql)) {
|
||||
bindQuery(ps, query);
|
||||
try (ResultSet rs = ps.executeQuery()) {
|
||||
List<EventFileRecord> out = new ArrayList<>();
|
||||
while (rs.next()) {
|
||||
out.add(record(rs));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("query duckdb hot event store failed", e);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Add query binding helper**
|
||||
|
||||
```java
|
||||
private static void bindQuery(PreparedStatement ps, EventFileQuery query) throws SQLException {
|
||||
int index = 1;
|
||||
ps.setString(index++, query.protocol().name());
|
||||
ps.setString(index++, query.dateFrom().toString());
|
||||
ps.setString(index++, query.dateTo().toString());
|
||||
if (query.vin() != null) {
|
||||
ps.setString(index++, query.vin());
|
||||
}
|
||||
if (query.eventType() != null) {
|
||||
ps.setString(index++, query.eventType());
|
||||
}
|
||||
if (query.eventTimeFrom() != null) {
|
||||
ps.setLong(index++, query.eventTimeFrom().toEpochMilli());
|
||||
}
|
||||
if (query.eventTimeTo() != null) {
|
||||
ps.setLong(index++, query.eventTimeTo().toEpochMilli());
|
||||
}
|
||||
ps.setInt(index, query.limit());
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Implement `findByRawArchiveUri`**
|
||||
|
||||
```java
|
||||
@Override
|
||||
public EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
|
||||
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
ensureInitialized();
|
||||
try (Connection connection = DriverManager.getConnection(jdbcUrl());
|
||||
PreparedStatement ps = connection.prepareStatement("""
|
||||
SELECT event_id, protocol, event_type, vin, event_time_ms, ingest_time_ms,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM event_records
|
||||
WHERE raw_archive_uri = ?
|
||||
ORDER BY event_type = 'RAW_ARCHIVE' DESC, ingest_time_ms DESC, event_id DESC
|
||||
LIMIT 1
|
||||
""")) {
|
||||
ps.setString(1, rawArchiveUri);
|
||||
try (ResultSet rs = ps.executeQuery()) {
|
||||
return rs.next() ? record(rs) : null;
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("query duckdb hot store by raw archive uri failed", e);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Add record mapper and helpers**
|
||||
|
||||
```java
|
||||
private EventFileRecord record(ResultSet rs) throws SQLException, IOException {
|
||||
return new EventFileRecord(
|
||||
rs.getString("event_id"),
|
||||
protocol(rs.getString("protocol")),
|
||||
rs.getString("event_type"),
|
||||
rs.getString("vin"),
|
||||
Instant.ofEpochMilli(rs.getLong("event_time_ms")),
|
||||
Instant.ofEpochMilli(rs.getLong("ingest_time_ms")),
|
||||
rs.getString("raw_archive_uri"),
|
||||
readMetadata(rs.getString("metadata_json")),
|
||||
rs.getString("payload_json"));
|
||||
}
|
||||
|
||||
private Map<String, String> readMetadata(String json) throws IOException {
|
||||
if (json == null || json.isBlank()) {
|
||||
return Map.of();
|
||||
}
|
||||
return objectMapper.readValue(json, STRING_MAP);
|
||||
}
|
||||
|
||||
private static ProtocolId protocol(String value) {
|
||||
if (value == null || value.isBlank()) {
|
||||
return ProtocolId.UNKNOWN;
|
||||
}
|
||||
try {
|
||||
return ProtocolId.valueOf(value);
|
||||
} catch (IllegalArgumentException ex) {
|
||||
return ProtocolId.UNKNOWN;
|
||||
}
|
||||
}
|
||||
|
||||
private String jdbcUrl() {
|
||||
return "jdbc:duckdb:" + dbPath;
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Run hot store tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
|
||||
```
|
||||
|
||||
Expected: all `DuckDbHotEventFileStoreTest` tests pass.
|
||||
|
||||
## Task 4: Preserve Legacy Store Tests and Update Defaults
|
||||
|
||||
**Files:**
|
||||
- Modify: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java`
|
||||
- Modify: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
|
||||
- Modify: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
|
||||
|
||||
- [ ] **Step 1: Keep Parquet tests explicitly legacy**
|
||||
|
||||
No behavior change is needed in `DuckDbParquetEventFileStoreTest`; leave it instantiating `DuckDbParquetEventFileStore` directly. Add a class comment:
|
||||
|
||||
```java
|
||||
/**
|
||||
* Compatibility coverage for the legacy Parquet sidecar backend.
|
||||
* Production history uses DuckDbHotEventFileStore through auto-configuration.
|
||||
*/
|
||||
class DuckDbParquetEventFileStoreTest {
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Set vehicle-history-app storage default**
|
||||
|
||||
Add to `modules/apps/vehicle-history-app/src/main/resources/application.yml`:
|
||||
|
||||
```yaml
|
||||
event-file-store:
|
||||
enabled: ${EVENT_FILE_STORE_ENABLED:true}
|
||||
storage: ${EVENT_FILE_STORE_STORAGE:duckdb-hot}
|
||||
path: ${EVENT_FILE_STORE_PATH:./target/event-store/}
|
||||
zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai}
|
||||
batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:1000}
|
||||
flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000}
|
||||
```
|
||||
|
||||
Keep existing indentation and only add `storage`; if `batch-size` already exists, update it to `1000`.
|
||||
|
||||
- [ ] **Step 3: Update app default test**
|
||||
|
||||
In `VehicleHistoryAppDefaultsTest`, assert:
|
||||
|
||||
```java
|
||||
assertThat(context.getEnvironment()
|
||||
.getProperty("lingniu.ingest.event-file-store.storage"))
|
||||
.isEqualTo("duckdb-hot");
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Run app default tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :vehicle-history-app -Dtest=VehicleHistoryAppDefaultsTest test
|
||||
```
|
||||
|
||||
Expected: default configuration test passes.
|
||||
|
||||
## Task 5: Verify History Ingest Still Archives RAW Bytes
|
||||
|
||||
**Files:**
|
||||
- Modify: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java`
|
||||
- Test: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java`
|
||||
|
||||
- [ ] **Step 1: Add an integration-style test using the hot store**
|
||||
|
||||
Add a test that writes a RAW envelope through `EventHistoryEnvelopeIngestor` into `DuckDbHotEventFileStore`, then finds it by URI.
|
||||
|
||||
```java
|
||||
@Test
|
||||
void rawArchiveEnvelopeCanBeFoundFromDuckDbHotStoreByUri(@TempDir Path tempDir) throws Exception {
|
||||
EventFileStore store = new DuckDbHotEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), OBJECT_MAPPER);
|
||||
CapturingArchiveStore archive = new CapturingArchiveStore();
|
||||
EventHistoryEnvelopeIngestor ingestor =
|
||||
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper(), archive);
|
||||
String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-hot.bin";
|
||||
String rawArchiveUri = "archive://" + rawArchiveKey;
|
||||
byte[] rawBytes = new byte[]{0x23, 0x23, 0x02, 0x01};
|
||||
|
||||
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
|
||||
.setSchemaVersion("1.0")
|
||||
.setEventId("raw-event-hot")
|
||||
.setVin("VINRAW001")
|
||||
.setSource("GB32960")
|
||||
.setProtocolVersion("V2016")
|
||||
.setEventTimeMs(1_782_112_400_000L)
|
||||
.setIngestTimeMs(1_782_112_401_000L)
|
||||
.putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey)
|
||||
.putMetadata(RawArchiveKeys.META_URI, rawArchiveUri)
|
||||
.setRawArchive(RawArchiveRef.newBuilder()
|
||||
.setUri(rawArchiveUri)
|
||||
.setSizeBytes(rawBytes.length)
|
||||
.setData(ByteString.copyFrom(rawBytes))
|
||||
.build())
|
||||
.build();
|
||||
|
||||
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
|
||||
|
||||
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
|
||||
assertThat(store.findByRawArchiveUri(rawArchiveUri))
|
||||
.isNotNull()
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.isEqualTo("raw-event-hot");
|
||||
assertThat(archive.bytesByKey).containsEntry(rawArchiveKey, rawBytes);
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the event history ingestor test**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-history-service -Dtest=EventHistoryEnvelopeIngestorTest test
|
||||
```
|
||||
|
||||
Expected: test passes.
|
||||
|
||||
- [ ] **Step 3: Run decoded frame service tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-history-service -Dtest=Gb32960DecodedFrameServiceTest test
|
||||
```
|
||||
|
||||
Expected: existing replay/snapshot tests pass. If they use a fake store, no change is needed.
|
||||
|
||||
## Task 6: Full Module Verification
|
||||
|
||||
**Files:**
|
||||
- No source changes unless failures expose missing imports or config assertions.
|
||||
|
||||
- [ ] **Step 1: Run sink module tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-file-store test
|
||||
```
|
||||
|
||||
Expected: all event-file-store tests pass, including both hot and legacy stores.
|
||||
|
||||
- [ ] **Step 2: Run event-history-service tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :event-history-service test
|
||||
```
|
||||
|
||||
Expected: all event-history-service tests pass.
|
||||
|
||||
- [ ] **Step 3: Package vehicle-history-app**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
mvn -pl :vehicle-history-app -am package -DskipTests
|
||||
```
|
||||
|
||||
Expected: package succeeds.
|
||||
|
||||
## Task 7: Local Runtime Verification
|
||||
|
||||
**Files:**
|
||||
- No committed source changes.
|
||||
|
||||
- [ ] **Step 1: Stop any old history service on port 20200**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
lsof -tiTCP:20200 -sTCP:LISTEN | xargs -r kill
|
||||
```
|
||||
|
||||
Expected: no command output, or the old process exits.
|
||||
|
||||
- [ ] **Step 2: Start vehicle-history-app with hot DuckDB store**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
EVENT_FILE_STORE_STORAGE=duckdb-hot \
|
||||
EVENT_FILE_STORE_PATH=./target/live-history-event-store \
|
||||
SINK_ARCHIVE_PATH=./target/live-history-archive \
|
||||
KAFKA_BROKERS=114.55.58.251:9092 \
|
||||
KAFKA_CONSUMER_ENABLED=true \
|
||||
KAFKA_GROUP_HISTORY=vehicle-history-hot-$(date +%Y%m%d%H%M%S) \
|
||||
java --sun-misc-unsafe-memory-access=allow \
|
||||
-jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
|
||||
```
|
||||
|
||||
Expected:
|
||||
|
||||
- app starts on `http://127.0.0.1:20200`;
|
||||
- logs show Kafka consumer subscribed;
|
||||
- `target/live-history-event-store/events.duckdb` is created.
|
||||
|
||||
- [ ] **Step 3: Verify health**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
curl -sS http://127.0.0.1:20200/actuator/health
|
||||
```
|
||||
|
||||
Expected:
|
||||
|
||||
```json
|
||||
{"status":"UP"}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Verify a recent VIN query**
|
||||
|
||||
Run with a VIN observed in live archive:
|
||||
|
||||
```bash
|
||||
curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/telemetry-snapshots?vin=LNXNEGRR9SR318194&platformAccount=Hyundai&dateFrom=2026-06-23&dateTo=2026-06-24&order=DESC&limit=3'
|
||||
```
|
||||
|
||||
Expected:
|
||||
|
||||
- response is a JSON array;
|
||||
- if live traffic for that VIN exists after service start, at least one snapshot appears;
|
||||
- `rawArchiveUris` point to existing files under `target/live-history-archive`.
|
||||
|
||||
- [ ] **Step 5: Verify raw frame replay**
|
||||
|
||||
Use one `rawArchiveUri` from Step 4:
|
||||
|
||||
```bash
|
||||
curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/frame?rawArchiveUri=archive://REPLACE_ME&platformAccount=Hyundai'
|
||||
```
|
||||
|
||||
Expected:
|
||||
|
||||
- response contains `vin`, `command`, `eventTime`, and parsed `blocks`;
|
||||
- no `raw archive is missing` warning for the selected URI.
|
||||
|
||||
## Self-Review Checklist
|
||||
|
||||
- [ ] The new hot store does not rewrite Parquet files on every append.
|
||||
- [ ] `appendAll` writes one batch in one transaction.
|
||||
- [ ] Duplicate `event_id` replay is idempotent.
|
||||
- [ ] Existing history HTTP APIs still use `EventFileStore`, so controller contracts are unchanged.
|
||||
- [ ] RAW bytes still land in `ArchiveStore`.
|
||||
- [ ] `findByRawArchiveUri` is backed by DuckDB index.
|
||||
- [ ] No MySQL credential, RDS host, username, or password appears in the plan or source files.
|
||||
- [ ] This plan does not claim the full production goal is complete; it only lands the history foundation.
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -1,46 +0,0 @@
|
||||
# JT808 Kafka Streaming Mileage Plan
|
||||
|
||||
## Current Goal
|
||||
|
||||
Consume JT808 Kafka location events and write the derived daily mileage into the common vehicle-stat metric repository.
|
||||
|
||||
## Current Design
|
||||
|
||||
- Only supported message backbone: Kafka.
|
||||
- Source topic: `vehicle.event.jt808.v1`
|
||||
- Runtime app: `vehicle-analytics-app`
|
||||
- Runtime state: none outside `vehicle_stat_metric`
|
||||
- Metric output: `VehicleStatRepository.recordDailyMileageSample(...)`
|
||||
- Production metric storage: JDBC/MySQL `vehicle_stat_metric`
|
||||
- Date boundary: `Asia/Shanghai`
|
||||
- Calculation method: `JT808_TOTAL_MILEAGE_DIFF`
|
||||
|
||||
JT808 daily mileage is calculated only from the GPS total mileage reported by JT808 location additional information:
|
||||
|
||||
```text
|
||||
daily_mileage_km = max_total_mileage_km - min_total_mileage_km
|
||||
```
|
||||
|
||||
The first valid local-day sample writes `daily_mileage_km=0.0`. Later ordered or replayed samples update the same metric row with:
|
||||
|
||||
```text
|
||||
metric_value = max_total_mileage_km - min_total_mileage_km
|
||||
metric_key = daily_mileage_km
|
||||
```
|
||||
|
||||
The local-day minimum and maximum GPS total mileage values stay on that same metric row as calculation source columns so restarts recover from MySQL without Redis or another state table.
|
||||
|
||||
## Runtime Properties
|
||||
|
||||
```text
|
||||
KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1
|
||||
VEHICLE_STAT_ENABLED=true
|
||||
VEHICLE_STAT_JT808_MILEAGE_ENABLED=true
|
||||
MYSQL_JDBC_URL=<jdbc-url>
|
||||
MYSQL_USERNAME=<user>
|
||||
MYSQL_PASSWORD=<password>
|
||||
```
|
||||
|
||||
## Notes
|
||||
|
||||
Do not create or write a protocol-specific JT808 daily-mileage table. Do not add another message backbone, distance accumulation, integral calculation, Redis state, or memory state back into this path unless the mileage definition changes again.
|
||||
@@ -1,720 +0,0 @@
|
||||
# Go Vehicle Ingest Redesign Phase 1 Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Build the first production-capable Go runtime for GB32960, JT808, and Yutong MQTT ingestion with unified Kafka, TDengine, MySQL statistics, and Redis realtime state boundaries.
|
||||
|
||||
**Architecture:** Create a new single Go module at `go/vehicle-gateway` and migrate only the useful pieces from the existing `go/ingest-edge` and `go/vehicle-state` prototypes. The gateway produces unified envelopes to Kafka; independent consumers write TDengine history, MySQL daily metrics, and Redis realtime state.
|
||||
|
||||
**Tech Stack:** Go 1.26+, Kafka (`segmentio/kafka-go`), Redis (`redis/go-redis`), MySQL (`go-sql-driver/mysql`), TDengine official Go connector, MQTT (`eclipse/paho.mqtt.golang`), standard library TCP.
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
- Create: `go/vehicle-gateway/go.mod`
|
||||
- Create: `go/vehicle-gateway/cmd/gateway/main.go`
|
||||
- Create: `go/vehicle-gateway/cmd/history-writer/main.go`
|
||||
- Create: `go/vehicle-gateway/cmd/stat-writer/main.go`
|
||||
- Create: `go/vehicle-gateway/cmd/realtime-api/main.go`
|
||||
- Create: `go/vehicle-gateway/internal/envelope/envelope.go`
|
||||
- Create: `go/vehicle-gateway/internal/envelope/envelope_test.go`
|
||||
- Create: `go/vehicle-gateway/internal/protocol/jt808/*`
|
||||
- Create: `go/vehicle-gateway/internal/protocol/gb32960/*`
|
||||
- Create: `go/vehicle-gateway/internal/protocol/yutongmqtt/*`
|
||||
- Create: `go/vehicle-gateway/internal/gateway/*`
|
||||
- Create: `go/vehicle-gateway/internal/identity/*`
|
||||
- Create: `go/vehicle-gateway/internal/eventbus/*`
|
||||
- Create: `go/vehicle-gateway/internal/history/*`
|
||||
- Create: `go/vehicle-gateway/internal/stats/*`
|
||||
- Create: `go/vehicle-gateway/internal/realtime/*`
|
||||
- Create: `go/vehicle-gateway/internal/observability/*`
|
||||
- Modify: `README.md`
|
||||
- Modify: `docs/target-architecture.md`
|
||||
- Create: `deploy/portainer/docker-compose-go.yml`
|
||||
|
||||
The existing `go/ingest-edge` and `go/vehicle-state` directories are migration sources only. After phase 1 is verified, remove or mark them superseded.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Create Single Go Module
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/go.mod`
|
||||
- Create: `go/vehicle-gateway/internal/observability/logger.go`
|
||||
- Create: `go/vehicle-gateway/cmd/gateway/main.go`
|
||||
|
||||
- [ ] **Step 1: Create module manifest**
|
||||
|
||||
Add `go/vehicle-gateway/go.mod`:
|
||||
|
||||
```go
|
||||
module lingniu-vehicle-ingest/go/vehicle-gateway
|
||||
|
||||
go 1.26
|
||||
|
||||
require (
|
||||
github.com/eclipse/paho.mqtt.golang v1.5.1
|
||||
github.com/go-sql-driver/mysql v1.9.3
|
||||
github.com/redis/go-redis/v9 v9.17.2
|
||||
github.com/segmentio/kafka-go v0.4.49
|
||||
github.com/taosdata/driver-go/v3 v3.8.1
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Add logger helper**
|
||||
|
||||
Add `go/vehicle-gateway/internal/observability/logger.go`:
|
||||
|
||||
```go
|
||||
package observability
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"os"
|
||||
)
|
||||
|
||||
func NewLogger(service string) *slog.Logger {
|
||||
handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{AddSource: true})
|
||||
return slog.New(handler).With("service", service)
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Add temporary gateway entrypoint**
|
||||
|
||||
Add `go/vehicle-gateway/cmd/gateway/main.go`:
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
|
||||
|
||||
func main() {
|
||||
logger := observability.NewLogger("vehicle-gateway")
|
||||
logger.Info("vehicle gateway scaffold started")
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Verify scaffold builds**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go mod tidy
|
||||
go test ./...
|
||||
go build ./cmd/gateway
|
||||
```
|
||||
|
||||
Expected: all commands exit `0`.
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Define Unified Envelope
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/envelope/envelope.go`
|
||||
- Create: `go/vehicle-gateway/internal/envelope/envelope_test.go`
|
||||
|
||||
- [ ] **Step 1: Write envelope tests**
|
||||
|
||||
Add `go/vehicle-gateway/internal/envelope/envelope_test.go`:
|
||||
|
||||
```go
|
||||
package envelope
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestFrameEnvelopeVehicleKeyPrefersVIN(t *testing.T) {
|
||||
e := FrameEnvelope{Protocol: ProtocolJT808, VIN: "LNBVIN00000000001", Phone: "013307795425"}
|
||||
if got := e.VehicleKey(); got != "LNBVIN00000000001" {
|
||||
t.Fatalf("VehicleKey() = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFrameEnvelopeVehicleKeyFallsBackToPhone(t *testing.T) {
|
||||
e := FrameEnvelope{Protocol: ProtocolJT808, Phone: "013307795425"}
|
||||
if got := e.VehicleKey(); got != "JT808:013307795425" {
|
||||
t.Fatalf("VehicleKey() = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFrameEnvelopeEventIDStable(t *testing.T) {
|
||||
e := FrameEnvelope{
|
||||
Protocol: ProtocolJT808,
|
||||
MessageID: "0x0200",
|
||||
Phone: "013307795425",
|
||||
Sequence: 1,
|
||||
EventTimeMS: 1782745114000,
|
||||
ReceivedAtMS: 1782745114999,
|
||||
RawHex: "7e02000000ff7e",
|
||||
}
|
||||
a := e.StableEventID()
|
||||
b := e.StableEventID()
|
||||
if a == "" || a != b {
|
||||
t.Fatalf("event id must be non-empty and stable: %q %q", a, b)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run tests and confirm failure**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/envelope
|
||||
```
|
||||
|
||||
Expected: fail because `FrameEnvelope` is not defined.
|
||||
|
||||
- [ ] **Step 3: Implement envelope**
|
||||
|
||||
Add `go/vehicle-gateway/internal/envelope/envelope.go`:
|
||||
|
||||
```go
|
||||
package envelope
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type Protocol string
|
||||
|
||||
const (
|
||||
ProtocolGB32960 Protocol = "GB32960"
|
||||
ProtocolJT808 Protocol = "JT808"
|
||||
ProtocolYutongMQTT Protocol = "YUTONG_MQTT"
|
||||
)
|
||||
|
||||
type ParseStatus string
|
||||
|
||||
const (
|
||||
ParseOK ParseStatus = "OK"
|
||||
ParsePartial ParseStatus = "PARTIAL"
|
||||
ParseBadFrame ParseStatus = "BAD_FRAME"
|
||||
)
|
||||
|
||||
type FrameEnvelope struct {
|
||||
EventID string `json:"event_id"`
|
||||
TraceID string `json:"trace_id"`
|
||||
Protocol Protocol `json:"protocol"`
|
||||
MessageID string `json:"message_id"`
|
||||
Sequence uint16 `json:"sequence"`
|
||||
VIN string `json:"vin,omitempty"`
|
||||
VehicleKeyHint string `json:"vehicle_key,omitempty"`
|
||||
Phone string `json:"phone,omitempty"`
|
||||
DeviceID string `json:"device_id,omitempty"`
|
||||
Plate string `json:"plate,omitempty"`
|
||||
SourceEndpoint string `json:"source_endpoint,omitempty"`
|
||||
EventTimeMS int64 `json:"event_time_ms"`
|
||||
ReceivedAtMS int64 `json:"received_at_ms"`
|
||||
RawHex string `json:"raw_hex,omitempty"`
|
||||
RawText string `json:"raw_text,omitempty"`
|
||||
Parsed map[string]any `json:"parsed,omitempty"`
|
||||
Fields map[string]any `json:"fields,omitempty"`
|
||||
ParseStatus ParseStatus `json:"parse_status"`
|
||||
ParseError string `json:"parse_error,omitempty"`
|
||||
}
|
||||
|
||||
func (e FrameEnvelope) VehicleKey() string {
|
||||
if key := strings.TrimSpace(e.VIN); key != "" {
|
||||
return key
|
||||
}
|
||||
if key := strings.TrimSpace(e.VehicleKeyHint); key != "" {
|
||||
return key
|
||||
}
|
||||
if key := strings.TrimSpace(e.Phone); key != "" {
|
||||
return string(e.Protocol) + ":" + key
|
||||
}
|
||||
if key := strings.TrimSpace(e.DeviceID); key != "" {
|
||||
return string(e.Protocol) + ":" + key
|
||||
}
|
||||
return string(e.Protocol) + ":unknown"
|
||||
}
|
||||
|
||||
func (e FrameEnvelope) StableEventID() string {
|
||||
if strings.TrimSpace(e.EventID) != "" {
|
||||
return e.EventID
|
||||
}
|
||||
input := fmt.Sprintf("%s|%s|%s|%d|%d|%s",
|
||||
e.Protocol, e.MessageID, e.VehicleKey(), e.Sequence, e.EventTimeMS, e.RawHex)
|
||||
sum := sha256.Sum256([]byte(input))
|
||||
return hex.EncodeToString(sum[:16])
|
||||
}
|
||||
|
||||
func (e FrameEnvelope) MarshalJSONBytes() ([]byte, error) {
|
||||
if e.EventID == "" {
|
||||
e.EventID = e.StableEventID()
|
||||
}
|
||||
if e.ParseStatus == "" {
|
||||
e.ParseStatus = ParseOK
|
||||
}
|
||||
return json.Marshal(e)
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Verify tests pass**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/envelope
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Implement JT808 Frame and 0200 Parser
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/protocol/jt808/parser.go`
|
||||
- Create: `go/vehicle-gateway/internal/protocol/jt808/parser_test.go`
|
||||
|
||||
- [ ] **Step 1: Add sample frame tests**
|
||||
|
||||
Use the production sample provided in the thread:
|
||||
|
||||
```text
|
||||
7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E
|
||||
```
|
||||
|
||||
Expected assertions:
|
||||
|
||||
- message id is `0x0200`
|
||||
- phone keeps normalized BCD string `013307795425`
|
||||
- sequence is `1`
|
||||
- `fields.total_mileage_km` is parsed from additional item `0x01`
|
||||
- latitude, longitude, speed, direction, alarm, status, and device time are present
|
||||
|
||||
- [ ] **Step 2: Implement parser**
|
||||
|
||||
Implementation rules:
|
||||
|
||||
- strip `0x7e` start/end delimiters
|
||||
- unescape `0x7d 0x02 -> 0x7e`
|
||||
- unescape `0x7d 0x01 -> 0x7d`
|
||||
- verify XOR checksum
|
||||
- parse 2011/2013 common header
|
||||
- parse BCD phone from 6 bytes and keep both raw and normalized values in `parsed.header`
|
||||
- parse 0200 fixed body
|
||||
- parse additional item list as `id,length,value_hex`
|
||||
- parse additional `0x01` as `total_mileage_km = uint32 / 10`
|
||||
|
||||
- [ ] **Step 3: Verify**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/protocol/jt808
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Implement GB32960 Frame and Data Unit Parser
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/protocol/gb32960/parser.go`
|
||||
- Create: `go/vehicle-gateway/internal/protocol/gb32960/parser_test.go`
|
||||
|
||||
- [ ] **Step 1: Add parser tests**
|
||||
|
||||
Tests must cover:
|
||||
|
||||
- `##` frame boundary
|
||||
- command id
|
||||
- response flag
|
||||
- 17-byte VIN
|
||||
- encryption flag
|
||||
- payload length
|
||||
- BCC verification
|
||||
- realtime data command `0x02`
|
||||
- reissue data command `0x03`
|
||||
- data unit `0x01` vehicle status fields
|
||||
- data unit `0x05` position fields
|
||||
|
||||
- [ ] **Step 2: Implement parser**
|
||||
|
||||
Implementation rules:
|
||||
|
||||
- parse header without allocating large temporary buffers
|
||||
- keep RAW as hex in envelope
|
||||
- write all recognized data units to `Parsed`
|
||||
- write core fields to `Fields`
|
||||
- unsupported data unit stays in `Parsed["unknown_units"]`
|
||||
|
||||
- [ ] **Step 3: Verify**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/protocol/gb32960
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Implement Kafka Sink
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/eventbus/kafka_sink.go`
|
||||
- Create: `go/vehicle-gateway/internal/eventbus/kafka_sink_test.go`
|
||||
|
||||
- [ ] **Step 1: Add sink interface**
|
||||
|
||||
Create a sink interface:
|
||||
|
||||
```go
|
||||
package eventbus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
||||
)
|
||||
|
||||
type Sink interface {
|
||||
PublishRaw(context.Context, envelope.FrameEnvelope) error
|
||||
PublishUnified(context.Context, envelope.FrameEnvelope) error
|
||||
Close() error
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Implement Kafka topic routing**
|
||||
|
||||
Rules:
|
||||
|
||||
- `GB32960 -> vehicle.raw.gb32960.v1`
|
||||
- `JT808 -> vehicle.raw.jt808.v1`
|
||||
- `YUTONG_MQTT -> vehicle.raw.yutong-mqtt.v1`
|
||||
- unified topic is `vehicle.event.unified.v1`
|
||||
- message key is `env.VehicleKey()`
|
||||
|
||||
- [ ] **Step 3: Verify routing with unit tests**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/eventbus
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 6: Implement TDengine History Writer
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/history/schema.go`
|
||||
- Create: `go/vehicle-gateway/internal/history/writer.go`
|
||||
- Create: `go/vehicle-gateway/internal/history/writer_test.go`
|
||||
- Create: `go/vehicle-gateway/cmd/history-writer/main.go`
|
||||
|
||||
- [ ] **Step 1: Add schema bootstrap SQL**
|
||||
|
||||
Implement schema strings for:
|
||||
|
||||
- database `lingniu_vehicle_ts`
|
||||
- stable `raw_frames`
|
||||
- stable `vehicle_locations`
|
||||
- stable `vehicle_mileage_points`
|
||||
|
||||
- [ ] **Step 2: Implement writer**
|
||||
|
||||
Rules:
|
||||
|
||||
- `AppendRawFrame` always writes one raw row.
|
||||
- `AppendLocation` writes only when longitude and latitude exist.
|
||||
- `AppendMileagePoint` writes only when `total_mileage_km` exists.
|
||||
- child table name is deterministic hash of `protocol + vehicle_key`.
|
||||
- escape tag values.
|
||||
|
||||
- [ ] **Step 3: Use TDengine official driver**
|
||||
|
||||
Import WebSocket driver in command:
|
||||
|
||||
```go
|
||||
import _ "github.com/taosdata/driver-go/v3/taosWS"
|
||||
```
|
||||
|
||||
Default driver name:
|
||||
|
||||
```text
|
||||
TDENGINE_DRIVER=taosWS
|
||||
```
|
||||
|
||||
Short-term compatibility:
|
||||
|
||||
```text
|
||||
TDENGINE_DRIVER=taosSql
|
||||
```
|
||||
|
||||
Only use `taosSql` when ECS TDengine WebSocket is not available.
|
||||
|
||||
- [ ] **Step 4: Verify**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/history
|
||||
go build ./cmd/history-writer
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 7: Implement MySQL Daily Metric Writer
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/stats/schema.go`
|
||||
- Create: `go/vehicle-gateway/internal/stats/daily_metric.go`
|
||||
- Create: `go/vehicle-gateway/internal/stats/daily_metric_test.go`
|
||||
- Create: `go/vehicle-gateway/cmd/stat-writer/main.go`
|
||||
|
||||
- [ ] **Step 1: Add schema bootstrap**
|
||||
|
||||
Implement `vehicle_daily_metric` schema from the design spec.
|
||||
|
||||
- [ ] **Step 2: Add metric derivation tests**
|
||||
|
||||
Test cases:
|
||||
|
||||
- no `total_mileage_km` produces no metric
|
||||
- one sample produces:
|
||||
- `daily_mileage_km = 0`
|
||||
- `daily_total_mileage_km = sample`
|
||||
- later larger sample updates:
|
||||
- `latest_total_mileage_km`
|
||||
- `daily_mileage_km`
|
||||
- `daily_total_mileage_km`
|
||||
- out-of-order smaller sample updates:
|
||||
- `first_total_mileage_km`
|
||||
- `daily_mileage_km`
|
||||
|
||||
- [ ] **Step 3: Implement MySQL upsert**
|
||||
|
||||
Use one table and one idempotent upsert:
|
||||
|
||||
```sql
|
||||
INSERT INTO vehicle_daily_metric
|
||||
(vin, stat_date, protocol, metric_key, metric_value, metric_unit,
|
||||
first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method)
|
||||
VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF')
|
||||
ON DUPLICATE KEY UPDATE
|
||||
first_total_mileage_km = LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)),
|
||||
latest_total_mileage_km = GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)),
|
||||
metric_value = CASE
|
||||
WHEN metric_key = 'daily_mileage_km'
|
||||
THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
|
||||
- LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
|
||||
WHEN metric_key = 'daily_total_mileage_km'
|
||||
THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
|
||||
ELSE VALUES(metric_value)
|
||||
END,
|
||||
sample_count = sample_count + 1,
|
||||
updated_at = CURRENT_TIMESTAMP
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Verify**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/stats
|
||||
go build ./cmd/stat-writer
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 8: Implement Redis Realtime API
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/realtime/repository.go`
|
||||
- Create: `go/vehicle-gateway/internal/realtime/repository_test.go`
|
||||
- Create: `go/vehicle-gateway/internal/realtime/http.go`
|
||||
- Create: `go/vehicle-gateway/cmd/realtime-api/main.go`
|
||||
|
||||
- [ ] **Step 1: Add repository contract**
|
||||
|
||||
Methods:
|
||||
|
||||
- `Update(ctx, envelope.FrameEnvelope) error`
|
||||
- `GetMerged(ctx, vin string) (Snapshot, error)`
|
||||
- `GetProtocol(ctx, vin string, protocol envelope.Protocol) (Snapshot, error)`
|
||||
- `IsOnline(ctx, vin string) (OnlineStatus, error)`
|
||||
|
||||
- [ ] **Step 2: Implement merge logic**
|
||||
|
||||
Rules:
|
||||
|
||||
- update `vehicle:latest:{vin}:{protocol}`
|
||||
- update `vehicle:latest:{vin}` with newest fields
|
||||
- update `vehicle:online:{vin}`
|
||||
- update sorted set `vehicle:last_seen`
|
||||
|
||||
- [ ] **Step 3: Implement HTTP API**
|
||||
|
||||
Routes:
|
||||
|
||||
- `GET /api/realtime/vehicles/{vin}`
|
||||
- `GET /api/realtime/vehicles/{vin}/online`
|
||||
- `GET /api/realtime/vehicles/{vin}/protocols/{protocol}`
|
||||
|
||||
- [ ] **Step 4: Verify**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
go test ./internal/realtime
|
||||
go build ./cmd/realtime-api
|
||||
```
|
||||
|
||||
Expected: pass.
|
||||
|
||||
---
|
||||
|
||||
### Task 9: Wire Gateway Runtime
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/internal/gateway/tcp_server.go`
|
||||
- Create: `go/vehicle-gateway/internal/gateway/mqtt_client.go`
|
||||
- Modify: `go/vehicle-gateway/cmd/gateway/main.go`
|
||||
|
||||
- [ ] **Step 1: Implement TCP server**
|
||||
|
||||
Rules:
|
||||
|
||||
- one goroutine per accepted connection
|
||||
- bounded max connections
|
||||
- read timeout and idle timeout
|
||||
- protocol-specific frame extractor
|
||||
- structured peer endpoint
|
||||
- graceful shutdown on SIGTERM
|
||||
|
||||
- [ ] **Step 2: Implement MQTT client**
|
||||
|
||||
Rules:
|
||||
|
||||
- connect with official production config from environment
|
||||
- subscribe configured topic list
|
||||
- convert each message into envelope
|
||||
- publish raw and unified events
|
||||
- reconnect with backoff
|
||||
|
||||
- [ ] **Step 3: Verify local JSON mode**
|
||||
|
||||
Run without Kafka:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
GB32960_TCP_ADDR=:132960 JT808_TCP_ADDR=:18080 go run ./cmd/gateway
|
||||
```
|
||||
|
||||
Expected: service starts and logs configured listeners.
|
||||
|
||||
---
|
||||
|
||||
### Task 10: Docker and ECS Deployment
|
||||
|
||||
**Files:**
|
||||
- Create: `go/vehicle-gateway/Dockerfile`
|
||||
- Create: `deploy/portainer/docker-compose-go.yml`
|
||||
- Modify: `docs/operations/current-ecs-deployment.md`
|
||||
|
||||
- [ ] **Step 1: Add multi-stage Dockerfile**
|
||||
|
||||
Build all commands:
|
||||
|
||||
- `gateway`
|
||||
- `history-writer`
|
||||
- `stat-writer`
|
||||
- `realtime-api`
|
||||
|
||||
- [ ] **Step 2: Add Portainer compose**
|
||||
|
||||
Services:
|
||||
|
||||
- `go-vehicle-gateway`
|
||||
- `go-history-writer`
|
||||
- `go-stat-writer`
|
||||
- `go-realtime-api`
|
||||
|
||||
Each service must include:
|
||||
|
||||
- restart policy
|
||||
- memory limit
|
||||
- Kafka env
|
||||
- MySQL/TDengine/Redis env as needed
|
||||
- logging options
|
||||
|
||||
- [ ] **Step 3: Verify Linux build**
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
cd go/vehicle-gateway
|
||||
GOOS=linux GOARCH=amd64 go build ./cmd/gateway
|
||||
GOOS=linux GOARCH=amd64 go build ./cmd/history-writer
|
||||
GOOS=linux GOARCH=amd64 go build ./cmd/stat-writer
|
||||
GOOS=linux GOARCH=amd64 go build ./cmd/realtime-api
|
||||
```
|
||||
|
||||
Expected: all commands exit `0`.
|
||||
|
||||
---
|
||||
|
||||
### Task 11: Production Verification
|
||||
|
||||
**Files:**
|
||||
- Create: `docs/operations/go-vehicle-gateway-verification.md`
|
||||
|
||||
- [ ] **Step 1: Record test commands**
|
||||
|
||||
Document commands to verify:
|
||||
|
||||
- gateway process health
|
||||
- Kafka topic consumption
|
||||
- TDengine row counts
|
||||
- MySQL daily metric rows
|
||||
- Redis realtime lookup
|
||||
|
||||
- [ ] **Step 2: Validate real traffic**
|
||||
|
||||
Evidence required:
|
||||
|
||||
- one real 32960 VIN with RAW, location, mileage point, daily metric, Redis snapshot
|
||||
- one real JT808 phone/VIN with RAW, location, mileage point, daily metric, Redis snapshot
|
||||
- one real Yutong MQTT VIN with RAW and Redis snapshot
|
||||
|
||||
- [ ] **Step 3: Keep old Java services until evidence is captured**
|
||||
|
||||
Only disable Java equivalents after the evidence file contains successful command outputs and timestamps.
|
||||
|
||||
---
|
||||
|
||||
## Self-Review Checklist
|
||||
|
||||
- The plan creates one new Go module instead of extending scattered prototypes.
|
||||
- The plan covers GB32960, JT808, and Yutong MQTT ingress.
|
||||
- The plan covers Kafka, TDengine, MySQL, and Redis.
|
||||
- The plan includes 32960 and 808 daily mileage and daily total mileage.
|
||||
- The plan includes local and ECS verification.
|
||||
- The plan does not restore Xinda Push.
|
||||
- The plan keeps Java services untouched until Go evidence exists.
|
||||
Reference in New Issue
Block a user