lingniu-vehicle-ingest/docs/target-architecture.md

# Target Architecture

This document defines the target architecture for Lingniu vehicle ingest. The
main goal is to keep protocol ingest, raw archive, historical detail storage,
realtime state, and statistics as separate responsibilities connected by stable
event contracts.

## Goals

- The ingest service receives protocol messages, archives raw frames, and
  publishes normalized full-field events to Kafka.
- Historical detail storage consumes Kafka events and stores raw parsed event
  details in Parquet, queried through DuckDB.
- Realtime vehicle state consumes Kafka events and writes hot state to Redis.
- Statistics consumes Kafka events and calculates daily operational metrics by
  configurable rules.
- Kafka consumers expose non-throwing ingest results for bad protobuf payloads,
  unsupported envelopes, and downstream failures so one bad record does not
  block a vehicle partition.
- Consumer services expose `EnvelopeConsumerProcessor` beans. Kafka workers or
  framework listeners should pass topic, partition, offset, key, and value into
  the processor; `SKIPPED`, `INVALID_ENVELOPE`, and `FAILED` results are routed
  through `EnvelopeDeadLetterSink`.
- `sink-mq` owns Kafka-specific consumer plumbing: `KafkaEnvelopeConsumerFactory`
  creates one independent Kafka consumer per service processor binding,
  `KafkaEnvelopeConsumerRunner` manages the long-running poll loop, and
  `KafkaEnvelopeDeadLetterSink` publishes failed records to the configured DLQ
  topic with diagnostic headers.
- Business consumers depend on internal fields, not protocol-specific field
  names.

## Non-Goals

- The ingest service must not write business tables.
- The ingest service must not serve historical detail queries.
- Parquet history storage must not store derived daily statistics.
- Redis hot state must not become the long-term historical store.
- Statistics services must not parse GB/T 32960 frames directly.

## Service Boundaries

### Ingest Service

The ingest service owns protocol IO and event publication.

Responsibilities:

- Accept GB/T 32960, JT/T 808, MQTT, Xinda Push, and future inbound protocols.
- Decode frames and acknowledge protocol messages where required.
- Emit `RawArchive` before deduplication and rate limiting when raw bytes are
  available.
- Write raw bytes through `sink-archive`.
- Map protocol messages into internal telemetry events.
- Publish internal events to Kafka with VIN as the partition key.

The ingest service must not depend on Redis, DuckDB query endpoints, or
statistics rule repositories.

### Event Contract

The event contract is the only dependency shared by ingest, history, realtime
state, and statistics.

Core event identity:

| Field | Meaning |
|---|---|
| `event_id` | Idempotency key for the parsed event |
| `trace_id` | Cross-service trace id |
| `vin` | Vehicle VIN and Kafka partition key |
| `source_protocol` | Protocol source such as `GB32960`; consumers may map unknown future source names to `UNKNOWN` and preserve the original source in metadata |
| `protocol_version` | Source protocol version |
| `event_type` | Business event type, for example `REALTIME`, `LOCATION`, `ALARM` |
| `event_time` | Device collection time |
| `ingest_time` | Platform receive time |
| `raw_archive_uri` | Optional reference to archived raw bytes |

Full-field telemetry is represented as internal field values:

| Field | Meaning |
|---|---|
| `field_key` | Stable internal field key such as `total_mileage_km` |
| `value_type` | `STRING`, `DOUBLE`, `LONG`, `BOOLEAN`, `INSTANT`, `JSON` |
| `value` | String representation used for durable serialization |
| `unit` | Stable unit when applicable |
| `quality` | `GOOD`, `ESTIMATED`, `INVALID`, `MISSING` |
| `source_path` | Protocol source path, for traceability |

The internal field list is defined in
`docs/vehicle-telemetry-internal-fields.md`. New protocol-specific fields start
as extension fields and are promoted only after a business meaning and unit are
stable.

### Historical Detail Service

The historical detail service owns raw parsed event detail storage and exports.

Responsibilities:

- Consume normalized telemetry events from Kafka.
- Store events in Parquet partitioned by protocol and event date.
- Query Parquet through DuckDB.
- Provide HTTP APIs for date-range query and export.
- Support ascending and descending time order.
- Support multi-day concatenated export.

Storage partition:

```text
<root>/
  protocol=GB32960/
    date=2026-06-22/
      part-*.parquet
```

The service stores detail events and field snapshots only. It does not calculate
or store daily mileage, daily hydrogen usage, or other derived metrics.

### Realtime State Service

The realtime state service owns hot vehicle state.

Responsibilities:

- Consume normalized telemetry events from Kafka.
- Keep latest vehicle state by VIN in Redis.
- Keep latest location by VIN in Redis.
- Keep latest safety state and hydrogen leak state by VIN in Redis.
- Optionally keep short recent windows for operations screens.

Suggested Redis keys:

| Key | Value |
|---|---|
| `vehicle:state:{vin}` | Latest full state snapshot |
| `vehicle:location:{vin}` | Latest location snapshot |
| `vehicle:safety:{vin}` | Latest safety snapshot |
| `vehicle:event:last:{vin}` | Latest event identity and timestamps |

Redis is for millisecond-level reads and recent state. It is not the historical
source of truth.

Current module: `vehicle-state-service`.

Implemented boundaries:

- `VehicleStateEnvelopeIngestor` parses Kafka envelope bytes.
- `VehicleStateUpdater` maps full-field telemetry snapshots to Redis JSON
  payloads.
- `VehicleStateRepository` isolates storage from event parsing.
- `VehicleStateController` exposes read-only latest state APIs.

### Statistics Service

The statistics service owns derived operational metrics.

Responsibilities:

- Consume normalized telemetry events from Kafka.
- Maintain day-level vehicle point state needed for statistics.
- Run scheduled and event-triggered calculations.
- Store derived results in the common metric table.

Current JT808 daily mileage strategy:

| Strategy | Formula |
|---|---|
| `JT808_TOTAL_MILEAGE_DIFF` | Current local day last valid JT808 GPS total mileage minus first valid JT808 GPS total mileage |

The derived value is stored as `metric_key = daily_mileage_km` in
`vehicle_stat_metric`; there is no protocol-specific JT808 daily-mileage table.

Hydrogen and electricity statistics use internal fields from the same event
contract. Refueling and charging resets must be handled in statistics logic, not
in protocol mappers.

Current module: `vehicle-stat-service`.

Implemented boundaries:

- `VehicleStatEnvelopeIngestor` parses Kafka envelope bytes.
- `VehicleStatEventProcessor` extracts statistic source points from internal
  fields such as `total_mileage_km`.
- `VehicleStatRepository` isolates point/result persistence from calculation.
- `VehicleStatRuleRepository` isolates per-vehicle calculation rules.
- `DailyVehicleStatService` calculates and saves day-level results.

## Data Flow

```mermaid
flowchart LR
  vehicle["Vehicle / Platform"] --> ingest["ingest service"]
  ingest --> archive["sink-archive<br/>raw bytes"]
  ingest --> kafka["Kafka<br/>full-field telemetry events"]
  kafka --> history["event-history-service"]
  kafka --> state["vehicle-state-service"]
  kafka --> stat["vehicle-stat-service"]
  history --> parquet["Parquet files"]
  history --> duckdb["DuckDB query"]
  state --> redis["Redis hot state"]
  stat --> statdb["Statistics DB"]
```

## Kafka Topic Boundaries

Suggested topics:

| Topic | Producer | Consumer |
|---|---|---|
| `vehicle.realtime` | protocol ingress | history, realtime state, statistics |
| `vehicle.location` | protocol ingress | history, realtime state, statistics |
| `vehicle.alarm` | protocol ingress | history, realtime state |
| `vehicle.session` | protocol ingress | history |
| `vehicle.media.meta` | protocol ingress | history, media tooling |
| `vehicle.raw.archive` | optional archive publisher | replay or audit services |
| `vehicle.dlq` | any consumer | operations and replay tooling |

All telemetry event topics use VIN as the partition key to preserve per-vehicle
ordering.

## Failure Handling

- Ingest must keep receiving traffic when history, Redis, or statistics services
  are down.
- Kafka publish failures go through the existing sink circuit breaker and DLQ
  strategy.
- History consumer commits Kafka offsets only after Parquet append succeeds.
- Redis state consumer can replay from Kafka to rebuild hot state.
- Statistics consumer stores checkpoints and can recompute a day from historical
  detail events when rules change.
- Raw archive failures are logged and alerted, but parsed event publication
  remains independent.

## Implementation Phases

### Phase 1: Full-Field Contract And Historical Detail

- Add a full-field telemetry contract to `ingest-api`.
- Map GB/T 32960 parsed blocks into internal field snapshots.
- Extend Kafka envelope schema for full-field events.
- Make `event-file-store` store full-field telemetry records.
- Add query and export APIs for date-range historical detail.

### Phase 2: Redis Hot State

- Add a Redis-backed realtime state module.
- Consume full-field telemetry events.
- Store latest state, location, and safety snapshots by VIN.
- Provide query APIs for operations screens.

### Phase 3: Statistics Service

- Add a statistics module with rule configuration.
- Consume full-field telemetry events.
- Implement daily mileage strategies.
- Add result persistence boundary and calculation tests.

## Acceptance Criteria

- Ingest has no compile-time dependency on history query controllers, Redis
  state repositories, or statistics calculators.
- Every service boundary communicates through Kafka events or explicit public
  APIs, not shared mutable state.
- GB/T 32960 detail fields are traceable from internal field key back to
  protocol source path.
- Historical query works across multiple days with ascending and descending
  order.
- Redis state can be rebuilt from Kafka replay.
- Daily mileage can be calculated by at least the two initial strategies.
- Tests cover mapper output, Kafka serialization, Parquet query order, Redis
  state update, and mileage strategy calculation.