<commit_before>package db

import (
	"encoding/json"

	"github.com/couchbaselabs/go-couchbase"
	"github.com/couchbaselabs/walrus"

	"github.com/couchbaselabs/sync_gateway/base"
)

// Bidirectional sync with an external Couchbase bucket.
// Watches the bucket's tap feed and applies changes to the matching managed document.
// Accepts local change notifications and makes equivalent changes to the external bucket.
// See: https://github.com/couchbase/sync_gateway/wiki/Bucket-Shadowing
type Shadower struct {
	context *DatabaseContext // Database
	bucket  base.Bucket      // External bucket we sync with
	tapFeed base.TapFeed     // Observes changes to bucket
}

// Creates a new Shadower.
func NewShadower(context *DatabaseContext, bucket base.Bucket) (*Shadower, error) {
	tapFeed, err := bucket.StartTapFeed(walrus.TapArguments{Backfill: 0})
	if err != nil {
		return nil, err
	}
	s := &Shadower{context: context, bucket: bucket, tapFeed: tapFeed}
	go s.readTapFeed()
	return s, nil
}

// Stops a Shadower. (Safe to call on a nil receiver)
func (s *Shadower) Stop() {
	if s != nil && s.tapFeed != nil {
		s.tapFeed.Close()
	}
}

func (s *Shadower) CurrentCheckpoint() uint64 {
	return 0 // TODO: Current tap protocol doesn't provide any value we can return here :(
}

// Main loop that pulls changes from the external bucket. (Runs in its own goroutine.)
func (s *Shadower) readTapFeed() {
	vbucketsFilling := 0
	for event := range s.tapFeed.Events() {
		switch event.Opcode {
		case walrus.TapBeginBackfill:
			if vbucketsFilling == 0 {
				base.LogTo("Shadow", "Reading history of external bucket")
			}
			vbucketsFilling++
			//base.LogTo("Shadow", "Reading history of external bucket")
		case walrus.TapMutation, walrus.TapDeletion:
			if event.Expiry == 0 {
				isDeletion := event.Opcode == walrus.TapDeletion
				err := s.pullDocument(string(event.Key), event.Value, isDeletion, event.Sequence, event.Flags)
				if err != nil {
					base.Warn("Error applying change from external bucket: %v", err)
				}
			}
		case walrus.TapEndBackfill:
			if vbucketsFilling--; vbucketsFilling == 0 {
				base.LogTo("Shadow", "Caught up with history of external bucket")
			}
		}
	}
	base.LogTo("Shadow", "End of tap feed(?)")
}

// Gets an external document and applies it as a new revision to the managed document.
func (s *Shadower) pullDocument(key string, value []byte, isDeletion bool, cas uint64, flags uint32) error {
	var body Body
	if isDeletion {
		body = Body{"_deleted": true}
	} else {
		if err := json.Unmarshal(value, &body); err != nil {
			return err<commit_msg>Bucket shadowing: Minor cleanup.<commit_after>package db

import (
	"encoding/json"
	"strings"

	"github.com/couchbaselabs/go-couchbase"
	"github.com/couchbaselabs/walrus"

	"github.com/couchbaselabs/sync_gateway/base"
)

// Bidirectional sync with an external Couchbase bucket.
// Watches the bucket's tap feed and applies changes to the matching managed document.
// Accepts local change notifications and makes equivalent changes to the external bucket.
// See: https://github.com/couchbase/sync_gateway/wiki/Bucket-Shadowing
type Shadower struct {
	context *DatabaseContext // Database
	bucket  base.Bucket      // External bucket we sync with
	tapFeed base.TapFeed     // Observes changes to bucket
}

// Creates a new Shadower.
func NewShadower(context *DatabaseContext, bucket base.Bucket) (*Shadower, error) {
	tapFeed, err := bucket.StartTapFeed(walrus.TapArguments{Backfill: 0})
	if err != nil {
		return nil, err
	}
	s := &Shadower{context: context, bucket: bucket, tapFeed: tapFeed}
	go s.readTapFeed()
	return s, nil
}

// Stops a Shadower. (Safe to call on a nil receiver)
func (s *Shadower) Stop() {
	if s != nil && s.tapFeed != nil {
		s.tapFeed.Close()
	}
}

// Main loop that pulls changes from the external bucket. (Runs in its own goroutine.)
func (s *Shadower) readTapFeed() {
	vbucketsFilling := 0
	for event := range s.tapFeed.Events() {
		switch event.Opcode {
		case walrus.TapBeginBackfill:
			if vbucketsFilling == 0 {
				base.LogTo("Shadow", "Reading history of external bucket")
			}
			vbucketsFilling++
			//base.LogTo("Shadow", "Reading history of external bucket")
		case walrus.TapMutation, walrus.TapDeletion:
			key := string(event.Key)
			// Ignore ephemeral documents or ones whose ID would conflict with our metadata
			if event.Expiry > 0 || strings.HasPrefix(key, kSyncKeyPrefix) {
				break
			}
			isDeletion := event.Opcode == walrus.TapDeletion
			err := s.pullDocument(key, event.Value, isDeletion, event.Sequence, event.Flags)
			if err != nil {
				base.Warn("Error applying change from external bucket: %v", err)
			}
		case walrus.TapEndBackfill:
			if vbucketsFilling--; vbucketsFilling == 0 {
				base.LogTo("Shadow", "Caught up with history of external bucket")
			}
		}
	}
	base.LogTo("Shadow", "End of tap feed(?)")
}

// Gets an external document and applies it as a new revision to the managed document.
func (s *Shadower) pullDocument(key string, value []byte, isDeletion bool, cas uint64, flags uint32) error {
	var body Body
	if isDeletion {
		body = Body{"_deleted": true}
	} else {
		if err := json.Unmarshal(value, &body); err != nil {
			return err
{'fraction_non_alphanumeric': 0.1140215716486903, 'fraction_numerical': 0.004044684129429892, 'mean_word_length': 3.844216417910448, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6015739', 'n_tokens_mistral': 1853, 'n_tokens_neox': 1695, 'n_words': 535}