From 14f9754db7d7d6cf2b45f7dffcc47efa257a93da Mon Sep 17 00:00:00 2001 From: pascal Date: Fri, 5 Jun 2026 15:15:04 +0200 Subject: [PATCH] expand network managers --- management/server/affectedpeers/resolver.go | 77 ++++++++++++++++++- management/server/group.go | 52 +++++++------ management/server/networks/manager.go | 2 +- .../server/networks/resources/manager.go | 31 +++++--- management/server/networks/routers/manager.go | 68 ++++++---------- management/server/peer.go | 27 +++++++ 6 files changed, 172 insertions(+), 85 deletions(-) diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index 917c91c6a..d96d342f4 100644 --- a/management/server/affectedpeers/resolver.go +++ b/management/server/affectedpeers/resolver.go @@ -24,6 +24,7 @@ import ( rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types" + networkTypes "github.com/netbirdio/netbird/management/server/networks/types" "github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/types" "github.com/netbirdio/netbird/route" @@ -60,8 +61,11 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap return snap, nil } - hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 - needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0 + // Changed resources contribute their group IDs to the changed-group set during + // the walk (see collectFromExplicitResources), so they drive the group walkers. + hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.Resources) > 0 + hasNetworkObject := len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0 || len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0 + needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject needsRoutersResources := needsPolicies // the resource<->router bridge can fire whenever policies/resources/networks are in play var err error @@ -127,6 +131,9 @@ type Change struct { ChangedPeerIDs []string Policies []*types.Policy Routes []*route.Route + Routers []*routerTypes.NetworkRouter + Resources []*resourceTypes.NetworkResource + Networks []*networkTypes.Network PostureCheckIDs []string ResourceIDs []string NetworkIDs []string @@ -147,6 +154,9 @@ func (c Change) isEmpty() bool { len(c.ChangedPeerIDs) == 0 && len(c.Policies) == 0 && len(c.Routes) == 0 && + len(c.Routers) == 0 && + len(c.Resources) == 0 && + len(c.Networks) == 0 && len(c.PostureCheckIDs) == 0 && len(c.ResourceIDs) == 0 && len(c.NetworkIDs) == 0 && @@ -165,8 +175,8 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) [] return nil } r := newResolver(ctx, snap, accountID, c) - log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d postureChecks=%v resources=%v networks=%v", - accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs) + log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v resourceIDs=%v networkIDs=%v", + accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs) r.walk() return r.expand() } @@ -222,6 +232,9 @@ func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change func (r *resolver) walk() { r.collectFromExplicitPolicies() r.collectFromExplicitRoutes(r.change.Routes) + r.collectFromExplicitRouters(r.change.Routers) + r.collectFromExplicitResources(r.change.Resources) + r.collectFromExplicitNetworks(r.change.Networks) r.collectFromPostureChecks(r.change.PostureCheckIDs) if len(r.changedGroupSet) > 0 || len(r.changedPeerSet) > 0 { @@ -339,6 +352,62 @@ func (r *resolver) collectFromExplicitRoutes(routes []*route.Route) { } } +// collectFromExplicitRouters folds the routing peers carried by changed router +// objects (old and/or new state) directly, and marks their networks so the +// router<->source bridge folds the source peers of policies serving them. Carrying +// the old router object here is how a repointed router's previous routing peers +// stay affected without a post-commit read. +func (r *resolver) collectFromExplicitRouters(routers []*routerTypes.NetworkRouter) { + for _, router := range routers { + if router == nil { + continue + } + log.WithContext(r.ctx).Tracef("collectFromExplicitRouters: changed router %s on network %s -> folding peerGroups=%v peer=%q and marking network for source bridge", + router.ID, router.NetworkID, router.PeerGroups, router.Peer) + addAll(r.groupSet, router.PeerGroups) + if router.Peer != "" { + r.peerSet[router.Peer] = struct{}{} + } + if router.NetworkID != "" { + r.networkIDs[router.NetworkID] = struct{}{} + } + } +} + +// collectFromExplicitResources marks the networks of changed resource objects +// (old and/or new state) so the bridge folds their routers and the source peers +// of policies targeting them, and treats each resource's group IDs as changed +// groups so policies targeting the resource through a now-detached (old) group — +// which the post-update group.Resources no longer links — still refresh. +func (r *resolver) collectFromExplicitResources(resources []*resourceTypes.NetworkResource) { + for _, resource := range resources { + if resource == nil { + continue + } + log.WithContext(r.ctx).Tracef("collectFromExplicitResources: changed resource %s on network %s -> marking network for bridge and treating groups %v as changed", + resource.ID, resource.NetworkID, resource.GroupIDs) + addAll(r.changedGroupSet, resource.GroupIDs) + if resource.NetworkID != "" { + r.networkIDs[resource.NetworkID] = struct{}{} + } + } +} + +// collectFromExplicitNetworks marks changed network objects (old and/or new +// state) so the bridge folds their routers and the source peers of policies +// serving their resources. A network has no groups/peers of its own. +func (r *resolver) collectFromExplicitNetworks(networks []*networkTypes.Network) { + for _, network := range networks { + if network == nil { + continue + } + log.WithContext(r.ctx).Tracef("collectFromExplicitNetworks: changed network %s -> marking for bridge", network.ID) + if network.ID != "" { + r.networkIDs[network.ID] = struct{}{} + } + } +} + func (r *resolver) collectFromPostureChecks(postureCheckIDs []string) { if len(postureCheckIDs) == 0 { return diff --git a/management/server/group.go b/management/server/group.go index 510931462..32dc6252f 100644 --- a/management/server/group.go +++ b/management/server/group.go @@ -221,10 +221,14 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us } var eventsToStore []func() + var snaps []*affectedpeers.Snapshot + var changes []affectedpeers.Change var globalErr error - groupIDs := make([]string, 0, len(groups)) + createdCount := 0 for _, newGroup := range groups { + change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}} + var snap *affectedpeers.Snapshot err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil { return err @@ -241,34 +245,31 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us return err } - groupIDs = append(groupIDs, newGroup.ID) - events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup) eventsToStore = append(eventsToStore, events...) - return nil + snap, err = affectedpeers.Load(ctx, transaction, accountID, change) + return err }) if err != nil { log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err) - if len(groupIDs) == 1 { + if createdCount == 0 { return err } globalErr = errors.Join(globalErr, err) // continue updating other groups + continue } + createdCount++ + snaps = append(snaps, snap) + changes = append(changes, change) } for _, storeEvent := range eventsToStore { storeEvent() } - affectedPeerIDs := am.ResolveAffectedPeers(ctx, am.Store, accountID, affectedpeers.Change{ChangedGroupIDs: groupIDs}) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateGroups %v: updating %d affected peers: %v", groupIDs, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateGroups %v: no affected peers", groupIDs) - } + am.dispatchAffected(ctx, accountID, snaps, changes) return globalErr } @@ -287,11 +288,13 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us } var eventsToStore []func() + var snaps []*affectedpeers.Snapshot + var changes []affectedpeers.Change var globalErr error - groupIDs := make([]string, 0, len(groups)) for _, newGroup := range groups { - events, err := am.updateSingleGroup(ctx, accountID, userID, newGroup) + change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}} + events, snap, err := am.updateSingleGroup(ctx, accountID, userID, newGroup, change) if err != nil { log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err) if len(groups) == 1 { @@ -301,26 +304,22 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us continue } eventsToStore = append(eventsToStore, events...) - groupIDs = append(groupIDs, newGroup.ID) + snaps = append(snaps, snap) + changes = append(changes, change) } for _, storeEvent := range eventsToStore { storeEvent() } - affectedPeerIDs := am.ResolveAffectedPeers(ctx, am.Store, accountID, affectedpeers.Change{ChangedGroupIDs: groupIDs}) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("UpdateGroups %v: updating %d affected peers: %v", groupIDs, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("UpdateGroups %v: no affected peers", groupIDs) - } + am.dispatchAffected(ctx, accountID, snaps, changes) return globalErr } -func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountID, userID string, newGroup *types.Group) ([]func(), error) { +func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountID, userID string, newGroup *types.Group, change affectedpeers.Change) ([]func(), *affectedpeers.Snapshot, error) { var events []func() + var snap *affectedpeers.Snapshot err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err := validateNewGroup(ctx, transaction, accountID, newGroup); err != nil { return err @@ -341,9 +340,12 @@ func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountI } events = am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup) - return nil + + var err error + snap, err = affectedpeers.Load(ctx, transaction, accountID, change) + return err }) - return events, err + return events, snap, err } // prepareGroupEvents prepares a list of event functions to be stored. diff --git a/management/server/networks/manager.go b/management/server/networks/manager.go index 4137c3122..d0b7a32c6 100644 --- a/management/server/networks/manager.go +++ b/management/server/networks/manager.go @@ -129,7 +129,7 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw var eventsToStore []func() var snap *affectedpeers.Snapshot - change := affectedpeers.Change{NetworkIDs: []string{networkID}} + change := affectedpeers.Change{Networks: []*types.Network{network}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { // Load before the cascade deletes: pre-state still references the network. var lerr error diff --git a/management/server/networks/resources/manager.go b/management/server/networks/resources/manager.go index ae60e34cf..2a5f10ebd 100644 --- a/management/server/networks/resources/manager.go +++ b/management/server/networks/resources/manager.go @@ -116,7 +116,7 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc var eventsToStore []func() var snap *affectedpeers.Snapshot - change := affectedpeers.Change{ResourceIDs: []string{resource.ID}} + change := affectedpeers.Change{Resources: []*types.NetworkResource{resource}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { var txErr error eventsToStore, snap, txErr = m.createResourceInTransaction(ctx, transaction, userID, resource, change) @@ -256,9 +256,8 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc if err != nil { return fmt.Errorf("failed to get old resource groups: %w", err) } - var oldGroupIDs []string for _, g := range oldGroups { - oldGroupIDs = append(oldGroupIDs, g.ID) + oldResource.GroupIDs = append(oldResource.GroupIDs, g.ID) } err = transaction.SaveNetworkResource(ctx, resource) @@ -276,12 +275,9 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc m.accountManager.StoreEvent(ctx, userID, resource.ID, resource.AccountID, activity.NetworkResourceUpdated, resource.EventMeta(network)) }) - // Pass both old and new resource group IDs so policies that targeted the - // resource via a now-detached group still refresh their source peers. - change = affectedpeers.Change{ - ResourceIDs: []string{resource.ID}, - ChangedGroupIDs: append(oldGroupIDs, resource.GroupIDs...), - } + // Carry both the old and new resource so policies that targeted the resource + // via a now-detached group still refresh their source peers. + change = affectedpeers.Change{Resources: []*types.NetworkResource{oldResource, resource}} if snap, err = affectedpeers.Load(ctx, transaction, resource.AccountID, change); err != nil { return err } @@ -377,9 +373,22 @@ func (m *managerImpl) DeleteResource(ctx context.Context, accountID, userID, net var events []func() var snap *affectedpeers.Snapshot - change := affectedpeers.Change{ResourceIDs: []string{resourceID}} + var change affectedpeers.Change err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - // Load before delete: pre-state snapshot still references the resource. + // Capture the resource and its groups before delete: the post-delete state + // no longer references it. + existing, err := transaction.GetNetworkResourceByID(ctx, store.LockingStrengthUpdate, accountID, resourceID) + if err != nil { + return fmt.Errorf("failed to get network resource: %w", err) + } + oldGroups, err := m.groupsManager.GetResourceGroupsInTransaction(ctx, transaction, store.LockingStrengthNone, accountID, resourceID) + if err != nil { + return fmt.Errorf("failed to get resource groups: %w", err) + } + for _, g := range oldGroups { + existing.GroupIDs = append(existing.GroupIDs, g.ID) + } + change = affectedpeers.Change{Resources: []*types.NetworkResource{existing}} if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { return err } diff --git a/management/server/networks/routers/manager.go b/management/server/networks/routers/manager.go index 2ed455b8a..aca6e5517 100644 --- a/management/server/networks/routers/manager.go +++ b/management/server/networks/routers/manager.go @@ -92,7 +92,7 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t var network *networkTypes.Network var snap *affectedpeers.Snapshot - change := affectedpeers.Change{NetworkIDs: []string{router.NetworkID}} + change := affectedpeers.Change{Routers: []*types.NetworkRouter{router}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { network, err = transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID) if err != nil { @@ -171,10 +171,9 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t var network *networkTypes.Network var snap *affectedpeers.Snapshot var change affectedpeers.Change - var oldRouting []string err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { var txErr error - network, snap, change, oldRouting, txErr = m.updateRouterInTransaction(ctx, transaction, router) + network, snap, change, txErr = m.updateRouterInTransaction(ctx, transaction, router) return txErr }) if err != nil { @@ -183,9 +182,7 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterUpdated, router.EventMeta(network)) - // The previous routing peers lose their routing role and are no longer reachable - // from the post-update state, so union them as directly affected after Expand. - affectedPeerIDs := append(snap.Expand(ctx, router.AccountID, change), oldRouting...) + affectedPeerIDs := snap.Expand(ctx, router.AccountID, change) if len(affectedPeerIDs) > 0 { log.WithContext(ctx).Debugf("UpdateRouter %s: updating %d affected peers: %v", router.ID, len(affectedPeerIDs), affectedPeerIDs) go m.accountManager.UpdateAffectedPeers(ctx, router.AccountID, affectedPeerIDs) @@ -196,66 +193,43 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t return router, nil } -func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, *affectedpeers.Snapshot, affectedpeers.Change, []string, error) { +func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, *affectedpeers.Snapshot, affectedpeers.Change, error) { network, err := transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID) if err != nil { - return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network: %w", err) + return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to get network: %w", err) } existing, err := transaction.GetNetworkRouterByID(ctx, store.LockingStrengthUpdate, router.AccountID, router.ID) if err != nil { - return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network router: %w", err) + return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to get network router: %w", err) } if existing.AccountID != router.AccountID { - return nil, nil, affectedpeers.Change{}, nil, status.NewNetworkRouterNotFoundError(router.ID) + return nil, nil, affectedpeers.Change{}, status.NewNetworkRouterNotFoundError(router.ID) } if existing.NetworkID != router.NetworkID { - return nil, nil, affectedpeers.Change{}, nil, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID) + return nil, nil, affectedpeers.Change{}, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID) } - // Capture the previous routing peers before persisting the update. - oldRouting := oldRoutingPeerIDs(ctx, transaction, router.AccountID, existing) - if err = transaction.UpdateNetworkRouter(ctx, router); err != nil { - return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to update network router: %w", err) + return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to update network router: %w", err) } if err = transaction.IncrementNetworkSerial(ctx, router.AccountID); err != nil { - return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to increment network serial: %w", err) + return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to increment network serial: %w", err) } - networkIDs := []string{router.NetworkID} - if existing.NetworkID != router.NetworkID { - networkIDs = append(networkIDs, existing.NetworkID) - } - - change := affectedpeers.Change{NetworkIDs: networkIDs} + // Carry both the previous and updated router so the bridge folds the old and + // new routing peers; a repoint loses the old peers' routing role and the + // post-update state can no longer reach them. + change := affectedpeers.Change{Routers: []*types.NetworkRouter{existing, router}} snap, err := affectedpeers.Load(ctx, transaction, router.AccountID, change) if err != nil { - return nil, nil, affectedpeers.Change{}, nil, err + return nil, nil, affectedpeers.Change{}, err } - return network, snap, change, oldRouting, nil -} - -// oldRoutingPeerIDs returns the peer IDs that served as the router's routing peers -// before an update (direct Peer plus PeerGroups members). -func oldRoutingPeerIDs(ctx context.Context, transaction store.Store, accountID string, existing *types.NetworkRouter) []string { - var ids []string - if existing.Peer != "" { - ids = append(ids, existing.Peer) - } - if len(existing.PeerGroups) > 0 { - groupPeers, err := transaction.GetPeerIDsByGroups(ctx, accountID, existing.PeerGroups) - if err != nil { - log.WithContext(ctx).Errorf("failed to get old router peer-group members for affected peers: %v", err) - } else { - ids = append(ids, groupPeers...) - } - } - return ids + return network, snap, change, nil } func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, networkID, routerID string) error { @@ -269,9 +243,15 @@ func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, netwo var event func() var snap *affectedpeers.Snapshot - change := affectedpeers.Change{NetworkIDs: []string{networkID}} + var change affectedpeers.Change err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - // Load before delete: pre-state still references the router and its peers. + // Capture the router before delete: its peers lose their routing role and + // the post-delete state can no longer reach them. + existing, err := transaction.GetNetworkRouterByID(ctx, store.LockingStrengthUpdate, accountID, routerID) + if err != nil { + return fmt.Errorf("failed to get network router: %w", err) + } + change = affectedpeers.Change{Routers: []*types.NetworkRouter{existing}} if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { return err } diff --git a/management/server/peer.go b/management/server/peer.go index 1ca97f4a1..bf5e69dfe 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -1478,6 +1478,33 @@ func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, ac } } +// dispatchAffected expands a set of (snapshot, change) pairs collected across +// several transactions, unions their affected peers, and dispatches one update. +// Used by batch operations that commit per-item transactions but want a single +// affected-peers pass — each snapshot is still loaded inside its own transaction. +func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) { + seen := make(map[string]struct{}) + var affectedPeerIDs []string + for i, snap := range snaps { + if snap == nil { + continue + } + for _, id := range snap.Expand(ctx, accountID, changes[i]) { + if _, ok := seen[id]; !ok { + seen[id] = struct{}{} + affectedPeerIDs = append(affectedPeerIDs, id) + } + } + } + + if len(affectedPeerIDs) > 0 { + log.WithContext(ctx).Debugf("dispatchAffected: account %s updating %d affected peers: %v", accountID, len(affectedPeerIDs), affectedPeerIDs) + am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) + } else { + log.WithContext(ctx).Tracef("dispatchAffected: account %s no affected peers", accountID) + } +} + // resolvePeerIDs resolves group IDs and direct peer IDs into a deduplicated peer ID list. func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Store, accountID string, groupIDs []string, directPeerIDs []string) []string { peerIDs, err := s.GetPeerIDsByGroups(ctx, accountID, groupIDs)