feat(api): 统一官方兼容接口响应协议

兼容接口现在以入口协议作为最终响应协议，同协议保留官方 Wire 响应，跨协议统一转换成功、任务状态与错误结构。同时修正异步提交状态边界，持久化兼容公开任务标识和官方提交响应，并新增迁移、流式响应及协议契约测试。验证：go vet ./...；go test ./...；govulncheck ./...；pnpm lint；pnpm test；pnpm build；pnpm audit --audit-level high；pnpm openapi；全部 CI 脚本。
2026-07-22 15:34:59 +08:00
parent 42e8b517fd
commit e07a997aa9
32 changed files with 2436 additions and 591 deletions
@@ -6,6 +6,7 @@ import (
 	"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"
@@ -22,17 +23,21 @@ const volcesContentsCompatibilityMarker = "volces_contents_generations_v3"
 // @Accept json
 // @Produce json
 // @Security BearerAuth
-// @Success 200 {object} map[string]any
+// @Success 200 {object} VolcesContentsGenerationTaskResponse
+// @Failure 400 {object} VolcesErrorEnvelope
+// @Failure 401 {object} VolcesErrorEnvelope
+// @Failure 429 {object} VolcesErrorEnvelope
+// @Failure 502 {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 {
-		writeError(w, http.StatusUnauthorized, "unauthorized")
+		writeVolcesError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
 		return
 	}
 	body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations")
 	if err != nil {
-		writeError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err))
+		writeVolcesError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err))
 		return
 	}
 	task, err := s.createVolcesCompatibleTask(r, user, body)
@@ -40,7 +45,7 @@ func (s *Server) createVolcesContentsGenerationTask(w http.ResponseWriter, r *ht
 		writeVolcesCompatibleTaskError(w, err)
 		return
 	}
-	writeJSON(w, http.StatusOK, volcesCompatibleTask(task))
+	writeJSON(w, http.StatusOK, volcesCompatibleCreateResponse(task))
 }

 // getVolcesContentsGenerationTask godoc
@@ -49,7 +54,8 @@ func (s *Server) createVolcesContentsGenerationTask(w http.ResponseWriter, r *ht
 // @Produce json
 // @Security BearerAuth
 // @Param taskID path string true "任务 ID"
-// @Success 200 {object} map[string]any
+// @Success 200 {object} VolcesContentsGenerationTaskResponse
+// @Failure 404 {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)
@@ -65,11 +71,13 @@ func (s *Server) getVolcesContentsGenerationTask(w http.ResponseWriter, r *http.
 // @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 {
-		writeError(w, http.StatusUnauthorized, "unauthorized")
+		writeVolcesError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
 		return
 	}
 	page := portraitAssetQueryInt(r, "page_num", "pageNumber", "page")
@@ -84,7 +92,7 @@ func (s *Server) listVolcesContentsGenerationTasks(w http.ResponseWriter, r *htt
 	})
 	if err != nil {
 		s.logger.Error("list Volces-compatible tasks failed", "error", err)
-		writeError(w, http.StatusInternalServerError, "list tasks failed")
+		writeVolcesError(w, http.StatusInternalServerError, "list tasks failed", "internal_error")
 		return
 	}
 	items := make([]any, 0)
@@ -94,8 +102,6 @@ func (s *Server) listVolcesContentsGenerationTasks(w http.ResponseWriter, r *htt
 	writeJSON(w, http.StatusOK, map[string]any{
 		"items": items, "total": tasks.Total,
 		"page_num": tasks.Page, "page_size": tasks.PageSize,
-		// data/page are retained as additive gateway fields for existing callers.
-		"data": items, "page": tasks.Page,
 	})
 }

@@ -106,7 +112,8 @@ func (s *Server) listVolcesContentsGenerationTasks(w http.ResponseWriter, r *htt
 // @Produce json
 // @Security BearerAuth
 // @Param taskID path string true "任务 ID"
-// @Success 200 {object} map[string]any
+// @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)
@@ -117,23 +124,20 @@ func (s *Server) deleteVolcesContentsGenerationTask(w http.ResponseWriter, r *ht
 	result, err := s.runner.CancelVolcesVideoTask(r.Context(), task, user)
 	if err != nil {
 		if errors.Is(err, runner.ErrTaskAccessDenied) {
-			writeError(w, http.StatusNotFound, "task not found")
+			writeVolcesError(w, http.StatusNotFound, "task not found", "not_found")
 			return
 		}
 		s.logger.Error("cancel Volces-compatible task failed", "error", err)
-		writeError(w, http.StatusInternalServerError, "cancel task failed")
+		writeVolcesError(w, http.StatusInternalServerError, "cancel task failed", "internal_error")
 		return
 	}
 	updated, err := s.store.GetTask(r.Context(), task.ID)
 	if err != nil {
-		writeError(w, http.StatusInternalServerError, "get cancelled task failed")
+		writeVolcesError(w, http.StatusInternalServerError, "get cancelled task failed", "internal_error")
 		return
 	}
-	response := volcesCompatibleTask(updated)
-	response["cancelled"] = result.Cancelled
-	response["cancellable"] = result.Cancellable
-	response["message"] = result.Message
-	writeJSON(w, http.StatusOK, response)
+	_ = result
+	writeJSON(w, http.StatusOK, volcesCompatibleTask(updated))
 }

 // createLegacyVolcesVideoGeneration godoc
@@ -202,34 +206,112 @@ func (s *Server) createVolcesCompatibleTask(r *http.Request, user *auth.User, bo
 		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
+	task.CompatibilityPublicID = task.ID
 	if err := s.runner.EnqueueAsyncTask(r.Context(), task); err != nil {
 		return store.GatewayTask{}, &clients.ClientError{Code: "enqueue_failed", Message: err.Error(), StatusCode: http.StatusServiceUnavailable, Retryable: true}
 	}
-	return task, nil
+	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,
+				Wire:       compatibilitySubmissionWire(current),
+			}
+		}
+		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 compatibilitySubmissionWire(task store.GatewayTask) *clients.WireResponse {
+	if task.CompatibilitySubmitHTTPStatus == 0 || strings.TrimSpace(task.CompatibilitySourceProtocol) == "" {
+		return nil
+	}
+	headers := map[string][]string{}
+	for name, value := range task.CompatibilitySubmitHeaders {
+		switch typed := value.(type) {
+		case []any:
+			for _, item := range typed {
+				if text := strings.TrimSpace(volcesCompatString(item)); text != "" {
+					headers[name] = append(headers[name], text)
+				}
+			}
+		case []string:
+			headers[name] = append([]string(nil), typed...)
+		case string:
+			headers[name] = []string{typed}
+		}
+	}
+	return &clients.WireResponse{
+		Protocol:   task.CompatibilitySourceProtocol,
+		StatusCode: task.CompatibilitySubmitHTTPStatus,
+		Headers:    headers,
+		Body:       cloneVolcesCompatibleMap(task.CompatibilitySubmitBody),
+		Converted:  task.CompatibilitySourceProtocol != task.CompatibilityProtocol,
+	}
 }

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

 func volcesCompatibleTask(task store.GatewayTask) map[string]any {
-	response := cloneVolcesCompatibleMap(task.Result)
-	if len(response) == 0 {
-		response = cloneVolcesCompatibleMap(task.RemoteTaskPayload)
+	if volcesTaskUsesNativeProtocol(task) {
+		if raw := cloneVolcesCompatibleMap(mapFromVolcesCompat(task.Result["raw"])); len(raw) > 0 {
+			return raw
+		}
+		if raw := cloneVolcesCompatibleMap(task.RemoteTaskPayload); len(raw) > 0 {
+			return raw
+		}
 	}
-	if response == nil {
-		response = map[string]any{}
-	}
-	response["id"] = task.ID
+	response := map[string]any{}
+	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()
-	if task.RemoteTaskID != "" {
-		response["upstream_task_id"] = task.RemoteTaskID
-	}
 	for _, key := range []string{"content", "seed", "resolution", "ratio", "duration", "frames", "framespersecond"} {
 		if response[key] == nil && task.Request[key] != nil {
 			response[key] = task.Request[key]
@@ -266,14 +347,45 @@ func volcesCompatibleTask(task store.GatewayTask) map[string]any {
 	if task.Status == "failed" || task.Status == "cancelled" {
 		response["error"] = map[string]any{"code": firstNonEmpty(task.ErrorCode, strings.ToUpper(task.Status)), "message": firstNonEmpty(task.ErrorMessage, task.Error, task.Message)}
 	}
-	response["gateway_task_id"] = task.ID
-	response["gateway_status"] = task.Status
-	response["billings"] = task.Billings
-	response["billing_summary"] = task.BillingSummary
-	response["final_charge_amount"] = task.FinalChargeAmount
 	return response
 }

+func volcesCompatibleCreateResponse(task store.GatewayTask) map[string]any {
+	if volcesTaskUsesNativeProtocol(task) {
+		if raw := cloneVolcesCompatibleMap(task.RemoteTaskPayload); len(raw) > 0 {
+			return raw
+		}
+	}
+	return map[string]any{"id": volcesCompatiblePublicID(task)}
+}
+
+func volcesCompatiblePublicID(task store.GatewayTask) string {
+	if strings.TrimSpace(task.CompatibilityPublicID) != "" {
+		return strings.TrimSpace(task.CompatibilityPublicID)
+	}
+	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 mapFromVolcesCompat(value any) map[string]any {
+	result, _ := value.(map[string]any)
+	return result
+}
+
 func volcesCompatibleTaskStatus(status string) string {
 	switch strings.ToLower(strings.TrimSpace(status)) {
 	case "succeeded", "success", "completed":
@@ -316,7 +428,11 @@ func writeVolcesCompatibleTaskError(w http.ResponseWriter, err error) {
 	} else if errors.As(err, &clientErr) {
 		status = http.StatusBadRequest
 	}
-	writeError(w, status, err.Error(), clients.ErrorCode(err))
+	if wire := clients.ErrorWireResponse(err); wireResponseMatches(wire, clients.ProtocolVolcesContents) {
+		writeWireResponse(w, wire)
+		return
+	}
+	writeVolcesError(w, status, err.Error(), clients.ErrorCode(err))
 }

 func volcesCompatString(value any) string {