diff --git a/dashboard/dist/index.html b/dashboard/dist/index.html index 137e840..5f090a6 100644 --- a/dashboard/dist/index.html +++ b/dashboard/dist/index.html @@ -21,7 +21,7 @@ min-height: 100vh; } - + diff --git a/dashboard/src/components/ServiceNav.tsx b/dashboard/src/components/ServiceNav.tsx index f8ab3fa..c83bd3d 100644 --- a/dashboard/src/components/ServiceNav.tsx +++ b/dashboard/src/components/ServiceNav.tsx @@ -6,7 +6,6 @@ interface ServiceInfo { baseUrl: string; status: string; lastHeartbeat: string; - role?: string; // "leader", "follower", or undefined (for non-meta-core services) } interface ServicesResponse { @@ -56,11 +55,7 @@ function ServiceNav() { if (loading) return null; - // Filter out follower meta-core instances (only show leader) - const filteredServices = services.filter( - s => s.name !== 'meta-core' || s.role === 'leader' - ); - const sortedServices = [...filteredServices].sort((a, b) => a.name.localeCompare(b.name)); + const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name)); if (sortedServices.length === 0) return null; diff --git a/internal/api/handlers.go b/internal/api/handlers.go index ec21cac..11a7063 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -400,7 +400,8 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) { http.ServeContent(w, r, fileInfo.Name(), fileInfo.ModTime(), file) } -// handleListServices handles GET /services +// handleListServices handles GET /services and GET /api/services +// Supports optional ?current= query parameter func (s *Server) handleListServices(w http.ResponseWriter, r *http.Request) { services, err := s.discovery.DiscoverAll() if err != nil { @@ -408,10 +409,18 @@ func (s *Server) handleListServices(w http.ResponseWriter, r *http.Request) { return } - writeJSON(w, http.StatusOK, map[string]interface{}{ + response := map[string]interface{}{ "services": services, "count": len(services), - }) + } + + // Add 'current' field if query parameter is provided + current := r.URL.Query().Get("current") + if current != "" { + response["current"] = current + } + + writeJSON(w, http.StatusOK, response) } // handleGetService handles GET /services/{name} diff --git a/internal/api/server.go b/internal/api/server.go index 75de929..a0aba66 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -90,7 +90,8 @@ func NewServer( // Create watchers poller if s.fileWatcher != nil { s.watchersPoller = watchers.NewPoller(watchersManager, s.fileWatcher, s.watcherDispatcher) - s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller) + // Pass republish callback to handlers (handles nil metaPublisher case) + s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller, s.RepublishMetadata) } } } @@ -173,7 +174,9 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST") // Service discovery + // Both /services and /api/services are supported for consistency with other services s.router.HandleFunc("/services", s.handleListServices).Methods("GET") + s.router.HandleFunc("/api/services", s.handleListServices).Methods("GET") s.router.HandleFunc("/services/cleanup/stats", s.handleCleanupStats).Methods("GET") s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET") @@ -271,6 +274,16 @@ func (s *Server) Start() error { s.metaPublisher = events.NewMetaPublisher(s.storage.GetRedisClient()) if err := s.metaPublisher.Start(); err != nil { log.Printf("[API] Warning: failed to start meta publisher: %v", err) + } else { + // Republish events for all existing metadata + go func() { + count, err := s.RepublishMetadata() + if err != nil { + log.Printf("[API] Warning: failed to republish metadata on startup: %v", err) + } else { + log.Printf("[API] Republished %d metadata events on startup", count) + } + }() } } @@ -326,6 +339,28 @@ func (s *Server) Stop() error { return s.server.Shutdown(ctx) } +// RepublishMetadata republishes all existing metadata to the meta:events stream +// This is called on startup and after reset-all +func (s *Server) RepublishMetadata() (int, error) { + if s.metaPublisher == nil { + return 0, fmt.Errorf("meta publisher not initialized") + } + if s.storage == nil || !s.storage.IsConnected() { + return 0, fmt.Errorf("storage not connected") + } + + // Create the republish function that wraps storage methods + republishFunc := func() ([]string, func(string) (map[string]string, error), error) { + hashIDs, err := s.storage.GetAllHashIDs() + if err != nil { + return nil, nil, err + } + return hashIDs, s.storage.GetMetadataFlat, nil + } + + return s.metaPublisher.RepublishAllMetadata(republishFunc) +} + // loggingMiddleware logs all requests func loggingMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/events/meta_publisher.go b/internal/events/meta_publisher.go index 68da340..9fa8d71 100644 --- a/internal/events/meta_publisher.go +++ b/internal/events/meta_publisher.go @@ -143,3 +143,63 @@ func (p *MetaPublisher) publishEvent(channel, operation string) { log.Printf("[MetaPublisher] Error publishing event: %v", err) } } + +// publishEventDirect publishes an event directly (not from keyspace notification) +func (p *MetaPublisher) publishEventDirect(operation, key string) { + fields := map[string]interface{}{ + "type": operation, + "key": key, + "ts": fmt.Sprintf("%d", time.Now().UnixMilli()), + } + + _, err := p.client.XAdd(p.ctx, &redis.XAddArgs{ + Stream: MetaEventsStream, + Values: fields, + }).Result() + + if err != nil { + log.Printf("[MetaPublisher] Error publishing direct event: %v", err) + } +} + +// RepublishFunc is a function type for getting metadata that can be republished +type RepublishFunc func() (hashIDs []string, getMetadata func(hashID string) (map[string]string, error), err error) + +// RepublishAllMetadata iterates over all existing file metadata and publishes events +// This is called on startup and can be triggered via API +func (p *MetaPublisher) RepublishAllMetadata(republishFunc RepublishFunc) (int, error) { + p.mu.RLock() + running := p.running + p.mu.RUnlock() + + if !running { + return 0, fmt.Errorf("publisher not running") + } + + // Get all hash IDs and metadata getter + hashIDs, getMetadata, err := republishFunc() + if err != nil { + return 0, fmt.Errorf("failed to get hash IDs: %w", err) + } + + log.Printf("[MetaPublisher] Republishing events for %d files...", len(hashIDs)) + + publishedCount := 0 + for _, hashID := range hashIDs { + // Get all properties for this file + metadata, err := getMetadata(hashID) + if err != nil || metadata == nil { + continue + } + + // Publish a set event for each property + for property := range metadata { + key := fmt.Sprintf("file:%s/%s", hashID, property) + p.publishEventDirect("set", key) + publishedCount++ + } + } + + log.Printf("[MetaPublisher] Republished %d property events", publishedCount) + return publishedCount, nil +} diff --git a/internal/watchers/handlers.go b/internal/watchers/handlers.go index 32fe44e..986851b 100644 --- a/internal/watchers/handlers.go +++ b/internal/watchers/handlers.go @@ -2,22 +2,28 @@ package watchers import ( "encoding/json" + "log" "net/http" "github.com/gorilla/mux" ) +// RepublishCallback is a function type for republishing metadata events +type RepublishCallback func() (int, error) + // Handlers provides HTTP handlers for watchers management type Handlers struct { - manager *Manager - poller *Poller + manager *Manager + poller *Poller + republishCallback RepublishCallback } // NewHandlers creates new watchers handlers -func NewHandlers(manager *Manager, poller *Poller) *Handlers { +func NewHandlers(manager *Manager, poller *Poller, republishCallback RepublishCallback) *Handlers { return &Handlers{ - manager: manager, - poller: poller, + manager: manager, + poller: poller, + republishCallback: republishCallback, } } @@ -209,10 +215,22 @@ func (h *Handlers) handleResetAll(w http.ResponseWriter, r *http.Request) { return } + // Republish all metadata events after reset + republishedCount := 0 + if h.republishCallback != nil { + count, err := h.republishCallback() + if err != nil { + log.Printf("[Handlers] Warning: failed to republish metadata after reset: %v", err) + } else { + republishedCount = count + } + } + writeJSON(w, http.StatusOK, map[string]interface{}{ - "status": "ok", - "message": "All watchers reset", - "totalFiles": totalCount, + "status": "ok", + "message": "All watchers reset", + "totalFiles": totalCount, + "republishedEvents": republishedCount, }) }