modelRT/model/parameter_data_object_init.go

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
	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 full and local-short Redis hashes for
// parameter attributes previously 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)*2)
	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,
		}, ".")

		if err := validateInitializedParameterToken(fullToken); err != nil {
			return nil, err
		}
		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,
		}, "/")
		if err := appendParameterDataObjectHash(&hashes, seenKeys, fullToken, owner, fields); err != nil {
			return nil, err
		}

		if !record.StationIsLocal {
			continue
		}
		if err := validateInitializedParameterToken(shortToken); err != nil {
			return nil, err
		}
		if err := appendParameterDataObjectHash(&hashes, seenKeys, shortToken, owner, fields); err != nil {
			return nil, err
		}
	}
	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 appendParameterDataObjectHash(
	hashes *[]parameterDataObjectHash,
	seenKeys map[string]string,
	key string,
	owner string,
	fields map[string]any,
) error {
	if existingOwner, exists := seenKeys[key]; exists {
		return fmt.Errorf(
			"ambiguous parameter token %q is produced by %q and %q",
			key,
			existingOwner,
			owner,
		)
	}
	seenKeys[key] = owner
	*hashes = append(*hashes, parameterDataObjectHash{Key: key, Fields: fields})
	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)
	}
	cleanupPipeline := rdb.TxPipeline()
	for start := 0; start < len(oldKeys); start += parameterDataObjectPipelineSize {
		end := min(start+parameterDataObjectPipelineSize, len(oldKeys))
		cleanupPipeline.Del(ctx, oldKeys[start:end]...)
	}
	cleanupPipeline.Del(ctx, constants.RedisParameterDataObjectKeySet)
	if _, err := cleanupPipeline.Exec(ctx); err != nil {
		return fmt.Errorf("remove stale parameter data-object hashes: %w", err)
	}

	for start := 0; start < len(hashes); start += parameterDataObjectPipelineSize {
		end := min(start+parameterDataObjectPipelineSize, len(hashes))
		pipeline := rdb.TxPipeline()
		keyMembers := make([]any, 0, end-start)
		for _, hash := range hashes[start:end] {
			pipeline.HSet(ctx, hash.Key, hash.Fields)
			keyMembers = append(keyMembers, hash.Key)
		}
		if len(keyMembers) > 0 {
			pipeline.SAdd(ctx, constants.RedisParameterDataObjectKeySet, keyMembers...)
		}
		if _, err := pipeline.Exec(ctx); err != nil {
			return fmt.Errorf("write parameter data-object hash batch starting at %d: %w", start, err)
		}
	}
	return nil
}