Go: fix nats and go command (#16828)

### Summary
1. update docker compose file to start NATS healthy
2. Add two commands
```
RAGFlow(admin)> live;
SUCCESS
RAGFlow(admin)> health;
+---------------+-------+
| field         | value |
+---------------+-------+
| storage       | ok    |
| message_queue | ok    |
| status        | ok    |
| db            | ok    |
| redis         | ok    |
| doc_engine    | ok    |
+---------------+-------+
```

---------

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

docker/docker-compose-base.yml

@@ -255,7 +255,7 @@ services:
       - ragflow-go
     image: nats:2.14.2
     ports:
-      - ${NATS_PORT}:4222
+      - ${NATS_PORT:-4222}:4222
       - "8222:8222"
     volumes:
       - nats_data:/data
@@ -265,10 +265,10 @@ services:
       - ragflow
     restart: unless-stopped
     healthcheck:
-      test: ["CMD", "nc", "-z", "localhost", "${NATS_PORT}"]
+      test: ["CMD", "nats-server", "--health-check"]
       interval: 10s
       timeout: 10s
-      retries: 120
+      retries: 30
 
   tei-cpu:
     profiles:

internal/admin/handler.go

@@ -56,9 +56,25 @@ type ErrorResponse struct {
 	Message string `json:"message"`
 }
 
-// Health check
-func (h *Handler) Health(c *gin.Context) {
-	c.JSON(200, gin.H{"status": "ok"})
+// Healthz to get system health
+func (h *Handler) Healthz(c *gin.Context) {
+	result, allOK := service.GetComponentsHealthz(c.Request.Context())
+	if allOK {
+		c.JSON(http.StatusOK, gin.H{
+			"code": 0,
+			"data": result,
+		})
+	} else {
+		c.JSON(http.StatusInternalServerError, gin.H{
+			"code": common.CodeServerError,
+			"data": result,
+		})
+	}
+}
+
+// Live endpoint
+func (h *Handler) Live(c *gin.Context) {
+	common.SuccessNoData(c, "")
 }
 
 // Ping ping endpoint

internal/admin/router.go

@@ -34,8 +34,10 @@ func NewRouter(handler *Handler) *Router {
 
 // Setup setup routes
 func (r *Router) Setup(engine *gin.Engine) {
-	// Health check
-	engine.GET("/health", r.handler.Health)
+	// Healthz to get system health
+	engine.GET("/healthz", r.handler.Healthz)
+	engine.GET("/", r.handler.Live)
+	engine.GET("/live", r.handler.Live)
 
 	// Admin API routes with prefix /api/v1/admin
 	admin := engine.Group("/api/v1/admin")

internal/cli/admin_command.go

@@ -87,6 +87,24 @@ func (c *CLI) AdminPingCacheCommand(cmd *Command) (ResponseIf, error) {
 	return HandleSimpleResponse(resp, fmt.Sprintf("ping cache"))
 }
 
+// AdminLiveServerCommand GET /live
+func (c *CLI) AdminLiveServerCommand(cmd *Command) (ResponseIf, error) {
+	resp, err := c.AdminServerClient.Request("GET", "/live", "none", nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to show live server: %w", err)
+	}
+	return HandleSimpleResponse(resp, fmt.Sprintf("show live server"))
+}
+
+// AdminHealthServerCommand GET /healthz
+func (c *CLI) AdminHealthServerCommand(cmd *Command) (ResponseIf, error) {
+	resp, err := c.AdminServerClient.Request("GET", "/healthz", "none", nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to show health server: %w", err)
+	}
+	return HandleCommonDataResponse(resp, fmt.Sprintf("show health server"))
+}
+
 // AdminShowVersionCommand show RAGFlow admin version
 func (c *CLI) AdminShowVersionCommand(cmd *Command) (ResponseIf, error) {
 	resp, err := c.AdminServerClient.Request("GET", "/admin/version", "web", nil, nil)

internal/cli/admin_parser.go

@@ -97,6 +97,18 @@ func (p *Parser) parseAdminPingServer() (*Command, error) {
 	return cmd, nil
 }
 
+func (p *Parser) parseAdminLiveServer() (*Command, error) {
+	p.nextToken() // consume LIVE
+	cmd := NewCommand("admin_live_server")
+	return cmd, nil
+}
+
+func (p *Parser) parseAdminHealthServer() (*Command, error) {
+	p.nextToken() // consume HEALTH
+	cmd := NewCommand("admin_health_server")
+	return cmd, nil
+}
+
 // endregion
 
 // region LIST commands

internal/cli/cli_http.go

@@ -51,6 +51,10 @@ func (c *CLI) ExecuteAdminCommand(cmd *Command) (ResponseIf, error) {
 		return c.AdminPingCacheCommand(cmd)
 	case "admin_ping_server":
 		return c.PingServerByCommand(cmd)
+	case "admin_live_server":
+		return c.AdminLiveServerCommand(cmd)
+	case "admin_health_server":
+		return c.AdminHealthServerCommand(cmd)
 	case "benchmark":
 		return c.RunBenchmark(cmd)
 	case "admin_list_services":

internal/cli/http_client.go

@@ -67,12 +67,19 @@ func (c *HTTPClient) APIBase() string {
 
 // NonAPIBase returns the non-API base URL
 func (c *HTTPClient) NonAPIBase() string {
-	return fmt.Sprintf("%s:%d/%s", c.Host, c.Port, c.APIVersion)
+	return fmt.Sprintf("%s:%d", c.Host, c.Port)
 }
 
 // BuildURL builds the full URL for a given path
-func (c *HTTPClient) BuildURL(path string) string {
-	base := c.APIBase()
+func (c *HTTPClient) BuildURL(path string, authKind string) string {
+	var base string
+
+	if authKind == "none" {
+		base = c.NonAPIBase()
+	} else {
+		base = c.APIBase()
+	}
+
 	if c.VerifySSL {
 		return fmt.Sprintf("https://%s%s", base, path)
 	}
@@ -126,7 +133,7 @@ func (c *HTTPClient) Request(method, path string, authKind string, headers map[s
 		return nil, fmt.Errorf("HTTP Client is nil")
 	}
 
-	url := c.BuildURL(path)
+	url := c.BuildURL(path, authKind)
 	mergedHeaders := c.Headers(authKind, headers)
 
 	var body io.Reader
@@ -196,7 +203,7 @@ func (c *HTTPClient) RequestWithIterations(method, path string, authKind string,
 		return response, nil
 	}
 
-	url := c.BuildURL(path)
+	url := c.BuildURL(path, authKind)
 	mergedHeaders := c.Headers(authKind, headers)
 
 	var body io.Reader
@@ -278,7 +285,7 @@ func (c *HTTPClient) RequestJSON(method, path string, authKind string, headers m
 
 // UploadMultipart uploads data using multipart/form-data
 func (c *HTTPClient) UploadMultipart(path string, contentType string, body io.Reader) error {
-	url := c.BuildURL(path)
+	url := c.BuildURL(path, "api")
 
 	req, err := http.NewRequest("POST", url, body)
 	if err != nil {
@@ -322,7 +329,7 @@ func (c *HTTPClient) UploadMultipart(path string, contentType string, body io.Re
 
 // RequestStream makes an HTTP request for SSE streaming and returns the response body reader
 func (c *HTTPClient) RequestStream(method, path string, authKind string, headers map[string]string, jsonBody map[string]interface{}) (io.ReadCloser, error) {
-	url := c.BuildURL(path)
+	url := c.BuildURL(path, authKind)
 	mergedHeaders := c.Headers(authKind, headers)
 
 	var body io.Reader

internal/cli/lexer.go

@@ -227,6 +227,10 @@ func (l *Lexer) lookupIdent(ident string) Token {
 		return Token{Type: TokenAPI, Value: ident}
 	case "ADD":
 		return Token{Type: TokenAdd, Value: ident}
+	case "LIVE":
+		return Token{Type: TokenLive, Value: ident}
+	case "HEALTH":
+		return Token{Type: TokenHealth, Value: ident}
 	case "HOST":
 		return Token{Type: TokenHost, Value: ident}
 	case "DELETE":

internal/cli/parser.go

@@ -86,6 +86,10 @@ func (p *Parser) parseAdminCommand() (*Command, error) {
 		return p.parseAdminLogout()
 	case TokenPing:
 		return p.parseAdminPingServer()
+	case TokenLive:
+		return p.parseAdminLiveServer()
+	case TokenHealth:
+		return p.parseAdminHealthServer()
 	case TokenList:
 		return p.parseAdminListCommands()
 	case TokenShow:

internal/cli/types.go

@@ -45,6 +45,8 @@ const (
 	TokenServer
 	TokenAPI
 	TokenAdd
+	TokenLive
+	TokenHealth
 	TokenHost
 	TokenDelete
 	TokenPassword

internal/service/system.go

@@ -75,12 +75,13 @@ type HealthzMeta struct {
 }
 
 type HealthzResponse struct {
-	DB        string                 `json:"db"`
-	Redis     string                 `json:"redis"`
-	DocEngine string                 `json:"doc_engine"`
-	Storage   string                 `json:"storage"`
-	Status    string                 `json:"status"`
-	Meta      map[string]HealthzMeta `json:"_meta,omitempty"`
+	DB           string                 `json:"db"`
+	Redis        string                 `json:"redis"`
+	DocEngine    string                 `json:"doc_engine"`
+	Storage      string                 `json:"storage"`
+	MessageQueue string                 `json:"message_queue"`
+	Status       string                 `json:"status"`
+	Meta         map[string]HealthzMeta `json:"_meta,omitempty"`
 }
 
 // GetVersion get RAGFlow version
@@ -312,8 +313,7 @@ func timedHealthCheck(check func() error) (bool, HealthzMeta) {
 	return true, meta
 }
 
-// Healthz runs lightweight dependency checks for /api/v1/system/healthz.
-func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) {
+func GetComponentsHealthz(ctx context.Context) (*HealthzResponse, bool) {
 	meta := map[string]HealthzMeta{}
 
 	dbOK, dbMeta := timedHealthCheck(func() error {
@@ -363,13 +363,32 @@ func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) {
 		meta["storage"] = storageMeta
 	}
 
-	allOK := dbOK && redisOK && docOK && storageOK
+	messageQueueOK, messageQueueMeta := timedHealthCheck(func() error {
+
+		msgQueueEngine := engine.GetMessageQueueEngine()
+		if msgQueueEngine == nil {
+			return fmt.Errorf("message queue is not initialized")
+		}
+
+		status := msgQueueEngine.CheckStatus()
+
+		if msgQueueEngine == nil || status != "CONNECTED" {
+			return fmt.Errorf("message queue is not healthy")
+		}
+		return nil
+	})
+	if !messageQueueOK {
+		meta["message_queue"] = messageQueueMeta
+	}
+
+	allOK := dbOK && redisOK && docOK && storageOK && messageQueueOK
 	result := &HealthzResponse{
-		DB:        okNok(dbOK),
-		Redis:     okNok(redisOK),
-		DocEngine: okNok(docOK),
-		Storage:   okNok(storageOK),
-		Status:    okNok(allOK),
+		DB:           okNok(dbOK),
+		Redis:        okNok(redisOK),
+		DocEngine:    okNok(docOK),
+		Storage:      okNok(storageOK),
+		MessageQueue: okNok(messageQueueOK),
+		Status:       okNok(allOK),
 	}
 	if len(meta) > 0 {
 		result.Meta = meta
@@ -377,6 +396,11 @@ func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) {
 	return result, allOK
 }
 
+// Healthz runs lightweight dependency checks for /api/v1/system/healthz.
+func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) {
+	return GetComponentsHealthz(ctx)
+}
+
 // ListAllVariables list all variables
 // Returns all system settings from database
 func (s *SystemService) ListAllVariables() ([]map[string]interface{}, error) {

internal/service/system_test.go

@@ -10,7 +10,7 @@ func TestSystemServiceHealthzReportsUnhealthyDependencies(t *testing.T) {
 	if result.Status != "nok" {
 		t.Fatalf("status=%q, want nok", result.Status)
 	}
-	if result.DB != "nok" || result.Redis != "nok" || result.DocEngine != "nok" || result.Storage != "nok" {
+	if result.DB != "nok" || result.Redis != "nok" || result.DocEngine != "nok" || result.Storage != "nok" || result.MessageQueue != "nok" {
 		t.Fatalf("unexpected health result: %+v", result)
 	}
 	if len(result.Meta) == 0 {