feat: streaming module updates from console service backend (#3064)

Fixes: #2805

backend/controller/console/console.go

@@ -327,6 +327,32 @@ func verbFromDecl(decl *schema.Verb, sch *schema.Schema) (*pbconsole.Verb, error
 }
 
 func (c *ConsoleService) StreamModules(ctx context.Context, req *connect.Request[pbconsole.StreamModulesRequest], stream *connect.ServerStream[pbconsole.StreamModulesResponse]) error {
+	deploymentChanges := make(chan dal.DeploymentNotification, 32)
+
+	// Subscribe to deployment changes.
+	c.dal.DeploymentChanges.Subscribe(deploymentChanges)
+	defer c.dal.DeploymentChanges.Unsubscribe(deploymentChanges)
+
+	err := c.sendStreamModulesResp(ctx, stream)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+
+		case <-deploymentChanges:
+			err = c.sendStreamModulesResp(ctx, stream)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func (c *ConsoleService) sendStreamModulesResp(ctx context.Context, stream *connect.ServerStream[pbconsole.StreamModulesResponse]) error {
 	deployments, err := c.dal.GetDeploymentsWithMinReplicas(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to get deployments: %w", err)
@@ -364,7 +390,6 @@ func (c *ConsoleService) StreamModules(ctx context.Context, req *connect.Request
 		return fmt.Errorf("failed to send StreamModulesResponse to stream: %w", err)
 	}
 
-	// TODO: handle deployment updates
 	return nil
 }
 

backend/controller/console/console_integration_test.go

@@ -56,6 +56,7 @@ func TestConsoleStreamModules(t *testing.T) {
 				assert.Equal(t, 2, len(stream.Msg().Modules))
 				assert.Equal(t, "console", stream.Msg().Modules[0].Name)
 				assert.Equal(t, "builtin", stream.Msg().Modules[1].Name)
+				break
 			}
 		}),
 	)