Go: fix 'list services' (#16598)

### Summary

```
RAGFlow(admin)> list services;
+-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+
| extra                                                                       | host      | id | name          | port | service_type  | status    |
+-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+
| map[database:1 mq_type:redis password:infini_rag_flow]                      | localhost | 0  | redis         | 6379 | message_queue | alive     |
| map[password:infini_rag_flow retrieval_type:elasticsearch username:elastic] | localhost | 1  | elasticsearch | 1200 | retrieval     | alive     |
|                                                                             | 0.0.0.0   | 2  | nats          | 4222 | message_queue | CONNECTED |
| map[meta_type:mysql password:infini_rag_flow username:root]                 | localhost | 3  | mysql         | 3306 | meta_data     | alive     |
| map[password:infini_rag_flow store_type:minio user:rag_flow]                | localhost | 4  | minio         | 9000 | file_store    | alive     |
+-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+

```

Signed-off-by: Jin Hai <haijin.chn@gmail.com>

internal/admin/service.go

@@ -976,7 +976,7 @@ func (s *Service) GenerateUserAPIToken(username string) (map[string]interface{},
 	}
 
 	// 4. Save API token
-	if err := s.apiTokenDAO.Create(apiToken); err != nil {
+	if err = s.apiTokenDAO.Create(apiToken); err != nil {
 		return nil, fmt.Errorf("failed to generate API key: %w", err)
 	}
 
@@ -1024,7 +1024,7 @@ func (s *Service) DeleteUserAPIToken(username, key string) error {
 func (s *Service) ListServices() ([]map[string]interface{}, error) {
 	allConfigs := server.GetAllConfigs()
 
-	var result []map[string]interface{}
+	var results []map[string]interface{}
 	for _, configDict := range allConfigs {
 		serviceType := configDict["service_type"]
 		if serviceType != "ragflow_server" {
@@ -1039,19 +1039,18 @@ func (s *Service) ListServices() ([]map[string]interface{}, error) {
 			} else {
 				configDict["status"] = "timeout"
 			}
-			result = append(result, configDict)
+			if serviceDetail != nil {
+				results = append(results, configDict)
+			}
 		}
 	}
 
-	id := len(result)
 	serverList := GlobalServerStore.ListInfos()
 	now := time.Now()
 	for _, serverStatus := range serverList {
 		serverItem := make(map[string]interface{})
 		serverItem["name"] = serverStatus.ServerName
 		serverItem["service_type"] = serverStatus.ServerType
-		serverItem["id"] = id
-		id++
 		serverItem["host"] = serverStatus.Host
 		serverItem["port"] = serverStatus.Port
 		// the difference between now and serverStatus.Timestamp is less than 5 seconds, then the server is alive
@@ -1060,9 +1059,14 @@ func (s *Service) ListServices() ([]map[string]interface{}, error) {
 		} else {
 			serverItem["status"] = "timeout"
 		}
-		result = append(result, serverItem)
+		results = append(results, serverItem)
 	}
-	return result, nil
+
+	for id, result := range results {
+		result["id"] = id
+	}
+
+	return results, nil
 }
 
 // GetServicesByType get services by type
@@ -1080,7 +1084,20 @@ func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[stri
 	case "meta_data":
 		return s.getMySQLStatus(name)
 	case "message_queue":
-		return s.getRedisInfo(name)
+		switch name {
+		case "redis":
+			return s.getRedisInfo(name)
+		case "nats":
+			host := configDict["host"].(string)
+			port := configDict["port"].(int)
+			return s.checkNatsAlive(name, host, port)
+		default:
+			return map[string]interface{}{
+				"service_name": name,
+				"status":       "unknown",
+				"message":      "Service type not supported",
+			}, nil
+		}
 	case "retrieval":
 		// Check the extra.retrieval_type to determine which retrieval service
 		if extra, ok := configDict["extra"].(map[string]interface{}); ok {
@@ -1095,14 +1112,8 @@ func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[stri
 		return s.checkRAGFlowServerAlive(name)
 	case "file_store":
 		return s.checkMinioAlive(name)
-	case "task_executor":
-		return s.checkTaskExecutorAlive(name)
 	default:
-		return map[string]interface{}{
-			"service_name": name,
-			"status":       "unknown",
-			"message":      "Service type not supported",
-		}, nil
+		return nil, nil
 	}
 }
 
@@ -1408,6 +1419,18 @@ func (s *Service) checkTaskExecutorAlive(name string) (map[string]interface{}, e
 	}, nil
 }
 
+// checkNatsAlive checks if NATS is alive
+func (s *Service) checkNatsAlive(name string, ip string, port int) (map[string]interface{}, error) {
+
+	msgQueueEngine := engine.GetMessageQueueEngine()
+	status := msgQueueEngine.CheckStatus()
+
+	return map[string]interface{}{
+		"service_name": name,
+		"status":       status,
+	}, nil
+}
+
 // ShutdownService shutdown service
 func (s *Service) ShutdownService(serviceID string) (map[string]interface{}, error) {
 	// TODO: Implement with proper service manager

internal/admin/state.go

@@ -53,6 +53,11 @@ func (s *ServerStore) UpdateServerInfo(serverName string, status *common.BaseMes
 		defer s.mu.Unlock()
 		s.servers[serverName] = status
 		return
+	case common.ServerTypeFileSyncer:
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.servers[serverName] = status
+		return
 	}
 }
 

internal/engine/engine.go

@@ -99,4 +99,5 @@ type MessageQueue interface {
 	GetMessages(messageCount int) ([]common.TaskHandle, error)
 	ListMessages(messageType string, pending bool) ([]map[string]string, error)
 	ShowMessageQueue() (map[string]string, error)
+	CheckStatus() string
 }

internal/engine/nats/nats.go

@@ -222,6 +222,11 @@ func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error)
 	return resultMessages, nil
 }
 
+func (n *NatsEngine) CheckStatus() string {
+	n.nc.Stats()
+	return n.nc.Status().String()
+}
+
 type NatsMessageHandle struct {
 	message jetstream.Msg
 }

internal/server/config.go

@@ -417,6 +417,7 @@ func Init(configPath string) error {
 			configDict["name"] = "nats"
 			configDict["host"] = host
 			configDict["port"] = port
+			configDict["service_type"] = "message_queue"
 		case "admin":
 			// Skip admin section
 			continue