From 12ac1dab3bb45dd15d3ff8fd74619b88c13fc0e0 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 20:44:21 +0800 Subject: [PATCH] feat: add jt808 raw frame history query --- .../vehicle-ingest-tdengine-verification.md | 3 + .../Jt808RawFrameHistoryController.java | 177 ++++++++++++++++++ .../config/EventHistoryAutoConfiguration.java | 8 + .../Jt808RawFrameHistoryControllerTest.java | 129 +++++++++++++ .../EventHistoryAutoConfigurationTest.java | 9 + .../TdengineHistoryQueries.java | 48 ++++- .../TdengineHistorySchema.java | 4 + .../TdengineRawFrameQuery.java | 37 +++- .../TdengineHistoryQueriesTest.java | 35 ++++ 9 files changed, 443 insertions(+), 7 deletions(-) create mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java create mode 100644 modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index fc15fcb4..863c37be 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -313,8 +313,11 @@ SELECT ts, frame_id, phone, message_id, metadata_json, raw_uri ```bash curl -sS 'http://127.0.0.1:20200/api/event-history/jt808/locations?phone=13079963301&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&order=DESC&limit=3' +curl -sS 'http://127.0.0.1:20200/api/event-history/jt808/raw-frames?phone=13079963320&messageId=256&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&order=DESC&limit=3' ``` +`/api/event-history/jt808/raw-frames` 用于排查注册、鉴权、心跳、位置等原始帧索引。返回项会保留 `messageId`、`messageIdHex`、`rawUri`、`parseStatus`、`peer`、`vin`、`phone` 和完整 `metadataJson`；注册帧的 `metadataJson` 包含 `jt808.register.province`、`jt808.register.city`、`jt808.register.maker`、`jt808.register.deviceType`、`jt808.register.deviceId`、`jt808.register.plateColor`、`jt808.register.plate`、`identitySource`、`identityResolved` 等字段。 + GB32960 链路已用正式平台登录和合成实时帧验证： - 正式平台 `115.29.187.205` 已连接 TCP `32960`，服务收到 `0x05 PLATFORM_LOGIN` 并返回 ACK；日志显示账号 `Hyundai`、策略 `ALLOW`。 diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java new file mode 100644 index 00000000..2bf059b4 --- /dev/null +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java @@ -0,0 +1,177 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.http.HttpStatus; +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 org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Locale; + +@RestController +@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true") +@ConditionalOnBean(TdengineHistoryReader.class) +@RequestMapping("/api/event-history/jt808") +@Tag(name = "jt-808-raw-frame-controller", description = "JT808 原始帧索引分页查询接口。") +public final class Jt808RawFrameHistoryController { + + private final TdengineHistoryReader reader; + + public Jt808RawFrameHistoryController(TdengineHistoryReader reader) { + if (reader == null) { + throw new IllegalArgumentException("reader must not be null"); + } + this.reader = reader; + } + + @GetMapping("/raw-frames") + @Operation( + summary = "查询 JT808 原始帧索引", + description = "按 phone、VIN、vehicleKey、messageId、parseStatus 和时间范围查询 TDengine raw_frames。" + + "返回完整 metadataJson，包含注册帧字段、身份解析来源、rawArchiveUri 等排障信息。" + + "分页使用 cursorTs + cursorId，不使用 offset。") + public RawFramePageResponse rawFrames( + @Parameter(description = "JT808 终端手机号或终端标识。", example = "13079963320") + @RequestParam(required = false) String phone, + @Parameter(description = "VIN。VIN 已反写后可用该字段查询。", example = "LNVIN000000000001") + @RequestParam(required = false) String vin, + @Parameter(description = "内部车辆键；传入后精确过滤 vehicle_key。", example = "jt808:13079963320") + @RequestParam(required = false) String vehicleKey, + @Parameter(description = "JT808 消息 ID，支持十进制，例如注册帧 256、位置帧 512。", example = "256") + @RequestParam(required = false) Integer messageId, + @Parameter(description = "解析状态，例如 SUCCEEDED、FAILED。", example = "SUCCEEDED") + @RequestParam(required = false) String parseStatus, + @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00") + @RequestParam String dateFrom, + @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00") + @RequestParam String dateTo, + @Parameter(description = "排序方向。", example = "DESC") + @RequestParam(defaultValue = "DESC") TdengineQueryOrder order, + @Parameter(description = "返回原始帧数量上限，最大 1000。", example = "100") + @RequestParam(defaultValue = "100") int limit, + @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z") + @RequestParam(required = false) String cursorTs, + @Parameter(description = "上一页返回的 nextCursor.id。", example = "jt808-frame-xxx") + @RequestParam(required = false) String cursorId) throws IOException { + if (allBlank(phone, vin, vehicleKey) && messageId == null && isBlank(parseStatus)) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, + "phone, vin, vehicleKey, messageId or parseStatus is required for jt808 raw frame query"); + } + QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo); + TdenginePage page = reader.queryRawFrames(new TdengineRawFrameQuery( + "JT808", + normalize(vehicleKey), + normalize(vin), + normalize(phone), + messageId, + normalize(parseStatus).toUpperCase(Locale.ROOT), + range.eventTimeFrom(), + range.eventTimeTo().plusMillis(1), + order, + limit, + cursor(cursorTs, cursorId))); + return RawFramePageResponse.from(page); + } + + private static TdenginePageCursor cursor(String cursorTs, String cursorId) { + boolean hasTs = cursorTs != null && !cursorTs.isBlank(); + boolean hasId = cursorId != null && !cursorId.isBlank(); + if (!hasTs && !hasId) { + return null; + } + if (!hasTs || !hasId) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together"); + } + return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim()); + } + + private static boolean allBlank(String... values) { + for (String value : values) { + if (!isBlank(value)) { + return false; + } + } + return true; + } + + private static boolean isBlank(String value) { + return value == null || value.isBlank(); + } + + private static String normalize(String value) { + return value == null ? "" : value.trim(); + } + + public record RawFramePageResponse( + List items, + CursorResponse nextCursor) { + + private static RawFramePageResponse from(TdenginePage page) { + CursorResponse next = page.nextCursor() + .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker())) + .orElse(null); + return new RawFramePageResponse( + page.items().stream().map(RawFrameResponse::from).toList(), + next); + } + } + + public record CursorResponse(String ts, String id) { + } + + public record RawFrameResponse( + String eventTime, + String receivedAt, + String frameId, + int messageId, + String messageIdHex, + int subType, + String rawUri, + String checksum, + long rawSizeBytes, + String parseStatus, + String parseError, + String peer, + String metadataJson, + String protocol, + String vehicleKey, + String vin, + String phone) { + + private static RawFrameResponse from(TdengineRawFrameRow row) { + return new RawFrameResponse( + row.ts().toString(), + row.receivedAt().toString(), + row.frameId(), + row.messageId(), + "0x%04X".formatted(row.messageId()), + row.subType(), + row.rawUri(), + row.checksum(), + row.rawSizeBytes(), + row.parseStatus(), + row.parseError(), + row.peer(), + row.metadataJson(), + row.protocol(), + row.vehicleKey(), + row.vin(), + row.phone()); + } + } +} 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 0251a87a..cf481fd5 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 @@ -8,6 +8,7 @@ import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; +import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController; import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController; import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; @@ -106,6 +107,13 @@ public class EventHistoryAutoConfiguration { return new Jt808LocationHistoryController(reader); } + @Bean + @ConditionalOnBean(TdengineHistoryReader.class) + @ConditionalOnMissingBean + public Jt808RawFrameHistoryController jt808RawFrameHistoryController(TdengineHistoryReader reader) { + return new Jt808RawFrameHistoryController(reader); + } + @Bean @ConditionalOnBean(TdengineHistoryReader.class) @ConditionalOnMissingBean diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java new file mode 100644 index 00000000..8103a5f0 --- /dev/null +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java @@ -0,0 +1,129 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery; +import com.lingniu.ingest.tdenginehistory.TdengineLocationRow; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class Jt808RawFrameHistoryControllerTest { + + @Test + void queriesRawFramesByPhoneMessageIdAndCursor() throws Exception { + CapturingReader reader = new CapturingReader(new TdenginePage<>( + List.of(row("frame-1", "2026-06-29T12:00:01Z")), + Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T12:00:01Z"), "frame-1")))); + Jt808RawFrameHistoryController controller = new Jt808RawFrameHistoryController(reader); + + Jt808RawFrameHistoryController.RawFramePageResponse response = controller.rawFrames( + "13079963320", + null, + null, + 0x0100, + null, + "2026-06-29T00:00:00+08:00", + "2026-06-29T23:59:59+08:00", + TdengineQueryOrder.DESC, + 20, + "2026-06-29T12:00:00Z", + "frame-0"); + + assertThat(reader.query.protocol()).isEqualTo("JT808"); + assertThat(reader.query.vehicleKey()).isBlank(); + assertThat(reader.query.phone()).isEqualTo("13079963320"); + assertThat(reader.query.messageId()).isEqualTo(0x0100); + assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor( + Instant.parse("2026-06-29T12:00:00Z"), "frame-0")); + assertThat(response.items()).hasSize(1); + Jt808RawFrameHistoryController.RawFrameResponse item = response.items().getFirst(); + assertThat(item.messageId()).isEqualTo(0x0100); + assertThat(item.messageIdHex()).isEqualTo("0x0100"); + assertThat(item.metadataJson()).contains("jt808.register.deviceId"); + assertThat(response.nextCursor()).isEqualTo(new Jt808RawFrameHistoryController.CursorResponse( + "2026-06-29T12:00:01Z", "frame-1")); + } + + @Test + void rejectsUnboundedRawFrameQuery() { + Jt808RawFrameHistoryController controller = new Jt808RawFrameHistoryController( + new CapturingReader(new TdenginePage<>(List.of(), Optional.empty()))); + + assertThatThrownBy(() -> controller.rawFrames( + null, + null, + null, + null, + null, + "2026-06-29", + "2026-06-29", + TdengineQueryOrder.DESC, + 20, + null, + null)) + .isInstanceOfSatisfying(ResponseStatusException.class, ex -> + assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST)); + } + + private static TdengineRawFrameRow row(String frameId, String ts) { + Instant instant = Instant.parse(ts); + return new TdengineRawFrameRow( + instant, + frameId, + instant.plusMillis(500), + 0x0100, + 0, + instant, + "archive://2026/06/29/JT808/unknown/" + frameId + ".bin", + "sha256:" + frameId, + 72, + "SUCCEEDED", + "", + "222.66.200.68:41000", + "{\"jt808.register.deviceId\":\"9963320\",\"jt808.register.plate\":\"沪A61559F\"}", + "JT808", + "jt808:13079963320", + "unknown", + "13079963320"); + } + + private static final class CapturingReader implements TdengineHistoryReader { + private final TdenginePage response; + private TdengineRawFrameQuery query; + + private CapturingReader(TdenginePage response) { + this.response = response; + } + + @Override + public TdenginePage queryRawFrames(TdengineRawFrameQuery query) throws IOException { + this.query = query; + return response; + } + + @Override + public TdenginePage queryLocations(TdengineLocationQuery query) { + throw new UnsupportedOperationException(); + } + + @Override + public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { + throw new UnsupportedOperationException(); + } + } +} 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 3cf8cba1..c643f54b 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 @@ -8,6 +8,7 @@ import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; +import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController; import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController; import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; @@ -91,6 +92,14 @@ class EventHistoryAutoConfigurationTest { .run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class)); } + @Test + void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() { + contextRunner + .withPropertyValues("lingniu.ingest.event-history.enabled=true") + .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class)) + .run(context -> assertThat(context).hasSingleBean(Jt808RawFrameHistoryController.class)); + } + @Test void createsTelemetryFieldHistoryControllerWhenTdengineReaderExists() { contextRunner diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java index d9b58694..3555119c 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java @@ -25,9 +25,12 @@ public final class TdengineHistoryQueries { } public TdengineQueryStatement rawFrames(TdengineRawFrameQuery query) { - String table = schema.rawFrameTable(query.protocol(), query.vehicleKey()); - return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(), - query.order(), query.limit(), query.cursor()); + if (query.childTableQuery()) { + String table = schema.rawFrameTable(query.protocol(), query.vehicleKey()); + return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(), + query.order(), query.limit(), query.cursor()); + } + return rawFramesStable(query); } public TdengineQueryStatement locations(TdengineLocationQuery query) { @@ -71,4 +74,43 @@ public final class TdengineHistoryQueries { values.add(limit); return new TdengineQueryStatement(sql.toString(), values); } + + private TdengineQueryStatement rawFramesStable(TdengineRawFrameQuery query) { + List values = new ArrayList<>(); + StringBuilder sql = new StringBuilder(384) + .append("SELECT ").append(RAW_COLUMNS) + .append(" FROM ").append(schema.rawFramesStableTable()) + .append(" WHERE ts >= ? AND ts < ?"); + values.add(query.from()); + values.add(query.to()); + addFilter(sql, values, "protocol", query.protocol()); + addFilter(sql, values, "vehicle_key", query.vehicleKey()); + addFilter(sql, values, "vin", query.vin()); + addFilter(sql, values, "phone", query.phone()); + if (query.messageId() != null) { + sql.append(" AND message_id = ?"); + values.add(query.messageId()); + } + addFilter(sql, values, "parse_status", query.parseStatus()); + if (query.cursor() != null) { + String op = query.order() == TdengineQueryOrder.ASC ? ">" : "<"; + sql.append(" AND (ts ").append(op).append(" ? OR (ts = ? AND frame_id ") + .append(op).append(" ?))"); + values.add(query.cursor().ts()); + values.add(query.cursor().ts()); + values.add(query.cursor().tieBreaker()); + } + sql.append(" ORDER BY ts ").append(query.order().name()) + .append(", frame_id ").append(query.order().name()) + .append(" LIMIT ?"); + values.add(query.limit()); + return new TdengineQueryStatement(sql.toString(), values); + } + + private static void addFilter(StringBuilder sql, List values, String column, String value) { + if (value != null && !value.isBlank()) { + sql.append(" AND ").append(column).append(" = ?"); + values.add(value); + } + } } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java index cb786aae..b51d0829 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java @@ -27,6 +27,10 @@ public final class TdengineHistorySchema { return "raw_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey); } + public String rawFramesStableTable() { + return "raw_frames"; + } + public String locationTable(String protocol, String vehicleKey) { return "loc_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey); } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java index 5bffc337..87f040c4 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java @@ -5,27 +5,56 @@ import java.time.Instant; public record TdengineRawFrameQuery( String protocol, String vehicleKey, + String vin, + String phone, + Integer messageId, + String parseStatus, Instant from, Instant to, TdengineQueryOrder order, int limit, TdenginePageCursor cursor ) { + public TdengineRawFrameQuery(String protocol, + String vehicleKey, + Instant from, + Instant to, + TdengineQueryOrder order, + int limit, + TdenginePageCursor cursor) { + this(protocol, vehicleKey, "", "", null, "", from, to, order, limit, cursor); + } + public TdengineRawFrameQuery { if (protocol == null || protocol.isBlank()) { throw new IllegalArgumentException("protocol must not be blank"); } - if (vehicleKey == null || vehicleKey.isBlank()) { - throw new IllegalArgumentException("vehicleKey must not be blank"); - } if (from == null || to == null || !from.isBefore(to)) { throw new IllegalArgumentException("query time range must be valid"); } + protocol = protocol.trim().toUpperCase(); + vehicleKey = trim(vehicleKey); + vin = trim(vin); + phone = trim(phone); + parseStatus = trim(parseStatus); order = order == null ? TdengineQueryOrder.DESC : order; limit = Math.max(1, Math.min(limit, 1001)); } TdengineRawFrameQuery withLimit(int newLimit) { - return new TdengineRawFrameQuery(protocol, vehicleKey, from, to, order, newLimit, cursor); + return new TdengineRawFrameQuery(protocol, vehicleKey, vin, phone, messageId, parseStatus, + from, to, order, newLimit, cursor); + } + + boolean childTableQuery() { + return !vehicleKey.isBlank() + && vin.isBlank() + && phone.isBlank() + && messageId == null + && parseStatus.isBlank(); + } + + private static String trim(String value) { + return value == null ? "" : value.trim(); } } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java index ac04677d..ee592efd 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java @@ -37,6 +37,41 @@ class TdengineHistoryQueriesTest { 101); } + @Test + void rawFrameQueryUsesStableWhenPhoneMessageIdOrParseStatusFiltersArePresent() { + TdengineRawFrameQuery query = new TdengineRawFrameQuery( + "JT808", + "", + "", + "13079963320", + 0x0100, + "SUCCEEDED", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.DESC, + 50, + new TdenginePageCursor(Instant.parse("2026-06-29T12:00:00Z"), "frame-100")); + + TdengineQueryStatement statement = queries.rawFrames(query); + + assertThat(statement.sql()) + .startsWith("SELECT ts, frame_id, received_at, message_id") + .contains(" FROM raw_frames") + .contains(" WHERE ts >= ? AND ts < ?") + .contains(" AND protocol = ?") + .contains(" AND phone = ?") + .contains(" AND message_id = ?") + .contains(" AND parse_status = ?") + .contains(" AND (ts < ? OR (ts = ? AND frame_id < ?))") + .contains(" ORDER BY ts DESC, frame_id DESC LIMIT ?"); + assertThat(statement.values()) + .containsExactly( + query.from(), query.to(), + "JT808", "13079963320", 0x0100, "SUCCEEDED", + query.cursor().ts(), query.cursor().ts(), query.cursor().tieBreaker(), + 50); + } + @Test void locationQueryCapsLimitAndUsesDescendingCursor() { TdengineLocationQuery query = new TdengineLocationQuery(