Update transparent proxy KeyValue with gRPC port configuration

cmd/prism-admin/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/hashicorp/go-hclog v1.6.2
 	github.com/hashicorp/raft v1.7.3
 	github.com/hashicorp/raft-boltdb v0.0.0-20250926130943-f41fa5f23d89
+	github.com/jrepp/prism-data-layer/pkg/launcherclient v0.0.0
 	github.com/prometheus/client_golang v1.23.2
 	github.com/spf13/cobra v1.10.1
 	github.com/spf13/viper v1.21.0
@@ -67,3 +68,5 @@ require (
 )
 
 replace github.com/jrepp/prism-data-layer/pkg/plugin => ../../pkg/plugin
+
+replace github.com/jrepp/prism-data-layer/pkg/launcherclient => ../../pkg/launcherclient

cmd/prism-admin/serve.go

@@ -11,6 +11,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/jrepp/prism-data-layer/pkg/launcherclient"
 	pb "github.com/jrepp/prism-data-layer/pkg/plugin/gen/prism"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
@@ -51,6 +52,10 @@ func init() {
 	serveCmd.Flags().String("data-dir", "", "Raft data directory")
 	serveCmd.Flags().String("db", "", "Database URN (e.g., sqlite:///var/lib/prism-admin/admin.db)")
 
+	// Launcher integration flags
+	serveCmd.Flags().String("admin", "", "Admin/launcher endpoint (launcher://host:port or host:port)")
+	serveCmd.Flags().String("process-id", "", "Process ID (assigned by launcher)")
+
 	// Bind legacy flags
 	viper.BindPFlag("server.port", serveCmd.Flags().Lookup("port"))
 	viper.BindPFlag("server.listen", serveCmd.Flags().Lookup("listen"))
@@ -72,6 +77,80 @@ func runServe(cmd *cobra.Command, args []string) error {
 	ctx := context.Background()
 	log := slog.Default()
 
+	// Check if launched by launcher (launcher integration)
+	adminEndpoint, _ := cmd.Flags().GetString("admin")
+	processID, _ := cmd.Flags().GetString("process-id")
+	var launcherClient *launcherclient.Client
+	var actualListener net.Listener
+
+	if adminEndpoint != "" && processID != "" {
+		launcherAddr, isLauncher, err := launcherclient.ParseAdminEndpoint(adminEndpoint)
+		if err != nil {
+			return fmt.Errorf("invalid admin endpoint: %w", err)
+		}
+
+		if isLauncher {
+			fmt.Printf("[INFO] Launched by launcher, connecting to %s with process ID %s\n", launcherAddr, processID)
+
+			// Get dynamic port from OS
+			listener, actualPort, err := launcherclient.GetActualPort()
+			if err != nil {
+				return fmt.Errorf("failed to get dynamic port: %w", err)
+			}
+			actualListener = listener
+			defer listener.Close()
+
+			fmt.Printf("[INFO] Got dynamic port: %d\n", actualPort)
+
+			// Create launcher client
+			launcherClient, err = launcherclient.New(&launcherclient.Config{
+				LauncherAddr: launcherAddr,
+				ProcessID:    processID,
+				ProcessType:  "admin",
+				ActualPort:   actualPort,
+			})
+			if err != nil {
+				return fmt.Errorf("failed to create launcher client: %w", err)
+			}
+			defer launcherClient.Close()
+
+			// Connect to launcher
+			if err := launcherClient.Connect(ctx); err != nil {
+				return fmt.Errorf("failed to connect to launcher: %w", err)
+			}
+
+			// Perform handshake and discover topology
+			topology, err := launcherClient.Handshake(ctx)
+			if err != nil {
+				return fmt.Errorf("handshake failed: %w", err)
+			}
+
+			fmt.Printf("[INFO] Handshake successful!\n")
+			fmt.Printf("[INFO]   Launcher ID: %s\n", topology.LauncherID)
+			fmt.Printf("[INFO]   Admin endpoints: %v\n", topology.AdminEndpoints)
+			fmt.Printf("[INFO]   Proxy endpoints: %v\n", topology.ProxyEndpoints)
+
+			// Start heartbeat loop
+			go func() {
+				ticker := time.NewTicker(30 * time.Second)
+				defer ticker.Stop()
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-ticker.C:
+						if err := launcherClient.SendHeartbeat(ctx); err != nil {
+							log.Error("heartbeat failed", "error", err)
+						}
+					}
+				}
+			}()
+
+			// Override gRPC port with actual port
+			cmd.Flags().Set("grpc-port", fmt.Sprintf("%d", actualPort))
+		}
+	}
+
 	// Load configuration
 	clusterCfg, controlPlaneCfg, _, err := LoadConfig()
 	if err != nil {
@@ -116,9 +195,12 @@ func runServe(cmd *cobra.Command, args []string) error {
 	}
 	defer raftNode.Stop()
 
-	// Bootstrap cluster (if not already bootstrapped)
-	if err := raftNode.Bootstrap(clusterCfg.Peers); err != nil {
-		return fmt.Errorf("failed to bootstrap cluster: %w", err)
+	// Bootstrap cluster (only first node should bootstrap)
+	// Other nodes will join the cluster once the first node is bootstrapped
+	if clusterCfg.NodeID == 1 {
+		if err := raftNode.Bootstrap(clusterCfg.Peers); err != nil {
+			return fmt.Errorf("failed to bootstrap cluster: %w", err)
+		}
 	}
 
 	// Wait for leader election
@@ -150,9 +232,18 @@ func runServe(cmd *cobra.Command, args []string) error {
 	// Start gRPC server
 	address := controlPlaneCfg.Listen
 
-	lis, err := net.Listen("tcp", address)
-	if err != nil {
-		return fmt.Errorf("failed to listen on %s: %w", address, err)
+	var lis net.Listener
+	if actualListener != nil {
+		// Use the listener from launcher client (dynamic port)
+		lis = actualListener
+		fmt.Printf("[INFO] Using launcher-assigned port: %s\n", lis.Addr().String())
+	} else {
+		// Standard mode - listen on configured port
+		var err error
+		lis, err = net.Listen("tcp", address)
+		if err != nil {
+			return fmt.Errorf("failed to listen on %s: %w", address, err)
+		}
 	}
 
 	grpcServer := grpc.NewServer()

cmd/pattern-launcher/main.go → cmd/prism-launcher/main.go

@@ -99,9 +99,13 @@ func main() {
 		log.Printf("No admin endpoint configured, running standalone")
 	}
 
+	// Create control service for process callbacks
+	controlService := launcher.NewControlService(*launcherID, adminClient)
+
 	// Create gRPC server
 	grpcServer := grpc.NewServer()
 	pb.RegisterPatternLauncherServer(grpcServer, service)
+	pb.RegisterLauncherControlServer(grpcServer, controlService)
 
 	// Enable reflection for grpcurl
 	reflection.Register(grpcServer)