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:

<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

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.