package model import ( "bytes" "context" "database/sql" "encoding/json" "fmt" "strings" "modelRT/constants" "modelRT/diagram" "modelRT/logger" "github.com/redis/go-redis/v9" ) const parameterDataObjectPipelineSize = 500 type parameterDataObjectHash struct { Key string Aliases []string Fields map[string]any } // ParameterInitializationRecord contains one parameter attribute together with // the hierarchy and metadata needed to create its Redis data-object hashes. type ParameterInitializationRecord struct { GridTag string `gorm:"column:grid_tag"` ZoneTag string `gorm:"column:zone_tag"` StationTag string `gorm:"column:station_tag"` StationIsLocal bool `gorm:"column:station_is_local"` ComponentUUID string `gorm:"column:component_uuid"` ComponentNSPath string `gorm:"column:component_nspath"` ComponentTag string `gorm:"column:component_tag"` AttributeGroup string `gorm:"column:attribute_group"` AttributeName string `gorm:"column:attribute_name"` AttributeValue string `gorm:"column:attribute_value"` AttributeType string `gorm:"column:attribute_type"` Description sql.NullString `gorm:"column:description"` DescriptionCount int64 `gorm:"column:description_count"` DynamicRecordCount int64 `gorm:"column:dynamic_record_count"` } // InitializeParameterDataObjects creates one full-token Redis hash and full or // local-short token aliases for each parameter loaded from PostgreSQL. func InitializeParameterDataObjects(ctx context.Context, records []ParameterInitializationRecord) error { hashes, err := buildParameterDataObjectHashes(records) if err != nil { return fmt.Errorf("build parameter data-object hashes: %w", err) } if err := storeParameterDataObjectHashes(ctx, diagram.GetRedisClientInstance(), hashes); err != nil { return fmt.Errorf("store parameter data-object hashes in redis: %w", err) } logger.Info(ctx, "initialize parameter data objects completed", "postgres_record_count", len(records), "redis_hash_count", len(hashes), ) return nil } func buildParameterDataObjectHashes(records []ParameterInitializationRecord) ([]parameterDataObjectHash, error) { hashes := make([]parameterDataObjectHash, 0, len(records)) seenKeys := make(map[string]string, len(records)*2) for _, record := range records { value, err := parameterRedisValue(record.AttributeValue) if err != nil { return nil, fmt.Errorf( "decode value for component %q group %q attribute %q: %w", record.ComponentTag, record.AttributeGroup, record.AttributeName, err, ) } fullToken := strings.Join([]string{ record.GridTag, record.ZoneTag, record.StationTag, record.ComponentNSPath, record.ComponentTag, record.AttributeGroup, record.AttributeName, }, ".") shortToken := strings.Join([]string{ record.ComponentNSPath, record.ComponentTag, record.AttributeGroup, record.AttributeName, }, ".") aliases := []string{fullToken} if record.StationIsLocal { aliases = append(aliases, shortToken) } fields := map[string]any{ "value": value, "meta": "PARAM", "type": record.AttributeType, "name": shortToken, "description": record.Description.String, "id": fullToken, } owner := strings.Join([]string{ record.ComponentUUID, record.AttributeGroup, record.AttributeName, }, "/") for _, alias := range aliases { if err := validateInitializedParameterToken(alias); err != nil { return nil, err } if existingOwner, exists := seenKeys[alias]; exists { return nil, fmt.Errorf( "ambiguous parameter token %q is produced by %q and %q", alias, existingOwner, owner, ) } seenKeys[alias] = owner } hashes = append(hashes, parameterDataObjectHash{Key: fullToken, Aliases: aliases, Fields: fields}) } return hashes, nil } func validateInitializedParameterToken(token string) error { dataObjectType, err := ClassifyDataObjectToken(token) if err != nil { return fmt.Errorf("generated invalid parameter token %q: %w", token, err) } if dataObjectType != constants.DataObjectTypeParameter { return fmt.Errorf("generated token %q is not a parameter", token) } return nil } func parameterRedisValue(rawJSON string) (any, error) { decoder := json.NewDecoder(bytes.NewBufferString(rawJSON)) decoder.UseNumber() var value any if err := decoder.Decode(&value); err != nil { return nil, err } switch typedValue := value.(type) { case nil: return "null", nil case string: return typedValue, nil case json.Number: return typedValue.String(), nil case bool: return typedValue, nil default: encoded, err := json.Marshal(typedValue) if err != nil { return nil, err } return string(encoded), nil } } func storeParameterDataObjectHashes( ctx context.Context, rdb *redis.Client, hashes []parameterDataObjectHash, ) error { if rdb == nil { return fmt.Errorf("redis client is nil") } oldKeys, err := rdb.SMembers(ctx, constants.RedisParameterDataObjectKeySet).Result() if err != nil { return fmt.Errorf("query previously initialized parameter keys: %w", err) } oldAliasKeys, err := rdb.SMembers(ctx, constants.RedisParameterDataObjectAliasKeySet).Result() if err != nil { return fmt.Errorf("query previously initialized parameter alias keys: %w", err) } currentKeys := make(map[string]struct{}, len(hashes)) currentAliasKeys := make(map[string]struct{}, len(hashes)*2) for start := 0; start < len(hashes); start += parameterDataObjectPipelineSize { end := min(start+parameterDataObjectPipelineSize, len(hashes)) pipeline := rdb.TxPipeline() keyMembers := make([]any, 0, end-start) aliasKeyMembers := make([]any, 0, (end-start)*2) for _, hash := range hashes[start:end] { pipeline.Del(ctx, hash.Key) pipeline.HSet(ctx, hash.Key, hash.Fields) keyMembers = append(keyMembers, hash.Key) currentKeys[hash.Key] = struct{}{} for _, alias := range hash.Aliases { aliasKey, err := DataObjectRedisAliasKey(constants.DataObjectTypeParameter, alias) if err != nil { return err } pipeline.Set(ctx, aliasKey, hash.Key, 0) aliasKeyMembers = append(aliasKeyMembers, aliasKey) currentAliasKeys[aliasKey] = struct{}{} } } if len(keyMembers) > 0 { pipeline.SAdd(ctx, constants.RedisParameterDataObjectKeySet, keyMembers...) } if len(aliasKeyMembers) > 0 { pipeline.SAdd(ctx, constants.RedisParameterDataObjectAliasKeySet, aliasKeyMembers...) } if _, err := pipeline.Exec(ctx); err != nil { return fmt.Errorf("write parameter data-object hash batch starting at %d: %w", start, err) } } if err := cleanupStaleParameterDataObjectKeys( ctx, rdb, oldKeys, oldAliasKeys, currentKeys, currentAliasKeys, ); err != nil { return err } return nil } func cleanupStaleParameterDataObjectKeys( ctx context.Context, rdb *redis.Client, oldKeys, oldAliasKeys []string, currentKeys, currentAliasKeys map[string]struct{}, ) error { pipeline := rdb.TxPipeline() for _, key := range oldKeys { if _, exists := currentKeys[key]; exists { continue } pipeline.Del(ctx, key) pipeline.SRem(ctx, constants.RedisParameterDataObjectKeySet, key) } for _, aliasKey := range oldAliasKeys { if _, exists := currentAliasKeys[aliasKey]; exists { continue } pipeline.Del(ctx, aliasKey) pipeline.SRem(ctx, constants.RedisParameterDataObjectAliasKeySet, aliasKey) } if len(currentKeys) == 0 { pipeline.Del(ctx, constants.RedisParameterDataObjectKeySet) } if len(currentAliasKeys) == 0 { pipeline.Del(ctx, constants.RedisParameterDataObjectAliasKeySet) } if _, err := pipeline.Exec(ctx); err != nil { return fmt.Errorf("remove stale parameter data-object keys: %w", err) } return nil }