182 lines
9.7 KiB
Go
182 lines
9.7 KiB
Go
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})
|
|
}
|
|
}
|