diachron/master/runexpress.go

package main

import (
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"
)

func runExpress(changes <-chan FileChange, numProcesses int, basePort int, pool *WorkerPool) {
	var currentProcesses []*exec.Cmd
	var mu sync.Mutex

	// Helper to start an express process on a specific port
	startExpress := func(port int) *exec.Cmd {
		listenAddr := fmt.Sprintf("127.0.0.1:%d", port)
		cmd := exec.Command("../express/run.sh", "--listen", listenAddr)
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr

		if err := cmd.Start(); err != nil {
			log.Printf("[express:%d] Failed to start: %v", port, err)
			return nil
		}

		log.Printf("[express:%d] Started (pid %d)", port, cmd.Process.Pid)

		// Monitor the process in background
		go func(p int) {
			err := cmd.Wait()
			if err != nil {
				log.Printf("[express:%d] Process exited: %v", p, err)
			} else {
				log.Printf("[express:%d] Process exited normally", p)
			}
		}(port)

		return cmd
	}

	// Helper to stop an express process
	stopExpress := func(cmd *exec.Cmd) {
		if cmd == nil || cmd.Process == nil {
			return
		}

		pid := cmd.Process.Pid
		log.Printf("[express] Stopping (pid %d)", pid)
		cmd.Process.Signal(syscall.SIGTERM)

		// Wait briefly for graceful shutdown
		done := make(chan struct{})
		go func() {
			cmd.Wait()
			close(done)
		}()

		select {
		case <-done:
			log.Printf("[express] Stopped gracefully (pid %d)", pid)
		case <-time.After(5 * time.Second):
			log.Printf("[express] Force killing (pid %d)", pid)
			cmd.Process.Kill()
		}
	}

	// Helper to stop all express processes
	stopAllExpress := func(processes []*exec.Cmd) {
		for _, cmd := range processes {
			stopExpress(cmd)
		}
	}

	// Helper to start all express processes and update the worker pool
	startAllExpress := func() []*exec.Cmd {
		processes := make([]*exec.Cmd, 0, numProcesses)
		addresses := make([]string, 0, numProcesses)
		for i := 0; i < numProcesses; i++ {
			port := basePort + i
			addr := fmt.Sprintf("127.0.0.1:%d", port)
			cmd := startExpress(port)
			if cmd != nil {
				processes = append(processes, cmd)
				addresses = append(addresses, addr)
			}
		}
		// Update the worker pool with new worker addresses
		pool.SetWorkers(addresses)
		return processes
	}

	// Helper to run the build
	runBuild := func() bool {
		log.Printf("[build] Starting ncc build...")

		cmd := exec.Command("../express/build.sh")

		stdout, _ := cmd.StdoutPipe()
		stderr, _ := cmd.StderrPipe()

		if err := cmd.Start(); err != nil {
			log.Printf("[build] Failed to start: %v", err)
			return false
		}

		// Copy output
		go io.Copy(os.Stdout, stdout)
		go io.Copy(os.Stderr, stderr)

		err := cmd.Wait()
		if err != nil {
			log.Printf("[build] Failed: %v", err)
			return false
		}

		log.Printf("[build] Success")
		return true
	}

	// Debounce timer
	var debounceTimer *time.Timer
	const debounceDelay = 100 * time.Millisecond

	// Initial build and start
	log.Printf("[master] Initial build...")
	if runBuild() {
		currentProcesses = startAllExpress()
	} else {
		log.Printf("[master] Initial build failed")
	}

	for change := range changes {
		log.Printf("[watch] %s: %s", change.Operation, change.Path)

		// Reset debounce timer
		if debounceTimer != nil {
			debounceTimer.Stop()
		}

		debounceTimer = time.AfterFunc(debounceDelay, func() {
			if !runBuild() {
				log.Printf("[master] Build failed, keeping current processes")
				return
			}

			mu.Lock()
			defer mu.Unlock()

			// Stop all old processes
			stopAllExpress(currentProcesses)

			// Start all new processes
			currentProcesses = startAllExpress()
		})
	}
}