fix: move generic schema writing to the JVM plugin

common/schema/schema.go

@@ -215,3 +215,43 @@ func FromProto(s *schemapb.Schema) (*Schema, error) {
 	}
 	return ValidateSchema(schema)
 }
+
+// ModuleDependencies returns the modules that the given module depends on
+// Dependency modules are the ones that are called by the given module, or that publish topics that the given module subscribes to
+func (s *Schema) ModuleDependencies(module string) map[string]*Module {
+	mods := map[string]*Module{}
+	for _, sch := range s.Modules {
+		mods[sch.Name] = sch
+	}
+	deps := make(map[string]*Module)
+	toProcess := []string{module}
+	for len(toProcess) > 0 {
+		dep := toProcess[0]
+		toProcess = toProcess[1:]
+		if deps[dep] != nil {
+			continue
+		}
+		dm := mods[dep]
+		deps[dep] = dm
+		for _, m := range dm.Decls {
+			if ref, ok := m.(*Verb); ok {
+				for _, ref := range ref.Metadata {
+					switch md := ref.(type) {
+					case *MetadataCalls:
+						for _, calls := range md.Calls {
+							if calls.Module != "" {
+								toProcess = append(toProcess, calls.Module)
+							}
+						}
+					case *MetadataSubscriber:
+						if md.Topic.Module != "" {
+							toProcess = append(toProcess, md.Topic.Module)
+						}
+					}
+				}
+			}
+		}
+	}
+	delete(deps, module)
+	return deps
+}

frontend/cli/cmd_new.go

@@ -67,7 +67,7 @@ func prepareNewCmd(ctx context.Context, k *kong.Kong, args []string) (optionalPl
 		return optionalPlugin, fmt.Errorf("could not load project config: %w", err)
 	}
 
-	plugin, err := languageplugin.New(ctx, filepath.Dir(projConfigPath), language, "new", false)
+	plugin, err := languageplugin.New(ctx, filepath.Dir(projConfigPath), language, "new")
 
 	if err != nil {
 		return optionalPlugin, fmt.Errorf("could not create plugin for %v: %w", language, err)

internal/buildengine/engine.go

@@ -1051,7 +1051,7 @@ func (e *Engine) gatherSchemas(
 }
 
 func (e *Engine) newModuleMeta(ctx context.Context, config moduleconfig.UnvalidatedModuleConfig) (moduleMeta, error) {
-	plugin, err := languageplugin.New(ctx, config.Dir, config.Language, config.Module, e.devMode)
+	plugin, err := languageplugin.New(ctx, config.Dir, config.Language, config.Module)
 	if err != nil {
 		return moduleMeta{}, fmt.Errorf("could not create plugin for %s: %w", config.Module, err)
 	}

internal/buildengine/languageplugin/plugin.go

@@ -96,8 +96,8 @@ type BuildContext struct {
 
 var ErrPluginNotRunning = errors.New("language plugin no longer running")
 
-// PluginFromConfig creates a new language plugin from the given config.
-func New(ctx context.Context, dir, language, name string, devMode bool) (p *LanguagePlugin, err error) {
+// New creates a new language plugin from the given config.
+func New(ctx context.Context, dir, language, name string) (p *LanguagePlugin, err error) {
 	impl, err := newClientImpl(ctx, dir, language, name)
 	if err != nil {
 		return nil, err

internal/buildengine/stubs.go

@@ -24,7 +24,7 @@ func GenerateStubs(ctx context.Context, projectRoot string, modules []*schema.Mo
 	if err != nil {
 		return err
 	}
-	return writeGenericSchemaFiles(modules, metas)
+	return nil
 }
 
 // CleanStubs removes all generated stubs.
@@ -107,34 +107,3 @@ func generateStubsForEachLanguage(ctx context.Context, projectRoot string, modul
 	}
 	return nil
 }
-
-func writeGenericSchemaFiles(modules []*schema.Module, metas map[string]moduleMeta) error {
-	sch := &schema.Schema{Modules: modules}
-	for _, meta := range metas {
-		module := meta.module.Config
-		if module.GeneratedSchemaDir == "" {
-			continue
-		}
-
-		modPath := module.Abs().GeneratedSchemaDir
-		err := os.MkdirAll(modPath, 0750)
-		if err != nil {
-			return fmt.Errorf("failed to create directory %s: %w", modPath, err)
-		}
-
-		for _, mod := range sch.Modules {
-			if mod.Name == module.Module {
-				continue
-			}
-			data, err := schema.ModuleToBytes(mod)
-			if err != nil {
-				return fmt.Errorf("failed to export module schema for module %s %w", mod.Name, err)
-			}
-			err = os.WriteFile(filepath.Join(modPath, mod.Name+".pb"), data, 0600)
-			if err != nil {
-				return fmt.Errorf("failed to write schema file for module %s %w", mod.Name, err)
-			}
-		}
-	}
-	return nil
-}

jvm-runtime/plugin/common/java_plugin_test.go

@@ -75,7 +75,7 @@ func TestJavaConfigDefaults(t *testing.T) {
 			dir, err := filepath.Abs(tt.dir)
 			assert.NoError(t, err)
 
-			plugin, err := languageplugin.New(ctx, t.TempDir(), "java", "test", false)
+			plugin, err := languageplugin.New(ctx, t.TempDir(), "java", "test")
 			assert.NoError(t, err)
 			t.Cleanup(func() {
 				_ = plugin.Kill() //nolint:errcheck

jvm-runtime/plugin/common/jvmcommon.go

@@ -28,6 +28,7 @@ import (
 	langpb "github.com/block/ftl/backend/protos/xyz/block/ftl/language/v1"
 	langconnect "github.com/block/ftl/backend/protos/xyz/block/ftl/language/v1/languagepbconnect"
 	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/builderrors"
 	"github.com/block/ftl/common/errors"
 	"github.com/block/ftl/common/plugin"
@@ -40,6 +41,8 @@ import (
 	"github.com/block/ftl/internal/flock"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/moduleconfig"
+	"github.com/block/ftl/internal/rpc"
+	"github.com/block/ftl/internal/schema/schemaeventsource"
 	"github.com/block/ftl/internal/watch"
 )
 
@@ -166,6 +169,10 @@ func (s *Service) Build(ctx context.Context, req *connect.Request[langpb.BuildRe
 	if err != nil {
 		return err
 	}
+	err = s.writeGenericSchemaFiles(buildCtx)
+	if err != nil {
+		return err
+	}
 	if req.Msg.RebuildAutomatically {
 		return s.runDevMode(ctx, req, buildCtx, stream)
 	}
@@ -296,6 +303,8 @@ func (s *Service) runQuarkusDev(ctx context.Context, req *connect.Request[langpb
 	debugPort, err := plugin.AllocatePort()
 	debugPort32 := int32(debugPort.Port)
 
+	eventSource := schemaeventsource.New(ctx, rpc.Dial(ftlv1connect.NewSchemaServiceClient, os.Getenv("FTL_BIND"), log.Debug))
+
 	if err == nil {
 		devModeBuild = fmt.Sprintf("%s -Ddebug=%d", devModeBuild, debugPort.Port)
 	}
@@ -334,6 +343,28 @@ func (s *Service) runQuarkusDev(ctx context.Context, req *connect.Request[langpb
 				}
 			}
 			return nil
+		case event := <-eventSource.Events():
+			// We want to write up-to-date versions of any schema files
+			// So that our code generation is up-to-date
+			view := eventSource.View()
+			// Iterate over the changed modules
+			for _, module := range event.Schema().Modules {
+				deps := view.ModuleDependencies(module.Name)
+				// We don't want to generate a schema if we are a dependency of this module
+				// This could make it easy to add circular dependencies
+				if _, ok := deps[buildCtx.Config.Module]; !ok {
+					bytes, err := schema.ModuleToBytes(module)
+					if err != nil {
+						logger.Errorf(err, "failed to write schema for module: %s", module.Name)
+						continue
+					}
+					err = os.WriteFile(filepath.Join(buildCtx.Config.GeneratedSchemaDir, module.Name+".pb"), bytes, 0600)
+					if err != nil {
+						logger.Errorf(err, "failed to write schema for module: %s", module.Name)
+						continue
+					}
+				}
+			}
 		case bc := <-events:
 			buildCtx = bc.buildCtx
 		case <-schemaChangeTicker.C:
@@ -519,6 +550,10 @@ func (s *Service) BuildContextUpdated(ctx context.Context, req *connect.Request[
 	if err != nil {
 		return nil, err
 	}
+	err = s.writeGenericSchemaFiles(buildCtx)
+	if err != nil {
+		return nil, err
+	}
 
 	s.updatesTopic.Publish(buildContextUpdatedEvent{
 		buildCtx: buildCtx,
@@ -795,3 +830,28 @@ func loadProtoErrors(config moduleconfig.AbsModuleConfig) (*langpb.ErrorList, er
 func ptr(s string) *string {
 	return &s
 }
+
+func (s *Service) writeGenericSchemaFiles(buildContext buildContext) error {
+
+	modPath := buildContext.Config.GeneratedSchemaDir
+	err := os.MkdirAll(modPath, 0750)
+	if err != nil {
+		return fmt.Errorf("failed to create directory %s: %w", modPath, err)
+	}
+
+	for _, mod := range buildContext.Schema.Modules {
+		if mod.Name == buildContext.Config.Module {
+			continue
+		}
+		data, err := schema.ModuleToBytes(mod)
+		if err != nil {
+			return fmt.Errorf("failed to export module schema for module %s %w", mod.Name, err)
+		}
+		err = os.WriteFile(filepath.Join(modPath, mod.Name+".pb"), data, 0600)
+		if err != nil {
+			return fmt.Errorf("failed to write schema file for module %s %w", mod.Name, err)
+		}
+	}
+
+	return nil
+}