package flow

import (
	"ai-agent/gateway"
	"ai-agent/workflow/consts/flow"
	"ai-agent/workflow/consts/node"
	"ai-agent/workflow/consts/public"
	nodeDao "ai-agent/workflow/dao/node"
	sessionDao "ai-agent/workflow/dao/session"
	flowDto "ai-agent/workflow/model/dto/flow"
	nodeDto "ai-agent/workflow/model/dto/node"
	sessionDto "ai-agent/workflow/model/dto/session"
	"ai-agent/workflow/model/entity"
	"ai-agent/workflow/service/flow/processor"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
	"github.com/cloudwego/eino-examples/compose/batch/batch"
	"github.com/cloudwego/eino/compose"
	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/google/uuid"
)

// StartLambda 启动节点
func StartLambda(ctx context.Context, input any) (any, error) {
	return input, nil
}

// FormLambda 表单调用节点
func FormLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}
	// 解析 valueSource 引用，填充表单节点输出配置（供下游引用）
	for _, output := range nodeInput.Config.OutputConfig {
		ProcessValueSourceRecursive(output, nodeInput.Global)
	}
	return nodeInput, nil
}

// ModelLambda 模型调用节点
func ModelLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}

	// 1. 解析 valueSource 引用，填充模型请求参数
	ProcessValueSourceRecursive(nodeInput.Config.ModelConfig.ModelRequestParams, nodeInput.Global)

	// 1.5 剔除 value 为空的字段；数组/枚举元素整体为空时移除整个元素（0/false 视为有效值）
	CleanEmptyModelParams(nodeInput.Config.ModelConfig.ModelRequestParams)

	// 2. 前置工具：决定模型调用入参（单次/多次）
	paramsList, err := invokePreTool(ctx, nodeInput.Config.PreTool, nodeInput.Config.ModelConfig.ModelRequestParams)
	if err != nil {
		return nil, err
	}

	// 3. 逐批调用模型，汇总输出（保持请求顺序），累计 token/费用供节点记录落库
	var outputRes []map[string]any
	var totalTokens int64
	var totalCost float64
	if nodeInput.Config.IsBatchExec && len(paramsList) > 1 {
		// 异步批量执行：并发请求模型，等待全部返回后再继续，避免下游读到空结果
		results := make([][]map[string]any, len(paramsList))
		tokenRes := make([]*gateway.ModelCallRes, len(paramsList))
		errs := make([]error, len(paramsList))
		var wg sync.WaitGroup
		for i, params := range paramsList {
			wg.Add(1)
			go func(i int, params map[string]any) {
				defer wg.Done()
				results[i], tokenRes[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params)
			}(i, params)
		}
		wg.Wait()
		for i, res := range results {
			if errs[i] != nil {
				return nil, errs[i]
			}
			if tokenRes[i] != nil {
				totalTokens += tokenRes[i].TotalTokens
				totalCost += tokenRes[i].Cost
			}
			outputRes = append(outputRes, res...)
		}
	} else {
		for _, params := range paramsList {
			res, modelRes, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params)
			if err != nil {
				return nil, err
			}
			if modelRes != nil {
				totalTokens += modelRes.TotalTokens
				totalCost += modelRes.Cost
			}
			outputRes = append(outputRes, res...)
		}
	}

	// 3.5 把本次节点消耗的 token/费用写入节点执行记录，供汇总节点聚合到 exec_workflow
	if nodeInput.NodeExecutionId > 0 && (totalTokens > 0 || totalCost > 0) {
		if _, err := nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{
			Id: nodeInput.NodeExecutionId,
			TokenInfo: []map[string]any{{
				"total_tokens": totalTokens,
				"total_fee":    totalCost,
			}},
		}); err != nil {
			return nil, fmt.Errorf("节点：%v 写入token信息失败: %v", nodeInput.Config.Name, err)
		}
	}

	// 4. 后置工具：加工模型输出（透传原始请求参数，供后置工具读取合并配置等）
	outputRes, err = invokePostTool(ctx, nodeInput.Config.PostTool, outputRes, nodeInput.Config.ModelConfig.ModelRequestParams)
	if err != nil {
		return nil, err
	}

	nodeInput.Config.OutputResult = outputRes
	return nodeInput, nil
}

// invokePreTool 执行前置处理器，把模型请求参数转换为模型调用入参列表。
// 前置处理器契约：入参即模型请求参数本体；返回值：
//   - map[string]any   一次模型调用，入参为返回值
//   - []map[string]any 多次模型调用，逐个入参请求
//   - nil              视为异常，节点失败（不允许静默跳过模型调用）
func invokePreTool(ctx context.Context, processorName string, modelParams map[string]any) (paramsList []map[string]any, err error) {
	if processorName == "" {
		return []map[string]any{modelParams}, nil
	}
	data, err := processor.Call(ctx, processorName, modelParams)
	if err != nil {
		return nil, fmt.Errorf("执行前置处理器[%s]失败: %v", processorName, err)
	}
	switch v := data.(type) {
	case nil:
		return nil, fmt.Errorf("前置处理器[%s]返回空", processorName)
	case map[string]any:
		return []map[string]any{v}, nil
	case []map[string]any:
		return v, nil
	default:
		return nil, fmt.Errorf("前置处理器[%s]返回类型不支持: %T", processorName, data)
	}
}

// invokePostTool 执行后置处理器，加工模型调用结果。
// 后置处理器契约：入参 {"output": 模型输出结果列表, "request": 原始模型请求参数}（列表须包成对象传入）；返回值：
//   - []map[string]any 替换模型输出
//   - map[string]any   替换为单条输出
//   - nil              保留原输出
func invokePostTool(ctx context.Context, processorName string, outputRes []map[string]any, requestParams map[string]any) ([]map[string]any, error) {
	if processorName == "" {
		return outputRes, nil
	}
	data, err := processor.Call(ctx, processorName, map[string]any{"output": outputRes, "request": requestParams})
	if err != nil {
		return nil, fmt.Errorf("执行后置处理器[%s]失败: %v", processorName, err)
	}
	switch v := data.(type) {
	case nil:
		return outputRes, nil
	case []map[string]any:
		return v, nil
	case map[string]any:
		return []map[string]any{v}, nil
	default:
		return nil, fmt.Errorf("后置处理器[%s]返回类型不支持: %T", processorName, data)
	}
}

func SubFlowLambda(ctx context.Context, input any) (any, error) {
	// 1. 类型断言（和其他节点保持一致的入参结构）
	nodeExecInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("子流程节点入参类型错误，期望*flowDto.NodeExecutionInput，实际%T", input)
	}
	// 2. 解析子流程配置
	subFlowConfig := nodeExecInput.Config.SubConfig
	if subFlowConfig == nil {
		return nil, fmt.Errorf("子流程节点缺少配置")
	}
	getRes, err := FlowUserService.Get(ctx, &flowDto.GetFlowUserReq{
		Id: subFlowConfig.WorkflowId,
	})
	if err != nil {
		return nil, err
	}
	// 3. 引入参数解析：把首页表单值/上游引用值/静态默认值写入子流程开始节点 outputConfig。
	//    须在 BuildGraph / ExtractFlowNodeFrom 之前执行，batchInputs 深拷贝的才是注入后的开始节点。
	injectSubFlowFields(nodeExecInput.Global, getRes.FlowContent, subFlowConfig.Fields)
	// 4. 并发数：从主流程开始节点 outputConfig 的 maxConcurrency 字段读取（前端把子流程节点生成次数表单字段聚合到主流程开始节点），读不到再用子流程节点配置兜底
	maxConcurrency := mainFlowMaxConcurrency(nodeExecInput.Global, subFlowConfig.MaxConcurrency)
	// 4. 编译子流程Graph（复用现有 BuildGraphFromFlowContent 逻辑）
	nodeList, subGraph := BuildGraph(ctx, getRes.FlowContent)
	// 4. 构建子流程Workflow（绑定START/END，和示例对齐）
	innerWorkflow := compose.NewWorkflow[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]()
	// 挂载子图节点并绑定全局START
	innerWorkflow.AddGraphNode("sub_flow_graph", subGraph).AddInput(compose.START)
	// 绑定子图输出到全局END
	innerWorkflow.End().AddInput("sub_flow_graph")
	// 生成次数（批量条数）：maxConcurrency<=0 时按 1 次兜底
	batchCount := maxConcurrency
	if batchCount <= 0 {
		batchCount = 1
	}
	// 5. 构建BatchNode（批量执行子流程，复用示例逻辑）
	batchNode := batch.NewBatchNode(&batch.NodeConfig[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]{
		Name:           fmt.Sprintf("sub_flow_batch_%s", nodeExecInput.Config.Id),
		InnerTask:      innerWorkflow,
		MaxConcurrency: batchCount,
	})

	// 6. 提取批量输入：按生成次数生成 N 份（每份独立克隆 ConfigMap，避免并发执行时节点输出写串）
	nodeInputParams := ExtractFlowNodeFrom(getRes.FlowContent)
	configMap := make(map[string]*entity.FlowNode)
	for _, cfg := range nodeInputParams {
		configMap[cfg.Id] = cfg
	}
	for _, i := range nodeList {
		configMap[i.Id] = &i
	}
	batchInputs := make([]*flowDto.FlowExecutionInput, 0, batchCount)
	for j := 0; j < batchCount; j++ {
		batchInputs = append(batchInputs, &flowDto.FlowExecutionInput{
			NodeGroupId: nodeExecInput.Global.NodeGroupId,
			ExecutionId: nodeExecInput.Global.ExecutionId,
			FlowId:      nodeExecInput.Global.FlowId,
			ConfigMap:   cloneConfigMap(configMap),
			SessionId:   nodeExecInput.Global.SessionId,
		})
	}
	// 7. 执行批量子流程
	batchOutput, err := batchNode.Invoke(ctx, batchInputs)
	if err != nil {
		return nil, fmt.Errorf("执行子流程BatchNode失败: %v", err)
	}
	// 8. 展平每份子流程执行的节点输出，写回当前节点 OutputResult 供下游引用
	var outputRes []map[string]any
	for _, single := range batchOutput {
		if single == nil {
			continue
		}
		outputRes = append(outputRes, collectFlowNodeResults(single)...)
	}
	g.Log().Info(ctx, fmt.Sprintf("子流程执行完成，共 %d 次，输出 %d 条", batchCount, len(outputRes)))
	nodeExecInput.Config.OutputResult = outputRes
	return nodeExecInput, nil
}

// injectSubFlowFields 将子流程节点引入参数（subConfig.Fields）解析后写入子流程开始节点
// outputConfig，使子流程启动时能读到首页表单值/上游引用值/静态默认值。
// 每个字段的取值优先级：valueSource 引用解析成功 → field.value → field.defaultValue；
// 匹配键为 field（前端约定以 field 为主，不兼容 path）。
func injectSubFlowFields(global *flowDto.FlowExecutionInput, subFlowContent *entity.FlowInfo, fields []map[string]any) {
	if global == nil || subFlowContent == nil || len(fields) == 0 {
		return
	}
	startNode := subFlowStartNode(subFlowContent)
	if startNode == nil {
		return
	}
	byField := make(map[string]map[string]any, len(startNode.OutputConfig))
	for _, output := range startNode.OutputConfig {
		byField[gconv.String(output["field"])] = output
	}
	for _, field := range fields {
		entry := byField[gconv.String(field["field"])]
		if entry == nil {
			continue
		}
		value := field["value"]
		if vs, has := field["valueSource"]; has && vs != nil {
			vsMap := gconv.Map(vs)
			vsNodeId := gconv.String(vsMap["nodeId"])
			vsField := gconv.String(vsMap["fieldName"])
			if vsField == "" {
				vsField = gconv.String(vsMap["field"])
			}
			if vsNodeId != "" && vsField != "" {
				if v, _, ok := resolveValueSource(global, vsNodeId, vsField); ok {
					value = v
				}
			}
		}
		if value == nil {
			value = field["defaultValue"]
		}
		if value != nil {
			entry["value"] = value
		}
	}
}

// subFlowStartNode 返回工作流开始节点
func subFlowStartNode(content *entity.FlowInfo) *entity.FlowNode {
	if content == nil {
		return nil
	}
	for i := range content.Nodes {
		if content.Nodes[i].Id == content.StartNodeId {
			return &content.Nodes[i]
		}
	}
	return nil
}

// mainFlowMaxConcurrency 取子流程批量执行并发数：从主流程开始节点 outputConfig
// 的 maxConcurrency 字段读取（前端把子流程节点的生成次数表单字段聚合到主流程开始节点），
// 读不到再用子流程节点配置的兜底值。
func mainFlowMaxConcurrency(global *flowDto.FlowExecutionInput, fallback int) int {
	if global == nil {
		return fallback
	}
	for _, n := range global.ConfigMap {
		if n == nil || n.NodeCode != node.NodeTypeStart {
			continue
		}
		for _, output := range n.OutputConfig {
			if gconv.String(output["field"]) != "maxConcurrency" && gconv.String(output["path"]) != "maxConcurrency" {
				continue
			}
			if v := gconv.Int(output["value"]); v > 0 {
				return v
			}
		}
		return fallback
	}
	return fallback
}

// cloneConfigMap 深拷贝 ConfigMap，保证各批次子流程并发执行时节点输出互不串扰。
// 浅拷贝会共享 *entity.FlowNode，并发写 OutputResult 产生竞态。
func cloneConfigMap(src map[string]*entity.FlowNode) map[string]*entity.FlowNode {
	dst := make(map[string]*entity.FlowNode, len(src))
	for k, v := range src {
		data, err := json.Marshal(v)
		if err != nil {
			dst[k] = v
			continue
		}
		n := new(entity.FlowNode)
		if err = json.Unmarshal(data, n); err != nil {
			dst[k] = v
			continue
		}
		dst[k] = n
	}
	return dst
}

// collectFlowNodeResults 收集一次子流程执行中所有已执行节点的输出，展平成 {字段:值} 列表
func collectFlowNodeResults(execInput *flowDto.FlowExecutionInput) []map[string]any {
	var res []map[string]any
	for _, executed := range execInput.ExecutedNodes {
		if nodeConfig := execInput.ConfigMap[executed.NodeId]; nodeConfig != nil {
			res = append(res, nodeConfig.OutputResult...)
		}
	}
	return res
}

// HttpLambda 构建HTTP(S)接口
func HttpLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}
	outputRes, err := HttpCallResultLambda(ctx, nodeInput)
	if err != nil {
		return nil, err
	}
	nodeInput.Config.OutputResult = outputRes
	return nodeInput, nil
}

func DataMergeLambda(ctx context.Context, input any) (res any, err error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("参数合并入参类型错误")
	}
	return nodeInput, nil
}

func SummaryLambda(ctx context.Context, input any) (any, error) {
	execInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("汇总节点入参类型错误，实际是 %T", input)
	}

	// 聚合所有已执行节点中需入库的文件结果（两层规则）
	summaryResult := collectSaveFileResults(ctx, execInput.Global)

	// 把汇总结果存入当前节点的输出
	g.Log().Info(ctx, fmt.Sprintf("结果汇总完成，汇总数据：%+v", summaryResult))

	err := gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		res, _, err := nodeDao.NodeExecutionDao.ListByFlowExecutionId(ctx, &nodeDto.ListNodeExecutionByFlowReq{
			NodeGroupId: execInput.Global.NodeGroupId,
		}, entity.NodeExecutionCol.TokenInfo)
		if err != nil {
			return err
		}
		var totalTokens int
		var totalFee float64
		for _, item := range res {
			for _, itemToken := range item.TokenInfo {
				m := gconv.Map(itemToken)
				totalTokens += gconv.Int(m["total_tokens"])
				totalFee += gconv.Float64(m["total_fee"])
			}
		}
		_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
			Id:          execInput.Global.ExecutionId,
			Status:      flow.FlowExecutionStatusSuccess.Code(),
			TotalTokens: totalTokens,
			TotalFee:    totalFee,
		})
		if err != nil {
			return err
		}
		if len(summaryResult) > 0 {
			_, err = sessionDao.ExecWorkflowResultDao.BatchInsert(ctx, summaryResult)
			if err != nil {
				return err
			}
		}
		return nil
	})

	return execInput, err
}

// collectSaveFileResults 按两层规则收集需入库的文件结果：
// 第一层：节点须开启"保存文件"（IsSaveFile）；
// 第二层：key 取自模型响应内容 respBody（即节点 OutputResult 的各字段），命中
// ModelResponseBodyMapping 才入库；原始响应体 key（respBody）恒入库（不要求映射声明）。
// 结果值为 http(s) URL 直接使用；非路径值（base64 图片/文本）先上传 OSS 换取 URL，
// 文本内容以 .inc 扩展名存储。
func collectSaveFileResults(ctx context.Context, execInput *flowDto.FlowExecutionInput) []*sessionDto.CreateWorkflowResultReq {
	if execInput == nil {
		return nil
	}
	var summaryResult []*sessionDto.CreateWorkflowResultReq
	for _, executedNode := range execInput.ExecutedNodes {
		nodeConfig := execInput.ConfigMap[executedNode.NodeId]
		if nodeConfig == nil || len(nodeConfig.OutputResult) == 0 || !nodeConfig.IsSaveFile {
			continue
		}
		// 第二层：key 取自模型响应内容 respBody（即节点 OutputResult 的各字段），
		// 命中 ModelResponseBodyMapping 才入库；原始响应体 key（respBody）恒入库
		saveKeys := nodeConfig.ModelConfig.ModelResponseBodyMapping
		for _, respBody := range nodeConfig.OutputResult {
			for key, val := range gconv.Map(respBody) {
				if _, ok := saveKeys[key]; !ok {
					if key != "respBody" {
						continue
					}
				}
				fileUrl, err := resolveSaveFileResult(ctx, val)
				if err != nil {
					g.Log().Warningf(ctx, "collectSaveFileResults 上传结果文件失败 key=%s err=%v", key, err)
					continue
				}
				summaryResult = append(summaryResult, &sessionDto.CreateWorkflowResultReq{
					SessionId:     execInput.SessionId,
					FlowId:        execInput.FlowId,
					ExecId:        execInput.ExecutionId,
					ResultFileUrl: fileUrl,
				})
			}
		}
	}
	return summaryResult
}

// resolveSaveFileResult 解析结果值为可入库的 URL：
//   - 已是 http(s) URL → 直接返回
//   - 非路径（base64 图片/文本）→ 上传 OSS 换取 URL
func resolveSaveFileResult(ctx context.Context, val any) (string, error) {
	isPath, path, fileBytes, ext := resolveFileContent(val)
	if isPath {
		return path, nil
	}
	if ext == "" {
		ext = ".png"
	}
	fileUrl, err := gateway.Upload(ctx, fmt.Sprintf("workflow_result_%s%s", uuid.NewString(), ext), fileBytes)
	if err != nil {
		return "", err
	}
	return fileUrl, nil
}

// resolveFileContent 判断结果值形态：
//   - 已是 URL 路径（http/https 开头）→ 直接使用
//   - data URI（data:<mime>;base64,<data>）→ 解码为字节，扩展名按 mime 推断
//   - 纯 base64（可解码且长度足以认为是编码数据）→ 解码为字节，默认 .png
//   - 其余（文本）→ 以 .inc 扩展名上传原文
func resolveFileContent(val any) (isPath bool, path string, fileBytes []byte, ext string) {
	s := gconv.String(val)
	if isFileURL(s) {
		return true, s, nil, ""
	}
	// data URI：data:<mime>;base64,<payload>
	if b, mime, ok := parseDataURI(s); ok {
		return false, "", b, extOfMime(mime)
	}
	// 纯 base64：可解码且长度足够，视为编码后的文件内容
	trimmed := strings.TrimSpace(s)
	if len(trimmed) >= 64 {
		if b, err := base64.StdEncoding.DecodeString(trimmed); err == nil && len(b) > 0 {
			return false, "", b, ".png"
		}
	}
	// 文本：以 .inc 存储
	return false, "", []byte(s), ".inc"
}

// isFileURL 判断字符串是否已是对外可访问的 URL 路径（http/https 开头）
func isFileURL(s string) bool {
	lower := strings.ToLower(s)
	return strings.HasPrefix(lower, "http://") || strings.HasPrefix(lower, "https://")
}

// extOfMime 按 MIME 类型推断文件扩展名
func extOfMime(mime string) string {
	switch strings.ToLower(strings.TrimSpace(mime)) {
	case "image/png", "png":
		return ".png"
	case "image/jpeg", "image/jpg", "jpeg", "jpg":
		return ".jpg"
	case "image/webp":
		return ".webp"
	case "image/gif":
		return ".gif"
	case "audio/mpeg", "audio/mp3", "mp3":
		return ".mp3"
	case "audio/wav", "wav":
		return ".wav"
	case "video/mp4", "mp4":
		return ".mp4"
	case "application/json", "json":
		return ".json"
	default:
		return ""
	}
}

// parseDataURI 解析 data URI：data:<mime>;base64,<payload>，返回解码字节与 mime
func parseDataURI(s string) ([]byte, string, bool) {
	const prefix = "data:"
	if !strings.HasPrefix(s, prefix) {
		return nil, "", false
	}
	rest := s[len(prefix):]
	comma := strings.Index(rest, ",")
	if comma < 0 {
		return nil, "", false
	}
	mime := rest[:comma]
	if semicolon := strings.Index(mime, ";"); semicolon >= 0 {
		mime = mime[:semicolon]
	}
	payload := strings.TrimPrefix(rest[comma+1:], "base64,")
	b, err := base64.StdEncoding.DecodeString(payload)
	if err != nil {
		return nil, "", false
	}
	return b, mime, true
}