package httpapi

import (
	"encoding/json"
	"errors"
	"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/runner"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)

const volcesContentsCompatibilityMarker = "volces_contents_generations_v3"

// createVolcesContentsGenerationTask godoc
// @Summary 创建火山内容生成任务
// @Description 统一公开入口兼容火山方舟内容生成任务。网关 task id 是查询与取消用的公开 id；上游 id 另以 upstream_task_id 保留。
// @Tags volces-compatible
// @Accept json
// @Produce json
// @Security BearerAuth
// @Success 200 {object} VolcesContentsGenerationTaskResponse
// @Failure 400 {object} VolcesErrorEnvelope
// @Failure 401 {object} VolcesErrorEnvelope
// @Failure 429 {object} VolcesErrorEnvelope
// @Failure 502 {object} VolcesErrorEnvelope
// @Failure 503 {object} VolcesErrorEnvelope
// @Router /api/v1/contents/generations/tasks [post]
func (s *Server) createVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeVolcesError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
		return
	}
	body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations")
	if err != nil {
		writeVolcesError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err))
		return
	}
	task, err := s.createVolcesCompatibleTask(r, user, body)
	if err != nil {
		writeVolcesCompatibleTaskError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, volcesCompatibleCreateResponse(task))
}

// getVolcesContentsGenerationTask godoc
// @Summary 查询火山内容生成任务
// @Tags volces-compatible
// @Produce json
// @Security BearerAuth
// @Param taskID path string true "任务 ID"
// @Success 200 {object} VolcesContentsGenerationTaskResponse
// @Failure 404 {object} VolcesErrorEnvelope
// @Failure 410 {object} VolcesErrorEnvelope
// @Router /api/v1/contents/generations/tasks/{taskID} [get]
func (s *Server) getVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) {
	task, ok := s.volcesCompatibleTaskForUser(w, r)
	if !ok {
		return
	}
	var err error
	task, err = s.hydrateTaskResult(r.Context(), task)
	if err != nil {
		writeVolcesError(w, statusFromRunError(err), err.Error(), clients.ErrorCode(err))
		return
	}
	writeJSON(w, http.StatusOK, volcesCompatibleTask(task))
}

// listVolcesContentsGenerationTasks godoc
// @Summary 列出火山内容生成任务
// @Tags volces-compatible
// @Produce json
// @Security BearerAuth
// @Success 200 {object} map[string]any
// @Failure 400 {object} VolcesErrorEnvelope
// @Failure 401 {object} VolcesErrorEnvelope
// @Router /api/v1/contents/generations/tasks [get]
func (s *Server) listVolcesContentsGenerationTasks(w http.ResponseWriter, r *http.Request) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeVolcesError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
		return
	}
	page := portraitAssetQueryInt(r, "page_num", "pageNumber", "page")
	pageSize := portraitAssetQueryInt(r, "page_size", "pageSize")
	tasks, err := s.store.ListVolcesCompatibleTasks(r.Context(), user, store.VolcesCompatibleTaskListFilter{
		CompatibilityMarker: volcesContentsCompatibilityMarker,
		Status:              r.URL.Query().Get("filter.status"),
		Model:               r.URL.Query().Get("filter.model"),
		TaskIDs:             r.URL.Query()["filter.task_ids"],
		Page:                page,
		PageSize:            pageSize,
	})
	if err != nil {
		s.logger.Error("list Volces-compatible tasks failed", "error", err)
		writeVolcesError(w, http.StatusInternalServerError, "list tasks failed", "internal_error")
		return
	}
	items := make([]any, 0)
	for _, task := range tasks.Items {
		items = append(items, volcesCompatibleTask(task))
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"items": items, "total": tasks.Total,
		"page_num": tasks.Page, "page_size": tasks.PageSize,
	})
}

// deleteVolcesContentsGenerationTask godoc
// @Summary 取消火山内容生成任务
// @Description 取消网关任务；对于已提交且保存了上游任务标识的 Volces 视频任务，同时调用火山 DELETE 接口并持久化取消状态。
// @Tags volces-compatible
// @Produce json
// @Security BearerAuth
// @Param taskID path string true "任务 ID"
// @Success 200 {object} VolcesContentsGenerationTaskResponse
// @Failure 404 {object} VolcesErrorEnvelope
// @Router /api/v1/contents/generations/tasks/{taskID} [delete]
func (s *Server) deleteVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) {
	task, ok := s.volcesCompatibleTaskForUser(w, r)
	if !ok {
		return
	}
	user, _ := auth.UserFromContext(r.Context())
	result, err := s.runner.CancelVolcesVideoTask(r.Context(), task, user)
	if err != nil {
		if errors.Is(err, runner.ErrTaskAccessDenied) {
			writeVolcesError(w, http.StatusNotFound, "task not found", "not_found")
			return
		}
		s.logger.Error("cancel Volces-compatible task failed", "error", err)
		writeVolcesError(w, http.StatusInternalServerError, "cancel task failed", "internal_error")
		return
	}
	updated, err := s.store.GetTask(r.Context(), task.ID)
	if err != nil {
		writeVolcesError(w, http.StatusInternalServerError, "get cancelled task failed", "internal_error")
		return
	}
	_ = result
	writeJSON(w, http.StatusOK, volcesCompatibleTask(updated))
}

// createLegacyVolcesVideoGeneration godoc
// @Summary 创建 server-main 兼容视频任务
// @Description 兼容 server-main 的 /api/v1/video/generations，返回 submitted 和 task_id；额外保留火山任务字段。
// @Tags media
// @Accept json
// @Produce json
// @Security BearerAuth
// @Success 200 {object} TaskAcceptedResponse
// @Router /api/v1/video/generations [post]
func (s *Server) createLegacyVolcesVideoGeneration(w http.ResponseWriter, r *http.Request) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeError(w, http.StatusUnauthorized, "unauthorized")
		return
	}
	body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations")
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err))
		return
	}
	task, err := s.createVolcesCompatibleTask(r, user, body)
	if err != nil {
		writeVolcesCompatibleTaskError(w, err)
		return
	}
	response := volcesCompatibleTask(task)
	writeJSON(w, http.StatusOK, easyAITaskAcceptedResponseWithExtensions(task, response))
}

// getEasyAITaskResult godoc
// @Summary 查询 server-main EasyAIClient 兼容任务结果
// @Description 查询当前用户的图片、视频、音频、视频高清和矢量化任务，并返回 EasyAI GeneratedResponse 兼容结构。
// @Tags tasks
// @Produce json
// @Security BearerAuth
// @Param taskID path string true "任务 ID"
// @Success 200 {object} EasyAIGeneratedResponse
// @Header 200 {string} X-Gateway-Response-Format "实际结果格式，异步轮询固定为 url"
// @Header 200 {string} Deprecation "output_content 与 result 兼容别名的废弃标记"
// @Failure 404 {object} EasyAIGeneratedResponse
// @Failure 410 {object} EasyAIGeneratedResponse
// @Failure 503 {object} ErrorEnvelope
// @Router /api/v1/ai/result/{taskID} [get]
func (s *Server) getEasyAITaskResult(w http.ResponseWriter, r *http.Request) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeEasyAIAsyncError(w, http.StatusUnauthorized, "unauthorized", nil, "unauthorized")
		return
	}
	task, err := s.store.GetTask(r.Context(), strings.TrimSpace(r.PathValue("taskID")))
	if err != nil {
		if store.IsNotFound(err) {
			writeEasyAIAsyncError(w, http.StatusNotFound, "task not found", nil, "not_found")
			return
		}
		s.logger.Error("get EasyAI-compatible task failed", "error", err)
		writeError(w, http.StatusInternalServerError, "get task failed", "internal_error")
		return
	}
	if !runner.TaskAccessibleToUser(task, user) {
		writeEasyAIAsyncError(w, http.StatusNotFound, "task not found", nil, "not_found")
		return
	}
	task, err = s.hydrateTaskResult(r.Context(), task)
	if err != nil {
		writeEasyAIAsyncError(w, statusFromRunError(err), err.Error(), nil, clients.ErrorCode(err))
		return
	}
	writeAsyncMediaResultJSON(w, easyAITaskResultResponse(task), s.billingMetrics)
}

func (s *Server) createVolcesCompatibleTask(r *http.Request, user *auth.User, body map[string]any) (store.GatewayTask, error) {
	model := strings.TrimSpace(volcesCompatString(body["model"]))
	if model == "" {
		return store.GatewayTask{}, &clients.ClientError{Code: "invalid_parameter", Message: "model is required", StatusCode: http.StatusBadRequest, Retryable: false}
	}
	if !apiKeyScopeAllowed(user, "videos.generations") {
		return store.GatewayTask{}, &clients.ClientError{Code: "forbidden", Message: "api key scope does not allow video generation", StatusCode: http.StatusForbidden, Retryable: false}
	}
	body["_gateway_compatibility"] = volcesContentsCompatibilityMarker
	body["_gateway_target_protocol"] = clients.ProtocolVolcesContents
	task, err := s.prepareAndCreateGatewayTask(r.Context(), r, user, "videos.generations", model, body, true)
	if err != nil {
		return store.GatewayTask{}, err
	}
	if err := s.store.SetTaskCompatibilitySubmission(r.Context(), task.ID, store.CompatibilitySubmission{
		TargetProtocol: clients.ProtocolVolcesContents,
		PublicID:       task.ID,
	}); err != nil {
		return store.GatewayTask{}, err
	}
	task.CompatibilityProtocol = clients.ProtocolVolcesContents
	if err := s.runner.SubmitAsyncTask(r.Context(), task); err != nil {
		return store.GatewayTask{}, err
	}
	return s.waitForCompatibilitySubmission(r, task)
}

func (s *Server) waitForCompatibilitySubmission(r *http.Request, task store.GatewayTask) (store.GatewayTask, error) {
	deadline := time.NewTimer(30 * time.Second)
	defer deadline.Stop()
	ticker := time.NewTicker(25 * time.Millisecond)
	defer ticker.Stop()
	for {
		current, err := s.store.GetTask(r.Context(), task.ID)
		if err != nil {
			return store.GatewayTask{}, err
		}
		if current.RemoteTaskID != "" || current.Status == "succeeded" {
			return current, nil
		}
		if current.Status == "failed" || current.Status == "cancelled" {
			status := http.StatusBadGateway
			switch current.ErrorCode {
			case "bad_request", "invalid_parameter", "unsupported_kind", "unsupported_model", "missing_credentials":
				status = http.StatusBadRequest
			case "permission_denied", "forbidden":
				status = http.StatusForbidden
			case "rate_limit":
				status = http.StatusTooManyRequests
			}
			return store.GatewayTask{}, &clients.ClientError{
				Code:       firstNonEmpty(current.ErrorCode, current.Status),
				Message:    firstNonEmpty(current.ErrorMessage, current.Error, current.Message),
				StatusCode: status,
				Retryable:  false,
			}
		}
		select {
		case <-r.Context().Done():
			return store.GatewayTask{}, r.Context().Err()
		case <-deadline.C:
			return store.GatewayTask{}, &clients.ClientError{Code: "submission_timeout", Message: "upstream did not confirm task submission before timeout", StatusCode: http.StatusGatewayTimeout, Retryable: true}
		case <-ticker.C:
		}
	}
}

func (s *Server) volcesCompatibleTaskForUser(w http.ResponseWriter, r *http.Request) (store.GatewayTask, bool) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok || user == nil {
		writeVolcesError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
		return store.GatewayTask{}, false
	}
	task, err := s.store.GetVolcesCompatibleTask(r.Context(), user, volcesContentsCompatibilityMarker, strings.TrimSpace(r.PathValue("taskID")))
	if err != nil {
		if store.IsNotFound(err) {
			writeVolcesError(w, http.StatusNotFound, "task not found", "not_found")
			return store.GatewayTask{}, false
		}
		s.logger.Error("get Volces-compatible task failed", "error", err)
		writeVolcesError(w, http.StatusInternalServerError, "get task failed", "internal_error")
		return store.GatewayTask{}, false
	}
	if !isVolcesCompatibleTask(task) || !kelingCompatTaskOwnedBy(task, user) {
		writeVolcesError(w, http.StatusNotFound, "task not found", "not_found")
		return store.GatewayTask{}, false
	}
	return task, true
}

func isVolcesCompatibleTask(task store.GatewayTask) bool {
	return task.Kind == "videos.generations" && strings.TrimSpace(volcesCompatString(task.Request["_gateway_compatibility"])) == volcesContentsCompatibilityMarker
}

func volcesCompatibleTask(task store.GatewayTask) map[string]any {
	response := map[string]any{}
	for _, key := range []string{"model", "seed", "resolution", "ratio", "duration", "frames", "framespersecond"} {
		if task.Result[key] != nil {
			response[key] = task.Result[key]
		}
	}
	if content, ok := task.Result["content"].(map[string]any); ok && len(content) > 0 {
		// Transitional read support for historical canonical results. New
		// results only persist data[] and reconstruct this protocol field.
		response["content"] = content
	} else if data, ok := task.Result["data"].([]any); ok && len(data) > 0 {
		if item, ok := data[0].(map[string]any); ok {
			content := map[string]any{}
			if videoURL := firstNonEmpty(volcesCompatString(item["url"]), volcesCompatString(item["video_url"])); videoURL != "" {
				content["video_url"] = videoURL
			}
			if lastFrameURL := volcesCompatString(item["last_frame_url"]); lastFrameURL != "" {
				content["last_frame_url"] = lastFrameURL
			}
			if len(content) > 0 {
				response["content"] = content
			}
		}
	}
	response["id"] = volcesCompatiblePublicID(task)
	response["model"] = firstNonEmpty(volcesCompatString(response["model"]), task.Model)
	response["status"] = volcesCompatibleTaskStatus(task.Status)
	response["created_at"] = task.CreatedAt.Unix()
	response["updated_at"] = task.UpdatedAt.Unix()
	for _, key := range []string{"content", "seed", "resolution", "ratio", "duration", "frames", "framespersecond"} {
		if response[key] == nil && task.Request[key] != nil {
			response[key] = task.Request[key]
		}
	}
	if usage := volcesCompatibleUsage(task.Usage); len(usage) > 0 {
		response["usage"] = usage
	} else if legacyUsage, ok := task.Result["usage"].(map[string]any); ok && len(legacyUsage) > 0 {
		response["usage"] = legacyUsage
	}
	if task.Status == "failed" || task.Status == "cancelled" {
		response["error"] = publicErrorMap(publicTaskError(task))
	}
	return response
}

func volcesCompatibleUsage(usage map[string]any) map[string]any {
	out := map[string]any{}
	for outputKey, inputKeys := range map[string][]string{
		"prompt_tokens":     {"inputTokens", "promptTokens"},
		"completion_tokens": {"outputTokens", "completionTokens"},
		"total_tokens":      {"totalTokens"},
	} {
		for _, inputKey := range inputKeys {
			if value := usage[inputKey]; value != nil {
				out[outputKey] = value
				break
			}
		}
	}
	return out
}

func volcesCompatibleCreateResponse(task store.GatewayTask) map[string]any {
	return map[string]any{"id": volcesCompatiblePublicID(task)}
}

func volcesCompatiblePublicID(task store.GatewayTask) string {
	if volcesTaskUsesNativeProtocol(task) && strings.TrimSpace(task.RemoteTaskID) != "" {
		return strings.TrimSpace(task.RemoteTaskID)
	}
	return task.ID
}

func volcesTaskUsesNativeProtocol(task store.GatewayTask) bool {
	if strings.TrimSpace(task.CompatibilitySourceProtocol) != "" {
		return task.CompatibilitySourceProtocol == clients.ProtocolVolcesContents && task.CompatibilityProtocol == clients.ProtocolVolcesContents
	}
	for index := len(task.Attempts) - 1; index >= 0; index-- {
		if provider := strings.ToLower(strings.TrimSpace(task.Attempts[index].Provider)); provider != "" {
			return provider == "volces"
		}
	}
	return false
}

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

func writeVolcesCompatibleTaskError(w http.ResponseWriter, err error) {
	status := http.StatusInternalServerError
	var staged *gatewayTaskCreationError
	if errors.As(err, &staged) {
		err = staged.Err
	}
	var trafficErr *taskTrafficError
	if errors.As(err, &trafficErr) {
		writeVolcesError(w, trafficErr.Status, trafficErr.Message, trafficErr.Code)
		return
	}
	var clientErr *clients.ClientError
	if errors.As(err, &clientErr) && clientErr.StatusCode > 0 {
		status = clientErr.StatusCode
	} else if errors.As(err, &clientErr) {
		status = http.StatusBadRequest
	} else if errors.Is(err, store.ErrTaskRequestBinaryNotMaterialized) {
		status = http.StatusBadRequest
	}
	writeVolcesError(w, status, err.Error(), clients.ErrorCode(err))
}

func volcesCompatString(value any) string {
	switch typed := value.(type) {
	case string:
		return strings.TrimSpace(typed)
	case json.Number:
		return typed.String()
	case float64:
		return strconv.FormatFloat(typed, 'f', -1, 64)
	default:
		return ""
	}
}
