diff --git a/vehicle-data-platform/apps/api/internal/openplatform/assets/docs.html b/vehicle-data-platform/apps/api/internal/openplatform/assets/docs.html index 1cf82f4d..e7959ecb 100644 --- a/vehicle-data-platform/apps/api/internal/openplatform/assets/docs.html +++ b/vehicle-data-platform/apps/api/internal/openplatform/assets/docs.html @@ -24,7 +24,7 @@ 羚牛 · 数据开放平台车辆数据 API Reference @@ -69,6 +69,16 @@ "cooperateOnly": true } +

历史时刻剩余氢质量 · 1.10.0

POST/api/v1/vehicles/hydrogen-remaining/history/query

仅查询2026-08-01起的真实历史 event_time，不使用未来采样、实时值或日用氢量填补。最近氢相关帧异常或字段不全时显式返回，不跳过它找旧正常值。

#

请求参数

字段必填说明
queries是1–200项
queries[].requestId是本批唯一非空，最多128字节，原样回显
queries[].vin / time是VIN规范化大写17位且不含I/O/Q；北京时间yyyy-MM-dd HH:mm:ss，不可未来或早于2026-08-01
queries[].protocol否GB32960 / YUTONG_MQTT / JT808；省略固定GB32960，后两者当前UNSUPPORTED，不跨协议回退
maxTimeDifferenceSeconds否整数0–300，默认300；0要求精确命中
应用与逐车授权需同时覆盖查询时刻和采样时刻（结束时间不含端点）；FORBIDDEN不泄露采样元数据。当前appKey无效/过期为401，非法结构400。

请求示例（非真实读数）

{
+  "queries": [{
+    "requestId": "job-start",
+    "vin": "LA9GG68L2PBAF4773",
+    "time": "2026-08-03 16:02:43",
+    "protocol": "GB32960"
+  }],
+  "maxTimeDifferenceSeconds": 300
+}

合法批量HTTP200/SUCCESS，逐项状态判断：NORMAL、NO_DATA、MISSING、STALE、UNSUPPORTED、INVALID、FORBIDDEN、ERROR。非NORMAL质量null，不丢项，不补零。

每app每实例30批/60秒，最多2个并发批；429读取Retry-After秒后重试。整批9秒预算、4个工作协程，超时项ERROR而不是NO_DATA。200项是结构上限，不保证200个不同点均能按时完成；建议从20个不同点开始，超时缩小批量并退避重试。

历史容量需要历史适用版本；当前表不能回套历史。当前只有REPORTED质量可用，缺历史容量版本时不估算，返回MISSING。同批按规范化VIN+time+protocol去重，同点共享单次查询/授权结果和来源版本，所有requestId独立回显；不同点仍是非原子快照，保留各项版本和原始证据。质量保留原精度，Seeker自行计算两端净减少量并在负差值时归零。
+

车辆单日用氢量

POST/api/v1/vehicles/hydrogen-consumption/query

查询车辆单日用氢量，单位 kg。NORMAL + OK + PRELIMINARY 仅供初步监控；正式报表要求 FINAL 并核对证据与版本。两个日统计接口没有共同快照，区间缺失或不同不能直接计算百公里氢耗。

#

请求参数

字段必填说明
date是日期，格式 yyyy-MM-dd
plateNumbers否车牌数组；省略或 [] 时查询全部授权车辆

成功返回 · 200

{
  "code": "SUCCESS",
  "data": [{
    "plateNumber": "浙F06618F",
    "hydrogenConsumptionKg": 12.315,
    "status": "NORMAL"
  }]
}
  • 400：日期或车牌格式不正确
  • 401：appKey 无效、停用或过期
  • 403：指定车辆未授权

车辆单日里程

POST/api/v1/vehicles/mileage/query

返回当日行驶里程、当日累计总里程和实际选用的数据协议，单位 km。

#

请求参数

字段必填说明
date是日期，格式 yyyy-MM-dd
plateNumbers否省略时查询全部授权车辆
protocolPriority否协议选源顺序，例如 ["GB32960","MQTT","JT808"]
缺数规则：若当日没有有效里程，日里程为 0；累计总里程沿用上一个有效统计周期，计算时间也显示该周期时间。

成功返回 · 200

{
  "code": "SUCCESS",
  "data": [{
    "dailyMileageKm": 182.437,
    "totalMileageKm": 12345.679,
    "sourceProtocol": "GB32960",
    "status": "NORMAL"
  }]
}
  • 400：日期、车牌或协议参数错误
  • 403：授权期未覆盖查询日
  • 无统计：单车以 NO_DATA 返回
@@ -98,6 +108,19 @@ ["traceId", "string", "本次请求的唯一追踪标识；出现问题时请完整提供给羚牛技术支持。"] ]; const fields = { + "historical-hydrogen": [ + ["data[].requestId / vin / queryTime", "string", "逐项对应请求；不得只依赖数组位置配对作业起终点。"], + ["data[].plateNumber", "string | null", "当前缺少历史车牌证据，为null，不把当前车牌冒充历史车牌。"], + ["data[].remainingHydrogenKg / remainingHydrogenKgStatus", "number | null / enum", "仅NORMAL给原始精度kg；其他状态null。有效质量超过容差STALE；最新INVALID/MISSING保留原状态。无历史氢相关帧NO_DATA；分块重组失败或扫描预算耗尽ERROR，不得回退。"], + ["data[].hydrogenRecordTime / timeDifferenceSeconds", "datetime | null / number | null", "真实event_time（RFC3339带时区）与查询采样偏差秒数，绝不使用查询之后的帧。"], + ["data[].hydrogenValueSource / hydrogenSourceProtocol", "string | null", "当前REPORTED表示终端上报（不等同直接测量），GB32960为实际氢协议。ESTIMATED保留但缺历史容积版本时不输出估算值。"], + ["data[].sourceRecordId", "string | null", "raw-sha256采样身份指纹，便于识别同一采样命中；内容版本另看sourceDataVersion，不是批次快照。"], + ["data[].sourceDataVersion", "string | null", "sha256原始JSON与解析状态内容指纹，识别同采样修订；不同起终点内容不同是正常的，不要求两端此字段相等。"], + ["data[].hydrogenCalculationVersion / hydrogenCapacityVersion", "string | null", "REPORTED_HYDROGEN_KG_V1解码转换版本；历史容量版本缺证据为null，不能用当前值回套。"], + ["data[].updatedAt", "datetime | null", "原始记录接收时间，不代替采集时间；即使时间不变，内容修订仍会改变sourceDataVersion。"], + ["data[].reasonCode / message", "string | null", "可程序识别原因与中文说明；ERROR应重试，不能当NO_DATA；FORBIDDEN不返回采样证据。"], + ["data[].hydrogenEstimatePressureMPa / hydrogenEstimateTemperatureC / hydrogenTankCapacityL / hydrogenPressureTemperatureSource", "number | null / string | null", "有同帧温压可返回诊断证据；MAX_SENSOR_AGGREGATE不保证同瓶。没有历史容量证据仍MISSING，容积null。"] + ], "daily-hydrogen": [ ["data[]", "array", "按请求车牌顺序返回的车辆数据数组。"], ["data[].vin", "string", "授权车辆 VIN，必须与实时、里程记录共同核对。"], diff --git a/vehicle-data-platform/apps/api/internal/openplatform/assets/openapi.yaml b/vehicle-data-platform/apps/api/internal/openplatform/assets/openapi.yaml index a529e3bb..08dfe1a2 100644 --- a/vehicle-data-platform/apps/api/internal/openplatform/assets/openapi.yaml +++ b/vehicle-data-platform/apps/api/internal/openplatform/assets/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: title: 车辆数据开放平台 API - version: 1.9.0 + version: 1.10.0 license: name: Proprietary description: | @@ -16,6 +16,55 @@ tags: - name: 开放平台管理 description: 仅车辆数据平台管理员可调用的应用和车辆授权管理接口 paths: + /api/v1/vehicles/hydrogen-remaining/history/query: + post: + tags: [合作方数据接口] + summary: 批量查询历史时刻剩余氢质量 + description: | + 按真实 event_time 查询不晚于指定时刻的最近氢相关帧，不跳过最新异常记录，不使用实时值或日用氢量填补。 + 历史支持从 2026-08-01 起；实际车辆/时间覆盖以保留的原始帧为准，日期可请求不代表每个时点均有有效氢量。 + 历史容量必须具有适用于历史时点的证据；不能用当前配置回套历史。当前缺少历史容量版本，只输出有证据的REPORTED质量，温压有效但无容量版本时 MISSING。 + 同批按规范化VIN+time+protocol去重，同一点共享一次查询/授权结果与来源版本，各requestId仍独立回显；不同点为非原子快照，不以同批证明两端可比。 + 200项仅为结构上限，不保证200个不同点能在9秒完成。建议先用20个不同点，ERROR超时时缩小批量并退避重试。 + 有效采样超容差为STALE，最新帧INVALID/MISSING优先保留原原因；分块原始帧重组失败或候选预算耗尽为ERROR，不伪装NO_DATA。 + 每项均返回 requestId；FORBIDDEN、ERROR 等失败不丢项，非 NORMAL 质量为 null。 + appKey须当前有效，并逐项校验请求时刻与采样时刻授权（结束时间不含端点）；FORBIDDEN不返回采样元数据。 + 每app每服务实例30批/60秒，最多2个并发批；429包含Retry-After。整批9秒预算、最多4个工作协程，超时项ERROR，可重试。 + operationId: queryHistoricalHydrogenRemaining + security: + - AppKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/HistoricalHydrogenQuery' + example: + queries: + - requestId: job-start + vin: LA9GG68L2PBAF4773 + time: '2026-08-03 16:02:43' + protocol: GB32960 + maxTimeDifferenceSeconds: 300 + responses: + '200': + description: 批量处理完成；每项状态独立判断，包括 FORBIDDEN 或 ERROR + content: + application/json: + schema: + $ref: '#/components/schemas/HistoricalHydrogenQueryResponse' + '400': + $ref: '#/components/responses/BadRequest' + '401': + $ref: '#/components/responses/Unauthorized' + '429': + description: 超出历史查询限流；读取 Retry-After 秒数后重试 + headers: + Retry-After: + schema: { type: integer, minimum: 1 } + description: 建议等待秒数 + '500': + $ref: '#/components/responses/InternalError' /api/v1/vehicles/hydrogen-consumption/query: post: tags: [合作方数据接口] @@ -529,7 +578,7 @@ components: properties: vin: type: string - pattern: '^[A-HJ-NPR-Z0-9]{17}$' + pattern: '^[A-HJ-NPR-Za-hj-npr-z0-9]{17}$' description: 已授权车辆 VIN time: type: string @@ -598,6 +647,67 @@ components: minLength: 1 maxLength: 32 description: 可选；省略或传空数组时查询当前有效授权的全部车辆 + HistoricalHydrogenQuery: + type: object + additionalProperties: false + required: [queries] + properties: + queries: + type: array + minItems: 1 + maxItems: 200 + items: + type: object + additionalProperties: false + required: [requestId, vin, time] + properties: + requestId: { type: string, minLength: 1, maxLength: 128, description: 本批唯一非空关联标识，最多128字节，原样返回 } + vin: { type: string, minLength: 17, maxLength: 17, pattern: '^[A-HJ-NPR-Za-hj-npr-z0-9]{17}$', description: 查询VIN按大写规范化，禁止I/O/Q } + time: { type: string, description: '北京时间 yyyy-MM-dd HH:mm:ss；不得早于2026-08-01或晚于请求时刻' } + protocol: { type: string, enum: [GB32960, YUTONG_MQTT, JT808], default: GB32960, description: 省略固定GB32960；显式协议不切换，YUTONG_MQTT/JT808当前UNSUPPORTED } + maxTimeDifferenceSeconds: + type: integer + minimum: 0 + maximum: 300 + default: 300 + description: 允许采样早于查询的最大秒数；0要求精确命中，不静默放宽 + HistoricalHydrogenQueryResponse: + allOf: + - $ref: '#/components/schemas/SuccessEnvelope' + - type: object + required: [data] + properties: + data: + type: array + items: { $ref: '#/components/schemas/HistoricalHydrogenResult' } + HistoricalHydrogenResult: + type: object + required: [requestId, vin, plateNumber, queryTime, remainingHydrogenKg, hydrogenRecordTime, timeDifferenceSeconds, remainingHydrogenKgStatus, hydrogenValueSource, hydrogenSourceProtocol, sourceRecordId, sourceDataVersion, hydrogenEstimatePressureMPa, hydrogenEstimateTemperatureC, hydrogenTankCapacityL, hydrogenPressureTemperatureSource, hydrogenCalculationVersion, hydrogenCapacityVersion, updatedAt, reasonCode, message] + properties: + requestId: { type: string, description: 本批对应请求标识 } + vin: { type: string } + plateNumber: { type: string, nullable: true, description: 无可证明历史车牌时为 null，不冒用当前车牌 } + queryTime: { type: string, description: 请求北京时间 } + remainingHydrogenKg: { type: number, nullable: true, minimum: 0, description: '全车剩余氢质量kg，保留源/模型计算精度；仅NORMAL可用，其他状态null。不得用缺值补0' } + hydrogenRecordTime: { type: string, format: date-time, nullable: true, description: 真实原始帧 event_time，RFC3339带时区 } + timeDifferenceSeconds: { type: number, nullable: true, minimum: 0, description: queryTime减真实采集时间的秒数 } + remainingHydrogenKgStatus: + type: string + enum: [NORMAL, NO_DATA, MISSING, STALE, UNSUPPORTED, INVALID, FORBIDDEN, ERROR] + description: 仅NORMAL参与业务计算；ERROR明确为处理失败，可按原因重试；FORBIDDEN不泄露采样 + hydrogenValueSource: { type: string, nullable: true, enum: [REPORTED, ESTIMATED], description: 终端上报不等同于直接测量；估算需历史温压/容积证据 } + hydrogenSourceProtocol: { type: string, nullable: true, enum: [GB32960, YUTONG_MQTT, JT808] } + sourceRecordId: { type: string, nullable: true, description: raw-sha256采样身份指纹；两端相同需识别采样分辨率不足，内容修订另看sourceDataVersion } + sourceDataVersion: { type: string, nullable: true, description: 'sha256原始JSON和解析状态内容指纹；同采样被修订时变化，不是两端必须相等的口径版本' } + hydrogenEstimatePressureMPa: { type: number, nullable: true, description: 有效同帧最大氢压MPa证据，缺历史容量仍可诊断返回 } + hydrogenEstimateTemperatureC: { type: number, nullable: true, description: 有效同帧最大氢温摄氏度证据 } + hydrogenTankCapacityL: { type: number, nullable: true, description: 当前缺历史容量版本始终null，不能回套当前容积 } + hydrogenPressureTemperatureSource: { type: string, nullable: true, enum: [MAX_SENSOR_AGGREGATE], description: 最大温压聚合不保证同瓶 } + hydrogenCalculationVersion: { type: string, nullable: true, description: 当前REPORTED_HYDROGEN_KG_V1，质量不额外舍入；ESTIMATED预留但当前缺历史容积版本不输出估算值 } + hydrogenCapacityVersion: { type: string, nullable: true, description: 历史时点适用容积版本；估算必需，无证据不估算 } + updatedAt: { type: string, format: date-time, nullable: true, description: 原始记录received_at，RFC3339带时区；不代替采集时间，同时间戳修订通过sourceDataVersion识别 } + reasonCode: { type: string, nullable: true, description: 可程序识别的状态原因 } + message: { type: string, nullable: true, description: 中文状态说明 } HydrogenStationQuery: type: object additionalProperties: false diff --git a/vehicle-data-platform/apps/api/internal/openplatform/docs_test.go b/vehicle-data-platform/apps/api/internal/openplatform/docs_test.go index 42426fce..c9c64ca4 100644 --- a/vehicle-data-platform/apps/api/internal/openplatform/docs_test.go +++ b/vehicle-data-platform/apps/api/internal/openplatform/docs_test.go @@ -116,3 +116,19 @@ func TestLiveContractDocumentationCoversSerializedFieldsAndQuality(t *testing.T) t.Error("HTML must not advertise sourceProtocol as realtime field") } } + +func TestHistoricalHydrogenDocumentationDefinesEvidenceAndFailureBoundaries(t *testing.T) { + for name, doc := range map[string]string{"OpenAPI": string(openAPISpec), "HTML": string(simpleDocsHTML)} { + for _, want := range []string{ + "/api/v1/vehicles/hydrogen-remaining/history/query", + "requestId", "maxTimeDifferenceSeconds", "sourceRecordId", + "hydrogenCapacityVersion", "sourceDataVersion", "reasonCode", "2026-08-01", "FORBIDDEN", "ERROR", + "STALE", "429", "Retry-After", "event_time", "历史容量", "非原子", + "20个不同点", "去重", + } { + if !strings.Contains(doc, want) { + t.Errorf("%s missing history contract evidence/failure boundary %q", name, want) + } + } + } +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/handler.go b/vehicle-data-platform/apps/api/internal/openplatform/handler.go index 291e5662..d25511a2 100644 --- a/vehicle-data-platform/apps/api/internal/openplatform/handler.go +++ b/vehicle-data-platform/apps/api/internal/openplatform/handler.go @@ -59,6 +59,7 @@ func NewExternalHandler(service *Service, portal *PortalService) *Handler { func (h *Handler) registerExternalDataRoutes() { h.mux.HandleFunc("POST "+HydrogenQueryPath, h.hydrogen) + h.mux.HandleFunc("POST "+HistoricalHydrogenQueryPath, h.historicalHydrogen) h.mux.HandleFunc("POST "+MileageQueryPath, h.mileage) h.mux.HandleFunc("POST "+MileageRangeQueryPath, h.mileageRange) h.mux.HandleFunc("POST "+TotalMileageQueryPath, h.totalMileage) @@ -109,6 +110,7 @@ func (h *Handler) registerPortalRoutes() { func NewDataHandler(service *Service) *Handler { handler := &Handler{service: service, mux: http.NewServeMux()} handler.mux.HandleFunc("POST "+HydrogenQueryPath, handler.hydrogen) + handler.mux.HandleFunc("POST "+HistoricalHydrogenQueryPath, handler.historicalHydrogen) handler.mux.HandleFunc("POST "+MileageQueryPath, handler.mileage) handler.mux.HandleFunc("POST "+MileageRangeQueryPath, handler.mileageRange) handler.mux.HandleFunc("POST "+TotalMileageQueryPath, handler.totalMileage) @@ -123,7 +125,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func IsPublicPath(path string) bool { - return path == HydrogenQueryPath || path == MileageQueryPath || path == MileageRangeQueryPath || path == TotalMileageQueryPath || path == StationaryVehicleQueryPath || path == RealtimeVehicleQueryPath || path == HydrogenStationQueryPath + return path == HistoricalHydrogenQueryPath || path == HydrogenQueryPath || path == MileageQueryPath || path == MileageRangeQueryPath || path == TotalMileageQueryPath || path == StationaryVehicleQueryPath || path == RealtimeVehicleQueryPath || path == HydrogenStationQueryPath } func (h *Handler) stationaryVehicles(w http.ResponseWriter, r *http.Request) { @@ -754,6 +756,7 @@ func requestRemoteAddress(r *http.Request) string { func dataProducts() []DataProduct { return []DataProduct{ + {Code: "historical_hydrogen_remaining", Name: "历史剩余氢质量", Description: "按VIN与历史北京时间批量查询全车剩余氢质量、采样时间及来源证据。", Version: "v1", Status: "available", Method: http.MethodPost, Path: HistoricalHydrogenQueryPath, Unit: "kg"}, { Code: "daily_hydrogen", Name: "单日用氢量", Description: "按车牌和自然日查询授权车辆的氢气消耗量。", diff --git a/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen.go b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen.go new file mode 100644 index 00000000..ff4a2e6e --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen.go @@ -0,0 +1,389 @@ +package openplatform + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "math" + "net/http" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "lingniu/vehicle-data-platform/apps/api/internal/vehicleprotocol" +) + +const HistoricalHydrogenQueryPath = "/api/v1/vehicles/hydrogen-remaining/history/query" +const historicalHydrogenBudget = 9 * time.Second + +var historicalVINPattern = regexp.MustCompile(`^[A-HJ-NPR-Z0-9]{17}$`) + +type HistoricalHydrogenQuery struct { + RequestID string `json:"requestId"` + VIN string `json:"vin"` + Time string `json:"time"` + Protocol string `json:"protocol,omitempty"` +} +type HistoricalHydrogenRequest struct { + Queries []HistoricalHydrogenQuery `json:"queries"` + MaxTimeDifferenceSeconds *int `json:"maxTimeDifferenceSeconds,omitempty"` +} + +// Optional numeric/protocol fields may be omitted, but explicit JSON null is +// not a valid integer or protocol identifier. Preserve strict unknown-field +// rejection even though this request has a custom decoder. +func (r *HistoricalHydrogenRequest) UnmarshalJSON(data []byte) error { + type plain HistoricalHydrogenRequest + var decoded plain + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&decoded); err != nil { + return err + } + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + return err + } + if value, exists := raw["maxTimeDifferenceSeconds"]; exists && bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return fmt.Errorf("maxTimeDifferenceSeconds must be an integer") + } + var queries []map[string]json.RawMessage + if err := json.Unmarshal(raw["queries"], &queries); err != nil { + return err + } + for _, query := range queries { + if value, exists := query["protocol"]; exists && bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return fmt.Errorf("protocol must be a string") + } + } + *r = HistoricalHydrogenRequest(decoded) + return nil +} + +type HistoricalHydrogenResult struct { + RequestID string `json:"requestId"` + VIN string `json:"vin"` + PlateNumber *string `json:"plateNumber"` + QueryTime string `json:"queryTime"` + RemainingHydrogenKg *float64 `json:"remainingHydrogenKg"` + HydrogenRecordTime *string `json:"hydrogenRecordTime"` + TimeDifferenceSeconds *float64 `json:"timeDifferenceSeconds"` + RemainingHydrogenKgStatus string `json:"remainingHydrogenKgStatus"` + HydrogenValueSource *string `json:"hydrogenValueSource"` + HydrogenSourceProtocol *string `json:"hydrogenSourceProtocol"` + SourceRecordID *string `json:"sourceRecordId"` + SourceDataVersion *string `json:"sourceDataVersion"` + HydrogenCalculationVersion *string `json:"hydrogenCalculationVersion"` + HydrogenCapacityVersion *string `json:"hydrogenCapacityVersion"` + UpdatedAt *string `json:"updatedAt"` + ReasonCode *string `json:"reasonCode"` + Message *string `json:"message"` + HydrogenEstimatePressureMPa *float64 `json:"hydrogenEstimatePressureMPa"` + HydrogenEstimateTemperatureC *float64 `json:"hydrogenEstimateTemperatureC"` + HydrogenTankCapacityL *float64 `json:"hydrogenTankCapacityL"` + HydrogenPressureTemperatureSource *string `json:"hydrogenPressureTemperatureSource"` +} + +type HistoricalHydrogenRepository interface { + HistoricalHydrogen(context.Context, string, time.Time, string) (*HistoricalHydrogenPoint, error) + HistoricalHydrogenAuthorized(context.Context, uint64, string, time.Time, time.Time) (bool, error) +} + +type parsedHistoricalHydrogenQuery struct { + HistoricalHydrogenQuery + at time.Time +} + +func (s *Service) validateHistoricalHydrogen(request HistoricalHydrogenRequest, now time.Time) ([]parsedHistoricalHydrogenQuery, int, error) { + if len(request.Queries) < 1 || len(request.Queries) > 200 { + return nil, 0, fmt.Errorf("%w: queries must contain 1-200 items", ErrInvalidRequest) + } + tolerance := 300 + if request.MaxTimeDifferenceSeconds != nil { + tolerance = *request.MaxTimeDifferenceSeconds + } + if tolerance < 0 || tolerance > 300 { + return nil, 0, fmt.Errorf("%w: tolerance must be 0-300", ErrInvalidRequest) + } + seen := make(map[string]bool, len(request.Queries)) + parsed := make([]parsedHistoricalHydrogenQuery, 0, len(request.Queries)) + lower := time.Date(2026, 8, 1, 0, 0, 0, 0, s.location) + for _, query := range request.Queries { + if strings.TrimSpace(query.RequestID) == "" || len(query.RequestID) > 128 || seen[query.RequestID] { + return nil, 0, fmt.Errorf("%w: requestId must be nonempty and unique", ErrInvalidRequest) + } + seen[query.RequestID] = true + query.VIN = strings.ToUpper(query.VIN) + if !historicalVINPattern.MatchString(query.VIN) { + return nil, 0, fmt.Errorf("%w: invalid vin", ErrInvalidRequest) + } + at, err := time.ParseInLocation("2006-01-02 15:04:05", query.Time, s.location) + if err != nil || at.Format("2006-01-02 15:04:05") != query.Time { + return nil, 0, fmt.Errorf("%w: invalid datetime", ErrInvalidRequest) + } + if at.After(now) || at.Before(lower) { + return nil, 0, fmt.Errorf("%w: time outside supported history", ErrInvalidRequest) + } + if query.Protocol == "" { + query.Protocol = vehicleprotocol.GB32960 + } else if protocol, ok := vehicleprotocol.Canonical(query.Protocol); !ok || protocol != query.Protocol { + return nil, 0, fmt.Errorf("%w: invalid protocol", ErrInvalidRequest) + } + parsed = append(parsed, parsedHistoricalHydrogenQuery{HistoricalHydrogenQuery: query, at: at}) + } + return parsed, tolerance, nil +} + +func (s *Service) QueryHistoricalHydrogen(ctx context.Context, appKey, traceID string, request HistoricalHydrogenRequest) ([]HistoricalHydrogenResult, error) { + now := s.now().In(s.location) + queries, tolerance, err := s.validateHistoricalHydrogen(request, now) + if err != nil { + return nil, err + } + if !appKeyPattern.MatchString(appKey) { + return nil, ErrUnauthorized + } + bounded, cancel := context.WithTimeout(ctx, historicalHydrogenBudget) + defer cancel() + app, err := s.repository.Authenticate(bounded, sha256.Sum256([]byte(strings.ToLower(appKey))), now, now, now) + if err != nil { + if errors.Is(err, ErrUnauthorized) { + return nil, ErrUnauthorized + } + return nil, err + } + release, err := s.historicalLimiter.acquire(app.ID, now) + if err != nil { + return nil, err + } + defer release() + results := make([]HistoricalHydrogenResult, len(queries)) + for i, query := range queries { + results[i] = historicalHydrogenFailure(query, "ERROR", "UPSTREAM_TIMEOUT", "历史查询超时，请重试") + } + repository, ok := s.repository.(HistoricalHydrogenRepository) + if !ok { + for i, query := range queries { + results[i] = historicalHydrogenFailure(query, "ERROR", "HISTORY_UNAVAILABLE", "历史数据服务暂不可用") + } + return results, nil + } + // Deduplicate identical sampling points within this batch. Every requestId + // still receives its own result, while duplicate endpoints share one source + // revision and one pair of authorization checks. + var unique []parsedHistoricalHydrogenQuery + var destinations [][]int + indexes := make(map[string]int, len(queries)) + for i, query := range queries { + key := query.VIN + "\x00" + query.Time + "\x00" + query.Protocol + index, exists := indexes[key] + if !exists { + index = len(unique) + indexes[key] = index + unique = append(unique, query) + destinations = append(destinations, nil) + } + destinations[index] = append(destinations[index], i) + } + type completed struct { + index int + result HistoricalHydrogenResult + } + jobs := make(chan int, len(unique)) + out := make(chan completed, len(unique)) + for i := range unique { + jobs <- i + } + close(jobs) + workers := 4 + if len(unique) < workers { + workers = len(unique) + } + for worker := 0; worker < workers; worker++ { + go func() { + for i := range jobs { + if bounded.Err() != nil { + return + } + out <- completed{i, s.queryHistoricalHydrogenItem(bounded, repository, app.ID, unique[i], tolerance, now)} + } + }() + } + for count := 0; count < len(unique); count++ { + select { + case result := <-out: + for _, destination := range destinations[result.index] { + item := result.result + item.RequestID = queries[destination].RequestID + results[destination] = item + } + case <-bounded.Done(): + return results, nil + } + } + // Use the remaining shared budget; avoid extending a timed-out batch for audit. + if bounded.Err() == nil { + _ = s.repository.Audit(bounded, app.ID, "historical_hydrogen_query", "success", traceID, len(results), "") + } + return results, nil +} + +func historicalHydrogenFailure(query parsedHistoricalHydrogenQuery, status, reason, message string) HistoricalHydrogenResult { + return HistoricalHydrogenResult{RequestID: query.RequestID, VIN: query.VIN, QueryTime: query.Time, RemainingHydrogenKgStatus: status, ReasonCode: historicalString(reason), Message: historicalString(message)} +} +func historicalString(value string) *string { + if value == "" { + return nil + } + return &value +} +func historicalTime(value time.Time, location *time.Location) *string { + if value.IsZero() { + return nil + } + text := value.In(location).Format(time.RFC3339Nano) + return &text +} + +func (s *Service) queryHistoricalHydrogenItem(ctx context.Context, repository HistoricalHydrogenRepository, appID uint64, query parsedHistoricalHydrogenQuery, tolerance int, now time.Time) HistoricalHydrogenResult { + failure := func(err error) HistoricalHydrogenResult { + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { + return historicalHydrogenFailure(query, "ERROR", "UPSTREAM_TIMEOUT", "历史查询超时，请重试") + } + return historicalHydrogenFailure(query, "ERROR", "UPSTREAM_FAILURE", "历史查询失败，请重试") + } + allowed, err := repository.HistoricalHydrogenAuthorized(ctx, appID, query.VIN, query.at, now) + if err != nil { + return failure(err) + } + if !allowed { + return historicalHydrogenFailure(query, "FORBIDDEN", "HISTORY_FORBIDDEN", "车辆或历史时刻未授权") + } + if query.Protocol != vehicleprotocol.GB32960 { + return historicalHydrogenFailure(query, "UNSUPPORTED", "PROTOCOL_UNSUPPORTED", "该协议暂不支持历史储氢质量") + } + point, err := repository.HistoricalHydrogen(ctx, query.VIN, query.at, query.Protocol) + if err != nil { + return failure(err) + } + if point == nil { + return historicalHydrogenFailure(query, "NO_DATA", "NO_HISTORICAL_SAMPLE", "查询时刻前没有历史储氢采样") + } + if point.VIN != query.VIN || point.Protocol != query.Protocol { + return historicalHydrogenFailure(query, "ERROR", "SOURCE_IDENTITY_MISMATCH", "历史采样身份不一致") + } + if point.ObservedAt.IsZero() || point.ObservedAt.After(query.at) { + return historicalHydrogenFailure(query, "ERROR", "INVALID_SAMPLE_TIME", "历史采样时间不符合查询条件") + } + allowed, err = repository.HistoricalHydrogenAuthorized(ctx, appID, query.VIN, point.ObservedAt, now) + if err != nil { + return failure(err) + } + if !allowed { + return historicalHydrogenFailure(query, "FORBIDDEN", "HISTORY_FORBIDDEN", "车辆或历史时刻未授权") + } + result := HistoricalHydrogenResult{RequestID: query.RequestID, VIN: query.VIN, QueryTime: query.Time, RemainingHydrogenKgStatus: point.Status, HydrogenRecordTime: historicalTime(point.ObservedAt, s.location), HydrogenValueSource: historicalString(point.ValueSource), HydrogenSourceProtocol: historicalString(point.Protocol), SourceRecordID: historicalString(point.RecordID), SourceDataVersion: historicalString(point.SourceDataVersion), HydrogenCalculationVersion: historicalString(point.CalculationVersion), HydrogenCapacityVersion: historicalString(point.CapacityVersion), UpdatedAt: historicalTime(point.UpdatedAt, s.location), ReasonCode: historicalString(point.ReasonCode), Message: historicalString(point.Message), HydrogenEstimatePressureMPa: point.EstimatePressureMPa, HydrogenEstimateTemperatureC: point.EstimateTemperatureC, HydrogenTankCapacityL: point.TankCapacityL, HydrogenPressureTemperatureSource: historicalString(point.PressureTemperatureSource)} + difference := query.at.Sub(point.ObservedAt).Seconds() + result.TimeDifferenceSeconds = &difference + switch result.RemainingHydrogenKgStatus { + case "NORMAL", "NO_DATA", "MISSING", "STALE", "UNSUPPORTED", "INVALID": + default: + return historicalHydrogenFailure(query, "ERROR", "INVALID_SOURCE_STATUS", "历史数据状态异常") + } + if result.RemainingHydrogenKgStatus == "NORMAL" { + if point.RemainingHydrogenKg == nil { + result.RemainingHydrogenKgStatus = "MISSING" + result.ReasonCode = historicalString("MISSING_MASS") + } else if math.IsNaN(*point.RemainingHydrogenKg) || math.IsInf(*point.RemainingHydrogenKg, 0) || *point.RemainingHydrogenKg < 0 { + result.RemainingHydrogenKgStatus = "INVALID" + result.ReasonCode = historicalString("INVALID_MASS") + } else if difference > float64(tolerance) { + result.RemainingHydrogenKgStatus = "STALE" + result.ReasonCode = historicalString("SAMPLE_OUTSIDE_TOLERANCE") + result.Message = historicalString("采样时间早于允许容差") + } else if point.ValueSource == "ESTIMATED" && (point.CalculationVersion == "" || point.CapacityVersion == "" || point.EstimatePressureMPa == nil || point.EstimateTemperatureC == nil || point.TankCapacityL == nil || point.PressureTemperatureSource == "") { + result.RemainingHydrogenKgStatus = "MISSING" + result.ReasonCode = historicalString("MISSING_ESTIMATE_EVIDENCE") + } else { + value := *point.RemainingHydrogenKg + result.RemainingHydrogenKg = &value + } + } + return result +} + +func (r *MySQLRepository) HistoricalHydrogenAuthorized(ctx context.Context, appID uint64, vin string, at, now time.Time) (bool, error) { + var exists bool + err := r.db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM vehicle_open_app a JOIN vehicle_open_app_vehicle g ON g.app_id=a.id WHERE a.id=? AND a.status='enabled' AND a.valid_from<=? AND (a.valid_to IS NULL OR a.valid_to>?) AND a.valid_from<=? AND (a.valid_to IS NULL OR a.valid_to>?) AND BINARY g.vin=BINARY ? AND g.valid_from<=? AND (g.valid_to IS NULL OR g.valid_to>?))`, appID, now, now, at, at, vin, at, at).Scan(&exists) + return exists, err +} + +type historicalHydrogenRateError struct{ RetryAfter int } + +func (e *historicalHydrogenRateError) Error() string { + return "historical hydrogen rate limit exceeded" +} + +type historicalHydrogenRateEntry struct { + start time.Time + count, active int +} +type historicalHydrogenLimiter struct { + mu sync.Mutex + entries map[uint64]*historicalHydrogenRateEntry +} + +func (l *historicalHydrogenLimiter) acquire(appID uint64, now time.Time) (func(), error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.entries == nil { + l.entries = make(map[uint64]*historicalHydrogenRateEntry) + } + for id, entry := range l.entries { + if entry.active == 0 && now.Sub(entry.start) >= 2*time.Minute { + delete(l.entries, id) + } + } + entry := l.entries[appID] + if entry == nil { + entry = &historicalHydrogenRateEntry{start: now} + l.entries[appID] = entry + } + if now.Sub(entry.start) >= time.Minute { + entry.start = now + entry.count = 0 + } + if entry.active >= 2 { + return nil, &historicalHydrogenRateError{RetryAfter: 1} + } + if entry.count >= 30 { + return nil, &historicalHydrogenRateError{RetryAfter: max(1, int(math.Ceil(time.Minute.Seconds()-now.Sub(entry.start).Seconds())))} + } + entry.count++ + entry.active++ + return func() { l.mu.Lock(); entry.active--; l.mu.Unlock() }, nil +} +func (h *Handler) historicalHydrogen(w http.ResponseWriter, r *http.Request) { + traceID := externalTraceID(r) + var request HistoricalHydrogenRequest + if !decodeExternalBody(w, r, traceID, &request) { + return + } + data, err := h.service.QueryHistoricalHydrogen(r.Context(), externalBearer(r), traceID, request) + if err != nil { + var limit *historicalHydrogenRateError + if errors.As(err, &limit) { + w.Header().Set("Retry-After", strconv.Itoa(limit.RetryAfter)) + writeExternal(w, http.StatusTooManyRequests, ExternalResponse{Code: "RATE_LIMITED", Message: "请求过于频繁，请稍后重试", TraceID: traceID}) + return + } + writeExternalError(w, traceID, err) + return + } + writeExternal(w, http.StatusOK, ExternalResponse{Code: "SUCCESS", Message: "success", Data: data, TraceID: traceID}) +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_acceptance_test.go b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_acceptance_test.go new file mode 100644 index 00000000..43bde07d --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_acceptance_test.go @@ -0,0 +1,39 @@ +package openplatform + +import ( + "context" + "testing" + "time" +) + +func TestHistoricalHydrogenCrossMidnightKeepsInstantAndRevisionEvidence(t *testing.T) { + service, repository, _ := historicalTestService() + start := time.Date(2026, 8, 3, 23, 59, 55, 0, service.location) + end := start.Add(20 * time.Second) + repository.lookup = func(_ context.Context, _ string, at time.Time, _ string) (*HistoricalHydrogenPoint, error) { + point := historicalTestPoint(at, 30.50012345) + point.RecordID = at.Format(time.RFC3339) + if at.Equal(end) { + point.RemainingHydrogenKg = realLiveFloat(29.80012345) + point.SourceDataVersion = "end-v2" + } + return point, nil + } + request := HistoricalHydrogenRequest{Queries: []HistoricalHydrogenQuery{ + {RequestID: "job-end", VIN: historicalTestVIN, Time: end.Format("2006-01-02 15:04:05")}, + {RequestID: "job-start", VIN: historicalTestVIN, Time: start.Format("2006-01-02 15:04:05")}, + }} + results, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "cross-midnight", request) + if err != nil || len(results) != 2 { + t.Fatalf("%+v %v", results, err) + } + for i, at := range []time.Time{end, start} { + got := results[i] + if got.RequestID != request.Queries[i].RequestID || got.QueryTime != request.Queries[i].Time || got.HydrogenRecordTime == nil || *got.HydrogenRecordTime != at.Format(time.RFC3339) || got.RemainingHydrogenKgStatus != "NORMAL" || *got.TimeDifferenceSeconds != 0 { + t.Fatalf("midnight instant lost: %+v", got) + } + } + if *results[0].RemainingHydrogenKg != 29.80012345 || *results[1].RemainingHydrogenKg != 30.50012345 || *results[0].SourceRecordID == *results[1].SourceRecordID || *results[0].SourceDataVersion == *results[1].SourceDataVersion { + t.Fatal("endpoint precision or distinct provenance lost") + } +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository.go b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository.go new file mode 100644 index 00000000..b1bef73b --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository.go @@ -0,0 +1,239 @@ +package openplatform + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "math" + "strconv" + "strings" + "time" +) + +// HistoricalHydrogenStart is the first supported event-time boundary. This is +// an API coverage boundary, not a claim that every VIN has retained records. +var HistoricalHydrogenStart = time.Date(2026, 8, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*60*60)) + +type HistoricalHydrogenPoint struct { + VIN, Protocol, RecordID, SourceDataVersion string + ObservedAt, ReceivedAt, UpdatedAt time.Time + RemainingHydrogenKg *float64 + Status, ValueSource, CalculationVersion, CapacityVersion, ReasonCode, Message string + EstimatePressureMPa, EstimateTemperatureC, TankCapacityL *float64 + PressureTemperatureSource string +} + +type historicalHydrogenFrame struct { + VIN, Protocol, EventID, FrameID, Parsed, ParseStatus string + EventMS, ReceivedMS, TimestampMS sql.NullInt64 +} + +func (r *MySQLRepository) HistoricalHydrogen(ctx context.Context, vin string, at time.Time, protocol string) (*HistoricalHydrogenPoint, error) { + if protocol == "" { + protocol = "GB32960" + } + if protocol != "GB32960" { + return &HistoricalHydrogenPoint{VIN: vin, Protocol: protocol, Status: "UNSUPPORTED", ReasonCode: "UNSUPPORTED_HYDROGEN_PROTOCOL", Message: "该协议暂不支持历史剩余氢质量"}, nil + } + if r.tdengine == nil { + return nil, errors.New("TDengine is not configured for historical hydrogen query") + } + query, err := historicalHydrogenQuery(r.tdDatabase, vin, at, protocol) + if err != nil { + return nil, err + } + for candidate := 0; candidate < 32; candidate++ { + var frame historicalHydrogenFrame + frame, err = r.loadHistoricalHydrogenCandidate(ctx, query) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + if frame.VIN != vin || frame.Protocol != protocol { + return nil, errors.New("historical hydrogen source identity mismatch") + } + if err := r.hydrateHistoricalHydrogenFrame(ctx, &frame); err != nil { + return nil, err + } + point := historicalHydrogenFromFrame(frame) + if point.ObservedAt.IsZero() || point.ObservedAt.After(at) || point.ObservedAt.Before(HistoricalHydrogenStart) { + return nil, errors.New("historical hydrogen source time out of bounds") + } + if point.ReasonCode != "MISSING_HYDROGEN_MEASUREMENT" { + return &point, nil + } + // A hydrated chunk can prove that a candidate has no hydrogen fields. Only + // in that case move to the next record; invalid/incomplete readings stop. + clause := fmt.Sprintf(" AND (event_time<%d OR (event_time=%d AND ts<%d))", frame.EventMS.Int64, frame.EventMS.Int64, frame.TimestampMS.Int64) + query = strings.Replace(query, " ORDER BY", clause+" ORDER BY", 1) + } + return nil, errors.New("historical hydrogen candidate budget exceeded") +} + +func historicalHydrogenQuery(database, vin string, at time.Time, protocol string) (string, error) { + if !validTDIdentifier(database) || vin == "" || protocol != "GB32960" || at.Before(HistoricalHydrogenStart) { + return "", errors.New("invalid historical hydrogen query") + } + quote := func(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" } + fields := []string{`parsed_json LIKE '%"chunked":true%'`} + for _, key := range append(append([]string{}, hydrogenMassFields...), realtimeHydrogenPressureField, realtimeHydrogenTemperatureField) { + fields = append(fields, "parsed_json LIKE "+quote("%\""+key+"\":%")) + } + // Raw rows are JSON-marshaled by ingestion (compact key-colon syntax). Do not + // filter invalid values/parse status or restrict arrival <= query event time: + // a delayed upload remains valid historical evidence for its collection time. + return `SELECT vin,protocol,event_id,frame_id,CAST(event_time AS BIGINT),CAST(received_at AS BIGINT),CAST(ts AS BIGINT),parsed_json,parse_status FROM ` + database + `.raw_frames WHERE vin=` + quote(vin) + ` AND protocol=` + quote(protocol) + ` AND event_time>=` + strconv.FormatInt(HistoricalHydrogenStart.UnixMilli(), 10) + ` AND event_time<=` + strconv.FormatInt(at.UnixMilli(), 10) + ` AND (` + strings.Join(fields, " OR ") + `) ORDER BY event_time DESC,ts DESC LIMIT 2`, nil +} + +func historicalHydrogenFromFrame(frame historicalHydrogenFrame) HistoricalHydrogenPoint { + point := HistoricalHydrogenPoint{VIN: frame.VIN, Protocol: frame.Protocol, Status: "MISSING"} + if frame.EventMS.Valid && frame.EventMS.Int64 > 0 { + point.ObservedAt = time.UnixMilli(frame.EventMS.Int64) + } + if frame.ReceivedMS.Valid && frame.ReceivedMS.Int64 > 0 { + point.ReceivedAt = time.UnixMilli(frame.ReceivedMS.Int64) + point.UpdatedAt = point.ReceivedAt + } + // The content fingerprint exposes corrections even if storage keeps the + // original arrival timestamp. It is not part of calculation comparability. + identity := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%s\x00%s\x00%d\x00%d", frame.VIN, frame.Protocol, frame.EventID, frame.FrameID, frame.EventMS.Int64, frame.TimestampMS.Int64))) + point.RecordID = "raw-sha256:" + hex.EncodeToString(identity[:]) + content := sha256.Sum256([]byte(frame.Parsed + "\x00" + frame.ParseStatus)) + point.SourceDataVersion = "sha256:" + hex.EncodeToString(content[:]) + fail := func(status, reason, message string) HistoricalHydrogenPoint { + point.Status = status + point.ReasonCode = reason + point.Message = message + return point + } + if frame.ParseStatus != "OK" { + return fail("INVALID", "RAW_FRAME_PARSE_FAILED", "历史原始记录解析异常") + } + var fields map[string]any + if json.Unmarshal([]byte(frame.Parsed), &fields) != nil { + return fail("INVALID", "INVALID_RAW_JSON", "历史原始数据格式异常") + } + for _, key := range hydrogenMassFields { + value, exists := fields[key] + if !exists { + continue + } + mass, valid := numericValue(value) + point.ValueSource = "REPORTED" + point.CalculationVersion = "REPORTED_HYDROGEN_KG_V1" + if !valid || math.IsNaN(mass) || math.IsInf(mass, 0) || mass < 0 || mass > 200 { + return fail("INVALID", "INVALID_MASS_READING", "历史剩余氢质量读数无效") + } + point.RemainingHydrogenKg = &mass + point.Status = "NORMAL" + return point + } + pv, pExists := fields[realtimeHydrogenPressureField] + tv, tExists := fields[realtimeHydrogenTemperatureField] + if !pExists && !tExists { + return fail("MISSING", "MISSING_HYDROGEN_MEASUREMENT", "历史原始记录缺少氢质量及温压") + } + pressure, pOK := numericValue(pv) + temperature, tOK := numericValue(tv) + if (pExists && (!pOK || !finiteRealtime(pressure) || pressure < 0 || pressure > 70)) || (tExists && (!tOK || !finiteRealtime(temperature) || temperature < -40 || temperature > 726.85)) { + return fail("INVALID", "INVALID_PRESSURE_TEMPERATURE", "历史温度或压力读数无效") + } + if !pExists || !tExists { + return fail("MISSING", "INCOMPLETE_PRESSURE_TEMPERATURE", "历史同帧温度或压力缺失") + } + point.EstimatePressureMPa = &pressure + point.EstimateTemperatureC = &temperature + point.PressureTemperatureSource = "MAX_SENSOR_AGGREGATE" + // Current capacity, current stream state and daily aggregate parameters do + // not establish a versioned total tank volume valid at this historical event. + return fail("MISSING", "MISSING_HISTORICAL_CAPACITY_VERSION", "缺少适用于该历史时点的储氢容积版本证据") +} + +func (r *MySQLRepository) hydrateHistoricalHydrogenFrame(ctx context.Context, frame *historicalHydrogenFrame) error { + var manifest struct { + Chunked bool `json:"chunked"` + PayloadKind string `json:"payload_kind"` + EventID string `json:"event_id"` + ChunkCount int `json:"chunk_count"` + } + if json.Unmarshal([]byte(frame.Parsed), &manifest) != nil || !manifest.Chunked { + return nil + } + if manifest.ChunkCount <= 0 || manifest.ChunkCount > 256 || (manifest.PayloadKind != "parsed_fields" && manifest.PayloadKind != "parsed_json") || (manifest.EventID != "" && manifest.EventID != frame.EventID) { + return errors.New("invalid historical hydrogen chunk manifest") + } + quote := func(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" } + query := `SELECT chunk_index,chunk_count,chunk_text FROM ` + r.tdDatabase + `.raw_frame_payload_chunks WHERE vin=` + quote(frame.VIN) + ` AND protocol=` + quote(frame.Protocol) + ` AND event_id=` + quote(frame.EventID) + ` AND frame_id=` + quote(frame.FrameID) + ` AND payload_kind=` + quote(manifest.PayloadKind) + ` ORDER BY chunk_index LIMIT ` + strconv.Itoa(manifest.ChunkCount+1) + rows, err := r.tdengine.QueryContext(ctx, query) + if err != nil { + return err + } + defer rows.Close() + parts := make([]string, manifest.ChunkCount) + seen := make([]bool, manifest.ChunkCount) + count := 0 + for rows.Next() { + var index, total int + var part string + if err := rows.Scan(&index, &total, &part); err != nil { + return err + } + if index < 0 || index >= len(parts) || seen[index] || total != len(parts) { + return errors.New("inconsistent historical hydrogen chunks") + } + seen[index] = true + parts[index] = part + count++ + } + if err := rows.Err(); err != nil { + return err + } + if count != len(parts) { + return errors.New("incomplete historical hydrogen chunks") + } + frame.Parsed = strings.Join(parts, "") + return nil +} + +// Read a second candidate to detect ambiguous ordering across TDengine child +// tables. Never select arbitrarily or skip a tied hydrogen frame when advancing +// past a hydrated non-hydrogen candidate. +func (r *MySQLRepository) loadHistoricalHydrogenCandidate(ctx context.Context, query string) (historicalHydrogenFrame, error) { + rows, err := r.tdengine.QueryContext(ctx, query) + if err != nil { + return historicalHydrogenFrame{}, err + } + defer rows.Close() + var first historicalHydrogenFrame + scan := func(frame *historicalHydrogenFrame) error { + return rows.Scan(&frame.VIN, &frame.Protocol, &frame.EventID, &frame.FrameID, &frame.EventMS, &frame.ReceivedMS, &frame.TimestampMS, &frame.Parsed, &frame.ParseStatus) + } + if !rows.Next() { + if err := rows.Err(); err != nil { + return first, err + } + return first, sql.ErrNoRows + } + if err := scan(&first); err != nil { + return first, err + } + if rows.Next() { + var second historicalHydrogenFrame + if err := scan(&second); err != nil { + return first, err + } + if first.EventMS == second.EventMS && first.TimestampMS == second.TimestampMS { + return first, errors.New("ambiguous historical hydrogen records at identical collection and storage timestamps") + } + } + if err := rows.Err(); err != nil { + return first, err + } + return first, nil +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository_test.go b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository_test.go new file mode 100644 index 00000000..0d4a17c0 --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_repository_test.go @@ -0,0 +1,146 @@ +package openplatform + +import ( + "context" + "database/sql" + "github.com/DATA-DOG/go-sqlmock" + "strings" + "testing" + "time" +) + +func historicalTestFrame(parsed string) historicalHydrogenFrame { + return historicalHydrogenFrame{VIN: "V1", Protocol: "GB32960", EventID: "E1", FrameID: "F1", Parsed: parsed, ParseStatus: "OK", EventMS: sql.NullInt64{Int64: HistoricalHydrogenStart.Add(time.Hour).UnixMilli(), Valid: true}, ReceivedMS: sql.NullInt64{Int64: HistoricalHydrogenStart.Add(2 * time.Hour).UnixMilli(), Valid: true}} +} +func TestHistoricalHydrogenRawPrecisionAndInvalidPriority(t *testing.T) { + for _, tc := range []struct { + parsed, status, reason string + mass *float64 + }{ + {`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":0}`, "NORMAL", "", realLiveFloat(0)}, + {`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":32.658123456}`, "NORMAL", "", realLiveFloat(32.658123456)}, + {`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":-1}`, "INVALID", "INVALID_MASS_READING", nil}, + {`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":null,"gd_fc_vehicle_hydrogen_mass_kg":10}`, "INVALID", "INVALID_MASS_READING", nil}, + {fmtRealtimePT(35, 15), "MISSING", "MISSING_HISTORICAL_CAPACITY_VERSION", nil}, + {fmtRealtimePT(0, 15), "MISSING", "MISSING_HISTORICAL_CAPACITY_VERSION", nil}, + {`{"gb32960.fuel_cell.max_hydrogen_pressure_mpa":null}`, "INVALID", "INVALID_PRESSURE_TEMPERATURE", nil}, + {`{"gb32960.fuel_cell.max_hydrogen_pressure_mpa":20}`, "MISSING", "INCOMPLETE_PRESSURE_TEMPERATURE", nil}, + } { + got := historicalHydrogenFromFrame(historicalTestFrame(tc.parsed)) + if got.Status != tc.status || got.ReasonCode != tc.reason || (got.RemainingHydrogenKg == nil) != (tc.mass == nil) || tc.mass != nil && *got.RemainingHydrogenKg != *tc.mass { + t.Fatalf("%+v", got) + } + } + frame := historicalTestFrame(`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":1}`) + first := historicalHydrogenFromFrame(frame) + frame.Parsed = `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":2}` + second := historicalHydrogenFromFrame(frame) + if first.RecordID != second.RecordID || first.SourceDataVersion == second.SourceDataVersion || first.UpdatedAt != second.UpdatedAt { + t.Fatal("revision tracking invalid") + } + frame.ParseStatus = "ERROR" + if got := historicalHydrogenFromFrame(frame); got.Status != "INVALID" { + t.Fatal(got) + } +} +func TestHistoricalHydrogenQueryEventTimeAndSafeBounds(t *testing.T) { + q, err := historicalHydrogenQuery("vehicle_ts", "VIN'1", HistoricalHydrogenStart.Add(time.Hour), "GB32960") + if err != nil { + t.Fatal(err) + } + for _, want := range []string{"vin='VIN''1'", "event_time>=", "event_time<=", "ORDER BY event_time DESC,ts DESC LIMIT 2", `"chunked":true`} { + if !strings.Contains(q, want) { + t.Fatal(q) + } + } + for _, bad := range []string{"parse_status='OK'", "ts<=", "IS NOT NULL"} { + if strings.Contains(q, bad) { + t.Fatal(q) + } + } + if _, err := historicalHydrogenQuery("bad;sql", "V1", HistoricalHydrogenStart, "GB32960"); err == nil { + t.Fatal("unsafe db") + } +} +func TestHistoricalHydrogenRepositoryPreservesDelayedUpload(t *testing.T) { + td, mock, _ := sqlmock.New() + defer td.Close() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + at := HistoricalHydrogenStart.Add(time.Hour) + mock.ExpectQuery("SELECT vin,protocol,event_id,frame_id").WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol", "event_id", "frame_id", "event", "received", "ts", "json", "status"}).AddRow("V1", "GB32960", "E1", "F1", at.UnixMilli(), at.Add(time.Hour).UnixMilli(), at.Add(time.Hour).UnixMilli(), `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":0}`, "OK")) + got, err := r.HistoricalHydrogen(context.Background(), "V1", at, "GB32960") + if err != nil || got == nil || got.Status != "NORMAL" || !got.ReceivedAt.After(at) { + t.Fatalf("%+v %v", got, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} +func TestHistoricalHydrogenChunkMissingFailsInsteadOfFallback(t *testing.T) { + td, mock, _ := sqlmock.New() + defer td.Close() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + frame := historicalTestFrame(`{"chunked":true,"payload_kind":"parsed_fields","event_id":"E1","chunk_count":2}`) + mock.ExpectQuery("SELECT chunk_index,chunk_count,chunk_text").WillReturnRows(sqlmock.NewRows([]string{"index", "count", "text"}).AddRow(0, 2, `{"x":`)) + if err := r.hydrateHistoricalHydrogenFrame(context.Background(), &frame); err == nil { + t.Fatal("missing chunk silently accepted") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} +func TestHistoricalHydrogenChunkHydration(t *testing.T) { + td, mock, _ := sqlmock.New() + defer td.Close() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + frame := historicalTestFrame(`{"chunked":true,"payload_kind":"parsed_fields","event_id":"E1","chunk_count":2}`) + mock.ExpectQuery("SELECT chunk_index,chunk_count,chunk_text").WillReturnRows(sqlmock.NewRows([]string{"index", "count", "text"}).AddRow(0, 2, `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":`).AddRow(1, 2, `0}`)) + if err := r.hydrateHistoricalHydrogenFrame(context.Background(), &frame); err != nil { + t.Fatal(err) + } + got := historicalHydrogenFromFrame(frame) + if got.Status != "NORMAL" || *got.RemainingHydrogenKg != 0 { + t.Fatal(got) + } +} + +func TestHistoricalHydrogenNewestInvalidNeverQueriesOlder(t *testing.T) { + td, mock, _ := sqlmock.New() + defer td.Close() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + at := HistoricalHydrogenStart.Add(time.Hour) + mock.ExpectQuery("SELECT vin,protocol,event_id,frame_id").WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol", "event_id", "frame_id", "event", "received", "ts", "json", "status"}).AddRow("V1", "GB32960", "E1", "F1", at.UnixMilli(), at.UnixMilli(), at.UnixMilli(), `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":null}`, "OK")) + got, err := r.HistoricalHydrogen(context.Background(), "V1", at, "GB32960") + if err != nil || got == nil || got.Status != "INVALID" { + t.Fatalf("%+v %v", got, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} +func TestHistoricalHydrogenRejectsFutureReturnedRow(t *testing.T) { + td, mock, _ := sqlmock.New() + defer td.Close() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + at := HistoricalHydrogenStart.Add(time.Hour) + mock.ExpectQuery("SELECT vin,protocol,event_id,frame_id").WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol", "event_id", "frame_id", "event", "received", "ts", "json", "status"}).AddRow("V1", "GB32960", "E1", "F1", at.Add(time.Second).UnixMilli(), at.UnixMilli(), at.UnixMilli(), `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":10}`, "OK")) + if _, err := r.HistoricalHydrogen(context.Background(), "V1", at, "GB32960"); err == nil { + t.Fatal("future row accepted") + } +} + +func TestHistoricalHydrogenTiedCandidatesFailWithoutSkipping(t *testing.T) { + for _, parsed := range []string{`{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":10}`, `{"chunked":true,"payload_kind":"parsed_fields","event_id":"E1","chunk_count":1}`} { + td, mock, _ := sqlmock.New() + r := &MySQLRepository{tdengine: td, tdDatabase: "vehicle_ts"} + at := HistoricalHydrogenStart.Add(time.Hour) + mock.ExpectQuery("SELECT vin,protocol,event_id,frame_id").WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol", "event_id", "frame_id", "event", "received", "ts", "json", "status"}).AddRow("V1", "GB32960", "E1", "F1", at.UnixMilli(), at.UnixMilli(), at.UnixMilli(), parsed, "OK").AddRow("V1", "GB32960", "E2", "F2", at.UnixMilli(), at.UnixMilli(), at.UnixMilli(), `{"gb32960.gd_fc_vehicle_info.hydrogen_mass_kg":20}`, "OK")) + if _, err := r.HistoricalHydrogen(context.Background(), "V1", at, "GB32960"); err == nil || !strings.Contains(err.Error(), "ambiguous") { + t.Fatalf("expected ambiguity, got %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } + td.Close() + } +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_test.go b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_test.go new file mode 100644 index 00000000..a9a3f12c --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/openplatform/historical_hydrogen_test.go @@ -0,0 +1,346 @@ +package openplatform + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" +) + +const historicalTestVIN = "LA9GG68L2PBAF4773" +const historicalTestKey = "0123456789abcdef0123456789abcdef" + +type historicalTestRepository struct { + *fakeRepository + lookup func(context.Context, string, time.Time, string) (*HistoricalHydrogenPoint, error) + allowed func(string, time.Time, time.Time) (bool, error) + lookups atomic.Int32 +} + +func (r *historicalTestRepository) HistoricalHydrogen(ctx context.Context, vin string, at time.Time, protocol string) (*HistoricalHydrogenPoint, error) { + r.lookups.Add(1) + if r.lookup != nil { + return r.lookup(ctx, vin, at, protocol) + } + return nil, nil +} +func (r *historicalTestRepository) HistoricalHydrogenAuthorized(_ context.Context, _ uint64, vin string, at, now time.Time) (bool, error) { + if r.allowed != nil { + return r.allowed(vin, at, now) + } + return true, nil +} +func historicalTestService() (*Service, *historicalTestRepository, time.Time) { + repository := &historicalTestRepository{fakeRepository: &fakeRepository{app: AppCredential{ID: 9}}} + service := NewService(repository) + now := time.Date(2026, 9, 9, 12, 0, 0, 0, service.location) + service.now = func() time.Time { return now } + return service, repository, now +} +func historicalTestRequest(at time.Time) HistoricalHydrogenRequest { + return HistoricalHydrogenRequest{Queries: []HistoricalHydrogenQuery{{RequestID: "start", VIN: historicalTestVIN, Time: at.Format("2006-01-02 15:04:05")}}} +} +func historicalTestPoint(at time.Time, kg float64) *HistoricalHydrogenPoint { + return &HistoricalHydrogenPoint{VIN: historicalTestVIN, Protocol: "GB32960", ObservedAt: at, RemainingHydrogenKg: &kg, Status: "NORMAL", ValueSource: "REPORTED", RecordID: "record", SourceDataVersion: "data-v1", CalculationVersion: "decode-v1", UpdatedAt: at} +} + +func TestHistoricalHydrogenValidation(t *testing.T) { + service, _, now := historicalTestService() + for _, tc := range []struct { + name string + mutate func(*HistoricalHydrogenRequest) + }{ + {"empty", func(r *HistoricalHydrogenRequest) { r.Queries = nil }}, + {"too many", func(r *HistoricalHydrogenRequest) { r.Queries = make([]HistoricalHydrogenQuery, 201) }}, + {"duplicate id", func(r *HistoricalHydrogenRequest) { r.Queries = append(r.Queries, r.Queries[0]) }}, + {"blank id", func(r *HistoricalHydrogenRequest) { r.Queries[0].RequestID = " " }}, + {"bad VIN", func(r *HistoricalHydrogenRequest) { r.Queries[0].VIN = "LA9GG68L2PBAI4773" }}, + {"fractional time", func(r *HistoricalHydrogenRequest) { r.Queries[0].Time = "2026-09-08 12:00:00.1" }}, + {"future", func(r *HistoricalHydrogenRequest) { + r.Queries[0].Time = now.Add(time.Second).Format("2006-01-02 15:04:05") + }}, + {"before coverage", func(r *HistoricalHydrogenRequest) { r.Queries[0].Time = "2026-07-31 23:59:59" }}, + {"protocol alias", func(r *HistoricalHydrogenRequest) { r.Queries[0].Protocol = "MQTT" }}, + {"negative tolerance", func(r *HistoricalHydrogenRequest) { v := -1; r.MaxTimeDifferenceSeconds = &v }}, + {"too large tolerance", func(r *HistoricalHydrogenRequest) { v := 301; r.MaxTimeDifferenceSeconds = &v }}, + } { + t.Run(tc.name, func(t *testing.T) { + request := historicalTestRequest(now.Add(-time.Hour)) + tc.mutate(&request) + if _, _, err := service.validateHistoricalHydrogen(request, now); !errors.Is(err, ErrInvalidRequest) { + t.Fatalf("err=%v", err) + } + }) + } + request := historicalTestRequest(now) + request.Queries[0].RequestID = " original id " + parsed, tolerance, err := service.validateHistoricalHydrogen(request, now) + if err != nil || tolerance != 300 || parsed[0].Protocol != "GB32960" || parsed[0].RequestID != " original id " { + t.Fatalf("%+v %d %v", parsed, tolerance, err) + } +} + +func TestHistoricalHydrogenNormalZeroPrecisionStaleAndInvalid(t *testing.T) { + for _, tc := range []struct { + name string + kg float64 + age time.Duration + tolerance int + status string + }{ + {"zero", 0, 0, 0, "NORMAL"}, {"precision", 12.3456789, time.Second, 300, "NORMAL"}, {"exact tolerance", 2, 300 * time.Second, 300, "NORMAL"}, {"stale", 2, 300*time.Second + time.Millisecond, 300, "STALE"}, {"not exact", 2, time.Millisecond, 0, "STALE"}, {"invalid negative", -1, 0, 300, "INVALID"}, + } { + t.Run(tc.name, func(t *testing.T) { + service, repository, now := historicalTestService() + at := now.Add(-time.Hour) + repository.lookup = func(context.Context, string, time.Time, string) (*HistoricalHydrogenPoint, error) { + return historicalTestPoint(at.Add(-tc.age), tc.kg), nil + } + request := historicalTestRequest(at) + request.MaxTimeDifferenceSeconds = &tc.tolerance + results, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", request) + if err != nil { + t.Fatal(err) + } + got := results[0] + if got.RemainingHydrogenKgStatus != tc.status { + t.Fatalf("%+v", got) + } + if tc.status == "NORMAL" { + if got.RemainingHydrogenKg == nil || *got.RemainingHydrogenKg != tc.kg { + t.Fatal("lost zero or precision") + } + } else if got.RemainingHydrogenKg != nil { + t.Fatal("non-normal quantity exposed") + } + if got.TimeDifferenceSeconds == nil || *got.TimeDifferenceSeconds != tc.age.Seconds() { + t.Fatalf("time delta=%v", got.TimeDifferenceSeconds) + } + }) + } +} + +func TestHistoricalHydrogenBatchForbiddenNeverLeaksSample(t *testing.T) { + service, repository, now := historicalTestService() + at := now.Add(-time.Hour) + repository.lookup = func(_ context.Context, _ string, at time.Time, _ string) (*HistoricalHydrogenPoint, error) { + return historicalTestPoint(at.Add(-time.Minute), 4), nil + } + repository.allowed = func(_ string, sample, _ time.Time) (bool, error) { return !sample.Before(at), nil } + results, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", historicalTestRequest(at)) + if err != nil { + t.Fatal(err) + } + got := results[0] + if got.RemainingHydrogenKgStatus != "FORBIDDEN" || got.RemainingHydrogenKg != nil || got.HydrogenRecordTime != nil || got.SourceRecordID != nil || got.HydrogenSourceProtocol != nil || got.SourceDataVersion != nil || got.UpdatedAt != nil { + t.Fatalf("sample leaked: %+v", got) + } + repository.allowed = func(string, time.Time, time.Time) (bool, error) { return false, nil } + repository.lookups.Store(0) + results, err = service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", historicalTestRequest(at)) + if err != nil || repository.lookups.Load() != 0 || results[0].RemainingHydrogenKgStatus != "FORBIDDEN" { + t.Fatalf("queried forbidden data: %+v %v", results, err) + } +} + +func TestHistoricalHydrogenMixedBatchAndNoProtocolFallback(t *testing.T) { + service, repository, now := historicalTestService() + at := now.Add(-time.Hour) + repository.lookup = func(_ context.Context, _ string, sample time.Time, _ string) (*HistoricalHydrogenPoint, error) { + if sample.Equal(at) { + return nil, errors.New("private driver query") + } + return nil, nil + } + request := historicalTestRequest(at) + request.Queries = append(request.Queries, HistoricalHydrogenQuery{RequestID: "none", VIN: historicalTestVIN, Time: at.Add(-time.Minute).Format("2006-01-02 15:04:05")}, HistoricalHydrogenQuery{RequestID: "unsupported", VIN: historicalTestVIN, Time: at.Format("2006-01-02 15:04:05"), Protocol: "JT808"}) + results, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", request) + if err != nil { + t.Fatal(err) + } + for i, want := range []string{"ERROR", "NO_DATA", "UNSUPPORTED"} { + if results[i].RequestID != request.Queries[i].RequestID || results[i].RemainingHydrogenKgStatus != want { + t.Fatalf("%+v", results) + } + } + if repository.lookups.Load() != 2 { + t.Fatalf("fallback read count=%d", repository.lookups.Load()) + } + encoded, _ := json.Marshal(results) + if strings.Contains(string(encoded), "private") { + t.Fatal("driver detail leaked") + } +} + +func TestHistoricalHydrogenRejectsFutureOrWrongIdentitySample(t *testing.T) { + for _, kind := range []string{"future", "vin", "protocol"} { + t.Run(kind, func(t *testing.T) { + service, repository, now := historicalTestService() + at := now.Add(-time.Hour) + repository.lookup = func(context.Context, string, time.Time, string) (*HistoricalHydrogenPoint, error) { + point := historicalTestPoint(at, 4) + switch kind { + case "future": + point.ObservedAt = at.Add(time.Second) + case "vin": + point.VIN = "OTHER" + case "protocol": + point.Protocol = "JT808" + } + return point, nil + } + result, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", historicalTestRequest(at)) + if err != nil || result[0].RemainingHydrogenKgStatus != "ERROR" || result[0].SourceRecordID != nil { + t.Fatalf("%+v %v", result, err) + } + }) + } +} + +func TestHistoricalHydrogenDeadlineKeepsEveryRequestID(t *testing.T) { + service, repository, now := historicalTestService() + repository.lookup = func(ctx context.Context, _ string, _ time.Time, _ string) (*HistoricalHydrogenPoint, error) { + <-ctx.Done() + return nil, ctx.Err() + } + request := historicalTestRequest(now.Add(-time.Hour)) + for i := 1; i < 8; i++ { + query := request.Queries[0] + query.RequestID = string(rune('a' + i)) + request.Queries = append(request.Queries, query) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + results, err := service.QueryHistoricalHydrogen(ctx, historicalTestKey, "trace", request) + if err != nil || len(results) != 8 { + t.Fatalf("%+v %v", results, err) + } + for i, got := range results { + if got.RequestID != request.Queries[i].RequestID || got.RemainingHydrogenKgStatus != "ERROR" || got.RemainingHydrogenKg != nil { + t.Fatalf("%+v", got) + } + } + if repository.lookups.Load() > 4 { + t.Fatal("more than four workers queried") + } +} + +func TestHistoricalHydrogenHandlerAuthenticationAndRateLimit(t *testing.T) { + service, repository, now := historicalTestService() + handler := NewDataHandler(service) + request := historicalTestRequest(now.Add(-time.Hour)) + request.Queries[0].Protocol = "JT808" + body, _ := json.Marshal(request) + call := func() *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodPost, HistoricalHydrogenQueryPath, strings.NewReader(string(body))) + r.Header.Set("Authorization", "Bearer "+historicalTestKey) + w := httptest.NewRecorder() + handler.ServeHTTP(w, r) + return w + } + repository.authErr = ErrUnauthorized + if w := call(); w.Code != 401 { + t.Fatalf("status=%d body=%s", w.Code, w.Body.String()) + } + repository.authErr = nil + for i := 0; i < 30; i++ { + if w := call(); w.Code != 200 { + t.Fatalf("request %d status=%d body=%s", i, w.Code, w.Body.String()) + } + } + w := call() + if w.Code != 429 || w.Header().Get("Retry-After") != "60" || !strings.Contains(w.Body.String(), "RATE_LIMITED") { + t.Fatalf("status=%d headers=%v body=%s", w.Code, w.Header(), w.Body.String()) + } + if !IsPublicPath(HistoricalHydrogenQueryPath) { + t.Fatal("missing public route") + } +} +func TestHistoricalHydrogenConcurrentLimitAndWindowReset(t *testing.T) { + var limiter historicalHydrogenLimiter + now := time.Now() + release1, err := limiter.acquire(1, now) + if err != nil { + t.Fatal(err) + } + release2, err := limiter.acquire(1, now) + if err != nil { + t.Fatal(err) + } + if _, err := limiter.acquire(1, now); err == nil { + t.Fatal("third concurrent batch allowed") + } + release1() + release2() + for i := 2; i < 30; i++ { + release, err := limiter.acquire(1, now) + if err != nil { + t.Fatal(err) + } + release() + } + if _, err := limiter.acquire(1, now); err == nil { + t.Fatal("31st batch allowed") + } + release, err := limiter.acquire(1, now.Add(time.Minute)) + if err != nil { + t.Fatal(err) + } + release() +} +func TestHistoricalHydrogenAuthorizationChecksAppAndGrantAtBothTimes(t *testing.T) { + db, mock, _ := sqlmock.New() + defer db.Close() + now := time.Now() + at := now.Add(-time.Hour) + mock.ExpectQuery("SELECT EXISTS.*a.valid_to>\\?.*a.valid_to>\\?.*BINARY g.vin=BINARY \\?.*g.valid_to>\\?").WithArgs(uint64(9), now, now, at, at, historicalTestVIN, at, at).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + allowed, err := NewMySQLRepository(db).HistoricalHydrogenAuthorized(context.Background(), 9, historicalTestVIN, at, now) + if err != nil || allowed { + t.Fatalf("allowed=%v err=%v", allowed, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} + +func TestHistoricalHydrogenTwoHundredItemsShareEighteenLookups(t *testing.T) { + service, repository, now := historicalTestService() + repository.lookup = func(_ context.Context, _ string, at time.Time, _ string) (*HistoricalHydrogenPoint, error) { + return historicalTestPoint(at, 12.3456789), nil + } + var request HistoricalHydrogenRequest + for i := 0; i < 200; i++ { + request.Queries = append(request.Queries, HistoricalHydrogenQuery{RequestID: strconv.Itoa(i), VIN: historicalTestVIN, Time: now.Add(-time.Duration(i%18) * time.Minute).Format("2006-01-02 15:04:05")}) + } + results, err := service.QueryHistoricalHydrogen(context.Background(), historicalTestKey, "trace", request) + if err != nil || len(results) != 200 || repository.lookups.Load() != 18 { + t.Fatalf("len=%d lookups=%d err=%v", len(results), repository.lookups.Load(), err) + } + for i, got := range results { + if got.RequestID != strconv.Itoa(i) || got.QueryTime != request.Queries[i].Time || got.RemainingHydrogenKgStatus != "NORMAL" || got.RemainingHydrogenKg == nil || *got.RemainingHydrogenKg != 12.3456789 { + t.Fatalf("index=%d result=%+v", i, got) + } + } +} + +func TestHistoricalHydrogenDecoderRejectsNullAndUnknownFields(t *testing.T) { + for _, body := range []string{ + `{"queries":[{"requestId":"a","vin":"LA9GG68L2PBAF4773","time":"2026-08-01 00:00:00"}],"maxTimeDifferenceSeconds":null}`, + `{"queries":[{"requestId":"a","vin":"LA9GG68L2PBAF4773","time":"2026-08-01 00:00:00","protocol":null}]}`, + `{"queries":[{"requestId":"a","vin":"LA9GG68L2PBAF4773","time":"2026-08-01 00:00:00","unexpected":true}]}`, + `{"queries":[],"unexpected":true}`, + } { + var request HistoricalHydrogenRequest + if err := json.Unmarshal([]byte(body), &request); err == nil { + t.Fatalf("accepted %s", body) + } + } +} diff --git a/vehicle-data-platform/apps/api/internal/openplatform/service.go b/vehicle-data-platform/apps/api/internal/openplatform/service.go index 0f69047c..a8c6c10b 100644 --- a/vehicle-data-platform/apps/api/internal/openplatform/service.go +++ b/vehicle-data-platform/apps/api/internal/openplatform/service.go @@ -347,9 +347,10 @@ func validCoordinate(longitude, latitude float64) bool { } type Service struct { - repository Repository - now func() time.Time - location *time.Location + historicalLimiter historicalHydrogenLimiter + repository Repository + now func() time.Time + location *time.Location } type parsedGrant struct { diff --git a/vehicle-data-platform/docs/oneos-historical-hydrogen-api-contract.md b/vehicle-data-platform/docs/oneos-historical-hydrogen-api-contract.md new file mode 100644 index 00000000..87050765 --- /dev/null +++ b/vehicle-data-platform/docs/oneos-historical-hydrogen-api-contract.md @@ -0,0 +1,87 @@ +# 历史剩余氢量接口契约 + +契约版本：1.10.0；修订日期：2026-09-09。此文档定义实现与使用边界，不代替上线记录或真实历史覆盖验收报告。 + +## 用途与业务边界 + +`POST /api/v1/vehicles/hydrogen-remaining/history/query` 批量查询指定历史时刻的全车剩余氢质量。认证继续使用 `Authorization: Bearer `，外层继续返回 `code / message / data / traceId`。实时剩余氢量和日用氢量接口保持原口径。 + +Seeker 负责保存和计算： + +```text +rawHydrogenDeltaKg = 起点剩余氢质量 − 终点剩余氢质量 +hydrogenConsumptionKg = max(rawHydrogenDeltaKg, 0) +negativeDeltaClamped = rawHydrogenDeltaKg < 0 +``` + +两端均 NORMAL 且来源、协议、转换版本、容量版本可比时才计算，使用 API 返回的未额外舍入质量值，最后按展示需要舍入。任一端缺失、未授权、异常或不可比较则结果 null。真实两端质量相同、或负差值归零，可以产生0；负的剩余质量本身属于 INVALID。此公式是业务约定的两端储氢净减少量，不代表区间内加氢修正后的实际燃料电池消耗量；不能用日用氢量替代或按时长/里程分摊。 + +## 请求与授权 + +| 参数 | 约束 | +| --- | --- | +| queries | 必填，1–200项，单点也使用数组 | +| queries[].requestId | 必填、非空、批内唯一、最多128字节，原样回显 | +| queries[].vin | 规范化大写17位VIN，不含I/O/Q | +| queries[].time | 严格北京时间 `yyyy-MM-dd HH:mm:ss`；从2026-08-01 00:00:00起，不允许未来时刻，更早或非法为400 | +| queries[].protocol | GB32960/YUTONG_MQTT/JT808；省略固定GB32960，不跨协议回退，后两者当前UNSUPPORTED | +| maxTimeDifferenceSeconds | 整数0–300，默认300；0要求精确采样，不静默放宽 | + +appKey当前无效、停用或过期返回401。合法批量返回HTTP200/SUCCESS，各项以 remainingHydrogenKgStatus 判定；查询时刻及实际采样时刻都必须落在应用和逐车授权范围内，授权结束时间为不含端点。单项FORBIDDEN仍回显requestId/vin/queryTime，采样时间、记录标识、来源、版本等均不泄露。整个请求结构非法返回400，不提交部分查询。 + +每个app、每个服务实例限制30批/60秒，最多2个并发批；429的 `Retry-After` 为等待秒数。整个批次9秒预算，最多4个并发工作协程；失败/超时明确为单项ERROR，其他已完成项仍可用。客户端按Retry-After或有限退避重试，不将错误作为无数据缓存；建议按VIN+time+protocol+容差去重，并将作业起终点放同批核对。此限制不是全服务集群共享令牌桶。**200项是请求结构上限，不是200个不同点在9秒完成的性能保证**；建议首次使用20个不同点，出现超时后缩小批量并退避重试。重复点较多的200项测试不能证明200个不同点的吞吐能力，具体候选性能证据见发布验收记录。 + +## 历史证据要求 + +匹配使用真实 `event_time`，只能采用不晚于查询时刻的历史氢相关采样。最新氢相关帧异常或字段不全时显式返回异常/缺失，不能跳过它拿更旧正常值掩盖。请求时间、接收时间、数据库更新时间和当前实时快照均不能冒充采集时间；不插值、不把日统计还原成瞬时质量。 + +有终端质量上报时 `hydrogenValueSource=REPORTED` 仅确认终端上报，不能保证终端内部直接测量。平台估算 `ESTIMATED` 必须使用历史同帧温压及适用于该历史时点、有可追溯版本的 VIN 全车水容积。**当前容积表的现值不是历史有效性证明**，缺少历史适用证据时返回 MISSING，不把修改后的当前容积套回8月记录。最大温压聚合也不保证来自同一瓶，仍需披露选取方法。 + +两端可能命中同一原始帧，调用方必须保留 sourceRecordId 与实际采集时间识别采样分辨率不足。同批按规范化VIN+time+protocol去重，同一点共享一次查询/授权结果与来源版本，每个requestId仍各有独立结果；不同点不保证原子数据快照，同批不等于所有点同一修订版本。保存每项数据版本、算法/配置版本、更新时间及 traceId，结果发生变化时重新核对两端。 + +## 返回字段与状态 + +所有字段存在，无法提供时为null；requestId、vin、queryTime和remainingHydrogenKgStatus为非空基本字段。 + +| 字段 | 当前实现 | +| --- | --- | +| plateNumber | 无历史车牌证据，null，不用当前车牌冒充 | +| remainingHydrogenKg | 仅NORMAL返回终端上报有效0–200kg；不额外round3，保留解析后数值精度 | +| hydrogenRecordTime / timeDifferenceSeconds | 真实event_time（RFC3339带时区）及与查询时刻的秒差，可含小数 | +| hydrogenValueSource / hydrogenSourceProtocol | 当前有效结果REPORTED / GB32960；ESTIMATED为契约预留，目前缺历史容量版本不输出估算值 | +| sourceRecordId | raw-sha256采样身份指纹，不包含底层连接配置 | +| sourceDataVersion | sha256原始JSON及parseStatus指纹；同一采样原地修订即使received_at不变也能识别 | +| hydrogenCalculationVersion | 当前REPORTED_HYDROGEN_KG_V1；算法口径版本不同默认不混算 | +| hydrogenCapacityVersion / hydrogenTankCapacityL | 当前无历史适用容积版本，两者null，不读取当前表假装历史有效 | +| updatedAt | 原始记录received_at，非API当前时间，也不是采集时间 | +| hydrogenEstimatePressureMPa / hydrogenEstimateTemperatureC / hydrogenPressureTemperatureSource | 有效历史同帧温压可供诊断返回，方法MAX_SENSOR_AGGREGATE；缺历史容量仍MISSING | +| reasonCode / message | 机器原因与中文说明；非正常情况可解释，不暴露底层凭证 | + +sourceDataVersion是单条内容修订证据，起终点本来就是不同内容，**不要求两端指纹相等**。可比性由VIN、数据来源、协议、算法及适用容量版本共同判断；更新时重新核对并留存两端证据。 + +| 状态 | 使用边界 | +| --- | --- | +| NORMAL | 有效质量且采样在容差内，才可参与已核实可比的差值计算 | +| NO_DATA | 支持的历史查询范围内没有氢相关帧；不代表更早数据不存在 | +| MISSING | 最近氢相关帧缺量、温压不全或缺历史容量版本；不跳过最新缺失帧 | +| STALE | 有效质量命中超容差，数值null，可返回采样诊断信息 | +| UNSUPPORTED | 显式请求当前不支持氢量的协议 | +| INVALID | 最新负质量、无效读数、温压异常或原始解析异常，数值null | +| FORBIDDEN | 查询时刻或采样时刻授权不满足，采样元数据全部null | +| ERROR | 超时、上游失败、分块重组失败或候选预算耗尽；可以重试，不当查无数据 | + +状态优先保留最新原始帧的INVALID/MISSING；只有原始质量有效时才因超容差变为STALE，不能用陈旧状态掩盖异常。分块报文会重组后确认是否氢相关；采集时间与入库时间均相同的候选发生冲突时返回ERROR，不任意择值；不完整分块处理失败为ERROR，最多检查32个候选的预算耗尽也为ERROR。 + +常见原因：NO_HISTORICAL_SAMPLE、SAMPLE_OUTSIDE_TOLERANCE、HISTORY_FORBIDDEN、PROTOCOL_UNSUPPORTED、UPSTREAM_TIMEOUT、UPSTREAM_FAILURE、INVALID_MASS_READING、INVALID_PRESSURE_TEMPERATURE、INCOMPLETE_PRESSURE_TEMPERATURE、MISSING_HISTORICAL_CAPACITY_VERSION、RAW_FRAME_PARSE_FAILED、INVALID_RAW_JSON。原因可随新增诊断扩展，客户端优先按状态处理未知原因。 + +## 覆盖与交付证据 + +接口仅支持2026-08-01北京时间起的请求与原始采样，较早请求400。这是接口能力边界，不表示此前所有原始数据不存在；更早保留范围不在本版承诺。日期可查询、授权覆盖、存在原始帧、存在有效氢量是四件不同的事，应逐车逐时点核对。生产原始历史采样测试、至少3车2段区间验证及缺失样本见本次发布验收报告；没有海港真实作业节点时，只能称历史采样区间验证，不宣称已通过海港作业联调。 + +## Seeker 接入检查 + +- VIN、requestId、queryTime 与请求逐项核对，不能以数组位置或当前车牌作为唯一关联依据。 +- 仅 NORMAL 进入差值计算；STALE、NO_DATA、MISSING、UNSUPPORTED、INVALID、FORBIDDEN、ERROR 均保持待数据/待核对，不补零。 +- 两端协议、REPORTED/ESTIMATED、转换版本及估算容量版本不同时默认不混算，须先核实可比性。 +- 原始质量、原始差值、归零标记、采样偏差、原始帧标识和版本均留存；展示舍入不回写覆盖计算证据。 +- 保留跨日作业两个时点，不切换为日报相减。集装箱分摊与碳排因子继续由 Seeker 维护。 diff --git a/vehicle-data-platform/docs/oneos-historical-hydrogen-api-release-20260909.md b/vehicle-data-platform/docs/oneos-historical-hydrogen-api-release-20260909.md new file mode 100644 index 00000000..f7147933 --- /dev/null +++ b/vehicle-data-platform/docs/oneos-historical-hydrogen-api-release-20260909.md @@ -0,0 +1,51 @@ +# 历史剩余氢量接口发布验收 + +## 发布结果 + +- 生产版本：`open-platform-historical-hydrogen-202609090100`；北京时间2026-09-09 00:57:48确认就绪。 +- 正式服务：[开放平台](https://open.d.lnoneos.com)，服务`lingniu-vehicle-open-platform`，端口20310。 +- 新接口：`POST /api/v1/vehicles/hydrogen-remaining/history/query`；[在线文档](https://open.d.lnoneos.com/open-api/docs/)及[OpenAPI 1.10.0](https://open.d.lnoneos.com/open-api/openapi.yaml)同步生效。 +- 最终候选与生产二进制SHA-256一致：`6bb05405b2dd126dd3a8fe13d226fd128a577154e00c2505bcba57f0aca5870b`。 +- 上一版`open-platform-hydrogen-estimate-202609082311`及原静态门户资源保留；无数据库迁移。发布脚本含失败自动回滚。 + +## 实现与边界 + +历史查询从北京时间2026-08-01起，按实际event_time匹配最近且不晚于请求时刻的氢相关原始帧，支持延迟上传；不会使用实时快照、日统计、插值或当前容积回套历史。缺历史容量版本的温压记录MISSING，当前有效历史质量仅REPORTED。真实零有效，负质量及显式无效值INVALID；非NORMAL质量null。最新异常不会被旧正常值掩盖。分块帧须完整重组，候选时间冲突或重组失败返回ERROR。 + +原始上报质量不额外舍入。返回采样时间、协议、稳定采样指纹sourceRecordId、独立内容版本sourceDataVersion、转换版本和接收时间updatedAt；内容修订即使保留接收时间也可通过指纹识别。历史车牌没有证据时null。采样与查询两个时刻均须满足应用和车辆授权，越权项不泄露采样元数据。 + +每批1–200项；requestId唯一并逐项回显。同批按VIN+time+protocol去重，相同点共享单次查询版本；不同点不保证原子快照。默认容差300秒，0要求精确命中。每应用每实例30批/60秒、2并发批；429附Retry-After。批次9秒、4工作协程，超时逐项ERROR。建议从20个不同点开始；200是结构上限，不能将重复点压测解读为200个不同历史点的性能保证。 + +完整契约见 `oneos-historical-hydrogen-api-contract.md`。作业净用氢量仍由Seeker按两端可比NORMAL值计算，不在平台新增日用氢量分摊。 + +## 验证证据 + +- `go test ./...`、`go test -race ./internal/openplatform`、`go vet ./internal/openplatform`、文档YAML校验、发布脚本自测与diff-check通过。 +- 单元/SQL测试覆盖真实零、原始精度、负数/null、非法及未来时间、容差边界、延迟上传、分块重组失败、同时间冲突、历史容量缺失、修订指纹、跨午夜、协议限制、查询/采样双重授权、混合错误、超时不丢项、200项去重映射、429并发/时间窗限制。 +- 生产只读覆盖调查确认8月1日已有氢质量原始帧，1024个VIN的既有应用及逐车授权覆盖8月1日至调查时刻；这不代表每台车每个时点有有效氢量。 +- 在生产机器的隔离候选进程127.0.0.1:20311测试后停止候选；未建立永久对外测试域名。 +- 选3台既有授权车辆，每车8月3日和15日各一个历史采样区间，每区间3条原始帧，共18帧。接口逐项比对原始质量、真实采集时间、VIN、requestId、来源及版本；全部NORMAL。 +- 最终候选18点2.989秒；200项含18个不同点2.997秒。正式HTTPS复测分别3.066秒和3.133秒，均全部NORMAL。 +- 正式HTTPS确认偏移1秒且容差0为STALE/null，默认容差命中同一sourceRecordId；部分FORBIDDEN和UNSUPPORTED不影响正常项；401、非法批次400、429及Retry-After通过。 +- 真实缺数样本：SAMPLE_VIN_1查询2026-08-01 00:00:00，返回NO_DATA、NO_HISTORICAL_SAMPLE，质量和采样信息为null，没有补零。 +- 候选与正式均回归100台实时、里程、日用氢量接口。正式耗时184/16/6毫秒；实时包含20条ESTIMATED kg与2个真实零，质量/满充百分比公式验证通过。日用氢量的NO_DATA/DATA_ANOMALY为数据状态，未把它们算作有效氢耗。 +- 正式健康、文档、OpenAPI和目录均200，systemd active；启动后日志未见新增服务异常。所有临时测试凭证及其授权已删除，未修改合作方既有密钥或授权。 + +## 脱敏历史样本 + +下表为真实历史遥测采样区间，**不是已取得海港作业节点证明的真实作业区间**。每区间中间一帧也已核验，完整脱敏返回保存在证据目录。两个端点相同质量是有效记录，不能据此推断全天或整次作业用氢为零。 + +| 车辆代号 | 采集时间（北京时间） | 剩余kg起点 → 终点 | 状态 | +| --- | --- | --- | --- | +| SAMPLE_VIN_1 | 2026-08-03T08:50:09+08:00 → 2026-08-03T08:50:29+08:00 | 2.9 → 2.9 | NORMAL | +| SAMPLE_VIN_1 | 2026-08-15T00:09:57+08:00 → 2026-08-15T00:10:19+08:00 | 5.5 → 5.5 | NORMAL | +| SAMPLE_VIN_2 | 2026-08-03T00:00:00+08:00 → 2026-08-03T00:00:20+08:00 | 3.3 → 3.3 | NORMAL | +| SAMPLE_VIN_2 | 2026-08-15T06:40:02+08:00 → 2026-08-15T06:40:21+08:00 | 1.9 → 1.9 | NORMAL | +| SAMPLE_VIN_3 | 2026-08-03T00:00:06+08:00 → 2026-08-03T00:00:26+08:00 | 5.7 → 5.7 | NORMAL | +| SAMPLE_VIN_3 | 2026-08-15T00:00:04+08:00 → 2026-08-15T00:00:24+08:00 | 4.4 → 4.4 | NORMAL | + +## 未覆盖的业务验收 + +海港真实作业开始/结束节点未提供，因此还未完成“3台车、每台2段真实海港作业”的端到端Seeker联调。本次完成历史接口与原始采样验收及部署。较早于2026-08-01的查询不在本版开放范围；当前容量历史版本证据不足，历史估算不输出数值。这些限制已同步在线契约，未伪造历史配置或作业记录。 + +脱敏响应、探针源码、候选与正式验收输出、发布SHA等位于工作区`outputs/historical-hydrogen-release-20260909/`。不含密钥；私有原始样本仅为服务器临时验收文件，验收后移除。