Add scheduled tasks (#13470)

### What problem does this PR solve?

1. RAGFlow server will send heartbeat periodically.
2. This PR will including:
- Scheduled task
- API server message sending
- Admin server API to receive the message.

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

---------

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

internal/admin/handler.go

@@ -19,10 +19,12 @@ package admin
 import (
 	"errors"
 	"net/http"
+	"ragflow/internal/common"
 	"ragflow/internal/server"
 	"ragflow/internal/service"
 	"ragflow/internal/utility"
 	"strconv"
+	"time"
 
 	"github.com/gin-gonic/gin"
 )
@@ -111,7 +113,7 @@ func (h *Handler) Login(c *gin.Context) {
 		return
 	}
 
-	user, code, err := h.userService.LoginByEmail(&req)
+	user, code, err := h.userService.LoginByEmail(&req, true)
 	if err != nil {
 		c.JSON(http.StatusUnauthorized, gin.H{
 			"code":    code,
@@ -135,8 +137,9 @@ func (h *Handler) Login(c *gin.Context) {
 	c.Header("Access-Control-Expose-Headers", "Authorization")
 
 	c.JSON(http.StatusOK, gin.H{
-		"code":    0,
-		"message": "Login successful",
+		"code":    common.CodeSuccess,
+		"message": "Welcome back!",
+		"data":    user,
 	})
 }
 
@@ -943,3 +946,31 @@ func (h *Handler) HandleNoRoute(c *gin.Context) {
 		Message: "The requested resource was not found",
 	})
 }
+
+// Reports handle heartbeat reports from servers
+func (h *Handler) Reports(c *gin.Context) {
+	var req common.BaseMessage
+	if err := c.ShouldBindJSON(&req); err != nil {
+		errorResponse(c, "Invalid request body: "+err.Error(), 400)
+		return
+	}
+
+	// Set default timestamp if not provided
+	if req.Timestamp.IsZero() {
+		req.Timestamp = time.Now()
+	}
+
+	// Only process heartbeat messages for now
+	if req.MessageType != common.MessageHeartbeat {
+		errorResponse(c, "Unsupported report type: "+string(req.MessageType), 400)
+		return
+	}
+
+	// Handle the heartbeat
+	if err := h.service.HandleHeartbeat(&req); err != nil {
+		errorResponse(c, "Failed to process heartbeat: "+err.Error(), 500)
+		return
+	}
+
+	successNoData(c, "Heartbeat received successfully")
+}

internal/admin/heartbeat.go

@@ -0,0 +1,76 @@
+package admin
+
+import (
+	"ragflow/internal/common"
+	"sync"
+	"time"
+)
+
+// ServerStatusStore is a thread-safe global server status storage
+type ServerStatusStore struct {
+	mu      sync.RWMutex
+	servers map[string]*common.BaseMessage // key: server_id
+}
+
+// GlobalServerStatusStore is the global instance
+var GlobalServerStatusStore = &ServerStatusStore{
+	servers: make(map[string]*common.BaseMessage),
+}
+
+// UpdateStatus updates or adds a server status
+func (s *ServerStatusStore) UpdateStatus(serverID string, status *common.BaseMessage) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.servers[serverID] = status
+}
+
+// GetStatus gets a single server status
+func (s *ServerStatusStore) GetStatus(serverID string) (*common.BaseMessage, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	status, ok := s.servers[serverID]
+	return status, ok
+}
+
+// GetAllStatuses gets all server statuses
+func (s *ServerStatusStore) GetAllStatuses() []*common.BaseMessage {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	result := make([]*common.BaseMessage, 0, len(s.servers))
+	for _, status := range s.servers {
+		result = append(result, status)
+	}
+	return result
+}
+
+// GetStatusesByType gets server statuses by type
+func (s *ServerStatusStore) GetStatusesByType(serverType common.ServerType) []*common.BaseMessage {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	result := make([]*common.BaseMessage, 0)
+	for _, status := range s.servers {
+		if status.ServerType == serverType {
+			result = append(result, status)
+		}
+	}
+	return result
+}
+
+// RemoveStatus removes a server status
+func (s *ServerStatusStore) RemoveStatus(serverID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.servers, serverID)
+}
+
+// CleanupStaleStatuses cleans up servers that haven't reported for a specified duration
+func (s *ServerStatusStore) CleanupStaleStatuses(maxAge time.Duration) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	now := time.Now()
+	for id, status := range s.servers {
+		if now.Sub(status.Timestamp) > maxAge {
+			delete(s.servers, id)
+		}
+	}
+}

internal/admin/router.go

@@ -47,6 +47,8 @@ func (r *Router) Setup(engine *gin.Engine) {
 		admin.GET("/ping", r.handler.Ping)
 		admin.POST("/login", r.handler.Login)
 
+		admin.POST("/reports", r.handler.Reports)
+
 		// Protected routes
 		protected := admin.Group("")
 		protected.Use(r.handler.AuthMiddleware())

internal/admin/service.go

@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"os"
 	"ragflow/internal/cache"
+	"ragflow/internal/common"
 	"ragflow/internal/dao"
 	"ragflow/internal/engine/elasticsearch"
 	"ragflow/internal/model"
@@ -732,3 +733,18 @@ func (s *Service) TestSandboxConnection(providerType string, config map[string]i
 		"connected":     true,
 	}, nil
 }
+
+// HandleHeartbeat handle heartbeat
+func (s *Service) HandleHeartbeat(msg *common.BaseMessage) error {
+	status := &common.BaseMessage{
+		ServerName: msg.ServerName,
+		ServerType: msg.ServerType,
+		Host:       msg.Host,
+		Port:       msg.Port,
+		Version:    msg.Version,
+		Timestamp:  msg.Timestamp,
+		Ext:        msg.Ext,
+	}
+	GlobalServerStatusStore.UpdateStatus(msg.ServerName, status)
+	return nil
+}