team-api/relay/handler/responses_lifecycle.go

package handler

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"time"

	"github.com/gogf/gf/v2/frame/g"

	"github.com/qianfree/team-api/relay/channel"
	"github.com/qianfree/team-api/relay/common"
	"github.com/qianfree/team-api/relay/constant"
	"github.com/qianfree/team-api/relay/override"
)

// Responses 生命周期动作（POST {id}/cancel 以 method 区分，其余按 HTTP method 透传）
const (
	responsesLifecycleCancel = "/cancel"
)

// HandleResponsesLifecycle 处理 Responses 生命周期端点：
// GET /v1/responses/{id}（retrieve）、POST /v1/responses/{id}/cancel、DELETE /v1/responses/{id}。
// 经 Redis 路由表（response_id → 渠道，Record 于直连转发时写入）还原原始渠道后透传上游。
// 无请求体、无计费（对齐异步任务查询端点先例）；错误以 *constant.RelayError 返回，
// 由调用方经 WriteRelayError 写为 OpenAI 格式。
func HandleResponsesLifecycle(ctx context.Context, method string, responseID string, rc *RelayContext, provider common.DataProvider) error {
	if responseID == "" {
		return constant.NewRequestError("response id is required", nil)
	}

	// 1. 查路由：miss（未记录/已过期/Redis 不可用）或渠道已不可物化均按 404 处理
	route, ok := common.DefaultResponseRouteStore.Lookup(ctx, rc.TenantID, responseID)
	if !ok {
		return &constant.RelayError{StatusCode: http.StatusNotFound, Message: "Response not found", Type: "invalid_request_error"}
	}

	selection, err := provider.MaterializeSelection(ctx, route.ChannelID, 0, route.ModelName)
	if err != nil {
		g.Log().Debugf(ctx, "[ResponsesLifecycle] materialize channel failed: responseID=%s channel=%d err=%v", responseID, route.ChannelID, err)
		return &constant.RelayError{StatusCode: http.StatusNotFound, Message: "Response not found", Type: "invalid_request_error"}
	}

	adaptor := channel.GetAdaptor(selection.ChannelType)
	if adaptor == nil {
		return constant.NewUpstreamError(http.StatusInternalServerError, fmt.Sprintf("unsupported channel type: %d", selection.ChannelType), nil)
	}

	info := buildLifecycleRelayInfo(rc, selection)
	info.RequestURLPath = "/responses/" + responseID
	// 守卫：路由记录只产生于 UpstreamSpeaksResponses 渠道，此处防御 chat-only 渠道
	//（GetRequestURL 会返回 /v1/chat/completions 导致拼错 URL）
	if !info.ChannelMeta.UpstreamSpeaksResponses() {
		return &constant.RelayError{StatusCode: http.StatusNotFound, Message: "Response not found", Type: "invalid_request_error"}
	}
	adaptor.Init(info)

	upstreamURL, err := buildLifecycleUpstreamURL(adaptor, info, responseID, method == http.MethodPost)
	if err != nil {
		return constant.NewUpstreamError(http.StatusInternalServerError, err.Error(), nil)
	}

	httpReq, err := http.NewRequestWithContext(ctx, method, upstreamURL, nil)
	if err != nil {
		return constant.NewUpstreamError(http.StatusInternalServerError, "create lifecycle request failed", err)
	}
	if err := adaptor.SetupRequestHeader(httpReq.Header, info); err != nil {
		return constant.NewUpstreamError(http.StatusInternalServerError, "setup request header failed", err)
	}
	// 与标准转发链路一致：应用渠道级 Header Override
	if hdrOverrides, hdrErr := override.ApplyHeaderOverride(info); hdrErr == nil && len(hdrOverrides) > 0 {
		override.MergeHeaderOverrides(httpReq.Header, hdrOverrides)
	}

	client := common.NewPooledClient(info.ChannelMeta.Settings.GetTimeoutSeconds(info.RelayMode), info.ChannelMeta.Settings.UseProxy)
	resp, err := client.Do(httpReq)
	if err != nil {
		return constant.NewUpstreamError(http.StatusBadGateway, "请求处理失败", err)
	}
	defer resp.Body.Close()

	// 原样透传上游响应（状态码 + Content-Type + body）
	if ct := resp.Header.Get("Content-Type"); ct != "" {
		rc.Writer.Header().Set("Content-Type", ct)
	}
	rc.Writer.WriteHeader(resp.StatusCode)
	_, _ = io.Copy(rc.Writer, resp.Body)

	// DELETE 成功后清理路由记录（上游已不可 retrieve）
	if method == http.MethodDelete && resp.StatusCode >= 200 && resp.StatusCode < 300 {
		common.DefaultResponseRouteStore.Delete(ctx, rc.TenantID, responseID)
	}
	return nil
}

// buildLifecycleRelayInfo 构建生命周期转发的最小 RelayInfo（不经调度 FSM，
// 字段拷贝自路由记录物化的 ChannelSelection，照 buildRelayInfo 的 ChannelMeta 部分；
// RequestURLPath 由调用方按目标 responseID 覆写）
func buildLifecycleRelayInfo(rc *RelayContext, selection *common.ChannelSelection) *common.RelayInfo {
	return &common.RelayInfo{
		Context:         context.Background(),
		TenantID:        rc.TenantID,
		UserID:          rc.UserID,
		ApiKeyID:        rc.ApiKeyID,
		ProjectID:       rc.ProjectID,
		RequestID:       rc.RequestID,
		RelayMode:       int(constant.RelayModeResponses),
		OriginModelName: selection.UpstreamModelName,
		StartTime:       time.Now(),
		ChannelMeta: &common.ChannelMeta{
			ChannelID:         selection.ChannelID,
			ChannelType:       selection.ChannelType,
			ChannelName:       selection.ChannelName,
			BaseURL:           selection.BaseURL,
			ApiKey:            selection.ApiKey,
			UpstreamModelName: selection.UpstreamModelName,
			IsModelMapped:     selection.IsModelMapped,
			Settings:          selection.Settings,

			SupportsResponses: selection.SupportsResponses,
			ChatViaResponses:  selection.ChatViaResponses,
		},
	}
}

// buildLifecycleUpstreamURL 构建生命周期端点的上游 URL：
// 复用 adaptor.GetRequestURL（RelayModeResponses + UpstreamSpeaksResponses 时返回
// POST 基址，如 openai 的 /v1/responses、codex 的 /backend-api/codex/responses），
// 追加 /{responseID}（cancel 再追加 /cancel），对各 Responses 上游天然形状正确。
func buildLifecycleUpstreamURL(adaptor common.Adaptor, info *common.RelayInfo, responseID string, isCancel bool) (string, error) {
	base, err := adaptor.GetRequestURL(info)
	if err != nil {
		return "", err
	}
	url := base + "/" + responseID
	if isCancel {
		url += responsesLifecycleCancel
	}
	return url, nil
}