From b85c2e129d15866443f41ea6fa225475ea304f89 Mon Sep 17 00:00:00 2001 From: douxu Date: Mon, 20 Jul 2026 15:46:31 +0800 Subject: [PATCH] feat(data-object): support parameter and measurement attribute updates - replace batch attribute updates with token, field, and value requests - support parameter value updates for writable attribute groups - support measurement mode switching and manual value updates - record bounded measurement operation logs in PostgreSQL - add jsonb array serialization for measurement operations - add transaction failure context and error-chain support - add unit tests for validation and update workflows --- common/errcode/bussiness_error.go | 3 + common/errcode/error.go | 4 +- database/query_component_measurement.go | 80 ++++ database/query_component_measurement_test.go | 30 ++ database/query_component_parameter.go | 27 ++ database/query_component_parameter_test.go | 37 ++ go.mod | 2 +- handler/data_object_attribute_update.go | 408 +++++++++++++------ handler/data_object_attribute_update_test.go | 265 ++++++++++++ orm/circuit_diagram_measurement.go | 27 +- orm/jsonb_serializer.go | 48 +++ orm/jsonb_serializer_test.go | 21 + 12 files changed, 812 insertions(+), 140 deletions(-) create mode 100644 handler/data_object_attribute_update_test.go create mode 100644 orm/jsonb_serializer_test.go diff --git a/common/errcode/bussiness_error.go b/common/errcode/bussiness_error.go index 6ab5999..f84cbe1 100644 --- a/common/errcode/bussiness_error.go +++ b/common/errcode/bussiness_error.go @@ -38,6 +38,9 @@ var ( // ErrCommitTxFailed indicates that the PostgreSQL transaction could not be committed successfully. ErrCommitTxFailed = newError(50005, "postgres database transaction commit failed") + // ErrMeasurementValueUpdateFailed indicates that a manual measurement value transaction failed. + ErrMeasurementValueUpdateFailed = newError(50006, "measurement manual value update failed") + // ErrCachedQueryFailed define variable to indicates an error occurred while attempting to fetch data from the Redis cache. ErrCachedQueryFailed = newError(60001, "query redis cached data failed") diff --git a/common/errcode/error.go b/common/errcode/error.go index 1d12276..ec0941e 100644 --- a/common/errcode/error.go +++ b/common/errcode/error.go @@ -66,8 +66,8 @@ func Wrap(msg string, err error) *AppError { return appErr } -// UnWrap define func return the error wrapped in structure -func (e *AppError) UnWrap() error { +// Unwrap returns the underlying cause for errors.Is and errors.As traversal. +func (e *AppError) Unwrap() error { return e.cause } diff --git a/database/query_component_measurement.go b/database/query_component_measurement.go index 66f6811..06562c4 100644 --- a/database/query_component_measurement.go +++ b/database/query_component_measurement.go @@ -3,6 +3,7 @@ package database import ( "context" + "encoding/json" "fmt" "strings" "time" @@ -16,6 +17,8 @@ import ( "gorm.io/gorm/clause" ) +const measurementOperationsLimit = 500 + // QueryMeasurementByID return the result of query circuit diagram component measurement info by id from postgresDB func QueryMeasurementByID(ctx context.Context, tx *gorm.DB, id int64) (orm.Measurement, error) { var measurement orm.Measurement @@ -42,6 +45,83 @@ func QueryMeasurementByToken(ctx context.Context, tx *gorm.DB, token string) (or return *measurement, nil } +// UpdateMeasurementMode stores the data-object mode representation in the +// measurement row: false is manual mode (0), true is automatic mode (1). +func UpdateMeasurementMode(ctx context.Context, db *gorm.DB, measurementID int64, automatic bool) error { + mode := int16(0) + if automatic { + mode = 1 + } + + result := db.WithContext(ctx). + Model(&orm.Measurement{}). + Where("id = ?", measurementID). + Update("mode", mode) + if result.Error != nil { + return fmt.Errorf("update measurement %d mode: %w", measurementID, result.Error) + } + if result.RowsAffected == 0 { + return fmt.Errorf("update measurement %d mode affected no rows", measurementID) + } + return nil +} + +// UpdateMeasurementModeWithOperation changes mode and appends its audit entry +// atomically. The operations array retains only its newest 500 entries. +func UpdateMeasurementModeWithOperation(ctx context.Context, db *gorm.DB, measurementID int64, automatic bool, timestamp time.Time) error { + mode := int16(0) + if automatic { + mode = 1 + } + operation := orm.JSONMap{ + "command": mode, + "timestamp": timestamp, + } + return updateMeasurementWithOperation(ctx, db, measurementID, map[string]any{"mode": mode}, operation) +} + +// AppendMeasurementValueOperation appends the audit result of a manual-value +// transaction without changing other measurement columns. +func AppendMeasurementValueOperation(ctx context.Context, db *gorm.DB, measurementID int64, transaction int, value float64, timestamp time.Time) error { + operation := orm.JSONMap{ + "transaction": transaction, + "value": value, + "timestamp": timestamp, + } + return updateMeasurementWithOperation(ctx, db, measurementID, nil, operation) +} + +func updateMeasurementWithOperation(ctx context.Context, db *gorm.DB, measurementID int64, updates map[string]any, operation orm.JSONMap) error { + encodedOperation, err := json.Marshal(operation) + if err != nil { + return fmt.Errorf("encode measurement %d operation: %w", measurementID, err) + } + + operationExpression := gorm.Expr(`( + CASE + WHEN cardinality(operations) >= ? + THEN operations[(cardinality(operations) - ? + 2):cardinality(operations)] + ELSE operations + END + ) || ARRAY[?::jsonb]`, measurementOperationsLimit, measurementOperationsLimit, string(encodedOperation)) + if updates == nil { + updates = make(map[string]any, 1) + } + updates["operations"] = operationExpression + + result := db.WithContext(ctx). + Model(&orm.Measurement{}). + Where("id = ?", measurementID). + Updates(updates) + if result.Error != nil { + return fmt.Errorf("update measurement %d operation: %w", measurementID, result.Error) + } + if result.RowsAffected == 0 { + return fmt.Errorf("update measurement %d operation affected no rows", measurementID) + } + return nil +} + // ValidateMeasurementToken checks whether token uniquely identifies an existing // measurement through the measurement, component, bay, station, zone, and grid // relationships. Supported formats are token1-token7, token4-token7, and diff --git a/database/query_component_measurement_test.go b/database/query_component_measurement_test.go index eee8f82..4c603e1 100644 --- a/database/query_component_measurement_test.go +++ b/database/query_component_measurement_test.go @@ -82,6 +82,36 @@ func TestBuildMeasurementTokenValidationQuery(t *testing.T) { } } +func TestUpdateMeasurementMode(t *testing.T) { + tests := []struct { + name string + automatic bool + wantMode int16 + }{ + {name: "manual", automatic: false, wantMode: 0}, + {name: "automatic", automatic: true, wantMode: 1}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = sqlDB.Close() }) + + db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{SkipDefaultTransaction: true}) + require.NoError(t, err) + + mock.ExpectExec(regexp.QuoteMeta(`UPDATE "measurement" SET "mode"=$1 WHERE id = $2`)). + WithArgs(tt.wantMode, int64(10)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + err = UpdateMeasurementMode(context.Background(), db, 10, tt.automatic) + require.NoError(t, err) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + func TestValidateMeasurementToken(t *testing.T) { tests := []struct { name string diff --git a/database/query_component_parameter.go b/database/query_component_parameter.go index 865185e..ceff620 100644 --- a/database/query_component_parameter.go +++ b/database/query_component_parameter.go @@ -145,6 +145,33 @@ func QueryParameterDataObjectValue(ctx context.Context, db *gorm.DB, parameter * return value, nil } +// UpdateParameterDataObjectValue writes token7 to the dynamic parameter row +// resolved from a data-object token. Component-table attributes are not +// supported by the data-object update API. +func UpdateParameterDataObjectValue(ctx context.Context, db *gorm.DB, parameter *ParameterDataObject, value any) error { + if parameter == nil { + return fmt.Errorf("parameter data object is nil") + } + if parameter.AttributeGroup == "component" { + return fmt.Errorf("component data-object updates are not supported") + } + if !parameterTableNamePattern.MatchString(parameter.TableName) { + return fmt.Errorf("invalid parameter table name %q", parameter.TableName) + } + + result := db.WithContext(ctx). + Table(parameter.TableName). + Where("global_uuid = ? AND attribute_group = ?", parameter.Component.GlobalUUID, parameter.AttributeGroup). + Update(parameter.AttributeName, value) + if result.Error != nil { + return fmt.Errorf("update parameter %q in table %q: %w", parameter.AttributeName, parameter.TableName, result.Error) + } + if result.RowsAffected == 0 { + return fmt.Errorf("update parameter %q in table %q affected no rows", parameter.AttributeName, parameter.TableName) + } + return nil +} + // QueryParameterAttributeDescription returns the display name registered for // token7 in basic.attribute. func QueryParameterAttributeDescription(ctx context.Context, db *gorm.DB, attributeName string) (string, error) { diff --git a/database/query_component_parameter_test.go b/database/query_component_parameter_test.go index 4e8cc43..c3a6418 100644 --- a/database/query_component_parameter_test.go +++ b/database/query_component_parameter_test.go @@ -153,6 +153,43 @@ func TestQueryParameterDataObjectValue(t *testing.T) { require.NoError(t, mock.ExpectationsWereMet()) } +func TestUpdateParameterDataObjectValue(t *testing.T) { + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = sqlDB.Close() }) + + db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{SkipDefaultTransaction: true}) + require.NoError(t, err) + + const componentUUID = "70c190f2-8a60-42a9-b143-ec5f87e0aa6b" + parsedUUID, err := uuid.FromString(componentUUID) + require.NoError(t, err) + parameter := &ParameterDataObject{ + Component: orm.Component{GlobalUUID: parsedUUID}, + TableName: "bus_bus_1_rated", + AttributeGroup: "rated", + AttributeName: "unom_kv", + } + + mock.ExpectExec(regexp.QuoteMeta(`UPDATE "bus_bus_1_rated" SET "unom_kv"=$1 WHERE global_uuid = $2 AND attribute_group = $3`)). + WithArgs("15.2", componentUUID, "rated"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + err = UpdateParameterDataObjectValue(context.Background(), db, parameter, "15.2") + require.NoError(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestUpdateParameterDataObjectValueRejectsComponent(t *testing.T) { + err := UpdateParameterDataObjectValue(context.Background(), &gorm.DB{}, &ParameterDataObject{ + TableName: "component", + AttributeGroup: "component", + AttributeName: "global_uuid", + }, "uuid") + require.Error(t, err) + assert.Contains(t, err.Error(), "not supported") +} + func TestQueryComponentParameterByDataObjectToken(t *testing.T) { sqlDB, mock, err := sqlmock.New() require.NoError(t, err) diff --git a/go.mod b/go.mod index 441ab18..d32ddda 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/gofrs/uuid v4.4.0+incompatible github.com/gomodule/redigo v1.8.9 github.com/gorilla/websocket v1.5.3 + github.com/jackc/pgx/v5 v5.5.5 github.com/json-iterator/go v1.1.12 github.com/natefinch/lumberjack v2.0.0+incompatible github.com/panjf2000/ants/v2 v2.10.0 @@ -62,7 +63,6 @@ require ( github.com/hashicorp/hcl v1.0.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect - github.com/jackc/pgx/v5 v5.5.5 // indirect github.com/jackc/puddle/v2 v2.2.1 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect diff --git a/handler/data_object_attribute_update.go b/handler/data_object_attribute_update.go index 6ef3099..9be0719 100644 --- a/handler/data_object_attribute_update.go +++ b/handler/data_object_attribute_update.go @@ -2,165 +2,325 @@ 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/network" + "modelRT/model" "modelRT/orm" "github.com/gin-gonic/gin" + "gorm.io/gorm" ) -// DataObjectAttributeUpdateHandler define data object attribute value update process API +type dataObjectAttributeUpdateRequest struct { + Token string `json:"token"` + Field string `json:"field"` + Value json.RawMessage `json:"value"` +} + +// DataObjectAttributeUpdateHandler updates the writable field of one data object. func DataObjectAttributeUpdateHandler(c *gin.Context) { - pgClient := database.GetPostgresDBClient() - var request network.ComponentAttributeUpdateInfo + ctx := c.Request.Context() + var request dataObjectAttributeUpdateRequest if err := c.ShouldBindJSON(&request); err != nil { - logger.Error(c, "unmarshal request params failed", "error", err) + logger.Error(ctx, "unmarshal data-object update request failed", "error", err) renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil) return } - updateResults := make(map[string]*errcode.AppError) - attriModifyConfs := make([]attributeModifyConfig, 0, len(request.AttributeConfigs)) - var attributeComponentTag string - for index, attribute := range request.AttributeConfigs { - slices := strings.Split(attribute.AttributeToken, ".") - if len(slices) < 7 { - updateResults[attribute.AttributeToken] = errcode.ErrInvalidToken - continue - } - - componentTag := slices[4] - if index == 0 { - attributeComponentTag = componentTag - } else if componentTag != attributeComponentTag { - updateResults[attribute.AttributeToken] = errcode.ErrCrossToken - continue - } - - attriModifyConfs = append(attriModifyConfs, attributeModifyConfig{ - attributeToken: attribute.AttributeToken, - attributeExtendType: slices[5], - attributeName: slices[6], - attributeOldVal: attribute.AttributeOldVal, - attributeNewVal: attribute.AttributeNewVal, - }) + 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 } - // open transaction - tx := pgClient.WithContext(c).Begin() + tx := database.GetPostgresDBClient().WithContext(ctx).Begin() if tx.Error != nil { - logger.Error(c, "begin postgres transaction failed", "error", tx.Error) + logger.Error(ctx, "begin data-object update transaction failed", "error", tx.Error) renderRespFailure(c, constants.RespCodeServerError, "begin postgres transaction failed", nil) return } - compInfo, err := database.QueryComponentByCompTag(c, tx, attributeComponentTag) - if err != nil { - logger.Error(c, "query component info by component tag failed", "error", err, "tag", attributeComponentTag) + 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, writeMeasurementManualValue, writeMeasurementManualValueToDataRT) + message = measurementResult.message + default: + err = fmt.Errorf("unsupported data object type %q", dataObjectType) + } - for _, attribute := range request.AttributeConfigs { - if _, exists := updateResults[attribute.AttributeToken]; !exists { - updateResults[attribute.AttributeToken] = errcode.ErrDBQueryFailed.WithCause(err) + 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) } } - - tx.Rollback() - - payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) - renderRespFailure(c, constants.RespCodeFailed, "query component metadata failed", payload) + 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 } - identifiers := make([]orm.ProjectIdentifier, len(attriModifyConfs)) - for i, mod := range attriModifyConfs { - identifiers[i] = orm.ProjectIdentifier{ - Token: mod.attributeToken, - Tag: compInfo.ModelName, - GroupName: mod.attributeExtendType, - } - } - tableNameMap, err := database.BatchGetProjectNames(tx, identifiers) - if err != nil { - tx.Rollback() - - for _, id := range identifiers { - if _, exists := updateResults[id.Token]; !exists { - updateResults[id.Token] = errcode.ErrRetrieveFailed.WithCause(err) - } - } - - payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) - renderRespFailure(c, constants.RespCodeFailed, "batch retrieve table names failed", payload) - return - } - - redisUpdateMap := make(map[string][]cacheUpdateItem) - for _, mod := range attriModifyConfs { - id := orm.ProjectIdentifier{Tag: compInfo.ModelName, GroupName: mod.attributeExtendType} - tableName, exists := tableNameMap[id] - if !exists { - updateResults[mod.attributeToken] = errcode.ErrFoundTargetFailed - continue - } - - result := tx.Table(tableName). - Where(fmt.Sprintf("%s = ? AND global_uuid = ?", mod.attributeName), mod.attributeOldVal, compInfo.GlobalUUID). - Updates(map[string]any{mod.attributeName: mod.attributeNewVal}) - - if result.Error != nil { - updateResults[mod.attributeToken] = errcode.ErrDBUpdateFailed - continue - } - if result.RowsAffected == 0 { - updateResults[mod.attributeToken] = errcode.ErrDBzeroAffectedRows - continue - } - - cacheKey := fmt.Sprintf("%s_%s", attributeComponentTag, mod.attributeExtendType) - redisUpdateMap[cacheKey] = append(redisUpdateMap[cacheKey], - cacheUpdateItem{ - token: mod.attributeToken, - name: mod.attributeName, - newVal: mod.attributeNewVal, - }) - } - - // commit transaction 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 } - for key, items := range redisUpdateMap { - hset := diagram.NewRedisHash(c, key, 5000, false) - - fields := make(map[string]any, len(items)) - for _, item := range items { - fields[item.name] = item.newVal - } - - if err := hset.SetRedisHashByMap(fields); err != nil { - logger.Error(c, "batch sync redis failed", "hash_key", key, "error", err) - - for _, item := range items { - if _, exists := updateResults[item.token]; exists { - updateResults[item.token] = errcode.ErrCacheSyncWarn.WithCause(err) - } - } - } - } - - payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) - if len(updateResults) > 0 { - renderRespFailure(c, constants.RespCodeFailed, "process completed with partial failures", payload) - return - } - renderRespSuccess(c, constants.RespCodeSuccess, "process completed successfully", payload) + 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 == "" { + return "", "", nil, fmt.Errorf("field is required") + } + + 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 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": + 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) (bool, error) { + var mode bool + if err := json.Unmarshal(raw, &mode); err == nil { + return mode, nil + } + + var text string + if err := json.Unmarshal(raw, &text); err != nil { + return false, fmt.Errorf("measurement mode must be true or false") + } + switch strings.ToLower(text) { + case "true": + return true, nil + case "false": + return false, nil + default: + return false, fmt.Errorf("measurement mode must be true or false") + } +} + +type measurementManualValueWriter func(context.Context, *orm.Measurement, float64) error + +type measurementDataRTWriter func(context.Context, orm.JSONMap, float64) error + +type measurementUpdateResult struct { + message string + measurementID int64 + value float64 + recordFailure bool +} + +func updateMeasurementDataObject( + ctx context.Context, + tx *gorm.DB, + token, field string, + value any, + writeManualValue measurementManualValueWriter, + writeDataRT measurementDataRTWriter, +) (measurementUpdateResult, error) { + measurement, _, err := database.QueryMeasurementByDataObjectToken(ctx, tx, token) + if err != nil { + return measurementUpdateResult{}, err + } + + lockedMeasurement, err := database.QueryMeasurementByID(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.(bool) + if !ok { + return measurementUpdateResult{}, fmt.Errorf("measurement mode has invalid type %T", value) + } + currentMode := lockedMeasurement.Mode != 0 + if currentMode == mode { + return measurementUpdateResult{message: fmt.Sprintf("measurement is already in %s mode", measurementModeName(mode))}, nil + } + if err := database.UpdateMeasurementModeWithOperation(ctx, tx, lockedMeasurement.ID, mode, time.Now().UTC()); err != nil { + return measurementUpdateResult{}, err + } + return measurementUpdateResult{message: fmt.Sprintf("measurement mode changed to %s", measurementModeName(mode))}, nil + case "value": + if lockedMeasurement.Mode != 0 { + 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 writeManualValue == nil { + return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement manual value writer is nil")) + } + if err := writeManualValue(ctx, &lockedMeasurement, manualValue); err != nil { + return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err) + } + if writeDataRT == nil { + return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement dataRT writer is nil")) + } + if err := writeDataRT(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 measurementModeName(automatic bool) string { + if automatic { + return "automatic" + } + return "manual" +} + +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 := diagram.NewRedisZSet(ctx, key, 0, false) + 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 writeMeasurementManualValueToDataRT(_ context.Context, _ orm.JSONMap, _ float64) error { + // TODO: call the dataRT HTTP API with the measurement data_source and manual value. + return nil } diff --git a/handler/data_object_attribute_update_test.go b/handler/data_object_attribute_update_test.go new file mode 100644 index 0000000..8e56135 --- /dev/null +++ b/handler/data_object_attribute_update_test.go @@ -0,0 +1,265 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "testing" + + "modelRT/common/errcode" + "modelRT/constants" + "modelRT/orm" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/postgres" + "gorm.io/gorm" +) + +func TestValidateDataObjectAttributeUpdateParameterGroups(t *testing.T) { + groups := []string{ + "rated", + "setup", + "model", + "stable", + "craft", + "integrity", + "behavior", + } + + for _, group := range groups { + t.Run(group, func(t *testing.T) { + request := dataObjectAttributeUpdateRequest{ + Token: fmt.Sprintf("nspath.component.%s.attribute", group), + Field: "VALUE", + Value: json.RawMessage(`"15.2"`), + } + + dataObjectType, field, value, err := validateDataObjectAttributeUpdate(request) + require.NoError(t, err) + assert.Equal(t, constants.DataObjectTypeParameter, dataObjectType) + assert.Equal(t, "value", field) + assert.Equal(t, "15.2", value) + }) + } +} + +func TestValidateDataObjectAttributeUpdateRejectsUnsupportedParameterGroups(t *testing.T) { + for _, group := range []string{"component", "base_extend"} { + t.Run(group, func(t *testing.T) { + _, _, _, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{ + Token: fmt.Sprintf("nspath.component.%s.attribute", group), + Field: "value", + Value: json.RawMessage(`"uuid"`), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "do not support token6="+group) + }) + } +} + +func TestValidateDataObjectAttributeUpdateMeasurementFields(t *testing.T) { + tests := []struct { + name string + field string + value string + expected any + wantError bool + }{ + {name: "numeric value", field: "value", value: `15.2`, expected: float64(15.2)}, + {name: "numeric string value", field: "value", value: `"15.2"`, expected: float64(15.2)}, + {name: "boolean automatic mode", field: "mode", value: `true`, expected: true}, + {name: "string manual mode", field: "mode", value: `"false"`, expected: false}, + {name: "invalid mode", field: "mode", value: `"manual"`, wantError: true}, + {name: "unsupported field", field: "name", value: `"measurement"`, wantError: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataObjectType, field, value, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{ + Token: "nspath.component.bay.measurement", + Field: tt.field, + Value: json.RawMessage(tt.value), + }) + if tt.wantError { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, constants.DataObjectTypeMeasurement, dataObjectType) + assert.Equal(t, tt.field, field) + assert.Equal(t, tt.expected, value) + }) + } +} + +func TestValidateDataObjectAttributeUpdateRejectsShortMeasurementToken(t *testing.T) { + _, _, _, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{ + Token: "nspath.measurement", + Field: "value", + Value: json.RawMessage(`1`), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "token6=bay") +} + +func TestValidateDataObjectAttributeUpdateRequiredFields(t *testing.T) { + tests := []struct { + name string + request dataObjectAttributeUpdateRequest + }{ + {name: "missing token", request: dataObjectAttributeUpdateRequest{Field: "value", Value: json.RawMessage(`1`)}}, + {name: "missing field", request: dataObjectAttributeUpdateRequest{Token: "nspath.measurement", Value: json.RawMessage(`1`)}}, + {name: "missing value", request: dataObjectAttributeUpdateRequest{Token: "nspath.measurement", Field: "value"}}, + {name: "null value", request: dataObjectAttributeUpdateRequest{Token: "nspath.measurement", Field: "value", Value: json.RawMessage(`null`)}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, _, err := validateDataObjectAttributeUpdate(tt.request) + require.Error(t, err) + }) + } +} + +func TestUpdateMeasurementDataObjectLocksRowAndUpdatesMode(t *testing.T) { + db, mock, closeDB := newDataObjectUpdateTestDB(t) + defer closeDB() + + mock.ExpectBegin() + tx := db.Begin() + require.NoError(t, tx.Error) + expectMeasurementResolution(mock, 0) + mock.ExpectExec(`UPDATE "measurement" SET .*"mode"=\$1.*"operations"=.*WHERE id = \$5`). + WithArgs(int16(1), 500, 500, sqlmock.AnyArg(), int64(10)). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectRollback() + + result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", true, nil, nil) + require.NoError(t, err) + assert.Contains(t, result.message, "automatic") + require.NoError(t, tx.Rollback().Error) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestUpdateMeasurementDataObjectReturnsMessageWhenModeIsUnchanged(t *testing.T) { + db, mock, closeDB := newDataObjectUpdateTestDB(t) + defer closeDB() + + mock.ExpectBegin() + tx := db.Begin() + require.NoError(t, tx.Error) + expectMeasurementResolution(mock, 1) + mock.ExpectRollback() + + result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", true, nil, nil) + require.NoError(t, err) + assert.Contains(t, result.message, "already") + require.NoError(t, tx.Rollback().Error) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestUpdateMeasurementDataObjectRejectsValueInAutomaticMode(t *testing.T) { + db, mock, closeDB := newDataObjectUpdateTestDB(t) + defer closeDB() + + mock.ExpectBegin() + tx := db.Begin() + require.NoError(t, tx.Error) + expectMeasurementResolution(mock, 1) + mock.ExpectRollback() + + _, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "read-only") + require.NoError(t, tx.Rollback().Error) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestUpdateMeasurementDataObjectWritesValueInManualMode(t *testing.T) { + db, mock, closeDB := newDataObjectUpdateTestDB(t) + defer closeDB() + + mock.ExpectBegin() + tx := db.Begin() + require.NoError(t, tx.Error) + expectMeasurementResolution(mock, 0) + mock.ExpectExec(`UPDATE "measurement" SET "operations"=.*WHERE id = \$4`). + WithArgs(500, 500, sqlmock.AnyArg(), int64(10)). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectRollback() + + called := false + writer := func(_ context.Context, measurement *orm.Measurement, value float64) error { + called = true + assert.Equal(t, int64(10), measurement.ID) + assert.Equal(t, float64(15.2), value) + return nil + } + dataRTCalled := false + dataRTWriter := func(_ context.Context, dataSource orm.JSONMap, value float64) error { + dataRTCalled = true + assert.Equal(t, float64(15.2), value) + assert.Equal(t, float64(1), dataSource["type"]) + return nil + } + result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), writer, dataRTWriter) + require.NoError(t, err) + assert.True(t, called) + assert.True(t, dataRTCalled) + assert.Contains(t, result.message, "updated") + assert.False(t, result.recordFailure) + require.NoError(t, tx.Rollback().Error) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestUpdateMeasurementDataObjectReturnsFailureResultAndAppError(t *testing.T) { + db, mock, closeDB := newDataObjectUpdateTestDB(t) + defer closeDB() + + mock.ExpectBegin() + tx := db.Begin() + require.NoError(t, tx.Error) + expectMeasurementResolution(mock, 0) + mock.ExpectRollback() + + writeErr := fmt.Errorf("write value failed") + writer := func(context.Context, *orm.Measurement, float64) error { return writeErr } + result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), writer, nil) + require.Error(t, err) + assert.ErrorIs(t, err, errcode.ErrMeasurementValueUpdateFailed) + assert.ErrorIs(t, err, writeErr) + assert.True(t, result.recordFailure) + assert.Equal(t, int64(10), result.measurementID) + assert.Equal(t, float64(15.2), result.value) + require.NoError(t, tx.Rollback().Error) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func newDataObjectUpdateTestDB(t *testing.T) (*gorm.DB, sqlmock.Sqlmock, func()) { + t.Helper() + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{SkipDefaultTransaction: true}) + require.NoError(t, err) + return db, mock, func() { _ = sqlDB.Close() } +} + +func expectMeasurementResolution(mock sqlmock.Sqlmock, mode int16) { + const componentUUID = "70c190f2-8a60-42a9-b143-ec5f87e0aa6b" + mock.ExpectQuery(`(?s)SELECT m\.\*.*WHERE c\.nspath = \$1.*AND m\.tag = \$2.*LIMIT 2`). + WithArgs("nspath", "measurement"). + WillReturnRows(sqlmock.NewRows([]string{ + "id", "tag", "mode", "data_source", "component_uuid", + }).AddRow(int64(10), "measurement", mode, `{"type":1,"io_address":{"station":"station","device":"device","channel":"tm1"}}`, componentUUID)) + mock.ExpectQuery(`(?s)SELECT global_uuid, nspath, tag, grid, zone, station.*WHERE global_uuid = \$1.*LIMIT 1`). + WithArgs(componentUUID). + WillReturnRows(sqlmock.NewRows([]string{"global_uuid", "nspath", "tag"}). + AddRow(componentUUID, "nspath", "component")) + mock.ExpectQuery(`(?s)SELECT \* FROM "measurement" WHERE id = \$1.*LIMIT \$2 FOR UPDATE`). + WithArgs(int64(10), 1). + WillReturnRows(sqlmock.NewRows([]string{ + "id", "tag", "mode", "data_source", "component_uuid", + }).AddRow(int64(10), "measurement", mode, `{"type":1,"io_address":{"station":"station","device":"device","channel":"tm1"}}`, componentUUID)) +} diff --git a/orm/circuit_diagram_measurement.go b/orm/circuit_diagram_measurement.go index 19dcd76..c52b237 100644 --- a/orm/circuit_diagram_measurement.go +++ b/orm/circuit_diagram_measurement.go @@ -9,19 +9,20 @@ import ( // Measurement structure define abstracted info set of electrical measurement type Measurement struct { - ID int64 `gorm:"column:id;primaryKey;autoIncrement"` - Tag string `gorm:"column:tag;size:64;not null;default:''"` - Name string `gorm:"column:name;size:64;not null;default:''"` - Type int16 `gorm:"column:type;not null;default:-1"` - Size int `gorm:"column:size;not null;default:-1"` - Mode int16 `gorm:"column:mode;not null;default:1"` - DataSource JSONMap `gorm:"column:data_source;type:jsonb;not null;default:'{}'"` - EventPlan JSONMap `gorm:"column:event_plan;type:jsonb;not null;default:'{}'"` - Binding JSONMap `gorm:"column:binding;type:jsonb;not null;default:'{\"ct\":{\"ratio\":1.0,\"polarity\":1,\"index\":0},\"pt\":{\"ratio\":1.0,\"polarity\":1,\"index\":0}}'"` - BayUUID uuid.UUID `gorm:"column:bay_uuid;type:uuid;not null"` - ComponentUUID uuid.UUID `gorm:"column:component_uuid;type:uuid;not null"` - Op int `gorm:"column:op;not null;default:-1"` - TS time.Time `gorm:"column:ts;type:timestamptz;not null;default:CURRENT_TIMESTAMP"` + ID int64 `gorm:"column:id;primaryKey;autoIncrement"` + Tag string `gorm:"column:tag;size:64;not null;default:'';uniqueIndex"` + Name string `gorm:"column:name;size:64;not null;default:''"` + Type int16 `gorm:"column:type;not null;default:-1"` + Size int `gorm:"column:size;not null;default:-1"` + Mode int16 `gorm:"column:mode;not null;default:1"` + Operations JSONMapArray `gorm:"column:operations;type:jsonb[];not null;default:'{}'"` + DataSource JSONMap `gorm:"column:data_source;type:jsonb;not null;default:'{}'"` + EventPlan JSONMap `gorm:"column:event_plan;type:jsonb;not null;default:'{}'"` + Binding JSONMap `gorm:"column:binding;type:jsonb;not null;default:'{\"ct\":{\"ratio\":1.0,\"polarity\":1,\"index\":0},\"pt\":{\"ratio\":1.0,\"polarity\":1,\"index\":0}}'"` + BayUUID uuid.UUID `gorm:"column:bay_uuid;type:uuid;not null"` + ComponentUUID uuid.UUID `gorm:"column:component_uuid;type:uuid;not null"` + Op int `gorm:"column:op;not null;default:-1"` + TS time.Time `gorm:"column:ts;type:timestamptz;not null;default:CURRENT_TIMESTAMP"` } // TableName func respresent return table name of Measurement diff --git a/orm/jsonb_serializer.go b/orm/jsonb_serializer.go index d43f7d2..c4898e4 100644 --- a/orm/jsonb_serializer.go +++ b/orm/jsonb_serializer.go @@ -5,6 +5,9 @@ import ( "database/sql/driver" "encoding/json" "errors" + "fmt" + + "github.com/jackc/pgx/v5/pgtype" ) // JSONMap define struct of implements the sql.Scanner and driver.Valuer interfaces for handling JSONB fields @@ -36,3 +39,48 @@ func (j *JSONMap) Scan(value any) error { } return json.Unmarshal(source, j) } + +// JSONMapArray represents a PostgreSQL jsonb[] column. +type JSONMapArray []JSONMap + +// Value encodes the slice as a PostgreSQL jsonb array. +func (j JSONMapArray) Value() (driver.Value, error) { + items := make(pgtype.FlatArray[map[string]any], len(j)) + for index, item := range j { + items[index] = map[string]any(item) + } + encoded, err := pgtype.NewMap().Encode(pgtype.JSONBArrayOID, pgtype.TextFormatCode, items, nil) + if err != nil { + return nil, fmt.Errorf("encode JSONMapArray: %w", err) + } + return string(encoded), nil +} + +// Scan decodes a PostgreSQL jsonb array. +func (j *JSONMapArray) Scan(value any) error { + if value == nil { + *j = nil + return nil + } + + var source []byte + switch typedValue := value.(type) { + case []byte: + source = typedValue + case string: + source = []byte(typedValue) + default: + return fmt.Errorf("unsupported data type %T for JSONMapArray Scan", value) + } + + var items pgtype.FlatArray[map[string]any] + if err := pgtype.NewMap().Scan(pgtype.JSONBArrayOID, pgtype.TextFormatCode, source, &items); err != nil { + return fmt.Errorf("decode JSONMapArray: %w", err) + } + result := make(JSONMapArray, len(items)) + for index, item := range items { + result[index] = JSONMap(item) + } + *j = result + return nil +} diff --git a/orm/jsonb_serializer_test.go b/orm/jsonb_serializer_test.go new file mode 100644 index 0000000..4478dec --- /dev/null +++ b/orm/jsonb_serializer_test.go @@ -0,0 +1,21 @@ +package orm + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestJSONMapArrayValueAndScan(t *testing.T) { + original := JSONMapArray{ + {"command": float64(0), "timestamp": "2026-07-20T00:00:00Z"}, + {"transaction": float64(1), "value": 15.2}, + } + + encoded, err := original.Value() + require.NoError(t, err) + + var decoded JSONMapArray + require.NoError(t, decoded.Scan(encoded)) + require.Equal(t, original, decoded) +}