package httpapi

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net/http"
	"strconv"
	"strings"
	"time"

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

const kelingOmniCompatibilityMarker = "keling_omni_v1"

type kelingCompatRequestIDKey struct{}

type KelingOmniVideoRequest struct {
	ModelName     string                   `json:"model_name" example:"kling-v3-omni"`
	Prompt        string                   `json:"prompt" example:"A quiet street in the rain with natural ambient sound"`
	MultiShot     bool                     `json:"multi_shot" example:"false"`
	ShotType      string                   `json:"shot_type,omitempty" example:"customize"`
	MultiPrompt   []KelingOmniMultiPrompt  `json:"multi_prompt,omitempty"`
	ImageList     []KelingOmniImageInput   `json:"image_list,omitempty"`
	ElementList   []KelingOmniElementInput `json:"element_list,omitempty"`
	VideoList     []KelingOmniVideoInput   `json:"video_list,omitempty"`
	Sound         string                   `json:"sound" enums:"on,off" example:"on"`
	Mode          string                   `json:"mode" enums:"std,pro,4k" example:"pro"`
	AspectRatio   string                   `json:"aspect_ratio" enums:"16:9,9:16,1:1" example:"9:16"`
	Duration      any                      `json:"duration" swaggertype:"string" example:"5"`
	WatermarkInfo KelingOmniWatermarkInfo  `json:"watermark_info,omitempty"`
	CallbackURL   string                   `json:"callback_url,omitempty"`
	ExternalTask  string                   `json:"external_task_id,omitempty"`
}

type KelingOmniMultiPrompt struct {
	Index    int    `json:"index" example:"1"`
	Prompt   string `json:"prompt" example:"A wide establishing shot"`
	Duration any    `json:"duration" swaggertype:"string" example:"3"`
}

type KelingOmniImageInput struct {
	ImageURL string `json:"image_url"`
	Type     string `json:"type,omitempty" enums:"first_frame,end_frame"`
}

type KelingOmniElementInput struct {
	ElementID any `json:"element_id"`
}

type KelingOmniVideoInput struct {
	VideoURL          string `json:"video_url"`
	ReferType         string `json:"refer_type,omitempty" enums:"base,feature"`
	KeepOriginalSound string `json:"keep_original_sound,omitempty" enums:"yes,no"`
}

type KelingOmniWatermarkInfo struct {
	Enabled bool `json:"enabled" example:"false"`
}

type KelingCompatibleEnvelope struct {
	Code      int    `json:"code" example:"0"`
	Message   string `json:"message" example:"SUCCEED"`
	RequestID string `json:"request_id"`
	Data      any    `json:"data,omitempty"`
}

type kelingCompatError struct {
	HTTPStatus int
	Code       int
	Message    string
	RequestID  string
}

func (e *kelingCompatError) Error() string {
	if e == nil {
		return "keling compatibility error"
	}
	return e.Message
}

func newKelingCompatError(status int, code int, message string) *kelingCompatError {
	return &kelingCompatError{HTTPStatus: status, Code: code, Message: message}
}

func (s *Server) requireKelingAPIKey(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requestID := newKelingCompatRequestID(r)
		r = r.WithContext(context.WithValue(r.Context(), kelingCompatRequestIDKey{}, requestID))
		user, err := s.auth.Authenticate(r)
		if err != nil {
			code := 1002
			message := "Authorization is invalid"
			if strings.TrimSpace(r.Header.Get("Authorization")) == "" {
				code = 1001
				message = "Authorization is required"
			}
			writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusUnauthorized, code, message))
			return
		}
		if user == nil || strings.TrimSpace(user.APIKeyID) == "" {
			writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusUnauthorized, 1002, "a Gateway API Key is required"))
			return
		}
		if !apiKeyScopeAllowed(user, "videos.generations") {
			writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusForbidden, 1103, "API Key scope does not allow video generation"))
			return
		}
		next.ServeHTTP(w, r.WithContext(auth.WithUser(r.Context(), user)))
	})
}

// createKelingOmniVideo godoc
// @Summary 创建 Kling Omni 视频任务
// @Description 兼容 Kling 旧版 /v1/videos/omni-video；Bearer token 必须为 Gateway API Key。任务固定异步执行，返回的 task_id 为网关任务 ID。
// @Tags kling-compatible
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param input body KelingOmniVideoRequest true "Kling Omni 官方兼容请求"
// @Success 200 {object} KelingCompatibleEnvelope
// @Failure 400 {object} KelingCompatibleEnvelope
// @Failure 401 {object} KelingCompatibleEnvelope
// @Failure 403 {object} KelingCompatibleEnvelope
// @Failure 429 {object} KelingCompatibleEnvelope
// @Failure 500 {object} KelingCompatibleEnvelope
// @Failure 503 {object} KelingCompatibleEnvelope
// @Router /api/v1/videos/omni-video [post]
func (s *Server) createKelingOmniVideo(w http.ResponseWriter, r *http.Request) {
	requestID := kelingCompatRequestID(r)
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusUnauthorized, 1002, "Authorization is invalid"))
		return
	}
	body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations")
	if err != nil {
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusBadRequest, 1201, err.Error()))
		return
	}
	normalized, compatErr := normalizeKelingOmniRequest(body)
	if compatErr != nil {
		writeKelingCompatError(w, requestID, compatErr)
		return
	}
	model := strings.TrimSpace(stringFromKelingCompat(normalized["model"]))
	if normalized["resolution"] == "2160p" {
		candidates, candidateErr := s.store.ListModelCandidates(r.Context(), model, "omni_video", user)
		if candidateErr != nil {
			writeKelingCompatError(w, requestID, kelingCompatGatewayError(candidateErr))
			return
		}
		if !kelingCompatCandidatesSupport4K(candidates) {
			writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusBadRequest, 1201, "mode=4k is not enabled by the selected model capabilities"))
			return
		}
	}
	task, createErr := s.prepareAndCreateGatewayTask(
		r.Context(),
		r,
		user,
		"videos.generations",
		model,
		normalized,
		true,
	)
	if createErr != nil {
		var staged *gatewayTaskCreationError
		if errors.As(createErr, &staged) && staged.Stage == gatewayTaskCreationPrepare {
			writeKelingCompatError(w, requestID, kelingCompatGatewayError(staged.Err))
			return
		}
		s.logger.Error("create Kling-compatible task failed", "error", createErr)
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusInternalServerError, 5000, "create task failed"))
		return
	}
	if err := s.store.SetTaskCompatibilitySubmission(r.Context(), task.ID, store.CompatibilitySubmission{
		TargetProtocol: clients.ProtocolKlingV1Omni,
		PublicID:       task.ID,
	}); err != nil {
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusInternalServerError, 5000, "create compatibility metadata failed"))
		return
	}
	if err := s.runner.EnqueueAsyncTask(r.Context(), task); err != nil {
		s.logger.Error("enqueue Kling-compatible task failed", "taskId", task.ID, "error", err)
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusServiceUnavailable, 5001, "video task queue is unavailable"))
		return
	}
	task, createErr = s.waitForCompatibilitySubmission(r, task)
	if createErr != nil {
		if wire := clients.ErrorWireResponse(createErr); wireResponseMatches(wire, clients.ProtocolKlingV1Omni) {
			writeWireResponse(w, wire)
			return
		}
		writeKelingCompatError(w, requestID, kelingCompatGatewayError(createErr))
		return
	}
	if kelingTaskUsesNativeProtocol(task) && len(task.CompatibilitySubmitBody) > 0 {
		writeJSON(w, task.CompatibilitySubmitHTTPStatus, task.CompatibilitySubmitBody)
		return
	}
	writeJSON(w, http.StatusOK, KelingCompatibleEnvelope{
		Code:      0,
		Message:   "SUCCEED",
		RequestID: requestID,
		Data:      kelingCompatTaskData(task),
	})
}

// getKelingOmniVideo godoc
// @Summary 查询 Kling Omni 视频任务
// @Description 按创建接口返回的网关 task_id 查询任务；仅允许创建任务的 Gateway 用户访问。
// @Tags kling-compatible
// @Produce json
// @Security BearerAuth
// @Param taskID path string true "网关任务 ID"
// @Success 200 {object} KelingCompatibleEnvelope
// @Failure 401 {object} KelingCompatibleEnvelope
// @Failure 403 {object} KelingCompatibleEnvelope
// @Failure 404 {object} KelingCompatibleEnvelope
// @Failure 500 {object} KelingCompatibleEnvelope
// @Router /api/v1/videos/omni-video/{taskID} [get]
func (s *Server) getKelingOmniVideo(w http.ResponseWriter, r *http.Request) {
	requestID := kelingCompatRequestID(r)
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusUnauthorized, 1002, "Authorization is invalid"))
		return
	}
	task, err := s.store.GetCompatibilityTask(r.Context(), clients.ProtocolKlingV1Omni, strings.TrimSpace(r.PathValue("taskID")))
	if err != nil && store.IsNotFound(err) {
		task, err = s.store.GetTask(r.Context(), strings.TrimSpace(r.PathValue("taskID")))
	}
	if err != nil {
		if store.IsNotFound(err) {
			writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusNotFound, 1203, "task not found"))
			return
		}
		s.logger.Error("get Kling-compatible task failed", "error", err)
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusInternalServerError, 5000, "query task failed"))
		return
	}
	if !kelingCompatTaskOwnedBy(task, user) || !isKelingCompatTask(task) {
		writeKelingCompatError(w, requestID, newKelingCompatError(http.StatusNotFound, 1203, "task not found"))
		return
	}
	if kelingTaskUsesNativeProtocol(task) {
		if raw, ok := task.Result["raw"].(map[string]any); ok && len(raw) > 0 {
			writeJSON(w, http.StatusOK, raw)
			return
		}
		if len(task.RemoteTaskPayload) > 0 {
			writeJSON(w, http.StatusOK, task.RemoteTaskPayload)
			return
		}
	}
	writeJSON(w, http.StatusOK, KelingCompatibleEnvelope{
		Code:      0,
		Message:   "SUCCEED",
		RequestID: requestID,
		Data:      kelingCompatTaskData(task),
	})
}

func normalizeKelingOmniRequest(input map[string]any) (map[string]any, *kelingCompatError) {
	if input == nil {
		input = map[string]any{}
	}
	if callbackURL := strings.TrimSpace(stringFromKelingCompat(input["callback_url"])); callbackURL != "" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "callback_url is not supported by this Gateway endpoint")
	}
	requestedModel := strings.TrimSpace(stringFromKelingCompat(input["model_name"]))
	if requestedModel == "" {
		requestedModel = "kling-video-o1"
	}
	model, maxDuration, ok := kelingCompatModel(requestedModel)
	if !ok {
		return nil, newKelingCompatError(http.StatusNotFound, 1203, "unsupported model_name: "+requestedModel)
	}

	mode := strings.ToLower(strings.TrimSpace(stringFromKelingCompat(input["mode"])))
	if mode == "" {
		mode = "pro"
	}
	resolutionByMode := map[string]string{"std": "720p", "pro": "1080p", "4k": "2160p"}
	resolution := resolutionByMode[mode]
	if resolution == "" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "mode must be std, pro, or 4k")
	}

	sound := strings.ToLower(strings.TrimSpace(stringFromKelingCompat(input["sound"])))
	if sound == "" {
		sound = "off"
	}
	if sound != "on" && sound != "off" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "sound must be on or off")
	}
	if model == klingO1Model && sound == "on" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "kling-video-o1 does not support generated audio; sound must be off")
	}

	content := make([]any, 0)
	prompt := strings.TrimSpace(stringFromKelingCompat(input["prompt"]))
	images, hasFirstFrame, imageErr := normalizeKelingImageList(input["image_list"])
	if imageErr != nil {
		return nil, imageErr
	}
	content = append(content, images...)
	elements, elementErr := normalizeKelingElementList(input["element_list"])
	if elementErr != nil {
		return nil, elementErr
	}
	content = append(content, elements...)
	videos, hasBaseVideo, hasVideo, videoErr := normalizeKelingVideoList(input["video_list"])
	if videoErr != nil {
		return nil, videoErr
	}
	content = append(content, videos...)
	if hasVideo && sound == "on" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "sound must be off when video_list is provided")
	}

	multiShot, multiShotPresent, boolErr := kelingCompatOptionalBool(input, "multi_shot")
	if boolErr != nil {
		return nil, boolErr
	}
	shotType := strings.ToLower(strings.TrimSpace(stringFromKelingCompat(input["shot_type"])))
	multiPrompts, shotDuration, multiPromptErr := normalizeKelingMultiPrompts(input["multi_prompt"])
	if multiPromptErr != nil {
		return nil, multiPromptErr
	}
	if !multiShotPresent {
		multiShot = false
	}
	if multiShot {
		if shotType != "customize" && shotType != "intelligence" {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "shot_type must be customize or intelligence when multi_shot is true")
		}
		if shotType == "customize" && len(multiPrompts) == 0 {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "multi_prompt is required for customized multi-shot generation")
		}
		if shotType == "intelligence" && len(multiPrompts) > 0 {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "multi_prompt is only supported when shot_type is customize")
		}
	} else if len(multiPrompts) > 0 || shotType != "" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "shot_type and multi_prompt require multi_shot=true")
	}
	if (len(multiPrompts) == 0 || shotType == "intelligence") && prompt == "" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "prompt is required")
	}
	if prompt != "" {
		content = append([]any{map[string]any{"type": "text", "text": prompt}}, content...)
	}
	content = append(content, multiPrompts...)

	duration, durationProvided, durationErr := kelingCompatOptionalInt(input, "duration")
	if durationErr != nil {
		return nil, durationErr
	}
	if hasBaseVideo {
		if durationProvided {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "duration is not supported for base video editing")
		}
	} else {
		if len(multiPrompts) > 0 {
			if durationProvided && duration != shotDuration {
				return nil, newKelingCompatError(http.StatusBadRequest, 1201, "duration must equal the sum of multi_prompt durations")
			}
			duration = shotDuration
		} else if !durationProvided {
			duration = 5
		}
		if duration < 3 || duration > maxDuration {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("duration for %s must be an integer between 3 and %d seconds", requestedModel, maxDuration))
		}
		if model == klingO1Model && (len(images) == 0 || hasFirstFrame) && duration != 5 && duration != 10 {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "kling-video-o1 text-to-video and first-frame generation only support 5 or 10 seconds")
		}
	}

	aspectRatio := strings.TrimSpace(stringFromKelingCompat(input["aspect_ratio"]))
	if aspectRatio != "" && aspectRatio != "16:9" && aspectRatio != "9:16" && aspectRatio != "1:1" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "aspect_ratio must be 16:9, 9:16, or 1:1")
	}
	if hasFirstFrame || hasBaseVideo {
		if aspectRatio != "" {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, "aspect_ratio is not supported with a first frame or base video")
		}
	} else if aspectRatio == "" {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, "aspect_ratio is required when no first frame or base video is provided")
	}

	watermarkEnabled, watermarkErr := kelingCompatWatermarkEnabled(input["watermark_info"])
	if watermarkErr != nil {
		return nil, watermarkErr
	}
	externalTaskID := strings.TrimSpace(stringFromKelingCompat(input["external_task_id"]))
	normalized := map[string]any{
		"model":                    model,
		"model_name":               requestedModel,
		"modelType":                "omni_video",
		"runMode":                  "real",
		"content":                  content,
		"resolution":               resolution,
		"mode":                     mode,
		"sound":                    sound,
		"audio":                    sound == "on",
		"multi_shot":               multiShot,
		"watermark":                watermarkEnabled,
		"watermark_info":           map[string]any{"enabled": watermarkEnabled},
		"external_task_id":         externalTaskID,
		"_gateway_compatibility":   kelingOmniCompatibilityMarker,
		"_gateway_target_protocol": clients.ProtocolKlingV1Omni,
	}
	if prompt != "" {
		normalized["prompt"] = prompt
	}
	if shotType != "" {
		normalized["shot_type"] = shotType
	}
	if !hasBaseVideo {
		normalized["duration"] = duration
	}
	if aspectRatio != "" {
		normalized["aspect_ratio"] = aspectRatio
	}
	return normalized, nil
}

func normalizeKelingImageList(value any) ([]any, bool, *kelingCompatError) {
	items, err := kelingCompatObjectList(value, "image_list")
	if err != nil {
		return nil, false, err
	}
	out := make([]any, 0, len(items))
	hasFirstFrame := false
	hasEndFrame := false
	for index, item := range items {
		url := strings.TrimSpace(stringFromKelingCompat(item["image_url"]))
		if url == "" {
			return nil, false, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("image_list[%d].image_url is required", index))
		}
		frameType := strings.TrimSpace(stringFromKelingCompat(item["type"]))
		role := "reference_image"
		switch frameType {
		case "":
		case "first_frame":
			role = "first_frame"
			hasFirstFrame = true
		case "end_frame":
			role = "last_frame"
			hasEndFrame = true
		default:
			return nil, false, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("image_list[%d].type must be first_frame or end_frame", index))
		}
		out = append(out, map[string]any{
			"type":      "image_url",
			"role":      role,
			"image_url": map[string]any{"url": url},
		})
	}
	if hasEndFrame && !hasFirstFrame {
		return nil, false, newKelingCompatError(http.StatusBadRequest, 1201, "end_frame requires first_frame")
	}
	return out, hasFirstFrame, nil
}

func normalizeKelingElementList(value any) ([]any, *kelingCompatError) {
	items, err := kelingCompatObjectList(value, "element_list")
	if err != nil {
		return nil, err
	}
	out := make([]any, 0, len(items))
	for index, item := range items {
		id := item["element_id"]
		if strings.TrimSpace(stringFromKelingCompat(id)) == "" {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("element_list[%d].element_id is required", index))
		}
		out = append(out, map[string]any{"type": "element", "element": map[string]any{"element_id": id}})
	}
	return out, nil
}

func normalizeKelingVideoList(value any) ([]any, bool, bool, *kelingCompatError) {
	items, err := kelingCompatObjectList(value, "video_list")
	if err != nil {
		return nil, false, false, err
	}
	if len(items) > 1 {
		return nil, false, false, newKelingCompatError(http.StatusBadRequest, 1201, "video_list supports at most one video")
	}
	out := make([]any, 0, len(items))
	hasBase := false
	for index, item := range items {
		url := strings.TrimSpace(stringFromKelingCompat(item["video_url"]))
		if url == "" {
			return nil, false, false, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("video_list[%d].video_url is required", index))
		}
		referType := strings.ToLower(strings.TrimSpace(stringFromKelingCompat(item["refer_type"])))
		if referType == "" {
			referType = "base"
		}
		if referType != "base" && referType != "feature" {
			return nil, false, false, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("video_list[%d].refer_type must be base or feature", index))
		}
		keepSound := strings.ToLower(strings.TrimSpace(stringFromKelingCompat(item["keep_original_sound"])))
		if keepSound != "" && keepSound != "yes" && keepSound != "no" {
			return nil, false, false, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("video_list[%d].keep_original_sound must be yes or no", index))
		}
		nested := map[string]any{"url": url, "refer_type": referType}
		if keepSound != "" {
			nested["keep_original_sound"] = keepSound
		}
		role := "video_feature"
		if referType == "base" {
			role = "video_base"
			hasBase = true
		}
		out = append(out, map[string]any{"type": "video_url", "role": role, "video_url": nested})
	}
	return out, hasBase, len(items) > 0, nil
}

func normalizeKelingMultiPrompts(value any) ([]any, int, *kelingCompatError) {
	items, err := kelingCompatObjectList(value, "multi_prompt")
	if err != nil {
		return nil, 0, err
	}
	if len(items) > 6 {
		return nil, 0, newKelingCompatError(http.StatusBadRequest, 1201, "multi_prompt supports at most six shots")
	}
	out := make([]any, 0, len(items))
	seen := map[int]bool{}
	total := 0
	for index, item := range items {
		shotIndex, ok := kelingCompatInt(item["index"])
		if !ok || shotIndex < 1 || shotIndex > 6 || seen[shotIndex] {
			return nil, 0, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("multi_prompt[%d].index must be a unique integer from 1 to 6", index))
		}
		seen[shotIndex] = true
		prompt := strings.TrimSpace(stringFromKelingCompat(item["prompt"]))
		if prompt == "" {
			return nil, 0, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("multi_prompt[%d].prompt is required", index))
		}
		duration, ok := kelingCompatInt(item["duration"])
		if !ok || duration < 1 {
			return nil, 0, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("multi_prompt[%d].duration must be an integer of at least 1 second", index))
		}
		total += duration
		out = append(out, map[string]any{
			"type":       "text",
			"role":       "shot_prompt",
			"shot_index": shotIndex,
			"text":       prompt,
			"duration":   duration,
		})
	}
	return out, total, nil
}

func kelingCompatModel(value string) (string, int, bool) {
	model, ok := canonicalKlingOmniModel(value)
	if !ok {
		return "", 0, false
	}
	if model == klingO1Model {
		return model, 10, true
	}
	return model, 15, true
}

func kelingCompatObjectList(value any, field string) ([]map[string]any, *kelingCompatError) {
	if value == nil {
		return nil, nil
	}
	raw, ok := value.([]any)
	if !ok {
		return nil, newKelingCompatError(http.StatusBadRequest, 1201, field+" must be an array")
	}
	out := make([]map[string]any, 0, len(raw))
	for index, item := range raw {
		object, ok := item.(map[string]any)
		if !ok {
			return nil, newKelingCompatError(http.StatusBadRequest, 1201, fmt.Sprintf("%s[%d] must be an object", field, index))
		}
		out = append(out, object)
	}
	return out, nil
}

func kelingCompatOptionalInt(body map[string]any, key string) (int, bool, *kelingCompatError) {
	value, present := body[key]
	if !present || value == nil || strings.TrimSpace(stringFromKelingCompat(value)) == "" {
		return 0, false, nil
	}
	parsed, ok := kelingCompatInt(value)
	if !ok {
		return 0, true, newKelingCompatError(http.StatusBadRequest, 1201, key+" must be an integer")
	}
	return parsed, true, nil
}

func kelingCompatOptionalBool(body map[string]any, key string) (bool, bool, *kelingCompatError) {
	value, present := body[key]
	if !present || value == nil {
		return false, false, nil
	}
	parsed, ok := value.(bool)
	if !ok {
		return false, true, newKelingCompatError(http.StatusBadRequest, 1201, key+" must be a boolean")
	}
	return parsed, true, nil
}

func kelingCompatWatermarkEnabled(value any) (bool, *kelingCompatError) {
	if value == nil {
		return false, nil
	}
	object, ok := value.(map[string]any)
	if !ok {
		return false, newKelingCompatError(http.StatusBadRequest, 1201, "watermark_info must be an object")
	}
	enabled, present := object["enabled"]
	if !present {
		return false, nil
	}
	result, ok := enabled.(bool)
	if !ok {
		return false, newKelingCompatError(http.StatusBadRequest, 1201, "watermark_info.enabled must be a boolean")
	}
	return result, nil
}

func kelingCompatInt(value any) (int, bool) {
	switch typed := value.(type) {
	case int:
		return typed, true
	case int64:
		return int(typed), true
	case float64:
		if math.Abs(typed-math.Round(typed)) > 1e-9 {
			return 0, false
		}
		return int(math.Round(typed)), true
	case json.Number:
		parsed, err := strconv.Atoi(typed.String())
		return parsed, err == nil
	case string:
		parsed, err := strconv.Atoi(strings.TrimSpace(typed))
		return parsed, err == nil
	default:
		return 0, false
	}
}

func stringFromKelingCompat(value any) string {
	switch typed := value.(type) {
	case string:
		return typed
	case json.Number:
		return typed.String()
	case float64:
		if math.Abs(typed-math.Round(typed)) < 1e-9 {
			return strconv.FormatInt(int64(math.Round(typed)), 10)
		}
		return strconv.FormatFloat(typed, 'f', -1, 64)
	case int:
		return strconv.Itoa(typed)
	case int64:
		return strconv.FormatInt(typed, 10)
	default:
		return ""
	}
}

func kelingCompatTaskData(task store.GatewayTask) map[string]any {
	data := map[string]any{
		"task_id":     kelingCompatPublicID(task),
		"task_status": kelingCompatTaskStatus(task.Status),
		"task_info": map[string]any{
			"external_task_id": strings.TrimSpace(stringFromKelingCompat(task.Request["external_task_id"])),
		},
		"created_at": task.CreatedAt.UnixMilli(),
		"updated_at": task.UpdatedAt.UnixMilli(),
		"watermark_info": map[string]any{
			"enabled": kelingCompatTaskWatermark(task.Request),
		},
	}
	if message := kelingCompatTaskMessage(task); message != "" {
		data["task_status_msg"] = message
	}
	if kelingCompatTaskStatus(task.Status) == "failed" {
		data["task_status_code"] = kelingCompatBusinessCode(task.ErrorCode, kelingCompatTaskMessage(task))
	}
	if videos := kelingCompatTaskVideos(task.Result); len(videos) > 0 {
		data["task_result"] = map[string]any{"videos": videos}
	}
	return data
}

func kelingCompatPublicID(task store.GatewayTask) string {
	if publicID := strings.TrimSpace(task.CompatibilityPublicID); publicID != "" {
		return publicID
	}
	return task.ID
}

func kelingTaskUsesNativeProtocol(task store.GatewayTask) bool {
	return task.CompatibilityProtocol == clients.ProtocolKlingV1Omni && task.CompatibilitySourceProtocol == clients.ProtocolKlingV1Omni
}

func kelingCompatTaskStatus(status string) string {
	switch strings.ToLower(strings.TrimSpace(status)) {
	case "succeeded", "success", "completed":
		return "succeed"
	case "failed", "cancelled", "canceled":
		return "failed"
	case "running", "processing":
		return "processing"
	default:
		return "submitted"
	}
}

func kelingCompatTaskVideos(result map[string]any) []any {
	raw, _ := result["data"].([]any)
	out := make([]any, 0, len(raw))
	for _, itemValue := range raw {
		item, _ := itemValue.(map[string]any)
		if item == nil {
			continue
		}
		url := strings.TrimSpace(stringFromKelingCompat(firstKelingCompatValue(item["url"], item["video_url"])))
		if url == "" {
			continue
		}
		video := map[string]any{"url": url}
		if id := strings.TrimSpace(stringFromKelingCompat(item["id"])); id != "" {
			video["id"] = id
		}
		if watermarkURL := strings.TrimSpace(stringFromKelingCompat(item["watermark_url"])); watermarkURL != "" {
			video["watermark_url"] = watermarkURL
		}
		if duration := strings.TrimSpace(stringFromKelingCompat(item["duration"])); duration != "" {
			video["duration"] = duration
		}
		out = append(out, video)
	}
	return out
}

func firstKelingCompatValue(values ...any) any {
	for _, value := range values {
		if strings.TrimSpace(stringFromKelingCompat(value)) != "" {
			return value
		}
	}
	return nil
}

func kelingCompatTaskMessage(task store.GatewayTask) string {
	return strings.TrimSpace(firstNonEmpty(task.ErrorMessage, task.Error, task.Message))
}

func kelingCompatTaskWatermark(request map[string]any) bool {
	if value, ok := request["watermark"].(bool); ok {
		return value
	}
	if info, ok := request["watermark_info"].(map[string]any); ok {
		value, _ := info["enabled"].(bool)
		return value
	}
	return false
}

func kelingCompatCandidatesSupport4K(candidates []store.RuntimeModelCandidate) bool {
	for _, candidate := range candidates {
		if strings.ToLower(strings.TrimSpace(candidate.Provider)) != "keling" {
			continue
		}
		capability, _ := candidate.Capabilities["omni_video"].(map[string]any)
		if capability == nil {
			capability, _ = candidate.Capabilities["omni"].(map[string]any)
		}
		for _, resolution := range kelingCompatStringList(capability["output_resolutions"]) {
			switch strings.ToLower(strings.TrimSpace(resolution)) {
			case "2160p", "4k":
				return true
			}
		}
	}
	return false
}

func kelingCompatStringList(value any) []string {
	switch typed := value.(type) {
	case []string:
		return typed
	case []any:
		result := make([]string, 0, len(typed))
		for _, item := range typed {
			if text := strings.TrimSpace(stringFromKelingCompat(item)); text != "" {
				result = append(result, text)
			}
		}
		return result
	default:
		return nil
	}
}

func kelingCompatGatewayError(err error) *kelingCompatError {
	if err == nil {
		return newKelingCompatError(http.StatusInternalServerError, 5000, "unknown gateway error")
	}
	codeText := clients.ErrorCode(err)
	businessCode := kelingCompatBusinessCode(codeText, err.Error())
	status := http.StatusInternalServerError
	switch businessCode {
	case 1101:
		status = http.StatusPaymentRequired
	case 1103:
		status = http.StatusForbidden
	case 1201:
		status = http.StatusBadRequest
	case 1203:
		status = http.StatusNotFound
	case 1302, 1303:
		status = http.StatusTooManyRequests
	case 5001:
		status = http.StatusBadGateway
	}
	return newKelingCompatError(status, businessCode, err.Error())
}

func kelingCompatBusinessCode(errorCode string, message string) int {
	combined := strings.ToLower(strings.TrimSpace(errorCode + " " + message))
	containsAny := func(values ...string) bool {
		for _, value := range values {
			if strings.Contains(combined, value) {
				return true
			}
		}
		return false
	}
	switch {
	case containsAny("insufficient_balance", "insufficient balance", "balance_not_enough", "wallet balance", "余额不足", "欠费", "quota exceeded"):
		return 1101
	case containsAny("permission_denied", "permission denied", "forbidden", "access denied", "scope does not allow"):
		return 1103
	case containsAny("concurrent", "concurrency"):
		return 1303
	case containsAny("rate_limit", "rate limit", "too many requests", "rpm", "tpm"):
		return 1302
	case containsAny("no_model_candidate", "no model candidate", "model_not_found", "unsupported model", "resource not found"):
		return 1203
	case containsAny("invalid_parameter", "invalid parameter", "bad_request", "parameter_preprocessing", "duration", "aspect_ratio"):
		return 1201
	case containsAny("upload_", "request_asset_", "network", "timeout", "upstream", "service unavailable", "bad gateway"):
		return 5001
	default:
		return 5000
	}
}

func isKelingCompatTask(task store.GatewayTask) bool {
	return task.Kind == "videos.generations" && strings.TrimSpace(stringFromKelingCompat(task.Request["_gateway_compatibility"])) == kelingOmniCompatibilityMarker
}

func kelingCompatTaskOwnedBy(task store.GatewayTask, user *auth.User) bool {
	if user == nil {
		return false
	}
	taskOwner := strings.TrimSpace(firstNonEmpty(task.GatewayUserID, task.UserID))
	requestOwner := strings.TrimSpace(firstNonEmpty(user.GatewayUserID, user.ID))
	return taskOwner != "" && requestOwner != "" && taskOwner == requestOwner
}

func newKelingCompatRequestID(r *http.Request) string {
	if r != nil {
		if value := strings.TrimSpace(firstNonEmpty(r.Header.Get("X-Request-ID"), r.Header.Get("X-Request-Id"))); value != "" {
			return value
		}
	}
	random := make([]byte, 16)
	if _, err := rand.Read(random); err == nil {
		return hex.EncodeToString(random)
	}
	return strconv.FormatInt(time.Now().UnixNano(), 36)
}

func kelingCompatRequestID(r *http.Request) string {
	if r != nil {
		if value, ok := r.Context().Value(kelingCompatRequestIDKey{}).(string); ok && strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return newKelingCompatRequestID(r)
}

func writeKelingCompatError(w http.ResponseWriter, requestID string, err *kelingCompatError) {
	if err == nil {
		err = newKelingCompatError(http.StatusInternalServerError, 5000, "internal error")
	}
	if err.RequestID != "" {
		requestID = err.RequestID
	}
	if requestID == "" {
		requestID = newKelingCompatRequestID(nil)
	}
	status := err.HTTPStatus
	if status == 0 {
		status = http.StatusInternalServerError
	}
	writeJSON(w, status, KelingCompatibleEnvelope{
		Code:      err.Code,
		Message:   err.Message,
		RequestID: requestID,
	})
}