change project structure

add across nodes ip limit
add user ip recorder
del config file watch

node/inboundbuilder.go

@@ -0,0 +1,267 @@
+// Package node the InbounderConfig used by add inbound
+package node
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/Yuzuki616/V2bX/api"
+	"github.com/Yuzuki616/V2bX/common/legoCmd"
+	"github.com/Yuzuki616/V2bX/conf"
+	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/uuid"
+	"github.com/xtls/xray-core/core"
+	coreConf "github.com/xtls/xray-core/infra/conf"
+)
+
+//InboundBuilder build Inbound config for different protocol
+func InboundBuilder(config *conf.ControllerConfig, nodeInfo *api.NodeInfo, tag string) (*core.InboundHandlerConfig, error) {
+	var proxySetting interface{}
+	if nodeInfo.NodeType == "V2ray" {
+		defer func() {
+			nodeInfo.V2ray = nil
+		}()
+		if nodeInfo.EnableVless {
+			nodeInfo.V2ray.Inbounds[0].Protocol = "vless"
+			// Enable fallback
+			if config.EnableFallback {
+				fallbackConfigs, err := buildVlessFallbacks(config.FallBackConfigs)
+				if err == nil {
+					proxySetting = &coreConf.VLessInboundConfig{
+						Decryption: "none",
+						Fallbacks:  fallbackConfigs,
+					}
+				} else {
+					return nil, err
+				}
+			} else {
+				proxySetting = &coreConf.VLessInboundConfig{
+					Decryption: "none",
+				}
+			}
+		} else {
+			nodeInfo.V2ray.Inbounds[0].Protocol = "vmess"
+			proxySetting = &coreConf.VMessInboundConfig{}
+		}
+	} else if nodeInfo.NodeType == "Trojan" {
+		defer func() {
+			nodeInfo.V2ray = nil
+			nodeInfo.Trojan = nil
+		}()
+		nodeInfo.V2ray = &api.V2rayConfig{}
+		nodeInfo.V2ray.Inbounds = make([]coreConf.InboundDetourConfig, 1)
+		nodeInfo.V2ray.Inbounds[0].Protocol = "trojan"
+		// Enable fallback
+		if config.EnableFallback {
+			fallbackConfigs, err := buildTrojanFallbacks(config.FallBackConfigs)
+			if err == nil {
+				proxySetting = &coreConf.TrojanServerConfig{
+					Fallbacks: fallbackConfigs,
+				}
+			} else {
+				return nil, err
+			}
+		} else {
+			proxySetting = &coreConf.TrojanServerConfig{}
+		}
+		nodeInfo.V2ray.Inbounds[0].PortList = &coreConf.PortList{
+			Range: []coreConf.PortRange{{From: uint32(nodeInfo.Trojan.LocalPort), To: uint32(nodeInfo.Trojan.LocalPort)}},
+		}
+		t := coreConf.TransportProtocol(nodeInfo.Trojan.TransportProtocol)
+		nodeInfo.V2ray.Inbounds[0].StreamSetting = &coreConf.StreamConfig{Network: &t}
+	} else if nodeInfo.NodeType == "Shadowsocks" {
+		defer func() {
+			nodeInfo.V2ray = nil
+		}()
+		nodeInfo.V2ray = &api.V2rayConfig{}
+		nodeInfo.V2ray.Inbounds = []coreConf.InboundDetourConfig{{Protocol: "shadowsocks"}}
+		proxySetting = &coreConf.ShadowsocksServerConfig{}
+		randomPasswd := uuid.New()
+		defaultSSuser := &coreConf.ShadowsocksUserConfig{
+			Cipher:   "aes-128-gcm",
+			Password: randomPasswd.String(),
+		}
+		proxySetting, _ := proxySetting.(*coreConf.ShadowsocksServerConfig)
+		proxySetting.Users = append(proxySetting.Users, defaultSSuser)
+		proxySetting.NetworkList = &coreConf.NetworkList{"tcp", "udp"}
+		proxySetting.IVCheck = true
+		if config.DisableIVCheck {
+			proxySetting.IVCheck = false
+		}
+		nodeInfo.V2ray.Inbounds[0].PortList = &coreConf.PortList{
+			Range: []coreConf.PortRange{{From: uint32(nodeInfo.SS.Port), To: uint32(nodeInfo.SS.Port)}},
+		}
+		t := coreConf.TransportProtocol(nodeInfo.SS.TransportProtocol)
+		nodeInfo.V2ray.Inbounds[0].StreamSetting = &coreConf.StreamConfig{Network: &t}
+	} else {
+		return nil, fmt.Errorf("unsupported node type: %s, Only support: V2ray, Trojan, Shadowsocks", nodeInfo.NodeType)
+	} /*else if nodeInfo.NodeType == "dokodemo-door" {
+		nodeInfo.V2ray = &api.V2rayConfig{}
+		nodeInfo.V2ray.Inbounds = make([]coreConf.InboundDetourConfig, 1)
+		nodeInfo.V2ray.Inbounds[0].Protocol = "dokodemo-door"
+		proxySetting = struct {
+			Host        string   `json:"address"`
+			NetworkList []string `json:"network"`
+		}{
+			Host:        "v1.mux.cool",
+			NetworkList: []string{"tcp", "udp"},
+		}
+	}*/
+	// Build Listen IP address
+	ipAddress := net.ParseAddress(config.ListenIP)
+	nodeInfo.V2ray.Inbounds[0].ListenOn = &coreConf.Address{Address: ipAddress}
+	// SniffingConfig
+	sniffingConfig := &coreConf.SniffingConfig{
+		Enabled:      true,
+		DestOverride: &coreConf.StringList{"http", "tls"},
+	}
+	if config.DisableSniffing {
+		sniffingConfig.Enabled = false
+	}
+	nodeInfo.V2ray.Inbounds[0].SniffingConfig = sniffingConfig
+
+	var setting json.RawMessage
+
+	// Build Protocol and Protocol setting
+
+	setting, err := json.Marshal(proxySetting)
+	if err != nil {
+		return nil, fmt.Errorf("marshal proxy %s config fialed: %s", nodeInfo.NodeType, err)
+	}
+	if *nodeInfo.V2ray.Inbounds[0].StreamSetting.Network == "tcp" {
+		if nodeInfo.V2ray.Inbounds[0].StreamSetting.TCPSettings != nil {
+			nodeInfo.V2ray.Inbounds[0].StreamSetting.TCPSettings.AcceptProxyProtocol = config.EnableProxyProtocol
+		} else {
+			tcpSetting := &coreConf.TCPConfig{
+				AcceptProxyProtocol: config.EnableProxyProtocol,
+			}
+			nodeInfo.V2ray.Inbounds[0].StreamSetting.TCPSettings = tcpSetting
+		}
+	} else if *nodeInfo.V2ray.Inbounds[0].StreamSetting.Network == "ws" {
+		nodeInfo.V2ray.Inbounds[0].StreamSetting.WSSettings = &coreConf.WebSocketConfig{
+			AcceptProxyProtocol: config.EnableProxyProtocol}
+	}
+	// Build TLS and XTLS settings
+	if nodeInfo.EnableTls && config.CertConfig.CertMode != "none" {
+		nodeInfo.V2ray.Inbounds[0].StreamSetting.Security = nodeInfo.TLSType
+		certFile, keyFile, err := getCertFile(config.CertConfig)
+		if err != nil {
+			return nil, err
+		}
+		if nodeInfo.TLSType == "tls" {
+			tlsSettings := &coreConf.TLSConfig{
+				RejectUnknownSNI: config.CertConfig.RejectUnknownSni,
+			}
+			tlsSettings.Certs = append(tlsSettings.Certs, &coreConf.TLSCertConfig{CertFile: certFile, KeyFile: keyFile, OcspStapling: 3600})
+
+			nodeInfo.V2ray.Inbounds[0].StreamSetting.TLSSettings = tlsSettings
+		} else if nodeInfo.TLSType == "xtls" {
+			xtlsSettings := &coreConf.XTLSConfig{
+				RejectUnknownSNI: config.CertConfig.RejectUnknownSni,
+			}
+			xtlsSettings.Certs = append(xtlsSettings.Certs, &coreConf.XTLSCertConfig{
+				CertFile:     certFile,
+				KeyFile:      keyFile,
+				OcspStapling: 3600})
+			nodeInfo.V2ray.Inbounds[0].StreamSetting.XTLSSettings = xtlsSettings
+		}
+	}
+	// Support ProxyProtocol for any transport protocol
+	if *nodeInfo.V2ray.Inbounds[0].StreamSetting.Network != "tcp" &&
+		*nodeInfo.V2ray.Inbounds[0].StreamSetting.Network != "ws" &&
+		config.EnableProxyProtocol {
+		sockoptConfig := &coreConf.SocketConfig{
+			AcceptProxyProtocol: config.EnableProxyProtocol,
+		}
+		nodeInfo.V2ray.Inbounds[0].StreamSetting.SocketSettings = sockoptConfig
+	}
+	nodeInfo.V2ray.Inbounds[0].Settings = &setting
+	nodeInfo.V2ray.Inbounds[0].Tag = tag
+	return nodeInfo.V2ray.Inbounds[0].Build()
+}
+
+func getCertFile(certConfig *conf.CertConfig) (certFile string, keyFile string, err error) {
+	if certConfig.CertMode == "file" {
+		if certConfig.CertFile == "" || certConfig.KeyFile == "" {
+			return "", "", fmt.Errorf("cert file path or key file path not exist")
+		}
+		return certConfig.CertFile, certConfig.KeyFile, nil
+	} else if certConfig.CertMode == "dns" {
+		lego, err := legoCmd.New()
+		if err != nil {
+			return "", "", err
+		}
+		certPath, keyPath, err := lego.DNSCert(certConfig.CertDomain, certConfig.Email, certConfig.Provider, certConfig.DNSEnv)
+		if err != nil {
+			return "", "", err
+		}
+		return certPath, keyPath, err
+	} else if certConfig.CertMode == "http" {
+		lego, err := legoCmd.New()
+		if err != nil {
+			return "", "", err
+		}
+		certPath, keyPath, err := lego.HTTPCert(certConfig.CertDomain, certConfig.Email)
+		if err != nil {
+			return "", "", err
+		}
+		return certPath, keyPath, err
+	}
+
+	return "", "", fmt.Errorf("unsupported certmode: %s", certConfig.CertMode)
+}
+
+func buildVlessFallbacks(fallbackConfigs []*conf.FallBackConfig) ([]*coreConf.VLessInboundFallback, error) {
+	if fallbackConfigs == nil {
+		return nil, fmt.Errorf("you must provide FallBackConfigs")
+	}
+
+	vlessFallBacks := make([]*coreConf.VLessInboundFallback, len(fallbackConfigs))
+	for i, c := range fallbackConfigs {
+
+		if c.Dest == "" {
+			return nil, fmt.Errorf("dest is required for fallback fialed")
+		}
+
+		var dest json.RawMessage
+		dest, err := json.Marshal(c.Dest)
+		if err != nil {
+			return nil, fmt.Errorf("marshal dest %s config fialed: %s", dest, err)
+		}
+		vlessFallBacks[i] = &coreConf.VLessInboundFallback{
+			Name: c.SNI,
+			Alpn: c.Alpn,
+			Path: c.Path,
+			Dest: dest,
+			Xver: c.ProxyProtocolVer,
+		}
+	}
+	return vlessFallBacks, nil
+}
+
+func buildTrojanFallbacks(fallbackConfigs []*conf.FallBackConfig) ([]*coreConf.TrojanInboundFallback, error) {
+	if fallbackConfigs == nil {
+		return nil, fmt.Errorf("you must provide FallBackConfigs")
+	}
+
+	trojanFallBacks := make([]*coreConf.TrojanInboundFallback, len(fallbackConfigs))
+	for i, c := range fallbackConfigs {
+
+		if c.Dest == "" {
+			return nil, fmt.Errorf("dest is required for fallback fialed")
+		}
+
+		var dest json.RawMessage
+		dest, err := json.Marshal(c.Dest)
+		if err != nil {
+			return nil, fmt.Errorf("marshal dest %s config fialed: %s", dest, err)
+		}
+		trojanFallBacks[i] = &coreConf.TrojanInboundFallback{
+			Name: c.SNI,
+			Alpn: c.Alpn,
+			Path: c.Path,
+			Dest: dest,
+			Xver: c.ProxyProtocolVer,
+		}
+	}
+	return trojanFallBacks, nil
+}

node/inboundbuilder_test.go

@@ -0,0 +1,100 @@
+package node_test
+
+import (
+	. "github.com/Yuzuki616/V2bX/node"
+	"testing"
+
+	"github.com/Yuzuki616/V2bX/api"
+)
+
+func TestBuildV2ray(t *testing.T) {
+	nodeInfo := &api.NodeInfo{
+		NodeType:          "V2ray",
+		NodeID:            1,
+		Port:              1145,
+		SpeedLimit:        0,
+		AlterID:           2,
+		TransportProtocol: "ws",
+		Host:              "test.test.tk",
+		Path:              "v2ray",
+		EnableTLS:         false,
+		TLSType:           "tls",
+	}
+	certConfig := &CertConfig{
+		CertMode:   "http",
+		CertDomain: "test.test.tk",
+		Provider:   "alidns",
+		Email:      "test@gmail.com",
+	}
+	config := &Config{
+		CertConfig: certConfig,
+	}
+	_, err := InboundBuilder(config, nodeInfo)
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func TestBuildTrojan(t *testing.T) {
+	nodeInfo := &api.NodeInfo{
+		NodeType:          "Trojan",
+		NodeID:            1,
+		Port:              1145,
+		SpeedLimit:        0,
+		AlterID:           2,
+		TransportProtocol: "tcp",
+		Host:              "trojan.test.tk",
+		Path:              "v2ray",
+		EnableTLS:         false,
+		TLSType:           "tls",
+	}
+	DNSEnv := make(map[string]string)
+	DNSEnv["ALICLOUD_ACCESS_KEY"] = "aaa"
+	DNSEnv["ALICLOUD_SECRET_KEY"] = "bbb"
+	certConfig := &CertConfig{
+		CertMode:   "dns",
+		CertDomain: "trojan.test.tk",
+		Provider:   "alidns",
+		Email:      "test@gmail.com",
+		DNSEnv:     DNSEnv,
+	}
+	config := &Config{
+		CertConfig: certConfig,
+	}
+	_, err := InboundBuilder(config, nodeInfo)
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func TestBuildSS(t *testing.T) {
+	nodeInfo := &api.NodeInfo{
+		NodeType:          "Shadowsocks",
+		NodeID:            1,
+		Port:              1145,
+		SpeedLimit:        0,
+		AlterID:           2,
+		TransportProtocol: "tcp",
+		Host:              "test.test.tk",
+		Path:              "v2ray",
+		EnableTLS:         false,
+		TLSType:           "tls",
+	}
+	DNSEnv := make(map[string]string)
+	DNSEnv["ALICLOUD_ACCESS_KEY"] = "aaa"
+	DNSEnv["ALICLOUD_SECRET_KEY"] = "bbb"
+	certConfig := &CertConfig{
+		CertMode:   "dns",
+		CertDomain: "trojan.test.tk",
+		Provider:   "alidns",
+		Email:      "test@me.com",
+		DNSEnv:     DNSEnv,
+	}
+	config := &Config{
+		CertConfig: certConfig,
+	}
+	_, err := InboundBuilder(config, nodeInfo)
+	if err != nil {
+		t.Error(err)
+	}
+}

node/node.go

@@ -0,0 +1,427 @@
+package node
+
+import (
+	"fmt"
+	"github.com/Yuzuki616/V2bX/common/limiter"
+	"github.com/Yuzuki616/V2bX/conf"
+	"github.com/Yuzuki616/V2bX/xray"
+	"github.com/go-resty/resty/v2"
+	"github.com/goccy/go-json"
+	"log"
+	"math"
+	"reflect"
+	"runtime"
+	"time"
+
+	"github.com/Yuzuki616/V2bX/api"
+	"github.com/Yuzuki616/V2bX/common/legoCmd"
+	"github.com/xtls/xray-core/common/protocol"
+	"github.com/xtls/xray-core/common/task"
+)
+
+type Node struct {
+	server                  *xray.Xray
+	config                  *conf.ControllerConfig
+	clientInfo              api.ClientInfo
+	apiClient               api.API
+	nodeInfo                *api.NodeInfo
+	Tag                     string
+	userList                *[]api.UserInfo
+	nodeInfoMonitorPeriodic *task.Periodic
+	userReportPeriodic      *task.Periodic
+	onlineIpReportPeriodic  *task.Periodic
+}
+
+// New return a Node service with default parameters.
+func New(server *xray.Xray, api api.API, config *conf.ControllerConfig) *Node {
+	controller := &Node{
+		server:    server,
+		config:    config,
+		apiClient: api,
+	}
+	return controller
+}
+
+// Start implement the Start() function of the service interface
+func (c *Node) Start() error {
+	c.clientInfo = c.apiClient.Describe()
+	// First fetch Node Info
+	newNodeInfo, err := c.apiClient.GetNodeInfo()
+	if err != nil {
+		return err
+	}
+	c.nodeInfo = newNodeInfo
+	c.Tag = c.buildNodeTag()
+	// Add new tag
+	err = c.addNewTag(newNodeInfo)
+	if err != nil {
+		log.Panic(err)
+		return err
+	}
+	// Update user
+	userInfo, err := c.apiClient.GetUserList()
+	if err != nil {
+		return err
+	}
+
+	err = c.addNewUser(userInfo, newNodeInfo)
+	if err != nil {
+		return err
+	}
+	//sync controller userList
+	c.userList = userInfo
+	if err := c.server.AddInboundLimiter(c.Tag, c.nodeInfo, userInfo); err != nil {
+		log.Print(err)
+	}
+	// Add Rule Manager
+	if !c.config.DisableGetRule {
+		if ruleList, protocolRule, err := c.apiClient.GetNodeRule(); err != nil {
+			log.Printf("Get rule list filed: %s", err)
+		} else if len(*ruleList) > 0 {
+			if err := c.server.UpdateRule(c.Tag, *ruleList); err != nil {
+				log.Print(err)
+			}
+			if len(*protocolRule) > 0 {
+				if err := c.server.UpdateProtocolRule(c.Tag, *protocolRule); err != nil {
+					log.Print(err)
+				}
+			}
+		}
+	}
+	c.nodeInfoMonitorPeriodic = &task.Periodic{
+		Interval: time.Duration(c.config.UpdatePeriodic) * time.Second,
+		Execute:  c.nodeInfoMonitor,
+	}
+	c.userReportPeriodic = &task.Periodic{
+		Interval: time.Duration(c.config.UpdatePeriodic) * time.Second,
+		Execute:  c.userInfoMonitor,
+	}
+	log.Printf("[%s: %d] Start monitor node status", c.nodeInfo.NodeType, c.nodeInfo.NodeId)
+	// delay to start nodeInfoMonitor
+	go func() {
+		time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second)
+		_ = c.nodeInfoMonitorPeriodic.Start()
+	}()
+
+	log.Printf("[%s: %d] Start report node status", c.nodeInfo.NodeType, c.nodeInfo.NodeId)
+	// delay to start userReport
+	go func() {
+		time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second)
+		_ = c.userReportPeriodic.Start()
+	}()
+	if c.config.EnableIpRecorder {
+		c.onlineIpReportPeriodic = &task.Periodic{
+			Interval: time.Duration(c.config.UpdatePeriodic) * 30,
+			Execute:  c.onlineIpReport,
+		}
+		log.Printf("[%s: %d] Start report online ip", c.nodeInfo.NodeType, c.nodeInfo.NodeId)
+		// delay to start onlineIpReport
+		go func() {
+			time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second)
+			_ = c.onlineIpReportPeriodic.Start()
+		}()
+	}
+	runtime.GC()
+	return nil
+}
+
+// Close implement the Close() function of the service interface
+func (c *Node) Close() error {
+	if c.nodeInfoMonitorPeriodic != nil {
+		err := c.nodeInfoMonitorPeriodic.Close()
+		if err != nil {
+			log.Panicf("node info periodic close failed: %s", err)
+		}
+	}
+
+	if c.nodeInfoMonitorPeriodic != nil {
+		err := c.userReportPeriodic.Close()
+		if err != nil {
+			log.Panicf("user report periodic close failed: %s", err)
+		}
+	}
+	if c.onlineIpReportPeriodic != nil {
+		err := c.onlineIpReportPeriodic.Close()
+		if err != nil {
+			log.Panicf("online ip report periodic close failed: %s", err)
+		}
+	}
+	return nil
+}
+
+func (c *Node) nodeInfoMonitor() (err error) {
+	// First fetch Node Info
+	newNodeInfo, err := c.apiClient.GetNodeInfo()
+	if err != nil {
+		log.Print(err)
+		return nil
+	}
+
+	var nodeInfoChanged = false
+	// If nodeInfo changed
+	if newNodeInfo != nil {
+		if c.nodeInfo.SS == nil || !reflect.DeepEqual(c.nodeInfo.SS, newNodeInfo.SS) {
+			// Remove old tag
+			oldtag := c.Tag
+			err := c.removeOldTag(oldtag)
+			if err != nil {
+				log.Print(err)
+				return nil
+			}
+			// Add new tag
+			c.nodeInfo = newNodeInfo
+			c.Tag = c.buildNodeTag()
+			err = c.addNewTag(newNodeInfo)
+			if err != nil {
+				log.Print(err)
+				return nil
+			}
+			nodeInfoChanged = true
+			// Remove Old limiter
+			if err = c.server.DeleteInboundLimiter(oldtag); err != nil {
+				log.Print(err)
+				return nil
+			}
+		}
+	}
+
+	// Check Rule
+	if !c.config.DisableGetRule {
+		if ruleList, protocolRule, err := c.apiClient.GetNodeRule(); err != nil {
+			log.Printf("Get rule list filed: %s", err)
+		} else if len(*ruleList) > 0 {
+			if err := c.server.UpdateRule(c.Tag, *ruleList); err != nil {
+				log.Print(err)
+			}
+			if len(*protocolRule) > 0 {
+				if err := c.server.UpdateProtocolRule(c.Tag, *protocolRule); err != nil {
+					log.Print(err)
+				}
+			}
+
+		}
+	}
+
+	// Check Cert
+	if c.nodeInfo.EnableTls && c.config.CertConfig.CertMode != "none" &&
+		(c.config.CertConfig.CertMode == "dns" || c.config.CertConfig.CertMode == "http") {
+		lego, err := legoCmd.New()
+		if err != nil {
+			log.Print(err)
+		}
+		// Xray-core supports the OcspStapling certification hot renew
+		_, _, err = lego.RenewCert(c.config.CertConfig.CertDomain, c.config.CertConfig.Email,
+			c.config.CertConfig.CertMode, c.config.CertConfig.Provider, c.config.CertConfig.DNSEnv)
+		if err != nil {
+			log.Print(err)
+		}
+	}
+	// Update User
+	newUserInfo, err := c.apiClient.GetUserList()
+	if err != nil {
+		log.Print(err)
+		return nil
+	}
+	if nodeInfoChanged {
+		c.userList = newUserInfo
+		newUserInfo = nil
+		err = c.addNewUser(c.userList, newNodeInfo)
+		if err != nil {
+			log.Print(err)
+			return nil
+		}
+		newNodeInfo = nil
+		// Add Limiter
+		if err := c.server.AddInboundLimiter(c.Tag, c.nodeInfo, c.userList); err != nil {
+			log.Print(err)
+			return nil
+		}
+		runtime.GC()
+	} else {
+		deleted, added := compareUserList(c.userList, newUserInfo)
+		if len(*deleted) > 0 {
+			deletedEmail := make([]string, len(*deleted))
+			for i := range *deleted {
+				deletedEmail[i] = fmt.Sprintf("%s|%s|%d", c.Tag,
+					(*deleted)[i].GetUserEmail(),
+					(*deleted)[i].UID)
+			}
+			err := c.server.RemoveUsers(deletedEmail, c.Tag)
+			if err != nil {
+				log.Print(err)
+			}
+		}
+		if len(*added) > 0 {
+			err = c.addNewUser(added, newNodeInfo)
+			if err != nil {
+				log.Print(err)
+			}
+			// Update Limiter
+			if err := c.server.UpdateInboundLimiter(c.Tag, c.nodeInfo, added); err != nil {
+				log.Print(err)
+			}
+		}
+		log.Printf("[%s: %d] %d user deleted, %d user added", c.nodeInfo.NodeType, c.nodeInfo.NodeId,
+			len(*deleted), len(*added))
+		c.userList = newUserInfo
+		newUserInfo = nil
+		runtime.GC()
+	}
+	return nil
+}
+
+func (c *Node) removeOldTag(oldtag string) (err error) {
+	err = c.server.RemoveInbound(oldtag)
+	if err != nil {
+		return err
+	}
+	err = c.server.RemoveOutbound(oldtag)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *Node) addNewTag(newNodeInfo *api.NodeInfo) (err error) {
+	inboundConfig, err := InboundBuilder(c.config, newNodeInfo, c.Tag)
+	if err != nil {
+		return err
+	}
+	err = c.server.AddInbound(inboundConfig)
+	if err != nil {
+
+		return err
+	}
+	outBoundConfig, err := OutboundBuilder(c.config, newNodeInfo, c.Tag)
+	if err != nil {
+
+		return err
+	}
+	err = c.server.AddOutbound(outBoundConfig)
+	if err != nil {
+
+		return err
+	}
+	return nil
+}
+
+func (c *Node) addNewUser(userInfo *[]api.UserInfo, nodeInfo *api.NodeInfo) (err error) {
+	users := make([]*protocol.User, 0)
+	if nodeInfo.NodeType == "V2ray" {
+		if nodeInfo.EnableVless {
+			users = c.buildVlessUsers(userInfo)
+		} else {
+			alterID := 0
+			alterID = (*userInfo)[0].V2rayUser.AlterId
+			if alterID >= 0 && alterID < math.MaxUint16 {
+				users = c.buildVmessUsers(userInfo, uint16(alterID))
+			} else {
+				users = c.buildVmessUsers(userInfo, 0)
+				return fmt.Errorf("AlterID should between 0 to 1<<16 - 1, set it to 0 for now")
+			}
+		}
+	} else if nodeInfo.NodeType == "Trojan" {
+		users = c.buildTrojanUsers(userInfo)
+	} else if nodeInfo.NodeType == "Shadowsocks" {
+		users = c.buildSSUsers(userInfo, getCipherFromString(nodeInfo.SS.CypherMethod))
+	} else {
+		return fmt.Errorf("unsupported node type: %s", nodeInfo.NodeType)
+	}
+	err = c.server.AddUsers(users, c.Tag)
+	if err != nil {
+		return err
+	}
+	log.Printf("[%s: %d] Added %d new users", c.nodeInfo.NodeType, c.nodeInfo.NodeId, len(*userInfo))
+	return nil
+}
+
+func compareUserList(old, new *[]api.UserInfo) (deleted, added *[]api.UserInfo) {
+	tmp := map[string]struct{}{}
+	tmp2 := map[string]struct{}{}
+	for i := range *old {
+		tmp[(*old)[i].GetUserEmail()] = struct{}{}
+	}
+	l := len(tmp)
+	for i := range *new {
+		e := (*new)[i].GetUserEmail()
+		tmp[e] = struct{}{}
+		tmp2[e] = struct{}{}
+		if l != len(tmp) {
+			*added = append(*added, (*new)[i])
+			l++
+		}
+	}
+	tmp = nil
+	l = len(tmp2)
+	for i := range *old {
+		tmp2[(*old)[i].GetUserEmail()] = struct{}{}
+		if l != len(tmp2) {
+			*deleted = append(*deleted, (*old)[i])
+			l++
+		}
+	}
+	return deleted, added
+}
+
+func (c *Node) userInfoMonitor() (err error) {
+	// Get User traffic
+	userTraffic := make([]api.UserTraffic, 0)
+	for i := range *c.userList {
+		up, down := c.server.GetUserTraffic(c.buildUserTag(&(*c.userList)[i]))
+		if up > 0 || down > 0 {
+			userTraffic = append(userTraffic, api.UserTraffic{
+				UID:      (*c.userList)[i].UID,
+				Upload:   up,
+				Download: down})
+		}
+	}
+	if len(userTraffic) > 0 && !c.config.DisableUploadTraffic {
+		err = c.apiClient.ReportUserTraffic(&userTraffic)
+		if err != nil {
+			log.Print(err)
+		} else {
+			log.Printf("[%s: %d] Report %d online users", c.nodeInfo.NodeType, c.nodeInfo.NodeId, len(userTraffic))
+		}
+	}
+	userTraffic = nil
+	runtime.GC()
+	return nil
+}
+
+func (c *Node) onlineIpReport() (err error) {
+	onlineIp, err := c.server.GetOnlineIps(c.Tag)
+	if err != nil {
+		log.Print(err)
+		return nil
+	}
+	rsp, err := resty.New().SetTimeout(time.Duration(c.config.IpRecorderConfig.Timeout) * time.Second).
+		R().
+		SetBody(onlineIp).
+		Post(c.config.IpRecorderConfig.Url +
+			"/api/v1/SyncOnlineIp?token=" +
+			c.config.IpRecorderConfig.Token)
+	if err != nil {
+		log.Print(err)
+		c.server.ClearOnlineIps(c.Tag)
+		return nil
+	}
+	log.Printf("[Node: %d] Report %d online ip", c.nodeInfo.NodeId, len(*onlineIp))
+	if rsp.StatusCode() == 200 {
+		onlineIp = &[]limiter.UserIp{}
+		err := json.Unmarshal(rsp.Body(), onlineIp)
+		if err != nil {
+			log.Print(err)
+			c.server.ClearOnlineIps(c.Tag)
+			return nil
+		}
+		c.server.UpdateOnlineIps(c.Tag, onlineIp)
+		log.Printf("[Node: %d] Updated %d online ip", c.nodeInfo.NodeId, len(*onlineIp))
+	} else {
+		c.server.ClearOnlineIps(c.Tag)
+	}
+	return nil
+}
+
+func (c *Node) buildNodeTag() string {
+	return fmt.Sprintf("%s_%s_%d", c.nodeInfo.NodeType, c.config.ListenIP, c.nodeInfo.NodeId)
+}

node/node_test.go

@@ -0,0 +1,78 @@
+package node_test
+
+import (
+	"fmt"
+	. "github.com/Yuzuki616/V2bX/node"
+	"os"
+	"os/signal"
+	"runtime"
+	"syscall"
+	"testing"
+
+	"github.com/Yuzuki616/V2bX/api"
+	_ "github.com/Yuzuki616/V2bX/example/distro/all"
+	"github.com/xtls/xray-core/core"
+	"github.com/xtls/xray-core/infra/conf"
+)
+
+func TestController(t *testing.T) {
+	serverConfig := &conf.Config{
+		Stats:     &conf.StatsConfig{},
+		LogConfig: &conf.LogConfig{LogLevel: "debug"},
+	}
+	policyConfig := &conf.PolicyConfig{}
+	policyConfig.Levels = map[uint32]*conf.Policy{0: &conf.Policy{
+		StatsUserUplink:   true,
+		StatsUserDownlink: true,
+	}}
+	serverConfig.Policy = policyConfig
+	config, _ := serverConfig.Build()
+
+	// config := &core.Config{
+	// 	App: []*serial.TypedMessage{
+	// 		serial.ToTypedMessage(&dispatcher.Config{}),
+	// 		serial.ToTypedMessage(&proxyman.InboundConfig{}),
+	// 		serial.ToTypedMessage(&proxyman.OutboundConfig{}),
+	// 		serial.ToTypedMessage(&stats.Config{}),
+	// 	}}
+
+	server, err := core.New(config)
+	defer server.Close()
+	if err != nil {
+		t.Errorf("failed to create instance: %s", err)
+	}
+	if err = server.Start(); err != nil {
+		t.Errorf("Failed to start instance: %s", err)
+	}
+	certConfig := &CertConfig{
+		CertMode:   "http",
+		CertDomain: "test.ss.tk",
+		Provider:   "alidns",
+		Email:      "ss@ss.com",
+	}
+	controlerconfig := &Config{
+		UpdatePeriodic: 5,
+		CertConfig:     certConfig,
+	}
+	apiConfig := &api.Config{
+		APIHost:  "http://127.0.0.1:667",
+		Key:      "123",
+		NodeID:   41,
+		NodeType: "V2ray",
+	}
+	apiclient := api.New(apiConfig)
+	c := New(server, apiclient, controlerconfig)
+	fmt.Println("Sleep 1s")
+	err = c.Start()
+	if err != nil {
+		t.Error(err)
+	}
+	//Explicitly triggering GC to remove garbage from config loading.
+	runtime.GC()
+
+	{
+		osSignals := make(chan os.Signal, 1)
+		signal.Notify(osSignals, os.Interrupt, os.Kill, syscall.SIGTERM)
+		<-osSignals
+	}
+}

node/outboundbuilder.go

@@ -0,0 +1,49 @@
+package node
+
+import (
+	"encoding/json"
+	"fmt"
+	conf2 "github.com/Yuzuki616/V2bX/conf"
+
+	"github.com/Yuzuki616/V2bX/api"
+	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/core"
+	"github.com/xtls/xray-core/infra/conf"
+)
+
+//OutboundBuilder build freedom outbund config for addoutbound
+func OutboundBuilder(config *conf2.ControllerConfig, nodeInfo *api.NodeInfo, tag string) (*core.OutboundHandlerConfig, error) {
+	outboundDetourConfig := &conf.OutboundDetourConfig{}
+	outboundDetourConfig.Protocol = "freedom"
+	outboundDetourConfig.Tag = tag
+
+	// Build Send IP address
+	if config.SendIP != "" {
+		ipAddress := net.ParseAddress(config.SendIP)
+		outboundDetourConfig.SendThrough = &conf.Address{Address: ipAddress}
+	}
+
+	// Freedom Protocol setting
+	var domainStrategy = "Asis"
+	if config.EnableDNS {
+		if config.DNSType != "" {
+			domainStrategy = config.DNSType
+		} else {
+			domainStrategy = "UseIP"
+		}
+	}
+	proxySetting := &conf.FreedomConfig{
+		DomainStrategy: domainStrategy,
+	}
+	// Used for Shadowsocks-Plugin
+	if nodeInfo.NodeType == "dokodemo-door" {
+		proxySetting.Redirect = fmt.Sprintf("127.0.0.1:%d", nodeInfo.V2ray.Inbounds[0].PortList.Range[0].From-1)
+	}
+	var setting json.RawMessage
+	setting, err := json.Marshal(proxySetting)
+	if err != nil {
+		return nil, fmt.Errorf("marshal proxy %s config fialed: %s", nodeInfo.NodeType, err)
+	}
+	outboundDetourConfig.Settings = &setting
+	return outboundDetourConfig.Build()
+}

node/userbuilder.go

@@ -0,0 +1,118 @@
+package node
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/Yuzuki616/V2bX/api"
+	"github.com/xtls/xray-core/common/protocol"
+	"github.com/xtls/xray-core/common/serial"
+	"github.com/xtls/xray-core/infra/conf"
+	"github.com/xtls/xray-core/proxy/shadowsocks"
+	"github.com/xtls/xray-core/proxy/trojan"
+	"github.com/xtls/xray-core/proxy/vless"
+)
+
+func (c *Node) buildVmessUsers(userInfo *[]api.UserInfo, serverAlterID uint16) (users []*protocol.User) {
+	users = make([]*protocol.User, len(*userInfo))
+	for i, user := range *userInfo {
+		users[i] = c.buildVmessUser(&user, serverAlterID)
+	}
+	return users
+}
+
+func (c *Node) buildVmessUser(userInfo *api.UserInfo, serverAlterID uint16) (user *protocol.User) {
+	vmessAccount := &conf.VMessAccount{
+		ID:       userInfo.V2rayUser.Uuid,
+		AlterIds: serverAlterID,
+		Security: "auto",
+	}
+	user = &protocol.User{
+		Level:   0,
+		Email:   c.buildUserTag(userInfo), // Uid: InboundTag|email|uid
+		Account: serial.ToTypedMessage(vmessAccount.Build()),
+	}
+	return user
+}
+
+func (c *Node) buildVlessUsers(userInfo *[]api.UserInfo) (users []*protocol.User) {
+	users = make([]*protocol.User, len(*userInfo))
+	for i := range *userInfo {
+		users[i] = c.buildVlessUser(&(*userInfo)[i])
+	}
+	return users
+}
+
+func (c *Node) buildVlessUser(userInfo *api.UserInfo) (user *protocol.User) {
+	vlessAccount := &vless.Account{
+		Id:   userInfo.V2rayUser.Uuid,
+		Flow: "xtls-rprx-direct",
+	}
+	user = &protocol.User{
+		Level:   0,
+		Email:   c.buildUserTag(userInfo),
+		Account: serial.ToTypedMessage(vlessAccount),
+	}
+	return user
+}
+
+func (c *Node) buildTrojanUsers(userInfo *[]api.UserInfo) (users []*protocol.User) {
+	users = make([]*protocol.User, len(*userInfo))
+	for i := range *userInfo {
+		users[i] = c.buildTrojanUser(&(*userInfo)[i])
+	}
+	return users
+}
+
+func (c *Node) buildTrojanUser(userInfo *api.UserInfo) (user *protocol.User) {
+	trojanAccount := &trojan.Account{
+		Password: userInfo.TrojanUser.Password,
+		Flow:     "xtls-rprx-direct",
+	}
+	user = &protocol.User{
+		Level:   0,
+		Email:   c.buildUserTag(userInfo),
+		Account: serial.ToTypedMessage(trojanAccount),
+	}
+	return user
+}
+
+func getCipherFromString(c string) shadowsocks.CipherType {
+	switch strings.ToLower(c) {
+	case "aes-128-gcm", "aead_aes_128_gcm":
+		return shadowsocks.CipherType_AES_128_GCM
+	case "aes-256-gcm", "aead_aes_256_gcm":
+		return shadowsocks.CipherType_AES_256_GCM
+	case "chacha20-poly1305", "aead_chacha20_poly1305", "chacha20-ietf-poly1305":
+		return shadowsocks.CipherType_CHACHA20_POLY1305
+	case "none", "plain":
+		return shadowsocks.CipherType_NONE
+	default:
+		return shadowsocks.CipherType_UNKNOWN
+	}
+}
+
+func (c *Node) buildSSUsers(userInfo *[]api.UserInfo, cypher shadowsocks.CipherType) (users []*protocol.User) {
+	users = make([]*protocol.User, len(*userInfo))
+	for i := range *userInfo {
+		c.buildSSUser(&(*userInfo)[i], cypher)
+	}
+	return users
+}
+
+func (c *Node) buildSSUser(userInfo *api.UserInfo, cypher shadowsocks.CipherType) (user *protocol.User) {
+	ssAccount := &shadowsocks.Account{
+		Password:   userInfo.Secret,
+		CipherType: cypher,
+	}
+	user = &protocol.User{
+		Level:   0,
+		Email:   c.buildUserTag(userInfo),
+		Account: serial.ToTypedMessage(ssAccount),
+	}
+	return user
+}
+
+func (c *Node) buildUserTag(user *api.UserInfo) string {
+	return fmt.Sprintf("%s|%s|%d", c.Tag, user.GetUserEmail(), user.UID)
+}