new-api/test/e2e/helpers/cluster.go

package helpers

import (
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strings"
	"time"
)

const (
	NodeTypeMaster = "master"
	NodeTypeSlave  = "slave"

	DefaultCNPort          = 3100
	DefaultOVPort          = 3101
	DefaultCNDBFileName    = "cn.db"
	DefaultOVDBFileName    = "ov.db"
	DefaultCNLogFileName   = "cn.log"
	DefaultOVLogFileName   = "ov.log"
	DefaultCNSessionName   = "cn_session"
	DefaultOVSessionName   = "ov_session"
	DefaultRootUsername    = "root"
	defaultNodeReadyTimout = 45 * time.Second
)

// The e2e cluster runs on loopback only; each secret below resolves from the
// environment first and falls back to a local-only value shared with the
// Playwright fixtures (see web/e2e/fixtures/cluster.ts).
func RootPassword() string {
	if password := strings.TrimSpace(os.Getenv("E2E_ROOT_PASSWORD")); password != "" {
		return password
	}
	return "e2e-local-root-pass"
}

func SessionSecret() string {
	if secret := strings.TrimSpace(os.Getenv("E2E_SESSION_SECRET")); secret != "" {
		return secret
	}
	return "e2e-local-session-secret"
}

func SyncAPIKey() string {
	if key := strings.TrimSpace(os.Getenv("E2E_SYNC_API_KEY")); key != "" {
		return key
	}
	return "e2e-local-sync-key"
}

type NodeProcess struct {
	Name        string
	Port        int
	BaseURL     string
	DBPath      string
	SessionName string
	NodeType    string
	LogPath     string

	cmd     *exec.Cmd
	logFile *os.File
	waitCh  chan error
}

type Cluster struct {
	RepoRoot   string
	RunDir     string
	BinaryPath string
	CNDBPath   string
	OVDBPath   string
	CN         *NodeProcess
	OV         *NodeProcess
}

type Environment struct {
	RepoRoot     string
	ArtifactRoot string
	RunDir       string
	BinaryPath   string
	CNDBPath     string
	OVDBPath     string
	Cluster      *Cluster
	CNClient     *APIClient
	OVClient     *APIClient
}

func BuildBinary(repoRoot, artifactRoot string) (string, error) {
	binDir := filepath.Join(artifactRoot, "bin")
	if err := os.MkdirAll(binDir, 0o755); err != nil {
		return "", fmt.Errorf("create binary dir: %w", err)
	}

	binaryName := "new-api-e2e"
	if runtime.GOOS == "windows" {
		binaryName += ".exe"
	}
	binaryPath := filepath.Join(binDir, binaryName)

	cmd := exec.Command("go", "build", "-o", binaryPath, ".")
	cmd.Dir = repoRoot
	output, err := cmd.CombinedOutput()
	if err != nil {
		return "", fmt.Errorf("build e2e binary: %w\n%s", err, string(output))
	}
	return binaryPath, nil
}

func PrepareEnvironment(repoRoot, artifactRoot, binaryPath, scenarioName string) (*Environment, error) {
	runDir := filepath.Join(artifactRoot, "runs", sanitizeName(scenarioName))
	if err := os.RemoveAll(runDir); err != nil {
		return nil, fmt.Errorf("reset run dir: %w", err)
	}
	if err := os.MkdirAll(runDir, 0o755); err != nil {
		return nil, fmt.Errorf("create run dir: %w", err)
	}

	cluster := NewCluster(repoRoot, runDir, binaryPath)
	if err := cluster.Start(NodeTypeMaster, NodeTypeMaster); err != nil {
		_ = cluster.Stop()
		return nil, err
	}

	cnClient, err := NewAPIClient(cluster.CN.BaseURL)
	if err != nil {
		_ = cluster.Stop()
		return nil, err
	}
	ovClient, err := NewAPIClient(cluster.OV.BaseURL)
	if err != nil {
		_ = cluster.Stop()
		return nil, err
	}

	if err := SetupRoot(cnClient, DefaultRootUsername, RootPassword()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("setup CN root: %w", err)
	}
	if err := SetupRoot(ovClient, DefaultRootUsername, RootPassword()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("setup OV root: %w", err)
	}

	if err := cluster.Stop(); err != nil {
		return nil, fmt.Errorf("stop bootstrap cluster: %w", err)
	}

	if err := SeedCNData(cluster.CNDBPath); err != nil {
		return nil, fmt.Errorf("seed CN data: %w", err)
	}
	if err := SeedOVData(cluster.OVDBPath); err != nil {
		return nil, fmt.Errorf("seed OV data: %w", err)
	}

	if err := cluster.Start(NodeTypeMaster, NodeTypeSlave); err != nil {
		_ = cluster.Stop()
		return nil, err
	}

	cnClient, err = NewAPIClient(cluster.CN.BaseURL)
	if err != nil {
		_ = cluster.Stop()
		return nil, err
	}
	ovClient, err = NewAPIClient(cluster.OV.BaseURL)
	if err != nil {
		_ = cluster.Stop()
		return nil, err
	}

	if err := Login(cnClient, DefaultRootUsername, RootPassword()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("login CN root: %w", err)
	}
	if err := Login(ovClient, DefaultRootUsername, RootPassword()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("login OV root: %w", err)
	}

	if err := ConfigureCNRegionSync(cnClient, cluster.CN.BaseURL, cluster.OV.BaseURL, SyncAPIKey()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("configure CN region sync: %w", err)
	}
	if err := ConfigureOVRegionSync(ovClient, cluster.CN.BaseURL, SyncAPIKey()); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("configure OV region sync: %w", err)
	}

	if err := WaitOptionApplied(cluster.OV.BaseURL, SyncAPIKey(), 5, 20*time.Second); err != nil {
		_ = cluster.Stop()
		return nil, fmt.Errorf("wait OV migration API ready: %w", err)
	}

	return &Environment{
		RepoRoot:     repoRoot,
		ArtifactRoot: artifactRoot,
		RunDir:       runDir,
		BinaryPath:   binaryPath,
		CNDBPath:     cluster.CNDBPath,
		OVDBPath:     cluster.OVDBPath,
		Cluster:      cluster,
		CNClient:     cnClient,
		OVClient:     ovClient,
	}, nil
}

func (e *Environment) Close() error {
	if e == nil || e.Cluster == nil {
		return nil
	}
	return e.Cluster.Stop()
}

func NewCluster(repoRoot, runDir, binaryPath string) *Cluster {
	return &Cluster{
		RepoRoot:   repoRoot,
		RunDir:     runDir,
		BinaryPath: binaryPath,
		CNDBPath:   filepath.Join(runDir, DefaultCNDBFileName),
		OVDBPath:   filepath.Join(runDir, DefaultOVDBFileName),
	}
}

func (c *Cluster) Start(cnNodeType, ovNodeType string) error {
	if err := os.MkdirAll(c.RunDir, 0o755); err != nil {
		return fmt.Errorf("create run dir: %w", err)
	}

	cn, err := c.startNode("cn", DefaultCNPort, c.CNDBPath, DefaultCNSessionName, cnNodeType, DefaultCNLogFileName)
	if err != nil {
		return err
	}
	c.CN = cn

	ov, err := c.startNode("ov", DefaultOVPort, c.OVDBPath, DefaultOVSessionName, ovNodeType, DefaultOVLogFileName)
	if err != nil {
		_ = c.CN.Stop()
		c.CN = nil
		return err
	}
	c.OV = ov

	if err := c.CN.WaitReady(defaultNodeReadyTimout); err != nil {
		_ = c.Stop()
		return err
	}
	if err := c.OV.WaitReady(defaultNodeReadyTimout); err != nil {
		_ = c.Stop()
		return err
	}
	return nil
}

func (c *Cluster) Stop() error {
	var errs []string
	if c.OV != nil {
		if err := c.OV.Stop(); err != nil {
			errs = append(errs, err.Error())
		}
		c.OV = nil
	}
	if c.CN != nil {
		if err := c.CN.Stop(); err != nil {
			errs = append(errs, err.Error())
		}
		c.CN = nil
	}
	if len(errs) == 0 {
		return nil
	}
	return errors.New(strings.Join(errs, "; "))
}

func (c *Cluster) startNode(name string, port int, dbPath, sessionName, nodeType, logFileName string) (*NodeProcess, error) {
	logPath := filepath.Join(c.RunDir, logFileName)
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		return nil, fmt.Errorf("open %s log file: %w", name, err)
	}

	systemLogDir := filepath.Join(c.RunDir, "system-logs", name)
	if err := os.MkdirAll(systemLogDir, 0o755); err != nil {
		_ = logFile.Close()
		return nil, fmt.Errorf("create %s system log dir: %w", name, err)
	}

	cmd := exec.Command(c.BinaryPath, "--log-dir", systemLogDir)
	cmd.Dir = c.RepoRoot
	cmd.Env = mergeEnv(os.Environ(), map[string]string{
		"PORT":                         fmt.Sprintf("%d", port),
		"SQLITE_PATH":                  dbPath,
		"SESSION_SECRET":               SessionSecret(),
		"SESSION_NAME":                 sessionName,
		"SESSION_SECURE":               "false",
		"SESSION_SAMESITE":             "strict",
		"GLOBAL_API_RATE_LIMIT_ENABLE": "false",
		"GLOBAL_WEB_RATE_LIMIT_ENABLE": "false",
		"CRITICAL_RATE_LIMIT_ENABLE":   "false",
		"GIN_MODE":                     "release",
		"NODE_TYPE":                    nodeType,
	})
	cmd.Stdout = logFile
	cmd.Stderr = logFile

	waitCh := make(chan error, 1)
	if err := cmd.Start(); err != nil {
		_ = logFile.Close()
		return nil, fmt.Errorf("start %s node: %w", name, err)
	}
	go func() {
		waitCh <- cmd.Wait()
	}()

	return &NodeProcess{
		Name:        name,
		Port:        port,
		BaseURL:     fmt.Sprintf("http://127.0.0.1:%d", port),
		DBPath:      dbPath,
		SessionName: sessionName,
		NodeType:    nodeType,
		LogPath:     logPath,
		cmd:         cmd,
		logFile:     logFile,
		waitCh:      waitCh,
	}, nil
}

func (n *NodeProcess) WaitReady(timeout time.Duration) error {
	client := &http.Client{Timeout: 1 * time.Second}
	deadline := time.Now().Add(timeout)

	for time.Now().Before(deadline) {
		select {
		case err := <-n.waitCh:
			if err == nil {
				return fmt.Errorf("%s node exited before ready; see %s", n.Name, n.LogPath)
			}
			return fmt.Errorf("%s node exited before ready: %w; see %s", n.Name, err, n.LogPath)
		default:
		}

		resp, err := client.Get(n.BaseURL + "/api/status")
		if err == nil {
			_, _ = io.Copy(io.Discard, resp.Body)
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				return nil
			}
		}
		time.Sleep(250 * time.Millisecond)
	}

	return fmt.Errorf("%s node did not become ready within %s; see %s", n.Name, timeout, n.LogPath)
}

func (n *NodeProcess) Stop() error {
	if n == nil {
		return nil
	}

	var errs []string
	killRequested := false
	if n.cmd != nil && n.cmd.Process != nil {
		if err := n.cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
			errs = append(errs, fmt.Sprintf("kill %s node: %v", n.Name, err))
		} else {
			killRequested = true
		}
		select {
		case err := <-n.waitCh:
			if err != nil && !killRequested && !strings.Contains(strings.ToLower(err.Error()), "signal: killed") {
				errs = append(errs, fmt.Sprintf("wait %s node: %v", n.Name, err))
			}
		case <-time.After(5 * time.Second):
			errs = append(errs, fmt.Sprintf("wait %s node timeout", n.Name))
		}
	}

	if n.logFile != nil {
		if err := n.logFile.Close(); err != nil {
			errs = append(errs, fmt.Sprintf("close %s log: %v", n.Name, err))
		}
		n.logFile = nil
	}
	n.cmd = nil

	if len(errs) == 0 {
		return nil
	}
	return errors.New(strings.Join(errs, "; "))
}

func sanitizeName(name string) string {
	replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_")
	return replacer.Replace(name)
}

func mergeEnv(base []string, overrides map[string]string) []string {
	filtered := make([]string, 0, len(base)+len(overrides))
	for _, entry := range base {
		parts := strings.SplitN(entry, "=", 2)
		if len(parts) != 2 {
			continue
		}
		if _, ok := overrides[parts[0]]; ok {
			continue
		}
		filtered = append(filtered, entry)
	}
	for key, value := range overrides {
		filtered = append(filtered, key+"="+value)
	}
	return filtered
}