easyai-ai-gateway/apps/api/internal/runner/param_processor.go

package runner

import (
	"strings"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)

type paramProcessContext struct {
	kind            string
	modelCapability map[string]any
	candidate       store.RuntimeModelCandidate
	log             *parameterPreprocessingLog
	aspectRatio     string
	resolution      string
	err             error
}

type paramProcessor interface {
	Name() string
	ShouldProcess(params map[string]any, modelType string, context *paramProcessContext) bool
	Process(params map[string]any, modelType string, context *paramProcessContext) bool
}

type ParamProcessorChain struct {
	processors []paramProcessor
}

type parameterPreprocessResult struct {
	Body map[string]any
	Log  parameterPreprocessingLog
	Err  error
}

type parameterPreprocessingLog struct {
	ModelType string                      `json:"modelType"`
	Input     map[string]any              `json:"actualInput"`
	Output    map[string]any              `json:"convertedOutput"`
	Changed   bool                        `json:"changed"`
	Changes   []parameterPreprocessChange `json:"changes"`
	Model     map[string]any              `json:"model,omitempty"`
}

type parameterPreprocessChange struct {
	Processor       string `json:"processor"`
	Action          string `json:"action"`
	Path            string `json:"path"`
	Before          any    `json:"before"`
	After           any    `json:"after"`
	Reason          string `json:"reason"`
	CapabilityPath  string `json:"capabilityPath,omitempty"`
	CapabilityValue any    `json:"capabilityValue,omitempty"`
}

func NewParamProcessorChain() ParamProcessorChain {
	return ParamProcessorChain{
		processors: []paramProcessor{
			resolutionNormalizeProcessor{},
			aspectRatioProcessor{},
			messageContentProcessor{},
			contentFilterProcessor{},
			inputAudioProcessor{},
			durationProcessor{},
			audioProcessor{},
			imageCountProcessor{},
		},
	}
}

func preprocessRequest(kind string, body map[string]any, candidate store.RuntimeModelCandidate) map[string]any {
	return preprocessRequestWithLog(kind, body, candidate).Body
}

func preprocessRequestWithLog(kind string, body map[string]any, candidate store.RuntimeModelCandidate) parameterPreprocessResult {
	params := cloneMap(body)
	modelType := strings.TrimSpace(candidate.ModelType)
	if modelType == "" {
		modelType = modelTypeFromKind(kind, params)
	}
	log := parameterPreprocessingLog{
		ModelType: modelType,
		Input:     cloneMap(params),
		Changes:   []parameterPreprocessChange{},
		Model: map[string]any{
			"modelName":         candidate.ModelName,
			"modelAlias":        candidate.ModelAlias,
			"providerModelName": candidate.ProviderModelName,
			"provider":          candidate.Provider,
			"platformId":        candidate.PlatformID,
			"platformModelId":   candidate.PlatformModelID,
		},
	}
	context := &paramProcessContext{
		kind:            kind,
		modelCapability: effectiveModelCapability(candidate),
		candidate:       candidate,
		log:             &log,
	}
	if kind == "videos.generations" {
		ensureVideoContent(params, context)
	}
	chain := NewParamProcessorChain()
	processed := chain.Process(params, modelType, context)
	log.Output = cloneMap(processed)
	log.Changed = len(log.Changes) > 0
	return parameterPreprocessResult{Body: processed, Log: log, Err: context.err}
}

func (chain ParamProcessorChain) Process(params map[string]any, modelType string, context *paramProcessContext) map[string]any {
	if params == nil {
		return map[string]any{}
	}
	for _, processor := range chain.processors {
		if !processor.ShouldProcess(params, modelType, context) {
			continue
		}
		if !processor.Process(params, modelType, context) {
			break
		}
		if context != nil && context.err != nil {
			break
		}
	}
	return params
}

func (context *paramProcessContext) recordChange(processor string, action string, path string, before any, after any, reason string, capabilityPath string, capabilityValue any) {
	if context == nil || context.log == nil {
		return
	}
	context.log.Changes = append(context.log.Changes, parameterPreprocessChange{
		Processor:       processor,
		Action:          action,
		Path:            path,
		Before:          cloneAny(before),
		After:           cloneAny(after),
		Reason:          reason,
		CapabilityPath:  capabilityPath,
		CapabilityValue: cloneAny(capabilityValue),
	})
}

func (context *paramProcessContext) reject(processor string, path string, before any, reason string, capabilityPath string, capabilityValue any) bool {
	if context != nil {
		context.recordChange(processor, "reject", path, before, nil, reason, capabilityPath, capabilityValue)
		context.err = parameterValidationError(reason)
	}
	return false
}

type parameterValidationError string

func (e parameterValidationError) Error() string {
	return string(e)
}

func parameterPreprocessingMetrics(log parameterPreprocessingLog) map[string]any {
	return map[string]any{
		"parameterPreprocessingSummary": parameterPreprocessingSummary(log),
	}
}

func parameterPreprocessingSummary(log parameterPreprocessingLog) map[string]any {
	summary := map[string]any{
		"modelType":   log.ModelType,
		"changed":     log.Changed,
		"changeCount": len(log.Changes),
	}
	if len(log.Changes) == 0 {
		return summary
	}
	actions := make([]string, 0)
	paths := make([]string, 0)
	capabilityPaths := make([]string, 0)
	for _, change := range log.Changes {
		appendUniqueString(&actions, change.Action)
		appendUniqueString(&paths, change.Path)
		appendUniqueString(&capabilityPaths, change.CapabilityPath)
	}
	summary["actions"] = actions
	summary["paths"] = paths
	if len(capabilityPaths) > 0 {
		summary["capabilityPaths"] = capabilityPaths
	}
	return summary
}