Merge branch 'main' into embedded-vnc

management/server/store/sql_store.go

@@ -599,6 +599,34 @@ func (s *SqlStore) ApproveAccountPeers(ctx context.Context, accountID string) (i
 	return int(result.RowsAffected), nil
 }
 
+// RefreshPeerLastSeen updates only peer_status_last_seen. Every other status
+// column is left untouched: peer_status_connected and
+// peer_status_session_started_at belong to the sync stream that owns the
+// session, and a blind write here would corrupt the fencing
+// MarkPeerConnectedIfNewerSession relies on.
+//
+// LastSeen comes from the database clock for the same reason it does there: a
+// Go-side timestamp is taken before the write and can land after a connect that
+// used CURRENT_TIMESTAMP, dragging the column backwards.
+//
+// staleBefore carries the caller's throttle into the same statement, so
+// concurrent requests for one peer collapse into a single write instead of
+// each racing on its own stale read. The column is nullable — Status is an
+// embedded pointer, so a peer stored without one leaves it NULL — and NULL
+// loses every comparison, hence the explicit branch for a peer never seen.
+func (s *SqlStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) {
+	result := s.db.WithContext(ctx).
+		Model(&nbpeer.Peer{}).
+		Where(accountAndIDQueryCondition, accountID, peerID).
+		Where("(peer_status_last_seen IS NULL OR peer_status_last_seen < ?)", staleBefore).
+		Update("peer_status_last_seen", gorm.Expr("CURRENT_TIMESTAMP"))
+	if result.Error != nil {
+		return false, status.Errorf(status.Internal, "refresh peer last seen: %v", result.Error)
+	}
+
+	return result.RowsAffected > 0, nil
+}
+
 // SaveUsers saves the given list of users to the database.
 func (s *SqlStore) SaveUsers(ctx context.Context, users []*types.User) error {
 	if len(users) == 0 {
@@ -6346,6 +6374,30 @@ func (s *SqlStore) CountProxiesByAccountID(ctx context.Context, accountID string
 	return count, nil
 }
 
+// HasActiveProxyAtClusterAddress reports whether any proxy — shared or
+// account-scoped — is currently active at the given cluster address, using
+// the same connected-within-threshold window as the other active-proxy
+// queries. Backs the agent-network settings delete guard: settings cannot be
+// deleted while a proxy declares the endpoint hostname as its address.
+//
+// The comparison folds case on both sides: the caller passes a normalized
+// (lowercase) hostname, but proxies declare their cluster address verbatim
+// and Connect stores it unchanged, so on case-sensitive collations a proxy
+// declaring "GW.Example.com" would otherwise slip past the guard. Hostnames
+// are case-insensitive per RFC 4343; the guard must be too.
+func (s *SqlStore) HasActiveProxyAtClusterAddress(ctx context.Context, clusterAddress string) (bool, error) {
+	var count int64
+	result := s.db.
+		Model(&proxy.Proxy{}).
+		Where("LOWER(cluster_address) = LOWER(?) AND status = ? AND last_seen > ?", clusterAddress, proxy.StatusConnected, time.Now().Add(-proxyActiveThreshold)).
+		Count(&count)
+	if result.Error != nil {
+		log.WithContext(ctx).Errorf("failed to count active proxies at cluster address: %v", result.Error)
+		return false, status.Errorf(status.Internal, "failed to count active proxies at cluster address")
+	}
+	return count > 0, nil
+}
+
 func (s *SqlStore) IsClusterAddressConflicting(ctx context.Context, clusterAddress, accountID string) (bool, error) {
 	var count int64
 	result := s.db.

management/server/store/sql_store_activity_test.go

@@ -0,0 +1,122 @@
+package store
+
+import (
+	"context"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+	"github.com/netbirdio/netbird/management/server/types"
+)
+
+const activityAccountID = "activityAccountId"
+
+func newActivityTestStore(t *testing.T) Store {
+	t.Helper()
+
+	store, cleanUp, err := NewTestStoreFromSQL(context.Background(), "", t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(cleanUp)
+
+	require.NoError(t, store.SaveAccount(context.Background(), &types.Account{
+		Id:        activityAccountID,
+		Domain:    "activity.example.com",
+		CreatedAt: time.Now().UTC(),
+	}))
+
+	return store
+}
+
+func TestRefreshPeerLastSeen(t *testing.T) {
+	ctx := context.Background()
+	store := newActivityTestStore(t)
+	stored := time.Now().UTC().Add(-3 * time.Hour)
+	require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored)))
+
+	refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour))
+	require.NoError(t, err)
+	assert.True(t, refreshed, "a peer seen three hours ago is stale enough to refresh")
+
+	peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
+	require.NoError(t, err)
+	assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time")
+	assert.True(t, peer.Status.LastSeen.After(stored), "last seen must move forward")
+}
+
+// TestRefreshPeerLastSeenHonoursCutoff covers the throttle the caller relies on:
+// two concurrent requests both read the same stale peer, but only the statement
+// that still finds LastSeen behind the cutoff writes.
+func TestRefreshPeerLastSeenHonoursCutoff(t *testing.T) {
+	ctx := context.Background()
+	store := newActivityTestStore(t)
+	stored := time.Now().UTC().Add(-10 * time.Minute)
+	require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored)))
+
+	refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour))
+	require.NoError(t, err)
+	assert.False(t, refreshed, "a peer seen inside the interval must not be written")
+
+	peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
+	require.NoError(t, err)
+	assert.WithinDuration(t, stored, peer.Status.LastSeen.UTC(), time.Second, "last seen must be left where it was")
+}
+
+// TestRefreshPeerLastSeenRecordsNeverSeenPeer covers the nullable column. Status
+// is an embedded pointer, so a peer stored without one leaves last seen NULL,
+// and NULL loses the cutoff comparison — such a peer would never record its
+// first activity.
+func TestRefreshPeerLastSeenRecordsNeverSeenPeer(t *testing.T) {
+	ctx := context.Background()
+	store := newActivityTestStore(t)
+	stored := activityPeer(time.Time{})
+	stored.Status = nil
+	require.NoError(t, store.AddPeerToAccount(ctx, stored))
+
+	refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour))
+	require.NoError(t, err)
+	assert.True(t, refreshed, "a peer that was never seen must record its first activity")
+
+	peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
+	require.NoError(t, err)
+	assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time")
+}
+
+// TestRefreshPeerLastSeenLeavesSessionStateAlone pins the column boundary: the
+// connected flag and the session token belong to the sync stream that owns the
+// peer's session, and a blind write here would corrupt its fencing. This is why
+// SavePeerStatus is not reused for an activity bump.
+func TestRefreshPeerLastSeenLeavesSessionStateAlone(t *testing.T) {
+	ctx := context.Background()
+	store := newActivityTestStore(t)
+
+	stored := activityPeer(time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC))
+	stored.Status.Connected = true
+	stored.Status.SessionStartedAt = 1234567890
+	require.NoError(t, store.AddPeerToAccount(ctx, stored))
+
+	refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour))
+	require.NoError(t, err)
+	require.True(t, refreshed, "the peer is stale enough to refresh")
+
+	peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
+	require.NoError(t, err)
+	assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should move forward")
+	assert.True(t, peer.Status.Connected, "connected flag must survive an activity write")
+	assert.Equal(t, int64(1234567890), peer.Status.SessionStartedAt, "session token must survive an activity write")
+}
+
+func activityPeer(lastSeen time.Time) *nbpeer.Peer {
+	return &nbpeer.Peer{
+		ID:        "activityPeer",
+		AccountID: activityAccountID,
+		Key:       "activityPeerKey",
+		IP:        netip.MustParseAddr("100.64.0.9"),
+		Name:      "activity-peer",
+		DNSLabel:  "activity-peer",
+		Status:    &nbpeer.PeerStatus{LastSeen: lastSeen},
+	}
+}

management/server/store/sql_store_agentnetwork.go

@@ -315,25 +315,65 @@ func (s *SqlStore) GetAllAgentNetworkSettings(ctx context.Context, lockStrength
 	return settings, nil
 }
 
-// GetAgentNetworkSettingsByCluster returns every Settings row pinned to
-// the given proxy cluster. Used by the bootstrap label generator to
-// build the set of subdomains already taken on a cluster.
-func (s *SqlStore) GetAgentNetworkSettingsByCluster(ctx context.Context, lockStrength LockingStrength, cluster string) ([]*agentNetworkTypes.Settings, error) {
+// GetAgentNetworkSettingsByProxyAddress returns every Settings row whose
+// gateway is served by the proxy declaring the given cluster address. Used by
+// cluster-scoped synthesis to find the accounts a shared proxy serves.
+func (s *SqlStore) GetAgentNetworkSettingsByProxyAddress(ctx context.Context, lockStrength LockingStrength, proxyAddress string) ([]*agentNetworkTypes.Settings, error) {
 	tx := s.db
 	if lockStrength != LockingStrengthNone {
 		tx = tx.Clauses(clause.Locking{Strength: string(lockStrength)})
 	}
 
 	var settings []*agentNetworkTypes.Settings
-	result := tx.Find(&settings, "cluster = ?", cluster)
+	result := tx.Find(&settings, "proxy_address = ?", proxyAddress)
 	if result.Error != nil {
-		log.WithContext(ctx).Errorf("failed to get agent network settings by cluster from store: %v", result.Error)
-		return nil, status.Errorf(status.Internal, "failed to get agent network settings by cluster from store")
+		log.WithContext(ctx).Errorf("failed to get agent network settings by proxy address from store: %v", result.Error)
+		return nil, status.Errorf(status.Internal, "failed to get agent network settings by proxy address from store")
 	}
 
 	return settings, nil
 }
 
+// GetAgentNetworkSettingsByDomain resolves the single Settings row holding the
+// given endpoint hostname — a point query on the domain unique index. Returns
+// status.NotFound when no account owns the domain.
+func (s *SqlStore) GetAgentNetworkSettingsByDomain(ctx context.Context, lockStrength LockingStrength, domain string) (*agentNetworkTypes.Settings, error) {
+	tx := s.db
+	if lockStrength != LockingStrengthNone {
+		tx = tx.Clauses(clause.Locking{Strength: string(lockStrength)})
+	}
+
+	var settings agentNetworkTypes.Settings
+	result := tx.Take(&settings, "domain = ?", domain)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, status.Errorf(status.NotFound, "agent network settings for domain %s not found", domain)
+		}
+
+		log.WithContext(ctx).Errorf("failed to get agent network settings by domain from store: %v", result.Error)
+		return nil, status.Errorf(status.Internal, "failed to get agent network settings by domain from store")
+	}
+
+	return &settings, nil
+}
+
+// CreateAgentNetworkSettings inserts a new settings row.
+//
+// Unlike SaveAgentNetworkSettings (an upsert) this is a plain INSERT, and it
+// returns the driver error unwrapped. Both properties are required by the
+// bootstrap allocator: an upsert would overwrite whichever row it collided
+// with, and the allocator classifies the rejection by matching the driver's
+// message — a unique violation on the account primary key means a concurrent
+// bootstrap for the same account won, and one on the domain index means the
+// hostname is taken.
+func (s *SqlStore) CreateAgentNetworkSettings(ctx context.Context, settings *agentNetworkTypes.Settings) error {
+	if err := s.db.Create(settings).Error; err != nil {
+		log.WithContext(ctx).Debugf("failed to create agent network settings: %v", err)
+		return err
+	}
+	return nil
+}
+
 // SaveAgentNetworkSettings upserts the per-account Agent Network
 // settings row.
 func (s *SqlStore) SaveAgentNetworkSettings(ctx context.Context, settings *agentNetworkTypes.Settings) error {
@@ -346,6 +386,25 @@ func (s *SqlStore) SaveAgentNetworkSettings(ctx context.Context, settings *agent
 	return nil
 }
 
+// DeleteAgentNetworkSettings removes the per-account Agent Network settings
+// row, releasing the account's endpoint. Returns status.NotFound when no row
+// exists. The guards on the delete (no providers, no proxy actively serving
+// the endpoint) live in the manager, which runs this inside a transaction
+// after re-checking them under a row lock.
+func (s *SqlStore) DeleteAgentNetworkSettings(ctx context.Context, accountID string) error {
+	result := s.db.Delete(&agentNetworkTypes.Settings{}, "account_id = ?", accountID)
+	if result.Error != nil {
+		log.WithContext(ctx).Errorf("failed to delete agent network settings from store: %v", result.Error)
+		return status.Errorf(status.Internal, "failed to delete agent network settings from store")
+	}
+
+	if result.RowsAffected == 0 {
+		return status.Errorf(status.NotFound, "agent network settings for account %s not found", accountID)
+	}
+
+	return nil
+}
+
 // IncrementAgentNetworkConsumption atomically upserts the consumption
 // row keyed on (account, dim_kind, dim_id, window_seconds, window_start)
 // and adds the supplied deltas. Concurrent calls from multiple proxy

management/server/store/sql_store_agentnetwork_budgetrule_test.go

@@ -88,9 +88,9 @@ func TestAgentNetworkSettings_RealStore_CollectionTogglesRoundTrip(t *testing.T)
 
 	const accountID = "acc-settings-toggles"
 	require.NoError(t, s.SaveAgentNetworkSettings(ctx, &agentNetworkTypes.Settings{
-		AccountID: accountID,
-		Cluster:   "eu.proxy.netbird.io",
-		Subdomain: "violet",
+		AccountID:    accountID,
+		Domain:       "violet.eu.proxy.netbird.io",
+		ProxyAddress: "eu.proxy.netbird.io",
 	}))
 
 	got, err := s.GetAgentNetworkSettings(ctx, LockingStrengthNone, accountID)

management/server/store/store.go

@@ -1,6 +1,6 @@
 package store
 
-//go:generate go run github.com/golang/mock/mockgen -package store -destination=store_mock.go -source=./store.go -build_flags=-mod=mod
+//go:generate go tool mockgen -package store -destination=store_mock.go -source=./store.go -build_flags=-mod=mod
 
 import (
 	"context"
@@ -180,6 +180,14 @@ type Store interface {
 	// Returns true when the update happened, false when this stream lost
 	// the race against a newer session.
 	MarkPeerConnectedIfNewerSession(ctx context.Context, accountID, peerID string, newSessionStartedAt int64) (bool, error)
+	// RefreshPeerLastSeen records that a peer was just seen, stamping the
+	// database clock like the other status writers. Connected and
+	// SessionStartedAt are left alone, so this never interferes with the
+	// session-ownership protocol MarkPeerConnectedIfNewerSession implements.
+	// The write only lands when the stored LastSeen is older than
+	// staleBefore, which keeps a caller's throttle atomic under concurrent
+	// requests for the same peer. Returns true when the update happened.
+	RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error)
 	// MarkPeerDisconnectedIfSameSession sets the peer to disconnected and
 	// resets SessionStartedAt to zero, but only when the stored
 	// SessionStartedAt equals the given sessionStartedAt. LastSeen is
@@ -328,6 +336,7 @@ type Store interface {
 	GetProxyByAccountID(ctx context.Context, accountID string) (*proxy.Proxy, error)
 	CountProxiesByAccountID(ctx context.Context, accountID string) (int64, error)
 	IsClusterAddressConflicting(ctx context.Context, clusterAddress, accountID string) (bool, error)
+	HasActiveProxyAtClusterAddress(ctx context.Context, clusterAddress string) (bool, error)
 	DeleteAccountCluster(ctx context.Context, clusterAddress, accountID string) error
 
 	GetCustomDomainsCounts(ctx context.Context) (total int64, validated int64, err error)
@@ -360,8 +369,11 @@ type Store interface {
 	DeleteAgentNetworkGuardrail(ctx context.Context, accountID, guardrailID string) error
 	GetAgentNetworkSettings(ctx context.Context, lockStrength LockingStrength, accountID string) (*agentNetworkTypes.Settings, error)
 	GetAllAgentNetworkSettings(ctx context.Context, lockStrength LockingStrength) ([]*agentNetworkTypes.Settings, error)
-	GetAgentNetworkSettingsByCluster(ctx context.Context, lockStrength LockingStrength, cluster string) ([]*agentNetworkTypes.Settings, error)
+	GetAgentNetworkSettingsByProxyAddress(ctx context.Context, lockStrength LockingStrength, proxyAddress string) ([]*agentNetworkTypes.Settings, error)
+	GetAgentNetworkSettingsByDomain(ctx context.Context, lockStrength LockingStrength, domain string) (*agentNetworkTypes.Settings, error)
+	CreateAgentNetworkSettings(ctx context.Context, settings *agentNetworkTypes.Settings) error
 	SaveAgentNetworkSettings(ctx context.Context, settings *agentNetworkTypes.Settings) error
+	DeleteAgentNetworkSettings(ctx context.Context, accountID string) error
 	IncrementAgentNetworkConsumption(ctx context.Context, accountID string, kind agentNetworkTypes.ConsumptionDimension, dimID string, windowSeconds int64, windowStart time.Time, tokensIn, tokensOut int64, costUSD float64) error
 	IncrementAgentNetworkConsumptionBatch(ctx context.Context, accountID string, keys []agentNetworkTypes.ConsumptionKey, tokensIn, tokensOut int64, costUSD float64) error
 	GetAgentNetworkConsumption(ctx context.Context, lockStrength LockingStrength, accountID string, kind agentNetworkTypes.ConsumptionDimension, dimID string, windowSeconds int64, windowStart time.Time) (*agentNetworkTypes.Consumption, error)
@@ -608,6 +620,9 @@ func getMigrationsPreAuto(ctx context.Context) []migrationFunc {
 		func(db *gorm.DB) error {
 			return migration.BackfillPublicIDs[posture.Checks](ctx, db)
 		},
+		func(db *gorm.DB) error {
+			return migration.MigrateAgentNetworkSettingsToDomain(ctx, db)
+		},
 	}
 }