feat: expose generic raw frame query

2026-06-30 10:22:05 +08:00
parent b977de5519
commit 68cd3d0d06
5 changed files with 346 additions and 0 deletions
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java
@@ -0,0 +1,194 @@
+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")
+@Tag(name = "Raw Frame API", description = "原始帧索引分页查询接口。")
+public final class RawFrameHistoryController {
+
+    private final TdengineHistoryReader reader;
+
+    public RawFrameHistoryController(TdengineHistoryReader reader) {
+        if (reader == null) {
+            throw new IllegalArgumentException("reader must not be null");
+        }
+        this.reader = reader;
+    }
+
+    @GetMapping("/raw-frames")
+    @Operation(
+            summary = "查询原始帧索引",
+            description = "按 protocol、vehicleKey、VIN、phone、messageId、parseStatus 和时间范围查询 TDengine raw_frames。"
+                    + "返回 metadataJson 与 rawUri，分页使用 cursorTs + cursorId。")
+    public RawFramePageResponse rawFrames(
+            @Parameter(description = "协议类型，例如 GB32960、JT808、MQTT_YUTONG、XINDA_PUSH。", example = "JT808")
+            @RequestParam String protocol,
+            @Parameter(description = "内部车辆键；传入后精确过滤 vehicle_key。", example = "jt808:g7gps")
+            @RequestParam(required = false) String vehicleKey,
+            @Parameter(description = "VIN。已完成身份反写的 raw 帧可用该字段查询。", example = "LB9A32A24P0LS1270")
+            @RequestParam(required = false) String vin,
+            @Parameter(description = "终端手机号或终端标识，主要用于 JT808。", example = "g7gps")
+            @RequestParam(required = false) String phone,
+            @Parameter(description = "消息 ID，支持十进制或 0x 十六进制，例如 512 或 0x0200。", example = "0x0200")
+            @RequestParam(required = false) String messageId,
+            @Parameter(description = "解析状态，例如 SUCCEEDED、FAILED、NOT_PARSED。", 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 = "frame-xxx")
+            @RequestParam(required = false) String cursorId) throws IOException {
+        Integer parsedMessageId = parseMessageId(messageId);
+        if (allBlank(vehicleKey, vin, phone) && parsedMessageId == null && isBlank(parseStatus)) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST,
+                    "vehicleKey, vin, phone, messageId or parseStatus is required for raw frame query");
+        }
+        QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
+        TdenginePage<TdengineRawFrameRow> page = reader.queryRawFrames(new TdengineRawFrameQuery(
+                normalize(protocol).toUpperCase(Locale.ROOT),
+                normalize(vehicleKey),
+                normalize(vin),
+                normalize(phone),
+                parsedMessageId,
+                normalize(parseStatus).toUpperCase(Locale.ROOT),
+                range.eventTimeFrom(),
+                range.eventTimeTo().plusMillis(1),
+                order,
+                limit,
+                cursor(cursorTs, cursorId)));
+        return RawFramePageResponse.from(page);
+    }
+
+    private static Integer parseMessageId(String messageId) {
+        String value = normalize(messageId);
+        if (value.isBlank()) {
+            return null;
+        }
+        try {
+            if (value.startsWith("0x") || value.startsWith("0X")) {
+                return Integer.parseInt(value.substring(2), 16);
+            }
+            return Integer.valueOf(value);
+        } catch (NumberFormatException ex) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "messageId must be decimal or 0x hex", ex);
+        }
+    }
+
+    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<RawFrameResponse> items,
+            CursorResponse nextCursor) {
+
+        private static RawFramePageResponse from(TdenginePage<TdengineRawFrameRow> 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());
+        }
+    }
+}
--- 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
@@ -10,6 +10,7 @@ import com.lingniu.ingest.eventhistory.Gb32960FrameController;
 import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
 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;
@@ -109,6 +110,13 @@ public class EventHistoryAutoConfiguration {
        return new LocationHistoryController(reader);
    }

+    @Bean
+    @ConditionalOnBean(TdengineHistoryReader.class)
+    @ConditionalOnMissingBean
+    public RawFrameHistoryController rawFrameHistoryController(TdengineHistoryReader reader) {
+        return new RawFrameHistoryController(reader);
+    }
+
    @Bean
    @ConditionalOnBean(TdengineHistoryReader.class)
    @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java
@@ -0,0 +1,132 @@
+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 RawFrameHistoryControllerTest {
+
+    @Test
+    void queriesRawFramesByProtocolHexMessageIdAndCursor() 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"))));
+        RawFrameHistoryController controller = new RawFrameHistoryController(reader);
+
+        RawFrameHistoryController.RawFramePageResponse response = controller.rawFrames(
+                "xinda_push",
+                null,
+                "LB9A32A24P0LS1270",
+                null,
+                "0x0200",
+                "not_parsed",
+                "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("XINDA_PUSH");
+        assertThat(reader.query.vin()).isEqualTo("LB9A32A24P0LS1270");
+        assertThat(reader.query.messageId()).isEqualTo(0x0200);
+        assertThat(reader.query.parseStatus()).isEqualTo("NOT_PARSED");
+        assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor(
+                Instant.parse("2026-06-29T12:00:00Z"), "frame-0"));
+        assertThat(response.items()).hasSize(1);
+        RawFrameHistoryController.RawFrameResponse item = response.items().getFirst();
+        assertThat(item.messageId()).isEqualTo(0x0200);
+        assertThat(item.messageIdHex()).isEqualTo("0x0200");
+        assertThat(item.rawUri()).contains("XINDA_PUSH");
+        assertThat(item.metadataJson()).contains("rawArchiveUri");
+        assertThat(response.nextCursor()).isEqualTo(new RawFrameHistoryController.CursorResponse(
+                "2026-06-29T12:00:01Z", "frame-1"));
+    }
+
+    @Test
+    void rejectsRawFrameQueryWithoutAnyNarrowingFilter() {
+        RawFrameHistoryController controller = new RawFrameHistoryController(
+                new CapturingReader(new TdenginePage<>(List.of(), Optional.empty())));
+
+        assertThatThrownBy(() -> controller.rawFrames(
+                "JT808",
+                null,
+                null,
+                null,
+                null,
+                null,
+                "2026-06-29",
+                "2026-06-30",
+                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),
+                0x0200,
+                0,
+                instant,
+                "archive://2026/06/29/XINDA_PUSH/unknown/" + frameId + ".bin",
+                "sha256:" + frameId,
+                72,
+                "NOT_PARSED",
+                "",
+                "xinda",
+                "{\"rawArchiveUri\":\"archive://2026/06/29/XINDA_PUSH/unknown/" + frameId + ".bin\"}",
+                "XINDA_PUSH",
+                "LB9A32A24P0LS1270",
+                "LB9A32A24P0LS1270",
+                "");
+    }
+
+    private static final class CapturingReader implements TdengineHistoryReader {
+        private final TdenginePage<TdengineRawFrameRow> response;
+        private TdengineRawFrameQuery query;
+
+        private CapturingReader(TdenginePage<TdengineRawFrameRow> response) {
+            this.response = response;
+        }
+
+        @Override
+        public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) throws IOException {
+            this.query = query;
+            return response;
+        }
+
+        @Override
+        public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TdenginePage<TdengineTelemetryFieldRow> queryTelemetryFields(TdengineTelemetryFieldQuery query) {
+            throw new UnsupportedOperationException();
+        }
+    }
+}
--- 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
@@ -10,6 +10,7 @@ import com.lingniu.ingest.eventhistory.Gb32960FrameController;
 import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
 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;
@@ -104,12 +105,21 @@ class EventHistoryAutoConfigurationTest {
                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
                .run(context -> {
                    assertThat(context).hasSingleBean(LocationHistoryController.class);
+                    assertThat(context).hasSingleBean(RawFrameHistoryController.class);
                    assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class);
                    assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
                    assertThat(context).doesNotHaveBean(Jt808RawFrameHistoryController.class);
                });
    }

+    @Test
+    void createsGenericRawFrameHistoryControllerWhenTdengineReaderExists() {
+        contextRunner
+                .withPropertyValues("lingniu.ingest.event-history.enabled=true")
+                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
+                .run(context -> assertThat(context).hasSingleBean(RawFrameHistoryController.class));
+    }
+
    @Test
    void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() {
        contextRunner