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
This commit is contained in:
parent
d26260b2a7
commit
b85c2e129d
|
|
@ -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")
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
2
go.mod
2
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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
Loading…
Reference in New Issue