refactor(session): add registry to manage SSH sessions

- Implement thread-safe session registry with sync.RWMutex
- Add Registry interface for session management operations
- Support Get, Register, Update, and Remove session operations
- Enable dynamic slug updates for existing sessions

session/handler.go

@@ -164,16 +164,9 @@ func (s *SSHSession) HandleTCPIPForward(req *ssh.Request) {
 }
 
 func (s *SSHSession) HandleHTTPForward(req *ssh.Request, portToBind uint16) {
-	slug := generateUniqueSlug()
-	if slug == "" {
-		err := req.Reply(false, nil)
-		if err != nil {
-			log.Println("Failed to reply to request:", err)
-		}
-		return
-	}
+	slug := random.GenerateRandomString(20)
 
-	if !registerClient(slug, s) {
+	if !s.registry.Register(slug, s) {
 		log.Printf("Failed to register client with slug: %s", slug)
 		err := req.Reply(false, nil)
 		if err != nil {
@@ -186,7 +179,7 @@ func (s *SSHSession) HandleHTTPForward(req *ssh.Request, portToBind uint16) {
 	err := binary.Write(buf, binary.BigEndian, uint32(portToBind))
 	if err != nil {
 		log.Println("Failed to write port to buffer:", err)
-		unregisterClient(slug)
+		s.registry.Remove(slug)
 		err = req.Reply(false, nil)
 		if err != nil {
 			log.Println("Failed to reply to request:", err)
@@ -198,7 +191,7 @@ func (s *SSHSession) HandleHTTPForward(req *ssh.Request, portToBind uint16) {
 	err = req.Reply(true, buf.Bytes())
 	if err != nil {
 		log.Println("Failed to reply to request:", err)
-		unregisterClient(slug)
+		s.registry.Remove(slug)
 		err = req.Reply(false, nil)
 		if err != nil {
 			log.Println("Failed to reply to request:", err)
@@ -271,25 +264,6 @@ func (s *SSHSession) HandleTCPForward(req *ssh.Request, addr string, portToBind
 	s.interaction.Start()
 }
 
-func generateUniqueSlug() string {
-	maxAttempts := 5
-
-	for i := 0; i < maxAttempts; i++ {
-		slug := random.GenerateRandomString(20)
-
-		clientsMutex.RLock()
-		_, exists := Clients[slug]
-		clientsMutex.RUnlock()
-
-		if !exists {
-			return slug
-		}
-	}
-
-	log.Println("Failed to generate unique slug after multiple attempts")
-	return ""
-}
-
 func readSSHString(reader *bytes.Reader) (string, error) {
 	var length uint32
 	if err := binary.Read(reader, binary.BigEndian, &length); err != nil {

session/interaction/interaction.go

@@ -114,7 +114,7 @@ func (i *Interaction) SetChannel(channel ssh.Channel) {
 	i.channel = channel
 }
 
-func (i *Interaction) SetSlugModificator(modificator func(oldSlug, newSlug string) bool) {
+func (i *Interaction) SetSlugModificator(modificator func(oldSlug, newSlug string) (success bool)) {
 	i.updateClientSlug = modificator
 }
 

session/registry.go

@@ -0,0 +1,66 @@
+package session
+
+import "sync"
+
+type Registry interface {
+	Get(slug string) (session *SSHSession, exist bool)
+	Update(oldSlug, newSlug string) (success bool)
+	Register(slug string, session *SSHSession) (success bool)
+	Remove(slug string)
+}
+type registry struct {
+	mu      sync.RWMutex
+	clients map[string]*SSHSession
+}
+
+func NewRegistry() Registry {
+	return &registry{
+		clients: make(map[string]*SSHSession),
+	}
+}
+
+func (r *registry) Get(slug string) (session *SSHSession, exist bool) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	session, exist = r.clients[slug]
+	return
+}
+
+func (r *registry) Update(oldSlug, newSlug string) (success bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if _, exists := r.clients[newSlug]; exists && newSlug != oldSlug {
+		return false
+	}
+
+	client, ok := r.clients[oldSlug]
+	if !ok {
+		return false
+	}
+
+	delete(r.clients, oldSlug)
+	client.slugManager.Set(newSlug)
+	r.clients[newSlug] = client
+	return true
+}
+
+func (r *registry) Register(slug string, session *SSHSession) (success bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if _, exists := r.clients[slug]; exists {
+		return false
+	}
+
+	r.clients[slug] = session
+	return true
+}
+
+func (r *registry) Remove(slug string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	delete(r.clients, slug)
+}

session/session.go

@@ -1,8 +1,8 @@
 package session
 
 import (
+	"fmt"
 	"log"
-	"sync"
 	"time"
 	"tunnel_pls/internal/config"
 	"tunnel_pls/session/forwarder"
@@ -13,11 +13,6 @@ import (
 	"golang.org/x/crypto/ssh"
 )
 
-var (
-	clientsMutex sync.RWMutex
-	Clients      = make(map[string]*SSHSession)
-)
-
 type Session interface {
 	HandleGlobalRequest(ch <-chan *ssh.Request)
 	HandleTCPIPForward(req *ssh.Request)
@@ -26,10 +21,13 @@ type Session interface {
 }
 
 type SSHSession struct {
-	lifecycle   lifecycle.SessionLifecycle
-	interaction interaction.Controller
-	forwarder   forwarder.ForwardingController
-	slugManager slug.Manager
+	initialReq    <-chan *ssh.Request
+	sshReqChannel <-chan ssh.NewChannel
+	lifecycle     lifecycle.SessionLifecycle
+	interaction   interaction.Controller
+	forwarder     forwarder.ForwardingController
+	slugManager   slug.Manager
+	registry      Registry
 }
 
 func (s *SSHSession) GetLifecycle() lifecycle.SessionLifecycle {
@@ -48,55 +46,64 @@ func (s *SSHSession) GetSlugManager() slug.Manager {
 	return s.slugManager
 }
 
-func New(conn *ssh.ServerConn, forwardingReq <-chan *ssh.Request, sshChan <-chan ssh.NewChannel) {
+func New(conn *ssh.ServerConn, forwardingReq <-chan *ssh.Request, sshChan <-chan ssh.NewChannel, sessionRegistry Registry) *SSHSession {
 	slugManager := slug.NewManager()
 	forwarderManager := forwarder.NewForwarder(slugManager)
 	interactionManager := interaction.NewInteraction(slugManager, forwarderManager)
 	lifecycleManager := lifecycle.NewLifecycle(conn, forwarderManager, slugManager)
 
 	interactionManager.SetLifecycle(lifecycleManager)
-	interactionManager.SetSlugModificator(updateClientSlug)
+	interactionManager.SetSlugModificator(sessionRegistry.Update)
 	forwarderManager.SetLifecycle(lifecycleManager)
-	lifecycleManager.SetUnregisterClient(unregisterClient)
+	lifecycleManager.SetUnregisterClient(sessionRegistry.Remove)
 
-	session := &SSHSession{
-		lifecycle:   lifecycleManager,
-		interaction: interactionManager,
-		forwarder:   forwarderManager,
-		slugManager: slugManager,
-	}
-
-	var once sync.Once
-	for channel := range sshChan {
-		ch, reqs, err := channel.Accept()
-		if err != nil {
-			log.Printf("failed to accept channel: %v", err)
-			continue
-		}
-		once.Do(func() {
-			session.lifecycle.SetChannel(ch)
-			session.interaction.SetChannel(ch)
-
-			tcpipReq := session.waitForTCPIPForward(forwardingReq)
-			if tcpipReq == nil {
-				log.Printf("Port forwarding request not received. Ensure you ran the correct command with -R flag. Example: ssh %s -p %s -R 80:localhost:3000", config.Getenv("DOMAIN", "localhost"), config.Getenv("PORT", "2200"))
-				if err := session.lifecycle.Close(); err != nil {
-					log.Printf("failed to close session: %v", err)
-				}
-				return
-			}
-			go session.HandleTCPIPForward(tcpipReq)
-		})
-		session.HandleGlobalRequest(reqs)
-	}
-	if err := session.lifecycle.Close(); err != nil {
-		log.Printf("failed to close session: %v", err)
+	return &SSHSession{
+		initialReq:    forwardingReq,
+		sshReqChannel: sshChan,
+		lifecycle:     lifecycleManager,
+		interaction:   interactionManager,
+		forwarder:     forwarderManager,
+		slugManager:   slugManager,
+		registry:      sessionRegistry,
 	}
 }
 
-func (s *SSHSession) waitForTCPIPForward(forwardingReq <-chan *ssh.Request) *ssh.Request {
+func (s *SSHSession) Start() error {
+	channel := <-s.sshReqChannel
+	ch, reqs, err := channel.Accept()
+	if err != nil {
+		log.Printf("failed to accept channel: %v", err)
+		return err
+	}
+	go s.HandleGlobalRequest(reqs)
+
+	tcpipReq := s.waitForTCPIPForward()
+	if tcpipReq == nil {
+		_, err := ch.Write([]byte(fmt.Sprintf("Port forwarding request not received. Ensure you ran the correct command with -R flag. Example: ssh %s -p %s -R 80:localhost:3000", config.Getenv("DOMAIN", "localhost"), config.Getenv("PORT", "2200"))))
+		if err != nil {
+			return err
+		}
+		if err := s.lifecycle.Close(); err != nil {
+			log.Printf("failed to close session: %v", err)
+		}
+		return fmt.Errorf("No forwarding Request")
+	}
+
+	s.lifecycle.SetChannel(ch)
+	s.interaction.SetChannel(ch)
+
+	go s.HandleTCPIPForward(tcpipReq)
+
+	if err := s.lifecycle.Close(); err != nil {
+		log.Printf("failed to close session: %v", err)
+		return err
+	}
+	return nil
+}
+
+func (s *SSHSession) waitForTCPIPForward() *ssh.Request {
 	select {
-	case req, ok := <-forwardingReq:
+	case req, ok := <-s.initialReq:
 		if !ok {
 			log.Println("Forwarding request channel closed")
 			return nil
@@ -114,41 +121,3 @@ func (s *SSHSession) waitForTCPIPForward(forwardingReq <-chan *ssh.Request) *ssh
 		return nil
 	}
 }
-
-func updateClientSlug(oldSlug, newSlug string) bool {
-	clientsMutex.Lock()
-	defer clientsMutex.Unlock()
-
-	if _, exists := Clients[newSlug]; exists && newSlug != oldSlug {
-		return false
-	}
-
-	client, ok := Clients[oldSlug]
-	if !ok {
-		return false
-	}
-
-	delete(Clients, oldSlug)
-	client.slugManager.Set(newSlug)
-	Clients[newSlug] = client
-	return true
-}
-
-func registerClient(slug string, session *SSHSession) bool {
-	clientsMutex.Lock()
-	defer clientsMutex.Unlock()
-
-	if _, exists := Clients[slug]; exists {
-		return false
-	}
-
-	Clients[slug] = session
-	return true
-}
-
-func unregisterClient(slug string) {
-	clientsMutex.Lock()
-	defer clientsMutex.Unlock()
-
-	delete(Clients, slug)
-}

session/slug/slug.go

@@ -1,32 +1,24 @@
 package slug
 
-import "sync"
-
 type Manager interface {
 	Get() string
 	Set(slug string)
 }
 
 type manager struct {
-	slug   string
-	slugMu sync.RWMutex
+	slug string
 }
 
 func NewManager() Manager {
 	return &manager{
-		slug:   "",
-		slugMu: sync.RWMutex{},
+		slug: "",
 	}
 }
 
 func (s *manager) Get() string {
-	s.slugMu.RLock()
-	defer s.slugMu.RUnlock()
 	return s.slug
 }
 
 func (s *manager) Set(slug string) {
-	s.slugMu.Lock()
 	s.slug = slug
-	s.slugMu.Unlock()
 }