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 index fc660e0e..4f6fe5b3 100644 --- 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 @@ -1,7 +1,11 @@ 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.eventfilestore.EventFileRecord; +import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.util.JsonFormat; @@ -21,20 +25,24 @@ import java.util.Map; 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()) { - throw new IllegalArgumentException("envelope telemetry_snapshot is required"); + 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 = new LinkedHashMap<>(envelope.getMetadataMap()); - if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) { - // 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。 - metadata.putIfAbsent("originalSource", envelope.getSource()); - } + Map metadata = metadata(envelope, protocol); return new EventFileRecord( envelope.getEventId(), protocol, @@ -48,6 +56,40 @@ public final class TelemetryEnvelopeRecordMapper { ); } + private static EventFileRecord rawArchiveRecord(VehicleEnvelope envelope) { + ProtocolId protocol = protocol(envelope.getSource()); + RawArchiveRef rawArchive = envelope.getRawArchive(); + Map metadata = metadata(envelope, protocol); + String rawArchiveUri = rawArchive.getUri(); + metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId()); + 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) + ); + } + + 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; @@ -66,4 +108,26 @@ public final class TelemetryEnvelopeRecordMapper { throw new IllegalArgumentException("failed to serialize telemetry_snapshot", e); } } + + private static String rawArchiveJson(VehicleEnvelope envelope, Map metadata) { + 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", rawArchive.getUri()); + payload.put("rawSizeBytes", rawArchive.getSizeBytes()); + payload.put("metadata", metadata); + try { + return OBJECT_MAPPER.writeValueAsString(payload); + } catch (JsonProcessingException e) { + throw new IllegalArgumentException("failed to serialize raw_archive", e); + } + } } 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 07ee5b6c..b391f9dd 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 @@ -5,6 +5,7 @@ import com.lingniu.ingest.api.consumer.EnvelopeIngestor; import com.lingniu.ingest.eventfilestore.EventFileQuery; import com.lingniu.ingest.eventfilestore.EventFileRecord; import com.lingniu.ingest.eventfilestore.EventFileStore; +import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.TelemetryField; import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; @@ -55,6 +56,43 @@ class EventHistoryEnvelopeIngestorTest { assertThat(store.records).isEmpty(); } + @Test + void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() { + CapturingStore store = new CapturingStore(); + EventHistoryEnvelopeIngestor ingestor = + new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper()); + + 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("archiveKey", "gb32960/2026/06/23/raw-event-1.bin") + .setRawArchive(RawArchiveRef.newBuilder() + .setUri("archive://gb32960/2026/06/23/raw-event-1.bin") + .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.eventType()).isEqualTo("RAW_ARCHIVE"); + assertThat(record.vin()).isEqualTo("VINRAW001"); + assertThat(record.rawArchiveUri()).isEqualTo("archive://gb32960/2026/06/23/raw-event-1.bin"); + assertThat(record.metadata()) + .containsEntry("protocolVersion", "V2016") + .containsEntry("archiveKey", "gb32960/2026/06/23/raw-event-1.bin"); + assertThat(record.payloadJson()).contains("\"rawSizeBytes\":128"); + assertThat(record.payloadJson()).doesNotContain("rawBytes"); + } + private static VehicleEnvelope envelope(String eventId) { return VehicleEnvelope.newBuilder() .setEventId(eventId) diff --git a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java index bf716ee2..111022f1 100644 --- a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java +++ b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java @@ -71,8 +71,7 @@ public final class EnvelopeMapper { int size = ra.rawBytes() == null ? 0 : ra.rawBytes().length; String key = rawArchiveKey(ra); String uri = rawArchiveUri(ra, key); - // RAW envelope 只携带引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。 - // 注意 KafkaEventSink 当前仍过滤 RawArchive；这段用于未来放开 raw 引用转发或消费者测试。 + // RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。 b.putMetadata(RawArchiveKeys.META_KEY, key); b.putMetadata(RawArchiveKeys.META_URI, uri); b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());