cli/pkg/interchain/relayer/relayer.go

// Copyright (C) 2025, Lux Industries Inc. All rights reserved
// See the file LICENSE for licensing terms.
package relayer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strings"
	"syscall"
	"time"

	"github.com/luxfi/cli/v2/pkg/application"
	"github.com/luxfi/cli/v2/pkg/binutils"
	"github.com/luxfi/cli/v2/pkg/constants"
	"github.com/luxfi/cli/v2/pkg/key"
	"github.com/luxfi/cli/v2/pkg/models"
	"github.com/luxfi/cli/v2/pkg/utils"
	"github.com/luxfi/cli/v2/pkg/ux"
	"github.com/luxfi/cli/v2/sdk/evm"
	// TODO: Fix warp imports
	// apiConfig "github.com/luxfi/warp/config"
	// offchainregistry "github.com/luxfi/warp/messages/off-chain-registry"
	// "github.com/luxfi/warp/relayer/config"
)

const (
	localRelayerSetupTime                  = 2 * time.Second
	localRelayerCheckPoolTime              = 100 * time.Millisecond
	localRelayerCheckTimeout               = 3 * time.Second
	defaultDBWriteIntervalSeconds          = 10
	defaultSignatureCacheSize              = 1024 * 1024
	defaultInitialConnectionTimeoutSeconds = 60
)

var relayerRequiredBalance = big.NewInt(0).Mul(big.NewInt(1e18), big.NewInt(500)) // 500 LUX

func GetDefaultRelayerKeyInfo(app *application.Lux) (string, string, string, error) {
	keyName := constants.WarpRelayerKeyName
	keyPath := app.GetKeyPath(keyName)
	var (
		k   *key.SoftKey
		err error
	)
	if utils.FileExists(keyPath) {
		k, err = key.LoadSoft(constants.LocalNetworkID, keyPath)
		if err != nil {
			return "", "", "", err
		}
	} else {
		k, err = key.NewSoft(0)
		if err != nil {
			return "", "", "", err
		}
		if err := k.Save(keyPath); err != nil {
			return "", "", "", err
		}
	}
	return keyName, k.C(), k.PrivKeyHex(), nil
}

func FundRelayer(
	rpcURL string,
	prefundedPrivateKey string,
	relayerAddress string,
) error {
	// get relayer balance
	client, err := evm.GetClient(rpcURL)
	if err != nil {
		return err
	}
	relayerBalance, err := client.GetAddressBalance(relayerAddress)
	if err != nil {
		return err
	}
	if relayerBalance.Cmp(relayerRequiredBalance) < 0 {
		toFund := big.NewInt(0).Sub(relayerRequiredBalance, relayerBalance)
		_, err := client.FundAddress(
			prefundedPrivateKey,
			relayerAddress,
			toFund,
		)
		if err != nil {
			return err
		}
	}
	return nil
}

type relayerRunFile struct {
	Pid int `json:"pid"`
}

func DeployRelayer(
	version string,
	binPath string,
	binDir string,
	configPath string,
	logFilePath string,
	runFilePath string,
	storageDir string,
) (string, error) {
	if err := RelayerCleanup(runFilePath, logFilePath, storageDir); err != nil {
		return "", err
	}
	if binPath == "" {
		var err error
		binPath, err = InstallRelayer(binDir, version)
		if err != nil {
			return "", err
		}
	}
	pid, err := executeRelayer(binPath, configPath, logFilePath)
	if err != nil {
		return "", err
	}
	return binPath, saveRelayerRunFile(runFilePath, pid)
}

func RelayerIsUp(runFilePath string) (bool, int, *os.Process, error) {
	if !utils.FileExists(runFilePath) {
		return false, 0, nil, nil
	}
	bs, err := os.ReadFile(runFilePath)
	if err != nil {
		return false, 0, nil, err
	}
	rf := relayerRunFile{}
	if err := json.Unmarshal(bs, &rf); err != nil {
		return false, 0, nil, err
	}
	proc, err := utils.GetProcess(rf.Pid)
	if err != nil {
		// after a reboot without network cleanup, it is expected that the file pid will exist but the process not
		return false, 0, nil, removeRelayerRunFile(runFilePath)
	}
	return true, rf.Pid, proc, nil
}

func RelayerCleanup(
	runFilePath string,
	logFilePath string,
	storageDir string,
) error {
	_ = os.Remove(logFilePath)
	if err := os.RemoveAll(storageDir); err != nil {
		return err
	}
	relayerIsUp, pid, proc, err := RelayerIsUp(runFilePath)
	if err != nil {
		return err
	}
	if relayerIsUp {
		waitedCh := make(chan struct{})
		go func() {
			for {
				if err := proc.Signal(syscall.Signal(0)); err != nil {
					if errors.Is(err, os.ErrProcessDone) {
						close(waitedCh)
						return
					} else {
						ux.Logger.PrintToUser("failure checking to process pid %d aliveness due to: %s", proc.Pid, err)
					}
				}
				time.Sleep(localRelayerCheckPoolTime)
			}
		}()
		if err := proc.Signal(os.Interrupt); err != nil {
			return fmt.Errorf("failed sending interrupt signal to relayer process with pid %d: %w", pid, err)
		}
		ctx, cancel := context.WithTimeout(context.Background(), localRelayerCheckTimeout)
		defer cancel()
		select {
		case <-ctx.Done():
			if err := proc.Signal(os.Kill); err != nil {
				return fmt.Errorf("failed killing relayer process with pid %d: %w", pid, err)
			}
		case <-waitedCh:
		}
		return removeRelayerRunFile(runFilePath)
	}
	return nil
}

func removeRelayerRunFile(runFilePath string) error {
	err := os.Remove(runFilePath)
	if err != nil {
		err = fmt.Errorf("failed removing relayer run file %s: %w", runFilePath, err)
	}
	return err
}

func saveRelayerRunFile(runFilePath string, pid int) error {
	rf := relayerRunFile{
		Pid: pid,
	}
	bs, err := json.Marshal(&rf)
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(runFilePath), constants.DefaultPerms755); err != nil {
		return err
	}
	if err := os.WriteFile(runFilePath, bs, constants.WriteReadReadPerms); err != nil {
		return fmt.Errorf("could not write awm relater run file to %s: %w", runFilePath, err)
	}
	return nil
}

func GetLatestRelayerReleaseVersion() (string, error) {
	downloader := application.NewDownloader()
	releaseURL := fmt.Sprintf("https://api.github.com/repos/%s/%s/releases/latest",
		constants.LuxOrg,
		constants.WarpServicesRepoName,
	)
	return downloader.GetLatestReleaseVersion(releaseURL)
}

func GetLatestRelayerPreReleaseVersion() (string, error) {
	downloader := application.NewDownloader()
	// Get all releases and find the latest prerelease
	allReleases, err := downloader.GetAllReleasesForRepo(
		constants.LuxOrg,
		constants.WarpServicesRepoName,
	)
	if err != nil {
		return "", err
	}
	if len(allReleases) == 0 {
		return "", fmt.Errorf("no releases found")
	}
	// Return the first one (most recent)
	return allReleases[0], nil
}

func InstallRelayer(binDir, version string) (string, error) {
	if version == "" || version == constants.LatestPreReleaseVersionTag {
		var err error
		version, err = GetLatestRelayerPreReleaseVersion()
		if err != nil {
			return "", err
		}
	}
	if version == constants.LatestReleaseVersionTag {
		var err error
		version, err = GetLatestRelayerReleaseVersion()
		if err != nil {
			return "", err
		}
	}
	ux.Logger.PrintToUser("Relayer version %s", version)
	versionBinDir := filepath.Join(binDir, version)
	binPath := filepath.Join(versionBinDir, constants.WarpRelayerBin)
	if utils.IsExecutable(binPath) {
		return binPath, nil
	}
	ux.Logger.PrintToUser("Installing Relayer")
	url, err := getRelayerURL(version)
	if err != nil {
		return "", err
	}
	bs, err := application.NewDownloader().Download(url)
	if err != nil {
		return "", err
	}
	if err := binutils.InstallArchive("tar.gz", bs, versionBinDir); err != nil {
		return "", err
	}
	return binPath, nil
}

func executeRelayer(binPath string, configPath string, logFile string) (int, error) {
	if err := os.MkdirAll(filepath.Dir(logFile), constants.DefaultPerms755); err != nil {
		return 0, err
	}
	logWriter, err := os.Create(logFile)
	if err != nil {
		return 0, err
	}

	ux.Logger.PrintToUser("Executing Relayer")

	cmd := exec.Command(binPath, "--config-file", configPath)
	cmd.Stdout = logWriter
	cmd.Stderr = logWriter
	if err := cmd.Start(); err != nil {
		return 0, err
	}

	ch := make(chan struct{})
	go func() {
		_ = cmd.Wait()
		ch <- struct{}{}
	}()
	time.Sleep(localRelayerSetupTime)
	select {
	case <-ch:
		return 0, fmt.Errorf("relayer process failed during setup")
	default:
	}

	err = waitForRelayerInitialization(
		configPath,
		logFile,
		0,
		0,
	)

	return cmd.Process.Pid, err
}

func getRelayerURL(version string) (string, error) {
	goarch, goos := runtime.GOARCH, runtime.GOOS
	if goos != "linux" && goos != "darwin" {
		return "", fmt.Errorf("OS not supported: %s", goos)
	}
	component := "warp-relayer"
	semanticVersion := strings.TrimPrefix(version, component+"/")
	if semanticVersion != version {
		return fmt.Sprintf(
			"https://github.com/%s/%s/releases/download/warp-relayer%%2F%s/warp-relayer_%s_%s_%s.tar.gz",
			constants.LuxOrg,
			constants.WarpServicesRepoName,
			semanticVersion,
			strings.TrimPrefix(semanticVersion, "v"),
			goos,
			goarch,
		), nil
	}
	semanticVersion = strings.TrimPrefix(version, component+"-")
	if semanticVersion != version {
		return fmt.Sprintf(
			"https://github.com/%s/%s/releases/download/warp-relayer-%s/warp-relayer_%s_%s_%s.tar.gz",
			constants.LuxOrg,
			constants.WarpServicesRepoName,
			semanticVersion,
			strings.TrimPrefix(semanticVersion, "v"),
			goos,
			goarch,
		), nil
	}
	return fmt.Sprintf(
		"https://github.com/%s/%s/releases/download/%s/warp-relayer_%s_%s_%s.tar.gz",
		constants.LuxOrg,
		constants.WarpServicesRepoName,
		semanticVersion,
		strings.TrimPrefix(semanticVersion, "v"),
		goos,
		goarch,
	), nil
}

func loadRelayerConfig(relayerConfigPath string) (*Config, error) {
	awmRelayerConfig := Config{}
	bs, err := os.ReadFile(relayerConfigPath)
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal(bs, &awmRelayerConfig); err != nil {
		return nil, err
	}
	return &awmRelayerConfig, nil
}

func saveRelayerConfig(relayerConfig *Config, relayerConfigPath string) error {
	if err := os.MkdirAll(filepath.Dir(relayerConfigPath), constants.DefaultPerms755); err != nil {
		return err
	}
	bs, err := json.MarshalIndent(relayerConfig, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(relayerConfigPath, bs, constants.WriteReadReadPerms)
}

func CreateBaseRelayerConfigIfMissing(
	relayerConfigPath string,
	logLevel string,
	storageLocation string,
	metricsPort uint16,
	network models.Network,
	allowPrivateIPs bool,
) error {
	if !utils.FileExists(relayerConfigPath) {
		return CreateBaseRelayerConfig(
			relayerConfigPath,
			logLevel,
			storageLocation,
			metricsPort,
			network,
			allowPrivateIPs,
		)
	}
	return nil
}

func CreateBaseRelayerConfig(
	relayerConfigPath string,
	logLevel string,
	storageLocation string,
	metricsPort uint16,
	network models.Network,
	allowPrivateIPs bool,
) error {
	// Get the appropriate endpoint based on network type
	var endpoint string
	switch network {
	case models.Mainnet:
		endpoint = constants.MainnetAPIEndpoint
	case models.Testnet:
		endpoint = constants.TestnetAPIEndpoint
	case models.Local:
		endpoint = constants.LocalAPIEndpoint
	default:
		return fmt.Errorf("unsupported network: %s", network)
	}
	
	awmRelayerConfig := &Config{
		LogLevel: logLevel,
		PChainAPI: &APIConfig{
			BaseURL:     endpoint,
			QueryParams: map[string]string{},
		},
		InfoAPI: &APIConfig{
			BaseURL:     endpoint,
			QueryParams: map[string]string{},
		},
		StorageLocation:                 storageLocation,
		ProcessMissedBlocks:             false,
		SourceBlockchains:               []*SourceBlockchain{},
		DestinationBlockchains:          []*DestinationBlockchain{},
		MetricsPort:                     metricsPort,
		DBWriteIntervalSeconds:          defaultDBWriteIntervalSeconds,
		SignatureCacheSize:              defaultSignatureCacheSize,
		AllowPrivateIPs:                 allowPrivateIPs,
		InitialConnectionTimeoutSeconds: defaultInitialConnectionTimeoutSeconds,
	}
	return saveRelayerConfig(awmRelayerConfig, relayerConfigPath)
}

func AddSourceAndDestinationToRelayerConfig(
	relayerConfigPath string,
	rpcEndpoint string,
	wsEndpoint string,
	subnetID string,
	blockchainID string,
	warpRegistryAddress string,
	warpMessengerAddress string,
	relayerRewardAddress string,
	relayerPrivateKey string,
) error {
	awmRelayerConfig, err := loadRelayerConfig(relayerConfigPath)
	if err != nil {
		return err
	}
	addSourceToRelayerConfig(
		awmRelayerConfig,
		rpcEndpoint,
		wsEndpoint,
		subnetID,
		blockchainID,
		warpRegistryAddress,
		warpMessengerAddress,
		relayerRewardAddress,
	)
	addDestinationToRelayerConfig(
		awmRelayerConfig,
		rpcEndpoint,
		subnetID,
		blockchainID,
		relayerPrivateKey,
	)
	return saveRelayerConfig(awmRelayerConfig, relayerConfigPath)
}

func AddSourceToRelayerConfig(
	relayerConfigPath string,
	rpcEndpoint string,
	wsEndpoint string,
	subnetID string,
	blockchainID string,
	warpRegistryAddress string,
	warpMessengerAddress string,
	relayerRewardAddress string,
) error {
	awmRelayerConfig, err := loadRelayerConfig(relayerConfigPath)
	if err != nil {
		return err
	}
	addSourceToRelayerConfig(
		awmRelayerConfig,
		rpcEndpoint,
		wsEndpoint,
		subnetID,
		blockchainID,
		warpRegistryAddress,
		warpMessengerAddress,
		relayerRewardAddress,
	)
	return saveRelayerConfig(awmRelayerConfig, relayerConfigPath)
}

func AddDestinationToRelayerConfig(
	relayerConfigPath string,
	rpcEndpoint string,
	subnetID string,
	blockchainID string,
	relayerPrivateKey string,
) error {
	awmRelayerConfig, err := loadRelayerConfig(relayerConfigPath)
	if err != nil {
		return err
	}
	addDestinationToRelayerConfig(
		awmRelayerConfig,
		rpcEndpoint,
		subnetID,
		blockchainID,
		relayerPrivateKey,
	)
	return saveRelayerConfig(awmRelayerConfig, relayerConfigPath)
}

func addSourceToRelayerConfig(
	relayerConfig *Config,
	rpcEndpoint string,
	wsEndpoint string,
	subnetID string,
	blockchainID string,
	warpRegistryAddress string,
	warpMessengerAddress string,
	relayerRewardAddress string,
) {
	if wsEndpoint == "" {
		wsEndpoint = strings.TrimPrefix(rpcEndpoint, "https")
		wsEndpoint = strings.TrimPrefix(wsEndpoint, "http")
		wsEndpoint = strings.TrimSuffix(wsEndpoint, "rpc")
		wsEndpoint = fmt.Sprintf("%s%s%s", "ws", wsEndpoint, "ws")
	}
	source := &SourceBlockchain{
		SubnetID:     subnetID,
		BlockchainID: blockchainID,
		VM:           EVM.String(),
		RPCEndpoint: APIConfig{
			BaseURL: rpcEndpoint,
		},
		WSEndpoint: APIConfig{
			BaseURL: wsEndpoint,
		},
		MessageContracts: map[string]MessageProtocolConfig{
			warpMessengerAddress: {
				MessageFormat: TELEPORTER.String(),
				Settings: map[string]interface{}{
					"reward-address": relayerRewardAddress,
				},
			},
			OffChainRegistrySourceAddress.Hex(): {
				MessageFormat: OFF_CHAIN_REGISTRY.String(),
				Settings: map[string]interface{}{
					"teleporter-registry-address": warpRegistryAddress,
				},
			},
		},
	}
	if !utils.Any(relayerConfig.SourceBlockchains, func(s *SourceBlockchain) bool { return s.BlockchainID == blockchainID }) {
		relayerConfig.SourceBlockchains = append(relayerConfig.SourceBlockchains, source)
	}
}

func addDestinationToRelayerConfig(
	relayerConfig *Config,
	rpcEndpoint string,
	subnetID string,
	blockchainID string,
	relayerFundedAddressKey string,
) {
	destination := &DestinationBlockchain{
		SubnetID:     subnetID,
		BlockchainID: blockchainID,
		VM:           EVM.String(),
		RPCEndpoint: APIConfig{
			BaseURL: rpcEndpoint,
		},
		AccountPrivateKey: relayerFundedAddressKey,
	}
	if !utils.Any(relayerConfig.DestinationBlockchains, func(s *DestinationBlockchain) bool { return s.BlockchainID == blockchainID }) {
		relayerConfig.DestinationBlockchains = append(relayerConfig.DestinationBlockchains, destination)
	}
}

func waitForRelayerInitialization(
	relayerConfigPath string,
	logPath string,
	checkInterval time.Duration,
	checkTimeout time.Duration,
) error {
	relayerConfig, err := loadRelayerConfig(relayerConfigPath)
	if err != nil {
		return err
	}
	sourceBlockchains := []string{}
	for _, source := range relayerConfig.SourceBlockchains {
		sourceBlockchains = append(sourceBlockchains, source.BlockchainID)
	}
	if checkInterval == 0 {
		checkInterval = 100 * time.Millisecond
	}
	if checkTimeout == 0 {
		checkTimeout = 120 * time.Second
	}
	t0 := time.Now()
	for {
		bs, err := os.ReadFile(logPath)
		if err != nil {
			return err
		}
		sourcesInitialized := 0
		for _, l := range strings.Split(string(bs), "\n") {
			for _, sourceBlockchain := range sourceBlockchains {
				if strings.Contains(l, "Listener initialized") && strings.Contains(l, sourceBlockchain) {
					sourcesInitialized++
				}
			}
		}
		if sourcesInitialized == len(sourceBlockchains) {
			break
		}
		elapsed := time.Since(t0)
		if elapsed > checkTimeout {
			return fmt.Errorf("timeout waiting for relayer initialization")
		}
		time.Sleep(checkInterval)
	}
	return nil
}