package httpapi import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "strings" "unicode/utf8" "data-terminal-cloud/internal/device" "data-terminal-cloud/internal/deviceauth" "data-terminal-cloud/internal/idempotency" ) type DeviceTelemetrySubmitter interface { Submit(context.Context, deviceauth.Identity, string, []byte, device.TelemetryBatchInput) (device.TelemetryBatchResponse, bool, error) } func SubmitDeviceTelemetryHandler(service DeviceTelemetrySubmitter) http.Handler { return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { identity, ok := deviceauth.FromContext(request.Context()) if !ok { WriteError(writer, request, http.StatusUnauthorized, "DEVICE_AUTHENTICATION_FAILED", "设备身份验证失败") return } idempotencyKey := strings.TrimSpace(request.Header.Get("Idempotency-Key")) if len(idempotencyKey) < 16 || len(idempotencyKey) > 128 { WriteAPIError(writer, request, http.StatusBadRequest, Error{ Code: "VALIDATION_FAILED", Message: "缺少有效的重复请求识别码", FieldErrors: []FieldError{{Field: "Idempotency-Key", Message: "长度必须为16至128字符"}}, }) return } rawBody, err := io.ReadAll(request.Body) if err != nil { WriteError(writer, request, http.StatusBadRequest, "INVALID_REQUEST_BODY", "请求正文无法读取") return } decoder := json.NewDecoder(bytes.NewReader(rawBody)) decoder.DisallowUnknownFields() var input device.TelemetryBatchInput if err := decoder.Decode(&input); err != nil || ensureJSONEOF(decoder) != nil { WriteAPIError(writer, request, http.StatusBadRequest, Error{ Code: "VALIDATION_FAILED", Message: "设备遥测参数不正确", FieldErrors: []FieldError{{Field: "body", Message: "必须是符合契约的JSON对象"}}, }) return } if fieldErrors := validateTelemetryBatch(input); len(fieldErrors) > 0 { WriteAPIError(writer, request, http.StatusBadRequest, Error{Code: "VALIDATION_FAILED", Message: "设备遥测参数不正确", FieldErrors: fieldErrors}) return } response, replayed, err := service.Submit(request.Context(), identity, idempotencyKey, rawBody, input) if err != nil { writeTelemetryError(writer, request, err) return } if replayed { writer.Header().Set("Idempotency-Replayed", "true") } WriteJSON(writer, request, http.StatusOK, response) }) } func validateTelemetryBatch(input device.TelemetryBatchInput) []FieldError { fieldErrors := make([]FieldError, 0) if len(input.Events) < 1 || len(input.Events) > 500 { return append(fieldErrors, FieldError{Field: "events", Message: "必须包含1至500条遥测事件"}) } eventKeys := make(map[string]struct{}, len(input.Events)) sequences := make(map[int64]struct{}, len(input.Events)) for index, event := range input.Events { prefix := fmt.Sprintf("events[%d]", index) eventKey := strings.TrimSpace(event.EventKey) if eventKey == "" || utf8.RuneCountInString(eventKey) > 255 || !isSafeErrorCode(eventKey) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".event_key", Message: "必须是最长255字符的安全业务键"}) } else if _, exists := eventKeys[eventKey]; exists { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".event_key", Message: "请求内不能重复"}) } else { eventKeys[eventKey] = struct{}{} } if event.SequenceNo < 1 { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".sequence_no", Message: "必须大于0"}) } else if _, exists := sequences[event.SequenceNo]; exists { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".sequence_no", Message: "请求内不能重复"}) } else { sequences[event.SequenceNo] = struct{}{} } if event.OccurredAt.IsZero() { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".occurred_at", Message: "必须是RFC3339时间"}) } if event.ConversationID != "" && !validUUID(event.ConversationID) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".conversation_id", Message: "必须是UUID"}) } switch event.EventType { case "status": if event.Status == nil || event.Error != nil || event.CommandResult != nil { fieldErrors = append(fieldErrors, FieldError{Field: prefix, Message: "status事件必须且只能包含status对象"}) continue } if strings.TrimSpace(event.Status.FirmwareVersion) == "" || utf8.RuneCountInString(event.Status.FirmwareVersion) > 80 { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.firmware_version", Message: "不能为空且最长80字符"}) } if event.Status.ReportedConfigRevision < 1 { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.reported_config_revision", Message: "必须大于0"}) } if strings.TrimSpace(event.Status.Network.Type) == "" || utf8.RuneCountInString(event.Status.Network.Type) > 40 { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.network.type", Message: "不能为空且最长40字符"}) } if event.Status.Network.RSSI != nil && (*event.Status.Network.RSSI < -127 || *event.Status.Network.RSSI > 0) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.network.rssi", Message: "必须为-127至0"}) } if event.Status.RecordingQueue.PendingSegments < 0 || event.Status.RecordingQueue.FailedSegments < 0 { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.recording_queue", Message: "分片数量不能为负数"}) } if utf8.RuneCountInString(event.Status.RecordingQueue.LastFailureCode) > 80 || !isSafeErrorCode(event.Status.RecordingQueue.LastFailureCode) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.recording_queue.last_failure_code", Message: "只能是最长80字符的错误码"}) } if event.Status.RecordingQueue.LastFailureAt != nil && event.Status.RecordingQueue.LastFailureAt.IsZero() { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.recording_queue.last_failure_at", Message: "必须是RFC3339时间"}) } if event.Status.ConversationStatus != "" && !oneOf(event.Status.ConversationStatus, "PREPARING", "ROOM_ISSUED", "STARTING_AGENT", "ACTIVE", "STOPPING", "COMPLETED", "FAILED", "EXPIRED") { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".status.conversation_status", Message: "不是有效的通话状态"}) } case "error": if event.Error == nil || event.Status != nil || event.CommandResult != nil { fieldErrors = append(fieldErrors, FieldError{Field: prefix, Message: "error事件必须且只能包含error对象"}) continue } if event.Error.Code == "" || utf8.RuneCountInString(event.Error.Code) > 80 || !isSafeErrorCode(event.Error.Code) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".error.code", Message: "必须是最长80字符的错误码"}) } if !oneOf(event.Error.Severity, "warning", "error", "critical") { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".error.severity", Message: "必须为warning、error或critical"}) } case "command_result": if event.CommandResult == nil || event.Status != nil || event.Error != nil { fieldErrors = append(fieldErrors, FieldError{Field: prefix, Message: "command_result事件必须且只能包含command_result对象"}) continue } if !validUUID(event.CommandResult.CommandID) { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".command_result.command_id", Message: "必须是UUID"}) } if !oneOf(event.CommandResult.Status, "ACKNOWLEDGED", "SUCCEEDED", "FAILED") { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".command_result.status", Message: "必须为ACKNOWLEDGED、SUCCEEDED或FAILED"}) } if event.CommandResult.ResultSummary == nil { fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".command_result.result_summary", Message: "必须是JSON对象"}) } else if path, found := forbiddenDeviceConfigField(event.CommandResult.ResultSummary, prefix+".command_result.result_summary"); found { fieldErrors = append(fieldErrors, FieldError{Field: path, Message: "命令结果不得包含密钥、令牌或凭据字段"}) } default: fieldErrors = append(fieldErrors, FieldError{Field: prefix + ".event_type", Message: "必须为status、error或command_result"}) } } return fieldErrors } func writeTelemetryError(writer http.ResponseWriter, request *http.Request, err error) { var conflict *device.TelemetryConflictError switch { case errors.As(err, &conflict): WriteAPIError(writer, request, http.StatusConflict, Error{Code: "TELEMETRY_EVENT_CONFLICT", Message: "遥测业务键与已保存事件冲突", Details: map[string]any{"conflicts": conflict.Conflicts}}) case errors.Is(err, device.ErrTelemetryConfigAhead): WriteError(writer, request, http.StatusConflict, "CONFIG_VERSION_CONFLICT", "设备报告的配置版本高于云端期望版本") case errors.Is(err, device.ErrTelemetryConversationMissing): WriteError(writer, request, http.StatusNotFound, "TELEMETRY_CONVERSATION_NOT_FOUND", "关联通话不存在或不属于当前设备") case errors.Is(err, device.ErrTelemetryCommandConflict): WriteError(writer, request, http.StatusConflict, "DEVICE_COMMAND_STATE_CONFLICT", "命令不存在、已过期或状态不允许回执") case errors.Is(err, idempotency.ErrKeyConflict): WriteError(writer, request, http.StatusConflict, "IDEMPOTENCY_KEY_REUSED", "重复请求识别码已用于不同请求") case errors.Is(err, idempotency.ErrInProgress): WriteAPIError(writer, request, http.StatusConflict, Error{Code: "REQUEST_IN_PROGRESS", Message: "相同请求正在处理中", Retryable: true}) default: WriteAPIError(writer, request, http.StatusServiceUnavailable, Error{Code: "DEVICE_TELEMETRY_UNAVAILABLE", Message: "暂时无法处理设备遥测", Retryable: true}) } }