feat: add telemetry field history api

2026-06-29 13:59:06 +08:00
parent 90342fe5bf
commit 192f115176
4 changed files with 358 additions and 0 deletions
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java
@@ -0,0 +1,189 @@
+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.TdengineTelemetryFieldQuery;
+import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
+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/telemetry")
+@Tag(name = "telemetry-field-history-controller", description = "遥测字段历史分页查询接口。")
+public final class TelemetryFieldHistoryController {
+
+    private final TdengineHistoryReader reader;
+
+    public TelemetryFieldHistoryController(TdengineHistoryReader reader) {
+        if (reader == null) {
+            throw new IllegalArgumentException("reader must not be null");
+        }
+        this.reader = reader;
+    }
+
+    @GetMapping("/fields")
+    @Operation(
+            summary = "查询单车单字段历史值",
+            description = "按协议、车辆标识、字段 key 和时间范围查询 TDengine telemetry_fields。分页使用 cursorTs + cursorId，不使用 offset。")
+    public TelemetryFieldPageResponse fields(
+            @Parameter(description = "协议名，例如 GB32960 或 JT808。", required = true, example = "GB32960")
+            @RequestParam String protocol,
+            @Parameter(description = "字段 key，例如 FUEL_CELL_V2016.hydrogenComsuxxx。", required = true)
+            @RequestParam String fieldKey,
+            @Parameter(description = "内部车辆键。传入后优先使用。", example = "LNVFC000000000001")
+            @RequestParam(required = false) String vehicleKey,
+            @Parameter(description = "VIN。GB32960 常用该字段作为车辆键。", example = "LNVFC000000000001")
+            @RequestParam(required = false) String vin,
+            @Parameter(description = "JT808 终端手机号或终端标识；JT808 会自动映射为 jt808:<phone>。", example = "g7gps")
+            @RequestParam(required = false) String phone,
+            @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 = "event-xxx#0")
+            @RequestParam(required = false) String cursorId) throws IOException {
+        String normalizedProtocol = require(protocol, "protocol is required for telemetry field query").toUpperCase(Locale.ROOT);
+        String normalizedFieldKey = require(fieldKey, "fieldKey is required for telemetry field query");
+        QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
+        TdenginePage<TdengineTelemetryFieldRow> page = reader.queryTelemetryFields(new TdengineTelemetryFieldQuery(
+                normalizedProtocol,
+                resolveVehicleKey(normalizedProtocol, vehicleKey, vin, phone),
+                normalizedFieldKey,
+                range.eventTimeFrom(),
+                range.eventTimeTo().plusMillis(1),
+                order,
+                limit,
+                cursor(cursorTs, cursorId)));
+        return TelemetryFieldPageResponse.from(page);
+    }
+
+    private static String resolveVehicleKey(String protocol, String vehicleKey, String vin, String phone) {
+        String explicitVehicleKey = trimToNull(vehicleKey);
+        if (explicitVehicleKey != null) {
+            return explicitVehicleKey;
+        }
+        String vinValue = trimToNull(vin);
+        if (vinValue != null) {
+            return vinValue;
+        }
+        String phoneValue = trimToNull(phone);
+        if (phoneValue != null) {
+            if ("JT808".equals(protocol) && !phoneValue.startsWith("jt808:")) {
+                return "jt808:" + phoneValue;
+            }
+            return phoneValue;
+        }
+        throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vehicleKey, vin or phone is required");
+    }
+
+    private static String require(String value, String message) {
+        String trimmed = trimToNull(value);
+        if (trimmed == null) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
+        }
+        return trimmed;
+    }
+
+    private static String trimToNull(String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        return value.trim();
+    }
+
+    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());
+    }
+
+    public record TelemetryFieldPageResponse(
+            List<TelemetryFieldResponse> items,
+            CursorResponse nextCursor) {
+
+        private static TelemetryFieldPageResponse from(TdenginePage<TdengineTelemetryFieldRow> page) {
+            CursorResponse next = page.nextCursor()
+                    .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
+                    .orElse(null);
+            return new TelemetryFieldPageResponse(
+                    page.items().stream().map(TelemetryFieldResponse::from).toList(),
+                    next);
+        }
+    }
+
+    public record CursorResponse(String ts, String id) {
+    }
+
+    public record TelemetryFieldResponse(
+            String eventTime,
+            String receivedAt,
+            String factId,
+            String frameId,
+            String fieldKey,
+            String valueType,
+            String valueText,
+            Double valueDouble,
+            Long valueLong,
+            String unit,
+            String quality,
+            String sourcePath,
+            String rawUri,
+            String metadataJson,
+            String protocol,
+            String vehicleKey,
+            String vin,
+            String phone) {
+
+        private static TelemetryFieldResponse from(TdengineTelemetryFieldRow row) {
+            return new TelemetryFieldResponse(
+                    row.ts().toString(),
+                    row.receivedAt().toString(),
+                    row.factId(),
+                    row.frameId(),
+                    row.fieldKey(),
+                    row.valueType(),
+                    row.valueText(),
+                    row.valueDouble(),
+                    row.valueLong(),
+                    row.unit(),
+                    row.quality(),
+                    row.sourcePath(),
+                    row.rawUri(),
+                    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
@@ -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.TelemetryFieldHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
 import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
 import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
@@ -98,6 +99,13 @@ public class EventHistoryAutoConfiguration {
        return new Jt808LocationHistoryController(reader);
    }

+    @Bean
+    @ConditionalOnBean(TdengineHistoryReader.class)
+    @ConditionalOnMissingBean
+    public TelemetryFieldHistoryController telemetryFieldHistoryController(TdengineHistoryReader reader) {
+        return new TelemetryFieldHistoryController(reader);
+    }
+
    static Path archiveRoot(String value) {
        if (value == null || value.isBlank()) {
            return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java
@@ -0,0 +1,152 @@
+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 TelemetryFieldHistoryControllerTest {
+
+    @Test
+    void queriesGb32960FieldByVinWithKeysetCursor() throws Exception {
+        CapturingReader reader = new CapturingReader(new TdenginePage<>(
+                List.of(row("field-1", "2026-06-29T05:00:01Z")),
+                Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "field-1"))));
+        TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader);
+
+        TelemetryFieldHistoryController.TelemetryFieldPageResponse response = controller.fields(
+                "GB32960",
+                "FUEL_CELL_V2016.hydrogenComsuxxx",
+                null,
+                "LNVFC000000000001",
+                null,
+                "2026-06-29T13:00:00+08:00",
+                "2026-06-29T13:10:00+08:00",
+                TdengineQueryOrder.ASC,
+                100,
+                "2026-06-29T05:00:00Z",
+                "field-0");
+
+        assertThat(reader.query.protocol()).isEqualTo("GB32960");
+        assertThat(reader.query.vehicleKey()).isEqualTo("LNVFC000000000001");
+        assertThat(reader.query.fieldKey()).isEqualTo("FUEL_CELL_V2016.hydrogenComsuxxx");
+        assertThat(reader.query.order()).isEqualTo(TdengineQueryOrder.ASC);
+        assertThat(reader.query.limit()).isEqualTo(100);
+        assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor(
+                Instant.parse("2026-06-29T05:00:00Z"), "field-0"));
+        assertThat(response.items())
+                .extracting(TelemetryFieldHistoryController.TelemetryFieldResponse::factId)
+                .containsExactly("field-1");
+        assertThat(response.items().getFirst().valueDouble()).isEqualTo(3.14);
+        assertThat(response.nextCursor()).isEqualTo(new TelemetryFieldHistoryController.CursorResponse(
+                "2026-06-29T05:00:01Z", "field-1"));
+    }
+
+    @Test
+    void mapsJt808PhoneToVehicleKeyWhenVehicleKeyIsAbsent() throws Exception {
+        CapturingReader reader = new CapturingReader(new TdenginePage<>(List.of(), Optional.empty()));
+        TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader);
+
+        controller.fields(
+                "JT808",
+                "location.speed_kmh",
+                null,
+                null,
+                "g7gps",
+                "2026-06-29",
+                "2026-06-29",
+                TdengineQueryOrder.DESC,
+                50,
+                null,
+                null);
+
+        assertThat(reader.query.vehicleKey()).isEqualTo("jt808:g7gps");
+    }
+
+    @Test
+    void rejectsMissingVehicleIdentity() {
+        TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(
+                new CapturingReader(new TdenginePage<>(List.of(), Optional.empty())));
+
+        assertThatThrownBy(() -> controller.fields(
+                "GB32960",
+                "vehicle.speed",
+                null,
+                null,
+                null,
+                "2026-06-29",
+                "2026-06-29",
+                TdengineQueryOrder.DESC,
+                50,
+                null,
+                null))
+                .isInstanceOfSatisfying(ResponseStatusException.class, ex ->
+                        assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST));
+    }
+
+    private static TdengineTelemetryFieldRow row(String factId, String ts) {
+        Instant instant = Instant.parse(ts);
+        return new TdengineTelemetryFieldRow(
+                instant,
+                factId,
+                "frame-1",
+                instant.plusMillis(500),
+                "FUEL_CELL_V2016.hydrogenComsuxxx",
+                "DOUBLE",
+                "3.14",
+                3.14,
+                null,
+                "kg/100km",
+                "GOOD",
+                "FUEL_CELL_V2016.hydrogenComsuxxx",
+                "archive://gb32960/frame-1.bin",
+                "{\"cmd\":\"REALTIME_REPORT\"}",
+                "GB32960",
+                "LNVFC000000000001",
+                "LNVFC000000000001",
+                "");
+    }
+
+    private static final class CapturingReader implements TdengineHistoryReader {
+        private final TdenginePage<TdengineTelemetryFieldRow> response;
+        private TdengineTelemetryFieldQuery query;
+
+        private CapturingReader(TdenginePage<TdengineTelemetryFieldRow> response) {
+            this.response = response;
+        }
+
+        @Override
+        public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TdenginePage<TdengineTelemetryFieldRow> queryTelemetryFields(TdengineTelemetryFieldQuery query)
+                throws IOException {
+            this.query = query;
+            return response;
+        }
+    }
+}
--- 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.TelemetryFieldHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
 import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
 import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
@@ -73,6 +74,14 @@ class EventHistoryAutoConfigurationTest {
                .run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class));
    }

+    @Test
+    void createsTelemetryFieldHistoryControllerWhenTdengineReaderExists() {
+        contextRunner
+                .withPropertyValues("lingniu.ingest.event-history.enabled=true")
+                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
+                .run(context -> assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class));
+    }
+
    @Test
    void parsesFileUriArchiveRoot() {
        assertThat(EventHistoryAutoConfiguration.archiveRoot("file:///tmp/lingniu-test-archive"))