feat: merge refactor/optimize-parameter-measurement-value-api into develop

- expose bay dev_* columns for components linked to bay measurements
- restrict parameter queries and traversal to supported attribute-group tables
- add regression tests for bay candidates and parameter table validation

.gitignore

@@ -32,6 +32,7 @@ go.work
 # ai config
 .cursor/
 .claude/
+.codewhale/
 .cursorrules
 .copilot/
 .chatgpt/

common/data_object_errors.go

@@ -0,0 +1,23 @@
+// Package common define common error variables
+package common
+
+import "errors"
+
+var (
+	// ErrUnsupportedParameterField indicates that a requested parameter field is not supported.
+	ErrUnsupportedParameterField = errors.New("unsupported parameter field")
+	// ErrInvalidParameterToken indicates that a token cannot represent a parameter.
+	ErrInvalidParameterToken = errors.New("invalid parameter token")
+	// ErrParameterTokenNotFound indicates that no parameter matches the token hierarchy.
+	ErrParameterTokenNotFound = errors.New("parameter token not found")
+	// ErrAmbiguousParameterToken indicates that a token matches more than one parameter.
+	ErrAmbiguousParameterToken = errors.New("ambiguous parameter token")
+	// ErrUnsupportedMeasurementField define error of unsupport measurement field
+	ErrUnsupportedMeasurementField = errors.New("unsupported measurement field")
+	// ErrInvalidMeasurementToken indicates that a token cannot represent a measurement.
+	ErrInvalidMeasurementToken = errors.New("invalid measurement token")
+	// ErrMeasurementTokenNotFound indicates that no measurement matches the token hierarchy.
+	ErrMeasurementTokenNotFound = errors.New("measurement token not found")
+	// ErrAmbiguousMeasurementToken indicates that a token matches more than one measurement.
+	ErrAmbiguousMeasurementToken = errors.New("ambiguous measurement token")
+)

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")
 

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
 }
 

constants/context.go

@@ -1,7 +1,13 @@
 // Package constants define constant variable
 package constants
 
+// ClientTokenContextName is the Gin key used for the configured client token.
+const ClientTokenContextName = "client_token"
+
 type contextKey string
 
 // MeasurementUUIDKey define measurement uuid key into context
 const MeasurementUUIDKey contextKey = "measurement_uuid"
+
+// CtxKeyClientToken is the typed standard-library context key for client token propagation.
+const CtxKeyClientToken contextKey = ClientTokenContextName

constants/data-object.go

@@ -0,0 +1,19 @@
+// Package constants define constant variable
+package constants
+
+// DataObjectType identifies the kind of object represented by a data object token.
+type DataObjectType string
+
+const (
+	// DataObjectTypeParameter represents a component parameter.
+	DataObjectTypeParameter DataObjectType = "parameter"
+	// DataObjectTypeMeasurement represents a component measurement.
+	DataObjectTypeMeasurement DataObjectType = "measurement"
+)
+
+const (
+	// MeasurementModeManual indicates that manual value entry is enabled.
+	MeasurementModeManual int16 = 0
+	// MeasurementModeAutomatic indicates that the measurement runs automatically.
+	MeasurementModeAutomatic int16 = 1
+)

constants/measurement.go

@@ -19,15 +19,15 @@ const (
 
 // channel name suffix
 const (
-	ChannelSuffixP      = "P"
-	ChannelSuffixQ      = "Q"
-	ChannelSuffixS      = "S"
-	ChannelSuffixPS     = "PS"
-	ChannelSuffixF      = "F"
-	ChannelSuffixDeltaF = "deltaF"
-	ChannelSuffixUAB    = "UAB"
-	ChannelSuffixUBC    = "UBC"
-	ChannelSuffixUCA    = "UCA"
+	ChannelSuffixP      = "p"
+	ChannelSuffixQ      = "q"
+	ChannelSuffixS      = "s"
+	ChannelSuffixPF     = "pf"
+	ChannelSuffixF      = "f"
+	ChannelSuffixDeltaF = "df"
+	ChannelSuffixUAB    = "uab"
+	ChannelSuffixUBC    = "ubc"
+	ChannelSuffixUCA    = "uca"
 )
 
 const (

constants/parameter_table.go

@@ -0,0 +1,26 @@
+// Package constants define constant variable
+package constants
+
+import "strings"
+
+var supportedParameterTableSuffixes = [...]string{
+	"base_extend",
+	"rated",
+	"setup",
+	"model",
+	"stable",
+	"craft",
+	"integrity",
+	"behavior",
+}
+
+// IsSupportedParameterTableName reports whether a dynamic parameter table has
+// one of the supported attribute-group suffixes.
+func IsSupportedParameterTableName(tableName string) bool {
+	for _, suffix := range supportedParameterTableSuffixes {
+		if strings.HasSuffix(tableName, "_"+suffix) {
+			return true
+		}
+	}
+	return false
+}

constants/parameter_table_test.go

@@ -0,0 +1,25 @@
+package constants
+
+import "testing"
+
+func TestIsSupportedParameterTableName(t *testing.T) {
+	tests := []struct {
+		name      string
+		tableName string
+		want      bool
+	}{
+		{name: "bay table is excluded", tableName: "ct_ct_demo_bay", want: false},
+		{name: "model table is included", tableName: "cable_cable_demo_model", want: true},
+		{name: "base extend table is included", tableName: "cable_cable_demo_base_extend", want: true},
+		{name: "suffix must start at separator", tableName: "cable_cable_demomodel", want: false},
+		{name: "empty table name", tableName: "", want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsSupportedParameterTableName(tt.tableName); got != tt.want {
+				t.Fatalf("IsSupportedParameterTableName(%q) = %v, want %v", tt.tableName, got, tt.want)
+			}
+		})
+	}
+}

constants/recommend_keys.go

@@ -4,6 +4,8 @@ package constants
 const (
 	// DefaultScore define the default score for redissearch suggestion
 	DefaultScore = 1.0
+	// ComponentConfigKey define component config token used at token6
+	ComponentConfigKey = "component"
 )
 
 const (
@@ -42,6 +44,9 @@ const (
 
 	// RedisSpecCompTagMeasSetKey define redis set key which store all measurement tag keys under specific component tag
 	RedisSpecCompTagMeasSetKey = "%s_measurement_tag_keys"
+
+	// RedisSpecCompNSPathMeasSetKey define redis set key which store all measurement tag keys under specific component nspath
+	RedisSpecCompNSPathMeasSetKey = "%s_nspath_measurement_tag_keys"
 )
 
 const (

database/query_bay_columns.go

@@ -0,0 +1,28 @@
+package database
+
+import (
+	"context"
+	"strings"
+
+	"modelRT/orm"
+
+	"gorm.io/gorm"
+)
+
+// QueryBayDevColumnNames returns the bay table columns exposed as token7
+// candidates under token6=bay.
+func QueryBayDevColumnNames(ctx context.Context, db *gorm.DB) ([]string, error) {
+	columnTypes, err := db.WithContext(ctx).Migrator().ColumnTypes((&orm.Bay{}).TableName())
+	if err != nil {
+		return nil, err
+	}
+
+	columnNames := make([]string, 0, len(columnTypes))
+	for _, columnType := range columnTypes {
+		columnName := columnType.Name()
+		if strings.HasPrefix(columnName, "dev_") {
+			columnNames = append(columnNames, columnName)
+		}
+	}
+	return columnNames, nil
+}

database/query_component_columns.go

@@ -0,0 +1,28 @@
+// Package database define database operation functions
+package database
+
+import (
+	"context"
+
+	"modelRT/orm"
+
+	"gorm.io/gorm"
+)
+
+// QueryComponentColumnNames returns all column names from the component table.
+func QueryComponentColumnNames(ctx context.Context, db *gorm.DB) ([]string, error) {
+	columnTypes, err := db.WithContext(ctx).Migrator().ColumnTypes((&orm.Component{}).TableName())
+	if err != nil {
+		return nil, err
+	}
+
+	columnNames := make([]string, 0, len(columnTypes))
+	for _, columnType := range columnTypes {
+		columnName := columnType.Name()
+		if columnName == "" {
+			continue
+		}
+		columnNames = append(columnNames, columnName)
+	}
+	return columnNames, nil
+}

database/query_component_measurement.go

@@ -3,24 +3,270 @@ package database
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"strings"
+	"time"
 
+	"modelRT/common"
+	"modelRT/constants"
 	"modelRT/orm"
+	"modelRT/sql"
 
 	"golang.org/x/sync/errgroup"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
-type ZoneWithParent struct {
-	orm.Zone
-	GridTag string `gorm:"column:grid_tag"`
+const (
+	measurementOperationsLimit    = 500
+	measurementOperationAppendSQL = "(array_append(operations, ?::jsonb))[GREATEST(cardinality(operations) - ? + 2, 1):]"
+)
+
+// QueryMeasurementByID returns a measurement by primary key without acquiring
+// a row lock. Call QueryMeasurementByIDForUpdate for write workflows.
+func QueryMeasurementByID(ctx context.Context, db *gorm.DB, id int64) (orm.Measurement, error) {
+	var measurement orm.Measurement
+	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	result := db.WithContext(cancelCtx).
+		Where(sql.MeasurementIDWhere, id).
+		Take(&measurement)
+
+	if result.Error != nil {
+		return orm.Measurement{}, fmt.Errorf("query measurement %d: %w", id, result.Error)
+	}
+	return measurement, nil
 }
 
-type StationWithParent struct {
-	orm.Zone
-	ZoneTag string `gorm:"column:zone_tag"`
+// QueryMeasurementByIDForUpdate locks a measurement row and loads only the
+// fields required by the data-object update workflow.
+func QueryMeasurementByIDForUpdate(ctx context.Context, tx *gorm.DB, id int64) (orm.Measurement, error) {
+	var measurement orm.Measurement
+	result := tx.WithContext(ctx).
+		Select("id", "mode", "data_source").
+		Where(sql.MeasurementIDWhere, id).
+		Clauses(clause.Locking{Strength: "UPDATE"}).
+		Take(&measurement)
+	if result.Error != nil {
+		return orm.Measurement{}, fmt.Errorf("lock measurement %d: %w", id, result.Error)
+	}
+	return measurement, nil
 }
 
+// QueryMeasurementByToken define function query circuit diagram component measurement info by token from postgresDB
+func QueryMeasurementByToken(ctx context.Context, tx *gorm.DB, token string) (orm.Measurement, error) {
+	measurement, _, err := QueryMeasurementByDataObjectToken(ctx, tx, token)
+	if err != nil {
+		return orm.Measurement{}, err
+	}
+	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 := constants.MeasurementModeManual
+	if automatic {
+		mode = constants.MeasurementModeAutomatic
+	}
+
+	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, mode int16, timestamp time.Time) error {
+	if mode != constants.MeasurementModeManual && mode != constants.MeasurementModeAutomatic {
+		return fmt.Errorf("measurement mode must be 0 or 1, got %d", mode)
+	}
+	operation := orm.JSONMap{
+		"command":   mode,
+		"timestamp": timestamp.UnixMilli(),
+	}
+	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.UnixMilli(),
+	}
+	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(
+		measurementOperationAppendSQL,
+		string(encodedOperation),
+		measurementOperationsLimit,
+	)
+	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
+// token4.token7.
+func ValidateMeasurementToken(ctx context.Context, db *gorm.DB, token string) error {
+	query, args, err := buildMeasurementTokenValidationQuery(token)
+	if err != nil {
+		return err
+	}
+
+	var count int64
+	if err := db.WithContext(ctx).Raw(query, args...).Scan(&count).Error; err != nil {
+		return fmt.Errorf("query measurement token %q: %w", token, err)
+	}
+
+	switch {
+	case count == 0:
+		return fmt.Errorf("%w: %q", common.ErrMeasurementTokenNotFound, token)
+	case count > 1:
+		return fmt.Errorf("%w: %q matched %d records", common.ErrAmbiguousMeasurementToken, token, count)
+	default:
+		return nil
+	}
+}
+
+// QueryMeasurementByDataObjectToken validates token and returns the existing
+// measurement and its owning component for attribute response construction.
+func QueryMeasurementByDataObjectToken(ctx context.Context, db *gorm.DB, token string) (*orm.Measurement, *orm.Component, error) {
+	validationQuery, args, err := buildMeasurementTokenValidationQuery(token)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	query := buildMeasurementRowsQuery(validationQuery)
+	var rows []orm.Measurement
+	if err := db.WithContext(ctx).Raw(query, args...).Scan(&rows).Error; err != nil {
+		return nil, nil, fmt.Errorf("query measurement token %q: %w", token, err)
+	}
+
+	switch len(rows) {
+	case 0:
+		return nil, nil, fmt.Errorf("%w: %q", common.ErrMeasurementTokenNotFound, token)
+	case 1:
+		// Continue by loading the owning component.
+	default:
+		return nil, nil, fmt.Errorf("%w: %q matched more than one record", common.ErrAmbiguousMeasurementToken, token)
+	}
+
+	var component orm.Component
+	result := db.WithContext(ctx).
+		Raw(compactMeasurementSQL(sql.MeasurementComponentByUUID), rows[0].ComponentUUID).
+		Scan(&component)
+	if result.Error != nil {
+		return nil, nil, fmt.Errorf("query component for measurement token %q: %w", token, result.Error)
+	}
+	if result.RowsAffected == 0 {
+		return nil, nil, fmt.Errorf("%w: component for %q", common.ErrMeasurementTokenNotFound, token)
+	}
+
+	return &rows[0], &component, nil
+}
+
+func buildMeasurementRowsQuery(validationQuery string) string {
+	measurementQuery := strings.Replace(
+		validationQuery,
+		sql.MeasurementCountSelect,
+		sql.MeasurementRowsSelect,
+		1,
+	)
+	return compactMeasurementSQL(strings.Join([]string{measurementQuery, sql.MeasurementLimitTwo}, "\n"))
+}
+
+func compactMeasurementSQL(statement string) string {
+	return strings.Join(strings.Fields(statement), " ")
+}
+
+func buildMeasurementTokenValidationQuery(token string) (string, []any, error) {
+	parts := strings.Split(token, ".")
+	for _, part := range parts {
+		if part == "" {
+			return "", nil, fmt.Errorf("%w %q: token segment cannot be empty", common.ErrInvalidMeasurementToken, token)
+		}
+	}
+
+	switch len(parts) {
+	case 7:
+		if parts[5] != "bay" {
+			return "", nil, fmt.Errorf("%w %q: token6 must be bay", common.ErrInvalidMeasurementToken, token)
+		}
+		query := compactMeasurementSQL(strings.Join([]string{
+			sql.MeasurementTokenValidationQueryBase,
+			sql.MeasurementSevenPartTokenWhere,
+		}, "\n"))
+		return query, []any{parts[0], parts[1], parts[2], parts[3], parts[4], parts[6]}, nil
+	case 4:
+		if parts[2] != "bay" {
+			return "", nil, fmt.Errorf("%w %q: token6 must be bay", common.ErrInvalidMeasurementToken, token)
+		}
+		query := compactMeasurementSQL(strings.Join([]string{
+			sql.MeasurementTokenValidationQueryBase,
+			sql.MeasurementFourPartTokenWhere,
+		}, "\n"))
+		return query, []any{parts[0], parts[1], parts[3]}, nil
+	case 2:
+		query := compactMeasurementSQL(strings.Join([]string{
+			sql.MeasurementTokenValidationQueryBase,
+			sql.MeasurementTwoPartTokenWhere,
+		}, "\n"))
+		return query, []any{parts[0], parts[1]}, nil
+	default:
+		return "", nil, fmt.Errorf("%w %q: expected 2, 4, or 7 segments, got %d", common.ErrInvalidMeasurementToken, token, len(parts))
+	}
+}
+
+// GetAllMeasurements define func to query all measurement info from postgresDB
+func GetAllMeasurements(ctx context.Context, tx *gorm.DB) ([]orm.Measurement, error) {
+	var measurements []orm.Measurement
+
+	// ctx超时判断
+	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	result := tx.WithContext(cancelCtx).Clauses(clause.Locking{Strength: "UPDATE"}).Find(&measurements)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+	return measurements, nil
+}
+
+// GetFullMeasurementSet queries all hierarchy tags required to build
+// measurement recommendations.
 func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSet, error) {
 	mSet := &orm.MeasurementSet{
 		GridToZoneTags:       make(map[string][]string),
@@ -28,14 +274,40 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe
 		StationToCompNSPaths: make(map[string][]string),
 		CompNSPathToCompTags: make(map[string][]string),
 		CompTagToMeasTags:    make(map[string][]string),
+		CompNSPathToMeasTags: make(map[string][]string),
 	}
 
 	g, gctx := errgroup.WithContext(ctx)
 	db = db.WithContext(gctx)
+	var bayLinkedCompTags []string
+	var bayDevColumnNames []string
+
+	g.Go(func() error {
+		var linkedComponents []struct {
+			CompTag string `gorm:"column:comp_tag"`
+		}
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementBayLinkedComponentTags)).Scan(&linkedComponents).Error; err != nil {
+			return fmt.Errorf("query bay-linked components: %w", err)
+		}
+		bayLinkedCompTags = make([]string, 0, len(linkedComponents))
+		for _, component := range linkedComponents {
+			bayLinkedCompTags = append(bayLinkedCompTags, component.CompTag)
+		}
+		return nil
+	})
+
+	g.Go(func() error {
+		var err error
+		bayDevColumnNames, err = QueryBayDevColumnNames(gctx, db)
+		if err != nil {
+			return fmt.Errorf("query bay dev columns: %w", err)
+		}
+		return nil
+	})
 
 	g.Go(func() error {
 		var grids []orm.Grid
-		if err := db.Table("grid").Select("tagname").Scan(&grids).Error; err != nil {
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementGridTags)).Scan(&grids).Error; err != nil {
 			return fmt.Errorf("query grids: %w", err)
 		}
 		for _, grid := range grids {
@@ -51,16 +323,13 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe
 			orm.Zone
 			GridTag string `gorm:"column:grid_tag"`
 		}
-		if err := db.Table("zone").
-			Select("zone.*, grid.tagname as grid_tag").
-			Joins("left join grid on zone.grid_id = grid.id").
-			Scan(&zones).Error; err != nil {
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementZoneHierarchy)).Scan(&zones).Error; err != nil {
 			return fmt.Errorf("query zones: %w", err)
 		}
-		for _, z := range zones {
-			mSet.AllZoneTags = append(mSet.AllZoneTags, z.TAGNAME)
-			if z.GridTag != "" {
-				mSet.GridToZoneTags[z.GridTag] = append(mSet.GridToZoneTags[z.GridTag], z.TAGNAME)
+		for _, zone := range zones {
+			mSet.AllZoneTags = append(mSet.AllZoneTags, zone.TAGNAME)
+			if zone.GridTag != "" {
+				mSet.GridToZoneTags[zone.GridTag] = append(mSet.GridToZoneTags[zone.GridTag], zone.TAGNAME)
 			}
 		}
 		return nil
@@ -71,40 +340,40 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe
 			orm.Station
 			ZoneTag string `gorm:"column:zone_tag"`
 		}
-		if err := db.Table("station").
-			Select("station.*, zone.tagname as zone_tag").
-			Joins("left join zone on station.zone_id = zone.id").
-			Scan(&stations).Error; err != nil {
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementStationHierarchy)).Scan(&stations).Error; err != nil {
 			return fmt.Errorf("query stations: %w", err)
 		}
-		for _, s := range stations {
-			mSet.AllStationTags = append(mSet.AllStationTags, s.TAGNAME)
-			if s.ZoneTag != "" {
-				mSet.ZoneToStationTags[s.ZoneTag] = append(mSet.ZoneToStationTags[s.ZoneTag], s.TAGNAME)
+		for _, station := range stations {
+			mSet.AllStationTags = append(mSet.AllStationTags, station.TAGNAME)
+			if station.ZoneTag != "" {
+				mSet.ZoneToStationTags[station.ZoneTag] = append(mSet.ZoneToStationTags[station.ZoneTag], station.TAGNAME)
 			}
 		}
 		return nil
 	})
 
 	g.Go(func() error {
-		var comps []struct {
+		var components []struct {
 			orm.Component
 			StationTag string `gorm:"column:station_tag"`
 		}
-		if err := db.Table("component").
-			Select("component.*, station.tagname as station_tag").
-			Joins("left join station on component.station_id = station.id").
-			Scan(&comps).Error; err != nil {
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementComponentHierarchy)).Scan(&components).Error; err != nil {
 			return fmt.Errorf("query components: %w", err)
 		}
-		for _, c := range comps {
-			mSet.AllCompNSPaths = append(mSet.AllCompNSPaths, c.NSPath)
-			mSet.AllCompTags = append(mSet.AllCompTags, c.Tag)
-			if c.StationTag != "" {
-				mSet.StationToCompNSPaths[c.StationTag] = append(mSet.StationToCompNSPaths[c.StationTag], c.NSPath)
+		for _, component := range components {
+			mSet.AllCompNSPaths = append(mSet.AllCompNSPaths, component.NSPath)
+			mSet.AllCompTags = append(mSet.AllCompTags, component.Tag)
+			if component.StationTag != "" {
+				mSet.StationToCompNSPaths[component.StationTag] = append(
+					mSet.StationToCompNSPaths[component.StationTag],
+					component.NSPath,
+				)
 			}
-			if c.NSPath != "" {
-				mSet.CompNSPathToCompTags[c.NSPath] = append(mSet.CompNSPathToCompTags[c.NSPath], c.Tag)
+			if component.NSPath != "" {
+				mSet.CompNSPathToCompTags[component.NSPath] = append(
+					mSet.CompNSPathToCompTags[component.NSPath],
+					component.Tag,
+				)
 			}
 		}
 		return nil
@@ -113,18 +382,23 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe
 	g.Go(func() error {
 		var measurements []struct {
 			orm.Measurement
-			CompTag string `gorm:"column:comp_tag"`
+			CompTag    string `gorm:"column:comp_tag"`
+			CompNSPath string `gorm:"column:comp_nspath"`
+			BayTag     string `gorm:"column:bay_tag"`
 		}
-		if err := db.Table("measurement").
-			Select("measurement.*, component.tag as comp_tag").
-			Joins("left join component on measurement.component_uuid = component.global_uuid").
-			Scan(&measurements).Error; err != nil {
+		if err := db.Raw(compactMeasurementSQL(sql.MeasurementTagHierarchy)).Scan(&measurements).Error; err != nil {
 			return fmt.Errorf("query measurements: %w", err)
 		}
-		for _, m := range measurements {
-			mSet.AllMeasTags = append(mSet.AllMeasTags, m.Tag)
-			if m.CompTag != "" {
-				mSet.CompTagToMeasTags[m.CompTag] = append(mSet.CompTagToMeasTags[m.CompTag], m.Tag)
+		for _, measurement := range measurements {
+			mSet.AllMeasTags = append(mSet.AllMeasTags, measurement.Tag)
+			if measurement.CompTag != "" {
+				mSet.CompTagToMeasTags[measurement.CompTag] = append(
+					mSet.CompTagToMeasTags[measurement.CompTag],
+					measurement.Tag,
+				)
+			}
+			if measurement.CompNSPath != "" && measurement.CompNSPath == measurement.BayTag {
+				mSet.CompNSPathToMeasTags[measurement.CompNSPath] = append(mSet.CompNSPathToMeasTags[measurement.CompNSPath], measurement.Tag)
 			}
 		}
 		return nil
@@ -134,6 +408,22 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe
 		return nil, err
 	}
 
+	appendBayDevCandidates(mSet, bayLinkedCompTags, bayDevColumnNames)
+
 	mSet.AllConfigTags = append(mSet.AllConfigTags, "bay")
 	return mSet, nil
 }
+
+func appendBayDevCandidates(mSet *orm.MeasurementSet, bayLinkedCompTags, bayDevColumnNames []string) {
+	if mSet == nil || len(bayLinkedCompTags) == 0 || len(bayDevColumnNames) == 0 {
+		return
+	}
+
+	mSet.AllMeasTags = append(mSet.AllMeasTags, bayDevColumnNames...)
+	for _, compTag := range bayLinkedCompTags {
+		mSet.CompTagToMeasTags[compTag] = append(
+			mSet.CompTagToMeasTags[compTag],
+			bayDevColumnNames...,
+		)
+	}
+}

database/query_component_measurement_bay_test.go

@@ -0,0 +1,52 @@
+package database
+
+import (
+	"testing"
+
+	"modelRT/orm"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAppendBayDevCandidatesRequiresBayLinkedComponent(t *testing.T) {
+	measurementSet := &orm.MeasurementSet{
+		AllMeasTags: []string{"current"},
+		CompTagToMeasTags: map[string][]string{
+			"linked-component":   {"current"},
+			"unlinked-component": {"voltage"},
+		},
+	}
+
+	appendBayDevCandidates(
+		measurementSet,
+		[]string{"linked-component"},
+		[]string{"dev_instruct", "dev_dyn_sense", "dev_fault_record"},
+	)
+
+	require.Equal(t,
+		[]string{"current", "dev_instruct", "dev_dyn_sense", "dev_fault_record"},
+		measurementSet.AllMeasTags,
+	)
+	require.Equal(t,
+		[]string{"current", "dev_instruct", "dev_dyn_sense", "dev_fault_record"},
+		measurementSet.CompTagToMeasTags["linked-component"],
+	)
+	require.Equal(t,
+		[]string{"voltage"},
+		measurementSet.CompTagToMeasTags["unlinked-component"],
+	)
+}
+
+func TestAppendBayDevCandidatesWithoutBayLinkDoesNothing(t *testing.T) {
+	measurementSet := &orm.MeasurementSet{
+		AllMeasTags: []string{"current"},
+		CompTagToMeasTags: map[string][]string{
+			"component": {"current"},
+		},
+	}
+
+	appendBayDevCandidates(measurementSet, nil, []string{"dev_instruct"})
+
+	require.Equal(t, []string{"current"}, measurementSet.AllMeasTags)
+	require.Equal(t, []string{"current"}, measurementSet.CompTagToMeasTags["component"])
+}

database/query_component_measurement_test.go

@@ -0,0 +1,271 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"testing"
+
+	"modelRT/common"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestBuildMeasurementTokenValidationQuery(t *testing.T) {
+	tests := []struct {
+		name      string
+		token     string
+		wantArgs  []any
+		wantWhere string
+		wantErr   bool
+	}{
+		{
+			name:      "seven-part token",
+			token:     "grid.zone.station.nspath.component.bay.measurement",
+			wantArgs:  []any{"grid", "zone", "station", "nspath", "component", "measurement"},
+			wantWhere: "WHERE g.tagname = ?",
+		},
+		{
+			name:      "four-part token",
+			token:     "nspath.component.bay.measurement",
+			wantArgs:  []any{"nspath", "component", "measurement"},
+			wantWhere: "WHERE c.nspath = ?",
+		},
+		{
+			name:      "two-part token",
+			token:     "nspath.measurement",
+			wantArgs:  []any{"nspath", "measurement"},
+			wantWhere: "WHERE c.nspath = ?",
+		},
+		{
+			name:    "non-bay group",
+			token:   "nspath.component.rated.attribute",
+			wantErr: true,
+		},
+		{
+			name:    "empty segment",
+			token:   "nspath..bay.measurement",
+			wantErr: true,
+		},
+		{
+			name:    "invalid segment count",
+			token:   "grid.zone.station",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			query, args, err := buildMeasurementTokenValidationQuery(tt.token)
+			if tt.wantErr {
+				require.Error(t, err)
+				assert.ErrorIs(t, err, common.ErrInvalidMeasurementToken)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Contains(t, query, "INNER JOIN component AS c ON c.global_uuid = m.component_uuid")
+			assert.Contains(t, query, "INNER JOIN bay AS b ON b.bay_uuid = m.bay_uuid")
+			assert.Contains(t, query, tt.wantWhere)
+			assert.NotContains(t, query, "grid_idWHERE")
+			assert.Regexp(t, `grid_id\s+WHERE`, query)
+			assert.NotContains(t, query, "\n")
+			assert.NotContains(t, query, "\t")
+			assert.Equal(t, tt.wantArgs, args)
+		})
+	}
+}
+
+func TestQueryMeasurementByIDDoesNotLockRead(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.ExpectQuery(`SELECT \* FROM "measurement" WHERE id = \$1 LIMIT \$2`).
+		WithArgs(int64(10), 1).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "mode"}).AddRow(int64(10), int16(1)))
+
+	measurement, err := QueryMeasurementByID(context.Background(), db, 10)
+	require.NoError(t, err)
+	assert.Equal(t, int64(10), measurement.ID)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestQueryMeasurementByIDForUpdateSelectsOnlyRequiredFields(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.ExpectQuery(`SELECT "id","mode","data_source" FROM "measurement" WHERE id = \$1 LIMIT \$2 FOR UPDATE`).
+		WithArgs(int64(10), 1).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "mode", "data_source"}).
+			AddRow(int64(10), int16(1), `{"type":1}`))
+
+	measurement, err := QueryMeasurementByIDForUpdate(context.Background(), db, 10)
+	require.NoError(t, err)
+	assert.Equal(t, int64(10), measurement.ID)
+	assert.Equal(t, int16(1), measurement.Mode)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestMeasurementOperationAppendSQLIsSingleLine(t *testing.T) {
+	assert.NotContains(t, measurementOperationAppendSQL, "\n")
+	assert.NotContains(t, measurementOperationAppendSQL, "\r")
+	assert.NotContains(t, measurementOperationAppendSQL, "\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
+		count    int64
+		queryErr error
+		wantErr  error
+	}{
+		{name: "exists", count: 1},
+		{name: "not found", count: 0, wantErr: common.ErrMeasurementTokenNotFound},
+		{name: "ambiguous", count: 2, wantErr: common.ErrAmbiguousMeasurementToken},
+		{name: "query failure", queryErr: errors.New("database unavailable")},
+	}
+
+	const token = "nspath.measurement"
+	query, _, err := buildMeasurementTokenValidationQuery(token)
+	require.NoError(t, err)
+	expectedQuery := query
+	for i := 1; strings.Contains(expectedQuery, "?"); i++ {
+		expectedQuery = strings.Replace(expectedQuery, "?", fmt.Sprintf("$%d", i), 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{})
+			require.NoError(t, err)
+
+			expectation := mock.ExpectQuery(regexp.QuoteMeta(expectedQuery)).
+				WithArgs("nspath", "measurement")
+			if tt.queryErr != nil {
+				expectation.WillReturnError(tt.queryErr)
+			} else {
+				expectation.WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(tt.count))
+			}
+
+			err = ValidateMeasurementToken(context.Background(), db, token)
+			if tt.wantErr != nil {
+				assert.ErrorIs(t, err, tt.wantErr)
+			} else if tt.queryErr != nil {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.queryErr.Error())
+			} else {
+				require.NoError(t, err)
+			}
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestBuildMeasurementRowsQuerySeparatesLimitClause(t *testing.T) {
+	validationQuery, _, err := buildMeasurementTokenValidationQuery("nspath.measurement")
+	require.NoError(t, err)
+
+	query := buildMeasurementRowsQuery(validationQuery)
+	assert.NotContains(t, query, "?LIMIT")
+	assert.Regexp(t, `m\.tag = \?\s+LIMIT 2$`, query)
+	assert.NotContains(t, query, "\n")
+	assert.NotContains(t, query, "\t")
+}
+
+func TestQueryMeasurementByDataObjectToken(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{})
+	require.NoError(t, err)
+
+	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",
+			"name",
+			"mode",
+			"size",
+			"data_source",
+			"event_plan",
+			"binding",
+			"component_uuid",
+		}).AddRow(
+			int64(10),
+			"measurement",
+			"A phase current",
+			int16(1),
+			10,
+			`{"type":1,"io_address":{"channel":"tm1"}}`,
+			`{"enabled":true}`,
+			`{"ct":{"ratio":1}}`,
+			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",
+			"grid",
+			"zone",
+			"station",
+		}).AddRow(componentUUID, "nspath", "component", "grid", "zone", "station"))
+
+	measurement, component, err := QueryMeasurementByDataObjectToken(context.Background(), db, "nspath.measurement")
+	require.NoError(t, err)
+	assert.Equal(t, int64(10), measurement.ID)
+	assert.Equal(t, int16(1), measurement.Mode)
+	assert.Equal(t, float64(1), measurement.DataSource["type"])
+	assert.Equal(t, "grid", component.GridName)
+	assert.Equal(t, "component", component.Tag)
+	require.NoError(t, mock.ExpectationsWereMet())
+}

database/query_component_parameter.go

@@ -0,0 +1,248 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"strings"
+
+	"modelRT/common"
+	"modelRT/constants"
+	"modelRT/model"
+	"modelRT/orm"
+	modelsql "modelRT/sql"
+
+	"gorm.io/gorm"
+)
+
+var parameterTableNamePattern = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
+
+// ParameterDataObject contains the resolved metadata needed to query a
+// parameter attribute after its token has passed hierarchy validation.
+type ParameterDataObject struct {
+	Component      orm.Component
+	Project        orm.ProjectManager
+	TableName      string
+	AttributeGroup string
+	AttributeName  string
+	AttributeType  string
+}
+
+// QueryParameterByDataObjectToken validates a four-part or seven-part
+// parameter token. The component group resolves directly to the component
+// table; other groups resolve through project_manager and a dynamic table.
+func QueryParameterByDataObjectToken(ctx context.Context, db *gorm.DB, token string) (*ParameterDataObject, error) {
+	componentQuery, componentArgs, parts, err := buildParameterComponentQuery(token)
+	if err != nil {
+		return nil, err
+	}
+
+	var components []orm.Component
+	if err := db.WithContext(ctx).Raw(componentQuery, componentArgs...).Scan(&components).Error; err != nil {
+		return nil, fmt.Errorf("query component for parameter token %q: %w", token, err)
+	}
+
+	switch len(components) {
+	case 0:
+		return nil, fmt.Errorf("%w: component hierarchy for %q", common.ErrParameterTokenNotFound, token)
+	case 1:
+		// Continue by resolving the component model and attribute group.
+	default:
+		return nil, fmt.Errorf("%w: component hierarchy for %q matched more than one record", common.ErrAmbiguousParameterToken, token)
+	}
+
+	attributeGroup := parts[len(parts)-2]
+	attributeName := parts[len(parts)-1]
+	component := components[0]
+	if attributeGroup == "component" {
+		attributeType, err := queryParameterAttributeType(ctx, db, "component", attributeName, token)
+		if err != nil {
+			return nil, err
+		}
+		return &ParameterDataObject{
+			Component:      component,
+			TableName:      "component",
+			AttributeGroup: attributeGroup,
+			AttributeName:  attributeName,
+			AttributeType:  attributeType,
+		}, nil
+	}
+
+	var projects []orm.ProjectManager
+	if err := db.WithContext(ctx).
+		Where("tag = ? AND group_name = ?", component.ModelName, attributeGroup).
+		Limit(2).
+		Find(&projects).Error; err != nil {
+		return nil, fmt.Errorf("query project mapping for parameter token %q: %w", token, err)
+	}
+
+	switch len(projects) {
+	case 0:
+		return nil, fmt.Errorf("%w: model %q does not define attribute group %q", common.ErrParameterTokenNotFound, component.ModelName, attributeGroup)
+	case 1:
+		// Continue by validating the dynamic table and attribute.
+	default:
+		return nil, fmt.Errorf("%w: model %q and attribute group %q matched more than one project", common.ErrAmbiguousParameterToken, component.ModelName, attributeGroup)
+	}
+
+	project := projects[0]
+	if !validParameterTableName(project.Name) {
+		return nil, fmt.Errorf("project mapping for parameter token %q contains invalid table name %q", token, project.Name)
+	}
+	attributeType, err := queryParameterAttributeType(ctx, db, project.Name, attributeName, token)
+	if err != nil {
+		return nil, err
+	}
+
+	var recordCount int64
+	if err := db.WithContext(ctx).
+		Table(project.Name).
+		Where("global_uuid = ? AND attribute_group = ?", component.GlobalUUID, attributeGroup).
+		Count(&recordCount).Error; err != nil {
+		return nil, fmt.Errorf("query dynamic record for parameter token %q: %w", token, err)
+	}
+
+	switch {
+	case recordCount == 0:
+		return nil, fmt.Errorf("%w: component %q has no %q parameter record", common.ErrParameterTokenNotFound, component.Tag, attributeGroup)
+	case recordCount > 1:
+		return nil, fmt.Errorf("%w: component %q has %d %q parameter records", common.ErrAmbiguousParameterToken, component.Tag, recordCount, attributeGroup)
+	}
+
+	return &ParameterDataObject{
+		Component:      component,
+		Project:        project,
+		TableName:      project.Name,
+		AttributeGroup: attributeGroup,
+		AttributeName:  attributeName,
+		AttributeType:  attributeType,
+	}, nil
+}
+
+// QueryParameterDataObjectValue returns token7 from the component row or from
+// a dynamic parameter row identified during token validation.
+func QueryParameterDataObjectValue(ctx context.Context, db *gorm.DB, parameter *ParameterDataObject) (any, error) {
+	if parameter == nil {
+		return nil, fmt.Errorf("parameter data object is nil")
+	}
+
+	var record map[string]any
+	query := db.WithContext(ctx).Table(parameter.TableName)
+	if parameter.AttributeGroup == "component" {
+		query = query.Where("tag = ?", parameter.Component.Tag)
+	} else {
+		query = query.Where("global_uuid = ? AND attribute_group = ?", parameter.Component.GlobalUUID, parameter.AttributeGroup)
+	}
+	result := query.Take(&record)
+	if result.Error != nil {
+		return nil, fmt.Errorf("query parameter value from table %q: %w", parameter.TableName, result.Error)
+	}
+
+	value, ok := record[parameter.AttributeName]
+	if !ok {
+		return nil, fmt.Errorf("parameter column %q is missing from table %q result", parameter.AttributeName, parameter.TableName)
+	}
+	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 !validParameterTableName(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
+}
+
+func validParameterTableName(tableName string) bool {
+	return parameterTableNamePattern.MatchString(tableName) && constants.IsSupportedParameterTableName(tableName)
+}
+
+// QueryParameterAttributeDescription returns the display name registered for
+// token7 in basic.attribute.
+func QueryParameterAttributeDescription(ctx context.Context, db *gorm.DB, attributeName string) (string, error) {
+	var rows []struct {
+		Description string `gorm:"column:attribute_name"`
+	}
+	if err := db.WithContext(ctx).
+		Raw(modelsql.ParameterAttributeDescription, attributeName).
+		Scan(&rows).Error; err != nil {
+		return "", fmt.Errorf("query parameter description for attribute %q: %w", attributeName, err)
+	}
+
+	switch len(rows) {
+	case 0:
+		return "", fmt.Errorf("parameter description not found for attribute %q", attributeName)
+	case 1:
+		return rows[0].Description, nil
+	default:
+		return "", fmt.Errorf("ambiguous parameter description for attribute %q", attributeName)
+	}
+}
+
+func queryParameterAttributeType(ctx context.Context, db *gorm.DB, tableName, attributeName, token string) (string, error) {
+	var attributeType string
+	result := db.WithContext(ctx).
+		Raw(modelsql.ParameterAttributeColumnType, tableName, attributeName).
+		Scan(&attributeType)
+	if result.Error != nil {
+		return "", fmt.Errorf("validate attribute column for parameter token %q: %w", token, result.Error)
+	}
+	if result.RowsAffected == 0 || attributeType == "" {
+		return "", fmt.Errorf("%w: column %q does not exist in parameter table %q", common.ErrParameterTokenNotFound, attributeName, tableName)
+	}
+	return strings.ToUpper(attributeType), nil
+}
+
+func buildParameterComponentQuery(token string) (string, []any, []string, error) {
+	dataObjectType, err := model.ClassifyDataObjectToken(token)
+	if err != nil {
+		return "", nil, nil, fmt.Errorf("%w %q: %v", common.ErrInvalidParameterToken, token, err)
+	}
+	if dataObjectType != constants.DataObjectTypeParameter {
+		return "", nil, nil, fmt.Errorf("%w %q: token does not identify a parameter", common.ErrInvalidParameterToken, token)
+	}
+
+	parts := strings.Split(token, ".")
+	var where string
+	var args []any
+	switch len(parts) {
+	case 7:
+		where = modelsql.ParameterSevenPartTokenWhere
+		args = []any{parts[0], parts[1], parts[2], parts[3], parts[4]}
+	case 4:
+		where = modelsql.ParameterFourPartTokenWhere
+		args = []any{parts[0], parts[1]}
+	default:
+		return "", nil, nil, fmt.Errorf("%w %q: expected 4 or 7 segments", common.ErrInvalidParameterToken, token)
+	}
+
+	query := compactParameterSQL(strings.Join([]string{
+		modelsql.ParameterComponentQueryBase,
+		where,
+		modelsql.ParameterLimitTwo,
+	}, "\n"))
+	return query, args, parts, nil
+}
+
+func compactParameterSQL(statement string) string {
+	return strings.Join(strings.Fields(statement), " ")
+}

database/query_component_parameter_test.go

@@ -0,0 +1,286 @@
+package database
+
+import (
+	"context"
+	"regexp"
+	"testing"
+
+	"modelRT/common"
+	"modelRT/orm"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/gofrs/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestBuildParameterComponentQuery(t *testing.T) {
+	tests := []struct {
+		name      string
+		token     string
+		wantArgs  []any
+		wantWhere string
+		wantErr   bool
+	}{
+		{
+			name:      "seven-part parameter",
+			token:     "grid.zone.station.nspath.component.stable.attribute",
+			wantArgs:  []any{"grid", "zone", "station", "nspath", "component"},
+			wantWhere: "WHERE g.tagname = ?",
+		},
+		{
+			name:      "four-part local parameter",
+			token:     "nspath.component.rated.attribute",
+			wantArgs:  []any{"nspath", "component"},
+			wantWhere: "s.is_local = TRUE",
+		},
+		{
+			name:    "measurement token",
+			token:   "nspath.component.bay.measurement",
+			wantErr: true,
+		},
+		{
+			name:    "empty segment",
+			token:   "nspath..stable.attribute",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			query, args, _, err := buildParameterComponentQuery(tt.token)
+			if tt.wantErr {
+				require.Error(t, err)
+				assert.ErrorIs(t, err, common.ErrInvalidParameterToken)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Contains(t, query, "INNER JOIN station AS s ON s.id = c.station_id")
+			assert.Contains(t, query, tt.wantWhere)
+			assert.Equal(t, tt.wantArgs, args)
+			assert.NotContains(t, query, "\n")
+			assert.NotContains(t, query, "\t")
+		})
+	}
+}
+
+func TestQueryParameterByDataObjectToken(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{})
+	require.NoError(t, err)
+
+	const (
+		token         = "grid.zone.station.nspath.component.stable.rated_voltage"
+		componentUUID = "70c190f2-8a60-42a9-b143-ec5f87e0aa6b"
+	)
+
+	mock.ExpectQuery(`(?s)SELECT c\.\*.*WHERE g\.tagname = \$1.*AND c\.tag = \$5.*LIMIT 2`).
+		WithArgs("grid", "zone", "station", "nspath", "component").
+		WillReturnRows(sqlmock.NewRows([]string{
+			"global_uuid", "nspath", "tag", "model_name", "station_id",
+		}).AddRow(componentUUID, "nspath", "component", "bus_1", int64(10)))
+
+	mock.ExpectQuery(`SELECT \* FROM "project_manager" WHERE tag = \$1 AND group_name = \$2 LIMIT \$3`).
+		WithArgs("bus_1", "stable", 2).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "name", "tag", "meta_model", "group_name", "link_type", "check_state", "ispublic",
+		}).AddRow(
+			int32(1),
+			"bus_bus_1_stable",
+			"bus_1",
+			"bus",
+			"stable",
+			int32(0),
+			`{"checkState":[{"name":"rated_voltage","checked":1}]}`,
+			false,
+		))
+
+	mock.ExpectQuery(`(?s)SELECT pg_catalog\.format_type.*pg_catalog\.pg_attribute.*c\.relname = \$1.*a\.attname = \$2.*LIMIT 1`).
+		WithArgs("bus_bus_1_stable", "rated_voltage").
+		WillReturnRows(sqlmock.NewRows([]string{"format_type"}).AddRow("double precision"))
+
+	mock.ExpectQuery(regexp.QuoteMeta(`SELECT count(*) FROM "bus_bus_1_stable" WHERE global_uuid = $1 AND attribute_group = $2`)).
+		WithArgs(componentUUID, "stable").
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(int64(1)))
+
+	parameter, err := QueryParameterByDataObjectToken(context.Background(), db, token)
+	require.NoError(t, err)
+	assert.Equal(t, "component", parameter.Component.Tag)
+	assert.Equal(t, "bus_bus_1_stable", parameter.Project.Name)
+	assert.Equal(t, "bus_bus_1_stable", parameter.TableName)
+	assert.Equal(t, "stable", parameter.AttributeGroup)
+	assert.Equal(t, "rated_voltage", parameter.AttributeName)
+	assert.Equal(t, "DOUBLE PRECISION", parameter.AttributeType)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestQueryParameterDataObjectValue(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{})
+	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},
+		Project: orm.ProjectManager{
+			Name: "bus_bus_1_stable",
+		},
+		TableName:      "bus_bus_1_stable",
+		AttributeGroup: "stable",
+		AttributeName:  "rated_voltage",
+	}
+
+	mock.ExpectQuery(regexp.QuoteMeta(`SELECT * FROM "bus_bus_1_stable" WHERE global_uuid = $1 AND attribute_group = $2 LIMIT $3`)).
+		WithArgs(componentUUID, "stable", 1).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"global_uuid", "attribute_group", "rated_voltage",
+		}).AddRow(componentUUID, "stable", float64(220)))
+
+	value, err := QueryParameterDataObjectValue(context.Background(), db, parameter)
+	require.NoError(t, err)
+	assert.Equal(t, float64(220), value)
+	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)
+	t.Cleanup(func() { _ = sqlDB.Close() })
+
+	db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{})
+	require.NoError(t, err)
+
+	const (
+		token         = "grid.zone.station.nspath.component.component.description"
+		componentUUID = "70c190f2-8a60-42a9-b143-ec5f87e0aa6b"
+	)
+	mock.ExpectQuery(`(?s)SELECT c\.\*.*WHERE g\.tagname = \$1.*AND c\.tag = \$5.*LIMIT 2`).
+		WithArgs("grid", "zone", "station", "nspath", "component").
+		WillReturnRows(sqlmock.NewRows([]string{
+			"global_uuid", "nspath", "tag", "model_name", "grid", "zone", "station", "station_id",
+		}).AddRow(componentUUID, "nspath", "component", "bus_1", "grid", "zone", "station", int64(10)))
+	mock.ExpectQuery(`(?s)SELECT pg_catalog\.format_type.*c\.relname = \$1.*a\.attname = \$2.*LIMIT 1`).
+		WithArgs("component", "description").
+		WillReturnRows(sqlmock.NewRows([]string{"format_type"}).AddRow("character varying(512)"))
+
+	parameter, err := QueryParameterByDataObjectToken(context.Background(), db, token)
+	require.NoError(t, err)
+	assert.Equal(t, "component", parameter.TableName)
+	assert.Equal(t, "component", parameter.AttributeGroup)
+	assert.Equal(t, "description", parameter.AttributeName)
+	assert.Equal(t, "CHARACTER VARYING(512)", parameter.AttributeType)
+	assert.Empty(t, parameter.Project.Name)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestQueryComponentParameterValue(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{})
+	require.NoError(t, err)
+
+	parameter := &ParameterDataObject{
+		Component:      orm.Component{Tag: "component"},
+		TableName:      "component",
+		AttributeGroup: "component",
+		AttributeName:  "description",
+	}
+	mock.ExpectQuery(regexp.QuoteMeta(`SELECT * FROM "component" WHERE tag = $1 LIMIT $2`)).
+		WithArgs("component", 1).
+		WillReturnRows(sqlmock.NewRows([]string{"tag", "description"}).AddRow("component", "测试组件"))
+
+	value, err := QueryParameterDataObjectValue(context.Background(), db, parameter)
+	require.NoError(t, err)
+	assert.Equal(t, "测试组件", value)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestQueryParameterAttributeDescription(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{})
+	require.NoError(t, err)
+
+	mock.ExpectQuery(`(?s)SELECT attribute_name.*FROM basic\.attribute.*WHERE attribute = \$1.*LIMIT 2`).
+		WithArgs("rated_voltage").
+		WillReturnRows(sqlmock.NewRows([]string{"attribute_name"}).AddRow("额定电压"))
+
+	description, err := QueryParameterAttributeDescription(context.Background(), db, "rated_voltage")
+	require.NoError(t, err)
+	assert.Equal(t, "额定电压", description)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestQueryParameterByDataObjectTokenComponentNotFound(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{})
+	require.NoError(t, err)
+
+	mock.ExpectQuery(`(?s)SELECT c\.\*.*WHERE c\.nspath = \$1.*AND c\.tag = \$2.*s\.is_local = TRUE.*LIMIT 2`).
+		WithArgs("nspath", "component").
+		WillReturnRows(sqlmock.NewRows([]string{"global_uuid"}))
+
+	_, err = QueryParameterByDataObjectToken(
+		context.Background(),
+		db,
+		"nspath.component.stable.rated_voltage",
+	)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, common.ErrParameterTokenNotFound)
+	require.NoError(t, mock.ExpectationsWereMet())
+}

database/query_measurement.go

@@ -1,62 +0,0 @@
-// Package database define database operation functions
-package database
-
-import (
-	"context"
-	"time"
-
-	"modelRT/orm"
-
-	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
-)
-
-// 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
-	// ctx超时判断
-	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-	result := tx.WithContext(cancelCtx).
-		Where("id = ?", id).
-		Clauses(clause.Locking{Strength: "UPDATE"}).
-		First(&measurement)
-
-	if result.Error != nil {
-		return orm.Measurement{}, result.Error
-	}
-	return measurement, nil
-}
-
-// QueryMeasurementByToken define function query circuit diagram component measurement info by token from postgresDB
-func QueryMeasurementByToken(ctx context.Context, tx *gorm.DB, token string) (orm.Measurement, error) {
-	// TODO parse token to avoid SQL injection
-
-	var component orm.Measurement
-	// ctx超时判断
-	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-	result := tx.WithContext(cancelCtx).
-		Where(" = ?", token).
-		Clauses(clause.Locking{Strength: "UPDATE"}).
-		First(&component)
-
-	if result.Error != nil {
-		return orm.Measurement{}, result.Error
-	}
-	return component, nil
-}
-
-// GetAllMeasurements define func to query all measurement info from postgresDB
-func GetAllMeasurements(ctx context.Context, tx *gorm.DB) ([]orm.Measurement, error) {
-	var measurements []orm.Measurement
-
-	// ctx超时判断
-	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-	result := tx.WithContext(cancelCtx).Clauses(clause.Locking{Strength: "UPDATE"}).Find(&measurements)
-	if result.Error != nil {
-		return nil, result.Error
-	}
-	return measurements, nil
-}

database/query_topologic.go

@@ -3,11 +3,8 @@ package database
 
 import (
 	"context"
-	"fmt"
 	"time"
 
-	"modelRT/constants"
-	"modelRT/diagram"
 	"modelRT/logger"
 	"modelRT/orm"
 	"modelRT/sql"
@@ -24,16 +21,19 @@ func QueryTopologic(ctx context.Context, tx *gorm.DB) ([]orm.Topologic, error) {
 	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	result := tx.WithContext(cancelCtx).Clauses(clause.Locking{Strength: "UPDATE"}).Raw(sql.RecursiveSQL, constants.UUIDNilStr).Scan(&topologics)
+	result := tx.WithContext(cancelCtx).
+		Clauses(clause.Locking{Strength: "UPDATE"}).
+		Find(&topologics)
 	if result.Error != nil {
-		logger.Error(ctx, "query circuit diagram topologic info by start node uuid failed", "start_node_uuid", constants.UUIDNilStr, "error", result.Error)
+		logger.Error(ctx, "query circuit diagram topologic info failed", "error", result.Error)
 		return nil, result.Error
 	}
 	return topologics, nil
 }
 
-// QueryTopologicByStartUUID returns all edges reachable from startUUID following
-// directed uuid_from → uuid_to edges in the topologic table.
+// QueryTopologicByStartUUID returns all directed edges reachable from startUUID.
+// It is used by point-to-point topology reachability checks and intentionally
+// does not depend on the legacy all-zero UUID virtual root.
 func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.UUID) ([]orm.Topologic, error) {
 	var topologics []orm.Topologic
 
@@ -42,7 +42,7 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.
 
 	result := tx.WithContext(cancelCtx).
 		Clauses(clause.Locking{Strength: "UPDATE"}).
-		Raw(sql.RecursiveSQL, startUUID).
+		Raw(sql.RecursiveTopologicByStartSQL, startUUID).
 		Scan(&topologics)
 	if result.Error != nil {
 		logger.Error(ctx, "query topologic by start uuid failed", "start_uuid", startUUID, "error", result.Error)
@@ -50,80 +50,3 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.
 	}
 	return topologics, nil
 }
-
-// QueryTopologicFromDB return the result of query topologic info from DB.
-// Returns the root node and a flat nodeMap for O(1) lookup by UUID.
-func QueryTopologicFromDB(ctx context.Context, tx *gorm.DB) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) {
-	topologicInfos, err := QueryTopologic(ctx, tx)
-	if err != nil {
-		logger.Error(ctx, "query topologic info failed", "error", err)
-		return nil, nil, err
-	}
-
-	tree, nodeMap, err := BuildMultiBranchTree(topologicInfos)
-	if err != nil {
-		logger.Error(ctx, "init topologic failed", "error", err)
-		return nil, nil, err
-	}
-	return tree, nodeMap, nil
-}
-
-// BuildMultiBranchTree return the multi branch tree by topologic info.
-// Returns the root node and a flat nodeMap for O(1) lookup by UUID.
-func BuildMultiBranchTree(topologics []orm.Topologic) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) {
-	nodeMap := make(map[uuid.UUID]*diagram.MultiBranchTreeNode, len(topologics)*2)
-
-	for _, topo := range topologics {
-		if _, exists := nodeMap[topo.UUIDFrom]; !exists {
-			// UUIDNil is the virtual root sentinel — skip creating a regular node for it
-			if topo.UUIDFrom != constants.UUIDNil {
-				nodeMap[topo.UUIDFrom] = &diagram.MultiBranchTreeNode{
-					ID:       topo.UUIDFrom,
-					Children: make([]*diagram.MultiBranchTreeNode, 0),
-				}
-			}
-		}
-
-		if _, exists := nodeMap[topo.UUIDTo]; !exists {
-			if topo.UUIDTo != constants.UUIDNil {
-				nodeMap[topo.UUIDTo] = &diagram.MultiBranchTreeNode{
-					ID:       topo.UUIDTo,
-					Children: make([]*diagram.MultiBranchTreeNode, 0),
-				}
-			}
-		}
-	}
-
-	for _, topo := range topologics {
-		var parent *diagram.MultiBranchTreeNode
-		if topo.UUIDFrom == constants.UUIDNil {
-			if _, exists := nodeMap[constants.UUIDNil]; !exists {
-				nodeMap[constants.UUIDNil] = &diagram.MultiBranchTreeNode{
-					ID:       constants.UUIDNil,
-					Children: make([]*diagram.MultiBranchTreeNode, 0),
-				}
-			}
-			parent = nodeMap[constants.UUIDNil]
-		} else {
-			parent = nodeMap[topo.UUIDFrom]
-		}
-
-		var child *diagram.MultiBranchTreeNode
-		if topo.UUIDTo == constants.UUIDNil {
-			child = &diagram.MultiBranchTreeNode{
-				ID: topo.UUIDTo,
-			}
-		} else {
-			child = nodeMap[topo.UUIDTo]
-		}
-		child.Parent = parent
-		parent.Children = append(parent.Children, child)
-	}
-
-	// return root vertex
-	root, exists := nodeMap[constants.UUIDNil]
-	if !exists {
-		return nil, nil, fmt.Errorf("root node not found")
-	}
-	return root, nodeMap, nil
-}

deploy/deploy.md

@@ -640,6 +640,12 @@ openssl x509 -in eventrt_client_cert.pem -noout -subject
 
 将服务器端三个证书文件打包为 K8s Secret（在证书文件所在目录执行）：
 
+```bash
+sh deploy/k8s/rabbitmq-certs-secret.sh
+```
+
+该脚本等价于：
+
 ```bash
 kubectl create secret generic rabbitmq-certs \
   --from-file=ca_certificate.pem=./ca_certificate.pem \
@@ -695,7 +701,11 @@ kubectl apply -f deploy/k8s/pg-service.yaml
 | **数据库** | `demo` | ConfigMap 中 `POSTGRES_DB` |
 | **用户名** | `postgres` | ConfigMap 中 `POSTGRES_USER` |
 | **密码** | `coslight` | ConfigMap `postgres-config` 中配置，生产环境迁移至 Secret |
-| **存储** | `2Gi` | PVC `postgres-data` |
+| **存储** | `6Gi` | PVC `postgres-data` |
+| **CPU** | `100m` 请求 / `500m` 上限 | StatefulSet `resources` 字段 |
+| **内存** | `256Mi` 请求 / `512Mi` 上限 | StatefulSet `resources` 字段 |
+
+> **注意：** 密码当前以明文形式存储在 `pg-configmap.yaml` 中，生产环境应将其迁移至 K8s Secret，并通过环境变量注入容器，避免将明文密码提交至版本库。
 
 ##### 4.4.1 等待 Pod 就绪
 
@@ -703,7 +713,23 @@ kubectl apply -f deploy/k8s/pg-service.yaml
 kubectl wait --for=condition=ready pod -l app=postgres --timeout=120s
 ```
 
-##### 4.4.2 初始化异步任务表
+##### 4.4.2 连接验证
+
+```bash
+# 快速检查 PostgreSQL 是否接受连接
+kubectl exec -it $(kubectl get pod -l app=postgres -o jsonpath='{.items[0].metadata.name}') \
+  -- pg_isready -U postgres -d demo
+
+# 进入 psql 执行简单查询确认数据库可用
+kubectl exec -it $(kubectl get pod -l app=postgres -o jsonpath='{.items[0].metadata.name}') \
+  -- psql -U postgres -d demo -c "SELECT current_database(), version();"
+
+# 列出所有数据库（确认 demo 库已创建）
+kubectl exec -it $(kubectl get pod -l app=postgres -o jsonpath='{.items[0].metadata.name}') \
+  -- psql -U postgres -c "\l"
+```
+
+##### 4.4.3 初始化异步任务表
 
 PostgreSQL 就绪后执行 1.4 节的建表 SQL，可通过以下方式进入容器执行：
 
@@ -717,14 +743,14 @@ kubectl exec -i $(kubectl get pod -l app=postgres -o jsonpath='{.items[0].metada
   -- psql -U postgres -d demo < /path/to/init.sql
 ```
 
-##### 4.4.3 状态检查
+##### 4.4.4 状态检查
 
 ```bash
 kubectl get pods -l app=postgres
 kubectl logs -l app=postgres --tail=30
 ```
 
-##### 4.4.4 清理
+##### 4.4.5 清理
 
 ```bash
 kubectl delete -f deploy/k8s/pg-service.yaml \
@@ -733,68 +759,73 @@ kubectl delete -f deploy/k8s/pg-service.yaml \
                -f deploy/k8s/pg-configmap.yaml
 ```
 
-#### 4.5 部署 MongoDB
-
-```bash
-kubectl apply -f deploy/k8s/mongodb-secret.yaml
-kubectl apply -f deploy/k8s/mongodb-pvc.yaml
-kubectl apply -f deploy/k8s/mongodb-statefulset.yaml
-kubectl apply -f deploy/k8s/mongodb-service.yaml
-```
-
-| 参数 | 值 | 说明 |
-| :--- | :--- | :--- |
-| **镜像** | `mongo:7.0` | MongoDB 7.0 |
-| **NodePort** | `30017` | 集群外访问端口 |
-| **用户名** | `admin` | Root 管理员 |
-| **密码** | `coslight` | Secret `mongodb-secret` 中配置，生产环境请替换强密码 |
-| **存储** | `2Gi` | PVC `mongodb-data` |
-
-> **注意：** 密码存储在 `mongodb-secret.yaml` 的 `stringData` 中，生产环境应替换为强密码，并避免将明文密码提交至版本库。
-
-##### 4.5.1 等待 Pod 就绪
-
-```bash
-kubectl wait --for=condition=ready pod -l app=mongodb --timeout=120s
-```
-
-##### 4.5.2 连接验证
-
-```bash
-kubectl exec -it $(kubectl get pod -l app=mongodb -o jsonpath='{.items[0].metadata.name}') \
-  -- mongosh -u admin -p coslight --authenticationDatabase admin
-```
-
-##### 4.5.3 状态检查
-
-```bash
-kubectl get pods -l app=mongodb
-kubectl logs -l app=mongodb --tail=30
-```
-
-##### 4.5.4 清理
-
-```bash
-kubectl delete -f deploy/k8s/mongodb-service.yaml \
-               -f deploy/k8s/mongodb-statefulset.yaml \
-               -f deploy/k8s/mongodb-pvc.yaml \
-               -f deploy/k8s/mongodb-secret.yaml
-```
-
 ### 5\. 部署 ModelRT（Kubernetes）
 
 所有资源部署在 `default` 命名空间，YAML 文件位于 `deploy/k8s/`。
 
 #### 5.1 构建并推送镜像
 
+镜像采用三阶段构建，最终基于 `scratch`：
+
+| 阶段 | 基础镜像 | 作用 |
+| :--- | :--- | :--- |
+| **builder** | `golang:1.26-alpine` | 编译 Go 二进制（`CGO_ENABLED=0`，`-trimpath -ldflags="-s -w"`） |
+| **certs** | `alpine:3.21` | 提取 CA 证书、时区数据及非 root 用户定义（UID 默认 `1000`） |
+| **runtime** | `scratch` | 仅含可执行文件与运行时依赖，无 shell、无包管理器 |
+
+**方式一：从源码构建并加载**
+
 ```bash
-# 在项目根目录执行
+# 在项目根目录执行（默认运行用户 UID=1000）
 docker build -f deploy/dockerfile/modelrt.Dockerfile -t coslight/modelrt:latest .
 
-# 推送到镜像仓库（或直接加载到 Minikube）
+# 自定义运行用户 UID
+docker build -f deploy/dockerfile/modelrt.Dockerfile \
+  --build-arg USER_ID=2000 \
+  -t coslight/modelrt:latest .
+
+# 加载到 Minikube（无需私有仓库）
 minikube image load coslight/modelrt:latest
 ```
 
+**方式二：直接加载已有本地镜像**
+
+Ubuntu 宿主机上已存在构建好的镜像（如 `modelrt:v1`）时，无需重新构建，直接导入 Minikube：
+
+```bash
+# 确认本地镜像存在
+docker images modelrt:v1
+
+# 加载到 Minikube
+minikube image load modelrt:v1
+
+# 验证镜像已进入 Minikube 缓存
+minikube image ls | grep modelrt
+```
+
+> **注意：** `deploy/k8s/modelrt-deployment.yaml` 中的 `image` 字段需与加载的镜像名称一致，并将 `imagePullPolicy` 设为 `Never`，防止 Minikube 尝试从远端拉取。
+
+#### 5.1.1 镜像冒烟测试
+
+```bash
+# 查看镜像大小（scratch 镜像预期 ≤ 25 MB）
+docker images coslight/modelrt:latest
+
+# 检查镜像元信息（确认 User、Cmd、架构）
+docker inspect coslight/modelrt:latest
+
+# 验证二进制可执行（无 config 时程序报错退出属预期行为，说明镜像构建正常）
+docker run --rm coslight/modelrt:latest
+
+# 挂载示例配置做完整启动验证（Ctrl+C 退出）
+docker run --rm \
+  -v "$(pwd)/configs/config.example.yaml:/app/configs/config.yaml" \
+  -p 8080:8080 \
+  coslight/modelrt:latest
+```
+
+> **注意：** `scratch` 镜像不含 shell，无法使用 `docker exec` 进入容器调试；如需排查问题，可临时将最终阶段改为 `alpine` 进行本地调试，确认后再切回 `scratch`。
+
 #### 5.2 创建客户端证书 Secret
 
 在 RabbitMQ TLS 证书生成完成后（见 4.2），进入证书文件所在目录执行：
@@ -864,7 +895,9 @@ kubectl delete secret modelrt-certs
 
 ### 6\. 部署可观测性栈（Kubernetes）
 
-在 `Kubernetes` 集群中部署 `Jaeger`（链路追踪）+ `Loki + Promtail + Grafana`（日志可视化）。所有资源部署在 `default` 命名空间，`YAML` 文件位于 `deploy/k8s/`。
+在 `Kubernetes` 集群中部署 `Jaeger`（链路追踪）+ `Loki + Alloy + Grafana`（日志可视化）。所有资源部署在 `default` 命名空间，`YAML` 文件位于 `deploy/k8s/`。
+
+> **日志采集器说明：** 集群内的日志采集由 `Grafana Alloy`（DaemonSet）负责，它通过 Kubernetes API 抓取带 `app` label 的 Pod 容器日志，解析 `zap` 输出的 JSON 字段后推送到 `Loki`。Alloy 已**替代**早期的 `Promtail`，两者推送目标（`loki-service:3100`）与标签解析完全一致，**不要同时部署**，否则会导致 Loki 中日志翻倍。
 
 #### 6.1 部署 Jaeger
 
@@ -882,14 +915,16 @@ kubectl apply -f deploy/k8s/loki-deployment.yaml
 kubectl apply -f deploy/k8s/loki-service.yaml
 ```
 
-#### 6.3 部署 Promtail
+#### 6.3 部署 Alloy
 
 ```bash
-kubectl apply -f deploy/k8s/promtail-rbac.yaml
-kubectl apply -f deploy/k8s/promtail-configmap.yaml
-kubectl apply -f deploy/k8s/promtail-daemonset.yaml
+kubectl apply -f deploy/k8s/alloy-rbac.yaml
+kubectl apply -f deploy/k8s/alloy-configmap.yaml
+kubectl apply -f deploy/k8s/alloy-daemonset.yaml
 ```
 
+> Alloy 以 DaemonSet 形式在每个节点运行，需要 `ServiceAccount` + `ClusterRole`（`alloy-rbac.yaml`）授予读取 `nodes/pods/pods/log` 的权限。采集与解析规则定义在 `alloy-configmap.yaml` 的 `config.alloy` 中。
+
 #### 6.4 部署 Grafana
 
 ```bash
@@ -907,9 +942,9 @@ kubectl apply -f deploy/k8s/jaeger-deployment.yaml \
               -f deploy/k8s/loki-pvc.yaml \
               -f deploy/k8s/loki-deployment.yaml \
               -f deploy/k8s/loki-service.yaml \
-              -f deploy/k8s/promtail-rbac.yaml \
-              -f deploy/k8s/promtail-configmap.yaml \
-              -f deploy/k8s/promtail-daemonset.yaml \
+              -f deploy/k8s/alloy-rbac.yaml \
+              -f deploy/k8s/alloy-configmap.yaml \
+              -f deploy/k8s/alloy-daemonset.yaml \
               -f deploy/k8s/grafana-configmap.yaml \
               -f deploy/k8s/grafana-deployment.yaml \
               -f deploy/k8s/grafana-service.yaml
@@ -955,7 +990,6 @@ Mac 本地端口  ──SSH隧道──▶  Ubuntu 宿主机 (192.168.1.101)
 
 ```bash
 ssh -L 5432:192.168.49.2:30432 \
-    -L 27017:192.168.49.2:30017 \
     -L 5671:192.168.49.2:30671 \
     -L 15671:192.168.49.2:31671 \
     -L 6379:192.168.49.2:30001 \
@@ -971,7 +1005,6 @@ ssh -L 5432:192.168.49.2:30432 \
 ```bash
 ssh -fN \
     -L 5432:192.168.49.2:30432 \
-    -L 27017:192.168.49.2:30017 \
     -L 5671:192.168.49.2:30671 \
     -L 15671:192.168.49.2:31671 \
     -L 6379:192.168.49.2:30001 \
@@ -987,7 +1020,6 @@ ssh -fN \
 | Mac 本地端口 | Minikube NodePort | 服务 | 说明 |
 | :--- | :--- | :--- | :--- |
 | `5432` | `30432` | PostgreSQL | 数据库连接 `localhost:5432` |
-| `27017` | `30017` | MongoDB | 数据库连接 `localhost:27017` |
 | `5671` | `30671` | RabbitMQ AMQP | ModelRT / EventRT 消息队列连接 |
 | `15671` | `31671` | RabbitMQ Management | RabbitMQ 管理界面 `http://localhost:15671` |
 | `6379` | `30001` | Redis | 分布式锁 / 数据存储 |
@@ -1011,14 +1043,111 @@ kill <PID>
 
 ### 8\. 后续操作（停止与清理）
 
-#### 8.1 停止容器
+#### 8.1 本地 Docker 部署清理
+
+适用于第 1、2 节使用 `docker run` 启动的 PostgreSQL 和 Redis 容器。
 
 ```bash
+# 停止容器
 docker stop postgres redis
-```
 
-#### 8.2 删除容器（删除后数据将丢失）
-
-```bash
+# 删除容器（容器内数据将同步丢失）
 docker rm postgres redis
 ```
+
+#### 8.2 本地运行清理
+
+适用于第 3 节以 `go run` 或编译后二进制方式在本地启动的 ModelRT 服务。
+
+前台运行时直接 `Ctrl+C` 终止；后台运行时查找并终止进程：
+
+```bash
+# 终止 go run 启动的进程
+pkill -f "go run main.go"
+
+# 或终止编译后的二进制进程
+pkill model-rt
+```
+
+#### 8.3 K8s(Minikube) 部署清理
+
+适用于第 4、5、6 节在 Minikube 中部署的所有资源。
+
+##### 8.3.1 分服务清理
+
+**仅停止（缩容至 0，PVC 数据保留）**
+
+将所有 Deployment 和 StatefulSet 缩容至 0 副本，Pod 停止运行但持久卷数据不删除，之后可直接缩容回 1 恢复服务。
+
+```bash
+# 停止所有 Deployment（Redis / RabbitMQ / ModelRT / Jaeger / Loki / Grafana）
+kubectl scale deployment --all --replicas=0
+
+# 停止所有 StatefulSet（PostgreSQL，PVC 数据保留）
+kubectl scale statefulset --all --replicas=0
+```
+
+恢复时：
+
+```bash
+kubectl scale deployment --all --replicas=1
+kubectl scale statefulset --all --replicas=1
+```
+
+> **注意：** DaemonSet（Alloy）无法通过 `scale` 停止，如需停用可手动删除其资源：`kubectl delete -f deploy/k8s/alloy-daemonset.yaml`。
+
+---
+
+**永久清理（删除所有资源，包含 PVC，数据不可恢复）**
+
+按部署顺序反向删除各服务资源：
+
+```bash
+# 可观测性栈（Grafana / Alloy / Loki / Jaeger）
+kubectl delete -f deploy/k8s/grafana-service.yaml \
+               -f deploy/k8s/grafana-deployment.yaml \
+               -f deploy/k8s/grafana-configmap.yaml \
+               -f deploy/k8s/alloy-daemonset.yaml \
+               -f deploy/k8s/alloy-configmap.yaml \
+               -f deploy/k8s/alloy-rbac.yaml \
+               -f deploy/k8s/loki-service.yaml \
+               -f deploy/k8s/loki-deployment.yaml \
+               -f deploy/k8s/loki-pvc.yaml \
+               -f deploy/k8s/loki-configmap.yaml \
+               -f deploy/k8s/jaeger-service.yaml \
+               -f deploy/k8s/jaeger-deployment.yaml
+
+# ModelRT 应用
+kubectl delete -f deploy/k8s/modelrt-service.yaml \
+               -f deploy/k8s/modelrt-deployment.yaml \
+               -f deploy/k8s/modelrt-configmap.yaml \
+               -f deploy/k8s/modelrt-secret.yaml
+kubectl delete secret modelrt-certs
+
+# PostgreSQL
+kubectl delete -f deploy/k8s/pg-service.yaml \
+               -f deploy/k8s/pg-statefulset.yaml \
+               -f deploy/k8s/pg-pvc.yaml \
+               -f deploy/k8s/pg-configmap.yaml
+
+# RabbitMQ
+kubectl delete -f deploy/k8s/rabbitmq-service.yaml \
+               -f deploy/k8s/rabbitmq-deployment.yaml \
+               -f deploy/k8s/rabbitmq-users-config.yaml \
+               -f deploy/k8s/rabbitmq-config.yaml \
+               -f deploy/k8s/rabbitmq-secret.yaml
+kubectl delete secret rabbitmq-certs
+
+# Redis
+kubectl delete -f deploy/k8s/redis-service.yaml \
+               -f deploy/k8s/redis-deployment.yaml
+```
+
+##### 8.3.2 一键清理
+
+> **注意：** 此操作会删除 `deploy/k8s/` 下所有 YAML 对应的 K8s 资源，包括 PVC，**持久化数据将永久丢失**，请确认后执行。
+
+```bash
+kubectl delete -f deploy/k8s/
+kubectl delete secret rabbitmq-certs modelrt-certs
+```

deploy/dockerfile/modelrt.Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.25-alpine AS builder
+FROM golang:1.26-alpine AS builder
 RUN apk --no-cache upgrade
 
 WORKDIR /app
@@ -11,8 +11,8 @@ RUN CGO_ENABLED=0 GOOS=linux go build \
     -mod=readonly \
     -o modelrt main.go
 
-# Prepare runtime dependencies in a pinned Alpine stage so they can be
-# copied into scratch without pulling any vulnerable OS packages at run time.
+# prepare runtime dependencies in a pinned alpine stage so they can be
+# copied into scratch without pulling any vulnerable os packages at run time.
 FROM alpine:3.21 AS certs
 ARG USER_ID=1000
 RUN apk --no-cache add ca-certificates tzdata && \
@@ -21,15 +21,14 @@ RUN apk --no-cache add ca-certificates tzdata && \
 FROM scratch
 # CA certificates required for TLS connections (RabbitMQ amqps://)
 COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
-# Timezone data
+# timezone data
 COPY --from=certs /usr/share/zoneinfo /usr/share/zoneinfo
-# Non-root user/group definitions
+# non-root user/group definitions
 COPY --from=certs /etc/passwd /etc/passwd
 COPY --from=certs /etc/group  /etc/group
 
 WORKDIR /app
 COPY --from=builder /app/modelrt ./modelrt
-COPY configs/config.example.yaml ./configs/config.example.yaml
 
 USER modelrt
 CMD ["/app/modelrt", "-modelRT_config_dir=/app/configs"]

deploy/k8s/alloy-configmap.yaml

@@ -0,0 +1,81 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: alloy-config
+  namespace: default
+data:
+  config.alloy: |
+    // 发现集群内所有 Pod
+    discovery.kubernetes "pods" {
+      role = "pod"
+    }
+
+    // 重写元数据标签,并只保留带 app label 的 Pod
+    discovery.relabel "pods" {
+      targets = discovery.kubernetes.pods.targets
+
+      rule {
+        source_labels = ["__meta_kubernetes_namespace"]
+        target_label  = "namespace"
+      }
+      rule {
+        source_labels = ["__meta_kubernetes_pod_name"]
+        target_label  = "pod"
+      }
+      rule {
+        source_labels = ["__meta_kubernetes_pod_container_name"]
+        target_label  = "container"
+      }
+      rule {
+        source_labels = ["__meta_kubernetes_pod_label_app"]
+        target_label  = "app"
+      }
+      // 只采集有 app label 的 Pod
+      rule {
+        source_labels = ["__meta_kubernetes_pod_label_app"]
+        action        = "keep"
+        regex         = ".+"
+      }
+    }
+
+    // 通过 Kubernetes API 抓取容器日志(无需挂载宿主机日志目录)
+    loki.source.kubernetes "pods" {
+      targets    = discovery.relabel.pods.output
+      forward_to = [loki.process.parse.receiver]
+    }
+
+    // 解析 zap 输出的 JSON 日志,并将关键字段提升为 Loki Label
+    loki.process "parse" {
+      forward_to = [loki.write.default.receiver]
+
+      // 解析结构化字段
+      stage.json {
+        expressions = {
+          level     = "level",
+          traceID   = "traceID",
+          spanID    = "spanID",
+          caller    = "caller",
+          pod       = "pod",
+          namespace = "namespace",
+          node      = "node",
+        }
+      }
+
+      // 提升为 Label,支持在 Grafana 中按实例/Trace 过滤
+      stage.labels {
+        values = {
+          level     = "",
+          traceID   = "",
+          pod       = "",
+          namespace = "",
+          node      = "",
+        }
+      }
+    }
+
+    // 推送到 Loki
+    loki.write "default" {
+      endpoint {
+        url = "http://loki-service:3100/loki/api/v1/push"
+      }
+    }

deploy/k8s/alloy-daemonset.yaml

@@ -0,0 +1,48 @@
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: alloy
+  namespace: default
+spec:
+  selector:
+    matchLabels:
+      app: alloy
+  template:
+    metadata:
+      labels:
+        app: alloy
+    spec:
+      serviceAccountName: alloy
+      tolerations:
+        - key: node-role.kubernetes.io/master
+          effect: NoSchedule
+      containers:
+        - name: alloy
+          image: grafana/alloy:v1.16.3
+          imagePullPolicy: IfNotPresent
+          args:
+            - run
+            - /etc/alloy/config.alloy
+            - --storage.path=/var/lib/alloy/data
+            - --server.http.listen-addr=0.0.0.0:12345
+          ports:
+            - containerPort: 12345
+              name: http
+          volumeMounts:
+            - name: config
+              mountPath: /etc/alloy
+            - name: data
+              mountPath: /var/lib/alloy/data
+          resources:
+            limits:
+              cpu: 200m
+              memory: 128Mi
+            requests:
+              cpu: 50m
+              memory: 64Mi
+      volumes:
+        - name: config
+          configMap:
+            name: alloy-config
+        - name: data
+          emptyDir: {}

deploy/k8s/alloy-rbac.yaml

@@ -0,0 +1,30 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: alloy
+  namespace: default
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: alloy
+rules:
+  - apiGroups: [""]
+    resources: ["nodes", "nodes/proxy", "services", "endpoints", "pods"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["pods/log"]
+    verbs: ["get", "list", "watch"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: alloy
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: alloy
+subjects:
+  - kind: ServiceAccount
+    name: alloy
+    namespace: default

deploy/k8s/grafana-configmap.yaml

@@ -10,7 +10,7 @@ data:
       - name: Loki
         type: loki
         access: proxy
-        url: http://loki:3100
+        url: http://loki-service:3100
         isDefault: true
         jsonData:
           # derivedFields: 从日志的 traceID 字段生成跳转链接到 Jaeger
@@ -23,4 +23,4 @@ data:
         type: jaeger
         uid: jaeger
         access: proxy
-        url: http://jaeger:16686
+        url: http://jaeger-service:16686

deploy/k8s/grafana-deployment.yaml

@@ -16,6 +16,7 @@ spec:
       containers:
         - name: grafana
           image: grafana/grafana:10.4.2
+          imagePullPolicy: IfNotPresent
           ports:
             - containerPort: 3000
           env:

deploy/k8s/grafana-service.yaml

@@ -1,14 +1,14 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: grafana
+  name: grafana-service
   namespace: default
 spec:
   ports:
     - name: http
       port: 3000
       targetPort: 3000
-      nodePort: 31000    # Grafana UI: http://<NodeIP>:31000
+      nodePort: 31000 # Grafana UI: http://<NodeIP>:31000
   selector:
     app: grafana
   type: NodePort

deploy/k8s/jaeger-deployment.yaml

@@ -15,6 +15,7 @@ spec:
       containers:
         - name: jaeger
           image: jaegertracing/all-in-one:1.56
+          imagePullPolicy: IfNotPresent
           env:
             - name: COLLECTOR_OTLP_ENABLED
               value: "true"

deploy/k8s/jaeger-service.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: jaeger
+  name: jaeger-service
   labels:
     app: jaeger
 spec:
@@ -9,19 +9,19 @@ spec:
     - name: ui
       port: 16686
       targetPort: 16686
-      nodePort: 31686   # Jaeger UI，浏览器访问 http://<NodeIP>:31686
+      nodePort: 31686 # Jaeger UI，浏览器访问 http://<NodeIP>:31686
     - name: collector-http
       port: 14268
       targetPort: 14268
-      nodePort: 31268   # Jaeger 原生 HTTP collector（非 OTel）
+      nodePort: 31268 # Jaeger 原生 HTTP collector（非 OTel）
     - name: otlp-http
       port: 4318
       targetPort: 4318
-      nodePort: 31318   # OTLP HTTP，集群外使用 <NodeIP>:31318
+      nodePort: 31318 # OTLP HTTP，集群外使用 <NodeIP>:31318
     - name: otlp-grpc
       port: 4317
       targetPort: 4317
-      nodePort: 31317   # OTLP gRPC，集群外使用 <NodeIP>:31317
+      nodePort: 31317 # OTLP gRPC，集群外使用 <NodeIP>:31317
   selector:
     app: jaeger
   type: NodePort

deploy/k8s/loki-deployment.yaml

@@ -20,6 +20,7 @@ spec:
       containers:
         - name: loki
           image: grafana/loki:2.9.4
+          imagePullPolicy: IfNotPresent
           args:
             - -config.file=/etc/loki/loki.yaml
           ports:

deploy/k8s/loki-service.yaml

@@ -1,14 +1,14 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: loki
+  name: loki-service
   namespace: default
 spec:
   ports:
     - name: http
       port: 3100
       targetPort: 3100
-      nodePort: 31100    # 集群外访问: http://<NodeIP>:31100
+      nodePort: 31100 # 集群外访问: http://<NodeIP>:31100
   selector:
     app: loki
   type: NodePort

deploy/k8s/modelrt-configmap.yaml

@@ -5,7 +5,7 @@ metadata:
 data:
   config.yaml: |
     postgres:
-      host: "192.168.1.101"
+      host: "postgres-service"
       port: 5432
       database: "demo"
       user: "postgres"
@@ -35,7 +35,7 @@ data:
         endpoint: ""        # Promtail handles log collection in K8s, direct push disabled
 
     otel:
-      endpoint: "jaeger:4318"
+      endpoint: "jaeger-service:4318"
       insecure: true
 
     ants:
@@ -77,7 +77,7 @@ data:
       service_addr: ":8080"
       service_name: "modelRT"
       secret_key: ""        # injected via env SERVICE_SECRET_KEY
-      deploy_env: "production"
+      deploy_env: "development"
 
     dataRT:
       host: "http://127.0.0.1"

deploy/k8s/modelrt-deployment.yaml

@@ -16,8 +16,9 @@ spec:
     spec:
       containers:
         - name: modelrt
-          image: coslight/modelrt:latest
+          image: modelrt:v1
           imagePullPolicy: IfNotPresent
+          command: ["/app/modelrt"]
           args:
             - "-modelRT_config_dir=/app/configs"
             - "-modelRT_config_name=config"

deploy/k8s/mongodb-service.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: mongodb
+  name: mongodb-service
   labels:
     app: mongodb
 spec:

deploy/k8s/mongodb-statefulset.yaml

@@ -34,9 +34,9 @@ spec:
                 - mongosh
                 - --eval
                 - "db.adminCommand('ping')"
-            initialDelaySeconds: 10
-            periodSeconds: 5
-            timeoutSeconds: 3
+            initialDelaySeconds: 30
+            periodSeconds: 10
+            timeoutSeconds: 10
             failureThreshold: 12
           livenessProbe:
             exec:
@@ -44,10 +44,10 @@ spec:
                 - mongosh
                 - --eval
                 - "db.adminCommand('ping')"
-            initialDelaySeconds: 30
-            periodSeconds: 20
-            timeoutSeconds: 3
-            failureThreshold: 3
+            initialDelaySeconds: 120
+            periodSeconds: 10
+            timeoutSeconds: 30
+            failureThreshold: 5
           resources:
             requests:
               cpu: 100m

deploy/k8s/pg-pvc.yaml

@@ -7,4 +7,4 @@ spec:
     - ReadWriteOnce
   resources:
     requests:
-      storage: 2Gi
+      storage: 6Gi

deploy/k8s/pg-service.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: postgres
+  name: postgres-service
   labels:
     app: postgres
 spec:

deploy/k8s/promtail-configmap.yaml

@@ -13,7 +13,7 @@ data:
       filename: /tmp/positions.yaml
 
     clients:
-      - url: http://loki:3100/loki/api/v1/push
+      - url: http://loki-service:3100/loki/api/v1/push
 
     scrape_configs:
       - job_name: kubernetes-pods

deploy/k8s/promtail-daemonset.yaml

@@ -19,6 +19,7 @@ spec:
       containers:
         - name: promtail
           image: grafana/promtail:2.9.4
+          imagePullPolicy: IfNotPresent
           args:
             - -config.file=/etc/promtail/promtail.yaml
           ports:

deploy/k8s/rabbitmq-certs-secret.sh

@@ -0,0 +1,14 @@
+#!/bin/sh
+# Create the rabbitmq server certificate secret.
+# Run this script from the directory that contains the three cert files,
+# or adjust the paths below to point at the actual files.
+#
+# Expected files (generated during RabbitMQ TLS setup):
+#   ca_certificate.pem
+#   server_certificate.pem
+#   server_key.pem
+
+kubectl create secret generic rabbitmq-certs \
+  --from-file=ca_certificate.pem=./ca_certificate.pem \
+  --from-file=server_certificate.pem=./server_certificate.pem \
+  --from-file=server_key.pem=./server_key.pem

deploy/k8s/rabbitmq-deployment.yaml

@@ -1,7 +1,7 @@
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: eventrt-rabbitmq
+  name: rabbitmq
 spec:
   replicas: 1
   selector:
@@ -15,6 +15,7 @@ spec:
       containers:
         - name: rabbitmq
           image: rabbitmq:4.1.1-management-alpine
+          imagePullPolicy: IfNotPresent
           ports:
             - containerPort: 4369
             - containerPort: 5671

deploy/k8s/rabbitmq-plugins-config.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: rabbit-plugins-conf
+data:
+  enabled_plugins: |
+    [rabbitmq_auth_mechanism_ssl, rabbitmq_management, rabbitmq_management_agent, rabbitmq_prometheus, rabbitmq_web_dispatch].

deploy/k8s/redis-deployment.yaml

@@ -15,6 +15,7 @@ spec:
       containers:
         - name: redis
           image: redis/redis-stack-server:latest
+          imagePullPolicy: IfNotPresent
           resources:
             limits:
               memory: "128Mi"

deploy/redis-test-data/real-time-compute/compute_data_injection.go

@@ -88,7 +88,7 @@ func generateNormalData(baseValue, normalBase float64) []float64 {
 func main() {
 	rootCtx := context.Background()
 
-	pgURI := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s", "192.168.1.101", 5432, "postgres", "coslight", "demo")
+	pgURI := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s", "localhost", 5432, "postgres", "coslight", "develop_env")
 
 	postgresDBClient, err := gorm.Open(postgres.Open(pgURI))
 	if err != nil {
@@ -164,7 +164,6 @@ func main() {
 			}
 
 			datas = generateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase)
-			// log.Printf("key:%s\n datas:%v\n", key, datas)
 
 			allHigh := true
 			for i := highStart; i < highEnd; i++ {

deploy/redis-test-data/util/rand.go

@@ -3,6 +3,7 @@ package util
 
 import (
 	"fmt"
+	"strings"
 
 	"modelRT/orm"
 )
@@ -61,7 +62,7 @@ func ProcessMeasurements(measurements []orm.Measurement) map[string]CalculationR
 		device, _ := ioAddress["device"].(string)
 		channel, _ := ioAddress["channel"].(string)
 
-		result := fmt.Sprintf("%s:%s:phasor:%s", station, device, channel)
+		result := strings.ToLower(fmt.Sprintf("%s:%s:phasor:%s", station, device, channel))
 		if measurement.EventPlan == nil {
 			continue
 		}

diagram/anchor_set.go

@@ -2,24 +2,20 @@
 package diagram
 
 import (
-	"errors"
 	"fmt"
-	"sync"
+
+	"modelRT/util"
 )
 
-// anchorValueOverview define struct of storage all anchor value
-var anchorValueOverview sync.Map
+// anchorValueOverview define struct of storage all anchor value keyed by component uuid
+var anchorValueOverview util.TypedMap[string, string]
 
 // GetAnchorValue define func of get circuit diagram data by componentID
 func GetAnchorValue(componentUUID string) (string, error) {
-	value, ok := diagramsOverview.Load(componentUUID)
+	anchorValue, ok := anchorValueOverview.Load(componentUUID)
 	if !ok {
 		return "", fmt.Errorf("can not find anchor value by componentUUID:%s", componentUUID)
 	}
-	anchorValue, ok := value.(string)
-	if !ok {
-		return "", errors.New("convert to string failed")
-	}
 	return anchorValue, nil
 }
 

diagram/component_set.go

@@ -2,32 +2,27 @@
 package diagram
 
 import (
-	"errors"
 	"fmt"
-	"sync"
 
 	"modelRT/orm"
+	"modelRT/util"
 )
 
-// diagramsOverview define struct of storage all circuit diagram data
-var diagramsOverview sync.Map
+// diagramsOverview define struct of storage all circuit diagram data keyed by component uuid
+var diagramsOverview util.TypedMap[string, *orm.Component]
 
 // GetComponentMap define func of get circuit diagram data by component uuid
 func GetComponentMap(componentUUID string) (*orm.Component, error) {
-	value, ok := diagramsOverview.Load(componentUUID)
+	componentInfo, ok := diagramsOverview.Load(componentUUID)
 	if !ok {
 		return nil, fmt.Errorf("can not find graph by global uuid:%s", componentUUID)
 	}
-	componentInfo, ok := value.(*orm.Component)
-	if !ok {
-		return nil, errors.New("convert to component map struct failed")
-	}
 	return componentInfo, nil
 }
 
 // UpdateComponentMap define func of update circuit diagram data by component uuid and component info
-func UpdateComponentMap(componentID int64, componentInfo *orm.Component) bool {
-	_, result := diagramsOverview.Swap(componentID, componentInfo)
+func UpdateComponentMap(componentUUID string, componentInfo *orm.Component) bool {
+	_, result := diagramsOverview.Swap(componentUUID, componentInfo)
 	return result
 }
 

diagram/context.go

@@ -0,0 +1,20 @@
+package diagram
+
+import (
+	"context"
+	"fmt"
+
+	"modelRT/common"
+	"modelRT/constants"
+)
+
+func clientTokenFromContext(ctx context.Context) (string, error) {
+	if ctx == nil {
+		return "", common.ErrGetClientToken
+	}
+	token, ok := ctx.Value(constants.CtxKeyClientToken).(string)
+	if !ok || token == "" {
+		return "", fmt.Errorf("%w: missing or invalid context value", common.ErrGetClientToken)
+	}
+	return token, nil
+}

diagram/context_test.go

@@ -0,0 +1,38 @@
+package diagram
+
+import (
+	"context"
+	"testing"
+
+	"modelRT/common"
+	"modelRT/constants"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClientTokenFromContext(t *testing.T) {
+	ctx := context.WithValue(context.Background(), constants.CtxKeyClientToken, "test-token")
+	token, err := clientTokenFromContext(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, "test-token", token)
+}
+
+func TestClientTokenFromContextReturnsErrorWhenMissing(t *testing.T) {
+	for _, ctx := range []context.Context{nil, context.Background()} {
+		_, err := clientTokenFromContext(ctx)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, common.ErrGetClientToken)
+	}
+}
+
+func TestRedisConstructorsReturnErrorInsteadOfPanickingWithoutToken(t *testing.T) {
+	ctx := context.Background()
+
+	_, err := NewRedisZSet(ctx, "zset", 0, false)
+	assert.ErrorIs(t, err, common.ErrGetClientToken)
+	_, err = NewRedisSet(ctx, "set", 0, false)
+	assert.ErrorIs(t, err, common.ErrGetClientToken)
+	_, err = NewRedisHash(ctx, "hash", 0, false)
+	assert.ErrorIs(t, err, common.ErrGetClientToken)
+}

diagram/graph.go

@@ -65,9 +65,7 @@ func (g *Graph) AddEdge(from, to uuid.UUID) {
 
 	// 创建新的拓扑信息时，如果被链接的点已经存在于游离节点中
 	// 则将其移除
-	if _, exist := g.FreeVertexs[toKey]; exist {
-		delete(g.FreeVertexs, toKey)
-	}
+	delete(g.FreeVertexs, toKey)
 }
 
 // DelNode delete a node to the graph

diagram/multi_branch_tree.go

@@ -1,125 +0,0 @@
-// Package diagram provide diagram data structure and operation
-package diagram
-
-import (
-	"fmt"
-
-	"github.com/gofrs/uuid"
-)
-
-var GlobalTree *MultiBranchTreeNode
-
-// MultiBranchTreeNode represents a topological structure using an multi branch tree
-type MultiBranchTreeNode struct {
-	ID       uuid.UUID              // 节点唯一标识
-	Parent   *MultiBranchTreeNode   // 指向父节点的指针
-	Children []*MultiBranchTreeNode // 指向所有子节点的指针切片
-}
-
-func NewMultiBranchTree(id uuid.UUID) *MultiBranchTreeNode {
-	return &MultiBranchTreeNode{
-		ID:       id,
-		Children: make([]*MultiBranchTreeNode, 0),
-	}
-}
-
-func (n *MultiBranchTreeNode) AddChild(child *MultiBranchTreeNode) {
-	child.Parent = n
-	n.Children = append(n.Children, child)
-}
-
-func (n *MultiBranchTreeNode) RemoveChild(childID uuid.UUID) bool {
-	for i, child := range n.Children {
-		if child.ID == childID {
-			n.Children = append(n.Children[:i], n.Children[i+1:]...)
-			child.Parent = nil
-			return true
-		}
-	}
-	return false
-}
-
-func (n *MultiBranchTreeNode) FindNodeByID(id uuid.UUID) *MultiBranchTreeNode {
-	if n.ID == id {
-		return n
-	}
-
-	for _, child := range n.Children {
-		if found := child.FindNodeByID(id); found != nil {
-			return found
-		}
-	}
-	return nil
-}
-
-func (n *MultiBranchTreeNode) PrintTree(level int) {
-	for range level {
-		fmt.Print("  ")
-	}
-
-	fmt.Printf("-ID: %s\n", n.ID)
-
-	for _, child := range n.Children {
-		child.PrintTree(level + 1)
-	}
-}
-
-// FindPath returns the ordered node sequence from startID to endID using the
-// supplied nodeMap for O(1) lookup. It walks each node up to the root to find
-// the LCA, then stitches the two half-paths together.
-// Returns nil when either node is absent from nodeMap or no path exists.
-func FindPath(startID, endID uuid.UUID, nodeMap map[uuid.UUID]*MultiBranchTreeNode) []*MultiBranchTreeNode {
-	startNode, ok := nodeMap[startID]
-	if !ok {
-		return nil
-	}
-	endNode, ok := nodeMap[endID]
-	if !ok {
-		return nil
-	}
-
-	// collect ancestors (inclusive) from a node up to the root sentinel
-	ancestors := func(n *MultiBranchTreeNode) []*MultiBranchTreeNode {
-		var chain []*MultiBranchTreeNode
-		for n != nil {
-			chain = append(chain, n)
-			n = n.Parent
-		}
-		return chain
-	}
-
-	startChain := ancestors(startNode) // [start, ..., root]
-	endChain := ancestors(endNode)     // [end,   ..., root]
-
-	// index startChain by ID for fast LCA detection
-	startIdx := make(map[uuid.UUID]int, len(startChain))
-	for i, node := range startChain {
-		startIdx[node.ID] = i
-	}
-
-	// find LCA: first node in endChain that also appears in startChain
-	lcaEndPos := -1
-	lcaStartPos := -1
-	for i, node := range endChain {
-		if j, found := startIdx[node.ID]; found {
-			lcaEndPos = i
-			lcaStartPos = j
-			break
-		}
-	}
-
-	if lcaEndPos < 0 {
-		return nil // disconnected
-	}
-
-	// path = startChain[0..lcaStartPos] reversed + endChain[lcaEndPos..0] reversed
-	path := make([]*MultiBranchTreeNode, 0, lcaStartPos+lcaEndPos+1)
-	for i := 0; i <= lcaStartPos; i++ {
-		path = append(path, startChain[i])
-	}
-	// append end-side (skip LCA to avoid duplication), reversed
-	for i := lcaEndPos - 1; i >= 0; i-- {
-		path = append(path, endChain[i])
-	}
-	return path
-}

diagram/redis_client.go

@@ -3,6 +3,8 @@ package diagram
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 
 	"github.com/redis/go-redis/v9"
 )
@@ -12,6 +14,46 @@ type RedisClient struct {
 	Client *redis.Client
 }
 
+// QueryLatestMeasurementValue returns the score whose member contains the
+// greatest numeric timestamp. Measurement ZSets currently store timestamp in
+// member and measurement value in score.
+func (rc *RedisClient) QueryLatestMeasurementValue(ctx context.Context, key string) (float64, error) {
+	if rc.Client == nil {
+		return 0, fmt.Errorf("redis client is not initialized")
+	}
+
+	members, err := rc.Client.ZRangeWithScores(ctx, key, 0, -1).Result()
+	if err != nil {
+		return 0, err
+	}
+	return latestMeasurementValue(members, key)
+}
+
+func latestMeasurementValue(members []redis.Z, key string) (float64, error) {
+	if len(members) == 0 {
+		return 0, fmt.Errorf("real-time measurement value not found for key %q", key)
+	}
+
+	var latestTimestamp int64
+	var latestValue float64
+	found := false
+	for _, member := range members {
+		timestamp, err := strconv.ParseInt(fmt.Sprint(member.Member), 10, 64)
+		if err != nil {
+			continue
+		}
+		if !found || timestamp > latestTimestamp {
+			latestTimestamp = timestamp
+			latestValue = member.Score
+			found = true
+		}
+	}
+	if !found {
+		return 0, fmt.Errorf("real-time measurement timestamps are invalid for key %q", key)
+	}
+	return latestValue, nil
+}
+
 // NewRedisClient define func of new redis client instance
 func NewRedisClient() *RedisClient {
 	return &RedisClient{

diagram/redis_client_test.go

@@ -0,0 +1,28 @@
+package diagram
+
+import (
+	"testing"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLatestMeasurementValueUsesMemberTimestamp(t *testing.T) {
+	value, err := latestMeasurementValue([]redis.Z{
+		{Member: "100", Score: 999},
+		{Member: "300", Score: 12},
+		{Member: "200", Score: 500},
+	}, "measurement-key")
+
+	require.NoError(t, err)
+	assert.Equal(t, float64(12), value)
+}
+
+func TestLatestMeasurementValueRejectsMissingOrInvalidTimestamps(t *testing.T) {
+	_, err := latestMeasurementValue(nil, "measurement-key")
+	require.Error(t, err)
+
+	_, err = latestMeasurementValue([]redis.Z{{Member: "invalid", Score: 1}}, "measurement-key")
+	require.Error(t, err)
+}

diagram/redis_hash.go

@@ -18,14 +18,17 @@ type RedisHash struct {
 }
 
 // NewRedisHash define func of new redis hash instance
-func NewRedisHash(ctx context.Context, hashKey string, lockLeaseTime uint64, needRefresh bool) *RedisHash {
-	token := ctx.Value("client_token").(string)
+func NewRedisHash(ctx context.Context, hashKey string, lockLeaseTime uint64, needRefresh bool) (*RedisHash, error) {
+	token, err := clientTokenFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
 	return &RedisHash{
 		ctx:           ctx,
 		hashKey:       hashKey,
 		rwLocker:      locker.InitRWLocker(hashKey, token, lockLeaseTime, needRefresh),
 		storageClient: GetRedisClientInstance(),
-	}
+	}, nil
 }
 
 // SetRedisHashByMap define func of set redis hash by map struct

diagram/redis_set.go

@@ -21,15 +21,18 @@ type RedisSet struct {
 }
 
 // NewRedisSet define func of new redis set instance
-func NewRedisSet(ctx context.Context, setKey string, lockLeaseTime uint64, needRefresh bool) *RedisSet {
-	token := ctx.Value("client_token").(string)
+func NewRedisSet(ctx context.Context, setKey string, lockLeaseTime uint64, needRefresh bool) (*RedisSet, error) {
+	token, err := clientTokenFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
 	return &RedisSet{
 		ctx:           ctx,
 		key:           setKey,
 		rwLocker:      locker.InitRWLocker(setKey, token, lockLeaseTime, needRefresh),
 		storageClient: GetRedisClientInstance(),
 		logger:        logger.GetLoggerInstance(),
-	}
+	}, nil
 }
 
 // SADD define func of add redis set by members

diagram/redis_zset.go

@@ -3,8 +3,6 @@ package diagram
 
 import (
 	"context"
-	"iter"
-	"maps"
 
 	locker "modelRT/distributedlock"
 	"modelRT/logger"
@@ -20,13 +18,16 @@ type RedisZSet struct {
 }
 
 // NewRedisZSet define func of new redis zset instance
-func NewRedisZSet(ctx context.Context, key string, lockLeaseTime uint64, needRefresh bool) *RedisZSet {
-	token := ctx.Value("client_token").(string)
+func NewRedisZSet(ctx context.Context, key string, lockLeaseTime uint64, needRefresh bool) (*RedisZSet, error) {
+	token, err := clientTokenFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
 	return &RedisZSet{
 		ctx:           ctx,
 		rwLocker:      locker.InitRWLocker(key, token, lockLeaseTime, needRefresh),
 		storageClient: GetRedisClientInstance(),
-	}
+	}, nil
 }
 
 // ZADD define func of add redis zset by members
@@ -46,6 +47,26 @@ func (rs *RedisZSet) ZADD(setKey string, score float64, member any) error {
 	return nil
 }
 
+// ZREPLACE atomically removes all existing members and adds one new member.
+func (rs *RedisZSet) ZREPLACE(setKey string, score float64, member any) error {
+	if err := rs.rwLocker.WLock(rs.ctx); err != nil {
+		logger.Error(rs.ctx, "lock wLock by setKey failed", "set_key", setKey, "error", err)
+		return err
+	}
+	defer rs.rwLocker.UnWLock(rs.ctx)
+
+	_, err := rs.storageClient.TxPipelined(rs.ctx, func(pipe redis.Pipeliner) error {
+		pipe.Del(rs.ctx, setKey)
+		pipe.ZAdd(rs.ctx, setKey, redis.Z{Score: score, Member: member})
+		return nil
+	})
+	if err != nil {
+		logger.Error(rs.ctx, "replace zset member failed", "set_key", setKey, "member", member, "error", err)
+		return err
+	}
+	return nil
+}
+
 // ZRANGE define func of returns the specified range of elements in the sorted set stored by key
 func (rs *RedisZSet) ZRANGE(setKey string, start, stop int64) ([]string, error) {
 	var results []string
@@ -70,55 +91,3 @@ func (rs *RedisZSet) ZRANGE(setKey string, start, stop int64) ([]string, error)
 	}
 	return results, nil
 }
-
-type Comparer[T any] interface {
-	Compare(T) int
-}
-
-type ComparableComparer[T any] interface {
-	Compare(T) int
-	comparable // 直接嵌入 comparable 约束
-}
-
-type methodNode[E Comparer[E]] struct {
-	value E
-	left  *methodNode[E]
-	right *methodNode[E]
-}
-
-type MethodTree[E Comparer[E]] struct {
-	root *methodNode[E]
-}
-
-type OrderedSet[E interface {
-	comparable
-	Comparer[E]
-}] struct {
-	tree     MethodTree[E]
-	elements map[E]bool
-}
-
-type ComparableOrderedSet[E ComparableComparer[E]] struct {
-	tree     MethodTree[E]
-	elements map[E]bool
-}
-
-type Set[E any] interface {
-	Insert(E)
-	Delete(E)
-	Has(E) bool
-	All() iter.Seq[E]
-}
-
-func InsertAll[E any](set Set[E], seq iter.Seq[E]) {
-	for v := range seq {
-		set.Insert(v)
-	}
-}
-
-type HashSet[E comparable] map[E]bool
-
-func (s HashSet[E]) Insert(v E)       { s[v] = true }
-func (s HashSet[E]) Delete(v E)       { delete(s, v) }
-func (s HashSet[E]) Has(v E) bool     { return s[v] }
-func (s HashSet[E]) All() iter.Seq[E] { return maps.Keys(s) }

diagram/topologic_set.go

@@ -2,32 +2,27 @@
 package diagram
 
 import (
-	"errors"
 	"fmt"
-	"sync"
+
+	"modelRT/util"
 )
 
-// graphOverview define struct of storage all circuit diagram topologic data
-var graphOverview sync.Map
+// graphOverview define struct of storage all circuit diagram topologic data keyed by pageID
+var graphOverview util.TypedMap[int64, *Graph]
 
 // PrintGrapMap define func of print circuit diagram topologic info data
 func PrintGrapMap() {
-	graphOverview.Range(func(key, value any) bool {
-		fmt.Println(key, value)
-		return true
-	})
+	for pageID, graph := range graphOverview.All() {
+		fmt.Println(pageID, graph)
+	}
 }
 
 // GetGraphMap define func of get circuit diagram topologic data by pageID
 func GetGraphMap(pageID int64) (*Graph, error) {
-	value, ok := graphOverview.Load(pageID)
+	graph, ok := graphOverview.Load(pageID)
 	if !ok {
 		return nil, fmt.Errorf("can not find graph by pageID:%d", pageID)
 	}
-	graph, ok := value.(*Graph)
-	if !ok {
-		return nil, errors.New("convert to graph struct failed")
-	}
 	return graph, nil
 }
 

diagram/topology_graph.go

@@ -0,0 +1,138 @@
+package diagram
+
+import (
+	"sync"
+
+	"modelRT/orm"
+
+	"github.com/gofrs/uuid"
+)
+
+// TopologyGraph represents directed topologic links with adjacency lists.
+// It preserves multiple parents for one node.
+type TopologyGraph struct {
+	Nodes      map[uuid.UUID]struct{}
+	OutEdges   map[uuid.UUID][]uuid.UUID
+	InEdges    map[uuid.UUID][]uuid.UUID
+	StartNodes []uuid.UUID
+	EndNodes   []uuid.UUID
+}
+
+var (
+	globalTopologyGraphMu sync.RWMutex
+	GlobalTopologyGraph   *TopologyGraph
+)
+
+// NewTopologyGraph builds a directed graph cache from topologic edges.
+func NewTopologyGraph(edges []orm.Topologic) *TopologyGraph {
+	graph := &TopologyGraph{
+		Nodes:    make(map[uuid.UUID]struct{}, len(edges)*2),
+		OutEdges: make(map[uuid.UUID][]uuid.UUID, len(edges)),
+		InEdges:  make(map[uuid.UUID][]uuid.UUID, len(edges)),
+	}
+
+	for _, edge := range edges {
+		from := edge.UUIDFrom
+		to := edge.UUIDTo
+
+		graph.Nodes[from] = struct{}{}
+		graph.Nodes[to] = struct{}{}
+		graph.OutEdges[from] = append(graph.OutEdges[from], to)
+		graph.InEdges[to] = append(graph.InEdges[to], from)
+	}
+
+	graph.StartNodes = graph.findStartNodes()
+	graph.EndNodes = graph.findEndNodes()
+	return graph
+}
+
+// SetGlobalTopologyGraph replaces the process-wide topology graph cache.
+func SetGlobalTopologyGraph(graph *TopologyGraph) {
+	globalTopologyGraphMu.Lock()
+	defer globalTopologyGraphMu.Unlock()
+	GlobalTopologyGraph = graph
+}
+
+// GetGlobalTopologyGraph returns the process-wide topology graph cache.
+func GetGlobalTopologyGraph() *TopologyGraph {
+	globalTopologyGraphMu.RLock()
+	defer globalTopologyGraphMu.RUnlock()
+	return GlobalTopologyGraph
+}
+
+func (g *TopologyGraph) findStartNodes() []uuid.UUID {
+	startNodes := make([]uuid.UUID, 0)
+	for id := range g.Nodes {
+		if len(g.InEdges[id]) == 0 && len(g.OutEdges[id]) > 0 {
+			startNodes = append(startNodes, id)
+		}
+	}
+	return startNodes
+}
+
+func (g *TopologyGraph) findEndNodes() []uuid.UUID {
+	endNodes := make([]uuid.UUID, 0)
+	for id := range g.Nodes {
+		if len(g.InEdges[id]) > 0 && len(g.OutEdges[id]) == 0 {
+			endNodes = append(endNodes, id)
+		}
+	}
+	return endNodes
+}
+
+// IsReachable reports whether end can be reached from start following directed
+// uuid_from -> uuid_to edges.
+func (g *TopologyGraph) IsReachable(start, end uuid.UUID) bool {
+	return len(g.FindPath(start, end)) > 0
+}
+
+// FindPath returns one shortest directed path from start to end, or nil when
+// no directed path exists.
+func (g *TopologyGraph) FindPath(start, end uuid.UUID) []uuid.UUID {
+	if g == nil {
+		return nil
+	}
+	if start == end {
+		if _, exists := g.Nodes[start]; exists {
+			return []uuid.UUID{start}
+		}
+		return nil
+	}
+
+	visited := map[uuid.UUID]struct{}{start: {}}
+	parent := make(map[uuid.UUID]uuid.UUID)
+	queue := []uuid.UUID{start}
+
+	for len(queue) > 0 {
+		cur := queue[0]
+		queue = queue[1:]
+
+		for _, next := range g.OutEdges[cur] {
+			if _, seen := visited[next]; seen {
+				continue
+			}
+			visited[next] = struct{}{}
+			parent[next] = cur
+
+			if next == end {
+				return reconstructTopologyGraphPath(parent, start, end)
+			}
+
+			queue = append(queue, next)
+		}
+	}
+
+	return nil
+}
+
+func reconstructTopologyGraphPath(parent map[uuid.UUID]uuid.UUID, start, end uuid.UUID) []uuid.UUID {
+	path := make([]uuid.UUID, 0)
+	for cur := end; cur != start; cur = parent[cur] {
+		path = append(path, cur)
+	}
+	path = append(path, start)
+	for i, j := 0, len(path)-1; i < j; i, j = i+1, j-1 {
+		path[i], path[j] = path[j], path[i]
+	}
+	return path
+}

diagram/topology_graph_test.go

@@ -0,0 +1,35 @@
+package diagram
+
+import (
+	"testing"
+
+	"modelRT/orm"
+
+	"github.com/gofrs/uuid"
+)
+
+func TestTopologyGraphSupportsMultiParentReachability(t *testing.T) {
+	startA := uuid.Must(uuid.NewV4())
+	startB := uuid.Must(uuid.NewV4())
+	shared := uuid.Must(uuid.NewV4())
+	end := uuid.Must(uuid.NewV4())
+
+	graph := NewTopologyGraph([]orm.Topologic{
+		{UUIDFrom: startA, UUIDTo: shared},
+		{UUIDFrom: startB, UUIDTo: shared},
+		{UUIDFrom: shared, UUIDTo: end},
+	})
+
+	if len(graph.StartNodes) != 2 {
+		t.Fatalf("expected 2 start nodes, got %d", len(graph.StartNodes))
+	}
+	if len(graph.InEdges[shared]) != 2 {
+		t.Fatalf("expected shared node to keep 2 parents, got %d", len(graph.InEdges[shared]))
+	}
+	if !graph.IsReachable(startA, end) {
+		t.Fatalf("expected %s to reach %s", startA, end)
+	}
+	if !graph.IsReachable(startB, end) {
+		t.Fatalf("expected %s to reach %s", startB, end)
+	}
+}

docs/docs.go

@@ -23,6 +23,57 @@ const docTemplate = `{
     "host": "{{.Host}}",
     "basePath": "{{.BasePath}}",
     "paths": {
+        "/data-object/recommend": {
+            "get": {
+                "description": "根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。",
+                "consumes": [
+                    "application/json"
+                ],
+                "produces": [
+                    "application/json"
+                ],
+                "tags": [
+                    "DataObject Recommend"
+                ],
+                "summary": "测量点推荐（搜索框自动补全）",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "example": "\"grid1\"",
+                        "description": "推荐关键词,例如 'grid1' 或 'grid1.'",
+                        "name": "input",
+                        "in": "query",
+                        "required": true
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "返回推荐列表成功",
+                        "schema": {
+                            "allOf": [
+                                {
+                                    "$ref": "#/definitions/network.SuccessResponse"
+                                },
+                                {
+                                    "type": "object",
+                                    "properties": {
+                                        "payload": {
+                                            "$ref": "#/definitions/network.DataObjectRecommendPayload"
+                                        }
+                                    }
+                                }
+                            ]
+                        }
+                    },
+                    "400": {
+                        "description": "返回推荐列表失败",
+                        "schema": {
+                            "$ref": "#/definitions/network.FailureResponse"
+                        }
+                    }
+                }
+            }
+        },
         "/data/realtime": {
             "get": {
                 "description": "根据用户输入的组件token,从 dataRT 服务中持续获取测点实时数据",
@@ -87,57 +138,6 @@ const docTemplate = `{
                 }
             }
         },
-        "/measurement/recommend": {
-            "get": {
-                "description": "根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Measurement Recommend"
-                ],
-                "summary": "测量点推荐（搜索框自动补全）",
-                "parameters": [
-                    {
-                        "type": "string",
-                        "example": "\"grid1\"",
-                        "description": "推荐关键词,例如 'grid1' 或 'grid1.'",
-                        "name": "input",
-                        "in": "query",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "返回推荐列表成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.MeasurementRecommendPayload"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
-                        "description": "返回推荐列表失败",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    }
-                }
-            }
-        },
         "/model/diagram_load/{page_id}": {
             "get": {
                 "description": "load circuit diagram info by page id",
@@ -332,34 +332,10 @@ const docTemplate = `{
                 ],
                 "responses": {
                     "200": {
-                        "description": "查询成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.AsyncTaskResultQueryResponse"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -388,40 +364,10 @@ const docTemplate = `{
                 ],
                 "responses": {
                     "200": {
-                        "description": "查询成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.AsyncTaskResult"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "404": {
-                        "description": "任务不存在",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -450,28 +396,10 @@ const docTemplate = `{
                 ],
                 "responses": {
                     "200": {
-                        "description": "任务取消成功",
-                        "schema": {
-                            "$ref": "#/definitions/network.SuccessResponse"
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误或任务无法取消",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "404": {
-                        "description": "任务不存在",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -559,23 +487,7 @@ const docTemplate = `{
                 }
             }
         },
-        "network.FailureResponse": {
-            "type": "object",
-            "properties": {
-                "code": {
-                    "type": "integer",
-                    "example": 3000
-                },
-                "msg": {
-                    "type": "string",
-                    "example": "process completed with partial failures"
-                },
-                "payload": {
-                    "type": "object"
-                }
-            }
-        },
-        "network.MeasurementRecommendPayload": {
+        "network.DataObjectRecommendPayload": {
             "type": "object",
             "properties": {
                 "input": {
@@ -596,10 +508,22 @@ const docTemplate = `{
                         " \"I_B_rms\"",
                         "\"I_C_rms\"]"
                     ]
+                }
+            }
+        },
+        "network.FailureResponse": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "integer",
+                    "example": 3000
                 },
-                "recommended_type": {
+                "msg": {
                     "type": "string",
-                    "example": "grid_tag"
+                    "example": "process completed with partial failures"
+                },
+                "payload": {
+                    "type": "object"
                 }
             }
         },

docs/swagger.json

@@ -17,6 +17,57 @@
     "host": "localhost:8080",
     "basePath": "/api/v1",
     "paths": {
+        "/data-object/recommend": {
+            "get": {
+                "description": "根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。",
+                "consumes": [
+                    "application/json"
+                ],
+                "produces": [
+                    "application/json"
+                ],
+                "tags": [
+                    "DataObject Recommend"
+                ],
+                "summary": "测量点推荐（搜索框自动补全）",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "example": "\"grid1\"",
+                        "description": "推荐关键词,例如 'grid1' 或 'grid1.'",
+                        "name": "input",
+                        "in": "query",
+                        "required": true
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "返回推荐列表成功",
+                        "schema": {
+                            "allOf": [
+                                {
+                                    "$ref": "#/definitions/network.SuccessResponse"
+                                },
+                                {
+                                    "type": "object",
+                                    "properties": {
+                                        "payload": {
+                                            "$ref": "#/definitions/network.DataObjectRecommendPayload"
+                                        }
+                                    }
+                                }
+                            ]
+                        }
+                    },
+                    "400": {
+                        "description": "返回推荐列表失败",
+                        "schema": {
+                            "$ref": "#/definitions/network.FailureResponse"
+                        }
+                    }
+                }
+            }
+        },
         "/data/realtime": {
             "get": {
                 "description": "根据用户输入的组件token,从 dataRT 服务中持续获取测点实时数据",
@@ -81,57 +132,6 @@
                 }
             }
         },
-        "/measurement/recommend": {
-            "get": {
-                "description": "根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Measurement Recommend"
-                ],
-                "summary": "测量点推荐（搜索框自动补全）",
-                "parameters": [
-                    {
-                        "type": "string",
-                        "example": "\"grid1\"",
-                        "description": "推荐关键词,例如 'grid1' 或 'grid1.'",
-                        "name": "input",
-                        "in": "query",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "返回推荐列表成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.MeasurementRecommendPayload"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
-                        "description": "返回推荐列表失败",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    }
-                }
-            }
-        },
         "/model/diagram_load/{page_id}": {
             "get": {
                 "description": "load circuit diagram info by page id",
@@ -326,34 +326,10 @@
                 ],
                 "responses": {
                     "200": {
-                        "description": "查询成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.AsyncTaskResultQueryResponse"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -382,40 +358,10 @@
                 ],
                 "responses": {
                     "200": {
-                        "description": "查询成功",
-                        "schema": {
-                            "allOf": [
-                                {
-                                    "$ref": "#/definitions/network.SuccessResponse"
-                                },
-                                {
-                                    "type": "object",
-                                    "properties": {
-                                        "payload": {
-                                            "$ref": "#/definitions/network.AsyncTaskResult"
-                                        }
-                                    }
-                                }
-                            ]
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "404": {
-                        "description": "任务不存在",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -444,28 +390,10 @@
                 ],
                 "responses": {
                     "200": {
-                        "description": "任务取消成功",
-                        "schema": {
-                            "$ref": "#/definitions/network.SuccessResponse"
-                        }
-                    },
-                    "400": {
                         "description": "请求参数错误或任务无法取消",
                         "schema": {
                             "$ref": "#/definitions/network.FailureResponse"
                         }
-                    },
-                    "404": {
-                        "description": "任务不存在",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
-                    },
-                    "500": {
-                        "description": "服务器内部错误",
-                        "schema": {
-                            "$ref": "#/definitions/network.FailureResponse"
-                        }
                     }
                 }
             }
@@ -553,23 +481,7 @@
                 }
             }
         },
-        "network.FailureResponse": {
-            "type": "object",
-            "properties": {
-                "code": {
-                    "type": "integer",
-                    "example": 3000
-                },
-                "msg": {
-                    "type": "string",
-                    "example": "process completed with partial failures"
-                },
-                "payload": {
-                    "type": "object"
-                }
-            }
-        },
-        "network.MeasurementRecommendPayload": {
+        "network.DataObjectRecommendPayload": {
             "type": "object",
             "properties": {
                 "input": {
@@ -590,10 +502,22 @@
                         " \"I_B_rms\"",
                         "\"I_C_rms\"]"
                     ]
+                }
+            }
+        },
+        "network.FailureResponse": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "integer",
+                    "example": 3000
                 },
-                "recommended_type": {
+                "msg": {
                     "type": "string",
-                    "example": "grid_tag"
+                    "example": "process completed with partial failures"
+                },
+                "payload": {
+                    "type": "object"
                 }
             }
         },

docs/swagger.yaml

@@ -59,18 +59,7 @@ definitions:
         example: 3
         type: integer
     type: object
-  network.FailureResponse:
-    properties:
-      code:
-        example: 3000
-        type: integer
-      msg:
-        example: process completed with partial failures
-        type: string
-      payload:
-        type: object
-    type: object
-  network.MeasurementRecommendPayload:
+  network.DataObjectRecommendPayload:
     properties:
       input:
         example: transformfeeder1_220.
@@ -86,9 +75,17 @@ definitions:
         items:
           type: string
         type: array
-      recommended_type:
-        example: grid_tag
+    type: object
+  network.FailureResponse:
+    properties:
+      code:
+        example: 3000
+        type: integer
+      msg:
+        example: process completed with partial failures
         type: string
+      payload:
+        type: object
     type: object
   network.RealTimeDataPayload:
     properties:
@@ -172,6 +169,37 @@ info:
   title: ModelRT 实时模型服务 API 文档
   version: "1.0"
 paths:
+  /data-object/recommend:
+    get:
+      consumes:
+      - application/json
+      description: 根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。
+      parameters:
+      - description: 推荐关键词,例如 'grid1' 或 'grid1.'
+        example: '"grid1"'
+        in: query
+        name: input
+        required: true
+        type: string
+      produces:
+      - application/json
+      responses:
+        "200":
+          description: 返回推荐列表成功
+          schema:
+            allOf:
+            - $ref: '#/definitions/network.SuccessResponse'
+            - properties:
+                payload:
+                  $ref: '#/definitions/network.DataObjectRecommendPayload'
+              type: object
+        "400":
+          description: 返回推荐列表失败
+          schema:
+            $ref: '#/definitions/network.FailureResponse'
+      summary: 测量点推荐（搜索框自动补全）
+      tags:
+      - DataObject Recommend
   /data/realtime:
     get:
       consumes:
@@ -212,37 +240,6 @@ paths:
       summary: 获取实时测点数据
       tags:
       - RealTime Component
-  /measurement/recommend:
-    get:
-      consumes:
-      - application/json
-      description: 根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。
-      parameters:
-      - description: 推荐关键词,例如 'grid1' 或 'grid1.'
-        example: '"grid1"'
-        in: query
-        name: input
-        required: true
-        type: string
-      produces:
-      - application/json
-      responses:
-        "200":
-          description: 返回推荐列表成功
-          schema:
-            allOf:
-            - $ref: '#/definitions/network.SuccessResponse'
-            - properties:
-                payload:
-                  $ref: '#/definitions/network.MeasurementRecommendPayload'
-              type: object
-        "400":
-          description: 返回推荐列表失败
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
-      summary: 测量点推荐（搜索框自动补全）
-      tags:
-      - Measurement Recommend
   /model/diagram_load/{page_id}:
     get:
       consumes:
@@ -361,26 +358,9 @@ paths:
       - application/json
       responses:
         "200":
-          description: 查询成功
-          schema:
-            allOf:
-            - $ref: '#/definitions/network.SuccessResponse'
-            - properties:
-                payload:
-                  $ref: '#/definitions/network.AsyncTaskResult'
-              type: object
-        "400":
           description: 请求参数错误
           schema:
             $ref: '#/definitions/network.FailureResponse'
-        "404":
-          description: 任务不存在
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
-        "500":
-          description: 服务器内部错误
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
       summary: 查询异步任务详情
       tags:
       - AsyncTask
@@ -399,21 +379,9 @@ paths:
       - application/json
       responses:
         "200":
-          description: 任务取消成功
-          schema:
-            $ref: '#/definitions/network.SuccessResponse'
-        "400":
           description: 请求参数错误或任务无法取消
           schema:
             $ref: '#/definitions/network.FailureResponse'
-        "404":
-          description: 任务不存在
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
-        "500":
-          description: 服务器内部错误
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
       summary: 取消异步任务
       tags:
       - AsyncTask
@@ -432,22 +400,9 @@ paths:
       - application/json
       responses:
         "200":
-          description: 查询成功
-          schema:
-            allOf:
-            - $ref: '#/definitions/network.SuccessResponse'
-            - properties:
-                payload:
-                  $ref: '#/definitions/network.AsyncTaskResultQueryResponse'
-              type: object
-        "400":
           description: 请求参数错误
           schema:
             $ref: '#/definitions/network.FailureResponse'
-        "500":
-          description: 服务器内部错误
-          schema:
-            $ref: '#/definitions/network.FailureResponse'
       summary: 查询异步任务结果
       tags:
       - AsyncTask

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

handler/alert_event_query.go

@@ -15,13 +15,14 @@ import (
 
 // QueryAlertEventHandler define query alert event process API
 func QueryAlertEventHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var targetLevel constants.AlertLevel
 
 	alertManger := alert.GetAlertMangerInstance()
 	levelStr := c.Query("level")
 	level, err := strconv.Atoi(levelStr)
 	if err != nil {
-		logger.Error(c, "convert alert level string to int failed", "error", err)
+		logger.Error(ctx, "convert alert level string to int failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: -1,

handler/anchor_point_replace.go

@@ -19,15 +19,16 @@ import (
 
 // ComponentAnchorReplaceHandler define component anchor point replace process API
 func ComponentAnchorReplaceHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var uuid, anchorName string
 
 	pgClient := database.GetPostgresDBClient()
-	cancelCtx, cancel := context.WithTimeout(c, 5*time.Second)
+	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
 	var request network.ComponetAnchorReplaceRequest
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "unmarshal component anchor point replace info failed", "error", err)
+		logger.Error(ctx, "unmarshal component anchor point replace info failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -42,7 +43,7 @@ func ComponentAnchorReplaceHandler(c *gin.Context) {
 	var componentInfo orm.Component
 	result := pgClient.WithContext(cancelCtx).Model(&orm.Component{}).Where("global_uuid = ?", uuid).Find(&componentInfo)
 	if result.Error != nil {
-		logger.Error(c, "query component detail info failed", "error", result.Error)
+		logger.Error(ctx, "query component detail info failed", "error", result.Error)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -54,7 +55,7 @@ func ComponentAnchorReplaceHandler(c *gin.Context) {
 
 	if result.RowsAffected == 0 {
 		err := fmt.Errorf("query component detail info by uuid failed:%w", errcode.ErrQueryRowZero)
-		logger.Error(c, "query component detail info from table is empty", "table_name", "component")
+		logger.Error(ctx, "query component detail info from table is empty", "table_name", "component")
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,

handler/async_task_create_handler.go

@@ -2,6 +2,8 @@
 package handler
 
 import (
+	"fmt"
+
 	"modelRT/constants"
 	"modelRT/database"
 	"modelRT/logger"
@@ -152,6 +154,7 @@ func validateBatchImportParams(params map[string]any) bool {
 func validateTestTaskParams(params map[string]any) bool {
 	// Test task has optional parameters, all are valid
 	// sleep_duration defaults to 60 seconds if not provided
+	fmt.Println("Test task parameters:", params)
 	return true
 }
 

handler/attr_delete.go

@@ -13,12 +13,13 @@ import (
 
 // AttrDeleteHandler deletes a data attribute
 func AttrDeleteHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.AttrDeleteRequest
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
 
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -27,7 +28,7 @@ func AttrDeleteHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal attribute delete request", "error", err)
+		logger.Error(ctx, "failed to unmarshal attribute delete request", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -35,9 +36,9 @@ func AttrDeleteHandler(c *gin.Context) {
 		return
 	}
 
-	rs := diagram.NewRedisString(c, request.AttrToken, clientToken, 10, true)
+	rs := diagram.NewRedisString(ctx, request.AttrToken, clientToken, 10, true)
 	if err := rs.GETDEL(request.AttrToken); err != nil {
-		logger.Error(c, "failed to delete attribute from Redis", "attr_token", request.AttrToken, "error", err)
+		logger.Error(ctx, "failed to delete attribute from Redis", "attr_token", request.AttrToken, "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code:    http.StatusBadRequest,
 			Msg:     err.Error(),

handler/attr_load.go

@@ -13,13 +13,14 @@ import (
 
 // AttrGetHandler retrieves the value of a data attribute
 func AttrGetHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.AttrGetRequest
 
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
 
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -28,7 +29,7 @@ func AttrGetHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal attribute get request", "error", err)
+		logger.Error(ctx, "failed to unmarshal attribute get request", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -37,12 +38,12 @@ func AttrGetHandler(c *gin.Context) {
 	}
 
 	pgClient := database.GetPostgresDBClient()
-	tx := pgClient.Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 
-	attrModel, err := database.ParseAttrToken(c, tx, request.AttrToken, clientToken)
+	attrModel, err := database.ParseAttrToken(ctx, tx, request.AttrToken, clientToken)
 	if err != nil {
 		tx.Rollback()
-		logger.Error(c, "failed to parse attribute token", "attr_token", request.AttrToken, "error", err)
+		logger.Error(ctx, "failed to parse attribute token", "attr_token", request.AttrToken, "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code:    http.StatusBadRequest,
 			Msg:     err.Error(),

handler/attr_update.go

@@ -13,13 +13,14 @@ import (
 
 // AttrSetHandler sets the value of a data attribute
 func AttrSetHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.AttrSetRequest
 
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
 
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -28,7 +29,7 @@ func AttrSetHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal attribute set request", "error", err)
+		logger.Error(ctx, "failed to unmarshal attribute set request", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -37,9 +38,9 @@ func AttrSetHandler(c *gin.Context) {
 	}
 
 	// The logic for handling Redis operations directly from the handler
-	rs := diagram.NewRedisString(c, request.AttrToken, clientToken, 10, true)
+	rs := diagram.NewRedisString(ctx, request.AttrToken, clientToken, 10, true)
 	if err := rs.Set(request.AttrToken, request.AttrValue); err != nil {
-		logger.Error(c, "failed to set attribute value in Redis", "attr_token", request.AttrToken, "error", err)
+		logger.Error(ctx, "failed to set attribute value in Redis", "attr_token", request.AttrToken, "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code:    http.StatusBadRequest,
 			Msg:     err.Error(),

handler/circuit_diagram_create.go

@@ -16,11 +16,12 @@ import (
 
 // CircuitDiagramCreateHandler define circuit diagram create process API
 func CircuitDiagramCreateHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 
 	var request network.CircuitDiagramCreateRequest
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "unmarshal circuit diagram create info failed", "error", err)
+		logger.Error(ctx, "unmarshal circuit diagram create info failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -32,7 +33,7 @@ func CircuitDiagramCreateHandler(c *gin.Context) {
 
 	graph, err := diagram.GetGraphMap(request.PageID)
 	if err != nil {
-		logger.Error(c, "get topologic data from set by pageID failed", "error", err)
+		logger.Error(ctx, "get topologic data from set by pageID failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -60,7 +61,7 @@ func CircuitDiagramCreateHandler(c *gin.Context) {
 				err = fmt.Errorf("convert uuid from string failed:%w:%w", err1, err2)
 			}
 
-			logger.Error(c, "format uuid from string failed", "error", err)
+			logger.Error(ctx, "format uuid from string failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -78,13 +79,13 @@ func CircuitDiagramCreateHandler(c *gin.Context) {
 	}
 
 	// open transaction
-	tx := pgClient.Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 
-	err = database.CreateTopologicIntoDB(c, tx, request.PageID, topologicCreateInfos)
+	err = database.CreateTopologicIntoDB(ctx, tx, request.PageID, topologicCreateInfos)
 	if err != nil {
 		tx.Rollback()
 
-		logger.Error(c, "create topologic info into DB failed", "topologic_info", topologicCreateInfos, "error", err)
+		logger.Error(ctx, "create topologic info into DB failed", "topologic_info", topologicCreateInfos, "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -102,11 +103,11 @@ func CircuitDiagramCreateHandler(c *gin.Context) {
 	}
 
 	for index, info := range request.ComponentInfos {
-		componentUUID, err := database.CreateComponentIntoDB(c, tx, info)
+		componentUUID, err := database.CreateComponentIntoDB(ctx, tx, info)
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "insert component info into DB failed", "error", err)
+			logger.Error(ctx, "insert component info into DB failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -125,7 +126,7 @@ func CircuitDiagramCreateHandler(c *gin.Context) {
 		// TODO 修复赋值问题
 		component, err := network.ConvertComponentCreateInfosToComponents(info)
 		if err != nil {
-			logger.Error(c, "convert component params info failed", "component_info", info, "error", err)
+			logger.Error(ctx, "convert component params info failed", "component_info", info, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,

handler/circuit_diagram_delete.go

@@ -21,11 +21,12 @@ import (
 
 // CircuitDiagramDeleteHandler define circuit diagram delete process API
 func CircuitDiagramDeleteHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 
 	var request network.CircuitDiagramDeleteRequest
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "unmarshal circuit diagram del info failed", "error", err)
+		logger.Error(ctx, "unmarshal circuit diagram del info failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -37,7 +38,7 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 
 	graph, err := diagram.GetGraphMap(request.PageID)
 	if err != nil {
-		logger.Error(c, "get topologic data from set by pageID failed", "error", err)
+		logger.Error(ctx, "get topologic data from set by pageID failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -65,7 +66,7 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 				err = fmt.Errorf("convert uuid from string failed:%w:%w", err1, err2)
 			}
 
-			logger.Error(c, "format uuid from string failed", "error", err)
+			logger.Error(ctx, "format uuid from string failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -83,14 +84,14 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 	}
 
 	// open transaction
-	tx := pgClient.Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 
 	for _, topologicDelInfo := range topologicDelInfos {
-		err = database.DeleteTopologicIntoDB(c, tx, request.PageID, topologicDelInfo)
+		err = database.DeleteTopologicIntoDB(ctx, tx, request.PageID, topologicDelInfo)
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "delete topologic info into DB failed", "topologic_info", topologicDelInfo, "error", err)
+			logger.Error(ctx, "delete topologic info into DB failed", "topologic_info", topologicDelInfo, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -107,7 +108,7 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "delete topologic info failed", "topologic_info", topologicDelInfo, "error", err)
+			logger.Error(ctx, "delete topologic info failed", "topologic_info", topologicDelInfo, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -126,14 +127,14 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 	}
 
 	for _, componentInfo := range request.ComponentInfos {
-		cancelCtx, cancel := context.WithTimeout(c, 5*time.Second)
+		cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		defer cancel()
 
 		globalUUID, err := uuid.FromString(componentInfo.UUID)
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "format uuid from string failed", "error", err)
+			logger.Error(ctx, "format uuid from string failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -157,7 +158,7 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 				err = fmt.Errorf("%w:please check uuid conditions", errcode.ErrDeleteRowZero)
 			}
 
-			logger.Error(c, "query component info into postgresDB failed", "component_global_uuid", componentInfo.UUID, "error", err)
+			logger.Error(ctx, "query component info into postgresDB failed", "component_global_uuid", componentInfo.UUID, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -179,7 +180,7 @@ func CircuitDiagramDeleteHandler(c *gin.Context) {
 				err = fmt.Errorf("%w:please check uuid conditions", errcode.ErrDeleteRowZero)
 			}
 
-			logger.Error(c, "delete component info into postgresDB failed", "component_global_uuid", componentInfo.UUID, "error", err)
+			logger.Error(ctx, "delete component info into postgresDB failed", "component_global_uuid", componentInfo.UUID, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,

handler/circuit_diagram_load.go

@@ -24,11 +24,12 @@ import (
 // @Failure 400 {object} network.FailureResponse "request process failed"
 // @Router /model/diagram_load/{page_id} [get]
 func CircuitDiagramLoadHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 
 	pageID, err := strconv.ParseInt(c.Query("page_id"), 10, 64)
 	if err != nil {
-		logger.Error(c, "get pageID from url param failed", "error", err)
+		logger.Error(ctx, "get pageID from url param failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -43,7 +44,7 @@ func CircuitDiagramLoadHandler(c *gin.Context) {
 
 	topologicInfo, err := diagram.GetGraphMap(pageID)
 	if err != nil {
-		logger.Error(c, "get topologic data from set by pageID failed", "error", err)
+		logger.Error(ctx, "get topologic data from set by pageID failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -62,9 +63,9 @@ func CircuitDiagramLoadHandler(c *gin.Context) {
 	componentParamMap := make(map[string]any)
 	for _, VerticeLink := range topologicInfo.VerticeLinks {
 		for _, componentUUID := range VerticeLink {
-			component, err := database.QueryComponentByUUID(c, pgClient, componentUUID)
+			component, err := database.QueryComponentByUUID(ctx, pgClient, componentUUID)
 			if err != nil {
-				logger.Error(c, "get component id info from DB by uuid failed", "error", err)
+				logger.Error(ctx, "get component id info from DB by uuid failed", "error", err)
 
 				resp := network.FailureResponse{
 					Code: http.StatusBadRequest,
@@ -79,7 +80,7 @@ func CircuitDiagramLoadHandler(c *gin.Context) {
 
 			componentParams, err := diagram.GetComponentMap(component.GlobalUUID.String())
 			if err != nil {
-				logger.Error(c, "get component data from set by uuid failed", "error", err)
+				logger.Error(ctx, "get component data from set by uuid failed", "error", err)
 
 				resp := network.FailureResponse{
 					Code: http.StatusBadRequest,
@@ -96,9 +97,9 @@ func CircuitDiagramLoadHandler(c *gin.Context) {
 	}
 
 	rootVertexUUID := topologicInfo.RootVertex.String()
-	rootComponent, err := database.QueryComponentByUUID(c, pgClient, topologicInfo.RootVertex)
+	rootComponent, err := database.QueryComponentByUUID(ctx, pgClient, topologicInfo.RootVertex)
 	if err != nil {
-		logger.Error(c, "get component id info from DB by uuid failed", "error", err)
+		logger.Error(ctx, "get component id info from DB by uuid failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -113,7 +114,7 @@ func CircuitDiagramLoadHandler(c *gin.Context) {
 
 	rootComponentParam, err := diagram.GetComponentMap(rootComponent.GlobalUUID.String())
 	if err != nil {
-		logger.Error(c, "get component data from set by uuid failed", "error", err)
+		logger.Error(ctx, "get component data from set by uuid failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,

handler/circuit_diagram_update.go

@@ -14,11 +14,12 @@ import (
 
 // CircuitDiagramUpdateHandler define circuit diagram update process API
 func CircuitDiagramUpdateHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 
 	var request network.CircuitDiagramUpdateRequest
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "unmarshal circuit diagram update info failed", "error", err)
+		logger.Error(ctx, "unmarshal circuit diagram update info failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -30,7 +31,7 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 
 	graph, err := diagram.GetGraphMap(request.PageID)
 	if err != nil {
-		logger.Error(c, "get topologic data from set by pageID failed", "error", err)
+		logger.Error(ctx, "get topologic data from set by pageID failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -47,7 +48,7 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 	for _, topologicLink := range request.TopologicLinks {
 		changeInfo, err := network.ParseUUID(topologicLink)
 		if err != nil {
-			logger.Error(c, "format uuid from string failed", "error", err)
+			logger.Error(ctx, "format uuid from string failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -63,14 +64,14 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 	}
 
 	// open transaction
-	tx := pgClient.Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 
 	for _, topologicChangeInfo := range topologicChangeInfos {
-		err = database.UpdateTopologicIntoDB(c, tx, request.PageID, topologicChangeInfo)
+		err = database.UpdateTopologicIntoDB(ctx, tx, request.PageID, topologicChangeInfo)
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "update topologic info into DB failed", "topologic_info", topologicChangeInfo, "error", err)
+			logger.Error(ctx, "update topologic info into DB failed", "topologic_info", topologicChangeInfo, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -87,7 +88,7 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 		if err != nil {
 			tx.Rollback()
 
-			logger.Error(c, "update topologic info failed", "topologic_info", topologicChangeInfo, "error", err)
+			logger.Error(ctx, "update topologic info failed", "topologic_info", topologicChangeInfo, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -102,9 +103,9 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 	}
 
 	for index, componentInfo := range request.ComponentInfos {
-		componentUUID, err := database.UpdateComponentIntoDB(c, tx, componentInfo)
+		componentUUID, err := database.UpdateComponentIntoDB(ctx, tx, componentInfo)
 		if err != nil {
-			logger.Error(c, "udpate component info into DB failed", "error", err)
+			logger.Error(ctx, "udpate component info into DB failed", "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -124,7 +125,7 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 		// TODO 修复赋值问题
 		component, err := network.ConvertComponentUpdateInfosToComponents(info)
 		if err != nil {
-			logger.Error(c, "convert component params info failed", "component_info", info, "error", err)
+			logger.Error(ctx, "convert component params info failed", "component_info", info, "error", err)
 
 			resp := network.FailureResponse{
 				Code: http.StatusBadRequest,
@@ -137,7 +138,7 @@ func CircuitDiagramUpdateHandler(c *gin.Context) {
 			c.JSON(http.StatusOK, resp)
 			return
 		}
-		diagram.UpdateComponentMap(info.ID, component)
+		diagram.UpdateComponentMap(info.UUID, component)
 	}
 
 	if len(request.FreeVertexs) > 0 {

handler/component_attribute_query.go

@@ -8,8 +8,6 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/gofrs/uuid"
-
 	"modelRT/common/errcode"
 	"modelRT/constants"
 	"modelRT/database"
@@ -18,16 +16,18 @@ import (
 	"modelRT/orm"
 
 	"github.com/gin-gonic/gin"
+	"github.com/gofrs/uuid"
 )
 
 // ComponentAttributeQueryHandler define circuit diagram component attribute value query process API
 func ComponentAttributeQueryHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 
 	tokens := c.Param("tokens")
 	if tokens == "" {
 		err := fmt.Errorf("tokens is missing from the path")
-		logger.Error(c, "query tokens from path failed", "error", err, "url", c.Request.RequestURI)
+		logger.Error(ctx, "query tokens from path failed", "error", err, "url", c.Request.RequestURI)
 		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
 		return
 	}
@@ -54,10 +54,18 @@ func ComponentAttributeQueryHandler(c *gin.Context) {
 	dbQueryMap := make(map[string][]cacheQueryItem)
 	var secondaryQueryCount int
 	for hSetKey, items := range cacheQueryMap {
-		hset := diagram.NewRedisHash(c, hSetKey, 5000, false)
+		hset, err := diagram.NewRedisHash(ctx, hSetKey, 5000, false)
+		if err != nil {
+			logger.Warn(ctx, "create redis hash failed", "key", hSetKey, "error", err)
+			for _, item := range items {
+				dbQueryMap[item.attributeCompTag] = append(dbQueryMap[item.attributeCompTag], item)
+				secondaryQueryCount++
+			}
+			continue
+		}
 		cacheData, err := hset.HGetAll()
 		if err != nil {
-			logger.Warn(c, "redis hgetall failed", "key", hSetKey, "err", err)
+			logger.Warn(ctx, "redis hgetall failed", "key", hSetKey, "err", err)
 		}
 		for _, item := range items {
 			if val, ok := cacheData[item.attributeName]; ok {
@@ -75,9 +83,9 @@ func ComponentAttributeQueryHandler(c *gin.Context) {
 		return
 	}
 
-	tx := pgClient.WithContext(c).Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 	if tx.Error != nil {
-		logger.Error(c, "begin postgres transaction failed", "error", tx.Error)
+		logger.Error(ctx, "begin postgres transaction failed", "error", tx.Error)
 		fillRemainingErrors(queryResults, tokenSlice, errcode.ErrBeginTxFailed)
 		payload := genQueryRespPayload(queryResults, tokenSlice)
 		renderRespFailure(c, constants.RespCodeServerError, "begin postgres database transaction failed", payload)
@@ -86,9 +94,9 @@ func ComponentAttributeQueryHandler(c *gin.Context) {
 	defer tx.Rollback()
 
 	allCompTags := slices.Collect(maps.Keys(dbQueryMap))
-	compModelMap, err := database.QueryComponentByCompTags(c, tx, allCompTags)
+	compModelMap, err := database.QueryComponentByCompTags(ctx, tx, allCompTags)
 	if err != nil {
-		logger.Error(c, "query component info from postgres database failed", "error", err)
+		logger.Error(ctx, "query component info from postgres database failed", "error", err)
 		fillRemainingErrors(queryResults, tokenSlice, errcode.ErrDBQueryFailed)
 		payload := genQueryRespPayload(queryResults, tokenSlice)
 		renderRespFailure(c, constants.RespCodeServerError, "query component meta failed", payload)
@@ -116,7 +124,7 @@ func ComponentAttributeQueryHandler(c *gin.Context) {
 
 	tableNameMap, err := database.BatchGetProjectNames(tx, identifiers)
 	if err != nil {
-		logger.Error(c, "batch get table names from postgres database failed", "error", err)
+		logger.Error(ctx, "batch get table names from postgres database failed", "error", err)
 		fillRemainingErrors(queryResults, tokenSlice, errcode.ErrRetrieveFailed)
 		payload := genQueryRespPayload(queryResults, tokenSlice)
 		renderRespFailure(c, constants.RespCodeServerError, "batch get table names from postgres database failed", payload)
@@ -151,10 +159,11 @@ func ComponentAttributeQueryHandler(c *gin.Context) {
 	}
 
 	if err := tx.Commit().Error; err != nil {
-		logger.Warn(c, "postgres transaction commit failed, but returning scanned data", "error", err)
+		logger.Warn(ctx, "postgres transaction commit failed, but returning scanned data", "error", err)
 	} else {
+		backfillCtx := context.WithoutCancel(ctx)
 		for hKey, items := range redisSyncMap {
-			go backfillRedis(c.Copy(), hKey, items)
+			go backfillRedis(backfillCtx, hKey, items)
 		}
 	}
 
@@ -184,7 +193,11 @@ func fillRemainingErrors(results map[string]queryResult, tokens []string, err *e
 }
 
 func backfillRedis(ctx context.Context, hSetKey string, items []cacheQueryItem) {
-	hset := diagram.NewRedisHash(ctx, hSetKey, 5000, false)
+	hset, err := diagram.NewRedisHash(ctx, hSetKey, 5000, false)
+	if err != nil {
+		logger.Error(ctx, "create redis hash for async backfill failed", "hash_key", hSetKey, "error", err)
+		return
+	}
 	fields := make(map[string]any, len(items))
 	for _, item := range items {
 		if item.attributeVal != "" {

handler/component_attribute_update.go

@@ -18,10 +18,11 @@ import (
 
 // ComponentAttributeUpdateHandler define circuit diagram component attribute value update process API
 func ComponentAttributeUpdateHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	pgClient := database.GetPostgresDBClient()
 	var request network.ComponentAttributeUpdateInfo
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "unmarshal request params failed", "error", err)
+		logger.Error(ctx, "unmarshal request params failed", "error", err)
 		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
 		return
 	}
@@ -54,16 +55,16 @@ func ComponentAttributeUpdateHandler(c *gin.Context) {
 	}
 
 	// open transaction
-	tx := pgClient.WithContext(c).Begin()
+	tx := pgClient.WithContext(ctx).Begin()
 	if tx.Error != nil {
-		logger.Error(c, "begin postgres transaction failed", "error", tx.Error)
+		logger.Error(ctx, "begin postgres transaction failed", "error", tx.Error)
 		renderRespFailure(c, constants.RespCodeServerError, "begin postgres transaction failed", nil)
 		return
 	}
 
-	compInfo, err := database.QueryComponentByCompTag(c, tx, attributeComponentTag)
+	compInfo, err := database.QueryComponentByCompTag(ctx, tx, attributeComponentTag)
 	if err != nil {
-		logger.Error(c, "query component info by component tag failed", "error", err, "tag", attributeComponentTag)
+		logger.Error(ctx, "query component info by component tag failed", "error", err, "tag", attributeComponentTag)
 
 		for _, attribute := range request.AttributeConfigs {
 			if _, exists := updateResults[attribute.AttributeToken]; !exists {
@@ -139,7 +140,14 @@ func ComponentAttributeUpdateHandler(c *gin.Context) {
 	}
 
 	for key, items := range redisUpdateMap {
-		hset := diagram.NewRedisHash(c, key, 5000, false)
+		hset, err := diagram.NewRedisHash(ctx, key, 5000, false)
+		if err != nil {
+			logger.Error(ctx, "create redis hash failed", "hash_key", key, "error", err)
+			for _, item := range items {
+				updateResults[item.token] = errcode.ErrCacheSyncWarn.WithCause(err)
+			}
+			continue
+		}
 
 		fields := make(map[string]any, len(items))
 		for _, item := range items {
@@ -147,7 +155,7 @@ func ComponentAttributeUpdateHandler(c *gin.Context) {
 		}
 
 		if err := hset.SetRedisHashByMap(fields); err != nil {
-			logger.Error(c, "batch sync redis failed", "hash_key", key, "error", err)
+			logger.Error(ctx, "batch sync redis failed", "hash_key", key, "error", err)
 
 			for _, item := range items {
 				if _, exists := updateResults[item.token]; exists {

handler/data_object_attribute_query.go

@@ -0,0 +1,337 @@
+// Package handler provides HTTP handlers for various endpoints.
+package handler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"modelRT/common"
+	"modelRT/common/errcode"
+	"modelRT/constants"
+	"modelRT/database"
+	"modelRT/diagram"
+	"modelRT/logger"
+	"modelRT/model"
+	"modelRT/orm"
+
+	"github.com/gin-gonic/gin"
+)
+
+// DataObjectAttributeQueryHandler define data object attribute value query process API
+func DataObjectAttributeQueryHandler(c *gin.Context) {
+	ctx := c.Request.Context()
+	pgClient := database.GetPostgresDBClient()
+
+	token, field, err := parseDataObjectAttributeQuery(c)
+	if err != nil {
+		logger.Error(ctx, "query token from query parameters failed", "error", err, "url", c.Request.RequestURI)
+		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+		return
+	}
+
+	dataObjectType, err := model.ClassifyDataObjectToken(token)
+	if err != nil {
+		logger.Error(ctx, "classify data object token failed", "error", err, "url", c.Request.RequestURI)
+		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+		return
+	}
+
+	if err := validateDataObjectField(dataObjectType, field); err != nil {
+		logger.Warn(ctx, "validate data object field failed", "token", token, "field", field, "error", err)
+		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+		return
+	}
+
+	var parameter *database.ParameterDataObject
+	var measurement *orm.Measurement
+	var measurementComponent *orm.Component
+	switch dataObjectType {
+	case constants.DataObjectTypeParameter:
+		// 参量支持两种形式token4.token5.token6.token7与token1.token2.token3.token4.token5.token6.token7
+		parameter, err = database.QueryParameterByDataObjectToken(ctx, pgClient, token)
+		if err != nil {
+			if errors.Is(err, common.ErrInvalidParameterToken) ||
+				errors.Is(err, common.ErrParameterTokenNotFound) ||
+				errors.Is(err, common.ErrAmbiguousParameterToken) {
+				logger.Warn(ctx, "validate parameter token failed", "token", token, "error", err)
+				renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+				return
+			}
+
+			logger.Error(ctx, "query parameter token from postgres failed", "token", token, "error", err)
+			renderRespFailure(c, constants.RespCodeServerError, "validate parameter token failed", nil)
+			return
+		}
+	case constants.DataObjectTypeMeasurement:
+		// 量测支持token1.token2.token3.token4.token5.token6.token7、token4.token5.token6.token7、token4.token7
+		measurement, measurementComponent, err = database.QueryMeasurementByDataObjectToken(ctx, pgClient, token)
+		if err != nil {
+			if errors.Is(err, common.ErrInvalidMeasurementToken) ||
+				errors.Is(err, common.ErrMeasurementTokenNotFound) ||
+				errors.Is(err, common.ErrAmbiguousMeasurementToken) {
+				logger.Warn(ctx, "validate measurement token failed", "token", token, "error", err)
+				renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+				return
+			}
+
+			logger.Error(ctx, "query measurement token from postgres failed", "token", token, "error", err)
+			renderRespFailure(c, constants.RespCodeServerError, "validate measurement token failed", nil)
+			return
+		}
+	}
+
+	switch dataObjectType {
+	case constants.DataObjectTypeParameter:
+		value, err := buildParameterAttributeValue(
+			ctx,
+			field,
+			parameter,
+			func(ctx context.Context, parameter *database.ParameterDataObject) (any, error) {
+				return database.QueryParameterDataObjectValue(ctx, pgClient, parameter)
+			},
+			func(ctx context.Context, attributeName string) (string, error) {
+				return database.QueryParameterAttributeDescription(ctx, pgClient, attributeName)
+			},
+		)
+		if err != nil {
+			if errors.Is(err, common.ErrUnsupportedParameterField) {
+				logger.Warn(ctx, "query unsupported parameter field", "token", token, "field", field, "error", err)
+				renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+				return
+			}
+
+			logger.Error(ctx, "build parameter attribute value failed", "token", token, "field", field, "error", err)
+			renderRespFailure(c, constants.RespCodeServerError, "query parameter attribute failed", nil)
+			return
+		}
+
+		result := dataObjectAttributeQueryResult{
+			Token: token,
+			Field: field,
+			Code:  errcode.ErrProcessSuccess.Code(),
+			Msg:   errcode.ErrProcessSuccess.Msg(),
+			Value: value,
+		}
+		renderRespSuccess(c, constants.RespCodeSuccess, "query parameter attribute success", map[string]any{
+			"attributes": []dataObjectAttributeQueryResult{result},
+		})
+	case constants.DataObjectTypeMeasurement:
+		value, err := buildMeasurementAttributeValue(
+			ctx,
+			field,
+			measurement,
+			measurementComponent,
+			queryMeasurementRealtimeValue,
+		)
+		if err != nil {
+			if errors.Is(err, common.ErrUnsupportedMeasurementField) {
+				logger.Warn(ctx, "query unsupported measurement field", "token", token, "field", field, "error", err)
+				renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+				return
+			}
+
+			logger.Error(ctx, "build measurement attribute value failed", "token", token, "field", field, "error", err)
+			renderRespFailure(c, constants.RespCodeServerError, "query measurement attribute failed", nil)
+			return
+		}
+
+		result := dataObjectAttributeQueryResult{
+			Token: token,
+			Field: field,
+			Code:  errcode.ErrProcessSuccess.Code(),
+			Msg:   errcode.ErrProcessSuccess.Msg(),
+			Value: value,
+		}
+		renderRespSuccess(c, constants.RespCodeSuccess, "query measurement attribute success", map[string]any{
+			"attributes": []dataObjectAttributeQueryResult{result},
+		})
+	default:
+		renderRespFailure(c, constants.RespCodeInvalidParams, "invalid data object type", nil)
+	}
+}
+
+func parseDataObjectAttributeQuery(c *gin.Context) (string, string, error) {
+	token := c.Query("token")
+	if token == "" {
+		return "", "", fmt.Errorf("token is missing from query parameters")
+	}
+
+	field := strings.ToLower(c.Query("field"))
+	if field == "" {
+		field = "value"
+	}
+	return token, field, nil
+}
+
+type measurementValueLoader func(context.Context, orm.JSONMap) (any, error)
+
+type parameterValueLoader func(context.Context, *database.ParameterDataObject) (any, error)
+
+type parameterDescriptionLoader func(context.Context, string) (string, error)
+
+var measurementDataObjectFields = map[string]struct{}{
+	"value":       {},
+	"mode":        {},
+	"meta":        {},
+	"type":        {},
+	"name":        {},
+	"description": {},
+	"id":          {},
+	"size":        {},
+	"data_source": {},
+	"event_plan":  {},
+	"binding":     {},
+}
+
+var parameterDataObjectFields = map[string]struct{}{
+	"value":       {},
+	"name":        {},
+	"meta":        {},
+	"type":        {},
+	"description": {},
+	"id":          {},
+}
+
+type dataObjectAttributeQueryResult struct {
+	Token string `json:"token"`
+	Field string `json:"field"`
+	Code  int    `json:"code"`
+	Msg   string `json:"msg"`
+	Value any    `json:"value"`
+}
+
+func validateDataObjectField(dataObjectType constants.DataObjectType, field string) error {
+	field = strings.ToLower(field)
+	switch dataObjectType {
+	case constants.DataObjectTypeMeasurement:
+		if _, ok := measurementDataObjectFields[field]; ok {
+			return nil
+		}
+		return fmt.Errorf("%w: %s", common.ErrUnsupportedMeasurementField, field)
+	case constants.DataObjectTypeParameter:
+		if _, ok := parameterDataObjectFields[field]; ok {
+			return nil
+		}
+		return fmt.Errorf("%w: %s", common.ErrUnsupportedParameterField, field)
+	default:
+		return fmt.Errorf("invalid data object type %q", dataObjectType)
+	}
+}
+
+func buildParameterAttributeValue(
+	ctx context.Context,
+	field string,
+	parameter *database.ParameterDataObject,
+	loadValue parameterValueLoader,
+	loadDescription parameterDescriptionLoader,
+) (any, error) {
+	if parameter == nil {
+		return nil, fmt.Errorf("parameter data object is nil")
+	}
+
+	component := parameter.Component
+	switch field {
+	case "value":
+		if loadValue == nil {
+			return nil, fmt.Errorf("parameter value loader is nil")
+		}
+		return loadValue(ctx, parameter)
+	case "meta":
+		return "PARAM", nil
+	case "type":
+		return parameter.AttributeType, nil
+	case "name":
+		return strings.Join([]string{
+			component.NSPath,
+			component.Tag,
+			parameter.AttributeGroup,
+			parameter.AttributeName,
+		}, "."), nil
+	case "description":
+		if loadDescription == nil {
+			return nil, fmt.Errorf("parameter description loader is nil")
+		}
+		return loadDescription(ctx, parameter.AttributeName)
+	case "id":
+		return strings.Join([]string{
+			component.GridName,
+			component.ZoneName,
+			component.StationName,
+			component.NSPath,
+			component.Tag,
+			parameter.AttributeGroup,
+			parameter.AttributeName,
+		}, "."), nil
+	default:
+		return nil, fmt.Errorf("%w: %s", common.ErrUnsupportedParameterField, field)
+	}
+}
+
+func buildMeasurementAttributeValue(
+	ctx context.Context,
+	field string,
+	measurement *orm.Measurement,
+	component *orm.Component,
+	loadValue measurementValueLoader,
+) (any, error) {
+	if measurement == nil {
+		return nil, fmt.Errorf("measurement is nil")
+	}
+	if component == nil {
+		return nil, fmt.Errorf("measurement component is nil")
+	}
+
+	switch field {
+	case "value":
+		if loadValue == nil {
+			return nil, fmt.Errorf("measurement value loader is nil")
+		}
+		return loadValue(ctx, measurement.DataSource)
+	case "mode":
+		return measurement.Mode, nil
+	case "meta":
+		return "MEASUREMENT", nil
+	case "type":
+		return model.MeasurementTypeFromDataSource(measurement.DataSource)
+	case "name":
+		// The resolved measurement and component prove that token4.token7 exists.
+		return component.NSPath + "." + measurement.Tag, nil
+	case "description":
+		return measurement.Name, nil
+	case "id":
+		return strings.Join([]string{
+			component.GridName,
+			component.ZoneName,
+			component.StationName,
+			component.NSPath,
+			component.Tag,
+			"bay",
+			measurement.Tag,
+		}, "."), nil
+	case "size":
+		return measurement.Size, nil
+	case "data_source":
+		return measurement.DataSource, nil
+	case "event_plan":
+		return measurement.EventPlan, nil
+	case "binding":
+		return measurement.Binding, nil
+	default:
+		return nil, fmt.Errorf("%w: %s", common.ErrUnsupportedMeasurementField, field)
+	}
+}
+
+func queryMeasurementRealtimeValue(ctx context.Context, dataSource orm.JSONMap) (any, error) {
+	queryKey, err := model.GenerateMeasureIdentifier(dataSource)
+	if err != nil {
+		return nil, fmt.Errorf("generate measurement redis key: %w", err)
+	}
+
+	value, err := diagram.NewRedisClient().QueryLatestMeasurementValue(ctx, queryKey)
+	if err != nil {
+		return nil, fmt.Errorf("query real-time measurement value by key %q: %w", queryKey, err)
+	}
+	return value, nil
+}

handler/data_object_attribute_query_test.go

@@ -0,0 +1,276 @@
+package handler
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"modelRT/common"
+	"modelRT/constants"
+	"modelRT/database"
+	"modelRT/orm"
+
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseDataObjectAttributeQuery(t *testing.T) {
+	const token = "nspath.component.bay.measurement"
+
+	tests := []struct {
+		name      string
+		target    string
+		wantToken string
+		wantField string
+		wantErr   string
+	}{
+		{
+			name:      "reads token and field from query parameters",
+			target:    "/data-object/attribute?token=" + token + "&field=NAME",
+			wantToken: token,
+			wantField: "name",
+		},
+		{
+			name:      "defaults missing field to value",
+			target:    "/data-object/attribute?token=" + token,
+			wantToken: token,
+			wantField: "value",
+		},
+		{
+			name:      "defaults empty field to value",
+			target:    "/data-object/attribute?token=" + token + "&field=",
+			wantToken: token,
+			wantField: "value",
+		},
+		{
+			name:    "rejects missing token",
+			target:  "/data-object/attribute?field=value",
+			wantErr: "token is missing from query parameters",
+		},
+		{
+			name:    "rejects empty token",
+			target:  "/data-object/attribute?token=&field=value",
+			wantErr: "token is missing from query parameters",
+		},
+		{
+			name:    "does not read legacy path parameters",
+			target:  "/data-object/attribute/" + token + "/value",
+			wantErr: "token is missing from query parameters",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, _ := gin.CreateTestContext(httptest.NewRecorder())
+			ctx.Request = httptest.NewRequest(http.MethodGet, tt.target, nil)
+
+			token, field, err := parseDataObjectAttributeQuery(ctx)
+			if tt.wantErr != "" {
+				require.EqualError(t, err, tt.wantErr)
+				assert.Empty(t, token)
+				assert.Empty(t, field)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantToken, token)
+			assert.Equal(t, tt.wantField, field)
+		})
+	}
+}
+
+func TestValidateDataObjectField(t *testing.T) {
+	tests := []struct {
+		name           string
+		token          string
+		dataObjectType constants.DataObjectType
+		field          string
+		wantErr        error
+	}{
+		{name: "bay value", token: "nspath.component.bay.measurement", dataObjectType: constants.DataObjectTypeMeasurement, field: "value"},
+		{name: "bay name", token: "nspath.component.bay.measurement", dataObjectType: constants.DataObjectTypeMeasurement, field: "name"},
+		{name: "bay binding", token: "nspath.component.bay.measurement", dataObjectType: constants.DataObjectTypeMeasurement, field: "binding"},
+		{name: "bay field is case insensitive", token: "nspath.component.bay.measurement", dataObjectType: constants.DataObjectTypeMeasurement, field: "DATA_SOURCE"},
+		{name: "bay unsupported", token: "nspath.component.bay.measurement", dataObjectType: constants.DataObjectTypeMeasurement, field: "unknown", wantErr: common.ErrUnsupportedMeasurementField},
+		{name: "parameter value", token: "nspath.component.rated.attribute", dataObjectType: constants.DataObjectTypeParameter, field: "value"},
+		{name: "parameter name", token: "nspath.component.rated.attribute", dataObjectType: constants.DataObjectTypeParameter, field: "name"},
+		{name: "parameter rejects mode", token: "nspath.component.rated.attribute", dataObjectType: constants.DataObjectTypeParameter, field: "mode", wantErr: common.ErrUnsupportedParameterField},
+		{name: "component name", token: "nspath.component.component.name", dataObjectType: constants.DataObjectTypeParameter, field: "name"},
+		{name: "component rejects mode", token: "nspath.component.component.name", dataObjectType: constants.DataObjectTypeParameter, field: "mode", wantErr: common.ErrUnsupportedParameterField},
+		{name: "parameter rejects size", token: "nspath.component.rated.attribute", dataObjectType: constants.DataObjectTypeParameter, field: "size", wantErr: common.ErrUnsupportedParameterField},
+		{name: "invalid type", token: "token", dataObjectType: constants.DataObjectType("unknown"), field: "value", wantErr: assert.AnError},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateDataObjectField(tt.dataObjectType, tt.field)
+			if tt.wantErr == nil {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			if tt.wantErr != assert.AnError {
+				assert.ErrorIs(t, err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestBuildParameterAttributeValue(t *testing.T) {
+	parameter := &database.ParameterDataObject{
+		Component: orm.Component{
+			GridName:    "grid000",
+			ZoneName:    "zone000",
+			StationName: "station000",
+			NSPath:      "110kV_TV",
+			Tag:         "cable_22",
+		},
+		AttributeGroup: "rated",
+		AttributeName:  "rated_voltage",
+		AttributeType:  "DOUBLE PRECISION",
+	}
+	loader := func(_ context.Context, actual *database.ParameterDataObject) (any, error) {
+		assert.Same(t, parameter, actual)
+		return float64(220), nil
+	}
+	descriptionLoader := func(_ context.Context, attributeName string) (string, error) {
+		assert.Equal(t, "rated_voltage", attributeName)
+		return "额定电压", nil
+	}
+
+	tests := []struct {
+		field    string
+		expected any
+	}{
+		{field: "value", expected: float64(220)},
+		{field: "meta", expected: "PARAM"},
+		{field: "type", expected: "DOUBLE PRECISION"},
+		{field: "name", expected: "110kV_TV.cable_22.rated.rated_voltage"},
+		{field: "description", expected: "额定电压"},
+		{field: "id", expected: "grid000.zone000.station000.110kV_TV.cable_22.rated.rated_voltage"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.field, func(t *testing.T) {
+			actual, err := buildParameterAttributeValue(
+				context.Background(),
+				tt.field,
+				parameter,
+				loader,
+				descriptionLoader,
+			)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, actual)
+		})
+	}
+}
+
+func TestBuildParameterAttributeValueRejectsUnsupportedField(t *testing.T) {
+	_, err := buildParameterAttributeValue(
+		context.Background(),
+		"unknown",
+		&database.ParameterDataObject{},
+		nil,
+		nil,
+	)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, common.ErrUnsupportedParameterField)
+}
+
+func TestBuildMeasurementAttributeValue(t *testing.T) {
+	dataSource := orm.JSONMap{
+		"type": float64(1),
+		"io_address": map[string]any{
+			"channel": "tm1p",
+		},
+	}
+	eventPlan := orm.JSONMap{"enabled": true}
+	binding := orm.JSONMap{"ct": map[string]any{"ratio": float64(2)}}
+	measurement := &orm.Measurement{
+		Tag:        "IA_rms",
+		Name:       "A相电流",
+		Mode:       1,
+		Size:       10,
+		DataSource: dataSource,
+		EventPlan:  eventPlan,
+		Binding:    binding,
+	}
+	component := &orm.Component{
+		GridName:    "grid000",
+		ZoneName:    "zone000",
+		StationName: "station000",
+		NSPath:      "110kV_TV",
+		Tag:         "cable_22",
+	}
+
+	loader := func(_ context.Context, source orm.JSONMap) (any, error) {
+		assert.Equal(t, dataSource, source)
+		return float64(220), nil
+	}
+
+	tests := []struct {
+		field    string
+		expected any
+	}{
+		{field: "value", expected: float64(220)},
+		{field: "mode", expected: int16(1)},
+		{field: "meta", expected: "MEASUREMENT"},
+		{field: "type", expected: "TM"},
+		{field: "name", expected: "110kV_TV.IA_rms"},
+		{field: "description", expected: "A相电流"},
+		{field: "id", expected: "grid000.zone000.station000.110kV_TV.cable_22.bay.IA_rms"},
+		{field: "size", expected: 10},
+		{field: "data_source", expected: dataSource},
+		{field: "event_plan", expected: eventPlan},
+		{field: "binding", expected: binding},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.field, func(t *testing.T) {
+			actual, err := buildMeasurementAttributeValue(context.Background(), tt.field, measurement, component, loader)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, actual)
+		})
+	}
+}
+
+func TestBuildMeasurementAttributeValueRejectsUnsupportedField(t *testing.T) {
+	_, err := buildMeasurementAttributeValue(
+		context.Background(),
+		"unknown",
+		&orm.Measurement{},
+		&orm.Component{},
+		nil,
+	)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, common.ErrUnsupportedMeasurementField)
+}
+
+func TestBuildMeasurementAttributeValueMode(t *testing.T) {
+	component := &orm.Component{}
+	tests := []struct {
+		name     string
+		mode     int16
+		expected int16
+	}{
+		{name: "collected value", mode: 1, expected: 1},
+		{name: "manually assigned value", mode: 0, expected: 0},
+		{name: "other positive mode", mode: 2, expected: 2},
+		{name: "negative mode", mode: -1, expected: -1},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual, err := buildMeasurementAttributeValue(
+				context.Background(),
+				"mode",
+				&orm.Measurement{Mode: tt.mode},
+				component,
+				nil,
+			)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, actual)
+		})
+	}
+}

handler/data_object_attribute_update.go

@@ -0,0 +1,425 @@
+// 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
+}

handler/data_object_attribute_update_test.go

@@ -0,0 +1,474 @@
+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: "automatic mode", field: "mode", value: `1`, expected: constants.MeasurementModeAutomatic},
+		{name: "manual mode", field: "mode", value: `0`, expected: constants.MeasurementModeManual},
+		{name: "boolean mode is rejected", field: "mode", value: `true`, wantError: true},
+		{name: "out of range mode", field: "mode", value: `2`, 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 TestValidateDataObjectAttributeUpdateAcceptsToken4Token7Measurement(t *testing.T) {
+	dataObjectType, field, value, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{
+		Token: "nspath.measurement",
+		Value: json.RawMessage(`15.2`),
+	})
+	require.NoError(t, err)
+	assert.Equal(t, constants.DataObjectTypeMeasurement, dataObjectType)
+	assert.Equal(t, "value", field)
+	assert.Equal(t, float64(15.2), value)
+}
+
+func TestValidateDataObjectAttributeUpdateRequiredFields(t *testing.T) {
+	tests := []struct {
+		name    string
+		request dataObjectAttributeUpdateRequest
+	}{
+		{name: "missing token", request: dataObjectAttributeUpdateRequest{Field: "value", 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 TestValidateDataObjectAttributeUpdateDefaultsEmptyFieldToValue(t *testing.T) {
+	tests := []struct {
+		name      string
+		request   dataObjectAttributeUpdateRequest
+		wantType  constants.DataObjectType
+		wantValue any
+	}{
+		{
+			name: "parameter omitted field",
+			request: dataObjectAttributeUpdateRequest{
+				Token: "nspath.component.rated.attribute",
+				Value: json.RawMessage(`"15.2"`),
+			},
+			wantType:  constants.DataObjectTypeParameter,
+			wantValue: "15.2",
+		},
+		{
+			name: "measurement whitespace field",
+			request: dataObjectAttributeUpdateRequest{
+				Token: "nspath.component.bay.measurement",
+				Field: "   ",
+				Value: json.RawMessage(`15.2`),
+			},
+			wantType:  constants.DataObjectTypeMeasurement,
+			wantValue: float64(15.2),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dataObjectType, field, value, err := validateDataObjectAttributeUpdate(tt.request)
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantType, dataObjectType)
+			assert.Equal(t, "value", field)
+			assert.Equal(t, tt.wantValue, value)
+		})
+	}
+}
+
+func TestMeasurementModeIsAutomatic(t *testing.T) {
+	automatic, err := measurementModeIsAutomatic(constants.MeasurementModeAutomatic)
+	require.NoError(t, err)
+	assert.True(t, automatic)
+
+	automatic, err = measurementModeIsAutomatic(constants.MeasurementModeManual)
+	require.NoError(t, err)
+	assert.False(t, automatic)
+
+	_, err = measurementModeIsAutomatic(-1)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid mode")
+}
+
+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 = \$4`).
+		WithArgs(constants.MeasurementModeAutomatic, sqlmock.AnyArg(), 500, int64(10)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectRollback()
+
+	startCalled := false
+	result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
+		startDataRTFunc: func(_ context.Context, dataSource orm.JSONMap, value *float64) error {
+			startCalled = true
+			assert.Equal(t, float64(1), dataSource["type"])
+			assert.Nil(t, value)
+			return nil
+		},
+	})
+	require.NoError(t, err)
+	assert.True(t, startCalled)
+	assert.Contains(t, result.message, "automatic")
+	require.NoError(t, tx.Rollback().Error)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateMeasurementModeToAutomaticReturnsErrorWhenDataRTStartFails(t *testing.T) {
+	db, mock, closeDB := newDataObjectUpdateTestDB(t)
+	defer closeDB()
+
+	mock.ExpectBegin()
+	tx := db.Begin()
+	require.NoError(t, tx.Error)
+	expectMeasurementResolution(mock, constants.MeasurementModeManual)
+	mock.ExpectExec(`UPDATE "measurement" SET .*"mode"=\$1.*"operations"=.*WHERE id = \$4`).
+		WithArgs(constants.MeasurementModeAutomatic, sqlmock.AnyArg(), 500, int64(10)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectRollback()
+
+	_, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
+		startDataRTFunc: func(context.Context, orm.JSONMap, *float64) error {
+			return fmt.Errorf("dataRT unavailable")
+		},
+	})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "start automatic measurement write")
+	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", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{})
+	require.NoError(t, err)
+	assert.Contains(t, result.message, "already")
+	require.NoError(t, tx.Rollback().Error)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateMeasurementModeToManualWithoutDataOnlyStopsDataRT(t *testing.T) {
+	db, mock, closeDB := newDataObjectUpdateTestDB(t)
+	defer closeDB()
+
+	mock.ExpectBegin()
+	tx := db.Begin()
+	require.NoError(t, tx.Error)
+	expectMeasurementResolution(mock, constants.MeasurementModeAutomatic)
+	mock.ExpectExec(`UPDATE "measurement" SET .*"mode"=\$1.*"operations"=.*WHERE id = \$4`).
+		WithArgs(constants.MeasurementModeManual, sqlmock.AnyArg(), 500, int64(10)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectRollback()
+
+	stopCalled := false
+	replaceCalled := false
+	result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, nil, measurementUpdateDependencies{
+		updateDataRTFunc: func(_ context.Context, dataSource orm.JSONMap, value *float64) error {
+			stopCalled = true
+			assert.Equal(t, float64(1), dataSource["type"])
+			assert.Nil(t, value)
+			return nil
+		},
+		replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64) error {
+			replaceCalled = true
+			return nil
+		},
+	})
+	require.NoError(t, err)
+	assert.True(t, stopCalled)
+	assert.False(t, replaceCalled)
+	assert.Contains(t, result.message, "manual")
+	require.NoError(t, tx.Rollback().Error)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateMeasurementModeToManualReplacesRedisValueWhenDataProvided(t *testing.T) {
+	db, mock, closeDB := newDataObjectUpdateTestDB(t)
+	defer closeDB()
+
+	mock.ExpectBegin()
+	tx := db.Begin()
+	require.NoError(t, tx.Error)
+	expectMeasurementResolution(mock, constants.MeasurementModeAutomatic)
+	mock.ExpectExec(`UPDATE "measurement" SET .*"mode"=\$1.*"operations"=.*WHERE id = \$4`).
+		WithArgs(constants.MeasurementModeManual, sqlmock.AnyArg(), 500, int64(10)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectRollback()
+
+	callOrder := make([]string, 0, 2)
+	result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`0`), measurementUpdateDependencies{
+		updateDataRTFunc: func(_ context.Context, _ orm.JSONMap, value *float64) error {
+			callOrder = append(callOrder, "stop-dataRT")
+			assert.Nil(t, value)
+			return nil
+		},
+		replaceRedisValueFunc: func(_ context.Context, measurement *orm.Measurement, value float64) error {
+			callOrder = append(callOrder, "replace-redis")
+			assert.Equal(t, int64(10), measurement.ID)
+			assert.Equal(t, float64(0), value)
+			return nil
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"stop-dataRT", "replace-redis"}, callOrder)
+	assert.Contains(t, result.message, "manual")
+	require.NoError(t, tx.Rollback().Error)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateMeasurementModeToManualDoesNotTouchRedisWhenDataRTStopFails(t *testing.T) {
+	db, mock, closeDB := newDataObjectUpdateTestDB(t)
+	defer closeDB()
+
+	mock.ExpectBegin()
+	tx := db.Begin()
+	require.NoError(t, tx.Error)
+	expectMeasurementResolution(mock, constants.MeasurementModeAutomatic)
+	mock.ExpectExec(`UPDATE "measurement" SET .*"mode"=\$1.*"operations"=.*WHERE id = \$4`).
+		WithArgs(constants.MeasurementModeManual, sqlmock.AnyArg(), 500, int64(10)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectRollback()
+
+	replaceCalled := false
+	_, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`15.2`), measurementUpdateDependencies{
+		updateDataRTFunc: func(_ context.Context, _ orm.JSONMap, value *float64) error {
+			assert.Nil(t, value)
+			return fmt.Errorf("dataRT unavailable")
+		},
+		replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64) error {
+			replaceCalled = true
+			return nil
+		},
+	})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "stop automatic measurement write")
+	assert.False(t, replaceCalled)
+	require.NoError(t, tx.Rollback().Error)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestParseOptionalMeasurementModeData(t *testing.T) {
+	for _, raw := range []json.RawMessage{nil, json.RawMessage(`null`)} {
+		value, err := parseOptionalMeasurementModeData(raw)
+		require.NoError(t, err)
+		assert.Nil(t, value)
+	}
+
+	value, err := parseOptionalMeasurementModeData(json.RawMessage(`"15.2"`))
+	require.NoError(t, err)
+	require.NotNil(t, value)
+	assert.Equal(t, 15.2, *value)
+
+	_, err = parseOptionalMeasurementModeData(json.RawMessage(`"invalid"`))
+	require.Error(t, err)
+}
+
+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, measurementUpdateDependencies{})
+	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 = \$3`).
+		WithArgs(sqlmock.AnyArg(), 500, 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
+		require.NotNil(t, value)
+		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), nil, measurementUpdateDependencies{
+		writeManualValueFunc: writer,
+		updateDataRTFunc:     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), nil, measurementUpdateDependencies{
+		writeManualValueFunc: writer,
+	})
+	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(`SELECT "id","mode","data_source" 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))
+}

handler/data_object_recommend.go

@@ -0,0 +1,125 @@
+// Package handler provides HTTP handlers for various endpoints.
+package handler
+
+import (
+	"errors"
+	"strings"
+
+	"modelRT/constants"
+	"modelRT/logger"
+	"modelRT/model"
+	"modelRT/network"
+
+	"github.com/gin-gonic/gin"
+)
+
+// DataObjectRecommendHandler define data-object recommend API
+// @Summary 测量点推荐（搜索框自动补全）
+// @Description 根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。
+// @Tags DataObject Recommend
+// @Accept json
+// @Produce json
+// @Param input query string true "推荐关键词,例如 'grid1' 或 'grid1.'" Example("grid1")
+// @Success 200 {object} network.SuccessResponse{payload=network.DataObjectRecommendPayload} "返回推荐列表成功"
+//
+//	@Example 200 {
+//	    "code": 200,
+//	    "msg": "success",
+//	    "payload": {
+//	        "input": "grid1.zone1.station1.ns1.tag1.bay.",
+//	        "offset": 21,
+//	        "recommended_list": [
+//	            "I11_A_rms",
+//	            "I11_B_rms.",
+//	            "I11_C_rms.",
+//	        ]
+//	    }
+//	}
+//
+// @Failure 400 {object} network.FailureResponse "返回推荐列表失败"
+//
+//	@Example 400 {
+//	    "code": 400,
+//	    "msg": "failed to get recommend data from redis",
+//	}
+//
+// @Router /data-object/recommend [get]
+func DataObjectRecommendHandler(c *gin.Context) {
+	ctx := c.Request.Context()
+	var request network.DataObjectRecommendRequest
+
+	if err := c.ShouldBindQuery(&request); err != nil {
+		logger.Error(ctx, "failed to bind data object recommend request", "error", err)
+		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
+		return
+	}
+	if err := validateDataObjectRecommendInput(request.Input); err != nil {
+		logger.Warn(ctx, "invalid data object recommend input", "input", request.Input, "error", err)
+		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), map[string]any{
+			"input": request.Input,
+		})
+		return
+	}
+	recommendResults := model.RedisSearchRecommend(ctx, request.Input)
+	payload := network.DataObjectRecommendPayload{
+		Input:           request.Input,
+		RecommendedList: make([]string, 0),
+	}
+	seen := make(map[string]struct{})
+	orderedResults := orderedRecommendResults(recommendResults)
+
+	for _, recommendResult := range orderedResults {
+		if recommendResult.Err != nil {
+			err := recommendResult.Err
+			logger.Error(ctx, "failed to get recommend data from redis", "input", request.Input, "error", err)
+			renderRespFailure(c, constants.RespCodeServerError, err.Error(), map[string]any{
+				"input": request.Input,
+			})
+			return
+		}
+
+		if recommendResult.Offset > payload.Offset {
+			payload.Offset = recommendResult.Offset
+		}
+	}
+
+	for _, recommendResult := range orderedResults {
+		for _, recommend := range recommendResult.QueryDatas {
+			if _, exists := seen[recommend]; !exists {
+				seen[recommend] = struct{}{}
+				payload.RecommendedList = append(payload.RecommendedList, recommend)
+			}
+		}
+	}
+
+	renderRespSuccess(c, constants.RespCodeSuccess, "success", payload)
+}
+
+func orderedRecommendResults(recommendResults map[string]model.SearchResult) []model.SearchResult {
+	orderedTypes := []string{
+		constants.CompNSPathRecommendHierarchyType.String(),
+		constants.GridRecommendHierarchyType.String(),
+		constants.ZoneRecommendHierarchyType.String(),
+		constants.StationRecommendHierarchyType.String(),
+		constants.CompTagRecommendHierarchyType.String(),
+		constants.ConfigRecommendHierarchyType.String(),
+		constants.MeasTagRecommendHierarchyType.String(),
+	}
+
+	results := make([]model.SearchResult, 0, len(recommendResults))
+	for _, recommendType := range orderedTypes {
+		result, ok := recommendResults[recommendType]
+		if !ok {
+			continue
+		}
+		results = append(results, result)
+	}
+	return results
+}
+
+func validateDataObjectRecommendInput(input string) error {
+	if strings.Contains(input, "..") {
+		return errors.New("input contains continuous dots")
+	}
+	return nil
+}

handler/diagram_node_link.go

@@ -41,11 +41,12 @@ var linkSetConfigs = map[int]linkSetConfig{
 
 // DiagramNodeLinkHandler defines the diagram node link process api
 func DiagramNodeLinkHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.DiagramNodeLinkRequest
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -54,7 +55,7 @@ func DiagramNodeLinkHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal diagram node process request", "error", err)
+		logger.Error(ctx, "failed to unmarshal diagram node process request", "error", err)
 
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -68,9 +69,9 @@ func DiagramNodeLinkHandler(c *gin.Context) {
 	nodeID := request.NodeID
 	nodeLevel := request.NodeLevel
 	action := request.Action
-	prevNodeInfo, currNodeInfo, err := database.QueryNodeInfoByID(c, pgClient, nodeID, nodeLevel)
+	prevNodeInfo, currNodeInfo, err := database.QueryNodeInfoByID(ctx, pgClient, nodeID, nodeLevel)
 	if err != nil {
-		logger.Error(c, "failed to query diagram node info by nodeID and level from postgres", "node_id", nodeID, "level", nodeLevel, "error", err)
+		logger.Error(ctx, "failed to query diagram node info by nodeID and level from postgres", "node_id", nodeID, "level", nodeLevel, "error", err)
 
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -84,8 +85,13 @@ func DiagramNodeLinkHandler(c *gin.Context) {
 		return
 	}
 
-	prevLinkSet, currLinkSet := generateLinkSet(c, nodeLevel, prevNodeInfo)
-	err = processLinkSetData(c, action, nodeLevel, prevLinkSet, currLinkSet, prevNodeInfo, currNodeInfo)
+	prevLinkSet, currLinkSet, err := generateLinkSet(ctx, nodeLevel, prevNodeInfo)
+	if err != nil {
+		logger.Error(ctx, "create diagram link redis sets failed", "node_id", nodeID, "level", nodeLevel, "error", err)
+		c.JSON(http.StatusOK, network.FailureResponse{Code: http.StatusInternalServerError, Msg: err.Error()})
+		return
+	}
+	err = processLinkSetData(ctx, action, nodeLevel, prevLinkSet, currLinkSet, prevNodeInfo, currNodeInfo)
 	if err != nil {
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -99,7 +105,7 @@ func DiagramNodeLinkHandler(c *gin.Context) {
 		return
 	}
 
-	logger.Info(c, "process diagram node link success", "node_id", nodeID, "level", nodeLevel, "action", request.Action)
+	logger.Info(ctx, "process diagram node link success", "node_id", nodeID, "level", nodeLevel, "action", request.Action)
 
 	c.JSON(http.StatusOK, network.SuccessResponse{
 		Code: http.StatusOK,
@@ -112,21 +118,27 @@ func DiagramNodeLinkHandler(c *gin.Context) {
 	})
 }
 
-func generateLinkSet(ctx context.Context, level int, prevNodeInfo orm.CircuitDiagramNodeInterface) (*diagram.RedisSet, *diagram.RedisSet) {
+func generateLinkSet(ctx context.Context, level int, prevNodeInfo orm.CircuitDiagramNodeInterface) (*diagram.RedisSet, *diagram.RedisSet, error) {
 	config, ok := linkSetConfigs[level]
 	// level not supported
 	if !ok {
-		return nil, nil
+		return nil, nil, nil
 	}
 
-	currLinkSet := diagram.NewRedisSet(ctx, config.CurrKey, 0, false)
+	currLinkSet, err := diagram.NewRedisSet(ctx, config.CurrKey, 0, false)
+	if err != nil {
+		return nil, nil, err
+	}
 	if config.PrevIsNil {
-		return nil, currLinkSet
+		return nil, currLinkSet, nil
 	}
 
 	prevLinkSetKey := fmt.Sprintf(config.PrevKeyTemplate, prevNodeInfo.GetTagName())
-	prevLinkSet := diagram.NewRedisSet(ctx, prevLinkSetKey, 0, false)
-	return prevLinkSet, currLinkSet
+	prevLinkSet, err := diagram.NewRedisSet(ctx, prevLinkSetKey, 0, false)
+	if err != nil {
+		return nil, nil, err
+	}
+	return prevLinkSet, currLinkSet, nil
 }
 
 func processLinkSetData(ctx context.Context, action string, level int, prevLinkSet, currLinkSet *diagram.RedisSet, prevNodeInfo, currNodeInfo orm.CircuitDiagramNodeInterface) error {

handler/history_data_query.go

@@ -16,11 +16,12 @@ import (
 
 // QueryHistoryDataHandler define query history data process API
 func QueryHistoryDataHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	token := c.Query("token")
 	beginStr := c.Query("begin")
 	begin, err := strconv.Atoi(beginStr)
 	if err != nil {
-		logger.Error(c, "convert begin param from string to int failed", "error", err)
+		logger.Error(ctx, "convert begin param from string to int failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -31,7 +32,7 @@ func QueryHistoryDataHandler(c *gin.Context) {
 	endStr := c.Query("end")
 	end, err := strconv.Atoi(endStr)
 	if err != nil {
-		logger.Error(c, "convert end param from string to int failed", "error", err)
+		logger.Error(ctx, "convert end param from string to int failed", "error", err)
 
 		resp := network.FailureResponse{
 			Code: http.StatusBadRequest,

handler/measurement_load.go

@@ -15,13 +15,14 @@ import (
 
 // MeasurementGetHandler define measurement query API
 func MeasurementGetHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.MeasurementGetRequest
 
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
 
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -30,7 +31,7 @@ func MeasurementGetHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal measurement get request", "error", err)
+		logger.Error(ctx, "failed to unmarshal measurement get request", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -38,10 +39,15 @@ func MeasurementGetHandler(c *gin.Context) {
 		return
 	}
 
-	zset := diagram.NewRedisZSet(c, request.MeasurementToken, 0, false)
+	zset, err := diagram.NewRedisZSet(ctx, request.MeasurementToken, 0, false)
+	if err != nil {
+		logger.Error(ctx, "failed to create measurement redis zset", "measurement_token", request.MeasurementToken, "error", err)
+		c.JSON(http.StatusOK, network.FailureResponse{Code: http.StatusInternalServerError, Msg: err.Error()})
+		return
+	}
 	points, err := zset.ZRANGE(request.MeasurementToken, 0, -1)
 	if err != nil {
-		logger.Error(c, "failed to get measurement data from redis", "measurement_token", request.MeasurementToken, "error", err)
+		logger.Error(ctx, "failed to get measurement data from redis", "measurement_token", request.MeasurementToken, "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusInternalServerError,
 			Msg:  err.Error(),
@@ -54,9 +60,9 @@ func MeasurementGetHandler(c *gin.Context) {
 	}
 
 	pgClient := database.GetPostgresDBClient()
-	measurementInfo, err := database.QueryMeasurementByID(c, pgClient, request.MeasurementID)
+	measurementInfo, err := database.QueryMeasurementByID(ctx, pgClient, request.MeasurementID)
 	if err != nil {
-		logger.Error(c, "failed to query measurement by id", "measurement_id", request.MeasurementID, "error", err)
+		logger.Error(ctx, "failed to query measurement by id", "measurement_id", request.MeasurementID, "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),

handler/measurement_recommend.go

@@ -1,121 +0,0 @@
-// Package handler provides HTTP handlers for various endpoints.
-package handler
-
-import (
-	"modelRT/logger"
-	"modelRT/model"
-	"modelRT/network"
-	"modelRT/util"
-	"net/http"
-
-	"github.com/gin-gonic/gin"
-)
-
-// MeasurementRecommendHandler define measurement recommend API
-// @Summary 测量点推荐（搜索框自动补全）
-// @Description 根据用户输入的字符串，从 Redis 中查询可能的测量点或结构路径，并提供推荐列表。
-// @Tags Measurement Recommend
-// @Accept json
-// @Produce json
-// @Param input query string true "推荐关键词,例如 'grid1' 或 'grid1.'" Example("grid1")
-// @Success 200 {object} network.SuccessResponse{payload=network.MeasurementRecommendPayload} "返回推荐列表成功"
-//
-//	@Example 200 {
-//	    "code": 200,
-//	    "msg": "success",
-//	    "payload": {
-//	        "input": "grid1.zone1.station1.ns1.tag1.bay.",
-//	        "offset": 21,
-//	        "recommended_list": [
-//	            "I11_A_rms",
-//	            "I11_B_rms.",
-//	            "I11_C_rms.",
-//	        ]
-//	    }
-//	}
-//
-// @Failure 400 {object} network.FailureResponse "返回推荐列表失败"
-//
-//	@Example 400 {
-//	    "code": 400,
-//	    "msg": "failed to get recommend data from redis",
-//	}
-//
-// @Router /measurement/recommend [get]
-func MeasurementRecommendHandler(c *gin.Context) {
-	var request network.MeasurementRecommendRequest
-
-	if err := c.ShouldBindQuery(&request); err != nil {
-		logger.Error(c, "failed to bind measurement recommend request", "error", err)
-		c.JSON(http.StatusOK, network.FailureResponse{
-			Code: http.StatusBadRequest,
-			Msg:  err.Error(),
-		})
-		return
-	}
-
-	recommendResults := model.RedisSearchRecommend(c, request.Input)
-	payloads := make([]network.MeasurementRecommendPayload, 0, len(recommendResults))
-	for _, recommendResult := range recommendResults {
-		if recommendResult.Err != nil {
-			err := recommendResult.Err
-			logger.Error(c, "failed to get recommend data from redis", "input", request.Input, "error", err)
-			c.JSON(http.StatusOK, network.FailureResponse{
-				Code: http.StatusInternalServerError,
-				Msg:  err.Error(),
-				Payload: map[string]any{
-					"input": request.Input,
-				},
-			})
-			return
-		}
-
-		var finalOffset int
-		recommends := recommendResult.QueryDatas
-		if recommendResult.IsFuzzy {
-			var maxOffset int
-			for index, recommend := range recommends {
-				offset := util.GetLongestCommonPrefixLength(request.Input, recommend)
-				if index == 0 || offset > maxOffset {
-					maxOffset = offset
-				}
-			}
-			finalOffset = maxOffset
-		} else {
-			var minOffset int
-			for index, recommend := range recommends {
-				offset := util.GetLongestCommonPrefixLength(request.Input, recommend)
-				if index == 0 || offset < minOffset {
-					minOffset = offset
-				}
-			}
-			finalOffset = minOffset
-		}
-
-		resultRecommends := make([]string, 0, len(recommends))
-		seen := make(map[string]struct{})
-
-		for _, recommend := range recommends {
-			recommendTerm := recommend[finalOffset:]
-			if len(recommendTerm) != 0 {
-				if _, exists := seen[recommendTerm]; !exists {
-					seen[recommendTerm] = struct{}{}
-					resultRecommends = append(resultRecommends, recommendTerm)
-				}
-			}
-		}
-
-		payloads = append(payloads, network.MeasurementRecommendPayload{
-			Input:           request.Input,
-			Offset:          finalOffset,
-			RecommendType:   recommendResult.RecommendType.String(),
-			RecommendedList: resultRecommends,
-		})
-	}
-
-	c.JSON(http.StatusOK, network.SuccessResponse{
-		Code:    http.StatusOK,
-		Msg:     "success",
-		Payload: &payloads,
-	})
-}

handler/measurement_recommend_test.go

@@ -0,0 +1,44 @@
+package handler
+
+import "testing"
+
+func TestValidateMeasurementRecommendInput(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		valid bool
+	}{
+		{
+			name:  "continuous dots",
+			input: "G..",
+			valid: false,
+		},
+		{
+			name:  "single separator",
+			input: "G.zone",
+			valid: true,
+		},
+		{
+			name:  "trailing dot",
+			input: "G.",
+			valid: true,
+		},
+		{
+			name:  "empty input",
+			input: "",
+			valid: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateDataObjectRecommendInput(tt.input)
+			if tt.valid && err != nil {
+				t.Fatalf("expected valid input, got error %v", err)
+			}
+			if !tt.valid && err == nil {
+				t.Fatalf("expected invalid input")
+			}
+		})
+	}
+}

handler/mesurement_link.go

@@ -18,11 +18,12 @@ import (
 
 // MeasurementLinkHandler defines the measurement link process api
 func MeasurementLinkHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.MeasurementLinkRequest
 	clientToken := c.GetString("client_token")
 	if clientToken == "" {
 		err := common.ErrGetClientToken
-		logger.Error(c, "failed to get client token from context", "error", err)
+		logger.Error(ctx, "failed to get client token from context", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -31,7 +32,7 @@ func MeasurementLinkHandler(c *gin.Context) {
 	}
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal measurement process request", "error", err)
+		logger.Error(ctx, "failed to unmarshal measurement process request", "error", err)
 
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -44,9 +45,9 @@ func MeasurementLinkHandler(c *gin.Context) {
 	pgClient := database.GetPostgresDBClient()
 	measurementID := request.MeasurementID
 	action := request.Action
-	measurementInfo, err := database.QueryMeasurementByID(c, pgClient, measurementID)
+	measurementInfo, err := database.QueryMeasurementByID(ctx, pgClient, measurementID)
 	if err != nil {
-		logger.Error(c, "failed to query measurement info by measurement id from postgres", "meauserement_id", measurementID, "error", err)
+		logger.Error(ctx, "failed to query measurement info by measurement id from postgres", "meauserement_id", measurementID, "error", err)
 
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -59,9 +60,9 @@ func MeasurementLinkHandler(c *gin.Context) {
 		return
 	}
 
-	componentInfo, err := database.QueryComponentByUUID(c, pgClient, measurementInfo.ComponentUUID)
+	componentInfo, err := database.QueryComponentByUUID(ctx, pgClient, measurementInfo.ComponentUUID)
 	if err != nil {
-		logger.Error(c, "failed to query component info by component uuid from postgres", "component_uuid", measurementInfo.ComponentUUID, "error", err)
+		logger.Error(ctx, "failed to query component info by component uuid from postgres", "component_uuid", measurementInfo.ComponentUUID, "error", err)
 
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
@@ -74,9 +75,19 @@ func MeasurementLinkHandler(c *gin.Context) {
 		return
 	}
 
-	allMeasSet := diagram.NewRedisSet(c, constants.RedisAllMeasTagSetKey, 0, false)
+	allMeasSet, err := diagram.NewRedisSet(ctx, constants.RedisAllMeasTagSetKey, 0, false)
+	if err != nil {
+		logger.Error(ctx, "create all-measurement redis set failed", "error", err)
+		c.JSON(http.StatusOK, network.FailureResponse{Code: http.StatusInternalServerError, Msg: err.Error()})
+		return
+	}
 	compMeasLinkKey := fmt.Sprintf(constants.RedisSpecCompTagMeasSetKey, componentInfo.Tag)
-	compMeasLinkSet := diagram.NewRedisSet(c, compMeasLinkKey, 0, false)
+	compMeasLinkSet, err := diagram.NewRedisSet(ctx, compMeasLinkKey, 0, false)
+	if err != nil {
+		logger.Error(ctx, "create component-measurement redis set failed", "set_key", compMeasLinkKey, "error", err)
+		c.JSON(http.StatusOK, network.FailureResponse{Code: http.StatusInternalServerError, Msg: err.Error()})
+		return
+	}
 
 	switch action {
 	case constants.SearchLinkAddAction:
@@ -84,18 +95,18 @@ func MeasurementLinkHandler(c *gin.Context) {
 		err2 := compMeasLinkSet.SADD(measurementInfo.Tag)
 		err = processActionError(err1, err2, action)
 		if err != nil {
-			logger.Error(c, "add measurement link process operation failed", "measurement_id", measurementID, "action", action, "error", err)
+			logger.Error(ctx, "add measurement link process operation failed", "measurement_id", measurementID, "action", action, "error", err)
 		}
 	case constants.SearchLinkDelAction:
 		err1 := allMeasSet.SREM(measurementInfo.Tag)
 		err2 := compMeasLinkSet.SREM(measurementInfo.Tag)
 		err = processActionError(err1, err2, action)
 		if err != nil {
-			logger.Error(c, "del measurement link process operation failed", "measurement_id", measurementID, "action", action, "error", err)
+			logger.Error(ctx, "del measurement link process operation failed", "measurement_id", measurementID, "action", action, "error", err)
 		}
 	default:
 		err = common.ErrUnsupportedLinkAction
-		logger.Error(c, "unsupport measurement link process action", "measurement_id", measurementID, "action", action, "error", err)
+		logger.Error(ctx, "unsupport measurement link process action", "measurement_id", measurementID, "action", action, "error", err)
 	}
 
 	if err != nil {
@@ -110,7 +121,7 @@ func MeasurementLinkHandler(c *gin.Context) {
 		return
 	}
 
-	logger.Info(c, "process measurement link success", "measurement_id", measurementID, "action", request.Action)
+	logger.Info(ctx, "process measurement link success", "measurement_id", measurementID, "action", request.Action)
 
 	c.JSON(http.StatusOK, network.SuccessResponse{
 		Code: http.StatusOK,

handler/real_time_data_pull.go

@@ -35,27 +35,28 @@ var pullUpgrader = websocket.Upgrader{
 // @Tags RealTime Component Websocket
 // @Router /monitors/data/realtime/stream/:clientID [get]
 func PullRealTimeDataHandler(c *gin.Context) {
+	requestCtx := c.Request.Context()
 	clientID := c.Param("clientID")
 	if clientID == "" {
 		err := fmt.Errorf("clientID is missing from the path")
-		logger.Error(c, "query clientID from path failed", "error", err, "url", c.Request.RequestURI)
+		logger.Error(requestCtx, "query clientID from path failed", "error", err, "url", c.Request.RequestURI)
 		renderWSRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
 		return
 	}
 
 	conn, err := pullUpgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
-		logger.Error(c, "upgrade http protocol to websocket protocol failed", "error", err)
+		logger.Error(requestCtx, "upgrade http protocol to websocket protocol failed", "error", err)
 		renderWSRespFailure(c, constants.RespCodeServerError, err.Error(), nil)
 		return
 	}
 	defer conn.Close()
 
-	ctx, cancel := context.WithCancel(c.Request.Context())
+	ctx, cancel := context.WithCancel(requestCtx)
 	defer cancel()
 
 	conn.SetCloseHandler(func(code int, text string) error {
-		logger.Info(c.Request.Context(), "websocket processor shutdown trigger",
+		logger.Info(requestCtx, "websocket processor shutdown trigger",
 			"clientID", clientID, "code", code, "reason", text)
 
 		// call cancel to notify other goroutines to stop working

handler/real_time_data_query.go

@@ -60,10 +60,11 @@ var wsUpgrader = websocket.Upgrader{
 //
 // @Router /data/realtime [get]
 func QueryRealTimeDataHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.RealTimeQueryRequest
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal real time query request", "error", err)
+		logger.Error(ctx, "failed to unmarshal real time query request", "error", err)
 		c.JSON(http.StatusOK, network.FailureResponse{
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
@@ -73,7 +74,7 @@ func QueryRealTimeDataHandler(c *gin.Context) {
 
 	conn, err := wsUpgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
-		logger.Error(c, "upgrade http protocol to websocket protocal failed", "error", err)
+		logger.Error(ctx, "upgrade http protocol to websocket protocal failed", "error", err)
 		return
 	}
 	defer conn.Close()
@@ -87,29 +88,29 @@ func QueryRealTimeDataHandler(c *gin.Context) {
 		case data := <-transportChannel:
 			respByte, err := jsoniter.Marshal(data)
 			if err != nil {
-				logger.Error(c, "marshal real time data to bytes failed", "error", err)
+				logger.Error(ctx, "marshal real time data to bytes failed", "error", err)
 				continue
 			}
 
 			err = conn.WriteMessage(websocket.TextMessage, respByte)
 			if err != nil {
-				logger.Error(c, "write message to websocket connection failed", "error", err)
+				logger.Error(ctx, "write message to websocket connection failed", "error", err)
 				continue
 			}
 		case <-closeChannel:
-			logger.Info(c, "data receiving goroutine has been closed")
+			logger.Info(ctx, "data receiving goroutine has been closed")
 			// TODO 优化时间控制
 			deadline := time.Now().Add(5 * time.Second)
 			err := conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "the session ended normally"), deadline)
 			if err != nil {
-				logger.Error(c, "sending close control message failed", "error", err)
+				logger.Error(ctx, "sending close control message failed", "error", err)
 			}
 			// gracefully close session processing
 			err = conn.Close()
 			if err != nil {
-				logger.Error(c, "websocket conn closed failed", "error", err)
+				logger.Error(ctx, "websocket conn closed failed", "error", err)
 			}
-			logger.Info(c, "websocket connection closed successfully.")
+			logger.Info(ctx, "websocket connection closed successfully.")
 		}
 	}
 }

handler/real_time_data_receive.go

@@ -17,9 +17,10 @@ var upgrader = websocket.Upgrader{
 
 // RealTimeDataReceivehandler define real time data receive and process API
 func RealTimeDataReceivehandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
-		logger.Error(c, "upgrade http protocol to websocket protocal failed", "error", err)
+		logger.Error(ctx, "upgrade http protocol to websocket protocal failed", "error", err)
 		return
 	}
 	defer conn.Close()
@@ -27,17 +28,17 @@ func RealTimeDataReceivehandler(c *gin.Context) {
 	for {
 		messageType, p, err := conn.ReadMessage()
 		if err != nil {
-			logger.Error(c, "read message from websocket connection failed", "error", err)
+			logger.Error(ctx, "read message from websocket connection failed", "error", err)
 
 			respByte := processResponse(-1, "read message from websocket connection failed", nil)
 			if len(respByte) == 0 {
-				logger.Error(c, "process message from byte failed", "error", err)
+				logger.Error(ctx, "process message from byte failed", "error", err)
 				continue
 			}
 
 			err = conn.WriteMessage(messageType, respByte)
 			if err != nil {
-				logger.Error(c, "write message to websocket connection failed", "error", err)
+				logger.Error(ctx, "write message to websocket connection failed", "error", err)
 				continue
 			}
 			continue
@@ -46,17 +47,17 @@ func RealTimeDataReceivehandler(c *gin.Context) {
 		var request network.RealTimeDataReceiveRequest
 		err = jsoniter.Unmarshal([]byte(p), &request)
 		if err != nil {
-			logger.Error(c, "unmarshal message from byte failed", "error", err)
+			logger.Error(ctx, "unmarshal message from byte failed", "error", err)
 
 			respByte := processResponse(-1, "unmarshal message from byte failed", nil)
 			if len(respByte) == 0 {
-				logger.Error(c, "process message from byte failed", "error", err)
+				logger.Error(ctx, "process message from byte failed", "error", err)
 				continue
 			}
 
 			err = conn.WriteMessage(messageType, respByte)
 			if err != nil {
-				logger.Error(c, "write message to websocket connection failed", "error", err)
+				logger.Error(ctx, "write message to websocket connection failed", "error", err)
 				continue
 			}
 			continue
@@ -70,13 +71,13 @@ func RealTimeDataReceivehandler(c *gin.Context) {
 		}
 		respByte := processResponse(0, "success", payload)
 		if len(respByte) == 0 {
-			logger.Error(c, "process message from byte failed", "error", err)
+			logger.Error(ctx, "process message from byte failed", "error", err)
 			continue
 		}
 
 		err = conn.WriteMessage(messageType, respByte)
 		if err != nil {
-			logger.Error(c, "write message to websocket connection failed", "error", err)
+			logger.Error(ctx, "write message to websocket connection failed", "error", err)
 			continue
 		}
 	}

handler/real_time_data_subscription.go

@@ -77,12 +77,13 @@ func init() {
 //
 // @Router /monitors/data/subscriptions [post]
 func RealTimeSubHandler(c *gin.Context) {
+	ctx := c.Request.Context()
 	var request network.RealTimeSubRequest
 	var subAction string
 	var clientID string
 
 	if err := c.ShouldBindJSON(&request); err != nil {
-		logger.Error(c, "failed to unmarshal real time query request", "error", err)
+		logger.Error(ctx, "failed to unmarshal real time query request", "error", err)
 		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
 		return
 	}
@@ -91,7 +92,7 @@ func RealTimeSubHandler(c *gin.Context) {
 		subAction = request.Action
 		id, err := uuid.NewV4()
 		if err != nil {
-			logger.Error(c, "failed to generate client id", "error", err)
+			logger.Error(ctx, "failed to generate client id", "error", err)
 			renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
 			return
 		}
@@ -114,9 +115,9 @@ func RealTimeSubHandler(c *gin.Context) {
 
 	switch subAction {
 	case constants.SubStartAction:
-		results, err := globalSubState.CreateConfig(c, tx, clientID, request.Measurements)
+		results, err := globalSubState.CreateConfig(ctx, tx, clientID, request.Measurements)
 		if err != nil {
-			logger.Error(c, "create real time data subscription config failed", "error", err)
+			logger.Error(ctx, "create real time data subscription config failed", "error", err)
 			renderRespFailure(c, constants.RespCodeFailed, err.Error(), network.RealTimeSubPayload{
 				ClientID:      clientID,
 				TargetResults: results,
@@ -130,9 +131,9 @@ func RealTimeSubHandler(c *gin.Context) {
 		})
 		return
 	case constants.SubStopAction:
-		results, err := globalSubState.RemoveTargets(c, clientID, request.Measurements)
+		results, err := globalSubState.RemoveTargets(ctx, clientID, request.Measurements)
 		if err != nil {
-			logger.Error(c, "remove target to real time data subscription config failed", "error", err)
+			logger.Error(ctx, "remove target to real time data subscription config failed", "error", err)
 			renderRespFailure(c, constants.RespCodeFailed, err.Error(), network.RealTimeSubPayload{
 				ClientID:      clientID,
 				TargetResults: results,
@@ -146,9 +147,9 @@ func RealTimeSubHandler(c *gin.Context) {
 		})
 		return
 	case constants.SubAppendAction:
-		results, err := globalSubState.AppendTargets(c, tx, clientID, request.Measurements)
+		results, err := globalSubState.AppendTargets(ctx, tx, clientID, request.Measurements)
 		if err != nil {
-			logger.Error(c, "append target to real time data subscription config failed", "error", err)
+			logger.Error(ctx, "append target to real time data subscription config failed", "error", err)
 			renderRespFailure(c, constants.RespCodeFailed, err.Error(), network.RealTimeSubPayload{
 				ClientID:      clientID,
 				TargetResults: results,
@@ -162,9 +163,9 @@ func RealTimeSubHandler(c *gin.Context) {
 		})
 		return
 	case constants.SubUpdateAction:
-		results, err := globalSubState.UpdateTargets(c, tx, clientID, request.Measurements)
+		results, err := globalSubState.UpdateTargets(ctx, tx, clientID, request.Measurements)
 		if err != nil {
-			logger.Error(c, "update target to real time data subscription config failed", "error", err)
+			logger.Error(ctx, "update target to real time data subscription config failed", "error", err)
 			renderRespFailure(c, constants.RespCodeFailed, err.Error(), network.RealTimeSubPayload{
 				ClientID:      clientID,
 				TargetResults: results,
@@ -179,7 +180,7 @@ func RealTimeSubHandler(c *gin.Context) {
 		return
 	default:
 		err := fmt.Errorf("%w: request action is %s", common.ErrUnsupportedSubAction, request.Action)
-		logger.Error(c, "unsupported action of real time data subscription request", "error", err)
+		logger.Error(ctx, "unsupported action of real time data subscription request", "error", err)
 		requestTargetsCount := processRealTimeRequestCount(request.Measurements)
 		results := processRealTimeRequestTargets(request.Measurements, requestTargetsCount, constants.CodeUnsupportSubOperation, err)
 		renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), network.RealTimeSubPayload{

logger/caller_test.go

@@ -0,0 +1,50 @@
+package logger_test
+
+import (
+	"context"
+	"io"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"modelRT/config"
+	"modelRT/constants"
+	"modelRT/logger"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCallerPointsToBusinessCode(t *testing.T) {
+	reader, writer, err := os.Pipe()
+	require.NoError(t, err)
+	originalStdout := os.Stdout
+	os.Stdout = writer
+	t.Cleanup(func() {
+		os.Stdout = originalStdout
+		_ = reader.Close()
+		_ = writer.Close()
+	})
+
+	logger.InitLoggerInstance(config.LoggerConfig{
+		Mode:  constants.DevelopmentLogMode,
+		Level: "info",
+	})
+
+	logger.Info(context.Background(), "facade caller test")
+	logger.NewGormLogger().Trace(context.Background(), time.Now(), func() (string, int64) {
+		return "SELECT 1", 1
+	}, nil)
+	require.NoError(t, writer.Close())
+
+	outputBytes, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	output := string(outputBytes)
+
+	assert.NotContains(t, output, "logger/facede.go")
+	assert.NotContains(t, output, `"func":"modelRT/logger.Info"`)
+	assert.NotContains(t, output, "gorm.io/gorm")
+	assert.Contains(t, output, "caller_test.go")
+	assert.True(t, strings.Count(output, "modelRT/logger_test.TestCallerPointsToBusinessCode") >= 2)
+}

logger/facede.go

@@ -43,8 +43,13 @@ func (f *facade) log(ctx context.Context, lvl zapcore.Level, msg string, kv ...a
 }
 
 func (f *facade) logSkip(ctx context.Context, lvl zapcore.Level, extraSkip int, msg string, kv ...any) {
-	fields := makeLogFieldsSkip(ctx, extraSkip, kv...)
+	caller := resolveLoggerCaller(extraSkip)
+	fields := makeLogFieldsWithCaller(ctx, caller, kv...)
 	ce := f._logger.Check(lvl, msg)
+	if ce == nil {
+		return
+	}
+	setCheckedEntryCaller(ce, caller)
 	ce.Write(fields...)
 }
 

logger/gorm_logger.go

@@ -4,6 +4,7 @@ package logger
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"gorm.io/gorm"
@@ -29,17 +30,17 @@ func (l *GormLogger) LogMode(_ gormLogger.LogLevel) gormLogger.Interface {
 
 // Info define func for implementing gormLogger.Interface
 func (l *GormLogger) Info(ctx context.Context, msg string, data ...any) {
-	Info(ctx, msg, "data", data)
+	Info(ctx, fmt.Sprintf(msg, data...))
 }
 
 // Warn define func for implementing gormLogger.Interface
 func (l *GormLogger) Warn(ctx context.Context, msg string, data ...any) {
-	Warn(ctx, msg, "data", data)
+	Warn(ctx, fmt.Sprintf(msg, data...))
 }
 
 // Error define func for implementing gormLogger.Interface
 func (l *GormLogger) Error(ctx context.Context, msg string, data ...any) {
-	Error(ctx, msg, "data", data)
+	Error(ctx, fmt.Sprintf(msg, data...))
 }
 
 // Trace define func for implementing gormLogger.Interface
@@ -49,12 +50,12 @@ func (l *GormLogger) Trace(ctx context.Context, begin time.Time, fc func() (sql
 	// get gorm exec sql and rows affected
 	sql, rows := fc()
 	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
-		ErrorSkip(ctx, 1, "SQL ERROR", "sql", sql, "rows", rows, "dur(ms)", duration)
+		ErrorSkip(ctx, 0, "SQL ERROR", "sql", sql, "rows", rows, "dur(ms)", duration)
 		return
 	}
 	if duration > l.SlowThreshold.Milliseconds() {
-		WarnSkip(ctx, 1, "SQL SLOW", "sql", sql, "rows", rows, "dur(ms)", duration)
+		WarnSkip(ctx, 0, "SQL SLOW", "sql", sql, "rows", rows, "dur(ms)", duration)
 	} else {
-		InfoSkip(ctx, 1, "SQL INFO", "sql", sql, "rows", rows, "dur(ms)", duration)
+		InfoSkip(ctx, 0, "SQL INFO", "sql", sql, "rows", rows, "dur(ms)", duration)
 	}
 }

logger/logger.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"path"
 	"runtime"
+	"strings"
 
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
@@ -41,8 +42,13 @@ func (l *logger) Error(msg string, kv ...any) {
 }
 
 func (l *logger) log(lvl zapcore.Level, msg string, kv ...any) {
-	fields := makeLogFields(l.ctx, kv...)
+	caller := resolveLoggerCaller(0)
+	fields := makeLogFieldsWithCaller(l.ctx, caller, kv...)
 	ce := l._logger.Check(lvl, msg)
+	if ce == nil {
+		return
+	}
+	setCheckedEntryCaller(ce, caller)
 	ce.Write(fields...)
 }
 
@@ -51,6 +57,10 @@ func makeLogFields(ctx context.Context, kv ...any) []zap.Field {
 }
 
 func makeLogFieldsSkip(ctx context.Context, extraSkip int, kv ...any) []zap.Field {
+	return makeLogFieldsWithCaller(ctx, resolveLoggerCaller(extraSkip), kv...)
+}
+
+func makeLogFieldsWithCaller(ctx context.Context, caller loggerCaller, kv ...any) []zap.Field {
 	if len(kv)%2 != 0 {
 		kv = append(kv, "unknown")
 	}
@@ -60,8 +70,7 @@ func makeLogFieldsSkip(ctx context.Context, extraSkip int, kv ...any) []zap.Fiel
 	spanID := spanCtx.SpanID().String()
 	kv = append(kv, "traceID", traceID, "spanID", spanID)
 
-	funcName, file, line := getLoggerCallerInfoSkip(extraSkip)
-	kv = append(kv, "func", funcName, "file", file, "line", line)
+	kv = append(kv, "func", caller.funcName, "file", caller.shortFile, "line", caller.line)
 	fields := make([]zap.Field, 0, len(kv)/2)
 	for i := 0; i < len(kv); i += 2 {
 		key := kv[i].(string)
@@ -95,13 +104,59 @@ func getLoggerCallerInfo() (funcName, file string, line int) {
 
 // getLoggerCallerInfoSkip returns caller info with additional skip frames beyond the standard depth.
 func getLoggerCallerInfoSkip(extraSkip int) (funcName, file string, line int) {
-	pc, file, line, ok := runtime.Caller(4 + extraSkip)
-	if !ok {
+	caller := resolveLoggerCaller(extraSkip)
+	return caller.funcName, caller.shortFile, caller.line
+}
+
+type loggerCaller struct {
+	pc        uintptr
+	funcName  string
+	fullFile  string
+	shortFile string
+	line      int
+}
+
+func resolveLoggerCaller(extraSkip int) loggerCaller {
+	pcs := make([]uintptr, 32)
+	count := runtime.Callers(2, pcs)
+	frames := runtime.CallersFrames(pcs[:count])
+
+	for {
+		frame, more := frames.Next()
+		if !isLoggerInfrastructureFrame(frame.Function) {
+			if extraSkip > 0 {
+				extraSkip--
+			} else {
+				return loggerCaller{
+					pc:        frame.PC,
+					funcName:  frame.Function,
+					fullFile:  frame.File,
+					shortFile: path.Base(frame.File),
+					line:      frame.Line,
+				}
+			}
+		}
+		if !more {
+			return loggerCaller{}
+		}
+	}
+}
+
+func isLoggerInfrastructureFrame(function string) bool {
+	return strings.HasPrefix(function, "modelRT/logger.") ||
+		strings.HasPrefix(function, "gorm.io/gorm")
+}
+
+func setCheckedEntryCaller(entry *zapcore.CheckedEntry, caller loggerCaller) {
+	if caller.pc == 0 {
 		return
 	}
-	file = path.Base(file)
-	funcName = runtime.FuncForPC(pc).Name()
-	return
+	entry.Entry.Caller = zapcore.EntryCaller{
+		Defined: true,
+		PC:      caller.pc,
+		File:    caller.fullFile,
+		Line:    caller.line,
+	}
 }
 
 // New returns a logger bound to ctx. Trace fields (traceID, spanID) are extracted

logger/loki_syncer.go

@@ -47,8 +47,7 @@ func newLokiSyncer(lCfg config.LokiConfig) *lokiSyncer {
 		client:   &http.Client{Timeout: 5 * time.Second},
 		ch:       make(chan string, 512),
 	}
-	ls.wg.Add(1)
-	go ls.run()
+	ls.wg.Go(ls.run)
 	return ls
 }
 
@@ -70,7 +69,6 @@ func (ls *lokiSyncer) Sync() error {
 }
 
 func (ls *lokiSyncer) run() {
-	defer ls.wg.Done()
 	ticker := time.NewTicker(2 * time.Second)
 	defer ticker.Stop()
 

main.go

@@ -59,8 +59,6 @@ var (
 // @title ModelRT 实时模型服务 API 文档
 // @version 1.0
 // @description 实时数据计算和模型运行服务的 API 服务
-// TODO termsOfService服务条款待后续优化
-// // @termsOfService http://swagger.io/terms/
 //
 // @contact.name douxu
 // TODO 修改支持的文档地址
@@ -179,8 +177,8 @@ func main() {
 	// init async task worker
 	taskWorker, err := task.InitTaskWorker(ctx, modelRTConfig, postgresDBClient)
 	if err != nil {
-		logger.Error(ctx, "Failed to initialize task worker", "error", err)
-		// Continue without task worker, but log warning
+		logger.Error(ctx, "failed to initialize task worker", "error", err)
+		// continue without task worker, but log warning
 	} else {
 		go taskWorker.Start()
 		defer taskWorker.Stop()
@@ -237,6 +235,18 @@ func main() {
 			panic(err)
 		}
 
+		componentColumnNames, err := database.QueryComponentColumnNames(ctx, tx)
+		if err != nil {
+			logger.Error(ctx, "query component table column names failed", "error", err)
+			panic(err)
+		}
+
+		err = model.StoreComponentColumnRecommend(ctx, fullParentPath, isLocalParentPath, componentColumnNames)
+		if err != nil {
+			logger.Error(ctx, "store component column recommend content failed", "error", err)
+			panic(err)
+		}
+
 		allMeasurement, err := database.GetAllMeasurements(ctx, tx)
 		if err != nil {
 			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
@@ -244,12 +254,12 @@ func main() {
 		}
 		go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement)
 
-		tree, _, err := database.QueryTopologicFromDB(ctx, tx)
+		topologics, err := database.QueryTopologic(ctx, tx)
 		if err != nil {
 			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
 			panic(err)
 		}
-		diagram.GlobalTree = tree
+		diagram.SetGlobalTopologyGraph(diagram.NewTopologyGraph(topologics))
 		return nil
 	})
 
@@ -258,7 +268,7 @@ func main() {
 		gin.SetMode(gin.ReleaseMode)
 	}
 	engine := gin.New()
-	// 添加CORS中间件
+	// add CORS middleware
 	engine.Use(cors.New(cors.Config{
 		AllowOrigins:     []string{"*"}, // 或指定具体域名
 		AllowMethods:     []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
@@ -278,7 +288,7 @@ func main() {
 		Handler: engine,
 	}
 
-	// creating a System Signal Receiver
+	// creating a system signal receiver
 	done := make(chan os.Signal, 10)
 	signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
 	go func() {

middleware/token.go

@@ -1,12 +1,20 @@
 // Package middleware define gin framework middlewares
 package middleware
 
-import "github.com/gin-gonic/gin"
+import (
+	"context"
+
+	"modelRT/constants"
+
+	"github.com/gin-gonic/gin"
+)
 
 // SetTokenMiddleware define a middleware for set token in context
 func SetTokenMiddleware(clientToken string) gin.HandlerFunc {
 	return func(c *gin.Context) {
-		c.Set("client_token", clientToken)
+		c.Set(constants.ClientTokenContextName, clientToken)
+		requestCtx := context.WithValue(c.Request.Context(), constants.CtxKeyClientToken, clientToken)
+		c.Request = c.Request.WithContext(requestCtx)
 		c.Next()
 	}
 }

middleware/token_test.go

@@ -0,0 +1,28 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"modelRT/constants"
+
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSetTokenMiddlewarePropagatesClientToken(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	router.Use(SetTokenMiddleware("test-token"))
+	router.GET("/test", func(c *gin.Context) {
+		assert.Equal(t, "test-token", c.GetString(constants.ClientTokenContextName))
+		assert.Equal(t, "test-token", c.Request.Context().Value(constants.CtxKeyClientToken))
+		c.Status(http.StatusNoContent)
+	})
+
+	request := httptest.NewRequest(http.MethodGet, "/test", nil)
+	response := httptest.NewRecorder()
+	router.ServeHTTP(response, request)
+	assert.Equal(t, http.StatusNoContent, response.Code)
+}

model/attribute_group_recommend_model.go

@@ -12,6 +12,7 @@ import (
 	"modelRT/orm"
 
 	"github.com/RediSearch/redisearch-go/v2/redisearch"
+	"golang.org/x/sync/errgroup"
 	"gorm.io/gorm"
 )
 
@@ -20,24 +21,38 @@ type columnParam struct {
 	AttributeGroup map[string]any
 }
 
+type attributeGroupRecommendJob struct {
+	AttributeSet    orm.AttributeSet
+	FullPath        string
+	IsLocalFullPath string
+	ColumnParam     columnParam
+}
+
 // TraverseAttributeGroupTables define func to traverse component attribute group tables
 func TraverseAttributeGroupTables(ctx context.Context, db *gorm.DB, compTagToFullPath map[string]string, isLocalCompTagToFullPath map[string]string, compAttrSet map[string]orm.AttributeSet) error {
 	var tableNames []string
 
 	excludedTables := []string{"component", ""}
+	var projectTableNames []string
 	result := db.Model(&orm.ProjectManager{}).
 		Where("name NOT IN ?", excludedTables).
-		Pluck("name", &tableNames)
+		Pluck("name", &projectTableNames)
 	if result.Error != nil && result.Error != gorm.ErrRecordNotFound {
 		logger.Error(ctx, "query name column data from postgres table failed", "err", result.Error)
 		return result.Error
 	}
+	for _, tableName := range projectTableNames {
+		if constants.IsSupportedParameterTableName(tableName) {
+			tableNames = append(tableNames, tableName)
+		}
+	}
 
 	if len(tableNames) == 0 {
 		logger.Info(ctx, "query from postgres successed, but no records found")
 		return nil
 	}
 
+	jobs := make([]attributeGroupRecommendJob, 0)
 	for _, tableName := range tableNames {
 		var records []map[string]any
 		err := db.Table(tableName).Find(&records).Error
@@ -102,13 +117,27 @@ func TraverseAttributeGroupTables(ctx context.Context, db *gorm.DB, compTagToFul
 				AttributeType:  attributeType,
 				AttributeGroup: attributeGroup,
 			}
-			go storeAttributeGroup(ctx, attrSet, fullPath, isLocalfullPath, columnParam)
+			jobs = append(jobs, attributeGroupRecommendJob{
+				AttributeSet:    attrSet,
+				FullPath:        fullPath,
+				IsLocalFullPath: isLocalfullPath,
+				ColumnParam:     columnParam,
+			})
 		}
 	}
-	return nil
+
+	group, groupCtx := errgroup.WithContext(ctx)
+	group.SetLimit(16)
+	for _, job := range jobs {
+		job := job
+		group.Go(func() error {
+			return storeAttributeGroup(groupCtx, job.AttributeSet, job.FullPath, job.IsLocalFullPath, job.ColumnParam)
+		})
+	}
+	return group.Wait()
 }
 
-func storeAttributeGroup(ctx context.Context, attributeSet orm.AttributeSet, fullPath string, isLocalFullPath string, colParams columnParam) {
+func storeAttributeGroup(ctx context.Context, attributeSet orm.AttributeSet, fullPath string, isLocalFullPath string, colParams columnParam) error {
 	rdb := diagram.GetRedisClientInstance()
 	pipe := rdb.Pipeline()
 
@@ -121,18 +150,25 @@ func storeAttributeGroup(ctx context.Context, attributeSet orm.AttributeSet, ful
 	attrbutesGroups := make([]any, 0, len(colParams.AttributeGroup)*2)
 	attributeGroupKey := fmt.Sprintf("%s_%s", attributeSet.CompTag, colParams.AttributeType)
 
-	sug := make([]redisearch.Suggestion, 0, len(colParams.AttributeGroup)*4)
+	sug := make([]redisearch.Suggestion, 0, len(colParams.AttributeGroup)*2+2)
+	configTerm := fmt.Sprintf("%s.%s", fullPath, colParams.AttributeType)
+	sug = append(sug, redisearch.Suggestion{
+		Term:  configTerm,
+		Score: constants.DefaultScore,
+	})
+	if isLocalFullPath != "" {
+		configTerm = fmt.Sprintf("%s.%s", isLocalFullPath, colParams.AttributeType)
+		sug = append(sug, redisearch.Suggestion{
+			Term:  configTerm,
+			Score: constants.DefaultScore,
+		})
+	}
+
 	for attrName, attrValue := range colParams.AttributeGroup {
 		attrbutesGroups = append(attrbutesGroups, attrName, attrValue)
 		attrNameMembers = append(attrNameMembers, attrName)
 
 		// add redis fuzzy search suggestion for token1-token7 type
-		configTerm := fmt.Sprintf("%s.%s", fullPath, colParams.AttributeType)
-		sug = append(sug, redisearch.Suggestion{
-			Term:  configTerm,
-			Score: constants.DefaultScore,
-		})
-
 		measTerm := fmt.Sprintf("%s.%s.%s", fullPath, colParams.AttributeType, attrName)
 		sug = append(sug, redisearch.Suggestion{
 			Term:  measTerm,
@@ -140,11 +176,9 @@ func storeAttributeGroup(ctx context.Context, attributeSet orm.AttributeSet, ful
 		})
 
 		// add redis fuzzy search suggestion for token4-token7 type
-		configTerm = fmt.Sprintf("%s.%s", isLocalFullPath, colParams.AttributeType)
-		sug = append(sug, redisearch.Suggestion{
-			Term:  configTerm,
-			Score: constants.DefaultScore,
-		})
+		if isLocalFullPath == "" {
+			continue
+		}
 
 		measTerm = fmt.Sprintf("%s.%s.%s", isLocalFullPath, colParams.AttributeType, attrName)
 		sug = append(sug, redisearch.Suggestion{
@@ -163,11 +197,24 @@ func storeAttributeGroup(ctx context.Context, attributeSet orm.AttributeSet, ful
 	}
 
 	if len(sug) > 0 {
-		ac.AddTerms(sug...)
+		if err := ac.AddTerms(sug...); err != nil {
+			logger.Error(ctx, "add attribute group recommend suggestions failed",
+				"component_tag", attributeSet.CompTag,
+				"attribute_type", colParams.AttributeType,
+				"error", err,
+			)
+			return fmt.Errorf("add attribute group recommend suggestions: %w", err)
+		}
 	}
 
 	_, err := pipe.Exec(ctx)
 	if err != nil {
-		logger.Error(ctx, "init component attribute group recommend content failed", "error", err)
+		logger.Error(ctx, "init component attribute group recommend content failed",
+			"component_tag", attributeSet.CompTag,
+			"attribute_type", colParams.AttributeType,
+			"error", err,
+		)
+		return fmt.Errorf("init component attribute group recommend content: %w", err)
 	}
+	return nil
 }

model/attribute_model.go

@@ -90,6 +90,6 @@ func (s *ShortAttrInfo) IsLocal() bool {
 }
 
 // GetAttrValue define return the attribute value
-func (l *ShortAttrInfo) GetAttrValue() any {
-	return l.AttrValue
+func (s *ShortAttrInfo) GetAttrValue() any {
+	return s.AttrValue
 }