package service

import (
	"context"
	"fmt"
	"model-gateway/consts/model"
	"model-gateway/dao"
	"model-gateway/model/dto"
	"model-gateway/model/entity"
	modelUtils "model-gateway/service/utils"
	"net/http"
	"time"

	"gitea.redpowerfuture.com/red-future/common/utils"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
)

var ModelCall = &modelCallService{}

type modelCallService struct{}

func (s *modelCallService) ModelCall(ctx context.Context, req *dto.ModelCallReq) (res *dto.ModelCallRes, err error) {
	// 1) 检查模型配置
	var modelInfo *entity.ModelManage
	modelInfo, err = dao.ModelManage.GetNotTenantId(ctx, &dto.GetModelManageReq{
		Id: req.ModelId,
	})
	if err != nil {
		return nil, fmt.Errorf("获取模型配置失败: %v", err)
	}
	if modelInfo == nil {
		return nil, fmt.Errorf("模型不存在")
	}
	// 引用行 → 解析为系统模型配置+本人 apiKey；系统模型已删除等解析失败 → 阻塞调用
	modelInfo, err = modelUtils.ResolveModelConfig(ctx, modelInfo)
	if err != nil {
		return nil, err
	}
	if modelInfo.Enabled != nil && !*modelInfo.Enabled {
		return nil, fmt.Errorf("模型不存在或未启用")
	}

	now := time.Now()
	userInfo, err := utils.GetUserInfo(ctx)
	if err != nil {
		return
	}
	if !g.IsEmpty(modelInfo.RefSystemModelId) {
		// 调用前检查模型计价配置（shop-user-trade）：未配置/未启用 → 阻塞调用（subject=解析后的系统模型 id）
		if err = modelBillable(ctx, modelInfo.Id); err != nil {
			return nil, err
		}
	}
	err = queue(ctx, modelInfo.ModelName, userInfo.TenantId, gconv.Int64(modelInfo.MaxConcurrency), func(ctx context.Context) (err error) {
		if *modelInfo.ResponseType == *model.ResponseTypeSync.Code() || *modelInfo.ResponseType == *model.ResponseTypeStream.Code() {
			var newRequestParams map[string]any
			var id int64
			id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, req)
			if err != nil {
				return err
			}
			if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
				return fmt.Errorf("保存模型请求参数失败")
			}
			if *modelInfo.ResponseType == *model.ResponseTypeSync.Code() {
				res, err = ModelSession.CreateSession(ctx, &dto.CallModelSessionReq{
					Id:            id,
					ModelInfo:     modelInfo,
					RequestParams: newRequestParams,
				})
			} else {
				res, err = ModelSession.CreateSessionStreamOnce(ctx, &dto.CallModelSessionReq{
					Id:            id,
					ModelInfo:     modelInfo,
					RequestParams: newRequestParams,
				})
			}
		}
		if *modelInfo.ResponseType == *model.ResponseTypeAsync.Code() {
			if g.IsEmpty(req.MsgTopic) {
				return fmt.Errorf("请指定消息主题")
			}
			var newRequestParams map[string]any
			var id int64
			id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, req)
			if err != nil {
				return err
			}
			if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
				return fmt.Errorf("保存模型请求参数失败")
			}
			res, err = ModelTaskStart.CreateTask(ctx, &dto.CallModelTaskStartReq{
				Id:            id,
				ModelInfo:     modelInfo,
				RequestParams: newRequestParams,
			})
		}
		return
	})
	return
}

func (s *modelCallService) ModelCallStream(ctx context.Context, w http.ResponseWriter, req *dto.ModelCallStreamReq) (err error) {
	// 1) 检查模型配置
	var modelInfo *entity.ModelManage
	modelInfo, err = dao.ModelManage.GetNotTenantId(ctx, &dto.GetModelManageReq{
		Id: req.ModelId,
	})
	if err != nil {
		return fmt.Errorf("获取模型配置失败: %v", err)
	}
	if modelInfo == nil {
		return fmt.Errorf("模型不存在")
	}
	// 引用行 → 解析为系统模型配置+本人 apiKey；系统模型已删除等解析失败 → 阻塞调用
	modelInfo, err = modelUtils.ResolveModelConfig(ctx, modelInfo)
	if err != nil {
		return err
	}
	if modelInfo.Enabled != nil && !*modelInfo.Enabled {
		return fmt.Errorf("模型不存在或未启用")
	}
	if *modelInfo.ResponseType == *model.ResponseTypeStream.Code() {
		now := time.Now()
		userInfo, err := utils.GetUserInfo(ctx)
		if err != nil {
			return err
		}
		if !g.IsEmpty(modelInfo.RefSystemModelId) {
			// 调用前检查模型计价配置（shop-user-trade）：未配置/未启用 → 阻塞调用（subject=解析后的系统模型 id）
			if err = modelBillable(ctx, modelInfo.Id); err != nil {
				return err
			}
		}
		err = queue(ctx, modelInfo.ModelName, userInfo.TenantId, gconv.Int64(modelInfo.MaxConcurrency), func(ctx context.Context) (err error) {
			var newRequestParams map[string]any
			var id int64
			id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, &dto.ModelCallReq{
				ModelId:        req.ModelId,
				RequestParams:  req.RequestParams,
				BusinessParams: req.BusinessParams,
				SessionId:      req.SessionId,
				BizName:        req.BizName,
			})
			if err != nil {
				return err
			}
			if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
				return fmt.Errorf("保存模型请求参数失败")
			}
			_, err = ModelSession.CreateSessionStream(ctx, w, &dto.CallModelSessionReq{
				Id:            id,
				ModelInfo:     modelInfo,
				RequestParams: newRequestParams,
			})
			return
		})
	} else {
		return fmt.Errorf("模型响应类型错误")
	}
	return err
}

// saveModelRequestParams 保存模型请求参数
func (s *modelCallService) saveModelRequestParams(ctx context.Context, now time.Time, modelInfo *entity.ModelManage, req *dto.ModelCallReq) (id int64, newRequestParams map[string]any, err error) {

	// 统一走模板校验+构建：requestParams 只装模板字段，businessParams 只装业务字段
	out, err := buildChatRequestParams(modelInfo, req.RequestParams, req.BusinessParams)
	if err != nil {
		return 0, nil, err
	}

	// 1) 上传模型原始请求参数文件（requestParams + businessParams 合并，保证审计完整）
	originalParams := make(map[string]any, len(req.RequestParams)+len(req.BusinessParams))
	for k, v := range req.RequestParams {
		originalParams[k] = v
	}
	for k, v := range req.BusinessParams {
		originalParams[k] = v
	}
	uploadOriginalReq, err := Upload(ctx, &dto.UploadFileBytesReq{
		FileBytes: gconv.Bytes(gconv.String(originalParams)),
		FileName:  fmt.Sprintf("modelRequestParams:%v.json", now.UnixMilli()),
	})
	if err != nil {
		return 0, nil, fmt.Errorf("上传模型原始请求参数文件失败: %v", err)
	}

	// 2) 上传模型解析成功的请求参数文件
	uploadNewReq, err := Upload(ctx, &dto.UploadFileBytesReq{
		FileBytes: gconv.Bytes(gconv.String(out)),
		FileName:  fmt.Sprintf("modelNewRequestParams:%v.json", now.UnixMilli()),
	})
	if err != nil {
		return 0, nil, fmt.Errorf("上传模型解析请求参数文件失败：%v", err)
	}

	// 3) 保存模型请求信息（快照媒体类型=shop 计费词汇 audio/video，空=无媒体引用，任务完成时直接用于算费；模型计费配置任务完成时按 modelId 现查）
	if *modelInfo.ResponseType == *model.ResponseTypeAsync.Code() {
		id, err = dao.ModelTaskStart.Insert(ctx, &dto.CreateModelTaskStartReq{
			ModelId:             req.ModelId,
			BizName:             req.BizName,
			MsgTopic:            req.MsgTopic,
			RequestPath:         uploadNewReq.FileURL,
			OriginalRequestPath: uploadOriginalReq.FileURL,
			MediaType:           modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, out),
		})
		if err != nil {
			return 0, nil, fmt.Errorf("保存模型请求信息失败: %v", err)
		}
	} else {
		id, err = dao.ModelSession.Insert(ctx, &dto.CreateModelSessionReq{
			ModelId:             req.ModelId,
			BizName:             req.BizName,
			SessionId:           req.SessionId,
			RequestPath:         uploadNewReq.FileURL,
			OriginalRequestPath: uploadOriginalReq.FileURL,
		})
		if err != nil {
			return 0, nil, fmt.Errorf("保存模型请求信息失败: %v", err)
		}
	}
	return id, out, nil
}

func queue(ctx context.Context, modelName string, tenantId uint64, maxCon int64, f func(ctx context.Context) (err error)) (err error) {
	const (
		keyExpireSec = 600              // 名额Key兜底过期时间 10min（进程崩溃后自愈）
		refreshStep  = keyExpireSec / 3 // 执行期间续期间隔
		waitInterval = 10 * time.Second // 超限轮询等待间隔
	)
	// Redis 操作统一使用独立上下文，避免外部 ctx canceled
	redisCtx := context.WithoutCancel(ctx)
	concurrencyKey := fmt.Sprintf("model:concurrency:%d:%s", tenantId, modelName)

	// 1) 原子占用并发名额：utils.SemaphoreAcquire 在 WATCH 事务内完成 判满→INCR→首设EXPIRE→超限不写，
	//    不再需要旧 reserveSlot 的「分布式锁 + Incr + 回滚」组合（组合已事务化，外层锁冗余）。
	//    超限（false）按 waitInterval 轮询重试；max<=0 视为不限制（SemaphoreAcquire 内部直接放行）。
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		ok, e := utils.SemaphoreAcquire(redisCtx, concurrencyKey, int(maxCon), keyExpireSec)
		if e != nil {
			return e
		}
		if ok {
			// 展示当前并发数（占用后 GET，与旧 reserveSlot 的 Incr 后计数值语义一致）
			if v, e := g.Redis().Get(redisCtx, concurrencyKey); e == nil {
				glog.Infof(ctx, "并发数: %s %d/%d", concurrencyKey, v.Int64(), maxCon)
			}
			break
		}
		glog.Infof(ctx, "并发超限等待: %s max=%d", concurrencyKey, maxCon)
		time.Sleep(waitInterval)
	}

	// 2) 执行业务期间周期续期名额Key：SemaphoreAcquire 仅在首次占用（计数从 0 起）时设 TTL，
	//    长耗时调用靠本循环持续保活——Key 过期后计数归零会突破 max 并发上限造成超发。
	stop := make(chan struct{})
	go refreshTTL(redisCtx, concurrencyKey, keyExpireSec, refreshStep, stop)
	// 3) 无论业务正常返回还是 panic，都停掉续期并释放名额（幂等，计数归零自动删除 key）
	defer func() {
		close(stop)
		_ = utils.SemaphoreRelease(redisCtx, concurrencyKey)
	}()

	// 4) 执行业务
	return f(ctx)
}

// refreshTTL 周期给名额 Key 续期，直到 stop 关闭；防止长耗时执行期间 Key 提前过期。
func refreshTTL(redisCtx context.Context, concurrencyKey string, keyExpireSec, step int64, stop <-chan struct{}) {
	interval := step
	if interval < 1 {
		interval = 1
	}
	ticker := time.NewTicker(time.Duration(interval) * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			if _, err := g.Redis().Expire(redisCtx, concurrencyKey, keyExpireSec); err != nil {
				glog.Errorf(context.TODO(), "redis refresh concurrency ttl err: %v", err)
			}
		case <-stop:
			return
		}
	}
}

// buildChatRequestParams 按模型配置的请求模板 + 业务字段映射构建请求体（ModelCall 请求路径共用）：
//  1. requestParams 只装模板字段，按 requestBodyMapping 模板校验（CheckParams）+ 构建（ParseConfigTemplate）；
//     未配置映射的字段（如未配置映射的 messages/tools）会被模板拒绝，明确报错
//  2. requestParams 为空时按配置模板构建请求结构（模板 defaultValue 生效），
//     避免"结构由模板声明、值全走业务字段"的场景因请求体为空而构建失败
//  3. businessParams 只装业务字段，按业务字段名（RequestBusinessFieldMapping 的 key）传值，
//     TakeBusinessFields 解析为写入路径，构建完成后由 WriteBusinessFields 按路径写入最终请求体
func buildChatRequestParams(modelInfo *entity.ModelManage, requestParams, businessParams map[string]any) (map[string]any, error) {

	// requestParams 可能混有扁平路径 key（messages.enumValues...）与已是对象/数组的值（stream）。
	// IsFlatMap 遇 map/slice 值即整体返回 false 会跳过 unflatten；sjson.Set 能处理任意值类型作为子树，
	// 带点 key 按路径展开、无点 key 直接赋值，故始终 unflatten
	var err error
	requestParams, err = utils.UnFlatBySjson(requestParams)
	if err != nil {
		return nil, err
	}

	rest := make(map[string]any, len(requestParams))
	for k, v := range requestParams {
		rest[k] = v
	}
	// 请求结构源：模板字段兜底（模板 value/defaultValue 生效）+ requestParams 覆盖同名；
	// 保证模板声明的结构字段（如 stream_options 对象）即使 requestParams 未传也进请求体
	// 用户已传字段按模板类型元数据递归合并（补 type 包装、补默认字段），数组字段仅在用户提供时才合并
	src := rest
	for k, v := range modelInfo.RequestBodyMapping {
		tmplMap, _ := v.(map[string]any)
		if tmplMap != nil {
			if t, _ := tmplMap["type"].(string); t == "array" {
				if _, has := src[k]; !has {
					continue
				}
				src[k] = modelUtils.MergeNode(src[k], v)
				continue
			}
		}
		if userVal, has := src[k]; has {
			src[k] = modelUtils.MergeNode(userVal, v)
		} else {
			src[k] = modelUtils.DeepCopyNode(v)
		}
	}
	if len(requestParams) > 0 {
		// requestParams 非空才按模板严格校验模板字段（空值回填 default）；
		// 为空时跳过，避免业务字段未写入就误报必填缺失
		if err := modelUtils.CheckParams(src, modelInfo.RequestBodyMapping); err != nil {
			return nil, err
		}
	}

	out := modelUtils.ParseConfigTemplate(src)
	if !g.IsEmpty(businessParams) {
		// 业务字段：businessParams 按业务字段名传值，解析为映射路径后写入
		bizValues, err := modelUtils.TakeBusinessFields(businessParams, modelInfo.RequestBusinessFieldMapping)
		if err != nil {
			return nil, err
		}
		// 业务字段按映射路径写入最终请求体（写 out 而非 rest：rest 只是模板字段容器，out 才是下发模型的请求体）
		if err = modelUtils.WriteBusinessFields(out, bizValues); err != nil {
			return nil, err
		}
		// 合并后按模板约束整体校验：必填/长度/范围
		if err = modelUtils.CheckBody(out, modelInfo.RequestBodyMapping); err != nil {
			return nil, err
		}
	}
	// 按模板声明的 type 归一字段值类型（模板字段 value / 业务字段写入值都可能与声明类型不符）
	out = modelUtils.CoerceBodyTypes(out, modelInfo.RequestBodyMapping)
	return out, nil
}