426 lines
15 KiB
Go
426 lines
15 KiB
Go
// Package handler provides HTTP handlers for various endpoints.
|
|
package handler
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"modelRT/common"
|
|
"modelRT/common/errcode"
|
|
"modelRT/constants"
|
|
"modelRT/database"
|
|
"modelRT/diagram"
|
|
"modelRT/logger"
|
|
"modelRT/model"
|
|
"modelRT/orm"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type dataObjectAttributeUpdateRequest struct {
|
|
Token string `json:"token"`
|
|
Field string `json:"field"`
|
|
Value json.RawMessage `json:"value"`
|
|
Data json.RawMessage `json:"data,omitempty"`
|
|
}
|
|
|
|
// DataObjectAttributeUpdateHandler updates the writable field of one data object.
|
|
func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
|
ctx := c.Request.Context()
|
|
var request dataObjectAttributeUpdateRequest
|
|
if err := c.ShouldBindJSON(&request); err != nil {
|
|
logger.Error(ctx, "unmarshal data-object update request failed", "error", err)
|
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
|
return
|
|
}
|
|
|
|
dataObjectType, field, value, err := validateDataObjectAttributeUpdate(request)
|
|
if err != nil {
|
|
logger.Warn(ctx, "validate data-object update request failed", "token", request.Token, "field", request.Field, "error", err)
|
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
|
return
|
|
}
|
|
|
|
tx := database.GetPostgresDBClient().WithContext(ctx).Begin()
|
|
if tx.Error != nil {
|
|
logger.Error(ctx, "begin data-object update transaction failed", "error", tx.Error)
|
|
renderRespFailure(c, constants.RespCodeServerError, "begin postgres transaction failed", nil)
|
|
return
|
|
}
|
|
transactionCompleted := false
|
|
defer func() {
|
|
if !transactionCompleted {
|
|
_ = tx.Rollback().Error
|
|
}
|
|
}()
|
|
|
|
message := "data-object attribute update success"
|
|
var measurementResult measurementUpdateResult
|
|
switch dataObjectType {
|
|
case constants.DataObjectTypeParameter:
|
|
parameter, queryErr := database.QueryParameterByDataObjectToken(ctx, tx, request.Token)
|
|
if queryErr == nil {
|
|
queryErr = database.UpdateParameterDataObjectValue(ctx, tx, parameter, value)
|
|
}
|
|
err = queryErr
|
|
case constants.DataObjectTypeMeasurement:
|
|
measurementResult, err = updateMeasurementDataObject(ctx, tx, request.Token, field, value, request.Data, measurementUpdateDependencies{
|
|
writeManualValueFunc: writeMeasurementManualValue,
|
|
updateDataRTFunc: callRealTimeDataWriteStopInterface,
|
|
startDataRTFunc: callRealTimeDataWriteStartInterface,
|
|
replaceRedisValueFunc: replaceMeasurementRedisValue,
|
|
})
|
|
message = measurementResult.message
|
|
default:
|
|
err = fmt.Errorf("unsupported data object type %q", dataObjectType)
|
|
}
|
|
|
|
if err != nil {
|
|
_ = tx.Rollback().Error
|
|
if measurementResult.recordFailure {
|
|
if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, time.Now().UTC()); logErr != nil {
|
|
logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
|
|
}
|
|
}
|
|
logger.Warn(ctx, "update data-object attribute failed", "token", request.Token, "field", field, "error", err)
|
|
if isInvalidDataObjectUpdateError(err) {
|
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
|
return
|
|
}
|
|
renderRespFailure(c, constants.RespCodeFailed, err.Error(), nil)
|
|
return
|
|
}
|
|
|
|
if err := tx.Commit().Error; err != nil {
|
|
logger.Error(ctx, "commit data-object update transaction failed", "token", request.Token, "field", field, "error", err)
|
|
renderRespFailure(c, constants.RespCodeServerError, "transaction commit failed", nil)
|
|
return
|
|
}
|
|
transactionCompleted = true
|
|
|
|
renderRespSuccess(c, constants.RespCodeSuccess, message, map[string]any{
|
|
"token": request.Token,
|
|
"field": field,
|
|
"value": value,
|
|
})
|
|
}
|
|
|
|
func validateDataObjectAttributeUpdate(request dataObjectAttributeUpdateRequest) (constants.DataObjectType, string, any, error) {
|
|
if request.Token == "" {
|
|
return "", "", nil, fmt.Errorf("token is required")
|
|
}
|
|
if len(bytes.TrimSpace(request.Value)) == 0 || bytes.Equal(bytes.TrimSpace(request.Value), []byte("null")) {
|
|
return "", "", nil, fmt.Errorf("value is required")
|
|
}
|
|
|
|
field := strings.ToLower(strings.TrimSpace(request.Field))
|
|
if field == "" {
|
|
field = "value"
|
|
}
|
|
|
|
dataObjectType, err := model.ClassifyDataObjectToken(request.Token)
|
|
if err != nil {
|
|
return "", "", nil, err
|
|
}
|
|
|
|
switch dataObjectType {
|
|
case constants.DataObjectTypeParameter:
|
|
parts := strings.Split(request.Token, ".")
|
|
attributeGroup := parts[len(parts)-2]
|
|
if !isWritableParameterAttributeGroup(attributeGroup) {
|
|
return "", "", nil, fmt.Errorf("parameter updates do not support token6=%s", attributeGroup)
|
|
}
|
|
if field != "value" {
|
|
return "", "", nil, fmt.Errorf("parameter data objects only support updating field value")
|
|
}
|
|
value, err := decodeDataObjectUpdateValue(request.Value)
|
|
return dataObjectType, field, value, err
|
|
case constants.DataObjectTypeMeasurement:
|
|
parts := strings.Split(request.Token, ".")
|
|
if len(parts) != 2 && parts[len(parts)-2] != "bay" {
|
|
return "", "", nil, fmt.Errorf("measurement updates require token4.token7 or token6=bay")
|
|
}
|
|
switch field {
|
|
case "value":
|
|
value, err := parseMeasurementUpdateValue(request.Value)
|
|
return dataObjectType, field, value, err
|
|
case "mode":
|
|
value, err := parseMeasurementUpdateMode(request.Value)
|
|
return dataObjectType, field, value, err
|
|
default:
|
|
return "", "", nil, fmt.Errorf("measurement data objects only support updating fields value and mode")
|
|
}
|
|
default:
|
|
return "", "", nil, fmt.Errorf("unsupported data object type %q", dataObjectType)
|
|
}
|
|
}
|
|
|
|
func isWritableParameterAttributeGroup(group string) bool {
|
|
switch group {
|
|
case "rated", "setup", "model", "stable", "craft", "integrity", "behavior", "base_extend":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func decodeDataObjectUpdateValue(raw json.RawMessage) (any, error) {
|
|
decoder := json.NewDecoder(bytes.NewReader(raw))
|
|
decoder.UseNumber()
|
|
var value any
|
|
if err := decoder.Decode(&value); err != nil {
|
|
return nil, fmt.Errorf("decode update value: %w", err)
|
|
}
|
|
if number, ok := value.(json.Number); ok {
|
|
if integer, err := number.Int64(); err == nil {
|
|
return integer, nil
|
|
}
|
|
decimal, err := number.Float64()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid numeric update value %q: %w", number, err)
|
|
}
|
|
return decimal, nil
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func parseMeasurementUpdateValue(raw json.RawMessage) (float64, error) {
|
|
var number float64
|
|
if err := json.Unmarshal(raw, &number); err == nil {
|
|
return number, nil
|
|
}
|
|
|
|
var text string
|
|
if err := json.Unmarshal(raw, &text); err != nil {
|
|
return 0, fmt.Errorf("measurement value must be a number or numeric string")
|
|
}
|
|
number, err := strconv.ParseFloat(text, 64)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("measurement value %q is not numeric: %w", text, err)
|
|
}
|
|
return number, nil
|
|
}
|
|
|
|
func parseMeasurementUpdateMode(raw json.RawMessage) (int16, error) {
|
|
var mode int16
|
|
if err := json.Unmarshal(raw, &mode); err != nil {
|
|
return 0, fmt.Errorf("measurement mode must be 0 (manual) or 1 (automatic)")
|
|
}
|
|
if mode != constants.MeasurementModeManual && mode != constants.MeasurementModeAutomatic {
|
|
return 0, fmt.Errorf("measurement mode must be 0 (manual) or 1 (automatic)")
|
|
}
|
|
return mode, nil
|
|
}
|
|
|
|
type measurementManualValueWriter func(context.Context, *orm.Measurement, float64) error
|
|
|
|
type measurementDataRTUpdater func(context.Context, orm.JSONMap, *float64) error
|
|
|
|
type measurementRedisValueReplacer func(context.Context, *orm.Measurement, float64) error
|
|
|
|
type measurementUpdateDependencies struct {
|
|
writeManualValueFunc measurementManualValueWriter
|
|
updateDataRTFunc measurementDataRTUpdater
|
|
startDataRTFunc measurementDataRTUpdater
|
|
replaceRedisValueFunc measurementRedisValueReplacer
|
|
}
|
|
|
|
type measurementUpdateResult struct {
|
|
message string
|
|
measurementID int64
|
|
value float64
|
|
recordFailure bool
|
|
}
|
|
|
|
func updateMeasurementDataObject(
|
|
ctx context.Context,
|
|
tx *gorm.DB,
|
|
token, field string,
|
|
value any,
|
|
modeData json.RawMessage,
|
|
dependencies measurementUpdateDependencies,
|
|
) (measurementUpdateResult, error) {
|
|
measurement, _, err := database.QueryMeasurementByDataObjectToken(ctx, tx, token)
|
|
if err != nil {
|
|
return measurementUpdateResult{}, err
|
|
}
|
|
|
|
lockedMeasurement, err := database.QueryMeasurementByIDForUpdate(ctx, tx, measurement.ID)
|
|
if err != nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("lock measurement %d for update: %w", measurement.ID, err)
|
|
}
|
|
|
|
switch field {
|
|
case "mode":
|
|
mode, ok := value.(int16)
|
|
if !ok {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement mode has invalid type %T", value)
|
|
}
|
|
currentMode, err := measurementModeIsAutomatic(lockedMeasurement.Mode)
|
|
if err != nil {
|
|
return measurementUpdateResult{}, err
|
|
}
|
|
targetAutomatic := mode == constants.MeasurementModeAutomatic
|
|
if currentMode == targetAutomatic {
|
|
return measurementUpdateResult{message: fmt.Sprintf("measurement is already in %s mode", measurementModeName(mode))}, nil
|
|
}
|
|
var manualValue *float64
|
|
if currentMode && mode == constants.MeasurementModeManual {
|
|
manualValue, err = parseOptionalMeasurementModeData(modeData)
|
|
if err != nil {
|
|
return measurementUpdateResult{}, err
|
|
}
|
|
}
|
|
if err := database.UpdateMeasurementModeWithOperation(ctx, tx, lockedMeasurement.ID, mode, time.Now().UTC()); err != nil {
|
|
return measurementUpdateResult{}, err
|
|
}
|
|
if currentMode && mode == constants.MeasurementModeManual {
|
|
if dependencies.updateDataRTFunc == nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement dataRT updater is nil")
|
|
}
|
|
if err := dependencies.updateDataRTFunc(ctx, lockedMeasurement.DataSource, nil); err != nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("stop automatic measurement write to dataRT: %w", err)
|
|
}
|
|
if manualValue != nil {
|
|
if dependencies.replaceRedisValueFunc == nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement redis value replacer is nil")
|
|
}
|
|
if err := dependencies.replaceRedisValueFunc(ctx, &lockedMeasurement, *manualValue); err != nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("replace measurement redis value: %w", err)
|
|
}
|
|
}
|
|
}
|
|
if !currentMode && mode == constants.MeasurementModeAutomatic {
|
|
if dependencies.startDataRTFunc == nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement dataRT starter is nil")
|
|
}
|
|
if err := dependencies.startDataRTFunc(ctx, lockedMeasurement.DataSource, nil); err != nil {
|
|
return measurementUpdateResult{}, fmt.Errorf("start automatic measurement write to dataRT: %w", err)
|
|
}
|
|
}
|
|
return measurementUpdateResult{message: fmt.Sprintf("measurement mode changed to %s", measurementModeName(mode))}, nil
|
|
case "value":
|
|
currentMode, err := measurementModeIsAutomatic(lockedMeasurement.Mode)
|
|
if err != nil {
|
|
return measurementUpdateResult{}, err
|
|
}
|
|
if currentMode {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement value is read-only while mode is automatic")
|
|
}
|
|
manualValue, ok := value.(float64)
|
|
if !ok {
|
|
return measurementUpdateResult{}, fmt.Errorf("measurement value has invalid type %T", value)
|
|
}
|
|
failureResult := measurementUpdateResult{
|
|
measurementID: lockedMeasurement.ID,
|
|
value: manualValue,
|
|
recordFailure: true,
|
|
}
|
|
if dependencies.writeManualValueFunc == nil {
|
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement manual value writer is nil"))
|
|
}
|
|
if err := dependencies.writeManualValueFunc(ctx, &lockedMeasurement, manualValue); err != nil {
|
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
|
}
|
|
if dependencies.updateDataRTFunc == nil {
|
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement dataRT updater is nil"))
|
|
}
|
|
if err := dependencies.updateDataRTFunc(ctx, lockedMeasurement.DataSource, &manualValue); err != nil {
|
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
|
}
|
|
if err := database.AppendMeasurementValueOperation(ctx, tx, lockedMeasurement.ID, 0, manualValue, time.Now().UTC()); err != nil {
|
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
|
}
|
|
return measurementUpdateResult{message: "measurement manual value updated"}, nil
|
|
default:
|
|
return measurementUpdateResult{}, fmt.Errorf("unsupported measurement update field %q", field)
|
|
}
|
|
}
|
|
|
|
func parseOptionalMeasurementModeData(raw json.RawMessage) (*float64, error) {
|
|
trimmed := bytes.TrimSpace(raw)
|
|
if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
|
|
return nil, nil
|
|
}
|
|
value, err := parseMeasurementUpdateValue(trimmed)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid measurement mode data: %w", err)
|
|
}
|
|
return &value, nil
|
|
}
|
|
|
|
func measurementModeName(mode int16) string {
|
|
if mode == constants.MeasurementModeAutomatic {
|
|
return "automatic"
|
|
}
|
|
return "manual"
|
|
}
|
|
|
|
func measurementModeIsAutomatic(mode int16) (bool, error) {
|
|
switch mode {
|
|
case constants.MeasurementModeManual:
|
|
return false, nil
|
|
case constants.MeasurementModeAutomatic:
|
|
return true, nil
|
|
default:
|
|
return false, fmt.Errorf("measurement has invalid mode %d", mode)
|
|
}
|
|
}
|
|
|
|
func isInvalidDataObjectUpdateError(err error) bool {
|
|
return errors.Is(err, common.ErrInvalidParameterToken) ||
|
|
errors.Is(err, common.ErrParameterTokenNotFound) ||
|
|
errors.Is(err, common.ErrAmbiguousParameterToken) ||
|
|
errors.Is(err, common.ErrInvalidMeasurementToken) ||
|
|
errors.Is(err, common.ErrMeasurementTokenNotFound) ||
|
|
errors.Is(err, common.ErrAmbiguousMeasurementToken)
|
|
}
|
|
|
|
func writeMeasurementManualValue(ctx context.Context, measurement *orm.Measurement, value float64) error {
|
|
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
|
if err != nil {
|
|
return fmt.Errorf("generate measurement redis key: %w", err)
|
|
}
|
|
zset, err := diagram.NewRedisZSet(ctx, key, 0, false)
|
|
if err != nil {
|
|
return fmt.Errorf("create measurement redis zset: %w", err)
|
|
}
|
|
if err := zset.ZADD(key, value, strconv.FormatInt(time.Now().UnixNano(), 10)); err != nil {
|
|
return fmt.Errorf("write manual measurement value to redis: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func callRealTimeDataWriteStopInterface(_ context.Context, _ orm.JSONMap, _ *float64) error {
|
|
// TODO: call the dataRT HTTP API. A nil value stops automatic writes;
|
|
// a non-nil value writes the supplied manual measurement value.
|
|
return nil
|
|
}
|
|
|
|
func callRealTimeDataWriteStartInterface(_ context.Context, _ orm.JSONMap, _ *float64) error {
|
|
// TODO: call the dataRT HTTP API to start automatic measurement writes.
|
|
return nil
|
|
}
|
|
|
|
func replaceMeasurementRedisValue(ctx context.Context, measurement *orm.Measurement, value float64) error {
|
|
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
|
if err != nil {
|
|
return fmt.Errorf("generate measurement redis key: %w", err)
|
|
}
|
|
zset, err := diagram.NewRedisZSet(ctx, key, 0, false)
|
|
if err != nil {
|
|
return fmt.Errorf("create measurement redis zset: %w", err)
|
|
}
|
|
if err := zset.ZREPLACE(key, value, strconv.FormatInt(time.Now().UnixNano(), 10)); err != nil {
|
|
return fmt.Errorf("replace manual measurement value in redis: %w", err)
|
|
}
|
|
return nil
|
|
}
|