From 1775e8fd1f034fabb0dd3b796f5caf62a789d610 Mon Sep 17 00:00:00 2001
From: Denise Li <dli@squareup.com>
Date: Wed, 9 Oct 2024 20:22:49 -0400
Subject: [PATCH] feat: streaming module updates from console service backend
 (#3064)

Fixes: https://github.com/TBD54566975/ftl/issues/2805
---
 backend/controller/console/console.go         | 27 ++++++++++++++++++-
 .../console/console_integration_test.go       |  1 +
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/backend/controller/console/console.go b/backend/controller/console/console.go
index c553f631b..81e069535 100644
--- a/backend/controller/console/console.go
+++ b/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
 }
 
diff --git a/backend/controller/console/console_integration_test.go b/backend/controller/console/console_integration_test.go
index aedd1a126..9e6a07d46 100644
--- a/backend/controller/console/console_integration_test.go
+++ b/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
 			}
 		}),
 	)