package flow

import (
	"ai-agent/workflow/consts/public"
	"ai-agent/workflow/model/entity"
	"context"

	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
)

const (
	FlowAsyncStateInflight = 0  // 任务已提交，结果未取（执行中/结果已发布未取/失败）
	FlowAsyncStateDone     = 1  // 成功，结果已缓存
	FlowAsyncStateFailed   = 2  // 确定失败
	FlowAsyncSegSentinel   = -1 // 非段异步调用的段序号哨兵值
)

var FlowAsyncTaskDao = &flowAsyncTaskDao{}

type flowAsyncTaskDao struct{}

// 缓存唯一键是 (node_group_id, node_id, segment_index)：node_group_id 是"逻辑运行(attempt)"标识，
// 全新/换参重跑换新组 = 换新键；续跑/恢复复用 exec 记录的组 = 读到同组活行。
// 删除只有软删且只作用于"终态组"（成功尾部 / 换参重跑废弃的旧组），此类组此后永不再写，
// 软删行不会被同键重存（组永不复活）——与 flow_segment_result 同一约定。
// execution_id 仅作溯源保留，不参与唯一键。

// Get 查询唯一键 (node_group_id, node_id, segment_index) 的记录
func (d *flowAsyncTaskDao) Get(ctx context.Context, nodeGroupId string, nodeId string, segIdx int) (res *entity.FlowAsyncTask, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask).
		Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId).
		Where(entity.FlowAsyncTaskCol.NodeId, nodeId).
		Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx).
		One()
	if err != nil {
		return nil, err
	}
	if r.IsEmpty() {
		return nil, nil
	}
	err = r.Struct(&res)
	return
}

// Upsert 提交时写入/更新 in-flight 行：唯一键冲突则更新 model_id/task_id/msg_topic/state（保留已完成结果不动）。
// 与 flow_segment_result 相同走 OnConflict().Save()；两点相对初稿的调整：
//  1. Result 列是 JSONB，空串会被 PG 拒绝（invalid input syntax for type json），
//     提交时本无结果，统一写 '{}' 占位（与表列默认值一致）。
//  2. 必须用 OnDuplicate 限定冲突更新列——GoFrame Save 默认把 Data 里所有列写进
//     ON CONFLICT DO UPDATE SET，若不限定会把已缓存的结果覆盖成 '{}'，与"保留已完成结果"矛盾。
//
// 提交/重提统一走本函数重置同组活行（state→in-flight + 新 task_id/msg_topic）：不再"删行重建"，
// 活行从未被软删，OnConflict 更新即可复位，无墓碑冲突、无物理删除。
func (d *flowAsyncTaskDao) Upsert(ctx context.Context, nodeGroupId string, execId int64, nodeId string, segIdx int, modelId, taskId int64, msgTopic string) error {
	rec := &entity.FlowAsyncTask{
		NodeGroupId:  nodeGroupId,
		ExecutionId:  execId,
		NodeId:       nodeId,
		SegmentIndex: segIdx,
		ModelId:      modelId,
		TaskId:       taskId,
		MsgTopic:     msgTopic,
		State:        FlowAsyncStateInflight,
		Result:       "{}",
	}
	_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).
		Model(ctx, public.TableNameFlowAsyncTask).
		Data(rec).
		OnConflict(entity.FlowAsyncTaskCol.NodeGroupId, entity.FlowAsyncTaskCol.NodeId, entity.FlowAsyncTaskCol.SegmentIndex).
		OnDuplicate(entity.FlowAsyncTaskCol.ModelId, entity.FlowAsyncTaskCol.TaskId, entity.FlowAsyncTaskCol.MsgTopic, entity.FlowAsyncTaskCol.State).
		Save()
	return err
}

// UpdateByKey 按唯一键更新 state/result（OmitNil 丢弃 nil 字段，map 值非 nil 全写入；
// state=0 也能落库）。Result 列是 JSONB，空串无法写入，统一落 '{}' 表示无结果。
func (d *flowAsyncTaskDao) UpdateByKey(ctx context.Context, nodeGroupId string, nodeId string, segIdx int, state int, result string) error {
	resultVal := result
	if resultVal == "" {
		resultVal = "{}"
	}
	_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask).
		Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId).
		Where(entity.FlowAsyncTaskCol.NodeId, nodeId).
		Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx).
		Data(map[string]any{
			entity.FlowAsyncTaskCol.State:  state,
			entity.FlowAsyncTaskCol.Result: resultVal,
		}).
		Update()
	return err
}

// DeleteByGroup 软删指定逻辑运行(组)的异步任务缓存（gfdb Model.Delete 在 deletedAt 配置下退化为软删）。
// 仅允许对"终态组"调用：① 工作流执行成功后（BuildExecution 尾部）；② 同一条 exec 换参重跑（forceNewRun）
// 废弃的旧组（execute 重置成功后回收）。终态组此后永不再被读写 → 软删行不复活、不占同键写入冲突。
// 失败/取消/重试耗尽一律不删（保留组行供同参数续跑复用）。绝无物理删除。
func (d *flowAsyncTaskDao) DeleteByGroup(ctx context.Context, nodeGroupId string) error {
	_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).
		Model(ctx, public.TableNameFlowAsyncTask).
		Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId).
		Delete()
	return err
}