diff --git a/docs/target-architecture.md b/docs/target-architecture.md index b5a969e1..57612218 100644 --- a/docs/target-architecture.md +++ b/docs/target-architecture.md @@ -1,271 +1,187 @@ # 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. +This document records the production target for `lingniu-vehicle-ingest`. +The project should stay small at runtime: protocol apps ingest and publish, +history consumes and indexes, analytics derives metrics, and business systems +read through explicit APIs or Kafka. + +## Active Scope + +Active production protocols: + +| Protocol | Runtime app | Event topic | Raw topic | +|---|---|---|---| +| GB/T 32960 | `gb32960-ingest-app` | `vehicle.event.gb32960.v1` | `vehicle.raw.gb32960.v1` | +| JT/T 808 | `jt808-ingest-app` | `vehicle.event.jt808.v1` | `vehicle.raw.jt808.v1` | +| Yutong MQTT | `yutong-mqtt-app` | `vehicle.event.mqtt-yutong.v1` | `vehicle.raw.mqtt-yutong.v1` | + +Xinda Push is legacy and is not part of the default production surface. +It may remain in source for reference, but new optimization work should target +the active protocols above. ## 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. +- Keep protocol IO, raw archive, history, latest-state, and statistics as + separate responsibilities. +- Publish normalized Kafka envelopes with VIN as the partition key whenever a + VIN is known. +- Store complete raw payload metadata and parsed JSON in TDengine `raw_frames`. +- Store query-friendly location rows separately from raw payload JSON. +- Keep derived statistics in metric tables, not protocol-specific daily tables. +- Use Redis only for hot state or streaming calculation state that can be + rebuilt from Kafka. +- Keep optional compatibility modules outside the production hot path. ## 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. +- Ingest apps do not write business tables. +- Ingest apps do not serve history query APIs. +- Redis is not the long-term historical store. +- `event-file-store` Parquet/DuckDB is not the default history store. +- `telemetry_fields` parsing is not handled by this project when a separate + field parsing service owns that job. -## Service Boundaries +## Runtime Responsibilities -### Ingest Service +### Protocol Apps -The ingest service owns protocol IO and event publication. +Apps: + +- `gb32960-ingest-app` +- `jt808-ingest-app` +- `yutong-mqtt-app` 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. +- Accept protocol traffic. +- Decode, authenticate, acknowledge, and maintain sessions where the protocol + requires it. +- Archive raw bytes through `sink-archive` when bytes are available. +- Publish parsed event envelopes to Kafka. +- Publish raw envelopes to Kafka. +- Resolve vehicle identity through the shared MySQL identity binding table. -The ingest service must not depend on Redis, DuckDB query endpoints, or -statistics rule repositories. +Protocol apps must stay independent from TDengine history readers, Redis state +repositories, and statistic calculators. ### Event Contract -The event contract is the only dependency shared by ingest, history, realtime -state, and statistics. +`sink-mq` owns the Kafka protobuf envelope and consumer plumbing. +Consumers should use `EnvelopeConsumerProcessor` so invalid protobuf, skipped +envelopes, and downstream failures are converted to structured results and DLQ +records instead of blocking a partition. -Core event identity: +Core identity fields: | Field | Meaning | |---|---| -| `event_id` | Idempotency key for the parsed event | +| `event_id` | Parsed event idempotency key | | `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 | +| `vin` | Vehicle id and Kafka partition key | +| `source` | Source protocol, for example `GB32960`, `JT808`, `YUTONG_MQTT` | +| `event_time_ms` | Device event time | +| `ingest_time_ms` | Platform receive time | +| `raw_uri` | `archive://...` reference to raw bytes when available | -Full-field telemetry is represented as internal field values: +Full-field telemetry uses `TelemetrySnapshot` fields with stable keys such as +`total_mileage_km`, `speed_kmh`, `longitude`, and `latitude`. -| 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 | +### History App -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. +App: `vehicle-history-app` 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. +- Consume active protocol event topics. +- Consume active protocol raw topics. +- Write raw frame rows into TDengine `raw_frames`. +- Write compact location rows into TDengine location tables. +- Keep `payloadJson.parsed` on raw rows for full raw inspection. +- Expose paged history APIs for raw frames and locations. +- Keep specialized APIs off by default unless explicitly enabled. -Storage partition: +Production default: + +- `TDENGINE_HISTORY_ENABLED=true` in deployment. +- `EVENT_FILE_STORE_ENABLED=false`. +- `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`. + +`event-file-store` remains an optional compatibility path for old low-level +record APIs or local investigations. It should not be treated as the primary +history backend. + +### Analytics App + +App: `vehicle-analytics-app` + +Responsibilities: + +- Consume JT808 event envelopes. +- Calculate 808 daily mileage from the reported GPS total mileage only: ```text -/ - protocol=GB32960/ - date=2026-06-22/ - part-*.parquet +daily_mileage_km = last_total_mileage_km - first_total_mileage_km +calculation_method = JT808_TOTAL_MILEAGE_DIFF ``` -The service stores detail events and field snapshots only. It does not calculate -or store daily mileage, daily hydrogen usage, or other derived metrics. +- Store the metric in `vehicle_stat_metric` with + `metric_key = daily_mileage_km`. +- Keep rolling daily calculation state in Redis. -### Realtime State Service +There is no JT808-specific daily mileage table and no in-memory production +state-store mode. -The realtime state service owns hot vehicle state. +### Latest State + +Module: `vehicle-state-service` 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. +- Consume normalized Kafka envelopes. +- Maintain latest state, location, and safety snapshots in Redis. +- Serve latest-state APIs for operational 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. +Redis state must be rebuildable from Kafka replay and should not be used as the +source of truth for historical queries. ## Data Flow ```mermaid flowchart LR - vehicle["Vehicle / Platform"] --> ingest["ingest service"] + vehicle["Vehicles / Platforms"] --> ingest["Protocol apps"] ingest --> archive["sink-archive
raw bytes"] - ingest --> kafka["Kafka
full-field telemetry events"] - kafka --> history["event-history-service"] + ingest --> kafka["Kafka
event + raw envelopes"] + kafka --> history["vehicle-history-app"] + kafka --> analytics["vehicle-analytics-app"] 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"] + history --> tdengine["TDengine
raw_frames + locations"] + analytics --> mysql["MySQL
vehicle_stat_metric"] + analytics --> redisCalc["Redis
mileage state"] + state --> redisState["Redis
latest state"] ``` -## 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. +- Protocol apps continue accepting traffic when history, analytics, or Redis + consumers are unavailable. +- Kafka producer failures are handled by sink retry/circuit-breaker behavior and + DLQ topics where configured. +- History consumers should commit offsets only after TDengine writes succeed. +- Analytics state can recover from Redis state or Kafka replay. +- Raw archive failures must be observable, but event publication and history + indexing remain separate concerns. ## 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. +- Active production deployment contains GB32960, JT808, Yutong MQTT, + vehicle-history, and vehicle-analytics apps. +- Xinda Push is absent from default deployment and history consumer bindings. +- History APIs read from TDengine by default; Parquet/DuckDB is optional + compatibility only. +- Raw frame queries can return complete parsed JSON through `payloadJson.parsed`. +- Location queries page over compact rows and reference raw records instead of + duplicating full raw JSON. +- JT808 daily mileage is stored only in `vehicle_stat_metric`. +- Runtime state needed for 808 mileage uses Redis, with no production memory + mode. +- Build validation covers the active modules and their composition tests.