[management] Refactor expose feature: move business logic from gRPC to manager

client/internal/expose/manager.go

@@ -58,7 +58,7 @@ func (m *Manager) Expose(ctx context.Context, req Request) (*Response, error) {
 }
 
 func (m *Manager) KeepAlive(ctx context.Context, domain string) error {
-	ticker := time.NewTicker(10 * time.Second)
+	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()
 	defer m.stop(domain)
 

management/internals/modules/reverseproxy/interface.go

@@ -21,8 +21,8 @@ type Manager interface {
 	GetServiceByID(ctx context.Context, accountID, serviceID string) (*Service, error)
 	GetAccountServices(ctx context.Context, accountID string) ([]*Service, error)
 	GetServiceIDByTargetID(ctx context.Context, accountID string, resourceID string) (string, error)
-	ValidateExposePermission(ctx context.Context, accountID, peerID string) error
-	CreateServiceFromPeer(ctx context.Context, accountID, peerID string, service *Service) (*Service, error)
-	DeleteServiceFromPeer(ctx context.Context, accountID, peerID, serviceID string) error
-	ExpireServiceFromPeer(ctx context.Context, accountID, peerID, serviceID string) error
+	CreateServiceFromPeer(ctx context.Context, accountID, peerID string, req *ExposeServiceRequest) (*ExposeServiceResponse, error)
+	RenewServiceFromPeer(ctx context.Context, accountID, peerID, domain string) error
+	StopServiceFromPeer(ctx context.Context, accountID, peerID, domain string) error
+	StartExposeReaper(ctx context.Context)
 }

management/internals/modules/reverseproxy/manager/expose_tracker.go

@@ -0,0 +1,163 @@
+package manager
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/netbirdio/netbird/shared/management/status"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	exposeTTL          = 90 * time.Second
+	exposeReapInterval = 30 * time.Second
+	maxExposesPerPeer  = 10
+)
+
+type trackedExpose struct {
+	mu          sync.Mutex
+	domain      string
+	accountID   string
+	peerID      string
+	lastRenewed time.Time
+	expiring    bool
+}
+
+type exposeTracker struct {
+	activeExposes  sync.Map
+	exposeCreateMu sync.Mutex
+	manager        *managerImpl
+}
+
+func exposeKey(peerID, domain string) string {
+	return peerID + ":" + domain
+}
+
+// TrackExposeIfAllowed atomically checks the per-peer limit and registers a new
+// active expose session under the same lock. Returns (true, false) if the expose
+// was already tracked (duplicate), (false, true) if tracking succeeded, and
+// (false, false) if the peer has reached the limit.
+func (t *exposeTracker) TrackExposeIfAllowed(peerID, domain, accountID string) (alreadyTracked, ok bool) {
+	t.exposeCreateMu.Lock()
+	defer t.exposeCreateMu.Unlock()
+
+	key := exposeKey(peerID, domain)
+	_, loaded := t.activeExposes.LoadOrStore(key, &trackedExpose{
+		domain:      domain,
+		accountID:   accountID,
+		peerID:      peerID,
+		lastRenewed: time.Now(),
+	})
+	if loaded {
+		return true, false
+	}
+
+	if t.CountPeerExposes(peerID) > maxExposesPerPeer {
+		t.activeExposes.Delete(key)
+		return false, false
+	}
+
+	return false, true
+}
+
+// UntrackExpose removes an active expose session from tracking.
+func (t *exposeTracker) UntrackExpose(peerID, domain string) {
+	t.activeExposes.Delete(exposeKey(peerID, domain))
+}
+
+// CountPeerExposes returns the number of active expose sessions for a peer.
+func (t *exposeTracker) CountPeerExposes(peerID string) int {
+	count := 0
+	t.activeExposes.Range(func(_, val any) bool {
+		if expose := val.(*trackedExpose); expose.peerID == peerID {
+			count++
+		}
+		return true
+	})
+	return count
+}
+
+// MaxExposesPerPeer returns the maximum number of concurrent exposes allowed per peer.
+func (t *exposeTracker) MaxExposesPerPeer() int {
+	return maxExposesPerPeer
+}
+
+// RenewTrackedExpose updates the in-memory lastRenewed timestamp for a tracked expose.
+// Returns false if the expose is not tracked or is being reaped.
+func (t *exposeTracker) RenewTrackedExpose(peerID, domain string) bool {
+	key := exposeKey(peerID, domain)
+	val, ok := t.activeExposes.Load(key)
+	if !ok {
+		return false
+	}
+
+	expose := val.(*trackedExpose)
+	expose.mu.Lock()
+	if expose.expiring {
+		expose.mu.Unlock()
+		return false
+	}
+	expose.lastRenewed = time.Now()
+	expose.mu.Unlock()
+
+	return true
+}
+
+// StopTrackedExpose removes an active expose session from tracking.
+// Returns false if the expose was not tracked.
+func (t *exposeTracker) StopTrackedExpose(peerID, domain string) bool {
+	key := exposeKey(peerID, domain)
+	_, ok := t.activeExposes.LoadAndDelete(key)
+	return ok
+}
+
+// StartExposeReaper starts a background goroutine that reaps expired expose sessions.
+func (t *exposeTracker) StartExposeReaper(ctx context.Context) {
+	go func() {
+		ticker := time.NewTicker(exposeReapInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				t.reapExpiredExposes()
+			}
+		}
+	}()
+}
+
+func (t *exposeTracker) reapExpiredExposes() {
+	t.activeExposes.Range(func(key, val any) bool {
+		expose := val.(*trackedExpose)
+		expose.mu.Lock()
+		expired := time.Since(expose.lastRenewed) > exposeTTL
+		if expired {
+			expose.expiring = true
+		}
+		expose.mu.Unlock()
+
+		if !expired {
+			return true
+		}
+
+		log.Infof("reaping expired expose session for peer %s, domain %s", expose.peerID, expose.domain)
+
+		err := t.manager.deleteServiceFromPeer(context.Background(), expose.accountID, expose.peerID, expose.domain, true)
+
+		s, _ := status.FromError(err)
+
+		switch {
+		case err == nil:
+			t.activeExposes.Delete(key)
+		case s.ErrorType == status.NotFound:
+			log.Debugf("service %s was already deleted", expose.domain)
+		default:
+			log.Errorf("failed to delete expired peer-exposed service for domain %s: %v", expose.domain, err)
+		}
+
+		return true
+	})
+}