From 64d06c90bae32442f52059de0cfca1a66e235372 Mon Sep 17 00:00:00 2001 From: douxu Date: Wed, 15 Jul 2026 17:33:55 +0800 Subject: [PATCH] feat: add measurement data-object attribute APIs - classify and validate measurement tokens in supported formats - query measurement attributes and latest values from PostgreSQL and Redis - add measurement mode support and centralized SQL definitions - register data-object attribute query and update routes - fix logger caller attribution and measurement identifier generation - add tests for token parsing, attribute queries, Redis values, and logging --- common/data_object_errors.go | 15 ++ constants/data-object.go | 12 + database/query_component_measurement.go | 234 ++++++++++++++---- database/query_component_measurement_test.go | 199 +++++++++++++++ database/query_measurement.go | 62 ----- .../compute_data_injection.go | 3 +- deploy/redis-test-data/util/rand.go | 3 +- diagram/redis_client.go | 42 ++++ diagram/redis_client_test.go | 28 +++ handler/component_attribute_query.go | 3 +- handler/data_object_attribute_query.go | 197 +++++++++++++++ handler/data_object_attribute_query_test.go | 115 +++++++++ handler/data_object_attribute_update.go | 166 +++++++++++++ logger/caller_test.go | 50 ++++ logger/facede.go | 7 +- logger/gorm_logger.go | 6 +- logger/logger.go | 71 +++++- model/data_object_token.go | 57 +++++ model/data_object_token_test.go | 101 ++++++++ model/measurement_attribute.go | 78 ++++++ model/measurement_attribute_test.go | 40 +++ model/measurement_identifier_test.go | 35 +++ model/measurement_protol_model.go | 18 +- orm/async_task.go | 2 +- orm/busbar_section.go | 10 +- orm/circuit_diagram_measurement.go | 1 + router/data-object.go | 16 ++ router/router.go | 1 + sql/measurement.go | 82 ++++++ sql/topologic.go | 4 +- 30 files changed, 1513 insertions(+), 145 deletions(-) create mode 100644 common/data_object_errors.go create mode 100644 constants/data-object.go create mode 100644 database/query_component_measurement_test.go delete mode 100644 database/query_measurement.go create mode 100644 diagram/redis_client_test.go create mode 100644 handler/data_object_attribute_query.go create mode 100644 handler/data_object_attribute_query_test.go create mode 100644 handler/data_object_attribute_update.go create mode 100644 logger/caller_test.go create mode 100644 model/data_object_token.go create mode 100644 model/data_object_token_test.go create mode 100644 model/measurement_attribute.go create mode 100644 model/measurement_attribute_test.go create mode 100644 model/measurement_identifier_test.go create mode 100644 router/data-object.go create mode 100644 sql/measurement.go diff --git a/common/data_object_errors.go b/common/data_object_errors.go new file mode 100644 index 0000000..71c0349 --- /dev/null +++ b/common/data_object_errors.go @@ -0,0 +1,15 @@ +// Package common define common error variables +package common + +import "errors" + +var ( + // 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") +) diff --git a/constants/data-object.go b/constants/data-object.go new file mode 100644 index 0000000..0c3212c --- /dev/null +++ b/constants/data-object.go @@ -0,0 +1,12 @@ +// 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" +) diff --git a/database/query_component_measurement.go b/database/query_component_measurement.go index e9e95ca..46ca19a 100644 --- a/database/query_component_measurement.go +++ b/database/query_component_measurement.go @@ -4,23 +4,174 @@ package database import ( "context" "fmt" + "strings" + "time" + "modelRT/common" "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"` +// 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(sql.MeasurementIDWhere, id). + Clauses(clause.Locking{Strength: "UPDATE"}). + First(&measurement) + + if result.Error != nil { + return orm.Measurement{}, result.Error + } + return measurement, nil } -type StationWithParent struct { - orm.Zone - ZoneTag string `gorm:"column:zone_tag"` +// 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 } +// 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), @@ -35,7 +186,7 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe 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 +202,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 +219,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 @@ -115,16 +263,16 @@ func GetFullMeasurementSet(ctx context.Context, db *gorm.DB) (*orm.MeasurementSe orm.Measurement CompTag string `gorm:"column:comp_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, + ) } } return nil diff --git a/database/query_component_measurement_test.go b/database/query_component_measurement_test.go new file mode 100644 index 0000000..eee8f82 --- /dev/null +++ b/database/query_component_measurement_test.go @@ -0,0 +1,199 @@ +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 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()) +} diff --git a/database/query_measurement.go b/database/query_measurement.go deleted file mode 100644 index aa51e89..0000000 --- a/database/query_measurement.go +++ /dev/null @@ -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 -} diff --git a/deploy/redis-test-data/real-time-compute/compute_data_injection.go b/deploy/redis-test-data/real-time-compute/compute_data_injection.go index 2f68ef7..8707336 100644 --- a/deploy/redis-test-data/real-time-compute/compute_data_injection.go +++ b/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++ { diff --git a/deploy/redis-test-data/util/rand.go b/deploy/redis-test-data/util/rand.go index 7df9c92..2e31302 100644 --- a/deploy/redis-test-data/util/rand.go +++ b/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 } diff --git a/diagram/redis_client.go b/diagram/redis_client.go index 4b673a8..247c98c 100644 --- a/diagram/redis_client.go +++ b/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{ diff --git a/diagram/redis_client_test.go b/diagram/redis_client_test.go new file mode 100644 index 0000000..7416ba1 --- /dev/null +++ b/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) +} diff --git a/handler/component_attribute_query.go b/handler/component_attribute_query.go index 44e9462..dfa4191 100644 --- a/handler/component_attribute_query.go +++ b/handler/component_attribute_query.go @@ -8,8 +8,6 @@ import ( "slices" "strings" - "github.com/gofrs/uuid" - "modelRT/common/errcode" "modelRT/constants" "modelRT/database" @@ -18,6 +16,7 @@ import ( "modelRT/orm" "github.com/gin-gonic/gin" + "github.com/gofrs/uuid" ) // ComponentAttributeQueryHandler define circuit diagram component attribute value query process API diff --git a/handler/data_object_attribute_query.go b/handler/data_object_attribute_query.go new file mode 100644 index 0000000..a611a94 --- /dev/null +++ b/handler/data_object_attribute_query.go @@ -0,0 +1,197 @@ +// 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 := c.Param("token") + field := c.Param("field") + if token == "" { + err := fmt.Errorf("tokens is missing from the path") + logger.Error(ctx, "query tokens from path failed", "error", err, "url", c.Request.RequestURI) + renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil) + return + } else if field == "" { + err := fmt.Errorf("field is missing from the path") + logger.Error(ctx, "query field from path failed", "error", err, "url", c.Request.RequestURI) + renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil) + return + } + + // TODO 根据传入的token查询是measurement还是attribute，分别查询不同的表 + // 量测支持token1.token2.token3.token4.token5.token6.token7、token4.token5.token6.token7、token4.token7 + // 参量支持两种形式token4.token5.token6.token7与token1.token2.token3.token4.token5.token6.token7 + 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 + } + + var measurement *orm.Measurement + var measurementComponent *orm.Component + switch dataObjectType { + case constants.DataObjectTypeParameter: + // TODO 验证参量token的合法性 + case constants.DataObjectTypeMeasurement: + 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: + // TODO 在根据传递的field来组装具体的查询结果 + // 目前先直接返回错误，后续再实现具体的查询逻辑 + renderRespFailure(c, constants.RespCodeServerError, "query parameter attribute is not implemented yet", nil) + case constants.DataObjectTypeMeasurement: + normalizedField := strings.ToLower(field) + value, err := buildMeasurementAttributeValue( + ctx, + normalizedField, + 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: normalizedField, + 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) + } +} + +type measurementValueLoader func(context.Context, orm.JSONMap) (any, error) + +type dataObjectAttributeQueryResult struct { + Token string `json:"token"` + Field string `json:"field"` + Code int `json:"code"` + Msg string `json:"msg"` + Value any `json:"value"` +} + +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": + fmt.Println("measurement.Mode:", measurement.Mode) + switch measurement.Mode { + case 0: + return false, nil + case 1: + return true, nil + default: + return nil, fmt.Errorf("invalid measurement mode %d: expected 0 or 1", measurement.Mode) + } + 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 +} diff --git a/handler/data_object_attribute_query_test.go b/handler/data_object_attribute_query_test.go new file mode 100644 index 0000000..af57498 --- /dev/null +++ b/handler/data_object_attribute_query_test.go @@ -0,0 +1,115 @@ +package handler + +import ( + "context" + "testing" + + "modelRT/common" + "modelRT/orm" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +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: true}, + {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{} + + manualMode, err := buildMeasurementAttributeValue( + context.Background(), + "mode", + &orm.Measurement{Mode: 0}, + component, + nil, + ) + require.NoError(t, err) + assert.Equal(t, false, manualMode) + + automaticMode, err := buildMeasurementAttributeValue( + context.Background(), + "mode", + &orm.Measurement{Mode: 1}, + component, + nil, + ) + require.NoError(t, err) + assert.Equal(t, true, automaticMode) + + _, err = buildMeasurementAttributeValue( + context.Background(), + "mode", + &orm.Measurement{Mode: 2}, + component, + nil, + ) + require.Error(t, err) + assert.Contains(t, err.Error(), "expected 0 or 1") +} diff --git a/handler/data_object_attribute_update.go b/handler/data_object_attribute_update.go new file mode 100644 index 0000000..6ef3099 --- /dev/null +++ b/handler/data_object_attribute_update.go @@ -0,0 +1,166 @@ +// Package handler provides HTTP handlers for various endpoints. +package handler + +import ( + "fmt" + "strings" + + "modelRT/common/errcode" + "modelRT/constants" + "modelRT/database" + "modelRT/diagram" + "modelRT/logger" + "modelRT/network" + "modelRT/orm" + + "github.com/gin-gonic/gin" +) + +// DataObjectAttributeUpdateHandler define data object attribute value update process API +func DataObjectAttributeUpdateHandler(c *gin.Context) { + pgClient := database.GetPostgresDBClient() + var request network.ComponentAttributeUpdateInfo + if err := c.ShouldBindJSON(&request); err != nil { + logger.Error(c, "unmarshal request params failed", "error", err) + renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil) + return + } + + updateResults := make(map[string]*errcode.AppError) + attriModifyConfs := make([]attributeModifyConfig, 0, len(request.AttributeConfigs)) + var attributeComponentTag string + for index, attribute := range request.AttributeConfigs { + slices := strings.Split(attribute.AttributeToken, ".") + if len(slices) < 7 { + updateResults[attribute.AttributeToken] = errcode.ErrInvalidToken + continue + } + + componentTag := slices[4] + if index == 0 { + attributeComponentTag = componentTag + } else if componentTag != attributeComponentTag { + updateResults[attribute.AttributeToken] = errcode.ErrCrossToken + continue + } + + attriModifyConfs = append(attriModifyConfs, attributeModifyConfig{ + attributeToken: attribute.AttributeToken, + attributeExtendType: slices[5], + attributeName: slices[6], + attributeOldVal: attribute.AttributeOldVal, + attributeNewVal: attribute.AttributeNewVal, + }) + } + + // open transaction + tx := pgClient.WithContext(c).Begin() + if tx.Error != nil { + logger.Error(c, "begin postgres transaction failed", "error", tx.Error) + renderRespFailure(c, constants.RespCodeServerError, "begin postgres transaction failed", nil) + return + } + + compInfo, err := database.QueryComponentByCompTag(c, tx, attributeComponentTag) + if err != nil { + logger.Error(c, "query component info by component tag failed", "error", err, "tag", attributeComponentTag) + + for _, attribute := range request.AttributeConfigs { + if _, exists := updateResults[attribute.AttributeToken]; !exists { + updateResults[attribute.AttributeToken] = errcode.ErrDBQueryFailed.WithCause(err) + } + } + + tx.Rollback() + + payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) + renderRespFailure(c, constants.RespCodeFailed, "query component metadata failed", payload) + return + } + + identifiers := make([]orm.ProjectIdentifier, len(attriModifyConfs)) + for i, mod := range attriModifyConfs { + identifiers[i] = orm.ProjectIdentifier{ + Token: mod.attributeToken, + Tag: compInfo.ModelName, + GroupName: mod.attributeExtendType, + } + } + tableNameMap, err := database.BatchGetProjectNames(tx, identifiers) + if err != nil { + tx.Rollback() + + for _, id := range identifiers { + if _, exists := updateResults[id.Token]; !exists { + updateResults[id.Token] = errcode.ErrRetrieveFailed.WithCause(err) + } + } + + payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) + renderRespFailure(c, constants.RespCodeFailed, "batch retrieve table names failed", payload) + return + } + + redisUpdateMap := make(map[string][]cacheUpdateItem) + for _, mod := range attriModifyConfs { + id := orm.ProjectIdentifier{Tag: compInfo.ModelName, GroupName: mod.attributeExtendType} + tableName, exists := tableNameMap[id] + if !exists { + updateResults[mod.attributeToken] = errcode.ErrFoundTargetFailed + continue + } + + result := tx.Table(tableName). + Where(fmt.Sprintf("%s = ? AND global_uuid = ?", mod.attributeName), mod.attributeOldVal, compInfo.GlobalUUID). + Updates(map[string]any{mod.attributeName: mod.attributeNewVal}) + + if result.Error != nil { + updateResults[mod.attributeToken] = errcode.ErrDBUpdateFailed + continue + } + if result.RowsAffected == 0 { + updateResults[mod.attributeToken] = errcode.ErrDBzeroAffectedRows + continue + } + + cacheKey := fmt.Sprintf("%s_%s", attributeComponentTag, mod.attributeExtendType) + redisUpdateMap[cacheKey] = append(redisUpdateMap[cacheKey], + cacheUpdateItem{ + token: mod.attributeToken, + name: mod.attributeName, + newVal: mod.attributeNewVal, + }) + } + + // commit transaction + if err := tx.Commit().Error; err != nil { + renderRespFailure(c, constants.RespCodeServerError, "transaction commit failed", nil) + return + } + + for key, items := range redisUpdateMap { + hset := diagram.NewRedisHash(c, key, 5000, false) + + fields := make(map[string]any, len(items)) + for _, item := range items { + fields[item.name] = item.newVal + } + + if err := hset.SetRedisHashByMap(fields); err != nil { + logger.Error(c, "batch sync redis failed", "hash_key", key, "error", err) + + for _, item := range items { + if _, exists := updateResults[item.token]; exists { + updateResults[item.token] = errcode.ErrCacheSyncWarn.WithCause(err) + } + } + } + } + + payload := genUpdateRespPayload(updateResults, request.AttributeConfigs) + if len(updateResults) > 0 { + renderRespFailure(c, constants.RespCodeFailed, "process completed with partial failures", payload) + return + } + renderRespSuccess(c, constants.RespCodeSuccess, "process completed successfully", payload) +} diff --git a/logger/caller_test.go b/logger/caller_test.go new file mode 100644 index 0000000..035d41f --- /dev/null +++ b/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) +} diff --git a/logger/facede.go b/logger/facede.go index 1533cdf..0e27628 100644 --- a/logger/facede.go +++ b/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...) } diff --git a/logger/gorm_logger.go b/logger/gorm_logger.go index 2387b5f..59d7979 100644 --- a/logger/gorm_logger.go +++ b/logger/gorm_logger.go @@ -50,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) } } diff --git a/logger/logger.go b/logger/logger.go index f7fa247..f40700d 100644 --- a/logger/logger.go +++ b/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 diff --git a/model/data_object_token.go b/model/data_object_token.go new file mode 100644 index 0000000..6d9a151 --- /dev/null +++ b/model/data_object_token.go @@ -0,0 +1,57 @@ +// Package model defines data models and domain rules for model runtime service. +package model + +import ( + "fmt" + "slices" + "strings" + + "modelRT/constants" +) + +var parameterAttributeGroups = map[string]struct{}{ + "base_extend": {}, + "rated": {}, + "setup": {}, + "model": {}, + "stable": {}, + "craft": {}, + "integrity": {}, + "behavior": {}, +} + +// ClassifyDataObjectToken determines whether token identifies a parameter or a +// measurement. Seven-part and four-part tokens are classified by token6, while +// two-part tokens are treated as measurements at the current stage. +func ClassifyDataObjectToken(token string) (constants.DataObjectType, error) { + parts := strings.Split(token, ".") + if slices.Contains(parts, "") { + return "", fmt.Errorf("invalid data object token %q: token segment cannot be empty", token) + } + + switch len(parts) { + case 2: + return constants.DataObjectTypeMeasurement, nil + case 4, 7: + token6Index := 2 + if len(parts) == 7 { + token6Index = 5 + } + + token6 := parts[token6Index] + if _, ok := parameterAttributeGroups[token6]; ok { + return constants.DataObjectTypeParameter, nil + } + if token6 == "bay" { + return constants.DataObjectTypeMeasurement, nil + } + // TODO 后续要完成对component这个token6类型的支持 + if token6 == "component" { + return "", fmt.Errorf("unsupported data object token %q: token6 component is not queryable", token) + } + + return "", fmt.Errorf("invalid data object token %q: unsupported token6 %q", token, token6) + default: + return "", fmt.Errorf("invalid data object token %q: expected 2, 4, or 7 segments, got %d", token, len(parts)) + } +} diff --git a/model/data_object_token_test.go b/model/data_object_token_test.go new file mode 100644 index 0000000..93a0a47 --- /dev/null +++ b/model/data_object_token_test.go @@ -0,0 +1,101 @@ +package model + +import ( + "testing" + + "modelRT/constants" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestClassifyDataObjectToken(t *testing.T) { + tests := []struct { + name string + token string + expected constants.DataObjectType + wantErr string + }{ + { + name: "seven-part measurement", + token: "grid.zone.station.nspath.component.bay.measurement", + expected: constants.DataObjectTypeMeasurement, + }, + { + name: "four-part measurement", + token: "nspath.component.bay.measurement", + expected: constants.DataObjectTypeMeasurement, + }, + { + name: "two-part measurement", + token: "nspath.measurement", + expected: constants.DataObjectTypeMeasurement, + }, + { + name: "seven-part parameter", + token: "grid.zone.station.nspath.component.rated.voltage", + expected: constants.DataObjectTypeParameter, + }, + { + name: "four-part parameter", + token: "nspath.component.base_extend.description", + expected: constants.DataObjectTypeParameter, + }, + { + name: "component group", + token: "nspath.component.component.name", + wantErr: "token6 component is not queryable", + }, + { + name: "unknown group", + token: "nspath.component.unknown.name", + wantErr: "unsupported token6", + }, + { + name: "invalid segment count", + token: "grid.zone.station", + wantErr: "expected 2, 4, or 7 segments", + }, + { + name: "empty segment", + token: "nspath..bay.measurement", + wantErr: "token segment cannot be empty", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + actual, err := ClassifyDataObjectToken(tt.token) + if tt.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + assert.Empty(t, actual) + return + } + + require.NoError(t, err) + assert.Equal(t, tt.expected, actual) + }) + } +} + +func TestClassifyDataObjectTokenParameterGroups(t *testing.T) { + groups := []string{ + "base_extend", + "rated", + "setup", + "model", + "stable", + "craft", + "integrity", + "behavior", + } + + for _, group := range groups { + t.Run(group, func(t *testing.T) { + actual, err := ClassifyDataObjectToken("nspath.component." + group + ".attribute") + require.NoError(t, err) + assert.Equal(t, constants.DataObjectTypeParameter, actual) + }) + } +} diff --git a/model/measurement_attribute.go b/model/measurement_attribute.go new file mode 100644 index 0000000..51c0cc0 --- /dev/null +++ b/model/measurement_attribute.go @@ -0,0 +1,78 @@ +package model + +import ( + "fmt" + "strings" + + "modelRT/constants" + "modelRT/orm" +) + +var allowedMeasurementTypes = map[string]struct{}{ + "TM": {}, + "TS": {}, + "TC": {}, + "TA": {}, + "SP": {}, +} + +// MeasurementTypeFromDataSource returns the two-character measurement type +// encoded in a CL3611 channel. Only TM, TS, TC, TA, and SP are valid. +func MeasurementTypeFromDataSource(dataSource orm.JSONMap) (string, error) { + dataSourceType, err := integerJSONValue(dataSource["type"]) + if err != nil { + return "", fmt.Errorf("invalid measurement data_source type: %w", err) + } + if dataSourceType != constants.DataSourceTypeCL3611 { + return "", fmt.Errorf("measurement type requires data_source type %d, got %d", constants.DataSourceTypeCL3611, dataSourceType) + } + + ioAddress, ok := dataSource["io_address"].(map[string]any) + if !ok { + if value, jsonMapOK := dataSource["io_address"].(orm.JSONMap); jsonMapOK { + ioAddress = map[string]any(value) + } else { + return "", fmt.Errorf("measurement data_source io_address is not an object") + } + } + + channel, ok := ioAddress["channel"].(string) + if !ok || len(channel) < 2 { + return "", fmt.Errorf("measurement data_source channel must contain at least two characters") + } + + measurementType := strings.ToUpper(channel[:2]) + if _, ok := allowedMeasurementTypes[measurementType]; !ok { + return "", fmt.Errorf("unsupported measurement type %q", measurementType) + } + return measurementType, nil +} + +func integerJSONValue(value any) (int, error) { + switch typed := value.(type) { + case int: + return typed, nil + case int8: + return int(typed), nil + case int16: + return int(typed), nil + case int32: + return int(typed), nil + case int64: + return int(typed), nil + case float32: + converted := int(typed) + if typed != float32(converted) { + return 0, fmt.Errorf("expected integer, got %v", typed) + } + return converted, nil + case float64: + converted := int(typed) + if typed != float64(converted) { + return 0, fmt.Errorf("expected integer, got %v", typed) + } + return converted, nil + default: + return 0, fmt.Errorf("expected integer, got %T", value) + } +} diff --git a/model/measurement_attribute_test.go b/model/measurement_attribute_test.go new file mode 100644 index 0000000..8ca8e38 --- /dev/null +++ b/model/measurement_attribute_test.go @@ -0,0 +1,40 @@ +package model + +import ( + "strings" + "testing" + + "modelRT/orm" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMeasurementTypeFromDataSource(t *testing.T) { + for _, measurementType := range []string{"TM", "TS", "TC", "TA", "SP"} { + t.Run(measurementType, func(t *testing.T) { + actual, err := MeasurementTypeFromDataSource(orm.JSONMap{ + "type": float64(1), + "io_address": map[string]any{ + "channel": strings.ToLower(measurementType) + "1_test", + }, + }) + require.NoError(t, err) + assert.Equal(t, measurementType, actual) + }) + } +} + +func TestMeasurementTypeFromDataSourceRejectsInvalidValues(t *testing.T) { + tests := []orm.JSONMap{ + {"type": float64(2), "io_address": map[string]any{"channel": "tm1"}}, + {"type": float64(1), "io_address": map[string]any{"channel": "xx1"}}, + {"type": float64(1), "io_address": map[string]any{"channel": "t"}}, + {"type": "1", "io_address": map[string]any{"channel": "tm1"}}, + } + + for _, dataSource := range tests { + _, err := MeasurementTypeFromDataSource(dataSource) + require.Error(t, err) + } +} diff --git a/model/measurement_identifier_test.go b/model/measurement_identifier_test.go new file mode 100644 index 0000000..aa8925b --- /dev/null +++ b/model/measurement_identifier_test.go @@ -0,0 +1,35 @@ +package model + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGenerateMeasureIdentifierSupportsJSONNumbers(t *testing.T) { + identifier, err := GenerateMeasureIdentifier(map[string]any{ + "type": float64(2), + "io_address": map[string]any{ + "station": "Station000", + "packet": float64(10), + "offset": float64(35), + }, + }) + + require.NoError(t, err) + assert.Equal(t, "station000:104:10:35", identifier) +} + +func TestGenerateMeasureIdentifierRejectsFractionalJSONNumbers(t *testing.T) { + _, err := GenerateMeasureIdentifier(map[string]any{ + "type": float64(2), + "io_address": map[string]any{ + "station": "station000", + "packet": float64(10.5), + "offset": float64(35), + }, + }) + + require.Error(t, err) +} diff --git a/model/measurement_protol_model.go b/model/measurement_protol_model.go index 228f51f..cfc4a1f 100644 --- a/model/measurement_protol_model.go +++ b/model/measurement_protol_model.go @@ -245,24 +245,18 @@ func GenerateMeasureIdentifier(source map[string]any) (string, error) { if !ok { return "", fmt.Errorf("Power104: missing packet field") } - var packet int - switch v := packetVal.(type) { - case int: - packet = v - default: - return "", fmt.Errorf("Power104:invalid packet format") + packet, err := integerJSONValue(packetVal) + if err != nil { + return "", fmt.Errorf("Power104:invalid packet format: %w", err) } offsetVal, ok := ioAddress["offset"] if !ok { return "", fmt.Errorf("Power104:missing offset field") } - var offset int - switch v := offsetVal.(type) { - case int: - offset = v - default: - return "", fmt.Errorf("Power104:invalid offset format") + offset, err := integerJSONValue(offsetVal) + if err != nil { + return "", fmt.Errorf("Power104:invalid offset format: %w", err) } return concatP104WithPlus(station, packet, offset), nil default: diff --git a/orm/async_task.go b/orm/async_task.go index 37709bf..d2900f5 100644 --- a/orm/async_task.go +++ b/orm/async_task.go @@ -117,7 +117,7 @@ func (a *AsyncTask) IsFailed() bool { return a.Status == AsyncTaskStatusFailed } -// IsValidTaskType checks if the task type is valid +// IsValidAsyncTaskType checks if the task type is valid func IsValidAsyncTaskType(taskType string) bool { switch AsyncTaskType(taskType) { case AsyncTaskTypeTopologyAnalysis, AsyncTaskTypePerformanceAnalysis, diff --git a/orm/busbar_section.go b/orm/busbar_section.go index d953d2f..b177385 100644 --- a/orm/busbar_section.go +++ b/orm/busbar_section.go @@ -91,10 +91,7 @@ func NewBusbarSection(name string) (*BusbarSection, error) { } func (b *BusbarSection) BusNameLenCheck() bool { - if len([]rune(b.BusbarName)) > 20 { - return false - } - return true + return len([]rune(b.BusbarName)) <= 20 } func (b *BusbarSection) BusVoltageCheck() bool { @@ -105,8 +102,5 @@ func (b *BusbarSection) BusVoltageCheck() bool { } func (b *BusbarSection) BusDescLenCheck() bool { - if len([]rune(b.BusbarDesc)) > 100 { - return false - } - return true + return len([]rune(b.BusbarDesc)) <= 100 } diff --git a/orm/circuit_diagram_measurement.go b/orm/circuit_diagram_measurement.go index e281559..19dcd76 100644 --- a/orm/circuit_diagram_measurement.go +++ b/orm/circuit_diagram_measurement.go @@ -14,6 +14,7 @@ type Measurement struct { Name string `gorm:"column:name;size:64;not null;default:''"` Type int16 `gorm:"column:type;not null;default:-1"` Size int `gorm:"column:size;not null;default:-1"` + Mode int16 `gorm:"column:mode;not null;default:1"` DataSource JSONMap `gorm:"column:data_source;type:jsonb;not null;default:'{}'"` EventPlan JSONMap `gorm:"column:event_plan;type:jsonb;not null;default:'{}'"` Binding JSONMap `gorm:"column:binding;type:jsonb;not null;default:'{\"ct\":{\"ratio\":1.0,\"polarity\":1,\"index\":0},\"pt\":{\"ratio\":1.0,\"polarity\":1,\"index\":0}}'"` diff --git a/router/data-object.go b/router/data-object.go new file mode 100644 index 0000000..b4850cb --- /dev/null +++ b/router/data-object.go @@ -0,0 +1,16 @@ +// Package router provides router config +package router + +import ( + "modelRT/handler" + + "github.com/gin-gonic/gin" +) + +// registerDataObjectRoutes define func of register data object routes +func registerDataObjectRoutes(rg *gin.RouterGroup, middlewares ...gin.HandlerFunc) { + g := rg.Group("/data-object/") + g.Use(middlewares...) + g.GET("attribute/:token/:field", handler.DataObjectAttributeQueryHandler) + g.POST("update", handler.DataObjectAttributeUpdateHandler) +} diff --git a/router/router.go b/router/router.go index f242cb9..b6ca534 100644 --- a/router/router.go +++ b/router/router.go @@ -27,5 +27,6 @@ func RegisterRoutes(engine *gin.Engine, clientToken string) { registerDataRoutes(routeGroup) registerMonitorRoutes(routeGroup) registerComponentRoutes(routeGroup, middleware.SetTokenMiddleware(clientToken)) + registerDataObjectRoutes(routeGroup, middleware.SetTokenMiddleware(clientToken)) registerAsyncTaskRoutes(routeGroup, middleware.SetTokenMiddleware(clientToken)) } diff --git a/sql/measurement.go b/sql/measurement.go new file mode 100644 index 0000000..641a201 --- /dev/null +++ b/sql/measurement.go @@ -0,0 +1,82 @@ +// Package sql defines reusable database SQL statements. +package sql + +const ( + // MeasurementCountSelect selects the number of measurements matching a + // token hierarchy and is used during token existence validation. + MeasurementCountSelect = "SELECT COUNT(*)" + + // MeasurementRowsSelect selects complete measurement rows after a token has + // been parsed into its hierarchy conditions. + MeasurementRowsSelect = "SELECT m.*" + + // MeasurementLimitTwo limits token resolution to two rows so callers can + // distinguish a unique match from an ambiguous token without loading all matches. + MeasurementLimitTwo = "LIMIT 2" + + // MeasurementIDWhere is the GORM condition used to query a measurement by + // its database primary key. + MeasurementIDWhere = "id = ?" + + // MeasurementTokenValidationQueryBase contains the common hierarchy joins + // used by all supported measurement token formats. + MeasurementTokenValidationQueryBase = MeasurementCountSelect + ` + FROM measurement AS m + INNER JOIN component AS c ON c.global_uuid = m.component_uuid + INNER JOIN bay AS b ON b.bay_uuid = m.bay_uuid + INNER JOIN station AS s ON s.id = c.station_id + INNER JOIN zone AS z ON z.id = s.zone_id + INNER JOIN grid AS g ON g.id = z.grid_id` + + // MeasurementSevenPartTokenWhere matches a complete token in the form + // token1.token2.token3.token4.token5.token6.token7. Token6 is validated as + // "bay" before this condition is used. + MeasurementSevenPartTokenWhere = `WHERE g.tagname = ? + AND z.tagname = ? AND s.tagname = ? + AND c.nspath = ? AND c.tag = ? + AND m.tag = ?` + + // MeasurementFourPartTokenWhere matches a local measurement token in the + // form token4.token5.token6.token7. Token6 is validated as "bay" before use. + MeasurementFourPartTokenWhere = `WHERE c.nspath = ? AND c.tag = ? + AND m.tag = ?` + + // MeasurementTwoPartTokenWhere matches the short measurement token format + // token4.token7 using component namespace path and measurement tag. + MeasurementTwoPartTokenWhere = `WHERE c.nspath = ? AND m.tag = ?` + + // MeasurementComponentByUUID returns the component hierarchy fields needed + // to construct a measurement's canonical name and seven-part ID. + MeasurementComponentByUUID = `SELECT global_uuid, nspath, + tag, grid, zone, station FROM component + WHERE global_uuid = ? LIMIT 1` + + // MeasurementGridTags returns every grid tag used to construct the first + // level of the measurement recommendation hierarchy. + MeasurementGridTags = `SELECT tagname FROM grid` + + // MeasurementZoneHierarchy returns zones together with their parent grid + // tags for building the grid-to-zone recommendation mapping. + MeasurementZoneHierarchy = `SELECT zone.*, + grid.tagname AS grid_tag FROM zone + LEFT JOIN grid ON zone.grid_id = grid.id` + + // MeasurementStationHierarchy returns stations together with their parent + // zone tags for building the zone-to-station recommendation mapping. + MeasurementStationHierarchy = `SELECT station.*, zone.tagname AS zone_tag + FROM station + LEFT JOIN zone ON station.zone_id = zone.id` + + // MeasurementComponentHierarchy returns components together with their + // parent station tags for building station, namespace, and component mappings. + MeasurementComponentHierarchy = `SELECT component.*, + station.tagname AS station_tag + FROM component LEFT JOIN station + ON component.station_id = station.id` + + // MeasurementTagHierarchy returns measurements together with their owning + // component tags for building the component-to-measurement mapping. + MeasurementTagHierarchy = `SELECT measurement.*, component.tag AS comp_tag + FROM measurement LEFT JOIN component + ON measurement.component_uuid = component.global_uuid` +) diff --git a/sql/topologic.go b/sql/topologic.go index f440e27..9c38eae 100644 --- a/sql/topologic.go +++ b/sql/topologic.go @@ -2,7 +2,7 @@ package sql // RecursiveSQL define topologic table recursive query statement -var RecursiveSQL = `WITH RECURSIVE recursive_tree as ( +const RecursiveSQL = `WITH RECURSIVE recursive_tree as ( SELECT uuid_from,uuid_to,flag FROM "topologic" WHERE uuid_from = ? @@ -16,7 +16,7 @@ var RecursiveSQL = `WITH RECURSIVE recursive_tree as ( // RecursiveTopologicByStartSQL returns every directed edge reachable from the // supplied start component. It tracks the visited node path inside PostgreSQL // so cycles in topologic data cannot recurse forever. -var RecursiveTopologicByStartSQL = `WITH RECURSIVE recursive_tree as ( +const RecursiveTopologicByStartSQL = `WITH RECURSIVE recursive_tree as ( SELECT uuid_from, uuid_to, flag, ARRAY[uuid_from, uuid_to] AS path FROM "topologic" WHERE uuid_from = ?