split selectors on discrete resource owners (#1889)

## Description

Switches the CLI from calling `DiscreteScopes` to `SplitByResourceOwner`
on the selector itself.  This func will take the original selector and produce
a slice of selectors, each one with a DiscreteOwner (the single user involved
in usage of that selector) and all include/filter scopes in that selector re-rooted
to that discrete owner.

Does not yet solve the per-category tuple, since we are still pivoting on the
scopes inside the selector.  That comes as a later change.

## Does this PR need a docs update or release note?

- [x] ⛔ No 

## Type of change

- [x] 🌻 Feature

## Issue(s)

* #1617

## Test Plan

- [x] ⚡ Unit test
- [x] 💚 E2E

src/cli/backup/exchange.go

@@ -282,12 +282,10 @@ func createExchangeCmd(cmd *cobra.Command, args []string) error {
 		bIDs []model.StableID
 	)
 
-	for _, scope := range sel.DiscreteScopes(users) {
+	for _, sel := range sel.SplitByResourceOwner(users) {
-		for _, selUser := range scope.Get(selectors.ExchangeUser) {
+		// TODO: pass in entire selector, not individual scopes
-			opSel := selectors.NewExchangeBackup([]string{selUser})
+		for _, scope := range sel.Scopes() {
-			opSel.Include([]selectors.ExchangeScope{scope.DiscreteCopy(selUser)})
+			bo, err := r.NewBackup(ctx, sel.Selector)
-
-			bo, err := r.NewBackup(ctx, opSel.Selector)
 			if err != nil {
 				errs = multierror.Append(errs, errors.Wrapf(
 					err,

src/cli/backup/onedrive.go

@@ -204,12 +204,10 @@ func createOneDriveCmd(cmd *cobra.Command, args []string) error {
 		bIDs []model.StableID
 	)
 
-	for _, scope := range sel.DiscreteScopes(users) {
+	for _, sel := range sel.SplitByResourceOwner(users) {
-		for _, selUser := range scope.Get(selectors.OneDriveUser) {
+		// TODO: pass in entire selector, not individual scopes
-			opSel := selectors.NewOneDriveBackup([]string{selUser})
+		for _, scope := range sel.Scopes() {
-			opSel.Include([]selectors.OneDriveScope{scope.DiscreteCopy(selUser)})
+			bo, err := r.NewBackup(ctx, sel.Selector)
-
-			bo, err := r.NewBackup(ctx, opSel.Selector)
 			if err != nil {
 				errs = multierror.Append(errs, errors.Wrapf(
 					err,

src/cli/backup/sharepoint.go

@@ -210,12 +210,10 @@ func createSharePointCmd(cmd *cobra.Command, args []string) error {
 		bIDs []model.StableID
 	)
 
-	for _, scope := range sel.DiscreteScopes(gc.GetSiteIDs()) {
+	for _, sel := range sel.SplitByResourceOwner(gc.GetSiteIDs()) {
-		for _, selSite := range scope.Get(selectors.SharePointSite) {
+		// TODO: pass in entire selector, not individual scopes
-			opSel := selectors.NewSharePointBackup([]string{selSite})
+		for _, scope := range sel.Scopes() {
-			opSel.Include([]selectors.SharePointScope{scope.DiscreteCopy(selSite)})
+			bo, err := r.NewBackup(ctx, sel.Selector)
-
-			bo, err := r.NewBackup(ctx, opSel.Selector)
 			if err != nil {
 				errs = multierror.Append(errs, errors.Wrapf(
 					err,

src/internal/connector/data_collections.go

@@ -77,7 +77,6 @@ func (gc *GraphConnector) DataCollections(
 		colls, err := sharepoint.DataCollections(
 			ctx,
 			sels,
-			gc.GetSiteIDs(),
 			gc.credentials.AzureTenantID,
 			gc.Service,
 			gc,
@@ -155,7 +154,7 @@ func (gc *GraphConnector) OneDriveDataCollections(
 	}
 
 	var (
-		scopes      = odb.DiscreteScopes(gc.GetUsers())
+		scopes      = odb.DiscreteScopes([]string{selector.DiscreteOwner})
 		collections = []data.Collection{}
 		errs        error
 	)

src/internal/connector/data_collections_test.go

@@ -199,7 +199,7 @@ func (suite *ConnectorDataCollectionIntegrationSuite) TestSharePointDataCollecti
 			getSelector: func() selectors.Selector {
 				sel := selectors.NewSharePointBackup(selSites)
 				sel.Include(sel.Libraries(selSites, selectors.Any()))
-
+				sel.DiscreteOwner = suite.site
 				return sel.Selector
 			},
 		},
@@ -209,7 +209,7 @@ func (suite *ConnectorDataCollectionIntegrationSuite) TestSharePointDataCollecti
 			getSelector: func() selectors.Selector {
 				sel := selectors.NewSharePointBackup(selSites)
 				sel.Include(sel.Lists(selSites, selectors.Any()))
-
+				sel.DiscreteOwner = suite.site
 				return sel.Selector
 			},
 		},
@@ -220,7 +220,6 @@ func (suite *ConnectorDataCollectionIntegrationSuite) TestSharePointDataCollecti
 			collections, err := sharepoint.DataCollections(
 				ctx,
 				test.getSelector(),
-				selSites,
 				connector.credentials.AzureTenantID,
 				connector.Service,
 				connector,

src/internal/connector/sharepoint/data_collections.go

@@ -26,7 +26,6 @@ type statusUpdater interface {
 func DataCollections(
 	ctx context.Context,
 	selector selectors.Selector,
-	siteIDs []string,
 	tenantID string,
 	serv graph.Servicer,
 	su statusUpdater,
@@ -38,7 +37,7 @@ func DataCollections(
 	}
 
 	var (
-		scopes      = b.DiscreteScopes(siteIDs)
+		scopes      = b.DiscreteScopes([]string{selector.DiscreteOwner})
 		collections = []data.Collection{}
 		errs        error
 	)

src/pkg/selectors/exchange.go

@@ -65,6 +65,17 @@ func (s Selector) ToExchangeBackup() (*ExchangeBackup, error) {
 	return &src, nil
 }
 
+func (s ExchangeBackup) SplitByResourceOwner(users []string) []ExchangeBackup {
+	sels := splitByResourceOwner[ExchangeScope](s.Selector, users, ExchangeUser)
+
+	ss := make([]ExchangeBackup, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, ExchangeBackup{exchange{sel}})
+	}
+
+	return ss
+}
+
 // NewExchangeRestore produces a new Selector with the service set to ServiceExchange.
 func NewExchangeRestore(users []string) *ExchangeRestore {
 	src := ExchangeRestore{
@@ -88,6 +99,17 @@ func (s Selector) ToExchangeRestore() (*ExchangeRestore, error) {
 	return &src, nil
 }
 
+func (sr ExchangeRestore) SplitByResourceOwner(users []string) []ExchangeRestore {
+	sels := splitByResourceOwner[ExchangeScope](sr.Selector, users, ExchangeUser)
+
+	ss := make([]ExchangeRestore, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, ExchangeRestore{exchange{sel}})
+	}
+
+	return ss
+}
+
 // Printable creates the minimized display of a selector, formatted for human readability.
 func (s exchange) Printable() Printable {
 	return toPrintable[ExchangeScope](s.Selector)
@@ -176,7 +198,15 @@ func (s *exchange) Scopes() []ExchangeScope {
 // If any Include scope's User category is set to Any, replaces that
 // scope's value with the list of userPNs instead.
 func (s *exchange) DiscreteScopes(userPNs []string) []ExchangeScope {
-	return discreteScopes[ExchangeScope](s.Selector, ExchangeUser, userPNs)
+	scopes := discreteScopes[ExchangeScope](s.Includes, ExchangeUser, userPNs)
+
+	ss := make([]ExchangeScope, 0, len(scopes))
+
+	for _, scope := range scopes {
+		ss = append(ss, ExchangeScope(scope))
+	}
+
+	return ss
 }
 
 type ExchangeItemScopeConstructor func([]string, []string, []string, ...option) []ExchangeScope
@@ -690,12 +720,6 @@ func (s ExchangeScope) setDefaults() {
 	}
 }
 
-// DiscreteCopy makes a shallow clone of the scope, then replaces the clone's
-// user comparison with only the provided user.
-func (s ExchangeScope) DiscreteCopy(user string) ExchangeScope {
-	return discreteCopy(s, user)
-}
-
 // ---------------------------------------------------------------------------
 // Backup Details Filtering
 // ---------------------------------------------------------------------------

src/pkg/selectors/onedrive.go

@@ -64,6 +64,17 @@ func (s Selector) ToOneDriveBackup() (*OneDriveBackup, error) {
 	return &src, nil
 }
 
+func (s OneDriveBackup) SplitByResourceOwner(users []string) []OneDriveBackup {
+	sels := splitByResourceOwner[ExchangeScope](s.Selector, users, OneDriveUser)
+
+	ss := make([]OneDriveBackup, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, OneDriveBackup{oneDrive{sel}})
+	}
+
+	return ss
+}
+
 // NewOneDriveRestore produces a new Selector with the service set to ServiceOneDrive.
 func NewOneDriveRestore(users []string) *OneDriveRestore {
 	src := OneDriveRestore{
@@ -87,6 +98,17 @@ func (s Selector) ToOneDriveRestore() (*OneDriveRestore, error) {
 	return &src, nil
 }
 
+func (s OneDriveRestore) SplitByResourceOwner(users []string) []OneDriveRestore {
+	sels := splitByResourceOwner[ExchangeScope](s.Selector, users, ExchangeUser)
+
+	ss := make([]OneDriveRestore, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, OneDriveRestore{oneDrive{sel}})
+	}
+
+	return ss
+}
+
 // Printable creates the minimized display of a selector, formatted for human readability.
 func (s oneDrive) Printable() Printable {
 	return toPrintable[OneDriveScope](s.Selector)
@@ -169,7 +191,15 @@ func (s *oneDrive) Scopes() []OneDriveScope {
 // If any Include scope's User category is set to Any, replaces that
 // scope's value with the list of userPNs instead.
 func (s *oneDrive) DiscreteScopes(userPNs []string) []OneDriveScope {
-	return discreteScopes[OneDriveScope](s.Selector, OneDriveUser, userPNs)
+	scopes := discreteScopes[OneDriveScope](s.Includes, OneDriveUser, userPNs)
+
+	ss := make([]OneDriveScope, 0, len(scopes))
+
+	for _, scope := range scopes {
+		ss = append(ss, OneDriveScope(scope))
+	}
+
+	return ss
 }
 
 // -------------------

src/pkg/selectors/selectors.go

@@ -98,6 +98,9 @@ type Selector struct {
 	// A record of the resource owners matched by this selector.
 	ResourceOwners filters.Filter `json:"resourceOwners,omitempty"`
 
+	// The single resource owner used by the selector after splitting.
+	DiscreteOwner string `json:"discreteOwner,omitempty"`
+
 	// A slice of exclusion scopes.  Exclusions apply globally to all
 	// inclusions/filters, with any-match behavior.
 	Excludes []scope `json:"exclusions,omitempty"`
@@ -120,10 +123,60 @@ func newSelector(s service, resourceOwners []string) Selector {
 
 // DiscreteResourceOwners returns the list of individual resourceOwners used
 // in the selector.
+// TODO(rkeepers): remove in favor of split and s.DiscreteOwner
 func (s Selector) DiscreteResourceOwners() []string {
 	return split(s.ResourceOwners.Target)
 }
 
+// isAnyResourceOwner returns true if the selector includes all resource owners.
+func isAnyResourceOwner(s Selector) bool {
+	return s.ResourceOwners.Comparator == filters.Passes
+}
+
+// isNoneResourceOwner returns true if the selector includes no resource owners.
+func isNoneResourceOwner(s Selector) bool {
+	return s.ResourceOwners.Comparator == filters.Fails
+}
+
+// SplitByResourceOwner makes one shallow clone for each resourceOwner in the
+// selector, specifying a new DiscreteOwner for each one.
+// If the original selector already specified a discrete slice of resource owners,
+// only those owners are used in the result.
+// If the original selector allowed Any() resource owner, the allOwners parameter
+// is used to populate the slice.  allOwners is assumed to be the complete slice of
+// resourceOwners in the tenant for the given service.
+// If the original selector specified None(), thus failing all resource owners,
+// an empty slice is returned.
+//
+// temporarily, clones all scopes in each selector and replaces the owners with
+// the discrete owner.
+func splitByResourceOwner[T scopeT, C categoryT](s Selector, allOwners []string, rootCat C) []Selector {
+	if isNoneResourceOwner(s) {
+		return []Selector{}
+	}
+
+	targets := allOwners
+
+	if !isAnyResourceOwner(s) {
+		targets = split(s.ResourceOwners.Target)
+	}
+
+	ss := make([]Selector, 0, len(targets))
+
+	for _, ro := range targets {
+		c := s
+		c.DiscreteOwner = ro
+
+		// TODO: when the rootCat gets removed from the scopes, we can remove this
+		c.Includes = discreteScopes[T](s.Includes, rootCat, []string{ro})
+		c.Filters = discreteScopes[T](s.Filters, rootCat, []string{ro})
+
+		ss = append(ss, c)
+	}
+
+	return ss
+}
+
 func (s Selector) String() string {
 	bs, err := json.Marshal(s)
 	if err != nil {
@@ -170,17 +223,17 @@ func scopes[T scopeT](s Selector) []T {
 // If discreteIDs is an empty slice, returns the normal scopes(s).
 // future TODO: if Includes is nil, return filters.
 func discreteScopes[T scopeT, C categoryT](
-	s Selector,
+	scopes []scope,
 	rootCat C,
 	discreteIDs []string,
-) []T {
+) []scope {
-	sl := []T{}
+	sl := []scope{}
 
 	if len(discreteIDs) == 0 {
-		return scopes[T](s)
+		return scopes
 	}
 
-	for _, v := range s.Includes {
+	for _, v := range scopes {
 		t := T(v)
 
 		if isAnyTarget(t, rootCat) {
@@ -189,7 +242,7 @@ func discreteScopes[T scopeT, C categoryT](
 			t = w
 		}
 
-		sl = append(sl, t)
+		sl = append(sl, scope(t))
 	}
 
 	return sl

src/pkg/selectors/selectors_test.go

@@ -254,3 +254,103 @@ func (suite *SelectorSuite) TestContains() {
 	assert.True(t, matches(does, key, target), "does contain")
 	assert.False(t, matches(doesNot, key, target), "does not contain")
 }
+
+func (suite *SelectorSuite) TestIsAnyResourceOwner() {
+	t := suite.T()
+	assert.False(t, isAnyResourceOwner(newSelector(ServiceUnknown, []string{"foo"})))
+	assert.False(t, isAnyResourceOwner(newSelector(ServiceUnknown, []string{})))
+	assert.False(t, isAnyResourceOwner(newSelector(ServiceUnknown, nil)))
+	assert.True(t, isAnyResourceOwner(newSelector(ServiceUnknown, []string{AnyTgt})))
+	assert.True(t, isAnyResourceOwner(newSelector(ServiceUnknown, Any())))
+}
+
+func (suite *SelectorSuite) TestIsNoneResourceOwner() {
+	t := suite.T()
+	assert.False(t, isNoneResourceOwner(newSelector(ServiceUnknown, []string{"foo"})))
+	assert.True(t, isNoneResourceOwner(newSelector(ServiceUnknown, []string{})))
+	assert.True(t, isNoneResourceOwner(newSelector(ServiceUnknown, nil)))
+	assert.True(t, isNoneResourceOwner(newSelector(ServiceUnknown, []string{NoneTgt})))
+	assert.True(t, isNoneResourceOwner(newSelector(ServiceUnknown, None())))
+}
+
+func (suite *SelectorSuite) TestSplitByResourceOnwer() {
+	allOwners := []string{"foo", "bar", "baz", "qux"}
+
+	table := []struct {
+		name           string
+		input          []string
+		expectLen      int
+		expectDiscrete []string
+	}{
+		{
+			name: "nil",
+		},
+		{
+			name:  "empty",
+			input: []string{},
+		},
+		{
+			name:  "noneTgt",
+			input: []string{NoneTgt},
+		},
+		{
+			name:  "none",
+			input: None(),
+		},
+		{
+			name:           "AnyTgt",
+			input:          []string{AnyTgt},
+			expectLen:      len(allOwners),
+			expectDiscrete: allOwners,
+		},
+		{
+			name:           "Any",
+			input:          Any(),
+			expectLen:      len(allOwners),
+			expectDiscrete: allOwners,
+		},
+		{
+			name:           "one owner",
+			input:          []string{"fnord"},
+			expectLen:      1,
+			expectDiscrete: []string{"fnord"},
+		},
+		{
+			name:           "two owners",
+			input:          []string{"fnord", "smarf"},
+			expectLen:      2,
+			expectDiscrete: []string{"fnord", "smarf"},
+		},
+		{
+			name:  "two owners and NoneTgt",
+			input: []string{"fnord", "smarf", NoneTgt},
+		},
+		{
+			name:           "two owners and AnyTgt",
+			input:          []string{"fnord", "smarf", AnyTgt},
+			expectLen:      len(allOwners),
+			expectDiscrete: allOwners,
+		},
+	}
+	for _, test := range table {
+		suite.T().Run(test.name, func(t *testing.T) {
+			s := newSelector(ServiceUnknown, test.input)
+			result := splitByResourceOwner[mockScope](s, allOwners, rootCatStub)
+
+			assert.Len(t, result, test.expectLen)
+
+			for _, expect := range test.expectDiscrete {
+				var found bool
+
+				for _, sel := range result {
+					if sel.DiscreteOwner == expect {
+						found = true
+						break
+					}
+				}
+
+				assert.Truef(t, found, "%s in list of discrete owners", expect)
+			}
+		})
+	}
+}

src/pkg/selectors/sharepoint.go

@@ -62,6 +62,17 @@ func (s Selector) ToSharePointBackup() (*SharePointBackup, error) {
 	return &src, nil
 }
 
+func (s SharePointBackup) SplitByResourceOwner(sites []string) []SharePointBackup {
+	sels := splitByResourceOwner[ExchangeScope](s.Selector, sites, SharePointSite)
+
+	ss := make([]SharePointBackup, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, SharePointBackup{sharePoint{sel}})
+	}
+
+	return ss
+}
+
 // NewSharePointRestore produces a new Selector with the service set to ServiceSharePoint.
 func NewSharePointRestore(sites []string) *SharePointRestore {
 	src := SharePointRestore{
@@ -85,6 +96,17 @@ func (s Selector) ToSharePointRestore() (*SharePointRestore, error) {
 	return &src, nil
 }
 
+func (s SharePointRestore) SplitByResourceOwner(users []string) []SharePointRestore {
+	sels := splitByResourceOwner[ExchangeScope](s.Selector, users, ExchangeUser)
+
+	ss := make([]SharePointRestore, 0, len(sels))
+	for _, sel := range sels {
+		ss = append(ss, SharePointRestore{sharePoint{sel}})
+	}
+
+	return ss
+}
+
 // Printable creates the minimized display of a selector, formatted for human readability.
 func (s sharePoint) Printable() Printable {
 	return toPrintable[SharePointScope](s.Selector)
@@ -167,7 +189,15 @@ func (s *sharePoint) Scopes() []SharePointScope {
 // If any Include scope's Site category is set to Any, replaces that
 // scope's value with the list of siteIDs instead.
 func (s *sharePoint) DiscreteScopes(siteIDs []string) []SharePointScope {
-	return discreteScopes[SharePointScope](s.Selector, SharePointSite, siteIDs)
+	scopes := discreteScopes[SharePointScope](s.Includes, SharePointSite, siteIDs)
+
+	ss := make([]SharePointScope, 0, len(scopes))
+
+	for _, scope := range scopes {
+		ss = append(ss, SharePointScope(scope))
+	}
+
+	return ss
 }
 
 // -------------------