package service

import (
	"context"
	"encoding/json"
	"fmt"
	"model-gateway/dao"
	"model-gateway/model/domain"
	"model-gateway/model/dto"
	"model-gateway/service/httpclient"
	modelUtils "model-gateway/service/utils"
	"net/http"
	"strings"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
)

// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）。
// 与同步请求一致：上游返回错误时按 shouldRetryWithMemory 判定是否指数退避重试（最多 modelCallMaxRetries 次）。
func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *dto.CallModelSessionReq) (docMsg *dto.ModelCallRes, err error) {
	startTime := time.Now()

	id := req.Id
	modelInfo := req.ModelInfo
	newRequestParams := req.RequestParams

	attempt := 0
LOOP:
	// 获取上游流式 reader（stream=false → w 不会被使用，传 nil）。
	// 非 2xx 状态/网络错误在此返回；按 shouldRetryWithMemory 判定是否指数退避重试，与同步请求一致。
	streamReader, err := httpclient.ModelHttpStreamRequest(ctx, nil, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
	if err != nil {
		if attempt < modelCallMaxRetries {
			if code, msg := streamErrorInfoOfError(err); shouldRetryWithMemory(ctx, modelInfo, code, msg, "") {
				attempt++
				wait := time.Duration(1<<attempt) * time.Second
				g.Log().Warningf(ctx, "模型流式请求异常，第 %d 次重试(等待 %v): code=%s err=%v", attempt+1, wait, code, err)
				if waitErr := retryWait(ctx, attempt); waitErr != nil {
					recordSessionError(context.WithoutCancel(ctx), id, startTime, "调用取消: "+waitErr.Error())
					return nil, waitErr
				}
				goto LOOP
			}
		}
		// 非重试错误/重试耗尽：请求失败即返回，把失败信息写入模型会话记录
		recordSessionError(ctx, id, startTime, err.Error())
		return nil, err
	}

	docMsg = new(dto.ModelCallRes)
	docMsg.TaskId = id
	var contentBuf strings.Builder

	// 记录流内 error 事件（OpenAI 兼容 error 分片），供流结束后统一判定重试/报错
	var streamErrCode, streamErrMsg, streamErrBody string

	// 路径预处理
	respMapping := make(map[string]string, len(modelInfo.ResponseBodyMapping))
	for k, _ := range modelInfo.ResponseBodyMapping {
		respMapping[k] = modelUtils.CleanFieldPath(k)
	}
	totalTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)
	promptTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)
	completionTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)

	httpclient.ParseSSEStream(ctx, streamReader, func(ctx context.Context, chunk map[string]any) error {
		// 流内错误事件（OpenAI 兼容 error 分片）：暂存错误码/消息，不做内容累加，由流结束后统一判定
		if code, msg := streamErrorOfChunk(chunk); code != "" {
			streamErrCode, streamErrMsg, streamErrBody = code, msg, gconv.String(chunk["error"])
			return nil
		}
		// 基于统一字段路径（GetByPath）在分片对象上取值，取首个数组元素文本累加
		for _, jsonPath := range respMapping {
			if realText := extractChunkText(ctx, modelUtils.GetByPathValue(chunk, jsonPath)); realText != "" {
				contentBuf.WriteString(realText)
			}
		}

		// Token 累加
		docMsg.TotalTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, totalTokenPath))
		docMsg.PromptTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, promptTokenPath))
		docMsg.CompletionTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, completionTokenPath))
		return nil
	})

	// 流内返回错误：丢弃本次部分内容，指数退避后重新请求（判定交给 shouldRetryWithMemory）
	if streamErrCode != "" {
		if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, streamErrCode, streamErrMsg, streamErrBody) {
			attempt++
			wait := time.Duration(1<<attempt) * time.Second
			g.Log().Warningf(ctx, "模型流式调用异常，第 %d 次重试(等待 %v): code=%s msg=%s", attempt+1, wait, streamErrCode, streamErrMsg)
			if waitErr := retryWait(ctx, attempt); waitErr != nil {
				recordSessionError(context.WithoutCancel(ctx), id, startTime, "调用取消: "+waitErr.Error())
				return nil, waitErr
			}
			goto LOOP
		}
		// 与同步一致：不可重试的错误码记录到 ErrorMsg 后正常走组装返回，不中断流程
		docMsg.ErrorMsg = streamErrMsg
	}

	// 流结束后组装（流内出错时内容为空，与同步一致不再组装/上传空内容）
	if streamErrCode == "" {
		for k, _ := range modelInfo.ResponseBodyMapping {
			docMsg.Content = map[string]any{k: contentBuf.String()}
		}
	}

	// 补充更新会话记录
	updateModelSessionReq := dto.UpdateModelSessionReq{
		Id:               id,
		DurationSeconds:  int64(time.Since(startTime).Seconds()),
		TotalTokens:      docMsg.TotalTokens,
		PromptTokens:     docMsg.PromptTokens,
		CompletionTokens: docMsg.CompletionTokens,
		ErrorMsg:         docMsg.ErrorMsg,
	}
	if !g.IsEmpty(docMsg.Content) {
		uploadNewResp, uploadErr := Upload(ctx, &dto.UploadFileBytesReq{
			FileBytes: gconv.Bytes(gconv.String(docMsg.Content)),
			FileName:  fmt.Sprintf("modelNewRespParams:%v.json", time.Now().UnixMilli()),
		})
		if uploadErr != nil {
			return nil, fmt.Errorf("上传模型返回参数文件失败：%v", uploadErr)
		}
		updateModelSessionReq.ResponsePath = uploadNewResp.FileURL
	}
	updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds())
	// 调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞）
	mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams)
	docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价
	docMsg.MediaType = mediaType
	docMsg.Cost = calcModelCost(ctx, modelInfo.Id,
		buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0))
	updateModelSessionReq.TotalCost = docMsg.Cost
	if _, updateErr := dao.ModelSession.Update(ctx, &updateModelSessionReq); updateErr != nil {
		g.Log().Errorf(ctx, "更新流式会话信息失败: %v", updateErr)
	}

	return docMsg, nil
}

// CreateSessionStream 流式调用上游模型 → SSE 逐分片推送给前端；流结束返回本次调用的 token/费用（docMsg.Cost）供调用方扣减。
func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.ResponseWriter, req *dto.CallModelSessionReq) (*dto.ModelCallRes, error) {
	startTime := time.Now()

	id := req.Id
	modelInfo := req.ModelInfo
	newRequestParams := req.RequestParams

	// 获取上游流式 reader 并设置 SSE 响应头
	streamReader, err := httpclient.ModelHttpStreamRequest(ctx, w, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
	if err != nil {
		// 请求建立前失败：把错误写入模型会话记录，避免留半截无错误信息记录
		recordSessionError(ctx, id, startTime, err.Error())
		return nil, err
	}

	flusher := w.(http.Flusher)

	docMsg := new(dto.ModelCallRes)
	var contentBuf strings.Builder

	// tool_calls 按 index 累加（OpenAI 兼容 delta），流末随 done 事件一次性返回
	toolAcc := make(map[int]*streamToolCallAcc)

	// 路径预处理
	respMapping := make(map[string]string, len(modelInfo.ResponseBodyMapping))
	for k, _ := range modelInfo.ResponseBodyMapping {
		respMapping[modelUtils.CleanFieldPath(k)] = modelUtils.CleanFieldPath(k)
	}
	totalTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)
	promptTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)
	completionTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)

	// 解析 ResponseBusinessFieldMapping 字段
	businessFieldRes := new(domain.ChatFieldsRes)
	err = gconv.Struct(modelInfo.ResponseBusinessFieldMapping, businessFieldRes)
	if err != nil {
		g.Log().Errorf(ctx, "解析 ResponseBusinessFieldMapping 失败: %v", err)
	}
	// tool_calls 读取路径：优先走 ResponseBusinessFieldMapping 的 tools 配置，缺省兜底 OpenAI 兼容路径
	//toolsPath := "choices[0].delta.tool_calls"
	toolsPath := businessFieldRes.Tools
	// reasoning_content 读取路径：走 ResponseBusinessFieldMapping 的 reasoning_content 配置，未配置则不返回思考内容
	reasoningPath := modelUtils.CleanFieldPath(businessFieldRes.ReasoningContent)

	httpclient.ParseSSEStream(ctx, streamReader, func(ctx context.Context, chunk map[string]any) error {
		// 基于统一字段路径（GetByPath）在分片对象上取值，取首个数组元素文本
		content := make(map[string]any, len(respMapping))
		for bizKey, jsonPath := range respMapping {
			if realText := extractChunkText(ctx, modelUtils.GetByPathValue(chunk, jsonPath)); realText != "" {
				content[bizKey] = realText
				contentBuf.WriteString(realText)
			}
		}

		// Token 累加（记录增量：usage 常在无文本/思考的末分片出现，需据此放行推送）
		prevTotal, prevPrompt, prevCompletion := docMsg.TotalTokens, docMsg.PromptTokens, docMsg.CompletionTokens
		docMsg.TotalTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, totalTokenPath))
		docMsg.PromptTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, promptTokenPath))
		docMsg.CompletionTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, completionTokenPath))
		tokenDelta := docMsg.TotalTokens != prevTotal || docMsg.PromptTokens != prevPrompt || docMsg.CompletionTokens != prevCompletion

		accumulateStreamToolCallsByPath(chunk, toolsPath, toolAcc)

		// 思考内容提取（独立业务字段，不进回答全文）
		var reasoningContent string
		if reasoningPath != "" {
			if v := modelUtils.GetByPathValue(chunk, reasoningPath); v != nil && !g.IsEmpty(v) {
				reasoningContent = firstText(v)
			}
		}

		// 纯 token 分片（无文本/思考）也放行：否则末分片 usage 被过滤，调用方拿不到 token 值
		if len(content) == 0 && reasoningContent == "" && !tokenDelta {
			return nil
		}

		// 逐 chunk SSE 推送给前端（字段名由 ModelCallStreamEvent 统一管理）
		event := &dto.ModelCallStreamEvent{
			Content:          content,
			ReasoningContent: reasoningContent,
			TotalTokens:      docMsg.TotalTokens,
			PromptTokens:     docMsg.PromptTokens,
			CompletionTokens: docMsg.CompletionTokens,
		}
		outBytes, err := json.Marshal(event)
		if err != nil {
			g.Log().Errorf(ctx, "[SSE] marshal response failed: %v", err)
			return nil
		}
		_, err = fmt.Fprintf(w, "data: %s\n\n", outBytes)
		if err != nil {
			g.Log().Errorf(ctx, "[SSE] write client failed: %v", err)
			return err
		}
		flusher.Flush()
		return nil
	})

	// 流结束：调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞）
	mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams)
	docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价
	docMsg.MediaType = mediaType
	docMsg.Cost = calcModelCost(ctx, modelInfo.Id,
		buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0))

	// 流末 done 事件：携带该步最终 token 与费用。工具调用时附带完整 tool_calls，
	// 纯文本流同样补发，使调用方拿到最终费用与 token；不识别 type=done 的消费方忽略该事件。
	event := &dto.ModelCallStreamEvent{
		Type:             "done",
		TotalTokens:      docMsg.TotalTokens,
		PromptTokens:     docMsg.PromptTokens,
		CompletionTokens: docMsg.CompletionTokens,
		Cost:             docMsg.Cost,
	}
	if tools := finalizeStreamToolCalls(toolAcc); len(tools) > 0 {
		var toolModels []dto.ModelTool
		if err := gconv.Structs(tools, &toolModels); err != nil {
			g.Log().Errorf(ctx, "[SSE] convert tools failed: %v", err)
		} else {
			event.Tools = toolModels
		}
	}
	outBytes, err := json.Marshal(event)
	if err == nil {
		_, _ = fmt.Fprintf(w, "data: %s\n\n", outBytes)
		flusher.Flush()
	}

	// 流结束后补充更新会话记录
	updateModelSessionReq := dto.UpdateModelSessionReq{
		Id:               id,
		DurationSeconds:  int64(time.Since(startTime).Seconds()),
		TotalTokens:      docMsg.TotalTokens,
		PromptTokens:     docMsg.PromptTokens,
		CompletionTokens: docMsg.CompletionTokens,
		TotalCost:        docMsg.Cost,
	}
	if !g.IsEmpty(contentBuf.String()) {
		uploadNewResp, uploadErr := Upload(ctx, &dto.UploadFileBytesReq{
			FileBytes: gconv.Bytes(gconv.String(map[string]any{"respBody": contentBuf.String()})),
			FileName:  fmt.Sprintf("modelNewRespParams:%v.json", time.Now().UnixMilli()),
		})
		if uploadErr != nil {
			return nil, fmt.Errorf("上传模型返回参数文件失败：%v", uploadErr)
		}
		updateModelSessionReq.ResponsePath = uploadNewResp.FileURL
	}
	updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds())
	if _, updateErr := dao.ModelSession.Update(ctx, &updateModelSessionReq); updateErr != nil {
		return nil, fmt.Errorf("更新会话信息失败: %v", updateErr)
	}

	return docMsg, nil
}

// recordSessionError 请求建立前失败（上游不可达/非 2xx 且非重试/重试耗尽/调用取消）时，
// 把错误与耗时写入模型会话记录，避免流式调用留半截无错误信息记录。
// 仅写 ErrorMsg/DurationSeconds（OmitEmpty 不会影响已落库字段）；ctx 已取消时须传 WithoutCancel(ctx)。
func recordSessionError(ctx context.Context, id int64, startTime time.Time, errMsg string) {
	if _, updateErr := dao.ModelSession.Update(ctx, &dto.UpdateModelSessionReq{
		Id:              id,
		DurationSeconds: int64(time.Since(startTime).Seconds()),
		ErrorMsg:        errMsg,
	}); updateErr != nil {
		g.Log().Errorf(ctx, "更新模型会话错误信息失败: %v", updateErr)
	}
}

// streamErrorOfChunk 从流式分片提取错误码与消息：优先 OpenAI 兼容 error 事件，顶层 code 兜底。
func streamErrorOfChunk(chunk map[string]any) (code, msg string) {
	if errObj := gconv.Map(chunk["error"]); errObj != nil {
		code = gconv.String(errObj["code"])
		msg = gconv.String(errObj["message"])
	}
	if code == "" {
		code = gconv.String(chunk["code"])
	}
	return
}

// streamErrorInfoOfError 从流式请求错误中提取错误码与消息（不做固定清单过滤，交由 shouldRetryWithMemory 判定）：
// 优先解析错误体 error.code/顶层 code 与 message，其次取非 2xx 的 HTTP 状态码。
// 纯网络错误等无错误码场景返回空串。
func streamErrorInfoOfError(err error) (code, msg string) {
	if err == nil {
		return "", ""
	}
	e := err.Error()
	// 非 2xx 时 httpclient.ModelHttpStreamRequest 返回 "[HTTP][Stream] 状态码异常: %d, body={...}"
	if idx := strings.Index(e, "body="); idx >= 0 {
		body := e[idx+len("body="):]
		var errResp struct {
			Error struct {
				Code    string `json:"code"`
				Message string `json:"message"`
			} `json:"error"`
			Code    string `json:"code"`
			Message string `json:"message"`
		}
		if json.Unmarshal([]byte(body), &errResp) == nil {
			if errResp.Error.Code != "" {
				return errResp.Error.Code, errResp.Error.Message
			}
			if errResp.Code != "" {
				return errResp.Code, errResp.Message
			}
		}
	}
	if idx := strings.Index(e, "状态码异常: "); idx >= 0 {
		codeStr := strings.TrimSpace(e[idx+len("状态码异常: "):])
		if comma := strings.IndexByte(codeStr, ','); comma >= 0 {
			codeStr = codeStr[:comma]
		}
		return codeStr, ""
	}
	return "", ""
}