Expand Stream and Collection interfaces for delta token-based incrementals (#1710)

## Description

Add functions to Collection and Stream interfaces that allow for getting
more information about the difference between the previous backup and
the currently in-progress one. These will allow delta token-based
incremental backups to determine how the state has evolved.

Current code does not use these functions and return values for them
are "default" values that should result in full backups even if
KopiaWrapper is updated to start checking the values and GraphConnector
still pulls all items

These functions are not used during restore and can return "default"
values

## Type of change

<!--- Please check the type of change your PR introduces: --->
- [x] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Test
- [ ] 💻 CI/Deployment
- [ ] 🐹 Trivial/Minor

## Issue(s)

* #1700 

## Test Plan

<!-- How will this be tested prior to merging.-->
- [ ] 💪 Manual
- [x] ⚡ Unit test
- [ ] 💚 E2E

src/internal/connector/exchange/exchange_data_collection.go

@@ -116,6 +116,18 @@ func (col *Collection) FullPath() path.Path {
 	return col.fullPath
 }
 
+// TODO(ashmrtn): Fill in with previous path once GraphConnector compares old
+// and new folder hierarchies.
+func (col Collection) PreviousPath() path.Path {
+	return nil
+}
+
+// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
+// hierarchies.
+func (col Collection) State() data.CollectionState {
+	return data.NewState
+}
+
 // populateByOptionIdentifier is a utility function that uses col.collectionType to be able to serialize
 // all the M365IDs defined in the jobs field. data channel is closed by this function
 func (col *Collection) populateByOptionIdentifier(
@@ -438,6 +450,11 @@ func (od *Stream) ToReader() io.ReadCloser {
 	return io.NopCloser(bytes.NewReader(od.message))
 }
 
+// TODO(ashmrtn): Fill in once delta tokens return deleted items.
+func (od Stream) Deleted() bool {
+	return false
+}
+
 func (od *Stream) Info() details.ItemInfo {
 	return details.ItemInfo{Exchange: od.info}
 }

src/internal/connector/graph/metadata_collection.go

@@ -40,6 +40,18 @@ func (md MetadataCollection) FullPath() path.Path {
 	return md.fullPath
 }
 
+// TODO(ashmrtn): Fill in with previous path once GraphConnector compares old
+// and new folder hierarchies.
+func (md MetadataCollection) PreviousPath() path.Path {
+	return nil
+}
+
+// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
+// hierarchies.
+func (md MetadataCollection) State() data.CollectionState {
+	return data.NewState
+}
+
 func (md MetadataCollection) Items() <-chan data.Stream {
 	res := make(chan data.Stream)
 
@@ -101,6 +113,11 @@ func (mi MetadataItem) UUID() string {
 	return mi.uuid
 }
 
+// TODO(ashmrtn): Fill in once we know how to handle this.
+func (mi MetadataItem) Deleted() bool {
+	return false
+}
+
 func (mi MetadataItem) ToReader() io.ReadCloser {
 	return io.NopCloser(bytes.NewReader(mi.data))
 }

src/internal/connector/mockconnector/mock_data_collection.go

@@ -91,6 +91,16 @@ func (medc *MockExchangeDataCollection) FullPath() path.Path {
 	return medc.fullPath
 }
 
+// TODO(ashmrtn): May want to allow setting this in the future for testing.
+func (medc MockExchangeDataCollection) PreviousPath() path.Path {
+	return nil
+}
+
+// TODO(ashmrtn): May want to allow setting this in the future for testing.
+func (medc MockExchangeDataCollection) State() data.CollectionState {
+	return data.NewState
+}
+
 // Items returns a channel that has the next items in the collection. The
 // channel is closed when there are no more items available.
 func (medc *MockExchangeDataCollection) Items() <-chan data.Stream {
@@ -125,6 +135,11 @@ func (med *MockExchangeData) UUID() string {
 	return med.ID
 }
 
+// TODO(ashmrtn): May want to allow setting this in the future for testing.
+func (med MockExchangeData) Deleted() bool {
+	return false
+}
+
 func (med *MockExchangeData) ToReader() io.ReadCloser {
 	if med.ReadErr != nil {
 		return io.NopCloser(errReader{med.ReadErr})

src/internal/connector/onedrive/collection.go

@@ -97,6 +97,18 @@ func (oc *Collection) FullPath() path.Path {
 	return oc.folderPath
 }
 
+// TODO(ashmrtn): Fill in with previous path once GraphConnector compares old
+// and new folder hierarchies.
+func (oc Collection) PreviousPath() path.Path {
+	return nil
+}
+
+// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
+// hierarchies.
+func (oc Collection) State() data.CollectionState {
+	return data.NewState
+}
+
 // Item represents a single item retrieved from OneDrive
 type Item struct {
 	id   string
@@ -112,6 +124,11 @@ func (od *Item) ToReader() io.ReadCloser {
 	return od.data
 }
 
+// TODO(ashmrtn): Fill in once delta tokens return deleted items.
+func (od Item) Deleted() bool {
+	return false
+}
+
 func (od *Item) Info() details.ItemInfo {
 	return details.ItemInfo{OneDrive: od.info}
 }

src/internal/connector/sharepoint/collection.go

@@ -69,6 +69,18 @@ func (sc *Collection) FullPath() path.Path {
 	return sc.fullPath
 }
 
+// TODO(ashmrtn): Fill in with previous path once GraphConnector compares old
+// and new folder hierarchies.
+func (sc Collection) PreviousPath() path.Path {
+	return nil
+}
+
+// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
+// hierarchies.
+func (sc Collection) State() data.CollectionState {
+	return data.NewState
+}
+
 func (sc *Collection) Items() <-chan data.Stream {
 	go sc.populate(context.TODO())
 	return sc.data
@@ -89,6 +101,11 @@ func (sd *Item) ToReader() io.ReadCloser {
 	return sd.data
 }
 
+// TODO(ashmrtn): Fill in once delta tokens return deleted items.
+func (sd Item) Deleted() bool {
+	return false
+}
+
 func (sd *Item) Info() details.ItemInfo {
 	return details.ItemInfo{SharePoint: sd.info}
 }

src/internal/data/data_collection.go

@@ -12,6 +12,15 @@ import (
 // standard ifaces
 // ------------------------------------------------------------------------------------------------
 
+type CollectionState int
+
+const (
+	NewState = CollectionState(iota)
+	NotMovedState
+	MovedState
+	DeletedState
+)
+
 // A Collection represents a compilation of data from the
 // same type application (e.g. mail)
 type Collection interface {
@@ -25,6 +34,19 @@ type Collection interface {
 	// generic. For example, a DataCollection for emails from a specific user
 	// would be {"<tenant id>", "exchange", "<user ID>", "emails"}.
 	FullPath() path.Path
+	// PreviousPath returns the path.Path this collection used to reside at
+	// (according to the M365 ID for the container) if the collection was moved or
+	// renamed. Returns nil if the collection is new or has been deleted.
+	PreviousPath() path.Path
+	// State represents changes to the Collection compared to the last backup
+	// involving the Collection. State changes are based on the M365 ID of the
+	// Collection, not just the path the collection resides at. Collections that
+	// are in the same location as they were in the previous backup should be
+	// marked as NotMovedState. Renaming or reparenting the Collection counts as
+	// Moved. Collections marked as Deleted will be removed from the current
+	// backup along with all items and Collections below them in the hierarchy
+	// unless said items/Collections were moved.
+	State() CollectionState
 }
 
 // Stream represents a single item within a Collection
@@ -34,6 +56,9 @@ type Stream interface {
 	ToReader() io.ReadCloser
 	// UUID provides a unique identifier for this data
 	UUID() string
+	// Deleted returns true if the item represented by this Stream has been
+	// deleted and should be removed from the current in-progress backup.
+	Deleted() bool
 }
 
 // StreamInfo is used to provide service specific

src/internal/data/data_collection_test.go

@@ -22,6 +22,14 @@ func (mc mockColl) FullPath() path.Path {
 	return mc.p
 }
 
+func (mc mockColl) PreviousPath() path.Path {
+	return nil
+}
+
+func (mc mockColl) State() CollectionState {
+	return NewState
+}
+
 type CollectionSuite struct {
 	suite.Suite
 }

src/internal/kopia/data_collection.go

@@ -35,6 +35,14 @@ func (kdc kopiaDataCollection) FullPath() path.Path {
 	return kdc.path
 }
 
+func (kdc kopiaDataCollection) PreviousPath() path.Path {
+	return nil
+}
+
+func (kdc kopiaDataCollection) State() data.CollectionState {
+	return data.NewState
+}
+
 type kopiaDataStream struct {
 	reader io.ReadCloser
 	uuid   string
@@ -49,6 +57,10 @@ func (kds kopiaDataStream) UUID() string {
 	return kds.uuid
 }
 
+func (kds kopiaDataStream) Deleted() bool {
+	return false
+}
+
 func (kds kopiaDataStream) Size() int64 {
 	return kds.size
 }