modelRT/handler/data_object_attribute_update.go

// Package handler provides HTTP handlers for various endpoints.
package handler

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"modelRT/client/manualsync"
	"modelRT/common"
	"modelRT/common/errcode"
	"modelRT/constants"
	"modelRT/database"
	"modelRT/diagram"
	"modelRT/logger"
	"modelRT/model"
	"modelRT/orm"
	redisrepository "modelRT/repository/redis"

	"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"`
}

const redisChangeRestoreTimeout = 5 * time.Second

// 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
		}
	}()

	redisClient := diagram.GetRedisClientInstance()
	redisChanges := redisrepository.NewRedisChangeSet(redisClient)
	canonicalRedisKey, err := model.ResolveDataObjectRedisKey(
		ctx,
		redisClient,
		dataObjectType,
		request.Token,
	)
	message := "data-object attribute update success"
	var measurementResult measurementUpdateResult
	switch {
	case err != nil:
		// The shared resolver error is handled by the common failure path below.
	case dataObjectType == constants.DataObjectTypeParameter:
		parameter, queryErr := database.QueryParameterByDataObjectToken(ctx, tx, request.Token)
		if queryErr == nil {
			queryErr = database.UpdateParameterDataObjectValue(ctx, tx, parameter, value)
		}
		if queryErr == nil {
			queryErr = redisChanges.AddHashChange(ctx, canonicalRedisKey, field, value)
		}
		err = queryErr
	case dataObjectType == constants.DataObjectTypeMeasurement:
		measurementResult, err = updateMeasurementDataObject(ctx, tx, request.Token, field, value, request.Data, measurementUpdateDependencies{
			writeManualValueFunc: func(ctx context.Context, measurement *orm.Measurement, value float64, timestamp time.Time) error {
				key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
				if err != nil {
					return fmt.Errorf("generate measurement redis key: %w", err)
				}
				return redisChanges.AddMeasurementValueChange(ctx, key, value, timestamp, false)
			},
			syncManualChangeFunc: manualsync.Sync,
			replaceRedisValueFunc: func(ctx context.Context, measurement *orm.Measurement, value float64, timestamp time.Time) error {
				key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
				if err != nil {
					return fmt.Errorf("generate measurement redis key: %w", err)
				}
				return redisChanges.AddMeasurementValueChange(ctx, key, value, timestamp, true)
			},
			nowFunc: time.Now,
		})
		if err == nil && measurementResult.modeChanged {
			err = redisChanges.AddHashChange(ctx, canonicalRedisKey, "mode", measurementResult.mode)
		}
		message = measurementResult.message
	default:
		err = fmt.Errorf("unsupported data object type %q", dataObjectType)
	}

	if err != nil {
		_ = tx.Rollback().Error
		if measurementResult.recordFailureOnError {
			if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); 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 := redisChanges.Apply(ctx); err != nil {
		_ = tx.Rollback().Error
		if measurementResult.recordFailureOnError {
			if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); logErr != nil {
				logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
			}
		}
		logger.Error(ctx, "apply redis data-object changes failed", "token", request.Token, "field", field, "error", err)
		renderRespFailure(c, constants.RespCodeFailed, err.Error(), nil)
		return
	}

	if err := tx.Commit().Error; err != nil {
		revertCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), redisChangeRestoreTimeout)
		defer cancel()
		if redisErr := redisChanges.Revert(revertCtx); redisErr != nil {
			logger.Error(ctx, "revert redis data-object changes failed", "token", request.Token, "field", field, "error", redisErr)
		}
		if measurementResult.recordFailureOnError {
			if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); logErr != nil {
				logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
			}
		}
		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, time.Time) error

type measurementManualChangeSyncer func(context.Context, orm.JSONMap, int16, *manualsync.SyntheticData) error

type measurementRedisValueReplacer func(context.Context, *orm.Measurement, float64, time.Time) error

type measurementUpdateDependencies struct {
	writeManualValueFunc  measurementManualValueWriter
	syncManualChangeFunc  measurementManualChangeSyncer
	replaceRedisValueFunc measurementRedisValueReplacer
	nowFunc               func() time.Time
}

type measurementUpdateResult struct {
	message              string
	measurementID        int64
	value                float64
	recordFailureOnError bool
	mode                 int16
	modeChanged          bool
	operationTime        time.Time
}

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
		}
		operationTime := measurementUpdateNow(dependencies)
		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, operationTime); err != nil {
			return measurementUpdateResult{}, err
		}
		syncManualMeasurementChange(
			ctx,
			dependencies.syncManualChangeFunc,
			lockedMeasurement.ID,
			lockedMeasurement.DataSource,
			mode,
			nil,
		)
		if currentMode && mode == constants.MeasurementModeManual {
			if manualValue != nil {
				if dependencies.replaceRedisValueFunc == nil {
					return measurementUpdateResult{}, fmt.Errorf("measurement redis value replacer is nil")
				}
				if err := dependencies.replaceRedisValueFunc(ctx, &lockedMeasurement, *manualValue, operationTime); err != nil {
					return measurementUpdateResult{}, fmt.Errorf("replace measurement redis value: %w", err)
				}
			}
		}
		return measurementUpdateResult{
			message:       fmt.Sprintf("measurement mode changed to %s", measurementModeName(mode)),
			mode:          mode,
			modeChanged:   true,
			operationTime: operationTime,
		}, 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)
		}
		operationTime := measurementUpdateNow(dependencies)
		failureResult := measurementUpdateResult{
			measurementID:        lockedMeasurement.ID,
			value:                manualValue,
			recordFailureOnError: true,
			operationTime:        operationTime,
		}
		if dependencies.writeManualValueFunc == nil {
			return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement manual value writer is nil"))
		}
		if err := dependencies.writeManualValueFunc(ctx, &lockedMeasurement, manualValue, operationTime); err != nil {
			return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
		}
		sample := manualsync.SyntheticData{Time: operationTime.UnixNano(), Value: manualValue}
		syncManualMeasurementChange(
			ctx,
			dependencies.syncManualChangeFunc,
			lockedMeasurement.ID,
			lockedMeasurement.DataSource,
			constants.MeasurementModeManual,
			&sample,
		)
		if err := database.AppendMeasurementValueOperation(ctx, tx, lockedMeasurement.ID, 0, manualValue, operationTime); err != nil {
			return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
		}
		failureResult.message = "measurement manual value updated"
		return failureResult, nil
	default:
		return measurementUpdateResult{}, fmt.Errorf("unsupported measurement update field %q", field)
	}
}

// TODO: This synchronous best-effort manual synchronization is not necessarily
// the final implementation. Its current behavior is to keep the local update
// successful when the downstream request fails (or the client is unavailable),
// log the error, and permanently drop that synchronization event without retry.
// The HTTP timeout still adds latency to the update request, and downstream state
// may diverge from local state. Revisit asynchronous delivery or a durable outbox
// if delivery reliability or request latency becomes important.
func syncManualMeasurementChange(
	ctx context.Context,
	syncer measurementManualChangeSyncer,
	measurementID int64,
	dataSource orm.JSONMap,
	mode int16,
	sample *manualsync.SyntheticData,
) {
	if syncer == nil {
		logManualMeasurementSyncError(ctx, "manual measurement synchronization skipped because sync client is unavailable",
			"measurement_id", measurementID,
			"mode", mode,
			"has_data", sample != nil,
		)
		return
	}
	if err := syncer(ctx, dataSource, mode, sample); err != nil {
		logManualMeasurementSyncError(ctx, "manual measurement synchronization failed; local update will continue",
			"measurement_id", measurementID,
			"mode", mode,
			"has_data", sample != nil,
			"error", err,
		)
	}
}

func logManualMeasurementSyncError(ctx context.Context, message string, fields ...any) {
	// The application initializes logging before serving requests. This guard keeps
	// the best-effort path safe in isolated unit tests and other pre-init callers.
	if logger.GetLoggerInstance() == nil {
		return
	}
	logger.Error(ctx, message, fields...)
}

func measurementUpdateNow(dependencies measurementUpdateDependencies) time.Time {
	if dependencies.nowFunc != nil {
		return dependencies.nowFunc().UTC()
	}
	return time.Now().UTC()
}

func measurementFailureTime(result measurementUpdateResult) time.Time {
	if !result.operationTime.IsZero() {
		return result.operationTime
	}
	return time.Now().UTC()
}

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