fix(scaling): yarn install queuing + refactor

Author: devthejo

cmd/server/main.go

@@ -88,6 +88,7 @@ func main() {
 		node.WithHealthCheckWait(cfg.HealthCheckWait),
 		node.WithHealthCheckInterval(cfg.HealthCheckInterval),
 		node.WithRequestTimeout(cfg.NodeRequestTimeout),
+		node.WithYarnQueue(cfg.MaxConcurrentYarnInstalls),
 	)
 	if err != nil {
 		err = pkgerrors.WrapWithCode(err, pkgerrors.ErrorCodeInternalError, "failed to create process manager")

pkg/config/config.go

@@ -31,6 +31,9 @@ type Config struct {
 	HealthCheckWait     time.Duration `envconfig:"HEALTH_CHECK_WAIT" default:"30s" description:"Timeout for health check"`
 	HealthCheckInterval time.Duration `envconfig:"HEALTH_CHECK_INTERVAL" default:"500ms" description:"Interval for health check polling"`
 	NodeRequestTimeout  time.Duration `envconfig:"NODE_REQUEST_TIMEOUT" default:"30s" description:"Timeout for Node.js requests"`
+
+	// Yarn configuration
+	MaxConcurrentYarnInstalls int `envconfig:"MAX_CONCURRENT_YARN_INSTALLS" default:"3" description:"Maximum concurrent yarn install operations"`
 }
 
 // LoadConfig loads configuration from environment variables and returns a Config
@@ -96,5 +99,8 @@ func (c *Config) Validate() error {
 	if c.NodeRequestTimeout <= 0 {
 		return fmt.Errorf("node request timeout must be positive")
 	}
+	if c.MaxConcurrentYarnInstalls <= 0 {
+		return fmt.Errorf("max concurrent yarn installs must be positive")
+	}
 	return nil
 }

pkg/node/dependency.go

@@ -0,0 +1,58 @@
+package node
+
+import (
+	"strings"
+
+	"github.com/socialgouv/xfuncjs-server/pkg/logger"
+)
+
+// DependencyResolver handles dependency resolution operations
+type DependencyResolver struct {
+	logger logger.Logger
+}
+
+// NewDependencyResolver creates a new dependency resolver
+func NewDependencyResolver(logger logger.Logger) *DependencyResolver {
+	return &DependencyResolver{
+		logger: logger.WithField("component", "dependency-resolver"),
+	}
+}
+
+// ResolveDependencies processes and resolves dependencies from the input specification
+func (dr *DependencyResolver) ResolveDependencies(inputDependencies map[string]string, workspaceMap map[string]string, workspaceRoot string, logger logger.Logger) (map[string]interface{}, error) {
+	dependencies := make(map[string]interface{})
+
+	logger.WithField("input_dependencies", inputDependencies).
+		Info("Starting dependency processing")
+
+	// Add user-specified dependencies
+	for k, v := range inputDependencies {
+		resolvedDep := v
+
+		if strings.HasPrefix(v, "link:") {
+			// Resolve workspace package dependencies
+			if resolved, _, resolveErr := ResolveWorkspacePackage(v, workspaceRoot, workspaceMap, logger); resolveErr != nil {
+				logger.WithField("dependency", k).
+					WithField("value", v).
+					WithField("error", resolveErr.Error()).
+					Warn("Failed to resolve workspace dependency, using original value")
+			} else {
+				resolvedDep = resolved
+			}
+		}
+
+		dependencies[k] = resolvedDep
+	}
+
+	logger.WithField("total_dependencies", len(dependencies)).
+		Info("Completed dependency processing")
+
+	return dependencies, nil
+}
+
+// ValidateDependencies validates that all dependencies are properly formatted
+func (dr *DependencyResolver) ValidateDependencies(dependencies map[string]interface{}) error {
+	// Add validation logic here if needed
+	// For now, we'll just return nil as the current implementation doesn't require validation
+	return nil
+}

pkg/node/options.go

@@ -36,6 +36,18 @@ func WithRequestTimeout(timeout time.Duration) ProcessManagerOption {
 	}
 }
 
+// WithYarnQueue sets the yarn installer with queue support
+func WithYarnQueue(maxConcurrentYarnInstalls int) ProcessManagerOption {
+	return func(pm *ProcessManager) {
+		// Initialize the global yarn queue if not already initialized
+		if GetYarnQueue() == nil {
+			InitializeYarnQueue(maxConcurrentYarnInstalls, pm.logger)
+		}
+		pm.yarnInstaller = NewYarnInstaller(GetYarnQueue(), pm.logger)
+		pm.dependencyResolver = NewDependencyResolver(pm.logger)
+	}
+}
+
 // SetNodeServerPort sets the base port for the Node.js HTTP servers
 func (pm *ProcessManager) SetNodeServerPort(port int) {
 	pm.nodeServerPort = port