Feature: Added HasWebsocketEvents field to the Orchestrator Host API, providing clients with a reliable source of truth for real-time event capabilities.

Improvement: Optimized database performance by implementing lightweight status updates for WebSocket events, avoiding full record rewrites.
Fix: Enhanced connection reliability by tying the "Connected" status strictly to verified pong responses rather than just TCP connection establishment.
Fix: Resolved delay in reconnection events by triggering an immediate ping upon connection, ensuring instant status updates.

Improvement: Refactored connection monitoring to automatically detect and handle stale WebSocket connections without redundant network traffic.

Author: sai-kumar-peddireddy

src/data/models/orchestrator_host.go

@@ -39,6 +39,7 @@ type OrchestratorHost struct {
 	UpdatedAt                 string                          `json:"updated_at,omitempty"`
 	RequiredClaims            []string                        `json:"required_claims,omitempty"`
 	RequiredRoles             []string                        `json:"required_roles,omitempty"`
+	HasWebsocketEvents        bool                            `json:"has_websocket_events,omitempty"`
 }
 
 func (o OrchestratorHost) GetHost() string {
@@ -223,6 +224,10 @@ func (o *OrchestratorHost) Diff(source OrchestratorHost) bool {
 		return true
 	}
 
+	if o.HasWebsocketEvents != source.HasWebsocketEvents {
+		return true
+	}
+
 	if o.ReverseProxy != nil && source.ReverseProxy == nil {
 		return true
 	}

src/data/orchestrator.go

@@ -238,6 +238,7 @@ func (j *JsonDatabase) UpdateOrchestratorHost(ctx basecontext.ApiContext, host *
 				j.data.OrchestratorHosts[i].ParallelsDesktopLicensed = host.ParallelsDesktopLicensed
 				// Reverse Proxy Hosts
 				j.data.OrchestratorHosts[i].IsReverseProxyEnabled = host.IsReverseProxyEnabled
+				j.data.OrchestratorHosts[i].HasWebsocketEvents = host.HasWebsocketEvents
 				j.data.OrchestratorHosts[i].ReverseProxy = host.ReverseProxy
 				j.data.OrchestratorHosts[i].ReverseProxyHosts = host.ReverseProxyHosts
 
@@ -353,6 +354,33 @@ func (j *JsonDatabase) UpdateOrchestratorHostTimestamp(ctx basecontext.ApiContex
 	return ErrOrchestratorHostNotFound
 }
 
+// UpdateOrchestratorHostWebsocketStatus updates only the HasWebsocketEvents field
+func (j *JsonDatabase) UpdateOrchestratorHostWebsocketStatus(ctx basecontext.ApiContext, hostID string, hasWebsocketEvents bool) (bool, error) {
+	if !j.IsConnected() {
+		return false, ErrDatabaseNotConnected
+	}
+
+	if hostID == "" {
+		return false, ErrOrchestratorHostEmptyIdOrHost
+	}
+
+	j.dataMutex.Lock()
+	defer j.dataMutex.Unlock()
+
+	for i, dbHost := range j.data.OrchestratorHosts {
+		if strings.EqualFold(dbHost.ID, hostID) {
+			if j.data.OrchestratorHosts[i].HasWebsocketEvents == hasWebsocketEvents {
+				return false, nil
+			}
+			j.data.OrchestratorHosts[i].HasWebsocketEvents = hasWebsocketEvents
+			ctx.LogDebugf("[Database] Host %s websocket status updated to %v", dbHost.Host, hasWebsocketEvents)
+			return true, nil
+		}
+	}
+
+	return false, ErrOrchestratorHostNotFound
+}
+
 func (j *JsonDatabase) GetOrchestratorAvailableResources(ctx basecontext.ApiContext) map[string]models.HostResourceItem {
 	j.dataMutex.RLock()
 	defer j.dataMutex.RUnlock()

src/mappers/orchestrator.go

@@ -26,6 +26,7 @@ func DtoOrchestratorHostToApiResponse(dto data_models.OrchestratorHost) models.O
 		RequiredClaims:           dto.RequiredClaims,
 		RequiredRoles:            dto.RequiredRoles,
 		State:                    dto.State,
+		HasWebsocketEvents:       dto.HasWebsocketEvents,
 	}
 
 	if dto.Resources != nil {

src/models/orchestrator_host.go

@@ -121,6 +121,7 @@ type OrchestratorHostResponse struct {
 	DetailedResources        *HostResources      `json:"detailed_resources,omitempty"`
 	RequiredClaims           []string            `json:"required_claims,omitempty"`
 	RequiredRoles            []string            `json:"required_roles,omitempty"`
+	HasWebsocketEvents       bool                `json:"has_websocket_events,omitempty"`
 }
 
 type OrchestratorHostUpdateRequest struct {

src/orchestrator/handlers/host_health_handler.go

@@ -67,6 +67,29 @@ func (h *HostHealthHandler) handlePong(ctx basecontext.ApiContext, hostID string
 		return
 	}
 
+	// If the host was not marked as having WebSocket events, update it now that we received a pong
+	if !host.HasWebsocketEvents {
+		updated, err := dbService.UpdateOrchestratorHostWebsocketStatus(ctx, hostID, true)
+		if err != nil {
+			ctx.LogErrorf("[HostHealthHandler] Error updating host %s websocket status: %v", hostID, err)
+		} else if updated {
+			// Broadcast WebSocket connection event now that we have confirmed capability via pong
+			if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
+				msg := models.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_CONNECTED", models.HostHealthUpdate{
+					HostID: host.ID,
+					State:  "websocket_connected",
+				})
+				go func() {
+					if err := emitter.Broadcast(msg); err != nil {
+						ctx.LogErrorf("[HostHealthHandler] Failed to broadcast event %s: %v", "HOST_WEBSOCKET_CONNECTED", err)
+					} else {
+						ctx.LogInfof("[HostHealthHandler] Broadcasted HOST_WEBSOCKET_CONNECTED event for host %s", hostID)
+					}
+				}()
+			}
+		}
+	}
+
 	host.UpdatedAt = helpers.GetUtcCurrentDateTime()
 	stateChanged := false
 

src/orchestrator/host_websocket_manager.go

@@ -8,6 +8,7 @@ import (
 	"github.com/Parallels/prl-devops-service/basecontext"
 	"github.com/Parallels/prl-devops-service/constants"
 	"github.com/Parallels/prl-devops-service/data/models"
+	event "github.com/Parallels/prl-devops-service/models"
 	"github.com/Parallels/prl-devops-service/orchestrator/interfaces"
 	"github.com/Parallels/prl-devops-service/serviceprovider"
 )
@@ -98,6 +99,28 @@ func (m *HostWebSocketManager) DisconnectHost(hostID string) {
 		client.Close()
 		delete(m.clients, hostID)
 		m.ctx.LogInfof("[HostWebSocketManager] Disconnected host %s", hostID)
+
+		// Update DB status
+		dbService, err := serviceprovider.GetDatabaseService(m.ctx)
+		if err == nil {
+			updated, _ := dbService.UpdateOrchestratorHostWebsocketStatus(m.ctx, hostID, false)
+			if updated {
+				if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
+					msg := event.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_DISCONNECTED",
+						event.HostHealthUpdate{
+							HostID: hostID,
+							State:  "websocket_disconnected",
+						})
+					go func() {
+						if err := emitter.Broadcast(msg); err != nil {
+							m.ctx.LogErrorf("[HostWebSocketManager] Failed to broadcast HOST_WEBSOCKET_DISCONNECTED event: %v", err)
+						} else {
+							m.ctx.LogInfof("[HostWebSocketManager] Broadcasted HOST_WEBSOCKET_DISCONNECTED event for host %s", hostID)
+						}
+					}()
+				}
+			}
+		}
 	}
 }
 

src/orchestrator/main.go

@@ -171,16 +171,35 @@ func (s *OrchestratorService) processHost(host models.OrchestratorHost) {
 		stalenessThreshold := s.refreshInterval * 3
 
 		if err == nil && time.Since(lastUpdated) < stalenessThreshold {
-			s.ctx.LogDebugf("[Orchestrator] Host %s is connected and fresh (last updated: %s), sending ping", host.Host, host.UpdatedAt)
-			if err := manager.SendPing(host.ID); err != nil {
-				s.ctx.LogWarnf("[Orchestrator] Failed to send ping to host %s: %v, falling back to HTTP health check", host.Host, err)
-				websocketPingFailed = true
-			} else {
-				// Ping successful, skip HTTP health check
-				return
+			s.ctx.LogDebugf("[Orchestrator] Host %s is connected and fresh (last updated: %s). Skipping HTTP health check.", host.Host, host.UpdatedAt)
+			// Ping successful (implied by freshness), skip HTTP health check
+			if !host.HasWebsocketEvents {
+				host.HasWebsocketEvents = true
+				_, _ = s.db.UpdateOrchestratorHostWebsocketStatus(s.ctx, host.ID, true)
 			}
+			return
 		} else {
 			s.ctx.LogWarnf("[Orchestrator] Host %s is connected but stale (last updated: %s). Falling back to HTTP health check.", host.Host, host.UpdatedAt)
+			websocketPingFailed = true
+			if host.HasWebsocketEvents {
+				host.HasWebsocketEvents = false
+				updated, _ := s.db.UpdateOrchestratorHostWebsocketStatus(s.ctx, host.ID, false)
+				if updated {
+					if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
+						msg := apimodels.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_DISCONNECTED", apimodels.HostHealthUpdate{
+							HostID: host.ID,
+							State:  "websocket_disconnected",
+						})
+						go func() {
+							if err := emitter.Broadcast(msg); err != nil {
+								s.ctx.LogErrorf("[Orchestrator] Failed to broadcast event %s: %v", "HOST_WEBSOCKET_DISCONNECTED", err)
+							} else {
+								s.ctx.LogInfof("[Orchestrator] Broadcasted HOST_WEBSOCKET_DISCONNECTED event for host %s (detected staleness in processHost)", host.Host)
+							}
+						}()
+					}
+				}
+			}
 		}
 	}
 

src/orchestrator/websocket_client.go

@@ -142,8 +142,6 @@ func (c *HostWebSocketClient) establishConnection(events []constants.EventType)
 	c.setConnected(true)
 	c.ctx.LogInfof("[HostWebSocketClient] Connected to host %s", c.hostName)
 
-	c.broadcastConnectionEvent()
-
 	return nil
 }
 
@@ -215,6 +213,11 @@ func (c *HostWebSocketClient) startPingRoutine() {
 	c.mu.Unlock()
 
 	go func() {
+		// Send initial ping immediately to confirm connection status
+		if err := c.SendPing(); err != nil {
+			c.ctx.LogDebugf("[HostWebSocketClient] Failed to send initial ping to host %s: %v", c.hostName, err)
+		}
+
 		for {
 			select {
 			case <-c.stopChan:
@@ -268,44 +271,29 @@ func (c *HostWebSocketClient) notifyDisconnection() {
 
 	c.setConnected(false)
 
-	if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
-		msg := api_models.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_DISCONNECTED", api_models.HostHealthUpdate{
-			HostID: c.hostID,
-			State:  "websocket_disconnected",
-		})
-		go func() {
-			if err := emitter.Broadcast(msg); err != nil {
-				c.ctx.LogErrorf("[HostWebSocketClient] Failed to broadcast HOST_WEBSOCKET_DISCONNECTED event: %v", err)
-			} else {
-				c.ctx.LogInfof("[HostWebSocketClient] Broadcasted HOST_WEBSOCKET_DISCONNECTED event for host %s", c.hostName)
+	// Update DB status
+	if dbService, err := serviceprovider.GetDatabaseService(c.ctx); err == nil {
+		updated, _ := dbService.UpdateOrchestratorHostWebsocketStatus(c.ctx, c.hostID, false)
+		if updated {
+			if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
+				msg := api_models.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_DISCONNECTED", api_models.HostHealthUpdate{
+					HostID: c.hostID,
+					State:  "websocket_disconnected",
+				})
+				go func() {
+					if err := emitter.Broadcast(msg); err != nil {
+						c.ctx.LogErrorf("[HostWebSocketClient] Failed to broadcast HOST_WEBSOCKET_DISCONNECTED event: %v", err)
+					} else {
+						c.ctx.LogInfof("[HostWebSocketClient] Broadcasted HOST_WEBSOCKET_DISCONNECTED event for host %s", c.hostName)
+					}
+				}()
 			}
-		}()
+		}
 	} else {
 		c.ctx.LogWarnf("[HostWebSocketClient] EventEmitter not available to broadcast disconnection for host %s", c.hostName)
 	}
 }
 
-func (c *HostWebSocketClient) broadcastConnectionEvent() {
-	c.ctx.LogInfof("[HostWebSocketClient] Host %s WebSocket connection established", c.hostName)
-
-	// Broadcast WebSocket connection event
-	if emitter := serviceprovider.GetEventEmitter(); emitter != nil && emitter.IsRunning() {
-		msg := api_models.NewEventMessage(constants.EventTypeOrchestrator, "HOST_WEBSOCKET_CONNECTED", api_models.HostHealthUpdate{
-			HostID: c.hostID,
-			State:  "websocket_connected",
-		})
-		go func() {
-			if err := emitter.Broadcast(msg); err != nil {
-				c.ctx.LogErrorf("[HostWebSocketClient] Failed to broadcast HOST_WEBSOCKET_CONNECTED event: %v", err)
-			} else {
-				c.ctx.LogInfof("[HostWebSocketClient] Broadcasted HOST_WEBSOCKET_CONNECTED event for host %s", c.hostName)
-			}
-		}()
-	} else {
-		c.ctx.LogWarnf("[HostWebSocketClient] EventEmitter not available to broadcast connection for host %s", c.hostName)
-	}
-}
-
 func (c *HostWebSocketClient) Probe() bool {
 	// Use a short timeout for probing
 	c.ctx.LogInfof("[HostWebSocketClient] Probing host %s for WebSocket support...", c.hostName)