ai-agent/workflow/dao/session/session_dao.go

package session

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

	"gitea.redpowerfuture.com/red-future/common/beans"
	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
)

var (
	SessionDao               = &sessionDao{}
	WorkflowSessionResultDao = &workflowSessionResultDao{}
	ChatSessionResultDao     = &chatSessionResultDao{}
)

type sessionDao struct{}

func (d *sessionDao) Insert(ctx context.Context, req *sessionDto.CreateSessionReq) (id int64, err error) {
	var s = new(entity.Session)
	if err = gconv.Struct(req, &s); err != nil {
		return
	}
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameSession).Insert(s)
	if err != nil {
		return
	}
	return r.LastInsertId()
}

func (d *sessionDao) Get(ctx context.Context, id int64) (res *entity.Session, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameSession).
		Where(entity.SessionCol.Id, id).
		Where("deleted_at IS NULL").
		One()
	if err != nil {
		return
	}
	err = r.Struct(&res)
	return
}

func (d *sessionDao) List(ctx context.Context, creator string, page *beans.Page) (res []*entity.Session, total int, err error) {
	m := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameSession).
		Where(entity.SessionCol.Creator, creator).
		Where("deleted_at IS NULL")
	m.OrderDesc(entity.SessionCol.CreatedAt)
	if page != nil {
		m.Page(int(page.PageNum), int(page.PageSize))
	}
	r, total, err := m.AllAndCount(false)
	if err != nil {
		return
	}
	err = r.Structs(&res)
	return
}

// DeleteCascade 软删会话，并级联软删该会话下所有普通对话结果（工作流结果保留）
func (d *sessionDao) DeleteCascade(ctx context.Context, id int64) (err error) {
	return gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameSession).
			Data(g.Map{
				entity.SessionCol.DeletedAt: gtime.Now(),
				entity.SessionCol.Updater:   "",
			}).
			Where(entity.SessionCol.Id, id).
			Where("deleted_at IS NULL").
			Update()
		if err != nil {
			return err
		}
		_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameChatSessionResult).
			Data(g.Map{
				entity.ChatSessionResultCol.DeletedAt: gtime.Now(),
				entity.ChatSessionResultCol.Updater:   "",
			}).
			Where(entity.ChatSessionResultCol.SessionId, id).
			Where("deleted_at IS NULL").
			Update()
		return err
	})
}

type workflowSessionResultDao struct{}

func (d *workflowSessionResultDao) Insert(ctx context.Context, w *entity.WorkflowSessionResult) (id int64, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).Insert(w)
	if err != nil {
		return
	}
	return r.LastInsertId()
}

func (d *workflowSessionResultDao) GetLatestBySessionAndFlow(ctx context.Context, sessionId, flowId int64) (res *entity.WorkflowSessionResult, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Where(entity.WorkflowSessionResultCol.SessionId, sessionId).
		Where(entity.WorkflowSessionResultCol.FlowId, flowId).
		Where("deleted_at IS NULL").
		OrderDesc(entity.WorkflowSessionResultCol.CreatedAt).
		One()
	if err != nil {
		return
	}
	err = r.Struct(&res)
	return
}

func (d *workflowSessionResultDao) ListBySession(ctx context.Context, sessionId int64) (res []*entity.WorkflowSessionResult, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Where(entity.WorkflowSessionResultCol.SessionId, sessionId).
		Where("deleted_at IS NULL").
		OrderDesc(entity.WorkflowSessionResultCol.CreatedAt).
		All()
	if err != nil {
		return
	}
	err = r.Structs(&res)
	return
}

func (d *workflowSessionResultDao) ListWorkflowResults(ctx context.Context, creator string, flowId int64, page *beans.Page) (res []*entity.WorkflowSessionResult, total int, err error) {
	m := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Where(entity.WorkflowSessionResultCol.Creator, creator).
		Where("deleted_at IS NULL")
	if flowId != 0 {
		m.Where(entity.WorkflowSessionResultCol.FlowId, flowId)
	}
	m.OrderDesc(entity.WorkflowSessionResultCol.CreatedAt)
	if page != nil {
		m.Page(int(page.PageNum), int(page.PageSize))
	}
	r, total, err := m.AllAndCount(false)
	if err != nil {
		return
	}
	err = r.Structs(&res)
	return
}

func (d *workflowSessionResultDao) UpdateRunning(ctx context.Context, id int64) (err error) {
	_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Data(g.Map{
			entity.WorkflowSessionResultCol.Status:  sessionDto.ResultStatusRunning,
			entity.WorkflowSessionResultCol.Updater: "",
		}).
		Where(entity.WorkflowSessionResultCol.Id, id).
		Update()
	return
}

func (d *workflowSessionResultDao) UpdateFailed(ctx context.Context, id int64, errorMsg string) (err error) {
	_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Data(g.Map{
			entity.WorkflowSessionResultCol.Status:       sessionDto.ResultStatusFailed,
			entity.WorkflowSessionResultCol.ErrorMessage: errorMsg,
			entity.WorkflowSessionResultCol.Updater:      "",
		}).
		Where(entity.WorkflowSessionResultCol.Id, id).
		Update()
	return
}

func (d *workflowSessionResultDao) UpdateSuccess(ctx context.Context, id int64, resultParams []map[string]any, totalTokens int, totalFee float64) (err error) {
	_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Data(g.Map{
			entity.WorkflowSessionResultCol.Status:       sessionDto.ResultStatusSuccess,
			entity.WorkflowSessionResultCol.ResultParams: resultParams,
			entity.WorkflowSessionResultCol.TotalTokens:  totalTokens,
			entity.WorkflowSessionResultCol.TotalFee:     totalFee,
			entity.WorkflowSessionResultCol.Updater:      "",
		}).
		Where(entity.WorkflowSessionResultCol.Id, id).
		Update()
	return
}

func (d *workflowSessionResultDao) SoftDelete(ctx context.Context, id int64) (err error) {
	_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameWorkflowSessionResult).
		Data(g.Map{
			entity.WorkflowSessionResultCol.DeletedAt: gtime.Now(),
			entity.WorkflowSessionResultCol.Updater:   "",
		}).
		Where(entity.WorkflowSessionResultCol.Id, id).
		Where("deleted_at IS NULL").
		Update()
	return
}

type chatSessionResultDao struct{}

func (d *chatSessionResultDao) Insert(ctx context.Context, c *entity.ChatSessionResult) (id int64, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameChatSessionResult).Insert(c)
	if err != nil {
		return
	}
	return r.LastInsertId()
}

func (d *chatSessionResultDao) ListBySession(ctx context.Context, sessionId int64) (res []*entity.ChatSessionResult, err error) {
	r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameChatSessionResult).
		Where(entity.ChatSessionResultCol.SessionId, sessionId).
		Where("deleted_at IS NULL").
		OrderDesc(entity.ChatSessionResultCol.CreatedAt).
		All()
	if err != nil {
		return
	}
	err = r.Structs(&res)
	return
}

func (d *chatSessionResultDao) SoftDelete(ctx context.Context, id int64) (err error) {
	_, err = gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameChatSessionResult).
		Data(g.Map{
			entity.ChatSessionResultCol.DeletedAt: gtime.Now(),
			entity.ChatSessionResultCol.Updater:   "",
		}).
		Where(entity.ChatSessionResultCol.Id, id).
		Where("deleted_at IS NULL").
		Update()
	return
}