rows) {
- }
- }
}
diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java
index 26bdaf88..d6ee000e 100644
--- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java
+++ b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java
@@ -144,8 +144,8 @@ public sealed interface VehicleEvent
* 原始报文归档事件：每条成功解码的入站帧由 Dispatcher 产出一条，携带原始字节和归档索引信息。
* 32960 历史全字段查询依赖它对应的 {@code archive://...} 文件能够被回读。
*
- * 当前 {@code KafkaEventSink} 默认不发送本类型，{@code EventFileStoreSink} 只保存索引不保存
- * bytes；因此启用新 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。
+ *
当前 {@code KafkaEventSink} 会发送本类型的索引信息，RAW bytes 由 archive 落盘保存；
+ * 因此启用 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。
*
* @param command 协议主命令码（如 32960 0x02/0x03 等），冗余在事件里便于按命令分片归档
* @param infoType 协议子类型（可为 0），同上
diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java
deleted file mode 100644
index 71233eb2..00000000
--- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java
+++ /dev/null
@@ -1,93 +0,0 @@
-package com.lingniu.ingest.api.history;
-
-import com.lingniu.ingest.api.ProtocolId;
-
-import java.time.Instant;
-import java.time.LocalDate;
-
-/**
- * Query criteria for the optional file-backed historical event index.
- */
-public record EventFileQuery(
- ProtocolId protocol,
- LocalDate dateFrom,
- LocalDate dateTo,
- Instant eventTimeFrom,
- Instant eventTimeTo,
- Order order,
- int limit,
- String vin,
- String eventType,
- Instant cursorEventTime,
- Instant cursorIngestTime,
- String cursorEventId
-) {
- public enum Order {
- ASC,
- DESC
- }
-
- public EventFileQuery {
- if (protocol == null) {
- throw new IllegalArgumentException("protocol must not be null");
- }
- if (dateFrom == null || dateTo == null) {
- throw new IllegalArgumentException("date range must not be null");
- }
- if (dateTo.isBefore(dateFrom)) {
- throw new IllegalArgumentException("dateTo must not be before dateFrom");
- }
- if (eventTimeFrom != null && eventTimeTo != null && eventTimeTo.isBefore(eventTimeFrom)) {
- throw new IllegalArgumentException("eventTimeTo must not be before eventTimeFrom");
- }
- order = order == null ? Order.ASC : order;
- limit = limit <= 0 ? 100 : limit;
- vin = vin == null || vin.isBlank() ? null : vin.trim();
- eventType = eventType == null || eventType.isBlank() ? null : eventType.trim();
- cursorEventId = cursorEventId == null || cursorEventId.isBlank() ? null : cursorEventId.trim();
- if ((cursorEventTime == null || cursorIngestTime == null || cursorEventId == null)
- && !(cursorEventTime == null && cursorIngestTime == null && cursorEventId == null)) {
- throw new IllegalArgumentException("cursor eventTime, ingestTime and eventId must be provided together");
- }
- }
-
- public EventFileQuery(ProtocolId protocol,
- LocalDate dateFrom,
- LocalDate dateTo,
- Instant eventTimeFrom,
- Instant eventTimeTo,
- Order order,
- int limit,
- String vin,
- String eventType) {
- this(protocol, dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, limit,
- vin, eventType, null, null, null);
- }
-
- public EventFileQuery(ProtocolId protocol,
- LocalDate dateFrom,
- LocalDate dateTo,
- Order order,
- int limit,
- String vin,
- String eventType) {
- this(protocol, dateFrom, dateTo, null, null, order, limit, vin, eventType);
- }
-
- public EventFileQuery(ProtocolId protocol,
- LocalDate dateFrom,
- LocalDate dateTo,
- Order order,
- int limit) {
- this(protocol, dateFrom, dateTo, order, limit, null, null);
- }
-
- public EventFileQuery(ProtocolId protocol,
- LocalDate dateFrom,
- LocalDate dateTo,
- Order order,
- int limit,
- String vin) {
- this(protocol, dateFrom, dateTo, order, limit, vin, null);
- }
-}
diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java
deleted file mode 100644
index 1f93cf64..00000000
--- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java
+++ /dev/null
@@ -1,41 +0,0 @@
-package com.lingniu.ingest.api.history;
-
-import com.lingniu.ingest.api.ProtocolId;
-
-import java.time.Instant;
-import java.util.Map;
-
-/**
- * Minimal historical event record shared by optional compatibility stores and history queries.
- */
-public record EventFileRecord(
- String eventId,
- ProtocolId protocol,
- String eventType,
- String vin,
- Instant eventTime,
- Instant ingestTime,
- String rawArchiveUri,
- Map metadata,
- String payloadJson
-) {
- public EventFileRecord {
- if (eventId == null || eventId.isBlank()) {
- throw new IllegalArgumentException("eventId must not be blank");
- }
- if (protocol == null) {
- throw new IllegalArgumentException("protocol must not be null");
- }
- if (eventTime == null) {
- throw new IllegalArgumentException("eventTime must not be null");
- }
- if (ingestTime == null) {
- throw new IllegalArgumentException("ingestTime must not be null");
- }
- eventType = eventType == null ? "" : eventType;
- vin = vin == null ? "" : vin;
- rawArchiveUri = rawArchiveUri == null ? "" : rawArchiveUri;
- metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
- payloadJson = payloadJson == null || payloadJson.isBlank() ? "{}" : payloadJson;
- }
-}
diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java
deleted file mode 100644
index 4864d1c8..00000000
--- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package com.lingniu.ingest.api.history;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- * Historical event index contract.
- *
- * The default production history path uses TDengine directly. This contract remains in core so
- * optional compatibility implementations can still plug into query code without pulling their
- * storage engines into the default reactor.
- */
-public interface EventFileStore {
-
- default void append(EventFileRecord record) throws IOException {
- appendAll(List.of(record));
- }
-
- void appendAll(List records) throws IOException;
-
- List query(EventFileQuery query) throws IOException;
-
- default EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
- return null;
- }
-}
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java
deleted file mode 100644
index 6ff8f016..00000000
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java
+++ /dev/null
@@ -1,122 +0,0 @@
-package com.lingniu.ingest.eventhistory;
-
-import com.lingniu.ingest.api.ProtocolId;
-import com.lingniu.ingest.api.history.EventFileQuery;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
-
-import java.io.IOException;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.OffsetDateTime;
-import java.time.ZoneId;
-import java.util.List;
-import java.util.Map;
-
-/**
- * 通用历史事件查询接口。
- *
- * 这是跨协议的低层记录查询，返回的是 EventFileStore 中的 envelope/snapshot 记录。
- * GB32960 业务查询应优先使用 {@link Gb32960FrameController}，因为它会回读 RAW 并按协议展开全部字段。
- */
-@RestController
-@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
-@ConditionalOnBean(EventFileStore.class)
-@RequestMapping("/api/event-history")
-public class EventHistoryController {
-
- private static final ZoneId DEFAULT_ZONE = ZoneId.of("Asia/Shanghai");
-
- private final EventFileStore store;
-
- public EventHistoryController(EventFileStore store) {
- this.store = store;
- }
-
- @GetMapping("/records")
- public List query(
- @RequestParam ProtocolId protocol,
- @RequestParam String dateFrom,
- @RequestParam String dateTo,
- @RequestParam(defaultValue = "ASC") EventFileQuery.Order order,
- @RequestParam(defaultValue = "100") int limit,
- @RequestParam(required = false) String vin,
- @RequestParam(required = false) String eventType,
- @RequestParam(required = false) String cursorEventTime,
- @RequestParam(required = false) String cursorIngestTime,
- @RequestParam(required = false) String cursorEventId) throws IOException {
- return queryRecords(protocol, dateFrom, dateTo, order, limit, vin, eventType,
- cursorEventTime, cursorIngestTime, cursorEventId).stream()
- .map(RecordResponse::from)
- .toList();
- }
-
- private List queryRecords(ProtocolId protocol,
- String dateFrom,
- String dateTo,
- EventFileQuery.Order order,
- int limit,
- String vin,
- String eventType,
- String cursorEventTime,
- String cursorIngestTime,
- String cursorEventId) throws IOException {
- QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
- return store.query(new EventFileQuery(
- protocol,
- range.dateFrom(),
- range.dateTo(),
- range.eventTimeFrom(),
- range.eventTimeTo(),
- order,
- limit,
- vin,
- eventType,
- instantParam(cursorEventTime),
- instantParam(cursorIngestTime),
- cursorEventId));
- }
-
- private static Instant instantParam(String raw) {
- if (raw == null || raw.isBlank()) {
- return null;
- }
- String value = raw.trim();
- try {
- return OffsetDateTime.parse(value).toInstant();
- } catch (RuntimeException ignored) {
- return LocalDateTime.parse(value).atZone(DEFAULT_ZONE).toInstant();
- }
- }
-
- public record RecordResponse(
- String eventId,
- ProtocolId protocol,
- String eventType,
- String vin,
- String eventTime,
- String ingestTime,
- String rawArchiveUri,
- Map metadata,
- String payloadJson) {
-
- private static RecordResponse from(EventFileRecord record) {
- return new RecordResponse(
- record.eventId(),
- record.protocol(),
- record.eventType(),
- record.vin(),
- record.eventTime().toString(),
- record.ingestTime().toString(),
- record.rawArchiveUri(),
- record.metadata(),
- record.payloadJson());
- }
- }
-}
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java
index 00ac451f..c3ca06d8 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java
@@ -3,8 +3,6 @@ package com.lingniu.ingest.eventhistory;
import com.google.protobuf.InvalidProtocolBufferException;
import com.lingniu.ingest.api.consumer.EnvelopeBatchIngestor;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import com.lingniu.ingest.tdenginehistory.TdengineEnvelopeRows;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
@@ -16,8 +14,7 @@ import java.util.ArrayList;
import java.util.List;
/**
- * Consumer-side entry point that stores one Kafka envelope value into the
- * historical detail file store.
+ * Consumer-side entry point that stores Kafka envelope values into TDengine history tables.
*
* 当前 32960 本机运行配置没有启用 Kafka consumer；这个类保留给“其他服务把 envelope
* 写回历史库”的解耦部署方式。失败时返回结构化结果，由 Kafka worker 决定是否进 DLQ。
@@ -26,62 +23,24 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
private static final Logger log = LoggerFactory.getLogger(EventHistoryEnvelopeIngestor.class);
- private final EventFileStore store;
- private final TelemetryEnvelopeRecordMapper mapper;
private final TdengineHistoryWriter tdengineWriter;
private final boolean telemetryFieldsEnabled;
- public EventHistoryEnvelopeIngestor(EventFileStore store, TelemetryEnvelopeRecordMapper mapper) {
- this(store, mapper, null);
- }
-
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter) {
this(tdengineWriter, false);
}
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter,
boolean telemetryFieldsEnabled) {
- this(null, null, tdengineWriter, false, telemetryFieldsEnabled);
- }
-
- public EventHistoryEnvelopeIngestor(EventFileStore store,
- TelemetryEnvelopeRecordMapper mapper,
- TdengineHistoryWriter tdengineWriter) {
- this(store, mapper, tdengineWriter, true, false);
- }
-
- public EventHistoryEnvelopeIngestor(EventFileStore store,
- TelemetryEnvelopeRecordMapper mapper,
- TdengineHistoryWriter tdengineWriter,
- boolean telemetryFieldsEnabled) {
- this(store, mapper, tdengineWriter, true, telemetryFieldsEnabled);
- }
-
- private EventHistoryEnvelopeIngestor(EventFileStore store,
- TelemetryEnvelopeRecordMapper mapper,
- TdengineHistoryWriter tdengineWriter,
- boolean requireStore,
- boolean telemetryFieldsEnabled) {
- if (store == null) {
- if (requireStore) {
- throw new IllegalArgumentException("store must not be null");
- }
+ if (tdengineWriter == null) {
+ throw new IllegalArgumentException("tdengineWriter must not be null");
}
- if (store != null && mapper == null) {
- throw new IllegalArgumentException("mapper must not be null");
- }
- this.store = store;
- this.mapper = mapper;
this.tdengineWriter = tdengineWriter;
this.telemetryFieldsEnabled = telemetryFieldsEnabled;
}
public void ingest(byte[] kafkaValue) throws IOException {
VehicleEnvelope envelope = parse(kafkaValue);
- if (store != null) {
- EventFileRecord record = mapper.toRecord(envelope);
- store.append(record);
- }
writeTdengineFacts(List.of(envelope));
}
@@ -101,9 +60,8 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
VehicleEnvelope envelope = null;
try {
envelope = parse(kafkaValue);
- EventFileRecord record = store == null ? null : mapper.toRecord(envelope);
results.add(null);
- valid.add(new BatchEntry(results.size() - 1, envelope, record));
+ valid.add(new BatchEntry(results.size() - 1, envelope));
} catch (IllegalArgumentException ex) {
results.add(envelope == null
? EnvelopeIngestResult.invalid(ex.getMessage())
@@ -114,9 +72,6 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
return List.copyOf(results);
}
try {
- if (store != null) {
- store.appendAll(valid.stream().map(BatchEntry::record).toList());
- }
writeTdengineFacts(valid.stream().map(BatchEntry::envelope).toList());
for (BatchEntry entry : valid) {
results.set(entry.index(), EnvelopeIngestResult.stored(
@@ -124,7 +79,7 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
entry.vin()));
}
} catch (IOException ex) {
- log.error("event history TDengine/file-store ingest failed batchSize={} validSize={} firstEventId={} firstVin={}",
+ log.error("event history TDengine ingest failed batchSize={} validSize={} firstEventId={} firstVin={}",
kafkaValues.size(),
valid.size(),
valid.getFirst().envelope().getEventId(),
@@ -140,13 +95,13 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
return List.copyOf(results);
}
- private record BatchEntry(int index, VehicleEnvelope envelope, EventFileRecord record) {
+ private record BatchEntry(int index, VehicleEnvelope envelope) {
private String eventId() {
- return record == null ? envelope.getEventId() : record.eventId();
+ return envelope.getEventId();
}
private String vin() {
- return record == null ? envelope.getVin() : record.vin();
+ return envelope.getVin();
}
}
@@ -162,9 +117,6 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
}
private void writeTdengineFacts(List envelopes) throws IOException {
- if (tdengineWriter == null) {
- return;
- }
var rawFrames = envelopes.stream()
.flatMap(envelope -> TdengineEnvelopeRows.rawFrame(envelope).stream())
.toList();
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java
index 77370116..562f21db 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java
@@ -2,10 +2,6 @@ package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.lingniu.ingest.api.ProtocolId;
-import com.lingniu.ingest.api.history.EventFileQuery;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.model.Gb32960Message;
import com.lingniu.ingest.protocol.gb32960.model.InfoBlock;
@@ -38,8 +34,8 @@ import java.util.Map;
/**
* GB32960 RAW 历史帧查询和业务快照组装服务。
*
- * 历史库里保存的是 RAW_ARCHIVE 索引，不保存完整解码后的宽表。查询时先从
- * {@link EventFileStore} 找到 archive:// URI，再读取本地 RAW .bin，用当前协议解析器即时解码。
+ *
TDengine raw_frames 保存 archive:// URI。查询时先从 TDengine 找到 RAW 索引，
+ * 再读取本地 RAW .bin，用当前协议解析器即时解码。
* 这种设计让写入路径保持轻量，也允许后续修复解析器后直接重放历史 RAW 得到新字段。
*
*
snapshot 的聚合单位是 {@code vin + eventTime}。同一时刻可能有多个 32960 子包，
@@ -50,37 +46,20 @@ public final class Gb32960DecodedFrameService {
private static final Logger log = LoggerFactory.getLogger(Gb32960DecodedFrameService.class);
private static final TypeReference> MAP_TYPE = new TypeReference<>() {};
private static final int OUTPUT_DOUBLE_SCALE = 6;
- private static final String RAW_ARCHIVE_EVENT_TYPE = "RAW_ARCHIVE";
private static final ZoneId DEFAULT_ZONE = ZoneId.of("Asia/Shanghai");
- private final EventFileStore store;
private final TdengineHistoryReader tdengineReader;
private final Gb32960MessageDecoder decoder;
private final Path archiveRoot;
private final ObjectMapper objectMapper;
- public Gb32960DecodedFrameService(EventFileStore store,
- Gb32960MessageDecoder decoder,
- Path archiveRoot,
- ObjectMapper objectMapper) {
- this(store, null, decoder, archiveRoot, objectMapper);
- }
-
public Gb32960DecodedFrameService(TdengineHistoryReader tdengineReader,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
- this(null, tdengineReader, decoder, archiveRoot, objectMapper);
- }
-
- public Gb32960DecodedFrameService(EventFileStore store,
- TdengineHistoryReader tdengineReader,
- Gb32960MessageDecoder decoder,
- Path archiveRoot,
- ObjectMapper objectMapper) {
- if (store == null && tdengineReader == null) {
+ if (tdengineReader == null) {
log.info("gb32960 decoded frame service has no history index backend; rawUri direct lookup remains available");
- } else if (store == null) {
+ } else {
log.info("gb32960 decoded frame service using TDengine raw_frames index");
}
if (decoder == null) {
@@ -89,7 +68,6 @@ public final class Gb32960DecodedFrameService {
if (archiveRoot == null) {
throw new IllegalArgumentException("archiveRoot must not be null");
}
- this.store = store;
this.tdengineReader = tdengineReader;
this.decoder = decoder;
this.archiveRoot = archiveRoot.toAbsolutePath().normalize();
@@ -98,7 +76,7 @@ public final class Gb32960DecodedFrameService {
public List query(LocalDate dateFrom,
LocalDate dateTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -109,39 +87,28 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
int frameLimit = Math.max(1, Math.min(limit, 1000));
- List fromEventStore = store == null ? List.of() : queryFromEventStore(
- dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
- if (!fromEventStore.isEmpty() || tdengineReader == null) {
- return fromEventStore;
+ if (tdengineReader == null) {
+ return List.of();
}
return queryFromTdengine(dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
}
public DecodedFramePage queryPage(QueryTimeRange range,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
TdenginePageCursor cursor) throws IOException {
int frameLimit = Math.max(1, Math.min(limit, 1000));
- if (tdengineReader != null && vin != null && !vin.isBlank()) {
- return queryPageFromTdengine(
- range.dateFrom(),
- range.dateTo(),
- range.eventTimeFrom(),
- range.eventTimeTo(),
- order,
- frameLimit,
- vin,
- platformAccount,
- cursor);
+ if (tdengineReader == null || vin == null || vin.isBlank()) {
+ return new DecodedFramePage(List.of(), null);
}
- List items = query(
+ return queryPageFromTdengine(
range.dateFrom(),
range.dateTo(),
range.eventTimeFrom(),
@@ -149,55 +116,15 @@ public final class Gb32960DecodedFrameService {
order,
frameLimit,
vin,
- platformAccount);
- return new DecodedFramePage(items, null);
- }
-
- private List queryFromEventStore(LocalDate dateFrom,
- LocalDate dateTo,
- java.time.Instant eventTimeFrom,
- java.time.Instant eventTimeTo,
- EventFileQuery.Order order,
- int frameLimit,
- String vin,
- String platformAccount) throws IOException {
- // 只查 RAW_ARCHIVE：REALTIME/LOCATION 等派生事件不再作为 32960 历史查询的数据源。
- List records = store.query(new EventFileQuery(
- ProtocolId.GB32960,
- dateFrom,
- dateTo,
- eventTimeFrom,
- eventTimeTo,
- order == null ? EventFileQuery.Order.DESC : order,
- frameLimit,
- vin,
- RAW_ARCHIVE_EVENT_TYPE));
-
- List out = new ArrayList<>();
- LinkedHashSet seenUris = new LinkedHashSet<>();
- for (EventFileRecord record : records) {
- String rawArchiveUri = rawArchiveUri(record);
- // 同一个 RAW 可能因为索引重建或事件补偿被看到多次，按 URI 去重保证返回帧唯一。
- if (rawArchiveUri == null || rawArchiveUri.isBlank() || !seenUris.add(rawArchiveUri)) {
- continue;
- }
- DecodedFrame frame = decodeIfAvailable(record, rawArchiveUri, platformAccount);
- if (frame == null) {
- continue;
- }
- out.add(frame);
- if (out.size() >= frameLimit) {
- break;
- }
- }
- return out;
+ platformAccount,
+ cursor);
}
private List queryFromTdengine(LocalDate dateFrom,
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int frameLimit,
String vin,
String platformAccount) throws IOException {
@@ -229,7 +156,7 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int frameLimit,
String vin,
String platformAccount,
@@ -261,7 +188,7 @@ public final class Gb32960DecodedFrameService {
}
public List snapshots(QueryTimeRange range,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -270,7 +197,7 @@ public final class Gb32960DecodedFrameService {
}
public List snapshotFields(QueryTimeRange range,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
@@ -286,7 +213,7 @@ public final class Gb32960DecodedFrameService {
}
public String snapshotFieldsCsv(QueryTimeRange range,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
@@ -324,18 +251,12 @@ public final class Gb32960DecodedFrameService {
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
return null;
}
- if (store != null) {
- EventFileRecord record = store.findByRawArchiveUri(rawArchiveUri);
- if (record != null) {
- return decode(record, rawArchiveUri(record), platformAccount);
- }
- }
return decodeWithoutIndex(rawArchiveUri, platformAccount);
}
public List snapshots(LocalDate dateFrom,
LocalDate dateTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -349,7 +270,7 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
- EventFileQuery.Order order,
+ HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -378,19 +299,6 @@ public final class Gb32960DecodedFrameService {
return out;
}
- private DecodedFrame decodeIfAvailable(EventFileRecord record,
- String rawArchiveUri,
- String platformAccount) throws IOException {
- try {
- return decode(record, rawArchiveUri, platformAccount);
- } catch (NoSuchFileException e) {
- // 索引存在但 RAW 文件被人工清理时不中断整次查询，返回仍可用的其他帧。
- log.warn("skip gb32960 frame because raw archive is missing eventId={} rawArchiveUri={}",
- record.eventId(), rawArchiveUri);
- return null;
- }
- }
-
private DecodedFrame decodeIfAvailable(TdengineRawFrameRow row, String platformAccount) throws IOException {
try {
return decode(row, platformAccount);
@@ -401,29 +309,6 @@ public final class Gb32960DecodedFrameService {
}
}
- private DecodedFrame decode(EventFileRecord record, String rawArchiveUri, String platformAccount) throws IOException {
- Path rawPath = archivePath(rawArchiveUri);
- byte[] rawBytes = Files.readAllBytes(rawPath);
- Gb32960Message message = decoder.decode(ByteBuffer.wrap(rawBytes), platformAccount(record, platformAccount));
- List> blocks = new ArrayList<>();
- for (InfoBlock block : message.infoBlocks()) {
- blocks.add(blockJson(block));
- }
- return new DecodedFrame(
- record.eventId(),
- message.header().vin(),
- message.header().command().name(),
- message.header().responseFlag().name(),
- message.header().protocolVersion().name(),
- message.header().encryptType().name(),
- message.header().dataLength(),
- message.header().eventTime() == null ? null : message.header().eventTime().toString(),
- record.ingestTime().toString(),
- rawArchiveUri,
- rawBytes.length,
- blocks);
- }
-
private DecodedFrame decode(TdengineRawFrameRow row, String platformAccount) throws IOException {
Path rawPath = archivePath(row.rawUri());
byte[] rawBytes = Files.readAllBytes(rawPath);
@@ -618,26 +503,6 @@ public final class Gb32960DecodedFrameService {
return path;
}
- private static String rawArchiveUri(EventFileRecord record) {
- if (record.rawArchiveUri() != null && !record.rawArchiveUri().isBlank()) {
- return record.rawArchiveUri();
- }
- if (record.metadata() == null) {
- return "";
- }
- return record.metadata().getOrDefault("rawArchiveUri", "");
- }
-
- private static String platformAccount(EventFileRecord record, String override) {
- if (override != null && !override.isBlank()) {
- return override;
- }
- if (record.metadata() == null) {
- return null;
- }
- return record.metadata().get("platformAccount");
- }
-
private String platformAccount(TdengineRawFrameRow row, String override) {
if (override != null && !override.isBlank()) {
return override;
@@ -663,8 +528,8 @@ public final class Gb32960DecodedFrameService {
return inclusive.plusMillis(1);
}
- private static TdengineQueryOrder tdengineOrder(EventFileQuery.Order order) {
- return order == EventFileQuery.Order.ASC ? TdengineQueryOrder.ASC : TdengineQueryOrder.DESC;
+ private static TdengineQueryOrder tdengineOrder(HistoryQueryOrder order) {
+ return order == HistoryQueryOrder.ASC ? TdengineQueryOrder.ASC : TdengineQueryOrder.DESC;
}
private static String rawArchiveEventId(String rawArchiveUri) {
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
index 951a4f67..9daf98a9 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
@@ -1,6 +1,5 @@
package com.lingniu.ingest.eventhistory;
-import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@@ -23,7 +22,7 @@ import java.util.List;
*
* 这个 Controller 只做参数校验、时间范围解析和 Swagger 说明，真正的 RAW 读取、
* 协议解码、snapshot 合并、字段投影都放在 {@link Gb32960DecodedFrameService}。
- * 这样可以保证接口层足够薄，后续如果要把查询服务拆出去，只需要迁移 service 和 store。
+ * 这样可以保证接口层足够薄，后续如果要把查询服务拆出去，只需要迁移 service。
*/
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
@@ -52,7 +51,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
- @RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
+ @RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回帧数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN；用于缩小到单车查询。", example = "LB9A32A20P0LS1257")
@@ -82,7 +81,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
- @RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
+ @RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回帧数量上限，最大 1000。", example = "100")
@RequestParam(defaultValue = "100") int limit,
@Parameter(description = "车辆 VIN，必填；用于命中 TDengine raw child table。", required = true, example = "LB9A32A20P0LS1257")
@@ -139,7 +138,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
- @RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
+ @RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN，必填；snapshot 查询只支持单车，避免跨车合并和低效扫描。", required = true, example = "LB9A32A20P0LS1257")
@@ -163,7 +162,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
- @RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
+ @RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN，必填；字段查询只支持单车。", required = true, example = "LB9A32A20P0LS1257")
@@ -189,7 +188,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
- @RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
+ @RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "1000")
@RequestParam(defaultValue = "1000") int limit,
@Parameter(description = "车辆 VIN，必填；字段导出只支持单车。", required = true, example = "LB9A32A20P0LS1257")
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java
new file mode 100644
index 00000000..8b41219a
--- /dev/null
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java
@@ -0,0 +1,6 @@
+package com.lingniu.ingest.eventhistory;
+
+public enum HistoryQueryOrder {
+ ASC,
+ DESC
+}
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java
deleted file mode 100644
index 2c22d708..00000000
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java
+++ /dev/null
@@ -1,173 +0,0 @@
-package com.lingniu.ingest.eventhistory;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.lingniu.ingest.api.ProtocolId;
-import com.lingniu.ingest.api.event.RawArchiveKeys;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
-import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.util.JsonFormat;
-
-import java.time.Instant;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-/**
- * Maps Kafka protobuf envelopes into records stored by the historical detail
- * file store.
- *
- *
这是跨协议 Kafka envelope 回灌历史库的通用适配器。它保存的是 protobuf
- * telemetry_snapshot 的 JSON 视图，不负责 GB32960 原始包解码；32960 全字段 snapshot
- * 查询仍以 rawArchiveUri 指向的 .bin 为准。
- */
-public final class TelemetryEnvelopeRecordMapper {
-
- private static final JsonFormat.Printer JSON_PRINTER = JsonFormat.printer();
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
- public EventFileRecord toRecord(VehicleEnvelope envelope) {
- if (envelope == null) {
- throw new IllegalArgumentException("envelope must not be null");
- }
- if (envelope.hasTelemetrySnapshot()) {
- return telemetryRecord(envelope);
- }
- if (envelope.hasRawArchive()) {
- return rawArchiveRecord(envelope);
- }
- throw new IllegalArgumentException("envelope telemetry_snapshot or raw_archive is required");
- }
-
- private static EventFileRecord telemetryRecord(VehicleEnvelope envelope) {
- ProtocolId protocol = protocol(envelope.getSource());
- Map metadata = metadata(envelope, protocol);
- return new EventFileRecord(
- envelope.getEventId(),
- protocol,
- envelope.getTelemetrySnapshot().getEventType(),
- envelope.getVin(),
- Instant.ofEpochMilli(envelope.getEventTimeMs()),
- Instant.ofEpochMilli(envelope.getIngestTimeMs()),
- envelope.getTelemetrySnapshot().getRawArchiveUri(),
- metadata,
- snapshotJson(envelope)
- );
- }
-
- private static EventFileRecord rawArchiveRecord(VehicleEnvelope envelope) {
- ProtocolId protocol = protocol(envelope.getSource());
- RawArchiveRef rawArchive = envelope.getRawArchive();
- Map metadata = metadata(envelope, protocol);
- String rawArchiveKey = rawArchiveKey(rawArchive.getUri(), metadata);
- String rawArchiveUri = rawArchive.getUri().isBlank() && !rawArchiveKey.isBlank()
- ? RawArchiveKeys.logicalUri(rawArchiveKey)
- : rawArchive.getUri();
- metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId());
- if (!rawArchiveKey.isBlank()) {
- metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey);
- }
- if (!rawArchiveUri.isBlank()) {
- metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri);
- }
- return new EventFileRecord(
- envelope.getEventId(),
- protocol,
- "RAW_ARCHIVE",
- envelope.getVin(),
- Instant.ofEpochMilli(envelope.getEventTimeMs()),
- Instant.ofEpochMilli(envelope.getIngestTimeMs()),
- rawArchiveUri,
- metadata,
- rawArchiveJson(envelope, metadata, rawArchiveKey, rawArchiveUri)
- );
- }
-
- private static String rawArchiveKey(String rawArchiveUri, Map metadata) {
- String key = metadata.getOrDefault(RawArchiveKeys.META_KEY, "");
- if (!key.isBlank()) {
- return key;
- }
- if (rawArchiveUri != null && rawArchiveUri.startsWith("archive://")) {
- return rawArchiveUri.substring("archive://".length());
- }
- return "";
- }
-
- private static Map metadata(VehicleEnvelope envelope, ProtocolId protocol) {
- Map metadata = new LinkedHashMap<>(envelope.getMetadataMap());
- if (!envelope.getProtocolVersion().isBlank()) {
- metadata.putIfAbsent("protocolVersion", envelope.getProtocolVersion());
- }
- if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
- // 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
- metadata.putIfAbsent("originalSource", envelope.getSource());
- }
- return metadata;
- }
-
- private static ProtocolId protocol(String source) {
- if (source == null || source.isBlank()) {
- return ProtocolId.UNKNOWN;
- }
- try {
- return ProtocolId.valueOf(source);
- } catch (IllegalArgumentException ex) {
- return ProtocolId.UNKNOWN;
- }
- }
-
- private static String snapshotJson(VehicleEnvelope envelope) {
- try {
- return JSON_PRINTER.print(envelope.getTelemetrySnapshot());
- } catch (InvalidProtocolBufferException e) {
- throw new IllegalArgumentException("failed to serialize telemetry_snapshot", e);
- }
- }
-
- private static String rawArchiveJson(VehicleEnvelope envelope,
- Map metadata,
- String rawArchiveKey,
- String rawArchiveUri) {
- Map payload = new LinkedHashMap<>();
- RawArchiveRef rawArchive = envelope.getRawArchive();
- payload.put("eventId", envelope.getEventId());
- payload.put("vin", envelope.getVin());
- payload.put("protocol", protocol(envelope.getSource()).name());
- if (!envelope.getSource().isBlank()) {
- payload.put("source", envelope.getSource());
- }
- payload.put("eventType", "RAW_ARCHIVE");
- payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString());
- payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString());
- payload.put("rawArchiveUri", rawArchiveUri);
- payload.put("rawArchiveKey", rawArchiveKey);
- if (metadata.containsKey("command")) {
- payload.put("command", metadata.get("command"));
- }
- if (metadata.containsKey("infoType")) {
- payload.put("infoType", metadata.get("infoType"));
- }
- payload.put("rawSizeBytes", rawArchive.getSizeBytes());
- putParsedJson(payload, rawArchive.getParsedJson());
- payload.put("metadata", metadata);
- try {
- return OBJECT_MAPPER.writeValueAsString(payload);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("failed to serialize raw_archive", e);
- }
- }
-
- private static void putParsedJson(Map payload, String parsedJson) {
- if (parsedJson == null || parsedJson.isBlank()) {
- return;
- }
- try {
- payload.put("parsed", OBJECT_MAPPER.readTree(parsedJson));
- } catch (JsonProcessingException ex) {
- payload.put("parsed", parsedJson);
- payload.put("parsedJsonError", ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage());
- }
- }
-}
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
index 47b5df20..1b68e036 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
@@ -2,8 +2,6 @@ package com.lingniu.ingest.eventhistory.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
-import com.lingniu.ingest.api.history.EventFileStore;
-import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -12,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
-import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.sink.kafka.KafkaSinkAutoConfiguration;
@@ -37,7 +34,7 @@ import java.nio.file.Path;
* 该模块有两类入口：
*
* - TDengine HTTP 查询入口：查询位置、RAW 帧和按需解码的 GB32960 帧。
- *
- 兼容 EventFileStore 入口：仅在旧索引显式启用时创建旧记录查询和旧写入路径。
+ *
- Kafka consumer 入口：消费 envelope 并写入 TDengine 历史表。
*
*
* 当前生产 history app 以 TDengine 为准；Kafka consumer 是否启动还取决于
@@ -52,22 +49,6 @@ import java.nio.file.Path;
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
public class EventHistoryAutoConfiguration {
- @Bean
- @ConditionalOnBean(EventFileStore.class)
- @ConditionalOnMissingBean
- public TelemetryEnvelopeRecordMapper telemetryEnvelopeRecordMapper() {
- return new TelemetryEnvelopeRecordMapper();
- }
-
- @Bean
- @ConditionalOnBean(EventFileStore.class)
- @ConditionalOnMissingBean(value = {TdengineHistoryWriter.class, EventHistoryEnvelopeIngestor.class})
- public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store,
- TelemetryEnvelopeRecordMapper mapper,
- ObjectProvider writer) {
- return new EventHistoryEnvelopeIngestor(store, mapper, writer.getIfAvailable());
- }
-
@Bean
@ConditionalOnBean(TdengineHistoryWriter.class)
@ConditionalOnMissingBean
@@ -94,24 +75,14 @@ public class EventHistoryAutoConfiguration {
return new EnvelopeConsumerProcessor("event-history-raw", ingestor, deadLetterSink);
}
- @Bean
- @ConditionalOnBean(EventFileStore.class)
- @ConditionalOnMissingBean
- public EventHistoryController eventHistoryController(EventFileStore store) {
- return new EventHistoryController(store);
- }
-
@Bean
@ConditionalOnBean(Gb32960MessageDecoder.class)
@ConditionalOnMissingBean
- public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider store,
- ObjectProvider reader,
+ public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider reader,
Gb32960MessageDecoder decoder,
@Value("${lingniu.ingest.event-history.archive-path:${SINK_ARCHIVE_PATH:./archive/}}")
String archivePath) {
- // 优先使用 EventFileStore 索引；TDengine-only 高吞吐运行时可直接从 raw_frames 找 rawUri。
- return new Gb32960DecodedFrameService(store.getIfAvailable(), reader.getIfAvailable(),
- decoder, archiveRoot(archivePath), null);
+ return new Gb32960DecodedFrameService(reader.getIfAvailable(), decoder, archiveRoot(archivePath), null);
}
@Bean
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java
index 597957ee..9e0adbf1 100644
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java
@@ -1,8 +1,7 @@
package com.lingniu.ingest.eventhistory;
-import com.lingniu.ingest.api.history.EventFileStore;
-import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import jakarta.servlet.http.HttpServletRequest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -14,8 +13,6 @@ import org.springframework.web.servlet.resource.NoResourceFoundException;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.List;
-
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.assertj.core.api.Assertions.assertThat;
@@ -87,18 +84,8 @@ class ApiExceptionHandlerTest {
}
private Gb32960DecodedFrameService service() {
- EventFileStore store = new EventFileStore() {
- @Override
- public void appendAll(List records) {
- }
-
- @Override
- public List query(com.lingniu.ingest.api.history.EventFileQuery query) {
- return List.of();
- }
- };
return new Gb32960DecodedFrameService(
- store,
+ (TdengineHistoryReader) null,
mock(Gb32960MessageDecoder.class),
archiveRoot,
null);
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java
deleted file mode 100644
index b4a4794c..00000000
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java
+++ /dev/null
@@ -1,82 +0,0 @@
-package com.lingniu.ingest.eventhistory;
-
-import com.lingniu.ingest.api.ProtocolId;
-import com.lingniu.ingest.api.history.EventFileQuery;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.time.Instant;
-import java.util.List;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-class EventHistoryControllerTest {
-
- @Test
- void queriesRecordsByDateRangeAndOrder() throws Exception {
- CapturingStore store = new CapturingStore(List.of(record("event-1")));
- EventHistoryController controller = new EventHistoryController(store);
-
- List records = controller.query(
- ProtocolId.GB32960,
- "2026-06-21",
- "2026-06-22",
- EventFileQuery.Order.DESC,
- 50,
- "VIN001",
- "RAW_ARCHIVE",
- "2026-06-22T08:00:00Z",
- "2026-06-22T08:00:01Z",
- "event-0");
-
- assertThat(records).extracting(EventHistoryController.RecordResponse::eventId).containsExactly("event-1");
- assertThat(records.getFirst().eventTime()).isEqualTo("2026-06-22T08:00:00Z");
- assertThat(records.getFirst().ingestTime()).isEqualTo("2026-06-22T08:00:01Z");
- assertThat(store.lastQuery.protocol()).isEqualTo(ProtocolId.GB32960);
- assertThat(store.lastQuery.dateFrom().toString()).isEqualTo("2026-06-21");
- assertThat(store.lastQuery.dateTo().toString()).isEqualTo("2026-06-22");
- assertThat(store.lastQuery.order()).isEqualTo(EventFileQuery.Order.DESC);
- assertThat(store.lastQuery.limit()).isEqualTo(50);
- assertThat(store.lastQuery.vin()).isEqualTo("VIN001");
- assertThat(store.lastQuery.eventType()).isEqualTo("RAW_ARCHIVE");
- assertThat(store.lastQuery.cursorEventTime()).isEqualTo(Instant.parse("2026-06-22T08:00:00Z"));
- assertThat(store.lastQuery.cursorIngestTime()).isEqualTo(Instant.parse("2026-06-22T08:00:01Z"));
- assertThat(store.lastQuery.cursorEventId()).isEqualTo("event-0");
- }
-
- private static EventFileRecord record(String eventId) {
- return new EventFileRecord(
- eventId,
- ProtocolId.GB32960,
- "REALTIME",
- "VIN001",
- Instant.parse("2026-06-22T08:00:00Z"),
- Instant.parse("2026-06-22T08:00:01Z"),
- "archive://raw/" + eventId + ".bin",
- Map.of(),
- "{\"fields\":[]}");
- }
-
- private static final class CapturingStore implements EventFileStore {
- private final List records;
- private EventFileQuery lastQuery;
-
- private CapturingStore(List records) {
- this.records = records;
- }
-
- @Override
- public void appendAll(List records) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public List query(EventFileQuery query) throws IOException {
- this.lastQuery = query;
- return records;
- }
- }
-}
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java
index 084f688b..fcd612be 100644
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java
@@ -1,14 +1,7 @@
package com.lingniu.ingest.eventhistory;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
-import com.lingniu.ingest.api.event.RawArchiveKeys;
-import com.lingniu.ingest.api.history.EventFileQuery;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
@@ -22,8 +15,6 @@ import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import org.junit.jupiter.api.Test;
-import java.io.IOException;
-import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -32,105 +23,36 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class EventHistoryEnvelopeIngestorTest {
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
- @Test
- void parsesKafkaEnvelopeBytesAndAppendsRecord() throws Exception {
- CapturingStore store = new CapturingStore();
- EventHistoryEnvelopeIngestor ingestor =
- new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
-
- assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
- ingestor.ingest(envelope("event-1").toByteArray());
-
- assertThat(store.records).extracting(EventFileRecord::eventId).containsExactly("event-1");
- assertThat(store.records.getFirst().payloadJson()).contains("total_mileage_km");
- }
-
@Test
void rejectsInvalidEnvelopeBytes() {
+ CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor =
- new EventHistoryEnvelopeIngestor(new CapturingStore(), new TelemetryEnvelopeRecordMapper());
+ new EventHistoryEnvelopeIngestor(tdengineWriter);
+ assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
assertThatThrownBy(() -> ingestor.ingest(new byte[]{0x01, 0x02}))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("VehicleEnvelope");
}
@Test
- void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutAppendingRecord() {
- CapturingStore store = new CapturingStore();
+ void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutWritingTdengine() {
+ CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor =
- new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
+ new EventHistoryEnvelopeIngestor(tdengineWriter);
var result = ingestor.tryIngest(new byte[]{0x01, 0x02});
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.INVALID_ENVELOPE);
assertThat(result.message()).contains("VehicleEnvelope");
- assertThat(store.records).isEmpty();
- }
-
- @Test
- void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() throws Exception {
- CapturingStore store = new CapturingStore();
- EventHistoryEnvelopeIngestor ingestor =
- new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
- String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-1.bin";
- String rawArchiveUri = "archive://" + rawArchiveKey;
-
- VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
- .setSchemaVersion("1.0")
- .setEventId("raw-event-1")
- .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)
- .putMetadata("command", "0x0002")
- .setRawArchive(RawArchiveRef.newBuilder()
- .setUri(rawArchiveUri)
- .setSizeBytes(128)
- .build())
- .build();
-
- EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
-
- assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
- assertThat(store.records).hasSize(1);
- EventFileRecord record = store.records.getFirst();
- assertThat(record.eventId()).isEqualTo("raw-event-1");
- assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960);
- assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
- assertThat(record.vin()).isEqualTo("VINRAW001");
- assertThat(record.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L));
- assertThat(record.rawArchiveUri()).isEqualTo(rawArchiveUri);
- assertThat(record.metadata())
- .containsEntry("protocolVersion", "V2016")
- .containsEntry(RawArchiveKeys.META_EVENT_ID, "raw-event-1")
- .containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey)
- .containsEntry(RawArchiveKeys.META_URI, rawArchiveUri)
- .containsEntry("command", "0x0002");
- JsonNode payload = OBJECT_MAPPER.readTree(record.payloadJson());
- assertThat(payload.get("eventId").asText()).isEqualTo("raw-event-1");
- assertThat(payload.get("protocol").asText()).isEqualTo("GB32960");
- assertThat(payload.get("eventType").asText()).isEqualTo("RAW_ARCHIVE");
- assertThat(payload.get("vin").asText()).isEqualTo("VINRAW001");
- assertThat(payload.get("eventTime").asText()).isEqualTo("2026-06-22T07:13:20Z");
- assertThat(payload.get("rawArchiveUri").asText()).isEqualTo(rawArchiveUri);
- assertThat(payload.get("rawArchiveKey").asText()).isEqualTo(rawArchiveKey);
- assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(128);
- assertThat(payload.get("metadata").get("command").asText()).isEqualTo("0x0002");
- assertThat(payload.toString()).doesNotContain("rawBytes");
+ assertThat(tdengineWriter.rawFrames).isEmpty();
+ assertThat(tdengineWriter.locations).isEmpty();
}
@Test
void tryIngestWritesRawAndLocationFactsToTdengineWhenWriterExists() throws Exception {
- CapturingStore store = new CapturingStore();
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
- EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(
- store, new TelemetryEnvelopeRecordMapper(), tdengineWriter);
+ EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter);
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
@@ -172,7 +94,6 @@ class EventHistoryEnvelopeIngestorTest {
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
- assertThat(store.records).hasSize(1);
assertThat(tdengineWriter.rawFrames)
.extracting(TdengineRawFrameRow::frameId)
.containsExactly("frame-jt808-1");
@@ -219,10 +140,8 @@ class EventHistoryEnvelopeIngestorTest {
@Test
void tryIngestAllBatchesFileStoreAndTdengineWrites() throws Exception {
- CapturingStore store = new CapturingStore();
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
- EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(
- store, new TelemetryEnvelopeRecordMapper(), tdengineWriter);
+ EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter);
VehicleEnvelope first = jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001");
VehicleEnvelope second = jt808LocationEnvelope("jt808-location-2", "frame-jt808-2", "013800000002");
@@ -237,9 +156,6 @@ class EventHistoryEnvelopeIngestorTest {
EnvelopeIngestResult.Status.STORED,
EnvelopeIngestResult.Status.INVALID_ENVELOPE,
EnvelopeIngestResult.Status.STORED);
- assertThat(store.appendAllCalls).isEqualTo(1);
- assertThat(store.records).extracting(EventFileRecord::eventId)
- .containsExactly("jt808-location-1", "jt808-location-2");
assertThat(tdengineWriter.rawFrames).extracting(TdengineRawFrameRow::frameId)
.containsExactly("frame-jt808-1", "frame-jt808-2");
assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId)
@@ -247,24 +163,6 @@ class EventHistoryEnvelopeIngestorTest {
assertThat(tdengineWriter.telemetryFields).isEmpty();
}
- private static VehicleEnvelope envelope(String eventId) {
- return VehicleEnvelope.newBuilder()
- .setEventId(eventId)
- .setVin("VIN001")
- .setSource("GB32960")
- .setEventTimeMs(1_782_112_400_000L)
- .setIngestTimeMs(1_782_112_401_000L)
- .setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
- .setEventType("REALTIME")
- .addFields(TelemetryField.newBuilder()
- .setKey("total_mileage_km")
- .setValueType("DOUBLE")
- .setValue("12345.6")
- .setUnit("km")
- .setQuality("GOOD")))
- .build();
- }
-
private static VehicleEnvelope jt808LocationEnvelope(String eventId, String frameId, String phone) {
return VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
@@ -304,22 +202,6 @@ class EventHistoryEnvelopeIngestorTest {
.build();
}
- private static final class CapturingStore implements EventFileStore {
- private final List records = new ArrayList<>();
- private int appendAllCalls;
-
- @Override
- public void appendAll(List records) {
- appendAllCalls++;
- this.records.addAll(records);
- }
-
- @Override
- public List query(EventFileQuery query) throws IOException {
- return List.copyOf(records);
- }
- }
-
private static final class CapturingTdengineWriter implements TdengineHistoryWriter {
private final List rawFrames = new ArrayList<>();
private final List locations = new ArrayList<>();
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java
index f0590cb8..600df17f 100644
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java
@@ -1,10 +1,6 @@
package com.lingniu.ingest.eventhistory;
-import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.codec.BccChecksum;
-import com.lingniu.ingest.api.history.EventFileQuery;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960BodyParser;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.codec.InfoBlockParser;
@@ -53,18 +49,10 @@ class Gb32960DecodedFrameServiceTest {
Path rawFile = archiveRoot.resolve(key);
Files.createDirectories(rawFile.getParent());
Files.write(rawFile, realtimeFrame());
+ String uri = "archive://" + key;
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(new EventFileRecord(
- "raw-1",
- ProtocolId.GB32960,
- "RAW_ARCHIVE",
- "LNVFC000000000001",
- Instant.parse("2026-06-22T10:00:00Z"),
- Instant.parse("2026-06-22T10:00:01Z"),
- "archive://" + key,
- Map.of("platformAccount", "Hyundai"),
- "{}"))),
+ new CapturingTdengineReader(List.of(rawRow("raw-1", uri))),
decoder(),
archiveRoot,
null);
@@ -72,7 +60,7 @@ class Gb32960DecodedFrameServiceTest {
List frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
null,
null);
@@ -109,7 +97,7 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of()),
+ (TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
@@ -123,42 +111,6 @@ class Gb32960DecodedFrameServiceTest {
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
- @Test
- void findsSingleFrameDirectlyByRawArchiveUriWhenEventFileStoreIsDisabled() throws Exception {
- String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000000.bin";
- writeArchive(key, realtimeFrame());
-
- Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- (EventFileStore) null,
- decoder(),
- archiveRoot,
- null);
-
- Gb32960DecodedFrameService.DecodedFrame frame =
- service.findByRawArchiveUri("archive://" + key, "Hyundai");
-
- assertThat(frame.eventId()).isEqualTo("1792527837000000");
- assertThat(frame.vin()).isEqualTo("LNVFC000000000001");
- }
-
- @Test
- void findsSingleFrameByRawArchiveUriIndexWhenAvailable() throws Exception {
- String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000001.bin";
- String uri = "archive://" + key;
- writeArchive(key, realtimeFrame());
-
- Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(record("indexed-raw", uri))),
- decoder(),
- archiveRoot,
- null);
-
- Gb32960DecodedFrameService.DecodedFrame frame = service.findByRawArchiveUri(uri, null);
-
- assertThat(frame.eventId()).isEqualTo("indexed-raw");
- assertThat(frame.rawArchiveUri()).isEqualTo(uri);
- }
-
@Test
void skipsMissingRawArchiveAndContinues() throws Exception {
String missingKey = "2026/06/22/GB32960/LNVFC000000000001/missing.bin";
@@ -168,9 +120,9 @@ class Gb32960DecodedFrameServiceTest {
Files.write(rawFile, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(
- record("missing", "archive://" + missingKey),
- record("available", "archive://" + availableKey))),
+ new CapturingTdengineReader(List.of(
+ rawRow("missing", "archive://" + missingKey),
+ rawRow("available", "archive://" + availableKey))),
decoder(),
archiveRoot,
null);
@@ -178,7 +130,7 @@ class Gb32960DecodedFrameServiceTest {
List frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
null,
null);
@@ -189,39 +141,10 @@ class Gb32960DecodedFrameServiceTest {
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
- @Test
- void queryRequestsOnlyRawArchiveRecordsForOneVin() throws Exception {
- String key = "2026/06/22/GB32960/LNVFC000000000001/raw-vin.bin";
- writeArchive(key, realtimeFrame());
- CapturingStore store = new CapturingStore(List.of(
- businessRecord("business-newer", "LNVFC000000000001"),
- record("raw-target", "archive://" + key),
- record("raw-other-vin", "archive://2026/06/22/GB32960/OTHER/raw.bin", "OTHER000000000000")));
-
- Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- store,
- decoder(),
- archiveRoot,
- null);
-
- List frames = service.query(
- LocalDate.parse("2026-06-22"),
- LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
- 1,
- "LNVFC000000000001",
- null);
-
- assertThat(store.lastQuery.eventType()).isEqualTo("RAW_ARCHIVE");
- assertThat(store.lastQuery.vin()).isEqualTo("LNVFC000000000001");
- assertThat(frames).hasSize(1);
- assertThat(frames.getFirst().eventId()).isEqualTo("raw-target");
- }
-
@Test
void snapshotsRequireVinSoSegmentsAreMergedOnlyWithinOneVehicle() {
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of()),
+ (TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
@@ -229,7 +152,7 @@ class Gb32960DecodedFrameServiceTest {
assertThatThrownBy(() -> service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
null,
null))
@@ -248,10 +171,10 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key3, stackFrame(8, 2, false));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(
- record("segment-3", "archive://" + key3),
- record("segment-2", "archive://" + key2),
- record("segment-1", "archive://" + key1))),
+ new CapturingTdengineReader(List.of(
+ rawRow("segment-3", "archive://" + key3),
+ rawRow("segment-2", "archive://" + key2),
+ rawRow("segment-1", "archive://" + key1))),
decoder(),
archiveRoot,
null);
@@ -259,7 +182,7 @@ class Gb32960DecodedFrameServiceTest {
List snapshots = service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -328,7 +251,7 @@ class Gb32960DecodedFrameServiceTest {
List snapshots = service.snapshots(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -376,7 +299,7 @@ class Gb32960DecodedFrameServiceTest {
List frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
null,
"Hyundai");
@@ -425,7 +348,7 @@ class Gb32960DecodedFrameServiceTest {
Gb32960DecodedFrameService.DecodedFramePage page = service.queryPage(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
- EventFileQuery.Order.ASC,
+ HistoryQueryOrder.ASC,
10,
vin,
null,
@@ -449,7 +372,7 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, stackFrame(1, 4, true));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(record("partial", "archive://" + key))),
+ new CapturingTdengineReader(List.of(rawRow("partial", "archive://" + key))),
decoder(),
archiveRoot,
null);
@@ -457,7 +380,7 @@ class Gb32960DecodedFrameServiceTest {
List snapshots = service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -488,14 +411,14 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(record("selected", "archive://" + key))),
+ new CapturingTdengineReader(List.of(rawRow("selected", "archive://" + key))),
decoder(),
archiveRoot,
null);
List snapshots = service.snapshotFields(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
vin,
null,
@@ -515,14 +438,14 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of(record("selected-csv", "archive://" + key))),
+ new CapturingTdengineReader(List.of(rawRow("selected-csv", "archive://" + key))),
decoder(),
archiveRoot,
null);
String csv = service.snapshotFieldsCsv(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
vin,
null,
@@ -536,14 +459,14 @@ class Gb32960DecodedFrameServiceTest {
@Test
void snapshotFieldsRejectsUnknownDictionaryFields() throws Exception {
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
- new CapturingStore(List.of()),
+ (TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
assertThatThrownBy(() -> service.snapshotFields(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
- EventFileQuery.Order.DESC,
+ HistoryQueryOrder.DESC,
10,
"LNVFC000000000001",
null,
@@ -562,34 +485,30 @@ class Gb32960DecodedFrameServiceTest {
return new Gb32960MessageDecoder(new Gb32960BodyParser(registry, selector));
}
- private static EventFileRecord record(String eventId, String rawArchiveUri) {
- return record(eventId, rawArchiveUri, "LNVFC000000000001");
+ private static TdengineRawFrameRow rawRow(String eventId, String rawArchiveUri) {
+ return rawRow(eventId, rawArchiveUri, "LNVFC000000000001");
}
- private static EventFileRecord record(String eventId, String rawArchiveUri, String vin) {
- return new EventFileRecord(
- eventId,
- ProtocolId.GB32960,
- "RAW_ARCHIVE",
- vin,
+ private static TdengineRawFrameRow rawRow(String eventId, String rawArchiveUri, String vin) {
+ return new TdengineRawFrameRow(
Instant.parse("2026-06-22T10:00:00Z"),
- Instant.parse("2026-06-22T10:00:01Z"),
- rawArchiveUri,
- Map.of("platformAccount", "Hyundai"),
- "{}");
- }
-
- private static EventFileRecord businessRecord(String eventId, String vin) {
- return new EventFileRecord(
eventId,
- ProtocolId.GB32960,
- "REALTIME",
+ Instant.parse("2026-06-22T10:00:01Z"),
+ 0x02,
+ 0,
+ Instant.parse("2026-06-22T10:00:00Z"),
+ rawArchiveUri,
+ "sha256:test",
+ 64,
+ "SUCCEEDED",
+ "",
+ "127.0.0.1:32960",
+ "{\"platformAccount\":\"Hyundai\"}",
+ "",
+ "GB32960",
vin,
- Instant.parse("2026-06-22T10:00:03Z"),
- Instant.parse("2026-06-22T10:00:04Z"),
- "archive://business/" + eventId + ".bin",
- Map.of("platformAccount", "Hyundai"),
- "{}");
+ vin,
+ "");
}
private static byte[] realtimeFrame() {
@@ -739,38 +658,6 @@ class Gb32960DecodedFrameServiceTest {
os.write((int) (v & 0xFF));
}
- private static final class CapturingStore implements EventFileStore {
- private final List records;
- private EventFileQuery lastQuery;
-
- private CapturingStore(List records) {
- this.records = records;
- }
-
- @Override
- public void appendAll(List records) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public List query(EventFileQuery query) {
- lastQuery = query;
- return records.stream()
- .filter(record -> query.vin() == null || query.vin().equals(record.vin()))
- .filter(record -> query.eventType() == null || query.eventType().equals(record.eventType()))
- .limit(query.limit())
- .toList();
- }
-
- @Override
- public EventFileRecord findByRawArchiveUri(String rawArchiveUri) {
- return records.stream()
- .filter(record -> record.rawArchiveUri().equals(rawArchiveUri))
- .findFirst()
- .orElse(null);
- }
- }
-
private static final class CapturingTdengineReader implements TdengineHistoryReader {
private final List rows;
private final TdenginePageCursor nextCursor;
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java
deleted file mode 100644
index 96538c3a..00000000
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java
+++ /dev/null
@@ -1,127 +0,0 @@
-package com.lingniu.ingest.eventhistory;
-
-import com.lingniu.ingest.api.ProtocolId;
-import com.lingniu.ingest.api.history.EventFileRecord;
-import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
-import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
-import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot;
-import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.jupiter.api.Test;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-class TelemetryEnvelopeRecordMapperTest {
-
- private final ObjectMapper objectMapper = new ObjectMapper();
-
- @Test
- void mapsFullFieldEnvelopeToEventFileRecord() {
- VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
- .setSchemaVersion("1.0")
- .setEventId("event-1")
- .setTraceId("trace-1")
- .setVin("VIN001")
- .setSource("GB32960")
- .setEventTimeMs(1_782_112_400_000L)
- .setIngestTimeMs(1_782_112_401_000L)
- .putMetadata("protocolVersion", "V2016")
- .setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
- .setEventType("REALTIME")
- .setRawArchiveUri("archive://raw/event-1.bin")
- .addFields(TelemetryField.newBuilder()
- .setKey("total_mileage_km")
- .setValueType("DOUBLE")
- .setValue("12345.6")
- .setUnit("km")
- .setQuality("GOOD")
- .setSourcePath("event.realtime.totalMileageKm"))
- .build())
- .build();
-
- EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
-
- assertThat(record.eventId()).isEqualTo("event-1");
- assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960);
- assertThat(record.eventType()).isEqualTo("REALTIME");
- assertThat(record.vin()).isEqualTo("VIN001");
- assertThat(record.rawArchiveUri()).isEqualTo("archive://raw/event-1.bin");
- assertThat(record.metadata()).containsEntry("protocolVersion", "V2016");
- assertThat(record.payloadJson()).contains("\"fields\"");
- assertThat(record.payloadJson()).contains("\"total_mileage_km\"");
- }
-
- @Test
- void rejectsEnvelopeWithoutTelemetrySnapshot() {
- VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
- .setEventId("event-1")
- .setVin("VIN001")
- .setSource("GB32960")
- .setEventTimeMs(1)
- .setIngestTimeMs(2)
- .build();
-
- assertThatThrownBy(() -> new TelemetryEnvelopeRecordMapper().toRecord(envelope))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("telemetry_snapshot");
- }
-
- @Test
- void unknownEnvelopeSourceIsStoredAsUnknownProtocolWithOriginalSourceMetadata() {
- VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
- .setSchemaVersion("1.0")
- .setEventId("event-future-1")
- .setTraceId("trace-future-1")
- .setVin("VIN-FUTURE-001")
- .setSource("FUTURE_OEM")
- .setEventTimeMs(1_782_112_400_000L)
- .setIngestTimeMs(1_782_112_401_000L)
- .putMetadata("tenant", "ln")
- .setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
- .setEventType("REALTIME")
- .addFields(TelemetryField.newBuilder()
- .setKey("speed_kmh")
- .setValueType("DOUBLE")
- .setValue("30.5")
- .setQuality("GOOD"))
- .build())
- .build();
-
- EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
-
- assertThat(record.protocol()).isEqualTo(ProtocolId.UNKNOWN);
- assertThat(record.metadata())
- .containsEntry("tenant", "ln")
- .containsEntry("originalSource", "FUTURE_OEM");
- }
-
- @Test
- void rawArchiveRecordStoresParsedJsonInPayloadOnly() throws Exception {
- VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
- .setSchemaVersion("1.0")
- .setEventId("raw-808-1")
- .setTraceId("trace-raw-1")
- .setVin("VIN808")
- .setSource("JT808")
- .setEventTimeMs(1_782_112_400_000L)
- .setIngestTimeMs(1_782_112_401_000L)
- .putMetadata("phone", "13307795425")
- .setRawArchive(RawArchiveRef.newBuilder()
- .setUri("archive://2026/06/30/JT808/VIN808/raw-808-1.bin")
- .setSizeBytes(64)
- .setParsedJson("{\"messageId\":\"0x0200\",\"body\":{\"speedKmh\":60.5}}")
- .build())
- .build();
-
- EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
-
- assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
- assertThat(record.metadata())
- .containsEntry("phone", "13307795425")
- .doesNotContainKey("rawArchiveParsedJson");
- var payload = objectMapper.readTree(record.payloadJson());
- assertThat(payload.path("parsed").path("messageId").asText()).isEqualTo("0x0200");
- assertThat(payload.path("parsed").path("body").path("speedKmh").asDouble()).isEqualTo(60.5);
- }
-}
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
index 40b80ca3..e5fcb08e 100644
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
@@ -2,8 +2,6 @@ package com.lingniu.ingest.eventhistory.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
-import com.lingniu.ingest.api.history.EventFileStore;
-import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -12,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
-import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
@@ -30,7 +27,7 @@ class EventHistoryAutoConfigurationTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {})
- .withBean(EventFileStore.class, () -> mock(EventFileStore.class));
+ .withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class));
@Test
void createsHistoryBeansWhenEnabled() {
@@ -39,13 +36,11 @@ class EventHistoryAutoConfigurationTest {
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.sink.kafka.consumer.enabled=true")
.run(context -> {
- assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys(
"eventHistoryEnvelopeConsumerProcessor",
"eventHistoryRawEnvelopeConsumerProcessor");
- assertThat(context).hasSingleBean(EventHistoryController.class);
assertThat(context).doesNotHaveBean(Gb32960DecodedFrameService.class);
assertThat(context).doesNotHaveBean(Gb32960FrameController.class);
});
@@ -54,9 +49,7 @@ class EventHistoryAutoConfigurationTest {
@Test
void backsOffWhenDisabled() {
contextRunner.run(context -> {
- assertThat(context).doesNotHaveBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).doesNotHaveBean(EventHistoryEnvelopeIngestor.class);
- assertThat(context).doesNotHaveBean(EventHistoryController.class);
});
}
@@ -75,7 +68,7 @@ class EventHistoryAutoConfigurationTest {
}
@Test
- void createsGb32960FrameBeansWithTdengineReaderWhenEventFileStoreIsDisabled() {
+ void createsGb32960FrameBeansWithTdengineReader() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withPropertyValues(
@@ -85,22 +78,19 @@ class EventHistoryAutoConfigurationTest {
.withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.run(context -> {
- assertThat(context).doesNotHaveBean(EventFileStore.class);
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(Gb32960FrameController.class);
});
}
@Test
- void createsHistoryConsumerWithTdengineWriterWhenEventFileStoreIsDisabled() {
+ void createsHistoryConsumerWithTdengineWriter() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {})
.withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class))
.withPropertyValues("lingniu.ingest.event-history.enabled=true")
.run(context -> {
- assertThat(context).doesNotHaveBean(EventFileStore.class);
- assertThat(context).doesNotHaveBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys(