CBG-3356: Add current version to ChangeEntry (#6575)

* CBG-3356: add CV to change entry, test that it corretcly populates when calling for changes. Tests need to activate a channel cache as backfill for channel cache not yet implemented

* updates to fix failing tests. Added cv to version type returned by Putting a doc and deleting a doc to make testing easier

* minor changes

* updates after rebase

* fix for test failure

* updates from rebase

* updated comment

* changes in response to commmets

* updates to fix test failures

* rebase + lint skip

* updates to update the doc id changes test I have to actually test the codepath

Author: gregns1

db/changes.go

@@ -44,19 +44,20 @@ type ChangesOptions struct {
 // A changes entry; Database.GetChanges returns an array of these.
 // Marshals into the standard CouchDB _changes format.
 type ChangeEntry struct {
-	Seq          SequenceID      `json:"seq"`
-	ID           string          `json:"id"`
-	Deleted      bool            `json:"deleted,omitempty"`
-	Removed      base.Set        `json:"removed,omitempty"`
-	Doc          json.RawMessage `json:"doc,omitempty"`
-	Changes      []ChangeRev     `json:"changes"`
-	Err          error           `json:"err,omitempty"` // Used to notify feed consumer of errors
-	allRemoved   bool            // Flag to track whether an entry is a removal in all channels visible to the user.
-	branched     bool
-	backfill     backfillFlag // Flag used to identify non-client entries used for backfill synchronization (di only)
-	principalDoc bool         // Used to indicate _user/_role docs
-	Revoked      bool         `json:"revoked,omitempty"`
-	collectionID uint32
+	Seq            SequenceID      `json:"seq"`
+	ID             string          `json:"id"`
+	Deleted        bool            `json:"deleted,omitempty"`
+	Removed        base.Set        `json:"removed,omitempty"`
+	Doc            json.RawMessage `json:"doc,omitempty"`
+	Changes        []ChangeRev     `json:"changes"`
+	Err            error           `json:"err,omitempty"` // Used to notify feed consumer of errors
+	allRemoved     bool            // Flag to track whether an entry is a removal in all channels visible to the user.
+	branched       bool
+	backfill       backfillFlag // Flag used to identify non-client entries used for backfill synchronization (di only)
+	principalDoc   bool         // Used to indicate _user/_role docs
+	Revoked        bool         `json:"revoked,omitempty"`
+	collectionID   uint32
+	CurrentVersion *SourceAndVersion `json:"current_version,omitempty"` // the current version of the change entry
 }
 
 const (
@@ -481,6 +482,12 @@ func makeChangeEntry(logEntry *LogEntry, seqID SequenceID, channel channels.ID)
 		principalDoc: logEntry.IsPrincipal,
 		collectionID: logEntry.CollectionID,
 	}
+	// populate CurrentVersion entry if log entry has sourceID and Version populated
+	// This allows current version to be nil in event of CV not being populated on log entry
+	// allowing omitempty to work as expected
+	if logEntry.SourceID != "" && logEntry.Version != 0 {
+		change.CurrentVersion = &SourceAndVersion{SourceID: logEntry.SourceID, Version: logEntry.Version}
+	}
 	if logEntry.Flags&channels.Removed != 0 {
 		change.Removed = base.SetOf(channel.Name)
 	}
@@ -1281,6 +1288,12 @@ func createChangesEntry(ctx context.Context, docid string, db *DatabaseCollectio
 	row.Seq = SequenceID{Seq: populatedDoc.Sequence}
 	row.SetBranched((populatedDoc.Flags & channels.Branched) != 0)
 
+	if populatedDoc.HLV != nil {
+		cv := SourceAndVersion{}
+		cv.SourceID, cv.Version = populatedDoc.HLV.GetCurrentVersion()
+		row.CurrentVersion = &cv
+	}
+
 	var removedChannels []string
 
 	userCanSeeDocChannel := false

db/changes_test.go

@@ -290,6 +290,39 @@ func TestDocDeletionFromChannelCoalescedRemoved(t *testing.T) {
 	printChanges(changes)
 }
 
+func TestCVPopulationOnChangeEntry(t *testing.T) {
+	db, ctx := setupTestDB(t)
+	defer db.Close(ctx)
+	collection := GetSingleDatabaseCollectionWithUser(t, db)
+	collectionID := collection.GetCollectionID()
+	bucketUUID := db.BucketUUID
+
+	collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
+
+	authenticator := db.Authenticator(base.TestCtx(t))
+	user, err := authenticator.NewUser("alice", "letmein", channels.BaseSetOf(t, "A"))
+	require.NoError(t, err)
+	require.NoError(t, authenticator.Save(user))
+
+	collection.user, _ = authenticator.GetUser("alice")
+
+	// Make channel active
+	_, err = db.channelCache.GetChanges(ctx, channels.NewID("A", collectionID), getChangesOptionsWithZeroSeq(t))
+	require.NoError(t, err)
+
+	_, doc, err := collection.Put(ctx, "doc1", Body{"channels": []string{"A"}})
+	require.NoError(t, err)
+
+	require.NoError(t, collection.WaitForPendingChanges(base.TestCtx(t)))
+
+	changes, err := collection.GetChanges(ctx, base.SetOf("A"), getChangesOptionsWithZeroSeq(t))
+	require.NoError(t, err)
+
+	assert.Equal(t, doc.ID, changes[0].ID)
+	assert.Equal(t, bucketUUID, changes[0].CurrentVersion.SourceID)
+	assert.Equal(t, doc.Cas, changes[0].CurrentVersion.Version)
+}
+
 func TestDocDeletionFromChannelCoalesced(t *testing.T) {
 	if base.TestUseXattrs() {
 		t.Skip("This test is known to be failing against couchbase server with XATTRS enabled.  Same error as TestDocDeletionFromChannelCoalescedRemoved")

db/database_test.go

@@ -1054,6 +1054,7 @@ func TestConflicts(t *testing.T) {
 	db, ctx := setupTestDB(t)
 	defer db.Close(ctx)
 	collection := GetSingleDatabaseCollectionWithUser(t, db)
+	bucketUUID := db.BucketUUID
 
 	collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
 
@@ -1125,15 +1126,19 @@ func TestConflicts(t *testing.T) {
 		Conflicts:  true,
 		ChangesCtx: base.TestCtx(t),
 	}
+	fetchedDoc, _, err := collection.GetDocWithXattr(ctx, "doc", DocUnmarshalCAS)
+	require.NoError(t, err)
+
 	changes, err := collection.GetChanges(ctx, channels.BaseSetOf(t, "all"), options)
 	assert.NoError(t, err, "Couldn't GetChanges")
 	assert.Equal(t, 1, len(changes))
 	assert.Equal(t, &ChangeEntry{
-		Seq:          SequenceID{Seq: 3},
-		ID:           "doc",
-		Changes:      []ChangeRev{{"rev": "2-b"}, {"rev": "2-a"}},
-		branched:     true,
-		collectionID: collectionID,
+		Seq:            SequenceID{Seq: 3},
+		ID:             "doc",
+		Changes:        []ChangeRev{{"rev": "2-b"}, {"rev": "2-a"}},
+		branched:       true,
+		collectionID:   collectionID,
+		CurrentVersion: &SourceAndVersion{SourceID: bucketUUID, Version: fetchedDoc.Cas},
 	}, changes[0],
 	)
 
@@ -1164,11 +1169,12 @@ func TestConflicts(t *testing.T) {
 	assert.NoError(t, err, "Couldn't GetChanges")
 	assert.Equal(t, 1, len(changes))
 	assert.Equal(t, &ChangeEntry{
-		Seq:          SequenceID{Seq: 4},
-		ID:           "doc",
-		Changes:      []ChangeRev{{"rev": "2-a"}, {"rev": rev3}},
-		branched:     true,
-		collectionID: collectionID,
+		Seq:            SequenceID{Seq: 4},
+		ID:             "doc",
+		Changes:        []ChangeRev{{"rev": "2-a"}, {"rev": rev3}},
+		branched:       true,
+		collectionID:   collectionID,
+		CurrentVersion: &SourceAndVersion{SourceID: bucketUUID, Version: doc.Cas},
 	}, changes[0])
 
 }

db/hybrid_logical_vector.go

@@ -30,8 +30,8 @@ type HybridLogicalVector struct {
 
 // SourceAndVersion is a structure used to add a new entry to a HLV
 type SourceAndVersion struct {
-	SourceID string
-	Version  uint64
+	SourceID string `json:"source_id"`
+	Version  uint64 `json:"version"`
 }
 
 func CreateVersion(source string, version uint64) SourceAndVersion {

rest/changes_test.go

@@ -273,3 +273,65 @@ func TestWebhookWinningRevChangedEvent(t *testing.T) {
 	assert.Equal(t, 5, int(atomic.LoadUint32(&WinningRevChangedCount)))
 	assert.Equal(t, 6, int(atomic.LoadUint32(&DocumentChangedCount)))
 }
+
+func TestCVPopulationOnChangesViaAPI(t *testing.T) {
+	rtConfig := RestTesterConfig{
+		SyncFn: `function(doc) {channel(doc.channels)}`,
+	}
+	rt := NewRestTester(t, &rtConfig)
+	defer rt.Close()
+	ctx := base.TestCtx(t)
+	collection := rt.GetSingleTestDatabaseCollection()
+	bucketUUID := rt.GetDatabase().BucketUUID
+	const DocID = "doc1"
+
+	// activate channel cache
+	_, err := rt.WaitForChanges(0, "/{{.keyspace}}/_changes", "", true)
+	require.NoError(t, err)
+
+	resp := rt.SendAdminRequest(http.MethodPut, "/{{.keyspace}}/"+DocID, `{"channels": ["ABC"]}`)
+	RequireStatus(t, resp, http.StatusCreated)
+
+	require.NoError(t, collection.WaitForPendingChanges(base.TestCtx(t)))
+
+	changes, err := rt.WaitForChanges(1, "/{{.keyspace}}/_changes", "", true)
+	require.NoError(t, err)
+
+	fetchedDoc, _, err := collection.GetDocWithXattr(ctx, DocID, db.DocUnmarshalCAS)
+	require.NoError(t, err)
+
+	assert.Equal(t, "doc1", changes.Results[0].ID)
+	assert.Equal(t, bucketUUID, changes.Results[0].CurrentVersion.SourceID)
+	assert.Equal(t, fetchedDoc.Cas, changes.Results[0].CurrentVersion.Version)
+}
+
+func TestCVPopulationOnDocIDChanges(t *testing.T) {
+	rtConfig := RestTesterConfig{
+		SyncFn: `function(doc) {channel(doc.channels)}`,
+	}
+	rt := NewRestTester(t, &rtConfig)
+	defer rt.Close()
+	ctx := base.TestCtx(t)
+	collection := rt.GetSingleTestDatabaseCollection()
+	bucketUUID := rt.GetDatabase().BucketUUID
+	const DocID = "doc1"
+
+	// activate channel cache
+	_, err := rt.WaitForChanges(0, "/{{.keyspace}}/_changes", "", true)
+	require.NoError(t, err)
+
+	resp := rt.SendAdminRequest(http.MethodPut, "/{{.keyspace}}/"+DocID, `{"channels": ["ABC"]}`)
+	RequireStatus(t, resp, http.StatusCreated)
+
+	require.NoError(t, collection.WaitForPendingChanges(base.TestCtx(t)))
+
+	changes, err := rt.WaitForChanges(1, fmt.Sprintf(`/{{.keyspace}}/_changes?filter=_doc_ids&doc_ids=%s`, DocID), "", true)
+	require.NoError(t, err)
+
+	fetchedDoc, _, err := collection.GetDocWithXattr(ctx, DocID, db.DocUnmarshalCAS)
+	require.NoError(t, err)
+
+	assert.Equal(t, "doc1", changes.Results[0].ID)
+	assert.Equal(t, bucketUUID, changes.Results[0].CurrentVersion.SourceID)
+	assert.Equal(t, fetchedDoc.Cas, changes.Results[0].CurrentVersion.Version)
+}

rest/changestest/changes_api_test.go

@@ -861,6 +861,7 @@ func TestChangesFromCompoundSinceViaDocGrant(t *testing.T) {
 	}
 }`})
 	defer rt.Close()
+	collection := rt.GetSingleTestDatabaseCollection()
 
 	// Create user with access to channel NBC:
 	ctx := rt.Context()
@@ -928,14 +929,18 @@ func TestChangesFromCompoundSinceViaDocGrant(t *testing.T) {
 	// Write another doc
 	_ = rt.PutDoc("mix-1", `{"channel":["ABC", "PBS", "HBO"]}`)
 
+	fetchedDoc, _, err := collection.GetDocWithXattr(ctx, "mix-1", db.DocUnmarshalSync)
+	require.NoError(t, err)
+	mixSource, mixVersion := fetchedDoc.HLV.GetCurrentVersion()
+
 	cacheWaiter.AddAndWait(1)
 
 	// Issue a changes request with a compound since value from the last changes response
 	// ensure we don't backfill from the start, but have everything from the compound sequence onwards
 	expectedResults = []string{
 		`{"seq":"8:2","id":"hbo-1","changes":[{"rev":"1-46f8c67c004681619052ee1a1cc8e104"}]}`,
 		`{"seq":8,"id":"grant-1","changes":[{"rev":"1-c5098bb14d12d647c901850ff6a6292a"}]}`,
-		`{"seq":9,"id":"mix-1","changes":[{"rev":"1-32f69cdbf1772a8e064f15e928a18f85"}]}`,
+		fmt.Sprintf(`{"seq":9,"id":"mix-1","changes":[{"rev":"1-32f69cdbf1772a8e064f15e928a18f85"}], "current_version":{"source_id": "%s", "version": %d}}`, mixSource, mixVersion),
 	}
 
 	t.Run("grant via existing channel", func(t *testing.T) {