revert: "chore: remove "ftl schema generate""

frontend/cli/cmd_schema.go

@@ -1,7 +1,8 @@
 package main
 
 type schemaCmd struct {
-	Get    getSchemaCmd    `default:"" cmd:"" help:"Retrieve the cluster FTL schema."`
-	Diff   schemaDiffCmd   `cmd:"" help:"Print any schema differences between this cluster and another cluster. Returns an exit code of 1 if there are differences."`
-	Import schemaImportCmd `cmd:"" help:"Import messages to the FTL schema."`
+	Get      getSchemaCmd      `default:"" cmd:"" help:"Retrieve the cluster FTL schema."`
+	Diff     schemaDiffCmd     `cmd:"" help:"Print any schema differences between this cluster and another cluster. Returns an exit code of 1 if there are differences."`
+	Generate schemaGenerateCmd `cmd:"" help:"Stream the schema from the cluster and generate files from the template."`
+	Import   schemaImportCmd   `cmd:"" help:"Import messages to the FTL schema."`
 }

frontend/cli/cmd_schema_generate.go

@@ -0,0 +1,186 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"connectrpc.com/connect"
+	"github.com/block/scaffolder"
+	"github.com/block/scaffolder/extensions/javascript"
+	"github.com/radovskyb/watcher"
+	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
+
+	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
+	"github.com/block/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/block/ftl/common/schema"
+	"github.com/block/ftl/common/slices"
+	"github.com/block/ftl/internal/log"
+)
+
+type schemaGenerateCmd struct {
+	Watch          time.Duration `short:"w" help:"Watch template directory at this frequency and regenerate on change."`
+	Template       string        `arg:"" help:"Template directory to use." type:"existingdir"`
+	Dest           string        `arg:"" help:"Destination directory to write files to (will be erased)."`
+	ReconnectDelay time.Duration `help:"Delay before attempting to reconnect to FTL." default:"5s"`
+}
+
+func (s *schemaGenerateCmd) Run(ctx context.Context, client ftlv1connect.SchemaServiceClient) error {
+	if s.Watch == 0 {
+		return s.oneOffGenerate(ctx, client)
+	}
+	return s.hotReload(ctx, client)
+}
+
+func (s *schemaGenerateCmd) oneOffGenerate(ctx context.Context, schemaClient ftlv1connect.SchemaServiceClient) error {
+	response, err := schemaClient.GetSchema(ctx, connect.NewRequest(&ftlv1.GetSchemaRequest{}))
+	if err != nil {
+		return fmt.Errorf("failed to get schema: %w", err)
+	}
+	modules, err := slices.MapErr(response.Msg.Schema.Modules, schema.ModuleFromProto)
+	if err != nil {
+		return fmt.Errorf("invalid module schema: %w", err)
+	}
+	return s.regenerateModules(log.FromContext(ctx), modules)
+}
+
+func (s *schemaGenerateCmd) hotReload(ctx context.Context, client ftlv1connect.SchemaServiceClient) error {
+	watch := watcher.New()
+	defer watch.Close()
+
+	absTemplatePath, err := filepath.Abs(s.Template)
+	if err != nil {
+		return fmt.Errorf("failed to get absolute path for template: %w", err)
+	}
+	absDestPath, err := filepath.Abs(s.Dest)
+	if err != nil {
+		return fmt.Errorf("failed to get absolute path for destination: %w", err)
+	}
+
+	if strings.HasPrefix(absDestPath, absTemplatePath) {
+		return fmt.Errorf("destination directory %s must not be inside the template directory %s", absDestPath, absTemplatePath)
+	}
+
+	logger := log.FromContext(ctx)
+	logger.Debugf("Watching %s", s.Template)
+
+	if err := watch.AddRecursive(s.Template); err != nil {
+		return fmt.Errorf("failed to watch template directory: %w", err)
+	}
+
+	wg, ctx := errgroup.WithContext(ctx)
+
+	moduleChange := make(chan []*schema.Module)
+
+	wg.Go(func() error {
+		for {
+			stream, err := client.PullSchema(ctx, connect.NewRequest(&ftlv1.PullSchemaRequest{}))
+			if err != nil {
+				return fmt.Errorf("failed to pull schema: %w", err)
+			}
+
+			modules := map[string]*schema.Module{}
+			regenerate := false
+			for stream.Receive() {
+				msg := stream.Msg()
+				switch msg.ChangeType {
+				case ftlv1.DeploymentChangeType_DEPLOYMENT_CHANGE_TYPE_ADDED, ftlv1.DeploymentChangeType_DEPLOYMENT_CHANGE_TYPE_CHANGED:
+					if msg.Schema == nil {
+						return fmt.Errorf("schema is nil for added/changed deployment %q", msg.GetDeploymentKey())
+					}
+					module, err := schema.ModuleFromProto(msg.Schema)
+					if err != nil {
+						return fmt.Errorf("failed to convert proto to module: %w", err)
+					}
+					modules[module.Name] = module
+
+				case ftlv1.DeploymentChangeType_DEPLOYMENT_CHANGE_TYPE_REMOVED:
+					if msg.Schema == nil {
+						return fmt.Errorf("schema is nil for removed deployment %q", msg.GetDeploymentKey())
+					}
+					if msg.ModuleRemoved {
+						delete(modules, msg.Schema.Name)
+					}
+				default:
+				}
+				if !msg.More {
+					regenerate = true
+				}
+				if !regenerate {
+					continue
+				}
+
+				moduleChange <- maps.Values(modules)
+			}
+
+			stream.Close()
+			logger.Debugf("Stream disconnected, attempting to reconnect...")
+			time.Sleep(s.ReconnectDelay)
+		}
+	})
+
+	wg.Go(func() error { return watch.Start(s.Watch) })
+
+	var previousModules []*schema.Module
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("context cancelled: %w", wg.Wait())
+
+		case event := <-watch.Event:
+			logger.Debugf("Template changed (%s), regenerating modules", event.Path)
+			if err := s.regenerateModules(logger, previousModules); err != nil {
+				return fmt.Errorf("failed to regenerate modules: %w", err)
+			}
+
+		case modules := <-moduleChange:
+			previousModules = modules
+			if err := s.regenerateModules(logger, modules); err != nil {
+				return fmt.Errorf("failed to regenerate modules: %w", err)
+			}
+		}
+	}
+}
+
+func (s *schemaGenerateCmd) regenerateModules(logger *log.Logger, modules []*schema.Module) error {
+	if err := os.RemoveAll(s.Dest); err != nil {
+		return fmt.Errorf("failed to remove destination directory: %w", err)
+	}
+
+	for _, module := range modules {
+		if err := scaffolder.Scaffold(s.Template, s.Dest, module,
+			scaffolder.Extend(javascript.Extension("template.js", javascript.WithLogger(makeJSLoggerAdapter(logger)))),
+		); err != nil {
+			return fmt.Errorf("failed to scaffold module %s: %w", module.Name, err)
+		}
+	}
+	logger.Debugf("Generated %d modules in %s", len(modules), s.Dest)
+	return nil
+}
+
+func makeJSLoggerAdapter(logger *log.Logger) func(args ...any) {
+	return func(args ...any) {
+		strs := slices.Map(args, func(v any) string { return fmt.Sprintf("%v", v) })
+		level := log.Debug
+		if prefix, ok := args[0].(string); ok {
+			switch prefix {
+			case "log:":
+				level = log.Info
+			case "debug:":
+				level = log.Debug
+			case "error:":
+				level = log.Error
+			case "warn:":
+				level = log.Warn
+			}
+		}
+		logger.Log(log.Entry{
+			Level:   level,
+			Message: strings.Join(strs[1:], " "),
+		})
+	}
+}

go.mod

@@ -51,6 +51,7 @@ require (
 	github.com/opencontainers/image-spec v1.1.0
 	github.com/otiai10/copy v1.14.1
 	github.com/posener/complete v1.2.3
+	github.com/radovskyb/watcher v1.0.7
 	github.com/rs/cors v1.11.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/swaggest/jsonschema-go v0.3.72
@@ -219,9 +220,12 @@ require (
 	github.com/chzyer/readline v1.5.1
 	github.com/danieljoos/wincred v1.2.2 // indirect
 	github.com/dlclark/regexp2 v1.11.4 // indirect
+	github.com/dop251/goja v0.0.0-20241009100908-5f46f2705ca3 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0 // indirect
 	github.com/hexops/gotextdiff v1.0.3
 	github.com/jackc/pgpassfile v1.0.0 // indirect