refactor: switch terminal to use schemaeventsource materialised view

frontend/cli/cmd_dev.go

@@ -59,7 +59,7 @@ func (d *devCmd) Run(
 		return errors.New("no directories specified")
 	}
 
-	terminal.LaunchEmbeddedConsole(ctx, k, bindContext, schemaClient)
+	terminal.LaunchEmbeddedConsole(ctx, k, bindContext, schemaEventSourceFactory())
 	var client buildengine.DeployClient = controllerClient
 	if d.ServeCmd.Provisioners > 0 {
 		client = rpc.ClientFromContext[provisionerconnect.ProvisionerServiceClient](ctx)

frontend/cli/cmd_interactive.go

@@ -6,15 +6,15 @@ import (
 
 	"github.com/alecthomas/kong"
 
-	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/TBD54566975/ftl/internal/schema/schemaeventsource"
 	"github.com/TBD54566975/ftl/internal/terminal"
 )
 
 type interactiveCmd struct {
 }
 
-func (i *interactiveCmd) Run(ctx context.Context, k *kong.Kong, binder terminal.KongContextBinder, client ftlv1connect.SchemaServiceClient) error {
-	err := terminal.RunInteractiveConsole(ctx, k, binder, client)
+func (i *interactiveCmd) Run(ctx context.Context, k *kong.Kong, binder terminal.KongContextBinder, eventSource schemaeventsource.EventSource) error {
+	err := terminal.RunInteractiveConsole(ctx, k, binder, eventSource)
 	if err != nil {
 		return fmt.Errorf("interactive console: %w", err)
 	}

frontend/cli/main.go

@@ -241,7 +241,7 @@ func makeBindContext(logger *log.Logger, cancel context.CancelFunc) terminal.Kon
 		})
 		kctx.FatalIfErrorf(err)
 
-		kongcompletion.Register(kctx.Kong, kongcompletion.WithPredictors(terminal.Predictors(ctx, schemaServiceClient)))
+		kongcompletion.Register(kctx.Kong, kongcompletion.WithPredictors(terminal.Predictors(schemaeventsource.New(ctx, schemaServiceClient).ViewOnly())))
 
 		verbServiceClient := rpc.Dial(ftlv1connect.NewVerbServiceClient, cli.Endpoint.String(), log.Error)
 		ctx = rpc.ContextWithClient(ctx, verbServiceClient)

internal/schema/schemaeventsource/schemaeventsource.go

@@ -18,6 +18,14 @@ import (
 	"github.com/TBD54566975/ftl/internal/schema"
 )
 
+// View is a read-only view of the schema.
+type View struct {
+	view *atomic.Value[*schema.Schema]
+}
+
+// Get returns the current schema.
+func (v *View) Get() *schema.Schema { return v.view.Load() }
+
 // Event represents a change in the schema.
 //
 //sumtype:decl
@@ -79,6 +87,17 @@ type EventSource struct {
 // channel.
 func (e EventSource) Events() <-chan Event { return e.events }
 
+// ViewOnly converts the EventSource into a read-only view of the schema.
+//
+// This will consume all events so the EventSource dodesn't block as the view is automatically updated.
+func (e EventSource) ViewOnly() View {
+	go func() {
+		for range e.Events() { //nolint:revive
+		}
+	}()
+	return View{e.view}
+}
+
 // View is the materialised view of the schema from "Events".
 func (e EventSource) View() *schema.Schema { return e.view.Load() }
 

internal/terminal/interactive.go

@@ -15,7 +15,7 @@ import (
 	"github.com/kballard/go-shellquote"
 	"github.com/posener/complete"
 
-	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/TBD54566975/ftl/internal/schema/schemaeventsource"
 )
 
 const interactivePrompt = "\033[32m>\033[0m "
@@ -26,15 +26,14 @@ type KongContextBinder func(ctx context.Context, kctx *kong.Context) context.Con
 
 type exitPanic struct{}
 
-func RunInteractiveConsole(ctx context.Context, k *kong.Kong, binder KongContextBinder, client ftlv1connect.SchemaServiceClient) error {
-
+func RunInteractiveConsole(ctx context.Context, k *kong.Kong, binder KongContextBinder, eventSource schemaeventsource.EventSource) error {
 	if !readline.DefaultIsTerminal() {
 		return nil
 	}
 	l, err := readline.NewEx(&readline.Config{
 		Prompt:          interactivePrompt,
 		InterruptPrompt: "^C",
-		AutoComplete:    &FTLCompletion{app: k, ctx: ctx, client: client},
+		AutoComplete:    &FTLCompletion{app: k, view: eventSource.ViewOnly()},
 		Listener: &ExitListener{cancel: func() {
 			_ = syscall.Kill(-syscall.Getpid(), syscall.SIGINT) //nolint:forcetypeassert,errcheck // best effort
 		}},
@@ -141,9 +140,8 @@ func errorf(format string, args ...any) {
 }
 
 type FTLCompletion struct {
-	app    *kong.Kong
-	client ftlv1connect.SchemaServiceClient
-	ctx    context.Context
+	app  *kong.Kong
+	view schemaeventsource.View
 }
 
 func (f *FTLCompletion) Do(line []rune, pos int) ([][]rune, int) {
@@ -197,7 +195,7 @@ func (f *FTLCompletion) Do(line []rune, pos int) ([][]rune, int) {
 		LastCompleted: lastCompleted,
 	}
 
-	command, err := kongcompletion.Command(parser, kongcompletion.WithPredictors(Predictors(f.ctx, f.client)))
+	command, err := kongcompletion.Command(parser, kongcompletion.WithPredictors(Predictors(f.view)))
 	if err != nil {
 		// TODO handle error
 		println(err.Error())

internal/terminal/predictors.go

@@ -1,41 +1,30 @@
 package terminal
 
 import (
-	"context"
-
-	"connectrpc.com/connect"
 	"github.com/posener/complete"
 
-	ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/TBD54566975/ftl/internal/schema"
+	"github.com/TBD54566975/ftl/internal/schema/schemaeventsource"
 )
 
-func Predictors(ctx context.Context, client ftlv1connect.SchemaServiceClient) map[string]complete.Predictor {
+func Predictors(view schemaeventsource.View) map[string]complete.Predictor {
 	return map[string]complete.Predictor{
-		"verbs": &verbPredictor{
-			Client: client,
-			Ctx:    ctx,
-		},
+		"verbs": &verbPredictor{view: view},
 	}
 }
 
 type verbPredictor struct {
-	Client ftlv1connect.SchemaServiceClient
-	Ctx    context.Context
+	view schemaeventsource.View
 }
 
 func (v *verbPredictor) Predict(args complete.Args) []string {
-	response, err := v.Client.GetSchema(v.Ctx, connect.NewRequest(&ftlv1.GetSchemaRequest{}))
-	if err != nil {
-		// Do we want to report errors here?
-		return nil
-	}
+	sch := v.view.Get()
 	ret := []string{}
-	for _, module := range response.Msg.Schema.Modules {
+	for _, module := range sch.Modules {
 		for _, dec := range module.Decls {
-			if dec.GetVerb() != nil {
-				verb := module.Name + "." + dec.GetVerb().Name
-				ret = append(ret, verb)
+			if verb, ok := dec.(*schema.Verb); ok {
+				ref := schema.Ref{Module: module.Name, Name: verb.Name}
+				ret = append(ret, ref.String())
 			}
 		}
 	}

internal/terminal/status.go

@@ -18,8 +18,8 @@ import (
 	"github.com/tidwall/pretty"
 	"golang.org/x/term"
 
-	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
 	"github.com/TBD54566975/ftl/internal/log"
+	"github.com/TBD54566975/ftl/internal/schema/schemaeventsource"
 )
 
 type BuildState string
@@ -491,11 +491,11 @@ func (r *terminalStatusLine) SetMessage(message string) {
 	r.manager.recalculateLines()
 }
 
-func LaunchEmbeddedConsole(ctx context.Context, k *kong.Kong, binder KongContextBinder, client ftlv1connect.SchemaServiceClient) {
+func LaunchEmbeddedConsole(ctx context.Context, k *kong.Kong, binder KongContextBinder, eventSource schemaeventsource.EventSource) {
 	sm := FromContext(ctx)
 	if _, ok := sm.(*terminalStatusManager); ok {
 		go func() {
-			err := RunInteractiveConsole(ctx, k, binder, client)
+			err := RunInteractiveConsole(ctx, k, binder, eventSource)
 			if err != nil {
 				fmt.Printf("\033[31mError: %s\033[0m\n", err)
 				return