> **Superseded:** This 2026-06-29 phase-1 implementation plan is historical
> context. Use `docs/target-architecture.md` and
> `docs/superpowers/specs/2026-06-29-vehicle-ingest-redesign.md` for the
> current production architecture: raw bytes are owned by `sink-archive`,
> history reads come from TDengine, and the former `raw-archive-store`
> prototype has been removed from the current build surface.

# Vehicle Ingest Redesign Phase 1 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build the foundation for the new GB32960/JT808 architecture: shared fact model, raw archive contract, and byte-light Kafka schema.

**Architecture:** Phase 1 adds new modules beside existing runtime paths. It does not change current Netty handlers, ACK behavior, or TDengine writes yet. The output is a tested set of immutable facts and contracts that later phases will wire into protocol apps and the history writer.

**Tech Stack:** Java 25, Maven, JUnit 5, AssertJ, Protobuf 4, existing `ProtocolId`.

---

## Scope

This plan implements only the first slice of the redesign:

- `modules/core/ingest-facts`: protocol-neutral fact records and helpers.
- `modules/sinks/raw-archive-store`: raw archive write/read contract and local URI-safe implementation skeleton.
- `modules/sinks/sink-kafka`: protobuf messages for raw frame facts and decoded facts.
- Parent Maven wiring and focused tests.

It intentionally does not yet modify:

- `gb32960-ingest-app`
- `jt808-ingest-app`
- existing `Dispatcher`
- existing `VehicleEvent`
- existing `vehicle-history-app` TDengine writer

Those changes belong to later implementation plans.

## File Structure

Create:

- `modules/core/ingest-facts/pom.xml`: Maven module for fact model.
- `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/ParseStatus.java`: parse status enum.
- `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/VehicleKey.java`: stable vehicle key derivation.
- `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameFact.java`: raw frame fact record.
- `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/DecodedFact.java`: sealed decoded fact hierarchy.
- `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/FactIds.java`: deterministic id helpers.
- `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/VehicleKeyTest.java`
- `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameFactTest.java`
- `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/FactIdsTest.java`
- `modules/sinks/raw-archive-store/pom.xml`: Maven module for raw archive contracts.
- `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriteRequest.java`
- `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReceipt.java`
- `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriter.java`
- `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReader.java`
- `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStore.java`
- `modules/sinks/raw-archive-store/src/test/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStoreTest.java`

Modify:

- `pom.xml`: add modules and dependency management entries.
- `modules/sinks/sink-kafka/pom.xml`: add dependency on `ingest-facts`.
- `modules/sinks/sink-kafka/src/main/proto/vehicle_envelope.proto`: add `RawFrameFactPayload`, `DecodedFactPayload`, and related enums/messages.
- `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/VehicleEnvelopeProtoCompatibilityTest.java`: verify new proto messages can round-trip.

---

### Task 1: Add `ingest-facts` Maven Module

**Files:**

- Modify: `pom.xml`
- Create: `modules/core/ingest-facts/pom.xml`
- Test later in Task 2 through Task 4.

- [ ] **Step 1: Add failing Maven target check**

Run:

```bash
mvn -pl modules/core/ingest-facts test
```

Expected: FAIL because Maven cannot find the selected project.

- [ ] **Step 2: Add module to root `pom.xml`**

Add this module after `modules/core/ingest-api`:

```xml
<module>modules/core/ingest-facts</module>
```

Add this dependency management entry after `ingest-api`:

```xml
<dependency>
    <groupId>com.lingniu.ingest</groupId>
    <artifactId>ingest-facts</artifactId>
    <version>${project.version}</version>
</dependency>
```

- [ ] **Step 3: Create module POM**

Create `modules/core/ingest-facts/pom.xml`:

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.lingniu.ingest</groupId>
        <artifactId>lingniu-vehicle-ingest</artifactId>
        <version>0.1.0-SNAPSHOT</version>
        <relativePath>../../../pom.xml</relativePath>
    </parent>
    <artifactId>ingest-facts</artifactId>
    <name>ingest-facts</name>
    <description>Protocol-neutral raw frame and decoded fact model for vehicle ingestion.</description>

    <dependencies>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>ingest-api</artifactId>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>
```

- [ ] **Step 4: Run Maven target check**

Run:

```bash
mvn -pl modules/core/ingest-facts test
```

Expected: PASS with no tests.

- [ ] **Step 5: Commit**

```bash
git add pom.xml modules/core/ingest-facts/pom.xml
git commit -m "build: add ingest facts module"
```

---

### Task 2: Implement Vehicle Key Derivation

**Files:**

- Create: `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/VehicleKey.java`
- Create: `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/VehicleKeyTest.java`

- [ ] **Step 1: Write failing tests**

Create `VehicleKeyTest.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

class VehicleKeyTest {

    @Test
    void usesVinForGb32960() {
        assertThat(VehicleKey.derive(ProtocolId.GB32960, " LB9A32A20P0LS1257 ", "", "abc"))
                .isEqualTo("LB9A32A20P0LS1257");
    }

    @Test
    void usesResolvedVinForJt808WhenPresent() {
        assertThat(VehicleKey.derive(ProtocolId.JT808, "VIN001", "013912345678", "abc"))
                .isEqualTo("VIN001");
    }

    @Test
    void usesPhoneForJt808WhenVinIsMissing() {
        assertThat(VehicleKey.derive(ProtocolId.JT808, "", "013912345678", "abc"))
                .isEqualTo("jt808:013912345678");
    }

    @Test
    void usesStableUnknownHashWhenVehicleIdentifiersAreMissing() {
        assertThat(VehicleKey.derive(ProtocolId.JT808, "", "", "abc"))
                .isEqualTo(VehicleKey.derive(ProtocolId.JT808, null, null, "abc"))
                .startsWith("unknown:JT808:");
    }

    @Test
    void rejectsMissingProtocol() {
        assertThatThrownBy(() -> VehicleKey.derive(null, "VIN001", "", "abc"))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessageContaining("protocol");
    }
}
```

- [ ] **Step 2: Run test and verify RED**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=VehicleKeyTest test
```

Expected: FAIL because `VehicleKey` does not exist.

- [ ] **Step 3: Implement `VehicleKey`**

Create `VehicleKey.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HexFormat;

public final class VehicleKey {

    private VehicleKey() {
    }

    public static String derive(ProtocolId protocol, String vin, String phone, String rawIdentitySeed) {
        if (protocol == null) {
            throw new IllegalArgumentException("protocol must not be null");
        }
        String normalizedVin = clean(vin);
        if (!normalizedVin.isBlank()) {
            return normalizedVin;
        }
        String normalizedPhone = clean(phone);
        if (protocol == ProtocolId.JT808 && !normalizedPhone.isBlank()) {
            return "jt808:" + normalizedPhone;
        }
        String seed = clean(rawIdentitySeed);
        if (seed.isBlank()) {
            seed = protocol.name();
        }
        return "unknown:" + protocol.name() + ":" + sha256(seed).substring(0, 16);
    }

    private static String clean(String value) {
        return value == null ? "" : value.trim();
    }

    private static String sha256(String value) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 is unavailable", e);
        }
    }
}
```

- [ ] **Step 4: Run test and verify GREEN**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=VehicleKeyTest test
```

Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/VehicleKey.java \
        modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/VehicleKeyTest.java
git commit -m "feat: add vehicle key derivation"
```

---

### Task 3: Implement Raw Frame Fact and Id Helpers

**Files:**

- Create: `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/ParseStatus.java`
- Create: `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/FactIds.java`
- Create: `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameFact.java`
- Create: `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/FactIdsTest.java`
- Create: `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameFactTest.java`

- [ ] **Step 1: Write failing tests**

Create `FactIdsTest.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;
import org.junit.jupiter.api.Test;

import java.time.Instant;

import static org.assertj.core.api.Assertions.assertThat;

class FactIdsTest {

    @Test
    void rawFrameIdIsStableForSameInputs() {
        String first = FactIds.rawFrameId(
                ProtocolId.GB32960,
                "VIN001",
                2,
                7,
                Instant.parse("2026-06-29T00:00:00Z"),
                "sha256:abc");
        String second = FactIds.rawFrameId(
                ProtocolId.GB32960,
                "VIN001",
                2,
                7,
                Instant.parse("2026-06-29T00:00:00Z"),
                "sha256:abc");

        assertThat(first).isEqualTo(second).startsWith("rf_");
    }

    @Test
    void decodedFactIdIncludesKind() {
        String id = FactIds.decodedFactId("rf_123", "location", 0);

        assertThat(id).isEqualTo("df_rf_123_location_0");
    }
}
```

Create `RawFrameFactTest.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;
import org.junit.jupiter.api.Test;

import java.time.Instant;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

class RawFrameFactTest {

    @Test
    void normalizesOptionalStringsAndMetadata() {
        RawFrameFact fact = new RawFrameFact(
                "frame-1",
                ProtocolId.JT808,
                "jt808:013912345678",
                null,
                "013912345678",
                0x0200,
                0,
                null,
                Instant.parse("2026-06-29T00:00:00Z"),
                null,
                "archive://2026/06/29/JT808/jt808-013912345678/frame-1.bin",
                "sha256:abc",
                128,
                ParseStatus.SUCCEEDED,
                null,
                Map.of("messageName", "LOCATION"));

        assertThat(fact.vin()).isEmpty();
        assertThat(fact.eventTime()).isEqualTo(fact.receivedAt());
        assertThat(fact.peer()).isEmpty();
        assertThat(fact.parseError()).isEmpty();
        assertThat(fact.metadata()).containsEntry("messageName", "LOCATION");
    }

    @Test
    void rejectsMissingRequiredFields() {
        assertThatThrownBy(() -> new RawFrameFact(
                "",
                ProtocolId.GB32960,
                "VIN001",
                "VIN001",
                "",
                2,
                0,
                Instant.parse("2026-06-29T00:00:00Z"),
                Instant.parse("2026-06-29T00:00:00Z"),
                "",
                "archive://x",
                "sha256:abc",
                1,
                ParseStatus.SUCCEEDED,
                "",
                Map.of()))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessageContaining("frameId");
    }
}
```

- [ ] **Step 2: Run tests and verify RED**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=FactIdsTest,RawFrameFactTest test
```

Expected: FAIL because `FactIds`, `RawFrameFact`, and `ParseStatus` do not exist.

- [ ] **Step 3: Implement `ParseStatus`**

Create `ParseStatus.java`:

```java
package com.lingniu.ingest.facts;

public enum ParseStatus {
    NOT_PARSED,
    SUCCEEDED,
    FAILED
}
```

- [ ] **Step 4: Implement `FactIds`**

Create `FactIds.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.HexFormat;

public final class FactIds {

    private FactIds() {
    }

    public static String rawFrameId(ProtocolId protocol,
                                    String vehicleKey,
                                    int messageId,
                                    int subType,
                                    Instant receivedAt,
                                    String checksum) {
        if (protocol == null) {
            throw new IllegalArgumentException("protocol must not be null");
        }
        String input = protocol.name() + "|"
                + clean(vehicleKey) + "|"
                + messageId + "|"
                + subType + "|"
                + (receivedAt == null ? "" : receivedAt.toString()) + "|"
                + clean(checksum);
        return "rf_" + sha256(input).substring(0, 32);
    }

    public static String decodedFactId(String frameId, String kind, int ordinal) {
        return "df_" + clean(frameId) + "_" + clean(kind) + "_" + ordinal;
    }

    private static String clean(String value) {
        return value == null ? "" : value.trim();
    }

    private static String sha256(String value) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 is unavailable", e);
        }
    }
}
```

- [ ] **Step 5: Implement `RawFrameFact`**

Create `RawFrameFact.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;

import java.time.Instant;
import java.util.Map;

public record RawFrameFact(
        String frameId,
        ProtocolId protocol,
        String vehicleKey,
        String vin,
        String phone,
        int messageId,
        int subType,
        Instant eventTime,
        Instant receivedAt,
        String peer,
        String rawUri,
        String checksum,
        long rawSizeBytes,
        ParseStatus parseStatus,
        String parseError,
        Map<String, String> metadata
) {
    public RawFrameFact {
        frameId = required(frameId, "frameId");
        if (protocol == null) {
            throw new IllegalArgumentException("protocol must not be null");
        }
        vehicleKey = required(vehicleKey, "vehicleKey");
        receivedAt = required(receivedAt, "receivedAt");
        eventTime = eventTime == null ? receivedAt : eventTime;
        rawUri = required(rawUri, "rawUri");
        checksum = required(checksum, "checksum");
        if (rawSizeBytes < 0) {
            throw new IllegalArgumentException("rawSizeBytes must not be negative");
        }
        parseStatus = parseStatus == null ? ParseStatus.NOT_PARSED : parseStatus;
        vin = clean(vin);
        phone = clean(phone);
        peer = clean(peer);
        parseError = clean(parseError);
        metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
    }

    private static String required(String value, String field) {
        String cleaned = clean(value);
        if (cleaned.isBlank()) {
            throw new IllegalArgumentException(field + " must not be blank");
        }
        return cleaned;
    }

    private static Instant required(Instant value, String field) {
        if (value == null) {
            throw new IllegalArgumentException(field + " must not be null");
        }
        return value;
    }

    private static String clean(String value) {
        return value == null ? "" : value.trim();
    }
}
```

- [ ] **Step 6: Run tests and verify GREEN**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=FactIdsTest,RawFrameFactTest test
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
git add modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts \
        modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts
git commit -m "feat: add raw frame fact model"
```

---

### Task 4: Implement Decoded Fact Records

**Files:**

- Create: `modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/DecodedFact.java`
- Create: `modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/DecodedFactTest.java`

- [ ] **Step 1: Write failing test**

Create `DecodedFactTest.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;
import org.junit.jupiter.api.Test;

import java.time.Instant;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;

class DecodedFactTest {

    @Test
    void locationFactCarriesRawReferenceAndMetadata() {
        DecodedFact.Location location = new DecodedFact.Location(
                "fact-1",
                "frame-1",
                ProtocolId.JT808,
                "jt808:013912345678",
                "",
                "013912345678",
                Instant.parse("2026-06-29T00:00:00Z"),
                Instant.parse("2026-06-29T00:00:01Z"),
                "archive://jt808/frame-1.bin",
                Map.of("messageId", "0x0200"),
                120.123456,
                30.123456,
                10.0,
                42.0,
                180.0,
                0L,
                2L,
                1000.5);

        assertThat(location.factId()).isEqualTo("fact-1");
        assertThat(location.rawUri()).isEqualTo("archive://jt808/frame-1.bin");
        assertThat(location.metadata()).containsEntry("messageId", "0x0200");
    }
}
```

- [ ] **Step 2: Run test and verify RED**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=DecodedFactTest test
```

Expected: FAIL because `DecodedFact` does not exist.

- [ ] **Step 3: Implement `DecodedFact`**

Create `DecodedFact.java`:

```java
package com.lingniu.ingest.facts;

import com.lingniu.ingest.api.ProtocolId;

import java.time.Instant;
import java.util.Map;

public sealed interface DecodedFact
        permits DecodedFact.Location,
                DecodedFact.Realtime,
                DecodedFact.Alarm,
                DecodedFact.Session,
                DecodedFact.Register,
                DecodedFact.Extension {

    String factId();
    String frameId();
    ProtocolId protocol();
    String vehicleKey();
    String vin();
    String phone();
    Instant eventTime();
    Instant receivedAt();
    String rawUri();
    Map<String, String> metadata();

    record Location(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            Double longitude,
            Double latitude,
            Double altitudeM,
            Double speedKmh,
            Double directionDeg,
            Long alarmFlag,
            Long statusFlag,
            Double totalMileageKm
    ) implements DecodedFact {
        public Location {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            metadata = BaseFields.metadata(metadata);
        }
    }

    record Realtime(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            Map<String, String> fields
    ) implements DecodedFact {
        public Realtime {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            metadata = BaseFields.metadata(metadata);
            fields = fields == null ? Map.of() : Map.copyOf(fields);
        }
    }

    record Alarm(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            String alarmLevel,
            Long alarmCode,
            String alarmName,
            Map<String, String> fields
    ) implements DecodedFact {
        public Alarm {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            alarmLevel = BaseFields.clean(alarmLevel);
            alarmName = BaseFields.clean(alarmName);
            metadata = BaseFields.metadata(metadata);
            fields = fields == null ? Map.of() : Map.copyOf(fields);
        }
    }

    record Session(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            String sessionType,
            Integer resultCode,
            Map<String, String> fields
    ) implements DecodedFact {
        public Session {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            sessionType = BaseFields.clean(sessionType);
            metadata = BaseFields.metadata(metadata);
            fields = fields == null ? Map.of() : Map.copyOf(fields);
        }
    }

    record Register(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            Map<String, String> registrationFields
    ) implements DecodedFact {
        public Register {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            metadata = BaseFields.metadata(metadata);
            registrationFields = registrationFields == null ? Map.of() : Map.copyOf(registrationFields);
        }
    }

    record Extension(
            String factId,
            String frameId,
            ProtocolId protocol,
            String vehicleKey,
            String vin,
            String phone,
            Instant eventTime,
            Instant receivedAt,
            String rawUri,
            Map<String, String> metadata,
            String extensionType,
            Map<String, String> fields
    ) implements DecodedFact {
        public Extension {
            BaseFields.validate(factId, frameId, protocol, vehicleKey, eventTime, receivedAt, rawUri);
            vin = BaseFields.clean(vin);
            phone = BaseFields.clean(phone);
            extensionType = BaseFields.clean(extensionType);
            metadata = BaseFields.metadata(metadata);
            fields = fields == null ? Map.of() : Map.copyOf(fields);
        }
    }

    final class BaseFields {
        private BaseFields() {
        }

        static void validate(String factId,
                             String frameId,
                             ProtocolId protocol,
                             String vehicleKey,
                             Instant eventTime,
                             Instant receivedAt,
                             String rawUri) {
            required(factId, "factId");
            required(frameId, "frameId");
            if (protocol == null) {
                throw new IllegalArgumentException("protocol must not be null");
            }
            required(vehicleKey, "vehicleKey");
            if (eventTime == null) {
                throw new IllegalArgumentException("eventTime must not be null");
            }
            if (receivedAt == null) {
                throw new IllegalArgumentException("receivedAt must not be null");
            }
            required(rawUri, "rawUri");
        }

        static String clean(String value) {
            return value == null ? "" : value.trim();
        }

        static Map<String, String> metadata(Map<String, String> metadata) {
            return metadata == null ? Map.of() : Map.copyOf(metadata);
        }

        private static void required(String value, String field) {
            if (clean(value).isBlank()) {
                throw new IllegalArgumentException(field + " must not be blank");
            }
        }
    }
}
```

- [ ] **Step 4: Run test and verify GREEN**

Run:

```bash
mvn -pl modules/core/ingest-facts -Dtest=DecodedFactTest test
```

Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/DecodedFact.java \
        modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/DecodedFactTest.java
git commit -m "feat: add decoded fact model"
```

---

### Task 5: Add Raw Archive Store Contract and Local Implementation

**Files:**

- Modify: `pom.xml`
- Create: `modules/sinks/raw-archive-store/pom.xml`
- Create: `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriteRequest.java`
- Create: `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReceipt.java`
- Create: `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriter.java`
- Create: `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReader.java`
- Create: `modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStore.java`
- Create: `modules/sinks/raw-archive-store/src/test/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStoreTest.java`

- [ ] **Step 1: Write failing local archive test**

Create `LocalRawArchiveStoreTest.java`:

```java
package com.lingniu.ingest.rawarchive;

import com.lingniu.ingest.api.ProtocolId;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import java.nio.file.Path;
import java.time.Instant;
import java.util.HexFormat;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

class LocalRawArchiveStoreTest {

    @TempDir
    Path tempDir;

    @Test
    void writesReadableRawObjectWithArchiveUriAndChecksum() throws Exception {
        LocalRawArchiveStore store = new LocalRawArchiveStore(tempDir, "archive://raw");
        byte[] bytes = HexFormat.of().parseHex("7e020000007e");

        RawArchiveReceipt receipt = store.write(new RawArchiveWriteRequest(
                ProtocolId.JT808,
                "jt808:013912345678",
                "frame-1",
                Instant.parse("2026-06-29T00:00:00Z"),
                bytes));

        assertThat(receipt.rawUri()).isEqualTo("archive://raw/2026/06/29/JT808/jt808_013912345678/frame-1.bin");
        assertThat(receipt.checksum()).startsWith("sha256:");
        assertThat(receipt.sizeBytes()).isEqualTo(bytes.length);
        assertThat(store.read(receipt.rawUri())).containsExactly(bytes);
    }

    @Test
    void rejectsBlankFrameId() {
        LocalRawArchiveStore store = new LocalRawArchiveStore(tempDir, "archive://raw");

        assertThatThrownBy(() -> store.write(new RawArchiveWriteRequest(
                ProtocolId.GB32960,
                "VIN001",
                "",
                Instant.parse("2026-06-29T00:00:00Z"),
                new byte[]{1})))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessageContaining("frameId");
    }
}
```

- [ ] **Step 2: Run test and verify RED**

Run:

```bash
mvn -pl modules/sinks/raw-archive-store test
```

Expected: FAIL because the module does not exist.

- [ ] **Step 3: Add module and POM wiring**

Add this root module after `modules/sinks/sink-archive`:

```xml
<module>modules/sinks/raw-archive-store</module>
```

Add dependency management after `sink-archive`:

```xml
<dependency>
    <groupId>com.lingniu.ingest</groupId>
    <artifactId>raw-archive-store</artifactId>
    <version>${project.version}</version>
</dependency>
```

Create `modules/sinks/raw-archive-store/pom.xml`:

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.lingniu.ingest</groupId>
        <artifactId>lingniu-vehicle-ingest</artifactId>
        <version>0.1.0-SNAPSHOT</version>
        <relativePath>../../../pom.xml</relativePath>
    </parent>
    <artifactId>raw-archive-store</artifactId>
    <name>raw-archive-store</name>
    <description>Raw frame archive writer and reader contracts.</description>

    <dependencies>
        <dependency>
            <groupId>com.lingniu.ingest</groupId>
            <artifactId>ingest-api</artifactId>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>
```

- [ ] **Step 4: Implement contracts**

Create `RawArchiveWriteRequest.java`:

```java
package com.lingniu.ingest.rawarchive;

import com.lingniu.ingest.api.ProtocolId;

import java.time.Instant;

public record RawArchiveWriteRequest(
        ProtocolId protocol,
        String vehicleKey,
        String frameId,
        Instant receivedAt,
        byte[] rawBytes
) {
    public RawArchiveWriteRequest {
        if (protocol == null) {
            throw new IllegalArgumentException("protocol must not be null");
        }
        vehicleKey = required(vehicleKey, "vehicleKey");
        frameId = required(frameId, "frameId");
        if (receivedAt == null) {
            throw new IllegalArgumentException("receivedAt must not be null");
        }
        if (rawBytes == null || rawBytes.length == 0) {
            throw new IllegalArgumentException("rawBytes must not be empty");
        }
        rawBytes = rawBytes.clone();
    }

    @Override
    public byte[] rawBytes() {
        return rawBytes.clone();
    }

    private static String required(String value, String field) {
        String cleaned = value == null ? "" : value.trim();
        if (cleaned.isBlank()) {
            throw new IllegalArgumentException(field + " must not be blank");
        }
        return cleaned;
    }
}
```

Create `RawArchiveReceipt.java`:

```java
package com.lingniu.ingest.rawarchive;

public record RawArchiveReceipt(String rawUri, String checksum, long sizeBytes) {
    public RawArchiveReceipt {
        rawUri = required(rawUri, "rawUri");
        checksum = required(checksum, "checksum");
        if (sizeBytes < 0) {
            throw new IllegalArgumentException("sizeBytes must not be negative");
        }
    }

    private static String required(String value, String field) {
        String cleaned = value == null ? "" : value.trim();
        if (cleaned.isBlank()) {
            throw new IllegalArgumentException(field + " must not be blank");
        }
        return cleaned;
    }
}
```

Create `RawArchiveWriter.java`:

```java
package com.lingniu.ingest.rawarchive;

import java.io.IOException;

public interface RawArchiveWriter {
    RawArchiveReceipt write(RawArchiveWriteRequest request) throws IOException;
}
```

Create `RawArchiveReader.java`:

```java
package com.lingniu.ingest.rawarchive;

import java.io.IOException;

public interface RawArchiveReader {
    byte[] read(String rawUri) throws IOException;
}
```

- [ ] **Step 5: Implement local store**

Create `LocalRawArchiveStore.java`:

```java
package com.lingniu.ingest.rawarchive;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.ZoneId;
import java.util.HexFormat;

public final class LocalRawArchiveStore implements RawArchiveWriter, RawArchiveReader {

    private static final ZoneId PARTITION_ZONE = ZoneId.of("Asia/Shanghai");

    private final Path root;
    private final String uriPrefix;

    public LocalRawArchiveStore(Path root, String uriPrefix) {
        if (root == null) {
            throw new IllegalArgumentException("root must not be null");
        }
        this.root = root.toAbsolutePath().normalize();
        String prefix = uriPrefix == null || uriPrefix.isBlank() ? "archive://raw" : uriPrefix.trim();
        this.uriPrefix = prefix.endsWith("/") ? prefix.substring(0, prefix.length() - 1) : prefix;
    }

    @Override
    public RawArchiveReceipt write(RawArchiveWriteRequest request) throws IOException {
        String key = key(request);
        Path target = resolveKey(key);
        Files.createDirectories(target.getParent());
        rejectSymlinkPath(target);
        byte[] bytes = request.rawBytes();
        Files.write(target, bytes, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
        return new RawArchiveReceipt(uriPrefix + "/" + key, checksum(bytes), bytes.length);
    }

    @Override
    public byte[] read(String rawUri) throws IOException {
        String key = keyFromUri(rawUri);
        Path target = resolveKey(key);
        rejectSymlinkPath(target);
        return Files.readAllBytes(target);
    }

    private String key(RawArchiveWriteRequest request) {
        var date = request.receivedAt().atZone(PARTITION_ZONE).toLocalDate();
        return "%04d/%02d/%02d/%s/%s/%s.bin".formatted(
                date.getYear(),
                date.getMonthValue(),
                date.getDayOfMonth(),
                request.protocol().name(),
                safeSegment(request.vehicleKey()),
                safeSegment(request.frameId()));
    }

    private String keyFromUri(String rawUri) {
        if (rawUri == null || rawUri.isBlank()) {
            throw new IllegalArgumentException("rawUri must not be blank");
        }
        String normalized = rawUri.trim();
        String expected = uriPrefix + "/";
        if (!normalized.startsWith(expected)) {
            throw new IllegalArgumentException("rawUri does not use expected prefix: " + rawUri);
        }
        return normalized.substring(expected.length());
    }

    private Path resolveKey(String key) {
        Path target = root.resolve(key).normalize();
        if (!target.startsWith(root)) {
            throw new IllegalArgumentException("raw archive key escapes root: " + key);
        }
        return target;
    }

    private void rejectSymlinkPath(Path target) throws IOException {
        Path current = root;
        Path relative = root.relativize(target);
        for (Path segment : relative) {
            current = current.resolve(segment);
            if (Files.isSymbolicLink(current)) {
                throw new IOException("raw archive path traverses symlink: " + current);
            }
            if (Files.notExists(current, LinkOption.NOFOLLOW_LINKS)) {
                return;
            }
        }
    }

    private static String safeSegment(String value) {
        String cleaned = value == null ? "" : value.trim();
        if (cleaned.isBlank()) {
            return "unknown";
        }
        return cleaned.replaceAll("[^A-Za-z0-9._-]", "_");
    }

    private static String checksum(byte[] bytes) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            return "sha256:" + HexFormat.of().formatHex(digest.digest(bytes));
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 is unavailable", e);
        }
    }
}
```

- [ ] **Step 6: Run tests and verify GREEN**

Run:

```bash
mvn -pl modules/sinks/raw-archive-store -Dtest=LocalRawArchiveStoreTest test
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
git add pom.xml modules/sinks/raw-archive-store
git commit -m "feat: add raw archive store contract"
```

---

### Task 6: Add Kafka Proto Messages for Facts

**Files:**

- Modify: `modules/sinks/sink-kafka/pom.xml`
- Modify: `modules/sinks/sink-kafka/src/main/proto/vehicle_envelope.proto`
- Create: `modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/VehicleEnvelopeProtoCompatibilityTest.java`

- [ ] **Step 1: Write failing proto compatibility test**

Create `VehicleEnvelopeProtoCompatibilityTest.java`:

```java
package com.lingniu.ingest.sink.kafka;

import com.lingniu.ingest.sink.kafka.proto.DecodedFactPayload;
import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto;
import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

class VehicleEnvelopeProtoCompatibilityTest {

    @Test
    void rawFrameFactPayloadRoundTripsThroughEnvelope() throws Exception {
        VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
                .setSchemaVersion("2.0")
                .setEventId("frame-1")
                .setVin("VIN001")
                .setSource("GB32960")
                .setEventTimeMs(1782662400000L)
                .setIngestTimeMs(1782662400001L)
                .setRawFrameFact(RawFrameFactPayload.newBuilder()
                        .setFrameId("frame-1")
                        .setVehicleKey("VIN001")
                        .setMessageId(2)
                        .setRawUri("archive://raw/2026/06/29/GB32960/VIN001/frame-1.bin")
                        .setChecksum("sha256:abc")
                        .setRawSizeBytes(128)
                        .setParseStatus(ParseStatusProto.PARSE_STATUS_SUCCEEDED)
                        .build())
                .build();

        VehicleEnvelope parsed = VehicleEnvelope.parseFrom(envelope.toByteArray());

        assertThat(parsed.hasRawFrameFact()).isTrue();
        assertThat(parsed.getRawFrameFact().getFrameId()).isEqualTo("frame-1");
        assertThat(parsed.getRawFrameFact().getParseStatus()).isEqualTo(ParseStatusProto.PARSE_STATUS_SUCCEEDED);
    }

    @Test
    void decodedFactPayloadRoundTripsThroughEnvelope() throws Exception {
        VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
                .setSchemaVersion("2.0")
                .setEventId("fact-1")
                .setVin("VIN001")
                .setSource("JT808")
                .setEventTimeMs(1782662400000L)
                .setIngestTimeMs(1782662400001L)
                .setDecodedFact(DecodedFactPayload.newBuilder()
                        .setFactId("fact-1")
                        .setFrameId("frame-1")
                        .setFactType("location")
                        .setVehicleKey("jt808:013912345678")
                        .setRawUri("archive://raw/2026/06/29/JT808/jt808_013912345678/frame-1.bin")
                        .putFields("longitude", "120.1")
                        .putFields("latitude", "30.1")
                        .build())
                .build();

        VehicleEnvelope parsed = VehicleEnvelope.parseFrom(envelope.toByteArray());

        assertThat(parsed.hasDecodedFact()).isTrue();
        assertThat(parsed.getDecodedFact().getFieldsMap()).containsEntry("longitude", "120.1");
    }
}
```

- [ ] **Step 2: Run test and verify RED**

Run:

```bash
mvn -pl modules/sinks/sink-kafka -Dtest=VehicleEnvelopeProtoCompatibilityTest test
```

Expected: FAIL because proto generated classes do not exist.

- [ ] **Step 3: Add `ingest-facts` dependency to sink-kafka**

Add this dependency to `modules/sinks/sink-kafka/pom.xml` after `ingest-api`:

```xml
<dependency>
    <groupId>com.lingniu.ingest</groupId>
    <artifactId>ingest-facts</artifactId>
</dependency>
```

- [ ] **Step 4: Extend proto schema**

Modify `vehicle_envelope.proto`.

Add to `VehicleEnvelope` after `TelemetrySnapshot telemetry_snapshot = 50;`:

```proto
  RawFrameFactPayload raw_frame_fact = 60;
  DecodedFactPayload decoded_fact = 61;
```

Add these messages after `TelemetryField`:

```proto
enum ParseStatusProto {
  PARSE_STATUS_UNSPECIFIED = 0;
  PARSE_STATUS_NOT_PARSED = 1;
  PARSE_STATUS_SUCCEEDED = 2;
  PARSE_STATUS_FAILED = 3;
}

message RawFrameFactPayload {
  string frame_id = 1;
  string vehicle_key = 2;
  string vin = 3;
  string phone = 4;
  int32 message_id = 5;
  int32 sub_type = 6;
  string raw_uri = 7;
  string checksum = 8;
  int64 raw_size_bytes = 9;
  ParseStatusProto parse_status = 10;
  string parse_error = 11;
  string peer = 12;
  map<string, string> metadata = 13;
}

message DecodedFactPayload {
  string fact_id = 1;
  string frame_id = 2;
  string fact_type = 3;
  string vehicle_key = 4;
  string vin = 5;
  string phone = 6;
  string raw_uri = 7;
  map<string, string> fields = 8;
  map<string, string> metadata = 9;
}
```

- [ ] **Step 5: Run test and verify GREEN**

Run:

```bash
mvn -pl modules/sinks/sink-kafka -Dtest=VehicleEnvelopeProtoCompatibilityTest test
```

Expected: PASS.

- [ ] **Step 6: Run existing sink-kafka tests**

Run:

```bash
mvn -pl modules/sinks/sink-kafka test
```

Expected: PASS. Existing envelope tests must continue to pass because new proto fields are additive.

- [ ] **Step 7: Commit**

```bash
git add modules/sinks/sink-kafka/pom.xml \
        modules/sinks/sink-kafka/src/main/proto/vehicle_envelope.proto \
        modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/VehicleEnvelopeProtoCompatibilityTest.java
git commit -m "feat: add fact payloads to kafka envelope"
```

---

### Task 7: Verify Phase 1 Build

**Files:**

- No new files.

- [ ] **Step 1: Run focused module tests**

Run:

```bash
mvn -pl modules/core/ingest-facts,modules/sinks/raw-archive-store,modules/sinks/sink-kafka -am test
```

Expected: PASS.

- [ ] **Step 2: Run dependency tree smoke check**

Run:

```bash
mvn -pl modules/sinks/sink-kafka -am -DskipTests package
```

Expected: PASS. Generated protobuf Java sources include `RawFrameFactPayload`, `DecodedFactPayload`, and `ParseStatusProto`.

- [ ] **Step 3: Search for accidental runtime wiring**

Run:

```bash
rg -n "RawFrameFact|DecodedFact|RawArchiveWriter|vehicle.raw-frame.v1|vehicle.decoded-fact.v1" \
  modules/apps modules/protocols modules/services modules/core modules/sinks
```

Expected: Matches only in the new modules, sink-kafka proto/test, and planned references. No existing handler behavior should be changed in Phase 1.

- [ ] **Step 4: Commit final verification note if any docs changed**

If the implementation updated this plan with checked boxes, commit it:

```bash
git add docs/superpowers/plans/2026-06-29-vehicle-ingest-redesign-phase1.md
git commit -m "docs: update phase 1 implementation checklist"
```

If no docs changed, no commit is needed.

---

## Self-Review

Spec coverage:

- Raw original frame preservation is covered by `RawFrameFact`, raw archive contracts, and local archive tests.
- Historical query storage is prepared by stable fact model and raw URI references; TDengine writer is intentionally deferred to Phase 3.
- Extension is covered by `DecodedFact.Extension` and byte-light proto maps.
- Performance foundations are covered by immutable facts, Kafka key design, and no handler wiring in Phase 1.

Placeholder scan:

- This plan contains no unfinished markers or unspecified implementation steps.

Type consistency:

- `RawFrameFact`, `ParseStatus`, `DecodedFact`, `RawArchiveWriteRequest`, `RawArchiveReceipt`, and proto payload names are used consistently across tasks.

Next plans:

- Phase 2: wire GB32960/JT808 durable raw boundary and fact publication.
- Phase 3: implement TDengine schema manager and history writer.
- Phase 4: implement history query APIs and raw replay endpoints.