ragflow/internal/admin/service.go

//
//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//

package admin

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/url"
	"ragflow/internal/agent/sandbox"
	"ragflow/internal/common"
	"ragflow/internal/dao"
	"ragflow/internal/engine"
	"ragflow/internal/engine/clickhouse"
	"ragflow/internal/engine/elasticsearch"
	"ragflow/internal/engine/kvrocks"
	"ragflow/internal/entity"
	modelModule "ragflow/internal/entity/models"
	"ragflow/internal/server"
	servicepkg "ragflow/internal/service"
	"ragflow/internal/storage"
	"regexp"
	"time"

	"go.uber.org/zap"
	"gorm.io/gorm"
)

// Service admin service layer
type Service struct {
	userDAO             *dao.UserDAO
	licenseDAO          *dao.LicenseDAO
	timeRecordDAO       *dao.TimeRecordDAO
	systemSettingsDAO   *dao.SystemSettingsDAO
	tenantDAO           *dao.TenantDAO
	userTenantDAO       *dao.UserTenantDAO
	fileDAO             *dao.FileDAO
	documentDAO         *dao.DocumentDAO
	taskDAO             *dao.TaskDAO
	kbDAO               *dao.KnowledgebaseDAO
	canvasDAO           *dao.UserCanvasDAO
	chatDAO             *dao.ChatDAO
	chatSessionDAO      *dao.ChatSessionDAO
	apiTokenDAO         *dao.APITokenDAO
	api4ConvDAO         *dao.API4ConversationDAO
	llmDAO              *dao.LLMDAO
	ingestionTaskDAO    *dao.IngestionTaskDAO
	ingestionTaskLogDao *dao.IngestionTaskLogDAO
	ingestionTaskSvc    *servicepkg.IngestionTaskService

	BillingProductDAO      *dao.BillingProductDAO
	BillingSubscriptionDAO *dao.BillingSubscriptionDAO
	MemoryDAO              *dao.MemoryDAO
	SearchDAO              *dao.SearchDAO
	deleteEngine           engine.DocEngine
	deleteStorage          storage.Storage
}

// NewService create admin service
func NewService() *Service {
	return &Service{
		userDAO:             dao.NewUserDAO(),
		licenseDAO:          dao.NewLicenseDAO(),
		timeRecordDAO:       dao.NewTimeRecordDAO(),
		systemSettingsDAO:   dao.NewSystemSettingsDAO(),
		tenantDAO:           dao.NewTenantDAO(),
		userTenantDAO:       dao.NewUserTenantDAO(),
		fileDAO:             dao.NewFileDAO(),
		documentDAO:         dao.NewDocumentDAO(),
		taskDAO:             dao.NewTaskDAO(),
		kbDAO:               dao.NewKnowledgebaseDAO(),
		canvasDAO:           dao.NewUserCanvasDAO(),
		chatDAO:             dao.NewChatDAO(),
		chatSessionDAO:      dao.NewChatSessionDAO(),
		apiTokenDAO:         dao.NewAPITokenDAO(),
		api4ConvDAO:         dao.NewAPI4ConversationDAO(),
		llmDAO:              dao.NewLLMDAO(),
		ingestionTaskDAO:    dao.NewIngestionTaskDAO(),
		ingestionTaskLogDao: dao.NewIngestionTaskLogDAO(),
		ingestionTaskSvc:    servicepkg.NewIngestionTaskService(),

		BillingProductDAO:      dao.NewBillingProductDAO(),
		BillingSubscriptionDAO: dao.NewBillingSubscriptionDAO(),
		MemoryDAO:              dao.NewMemoryDAO(),
		SearchDAO:              dao.NewSearchDAO(),
	}
}

// Logout user logout
func (s *Service) Logout(ctx context.Context, user interface{}) error {
	// Invalidate token by setting it to INVALID_ prefix
	if u, ok := user.(*entity.User); ok {
		invalidToken := "INVALID_" + generateRandomHex(16)
		return s.userDAO.UpdateAccessToken(ctx, dao.DB, u, invalidToken)
	}
	return nil
}

// ListIngestionTasks list all ingestion tasks for admin user
func (s *Service) ListIngestionTasks(ctx context.Context) ([]map[string]interface{}, error) {
	return s.ingestionTaskSvc.ListAllForAdmin(ctx)
}

func (s *Service) RemoveIngestionTasks(ctx context.Context, tasks []string) ([]map[string]string, error) {
	return s.ingestionTaskSvc.RemoveMany(ctx, tasks, nil)
}

func (s *Service) StopIngestionTasks(ctx context.Context, tasks []string) ([]*entity.IngestionTask, error) {
	return s.ingestionTaskSvc.RequestStopMany(ctx, tasks, nil)
}

// GetUserByToken get user by access token
func (s *Service) GetUserByToken(ctx context.Context, token string) (*entity.User, error) {
	user, err := s.userDAO.GetByAccessToken(ctx, dao.DB, token)
	if err != nil {
		return nil, common.ErrInvalidToken
	}

	if user.IsSuperuser == nil || !*user.IsSuperuser {
		return nil, common.ErrNotAdmin
	}

	if user.IsActive != "1" {
		return nil, fmt.Errorf("user inactive")
	}

	return user, nil
}

// generateRandomHex generate random hex string
func generateRandomHex(n int) string {
	bytes := make([]byte, n)
	rand.Read(bytes)
	return hex.EncodeToString(bytes)
}

// ListUsers list all users
func (s *Service) ListUsers(ctx context.Context, pageIndex, pageSize int, name, status, sort, orderBy string) ([]map[string]interface{}, error) {
	users, _, err := s.userDAO.List(ctx, dao.DB, (pageIndex-1)*pageSize, pageSize, name, status, sort, orderBy)
	if err != nil {
		return nil, err
	}

	result := make([]map[string]interface{}, 0, len(users))
	for _, user := range users {
		result = append(result, map[string]interface{}{
			"email":        user.Email,
			"nickname":     user.Nickname,
			"create_date":  user.CreateTime,
			"is_active":    user.IsActive,
			"is_superuser": user.IsSuperuser,
		})
	}
	return result, nil
}

// CreateUser create a new user
// Parameters:
//   - username: email address of the user
//   - password: encrypted password (base64 encoded RSA encrypted)
//   - role: user role ("user" or "admin")
//
// Returns:
//   - map[string]interface{}: user information without password
//   - error: error message
func (s *Service) CreateUser(ctx context.Context, username, password, role string) (map[string]interface{}, error) {
	emailRegex := regexp.MustCompile(`^[\w\._-]+@([\w_-]+\.)+[\w-]{2,}$`)
	if !emailRegex.MatchString(username) {
		return nil, fmt.Errorf("invalid email address: %s", username)
	}

	existUser, _ := s.userDAO.GetByEmail(ctx, dao.DB, username)
	if existUser != nil {
		return nil, fmt.Errorf("user '%s' already exists", username)
	}

	decryptedPassword, err := common.DecryptPassword(password)
	if err != nil {
		return nil, fmt.Errorf("failed to decrypt password: %w", err)
	}

	hashedPassword, err := common.GenerateWerkzeugPasswordHash(decryptedPassword)
	if err != nil {
		return nil, fmt.Errorf("failed to hash password: %w", err)
	}

	userID := common.GenerateToken()
	accessToken := common.GenerateToken()
	status := "1"
	loginChannel := "password"
	isSuperuser := role == "admin"

	user := &entity.User{
		ID:              userID,
		AccessToken:     &accessToken,
		Email:           username,
		Nickname:        "",
		Password:        &hashedPassword,
		Status:          &status,
		IsActive:        "1",
		IsAuthenticated: "1",
		IsAnonymous:     "0",
		LoginChannel:    &loginChannel,
		IsSuperuser:     &isSuperuser,
	}

	// Start transaction for creating user and related data
	tx := dao.DB.Begin()
	if tx.Error != nil {
		return nil, fmt.Errorf("failed to begin transaction: %w", tx.Error)
	}

	// Rollback helper function
	rollbackTx := func() {
		if rbErr := tx.Rollback(); rbErr.Error != nil {
			common.Error("failed to rollback transaction", rbErr.Error)
		}
	}

	// 1. Create user
	if err = tx.Create(user).Error; err != nil {
		rollbackTx()
		return nil, fmt.Errorf("failed to create user: %w", err)
	}

	// 2. Create tenant (tenant_id = user_id)
	// tenant name = nickname + "'s Kingdom" (same as Python)
	tenantName := user.Nickname + "'s Kingdom"

	// Get default model IDs from config
	cfg := server.GetConfig()
	chatModel := ""
	embeddingModel := ""
	asrModel := ""
	vlmModel := ""
	rerankModel := ""
	var ttsModel *string = nil
	var ocrModel *string = nil
	parserIDs := "naive:General,qa:Q&A,manual:Manual,table:Table,paper:Paper,book:Book,laws:Laws,presentation:Presentation,picture:Picture,one:One,audio:Audio,email:Email"

	if cfg != nil {
		chatModel = cfg.GetDefaultChatModel().Name
		embeddingModel = cfg.GetDefaultEmbeddingModel().Name
		asrModel = cfg.GetDefaultASRModel().Name
		vlmModel = cfg.GetDefaultVisionModel().Name
		rerankModel = cfg.GetDefaultRerankModel().Name
		ttsModelStr := cfg.GetDefaultTTSModel().Name
		ttsModel = &ttsModelStr
		ocrModelStr := cfg.GetDefaultOCRModel().Name
		ocrModel = &ocrModelStr
	}

	tenantStatus := "1"
	tenant := &entity.Tenant{
		ID:        userID,
		Name:      &tenantName,
		LLMID:     chatModel,
		EmbdID:    embeddingModel,
		ASRID:     asrModel,
		Img2TxtID: vlmModel,
		RerankID:  rerankModel,
		TTSID:     ttsModel,
		OCRID:     ocrModel,
		ParserIDs: parserIDs,
		Credit:    512,
		Status:    &tenantStatus,
	}
	if err = tx.Create(tenant).Error; err != nil {
		rollbackTx()
		return nil, fmt.Errorf("failed to create tenant: %w", err)
	}

	// 3. Create user-tenant relation
	userTenantStatus := "1"
	userTenant := &entity.UserTenant{
		ID:        common.GenerateToken(),
		UserID:    userID,
		TenantID:  userID,
		Role:      "owner",
		InvitedBy: userID,
		Status:    &userTenantStatus,
	}
	if err = tx.Create(userTenant).Error; err != nil {
		rollbackTx()
		return nil, fmt.Errorf("failed to create user-tenant relation: %w", err)
	}

	// 4. Create root file folder
	fileID := common.GenerateToken()
	fileLocation := ""
	file := &entity.File{
		ID:        fileID,
		ParentID:  fileID,
		TenantID:  userID,
		CreatedBy: userID,
		Name:      "/",
		Type:      "folder",
		Size:      0,
		Location:  &fileLocation,
	}
	if err = tx.Create(file).Error; err != nil {
		rollbackTx()
		return nil, fmt.Errorf("failed to create root file folder: %w", err)
	}

	// Commit transaction
	if err = tx.Commit().Error; err != nil {
		return nil, fmt.Errorf("failed to commit transaction: %w", err)
	}

	common.Info("Create user success with tenant and related data", zap.String("username", username))

	return map[string]interface{}{
		"id":           user.ID,
		"email":        user.Email,
		"nickname":     user.Nickname,
		"is_active":    user.IsActive,
		"is_superuser": isSuperuser,
		"create_date":  user.CreateDate,
	}, nil
}

// GetUserDetails get user details
func (s *Service) GetUserDetails(ctx context.Context, username string) (map[string]interface{}, error) {
	user, err := s.userDAO.GetByEmail(ctx, dao.DB, username)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrUserNotFound
		}
		return nil, fmt.Errorf("failed to get user: %w", err)
	}

	return map[string]interface{}{
		"avatar":          user.Avatar,
		"email":           user.Email,
		"language":        user.Language,
		"last_login_time": user.LastLoginTime,
		"is_active":       user.IsActive,
		"is_anonymous":    user.IsAnonymous,
		"login_channel":   user.LoginChannel,
		"status":          user.Status,
		"is_superuser":    user.IsSuperuser,
		"create_date":     user.CreateDate,
		"update_date":     user.UpdateDate,
	}, nil
}

// DeleteUserResult result of delete user operation
type DeleteUserResult struct {
	Username        string   `json:"username"`
	LangfuseCount   int      `json:"langfuse_count"`
	MetadataTable   string   `json:"metadata_table"`
	TenantCount     int      `json:"tenant_count"`
	UserTenantCount int      `json:"user_tenant_count"`
	UserCount       int      `json:"user_count"`
	DeletedDetails  []string `json:"deleted_details"`
}

// DeleteUser delete user with cascade delete of all related data
// Parameters:
//   - username: email address of the user to delete
//
// Returns:
//   - *DeleteUserResult
//   - error: error message
func (s *Service) DeleteUser(ctx context.Context, username string) (*DeleteUserResult, error) {
	userList, err := s.userDAO.ListByEmail(ctx, dao.DB, username)
	if err != nil {
		return nil, fmt.Errorf("failed to get user: %w", err)
	}
	if len(userList) == 0 {
		return nil, fmt.Errorf("user '%s' not found", username)
	}

	if len(userList) > 1 {
		return nil, fmt.Errorf("exist more than 1 user: %s", username)
	}

	user := userList[0]

	// Check if user is active - cannot delete active users
	if user.IsActive == "1" {
		return nil, fmt.Errorf("user '%s' is active and can't be deleted. Please deactivate the user first", username)
	}

	// Check if user is superuser - cannot delete admin accounts
	if user.IsSuperuser != nil && *user.IsSuperuser {
		return nil, fmt.Errorf("user '%s' is admin account and cannot be deleted", username)
	}

	return s.deleteUserData(ctx, user)
}

// ChangePassword change user password
// Parameters:
//   - username: email address of the user
//   - newPassword: new encrypted password (base64 encoded RSA encrypted)
//
// Returns:
//   - error: error message
func (s *Service) ChangePassword(ctx context.Context, username, newPassword string) error {
	userList, err := s.userDAO.ListByEmail(ctx, dao.DB, username)
	if err != nil || len(userList) == 0 {
		return fmt.Errorf("user '%s' not found", username)
	}

	if len(userList) > 1 {
		return fmt.Errorf("exist more than 1 user: %s", username)
	}

	user := userList[0]

	decryptedPassword, err := common.DecryptPassword(newPassword)
	if err != nil {
		return fmt.Errorf("failed to decrypt password: %w", err)
	}

	if user.Password != nil && common.CheckWerkzeugPassword(decryptedPassword, *user.Password) {
		return nil
	}

	hashedPassword, err := common.GenerateWerkzeugPasswordHash(decryptedPassword)
	if err != nil {
		return fmt.Errorf("failed to hash password: %w", err)
	}

	user.Password = &hashedPassword

	if err = s.userDAO.Update(ctx, dao.DB, user); err != nil {
		return fmt.Errorf("failed to update user: %w", err)
	}

	return nil
}

// UpdateUserActivateStatus update user activate status
// Parameters:
//   - username: email address of the user
//   - isActive: true to activate, false to deactivate
//
// Returns:
//   - error: error message
func (s *Service) UpdateUserActivateStatus(ctx context.Context, username string, isActive bool) error {
	userList, err := s.userDAO.ListByEmail(ctx, dao.DB, username)
	if err != nil || len(userList) == 0 {
		return fmt.Errorf("user '%s' not found", username)
	}

	if len(userList) > 1 {
		return fmt.Errorf("exist more than 1 user: %s", username)
	}

	user := userList[0]

	targetStatus := "0"
	if isActive {
		targetStatus = "1"
	}

	if user.IsActive == targetStatus {
		return nil
	}

	user.IsActive = targetStatus
	if !isActive {
		invalidToken := "INVALID_" + common.GenerateToken()
		user.AccessToken = &invalidToken
	}

	if err = s.userDAO.Update(ctx, dao.DB, user); err != nil {
		return fmt.Errorf("failed to update user: %w", err)
	}

	return nil
}

// GrantAdmin grant admin privileges
// Parameters:
//   - username: email address of the user
//
// Returns:
//   - error: error message
func (s *Service) GrantAdmin(ctx context.Context, username string) error {
	userList, err := s.userDAO.ListByEmail(ctx, dao.DB, username)
	if err != nil || len(userList) == 0 {
		return fmt.Errorf("user '%s' not found", username)
	}

	if len(userList) > 1 {
		return fmt.Errorf("exist more than 1 user: %s", username)
	}

	user := userList[0]

	if user.IsSuperuser != nil && *user.IsSuperuser {
		return nil
	}

	isSuperuser := true
	user.IsSuperuser = &isSuperuser

	if err = s.userDAO.Update(ctx, dao.DB, user); err != nil {
		return fmt.Errorf("failed to update user: %w", err)
	}

	return nil
}

// RevokeAdmin revoke admin privileges
// Parameters:
//   - username: email address of the user
//
// Returns:
//   - error: error message
func (s *Service) RevokeAdmin(ctx context.Context, username string) error {
	userList, err := s.userDAO.ListByEmail(ctx, dao.DB, username)
	if err != nil || len(userList) == 0 {
		return fmt.Errorf("user '%s' not found", username)
	}

	if len(userList) > 1 {
		return fmt.Errorf("exist more than 1 user: %s", username)
	}

	user := userList[0]

	if user.IsSuperuser == nil || !*user.IsSuperuser {
		return nil
	}

	isSuperuser := false
	user.IsSuperuser = &isSuperuser

	if err = s.userDAO.Update(ctx, dao.DB, user); err != nil {
		return fmt.Errorf("failed to update user: %w", err)
	}

	return nil
}

// API Key methods

// ListUserAPITokens get user API keys
func (s *Service) ListUserAPITokens(ctx context.Context, username string) ([]map[string]interface{}, error) {
	// 1. Get user details
	user, err := s.userDAO.GetByEmail(ctx, dao.DB, username)
	if err != nil {
		return nil, fmt.Errorf("user not found: %w", err)
	}

	// 2. Get user's tenants
	userTenants, err := s.userTenantDAO.GetByUserID(ctx, dao.DB, user.ID)
	if err != nil || len(userTenants) == 0 {
		return nil, fmt.Errorf("tenant not found")
	}

	tenantID := userTenants[0].TenantID

	// 3. Get API tokens by tenant ID
	tokens, err := s.apiTokenDAO.GetByTenantID(ctx, dao.DB, tenantID)
	if err != nil {
		return nil, fmt.Errorf("failed to get API tokens: %w", err)
	}

	// 4. Convert to map slice
	result := make([]map[string]interface{}, 0, len(tokens))
	for _, token := range tokens {
		result = append(result, map[string]interface{}{
			"tenant_id":   token.TenantID,
			"token":       token.Token,
			"beta":        token.Beta,
			"dialog_id":   token.DialogID,
			"source":      token.Source,
			"create_time": token.CreateTime,
			"create_date": token.CreateDate,
			"update_time": token.UpdateTime,
			"update_date": token.UpdateDate,
		})
	}

	return result, nil
}

// GenerateUserAPIToken generate API key for user
func (s *Service) GenerateUserAPIToken(ctx context.Context, username string) (map[string]interface{}, error) {
	// 1. Get user details
	user, err := s.userDAO.GetByEmail(ctx, dao.DB, username)
	if err != nil {
		return nil, fmt.Errorf("user not found: %w", err)
	}

	// 2. Get user's tenants
	userTenants, err := s.userTenantDAO.GetByUserID(ctx, dao.DB, user.ID)
	if err != nil || len(userTenants) == 0 {
		return nil, fmt.Errorf("tenant not found")
	}

	tenantID := userTenants[0].TenantID

	// 3. Generate API token
	key := common.GenerateAPIToken()
	beta := common.GenerateBetaAPIToken()

	apiToken := &entity.APIToken{
		TenantID: tenantID,
		Token:    key,
		Beta:     &beta,
	}

	// 4. Save API token
	if err = s.apiTokenDAO.Create(ctx, dao.DB, apiToken); err != nil {
		return nil, fmt.Errorf("failed to generate API key: %w", err)
	}

	return map[string]interface{}{
		"tenant_id":   tenantID,
		"token":       key,
		"beta":        beta,
		"create_time": apiToken.CreateTime,
		"create_date": apiToken.CreateDate,
		"update_time": apiToken.UpdateTime,
		"update_date": apiToken.UpdateDate,
	}, nil
}

// DeleteUserAPIToken delete user API key
func (s *Service) DeleteUserAPIToken(ctx context.Context, username, key string) error {
	// 1. Get user details
	user, err := s.userDAO.GetByEmail(ctx, dao.DB, username)
	if err != nil {
		return fmt.Errorf("user not found: %w", err)
	}

	// 2. Get user's tenants
	userTenants, err := s.userTenantDAO.GetByUserID(ctx, dao.DB, user.ID)
	if err != nil || len(userTenants) == 0 {
		return fmt.Errorf("tenant not found")
	}

	tenantID := userTenants[0].TenantID

	// 3. Delete API token
	rowsAffected, err := s.apiTokenDAO.DeleteByTenantIDAndToken(ctx, dao.DB, tenantID, key)
	if err != nil {
		return fmt.Errorf("failed to delete API key: %w", err)
	}

	if rowsAffected == 0 {
		return fmt.Errorf("API key not found or could not be deleted")
	}

	return nil
}

type ServiceStatus struct {
	Type    string `json:"type"`
	Name    string `json:"name"`
	Host    string `json:"host"`
	Port    int    `json:"port"`
	Status  string `json:"status"`
	Elapsed string `json:"elapsed"`
	Message string `json:"message"`
}

func newServiceStatus(typeStr, nameStr, host string, port int, statusStr string, startTime time.Time, messageStr string) ServiceStatus {
	return ServiceStatus{
		Type:    typeStr,
		Name:    nameStr,
		Host:    host,
		Port:    port,
		Status:  statusStr,
		Elapsed: fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
		Message: messageStr,
	}
}

func serviceEndpoint(raw string, defaultPort int) (string, int) {
	if parsed, err := url.Parse(raw); err == nil && parsed.Hostname() != "" {
		raw = parsed.Host
	}
	host, port, err := net.SplitHostPort(raw)
	if err == nil {
		var portNumber int
		fmt.Sscanf(port, "%d", &portNumber)
		return host, portNumber
	}
	return raw, defaultPort
}

// ListServices get all services
func (s *Service) ListServices(ctx context.Context) ([]ServiceStatus, error) {

	var results []ServiceStatus

	globalConfig := server.GetConfig()
	// Database
	databaseType := globalConfig.DatabaseType()
	switch databaseType {
	case "mysql":
		mysqlStatus := s.getMySQLStatus(ctx)
		results = append(results, mysqlStatus)
	default:
		mysqlConfig := globalConfig.GetMySQLConfig()
		results = append(results, newServiceStatus("database", databaseType, mysqlConfig.Host, mysqlConfig.Port, "not available", time.Now(), "not supported database type"))
	}

	// Doc engine
	docEngineImpl := engine.Get()
	docHost, docPort := "-", 0
	if docEngineImpl.GetType() == "infinity" {
		docHost, docPort = serviceEndpoint(globalConfig.GetInfinityConfig().URI, 0)
	} else if docEngineImpl.GetType() == "elasticsearch" {
		docHost, docPort = serviceEndpoint(globalConfig.GetElasticsearchConfig().Hosts, 0)
	}
	err := docEngineImpl.Ping(ctx)
	if err == nil {
		results = append(results, newServiceStatus("doc_engine", docEngineImpl.GetType(), docHost, docPort, "alive", time.Now(), ""))
	} else {
		results = append(results, newServiceStatus("doc_engine", docEngineImpl.GetType(), docHost, docPort, "timeout", time.Now(), err.Error()))
	}

	// storage engine
	storageImpl := storage.GetStorageFactory().GetStorage()
	storageHost, storagePort := "-", 0
	if storageImpl.Type() == "minio" {
		storageHost, storagePort = serviceEndpoint(globalConfig.GetMinioConfig().Host, 0)
	}
	storageHealth := storageImpl.Health(ctx)
	if storageHealth {
		results = append(results, newServiceStatus("storage_engine", storageImpl.Type(), storageHost, storagePort, "alive", time.Now(), ""))
	} else {
		results = append(results, newServiceStatus("storage_engine", storageImpl.Type(), storageHost, storagePort, "timeout", time.Now(), ""))
	}

	// cache engine
	cacheType := globalConfig.CacheEngineType()
	switch cacheType {
	// The Go stack talks to Kvrocks. "redis" is accepted for backwards
	// compatibility with the shared service_conf.yaml.template; both map to
	// the same Kvrocks backend, so probe the same connection.
	case "redis", "kvrocks":
		kvrocksStatus := s.getKvrocksStatus(ctx)
		results = append(results, kvrocksStatus)
	default:
		redisConfig := globalConfig.GetKvrocksConfig()
		results = append(results, newServiceStatus("cache", cacheType, redisConfig.Host, redisConfig.Port, "not available", time.Now(), "not supported cache type"))
	}

	// message queue
	messageQueueImpl := engine.GetMessageQueueEngine()
	messageQueueStatus := messageQueueImpl.CheckStatus()
	if messageQueueStatus == "CONNECTED" {
		messageQueueStatus = "alive"
	}
	natsConfig := globalConfig.GetNATSConfig()
	results = append(results, newServiceStatus("message_queue", messageQueueImpl.Type(), natsConfig.Host, natsConfig.Port, messageQueueStatus, time.Now(), ""))

	results = append(results, s.GetEEServicesStatus(ctx)...)

	serverList := GlobalServerStore.ListInfos()
	for _, serverStatus := range serverList {
		now := time.Now()
		serverItem := newServiceStatus(string(serverStatus.ServerType), serverStatus.ServerName, serverStatus.Host, serverStatus.Port, "", serverStatus.Timestamp, "")
		// the difference between now and serverStatus.Timestamp is less than 5 seconds, then the server is alive
		if now.Sub(serverStatus.Timestamp) < 45*time.Second {
			serverItem.Status = "alive"
		} else {
			serverItem.Status = "timeout"
		}
		results = append(results, serverItem)
	}

	return results, nil
}

// GetServicesByType get services by type
func (s *Service) GetServicesByType(serviceType string) ([]map[string]interface{}, error) {
	return nil, errors.New("get_services_by_type: not implemented")
}

// GetServiceDetails get service details
func (s *Service) GetServiceDetails(configDict map[string]interface{}) ([]ServiceStatus, error) {
	//serviceType, _ := configDict["service_type"].(string)
	//name, _ := configDict["name"].(string)
	//ctx := context.Background()
	//
	//// Call detail function based on service type
	//switch serviceType {
	//case "meta_data":
	//	return s.getMySQLStatus(ctx), nil
	//case "cache":
	//	return s.getKvrocksStatus(ctx), nil
	//case "message_queue":
	//	host := configDict["host"].(string)
	//	port := configDict["port"].(int)
	//	return s.checkNatsAlive(name, host, port)
	//case "retrieval":
	//	// Check the extra.retrieval_type to determine which retrieval service
	//	if extra, ok := configDict["extra"].(map[string]interface{}); ok {
	//		if retrievalType, ok := extra["retrieval_type"].(string); ok {
	//			if retrievalType == "infinity" {
	//				return s.getInfinityStatus(name)
	//			}
	//		}
	//	}
	//	return s.getESClusterStats(name)
	//case "ragflow_server":
	//	return s.checkRAGFlowServerAlive(name)
	//case "file_store":
	//	return s.checkMinioAlive(name)
	//case "olap":
	//	return s.checkOlapAlive(name)
	//case "tracing":
	//	return s.checkTracingAlive(name)
	//default:
	//	return nil, nil
	//}
	return nil, nil
}

// getMySQLStatus gets MySQL service status
func (s *Service) getMySQLStatus(ctx context.Context) ServiceStatus {

	serviceType := "database"
	name := "mysql"
	startTime := time.Now()

	sqlDB, err := dao.DB.DB()
	if err != nil {
		mysqlConfig := server.GetConfig().GetMySQLConfig()
		return newServiceStatus(serviceType, name, mysqlConfig.Host, mysqlConfig.Port, "not connected", startTime, err.Error())
	}

	// Execute SELECT 1 to check connectivity
	err = sqlDB.PingContext(ctx)
	if err != nil {
		mysqlConfig := server.GetConfig().GetMySQLConfig()
		return newServiceStatus(serviceType, name, mysqlConfig.Host, mysqlConfig.Port, "timeout", startTime, err.Error())
	}

	mysqlConfig := server.GetConfig().GetMySQLConfig()
	return newServiceStatus(serviceType, name, mysqlConfig.Host, mysqlConfig.Port, "alive", startTime, "")
}

// getKvrocksStatus gets the Kvrocks service status.
func (s *Service) getKvrocksStatus(ctx context.Context) ServiceStatus {

	serviceType := "cache"
	name := "kvrocks"

	startTime := time.Now()
	kvrocksConfig := server.GetConfig().GetKvrocksConfig()

	kvrocksClient := kvrocks.Get()
	if kvrocksClient.Health(ctx) {
		return newServiceStatus(serviceType, name, kvrocksConfig.Host, kvrocksConfig.Port, "alive", startTime, "")
	}

	return newServiceStatus(serviceType, name, kvrocksConfig.Host, kvrocksConfig.Port, "timeout", startTime, "Kvrocks health check failed")
}

// getESClusterStats gets Elasticsearch cluster stats
func (s *Service) getESClusterStats(ctx context.Context, serviceType string) map[string]interface{} {
	name := "elasticsearch"
	startTime := time.Now()

	cfg := server.GetConfig()
	if cfg == nil || !cfg.IsElasticConfigured() {
		return map[string]interface{}{
			"type":    serviceType,
			"name":    name,
			"status":  "timeout",
			"elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
			"message": "error: Elasticsearch configuration not found",
		}
	}

	// Create ES engine and get cluster stats
	esEngine, err := elasticsearch.NewEngine(ctx, cfg.GetElasticsearchConfig())
	if err != nil {
		return map[string]interface{}{
			"type":    serviceType,
			"name":    name,
			"status":  "timeout",
			"elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
			"message": fmt.Sprintf("error: %s", err.Error()),
		}
	}
	defer esEngine.Close()

	clusterStats, err := esEngine.GetClusterStats(ctx)
	if err != nil {
		return map[string]interface{}{
			"type":    serviceType,
			"name":    name,
			"status":  "timeout",
			"elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
			"message": fmt.Sprintf("error: %s", err.Error()),
		}
	}

	return map[string]interface{}{
		"type":    serviceType,
		"name":    name,
		"status":  "alive",
		"elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
		"message": clusterStats,
	}
}

// getInfinityStatus gets Infinity service status
func (s *Service) getInfinityStatus(serviceType string) map[string]interface{} {
	name := "infinity"
	startTime := time.Now()
	// TODO: Implement actual Infinity health check
	return map[string]interface{}{
		"type":    serviceType,
		"name":    name,
		"status":  "unknown",
		"elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()),
		"message": "Infinity health check not implemented",
	}
}

// checkTaskExecutorAlive checks if task executor is alive
func (s *Service) checkTaskExecutorAlive(name string) (map[string]interface{}, error) {
	// TODO: Implement actual task executor health check
	return map[string]interface{}{
		"service_name": name,
		"status":       "unknown",
		"message":      "Task executor health check not implemented",
	}, nil
}

// checkNatsAlive checks if NATS is alive
func (s *Service) checkNatsAlive(name string, ip string, port int) (map[string]interface{}, error) {

	msgQueueEngine := engine.GetMessageQueueEngine()
	if msgQueueEngine == nil {
		return map[string]interface{}{
			"service_name": name,
			"status":       "Message queue engine not initialized",
		}, nil
	}

	status := msgQueueEngine.CheckStatus()

	return map[string]interface{}{
		"service_name": name,
		"status":       status,
	}, nil
}

// checkTracingAlive checks if tracing is alive
func (s *Service) checkTracingAlive(name string) (map[string]interface{}, error) {
	return map[string]interface{}{
		"service_name": name,
		"status":       "unknown",
		"message":      "Tracing health check not implemented",
	}, nil
}

// checkOlapAlive checks if ClickHouse is alive
func (s *Service) checkOlapAlive(name string) (map[string]interface{}, error) {
	clickhouseDriver := clickhouse.GetDriver()
	if clickhouseDriver == nil {
		return map[string]interface{}{
			"service_name": name,
			"status":       "unknown",
		}, nil
	}

	status, err := clickhouseDriver.Status()
	if err != nil {
		return map[string]interface{}{
			"service_name": name,
			"status":       "timeout",
			"message":      fmt.Sprintf("error: %s", err.Error()),
		}, nil
	}

	return map[string]interface{}{
		"service_name": name,
		"status":       "alive",
		"message":      status,
	}, nil
}

// ShutdownService shutdown service
func (s *Service) ShutdownService(serviceName string) (map[string]interface{}, error) {
	// TODO: Implement with proper service manager
	return map[string]interface{}{
		"command":      "shutdown service",
		"service_name": serviceName,
		"error":        "shutdown service not implemented",
	}, nil
}

// StartService start service
func (s *Service) StartService(serviceName string) (map[string]interface{}, error) {
	return map[string]interface{}{
		"command":      "start service",
		"service_name": serviceName,
		"error":        "command 'start service' isn't implemented",
	}, nil
}

// RestartService restart service
func (s *Service) RestartService(serviceName string) (map[string]interface{}, error) {
	return map[string]interface{}{
		"command":      "restart service",
		"service_name": serviceName,
		"error":        "command 'restart service' isn't implemented",
	}, nil
}

// AdminException admin exception error
type AdminException struct {
	Message string
	Code    int
}

// Error implement error interface
func (e *AdminException) Error() string {
	return e.Message
}

// NewAdminException create admin exception
func NewAdminException(message string) *AdminException {
	return &AdminException{
		Message: message,
		Code:    400,
	}
}

// GetVariable get variable by name
// Returns the exact system setting with the given name, or settings matching the
// given name prefix when an exact setting does not exist.
func (s *Service) GetVariable(ctx context.Context, varName string) ([]map[string]interface{}, error) {
	settings, err := s.systemSettingsDAO.GetByName(ctx, dao.DB, varName)
	if err != nil {
		return nil, err
	}

	if len(settings) == 0 {
		settings, err = s.systemSettingsDAO.GetByNamePrefix(ctx, dao.DB, varName)
		if err != nil {
			return nil, err
		}
		if len(settings) == 0 {
			return nil, NewAdminException("Can't get setting: " + varName)
		}
	}
	return entity.FormatSystemSettings(settings), nil
}

// ListAllVariables list all variables
// Returns all system settings from database
func (s *Service) ListAllVariables(ctx context.Context) ([]map[string]interface{}, error) {
	settings, err := s.systemSettingsDAO.GetAll(ctx, dao.DB)
	if err != nil {
		return nil, err
	}

	return entity.FormatSystemSettings(settings), nil
}

// SetVariable set variable
// Creates or updates a system setting
// If the setting exists, updates it; otherwise creates a new one
func (s *Service) SetVariable(ctx context.Context, varName, varValue string) error {
	return s.setVariable(ctx, dao.DB, varName, varValue)
}

func (s *Service) setVariable(ctx context.Context, db *gorm.DB, varName, varValue string) error {
	settings, err := s.systemSettingsDAO.GetByName(ctx, db, varName)
	if err != nil {
		return err
	}

	if len(settings) == 1 {
		setting := &settings[0]
		if err = entity.ValidateSystemSettingValue(*setting, varValue); err != nil {
			return err
		}
		setting.Value = varValue
		return s.systemSettingsDAO.UpdateByName(ctx, db, varName, setting)
	} else if len(settings) > 1 {
		return NewAdminException("Can't update more than 1 setting: " + varName)
	}

	dataType := entity.InferSystemSettingDataType(varName)
	newSetting := &entity.SystemSettings{
		Name:     varName,
		Value:    varValue,
		Source:   "admin",
		DataType: dataType,
	}
	if err = entity.ValidateSystemSettingValue(*newSetting, varValue); err != nil {
		return err
	}
	return s.systemSettingsDAO.Create(ctx, db, newSetting)
}

// Config methods

// ListAllConfigs list all configs
// Returns all service configurations from the config file
func (s *Service) ListAllConfigs() ([]map[string]interface{}, error) {
	result, err := server.GetAllConfigs()
	if err != nil {
		return nil, err
	}
	return result, nil
}

// Environment methods

// ListEnvironments list all environments
func (s *Service) ListEnvironments() ([]map[string]interface{}, error) {
	result := make([]map[string]interface{}, 0)

	globalConfig := server.GetConfig()

	// DOC_ENGINE
	docEngine := globalConfig.GetEnvDocumentEngineType()
	if docEngine == "" {
		docEngine = "elasticsearch"
	}
	result = append(result, map[string]interface{}{
		"env":   "DOCUMENT_ENGINE",
		"value": docEngine,
	})

	// DEFAULT_SUPERUSER_EMAIL
	defaultSuperuserEmail := globalConfig.GetEnvDefaultSuperUserEmail()
	if defaultSuperuserEmail == "" {
		defaultSuperuserEmail = "[email protected]"
	}
	result = append(result, map[string]interface{}{
		"env":   "DEFAULT_SUPERUSER_EMAIL",
		"value": defaultSuperuserEmail,
	})

	// DB_TYPE
	dbType := globalConfig.GetEnvDatabaseType()
	if dbType == "" {
		dbType = "mysql"
	}
	result = append(result, map[string]interface{}{
		"env":   "DATABASE",
		"value": dbType,
	})

	// DEVICE
	device := globalConfig.GetEnvDeviceType()
	if device == "" {
		device = "cpu"
	}
	result = append(result, map[string]interface{}{
		"env":   "DEVICE",
		"value": device,
	})

	// STORAGE_IMPL
	storageImpl := globalConfig.GetEnvStorageType()
	if storageImpl == "" {
		storageImpl = "minio"
	}
	result = append(result, map[string]interface{}{
		"env":   "STORAGE",
		"value": storageImpl,
	})

	return result, nil
}

// Version methods

// GetVersion get RAGFlow version
func (s *Service) GetVersion() (string, string) {
	return common.GetRAGFlowVersion(), common.GetRAGFlowType()
}

// Sandbox methods

// ListSandboxProviders list sandbox providers
func (s *Service) ListSandboxProviders() ([]map[string]interface{}, error) {
	return sandbox.ListProviders(), nil
}

// GetSandboxProviderSchema get sandbox provider schema
func (s *Service) GetSandboxProviderSchema(providerID string) (map[string]interface{}, error) {
	schema, err := sandbox.ConfigSchema(providerID)
	if err != nil {
		return nil, common.NewCodedError(common.CodeBadRequest, err.Error())
	}
	return schema, nil
}

// GetSandboxConfig get sandbox config
func (s *Service) GetSandboxConfig(ctx context.Context) (map[string]interface{}, error) {
	settings, err := s.systemSettingsDAO.GetByNamePrefix(ctx, dao.DB, "sandbox.")
	if err != nil {
		return nil, err
	}

	providerType := "self_managed"
	values := make(map[string]string, len(settings))
	for _, setting := range settings {
		if _, exists := values[setting.Name]; exists {
			return nil, fmt.Errorf("duplicate sandbox setting: %s", setting.Name)
		}
		values[setting.Name] = setting.Value
	}
	if value, exists := values["sandbox.provider_type"]; exists {
		providerType = value
	}
	schema, err := sandbox.ConfigSchema(providerType)
	if err != nil {
		return nil, common.NewCodedError(common.CodeBadRequest, err.Error())
	}

	raw := values["sandbox."+providerType]
	var config map[string]interface{}
	if raw != "" {
		var value interface{}
		if err := json.Unmarshal([]byte(raw), &value); err != nil {
			config = nil
		} else if value == nil {
			return nil, common.NewCodedError(common.CodeBadRequest,
				fmt.Sprintf("sandbox.%s must contain a JSON object", providerType))
		} else if object, ok := value.(map[string]interface{}); ok {
			config = object
		} else {
			return nil, common.NewCodedError(common.CodeBadRequest,
				fmt.Sprintf("sandbox.%s must contain a JSON object", providerType))
		}
	}
	if len(config) == 0 {
		config = make(map[string]interface{})
		for name, field := range schema {
			fieldSchema, ok := field.(map[string]any)
			if !ok {
				continue
			}
			if fieldSchema["readonly"] == true {
				continue
			}
			if value, ok := fieldSchema["default"]; ok && value != nil {
				config[name] = value
			}
		}
	}
	return map[string]interface{}{"provider_type": providerType, "config": config}, nil
}

// SetSandboxConfig set sandbox config
func (s *Service) SetSandboxConfig(ctx context.Context, providerType string, config map[string]interface{}, setActive bool) (map[string]interface{}, error) {
	if config == nil {
		return nil, common.NewCodedError(common.CodeBadRequest, "config must be a JSON object")
	}
	if err := sandbox.ValidateConfig(providerType, config); err != nil {
		return nil, common.NewCodedError(common.CodeBadRequest, err.Error())
	}
	encoded, err := json.Marshal(config)
	if err != nil {
		return nil, common.NewCodedError(common.CodeBadRequest, fmt.Sprintf("invalid sandbox configuration: %v", err))
	}
	err = s.systemSettingsDAO.Transaction(ctx, dao.DB, func(tx *gorm.DB) error {
		if setActive {
			if err := s.setVariable(ctx, tx, "sandbox.provider_type", providerType); err != nil {
				return err
			}
		}
		return s.setVariable(ctx, tx, "sandbox."+providerType, string(encoded))
	})
	if err != nil {
		return nil, err
	}
	return map[string]interface{}{"provider_type": providerType, "config": config}, nil
}

// TestSandboxConnection test sandbox connection
func (s *Service) TestSandboxConnection(ctx context.Context, providerType string, config map[string]interface{}) (map[string]interface{}, error) {
	result, err := sandbox.TestConnection(ctx, providerType, config)
	if err != nil {
		return nil, common.NewCodedError(common.CodeBadRequest, err.Error())
	}
	return result, nil
}

var heartBeatCount int64 = 0

// HandleHeartbeat handle heartbeat
func (s *Service) HandleHeartbeat(message *common.BaseMessage) (common.ErrorCode, map[string]interface{}, string) {
	heartBeatCount++

	status := &common.BaseMessage{
		ServerName: message.ServerName,
		ServerType: message.ServerType,
		Host:       message.Host,
		Port:       message.Port,
		Version:    message.Version,
		Timestamp:  message.Timestamp,
		Ext:        message.Ext,
	}
	return UpdateServer(message.ServerName, status)
}

// InitDefaultAdmin initialize default admin user
// This matches Python's init_default_admin behavior
func (s *Service) InitDefaultAdmin() error {
	// Default superuser settings (matching Python's DEFAULT_SUPERUSER_* defaults)
	defaultNickname := "admin"
	defaultEmail := "[email protected]"
	defaultPassword := "admin"

	// Query superusers
	var users []*entity.User
	err := dao.DB.Where("is_superuser = ? AND status = ?", true, "1").Find(&users).Error
	if err != nil {
		return fmt.Errorf("failed to query superusers: %w", err)
	}

	if len(users) == 0 {
		userID := common.GenerateToken()
		accessToken := common.GenerateToken()
		status := "1"
		loginChannel := "password"
		isSuperuser := true

		// Python: password = encode_to_base64(password) = base64.b64encode(password)
		// Then: generate_password_hash(base64_password) creates werkzeug hash
		password := base64.StdEncoding.EncodeToString([]byte(defaultPassword))
		hashedPassword, err := common.GenerateWerkzeugPasswordHash(password)
		if err != nil {
			return fmt.Errorf("failed to hash password: %w", err)
		}

		user := &entity.User{
			ID:              userID,
			Email:           defaultEmail,
			Nickname:        defaultNickname,
			Password:        &hashedPassword,
			AccessToken:     &accessToken,
			Status:          &status,
			IsActive:        "1",
			IsAuthenticated: "1",
			IsAnonymous:     "0",
			LoginChannel:    &loginChannel,
			IsSuperuser:     &isSuperuser,
		}

		if err = dao.DB.Create(user).Error; err != nil {
			return fmt.Errorf("can't init admin: %w", err)
		}

		if err = s.addTenantForAdmin(userID, defaultNickname); err != nil {
			return fmt.Errorf("failed to add tenant for admin: %w", err)
		}

		common.Info("Init default super user successfully")
		return nil
	}

	for _, user := range users {
		if user.IsActive != "1" {
			return fmt.Errorf("no active admin. Please update 'is_active' in db manually")
		}
	}

	for _, user := range users {
		if user.Email == defaultEmail {
			// Check if tenant exists
			var count int64
			dao.DB.Model(&entity.UserTenant{}).Where("user_id = ? AND status = ?", user.ID, "1").Count(&count)
			if count == 0 {
				nickname := defaultNickname
				if user.Nickname != "" {
					nickname = user.Nickname
				}
				if err = s.addTenantForAdmin(user.ID, nickname); err != nil {
					return err
				}
			}
			break
		}
	}

	common.Info("Init default super user successfully")
	return nil
}

// addTenantForAdmin add tenant for admin user
func (s *Service) addTenantForAdmin(userID, nickname string) error {
	status := "1"
	role := "owner"
	tenantName := nickname + "'s Kingdom"

	tenant := &entity.Tenant{
		ID:   userID,
		Name: &tenantName,
	}

	if err := dao.DB.Create(tenant).Error; err != nil {
		return err
	}

	userTenant := &entity.UserTenant{
		TenantID:  userID,
		UserID:    userID,
		InvitedBy: userID,
		Role:      role,
		Status:    &status,
	}

	return dao.DB.Create(userTenant).Error
}

// ListAllModels list all models
func (s *Service) ListAllModels(pageIndex, pageSize int) ([]map[string]interface{}, error) {
	models, err := dao.GetModelProviderManager().ListAllModels()
	if err != nil {
		return nil, err
	}
	if pageSize > 0 && pageIndex >= 0 && pageIndex*pageSize < len(models) {
		return models[pageIndex*pageSize : (pageIndex+1)*pageSize], nil
	}
	return models, nil
}

func (s *Service) GetModelByModelName(modelName string) (*modelModule.Model, error) {
	return dao.GetModelProviderManager().GetModelByNameOrAlias(modelName), nil
}