Improves sending pending prompts via websocket and adds config for log enabled for websocket

app/boot_levels.go

@@ -320,6 +320,7 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
 	corredor.Service().SetRoleFinder(sysService.DefaultRole)
 
 	app.WsServer = websocket.Websocket(&websocket.Config{
+		LogEnabled:  app.Opt.Websocket.LogEnabled,
 		Timeout:     app.Opt.Websocket.Timeout,
 		PingTimeout: app.Opt.Websocket.PingTimeout,
 		PingPeriod:  app.Opt.Websocket.PingPeriod,

automation/service/session.go

@@ -236,7 +236,6 @@ func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool) (ses *
 
 func (svc *session) Watch(ctx context.Context) {
 	gcTicker := time.NewTicker(time.Second)
-	promptTicker := time.NewTicker(time.Second)
 
 	go func() {
 		defer sentry.Recover()
@@ -266,6 +265,18 @@ func (svc *session) Watch(ctx context.Context) {
 				// @todo cleanup pool when sessions are complete
 
 			case <-gcTicker.C:
+				// Sends pending prompt via websocket
+				ws := websocket.Session(ctx, nil, nil)
+				userIds := ws.GetActiveUserIDs()
+				for _, s := range svc.pool {
+					for _, u := range userIds {
+						pp := s.PendingPrompts(u)
+						if len(pp) > 0 {
+							_ = ws.Send(websocket.Message(websocket.StatusOK, s.WorkflowID, pp), u)
+						}
+					}
+				}
+
 				svc.gc()
 			}
 		}
@@ -274,29 +285,6 @@ func (svc *session) Watch(ctx context.Context) {
 		//svc.suspendAll(ctx)
 	}()
 
-	// Prompt routine
-	go func() {
-		defer sentry.Recover()
-		defer promptTicker.Stop()
-		defer svc.log.Info("stopped")
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-promptTicker.C:
-				activeSessions := websocket.GetActiveSessions()
-				for _, s := range activeSessions {
-					var ctxr = context.Background()
-					pp := svc.PendingPrompts(auth.SetIdentityToContext(ctxr, s.User()))
-					if len(pp) > 0 {
-						_ = s.Send(websocket.Message(websocket.StatusOK, websocket.WorkflowApplication, pp), s.User().Identity())
-					}
-				}
-			}
-		}
-	}()
-
 	svc.log.Debug("watcher initialized")
 }
 

pkg/options/websocket.gen.go

@@ -14,6 +14,7 @@ import (
 
 type (
 	WebsocketOpt struct {
+		LogEnabled  bool          `env:"WEBSOCKET_LOG_ENABLED"`
 		Timeout     time.Duration `env:"WEBSOCKET_TIMEOUT"`
 		PingTimeout time.Duration `env:"WEBSOCKET_PING_TIMEOUT"`
 		PingPeriod  time.Duration `env:"WEBSOCKET_PING_PERIOD"`

pkg/options/websocket.yaml

@@ -6,6 +6,10 @@ docs:
   description: A Websocket server emphasize the trigger events and actions.
 
 props:
+  - name: LogEnabled
+    type: bool
+    description: Enable extra logging for authentication flows
+
   - name: Timeout
     type: time.Duration
     default: 15 * time.Second

websocket/config.go

@@ -6,6 +6,7 @@ import (
 
 type (
 	Config struct {
+		LogEnabled  bool
 		Timeout     time.Duration
 		PingTimeout time.Duration
 		PingPeriod  time.Duration

websocket/encoder.go

@@ -7,14 +7,12 @@ import (
 const (
 	StatusOK    = "ok"
 	StatusError = "error"
-
-	WorkflowApplication = "Workflow"
 )
 
 type (
 	message struct {
 		Status      string      `json:"status"`
-		Application string      `json:"application"`
+		Application uint64      `json:"application"`
 		Data        interface{} `json:"data"`
 	}
 
@@ -23,7 +21,7 @@ type (
 	}
 )
 
-func Message(status, application string, data interface{}) *message {
+func Message(status string, application uint64, data interface{}) *message {
 	return &message{
 		Status:      status,
 		Application: application,

websocket/router.go

@@ -10,7 +10,5 @@ import (
 // If it's valid then we keep the connection open or close it
 func (ws *websocket) MountRoutes(r chi.Router) {
 	// Initialize handlers & controllers.
-	r.Group(func(r chi.Router) {
-		r.Get("/", ws.Open)
-	})
+	r.Get("/", ws.Open)
 }

websocket/session.go

@@ -310,6 +310,16 @@ func (s *session) sendBytes(p []byte) error {
 	return nil
 }
 
-func GetActiveSessions() map[uint64]*session {
-	return sessions
+func (s *session) Walk(callback func(*session)) {
+	for _, sess := range sessions {
+		callback(sess)
+	}
+}
+
+// GetActiveUserIDs return userIDs from active ws sessions
+func (s *session) GetActiveUserIDs() (ids []uint64) {
+	s.Walk(func(sess *session) {
+		ids = append(ids, sess.user.Identity())
+	})
+	return
 }

websocket/types.go

@@ -31,6 +31,7 @@ func (a *Auth) ParseWithClaims() (jwt.MapClaims, error) {
 	}
 }
 
-func Unmarshal(raw []byte) (p *Payload, err error) {
-	return p, json.Unmarshal(raw, p)
+func Unmarshal(raw []byte) (*Payload, error) {
+	var p Payload
+	return &p, json.Unmarshal(raw, &p)
 }

websocket/websocket.go

@@ -27,6 +27,10 @@ type (
 )
 
 func Websocket(config *Config, logger *zap.Logger) *websocket {
+	if !config.LogEnabled {
+		logger = zap.NewNop()
+	}
+
 	return &websocket{
 		config: config,
 		logger: logger,