bootstrap.go

// Copyright 2022-Present Couchbase, Inc.
//
// Use of this software is governed by the Business Source License included
// in the file licenses/BSL-Couchbase.txt.  As of the Change Date specified
// in that file, in accordance with the Business Source License, use of this
// software will be governed by the Apache License, Version 2.0, included in
// the file licenses/APL2.txt.

package base

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"reflect"
	"sort"
	"sync"
	"time"

	"dario.cat/mergo"
	"github.com/couchbase/gocb/v2"
)

// BootstrapConnection is the interface that can be used to bootstrap Sync Gateway against a Couchbase Server cluster.
// Manages retrieval of set of buckets, and generic interaction with bootstrap metadata documents from those buckets.
type BootstrapConnection interface {
	// GetConfigBuckets returns a list of bucket names where a bootstrap metadata documents could reside.
	GetConfigBuckets(context.Context) ([]string, error)
	// GetMetadataDocument fetches a bootstrap metadata document for a given bucket and key, along with the CAS of the config document.
	GetMetadataDocument(ctx context.Context, bucket, key string, valuePtr any) (cas uint64, err error)
	// InsertMetadataDocument saves a new bootstrap metadata document for a given bucket and key.
	InsertMetadataDocument(ctx context.Context, bucket, key string, value any) (newCAS uint64, err error)
	// DeleteMetadataDocument deletes an existing bootstrap metadata document for a given bucket and key.
	DeleteMetadataDocument(ctx context.Context, bucket, key string, cas uint64) (err error)
	// UpdateMetadataDocument updates an existing bootstrap metadata document for a given bucket and key. updateCallback can return nil to remove the config.  Retries on CAS failure.
	UpdateMetadataDocument(ctx context.Context, bucket, key string, updateCallback func(rawBucketConfig []byte, rawBucketConfigCas uint64) (updatedConfig []byte, err error)) (newCAS uint64, err error)
	// WriteMetadataDocument writes a bootstrap metadata document for a given bucket and key.  Does not retry on CAS failure.
	WriteMetadataDocument(ctx context.Context, bucket, key string, cas uint64, valuePtr any) (casOut uint64, err error)
	// TouchMetadataDocument sets the specified property in a bootstrap metadata document for a given bucket and key.  Used to
	// trigger CAS update on the document, to block any racing updates. Does not retry on CAS failure.
	TouchMetadataDocument(ctx context.Context, bucket, key string, property string, value string, cas uint64) (casOut uint64, err error)
	// KeyExists checks whether the specified key exists in the bucket's default collection
	KeyExists(ctx context.Context, bucket, key string) (exists bool, err error)
	// GetDocument retrieves the document with the specified key from the bucket's default collection.
	// Returns exists=false if key is not found, returns error for any other error.
	GetDocument(ctx context.Context, bucket, docID string, rv any) (exists bool, err error)
	// GetRawDocument retrieves the document with the specified key from the bucket's default collection as raw bytes.
	// Returns exists=false if key is not found, returns error for any other error.
	GetRawDocument(ctx context.Context, bucket, docID string) (value []byte, exists bool, err error)
	// Close releases any long-lived connections
	Close()
}

// CouchbaseClusterSpec define how to make a connection to Couchbase Server
type CouchbaseClusterSpec struct {
	Server               string // connection string to connect to the Couchbase cluster
	Username             string // RBAC username to authenticate with the cluster
	Password             string // RBAC password to authenticate with the cluster
	X509Certpath         string // X.509 cert path to authenticate with the cluster
	X509Keypath          string // X.509 key path to authenticate with the cluster
	CACertpath           string // CA cert path to use for TLS connections
	TLSSkipVerify        bool   // If true, do not validate TLS certificate
	UseGOCBFastFailRetry bool   // When true, readiness checks fail fast instead of using the best-effort retry strategy
}

// CouchbaseCluster is a GoCBv2 implementation of BootstrapConnection
type CouchbaseCluster struct {
	server                  string
	clusterOptions          gocb.ClusterOptions
	forcePerBucketAuth      bool // Forces perBucketAuth authenticators to be used to connect to the bucket
	perBucketAuth           map[string]*gocb.Authenticator
	bucketConnectionMode    BucketConnectionMode    // Whether to cache cluster connections
	cachedClusterConnection *gocb.Cluster           // Cached cluster connection, should only be used by GetConfigBuckets
	cachedBucketConnections cachedBucketConnections // Per-bucket cached connections
	cachedConnectionLock    sync.Mutex              // mutex for access to cachedBucketConnections
	configPersistence       ConfigPersistence       // ConfigPersistence mode
	useGOCBFastFailRetry    bool                    // When true, readiness checks fail fast instead of using the best-effort retry strategy
}

type BucketConnectionMode int

const (
	// CachedClusterConnections mode reuses a cached cluster connection.  Should be used for recurring operations
	CachedClusterConnections BucketConnectionMode = iota
	// PerUseClusterConnections mode establishes a new cluster connection per cluster operation.  Should be used for adhoc operations
	PerUseClusterConnections
)

type cachedBucket struct {
	bucket        *gocb.Bucket // underlying bucket
	bucketCloseFn func()       // teardown function which will close the gocb connection
	refcount      int          // count of how many functions are using this cachedBucket
	shouldClose   bool         // mark this cachedBucket as needing to be closed with ref
}

// cahedBucketConnections is a lockable map cached buckets containing refcounts
type cachedBucketConnections struct {
	buckets map[string]*cachedBucket
	lock    sync.Mutex
}

// removeOutdatedBuckets marks any active buckets for closure and removes the cached connections.
func (c *cachedBucketConnections) removeOutdatedBuckets(activeBuckets Set) {
	c.lock.Lock()
	defer c.lock.Unlock()
	for bucketName, bucket := range c.buckets {
		_, exists := activeBuckets[bucketName]
		if exists {
			continue
		}
		bucket.shouldClose = true
		c._teardown(bucketName)
	}
}

// closeAll removes all cached bucekts
func (c *cachedBucketConnections) closeAll() {
	c.lock.Lock()
	defer c.lock.Unlock()
	for _, bucket := range c.buckets {
		bucket.shouldClose = true
		bucket.bucketCloseFn()
	}
}

// teardown closes the cached bucket connection while locked, suitable for CouchbaseCluster.getBucket() teardowns
func (c *cachedBucketConnections) teardown(bucketName string) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.buckets[bucketName].refcount--
	c._teardown(bucketName)
}

// _teardown closes expects the lock to be acquired before calling this function and the reference count to be up to date.
func (c *cachedBucketConnections) _teardown(bucketName string) {
	if !c.buckets[bucketName].shouldClose || c.buckets[bucketName].refcount > 0 {
		return
	}
	c.buckets[bucketName].bucketCloseFn()
	delete(c.buckets, bucketName)
}

// get returns a cachedBucket for a given bucketName, or nil if it doesn't exist
func (c *cachedBucketConnections) _get(bucketName string) *cachedBucket {
	bucket, ok := c.buckets[bucketName]
	if !ok {
		return nil
	}
	c.buckets[bucketName].refcount++
	return bucket
}

// set adds a cachedBucket for a given bucketName, or nil if it doesn't exist
func (c *cachedBucketConnections) _set(bucketName string, bucket *cachedBucket) {
	c.buckets[bucketName] = bucket
}

var _ BootstrapConnection = &CouchbaseCluster{}

// NewCouchbaseCluster creates and opens a Couchbase Server cluster connection.
func NewCouchbaseCluster(ctx context.Context, clusterSpec CouchbaseClusterSpec,
	forcePerBucketAuth bool, perBucketCreds PerBucketCredentialsConfig,
	useXattrConfig bool, bucketMode BucketConnectionMode) (*CouchbaseCluster, error) {
	securityConfig, err := GoCBv2SecurityConfig(ctx, Ptr(clusterSpec.TLSSkipVerify), clusterSpec.CACertpath)
	if err != nil {
		return nil, err
	}

	clusterAuthConfig, err := GoCBv2Authenticator(
		clusterSpec.Username, clusterSpec.Password,
		clusterSpec.X509Certpath, clusterSpec.X509Keypath,
	)
	if err != nil {
		return nil, err
	}

	// Populate individual bucket credentials
	perBucketAuth := make(map[string]*gocb.Authenticator, len(perBucketCreds))
	for bucket, credentials := range perBucketCreds {
		authenticator, err := GoCBv2Authenticator(
			credentials.Username, credentials.Password,
			credentials.X509CertPath, credentials.X509KeyPath,
		)
		if err != nil {
			return nil, err
		}
		perBucketAuth[bucket] = &authenticator
	}

	clusterOptions := gocb.ClusterOptions{
		Authenticator:  clusterAuthConfig,
		SecurityConfig: securityConfig,
		TimeoutsConfig: GoCBv2TimeoutsConfig(nil, nil),
		RetryStrategy:  gocb.NewBestEffortRetryStrategy(nil),
	}

	cbCluster := &CouchbaseCluster{
		server:               clusterSpec.Server,
		forcePerBucketAuth:   forcePerBucketAuth,
		perBucketAuth:        perBucketAuth,
		clusterOptions:       clusterOptions,
		bucketConnectionMode: bucketMode,
		useGOCBFastFailRetry: clusterSpec.UseGOCBFastFailRetry,
	}

	if bucketMode == CachedClusterConnections {
		cbCluster.cachedBucketConnections = cachedBucketConnections{buckets: make(map[string]*cachedBucket)}
	}

	cbCluster.configPersistence = &DocumentBootstrapPersistence{}
	if useXattrConfig {
		cbCluster.configPersistence = &XattrBootstrapPersistence{}
	}

	return cbCluster, nil
}

// UseGOCBFastFailRetry returns whether gocb operations on this cluster should fail fast instead of using the
// best-effort retry strategy.
func (cc *CouchbaseCluster) UseGOCBFastFailRetry() bool {
	return cc.useGOCBFastFailRetry
}

// connect attempts to open a gocb.Cluster connection. Callers will be responsible for closing the connection.
// Pass an authenticator to use that to connect instead of using the cluster credentials.
func (cc *CouchbaseCluster) connect(auth *gocb.Authenticator) (*gocb.Cluster, error) {
	clusterOptions := cc.clusterOptions
	if auth != nil {
		clusterOptions.Authenticator = *auth
		clusterOptions.Username = ""
		clusterOptions.Password = ""
	}

	cluster, err := gocb.Connect(cc.server, clusterOptions)
	if err != nil {
		return nil, err
	}

	err = cluster.WaitUntilReady(time.Second*10, &gocb.WaitUntilReadyOptions{
		DesiredState:  gocb.ClusterStateOnline,
		ServiceTypes:  []gocb.ServiceType{gocb.ServiceTypeManagement},
		RetryStrategy: goCBRetryStrategy(cc.useGOCBFastFailRetry),
	})
	if err != nil {
		_ = cluster.Close(nil)
		return nil, err
	}

	return cluster, nil
}

func (cc *CouchbaseCluster) getClusterConnection() (*gocb.Cluster, error) {

	if cc.bucketConnectionMode == PerUseClusterConnections {
		return cc.connect(nil)
	}

	cc.cachedConnectionLock.Lock()
	defer cc.cachedConnectionLock.Unlock()
	if cc.cachedClusterConnection != nil {
		return cc.cachedClusterConnection, nil
	}

	clusterConnection, err := cc.connect(nil)
	if err != nil {
		return nil, err
	}
	cc.cachedClusterConnection = clusterConnection
	return cc.cachedClusterConnection, nil

}

func (cc *CouchbaseCluster) GetConfigBuckets(context.Context) ([]string, error) {
	if cc == nil {
		return nil, errors.New("nil CouchbaseCluster")
	}

	connection, err := cc.getClusterConnection()
	if err != nil {
		return nil, err
	}

	defer func() {
		if cc.bucketConnectionMode == PerUseClusterConnections {
			_ = connection.Close(nil)
		}
	}()

	buckets, err := connection.Buckets().GetAllBuckets(nil)
	if err != nil {
		cc.cachedClusterConnection = nil
		return nil, err
	}

	bucketList := make([]string, 0, len(buckets))
	for bucketName := range buckets {
		bucketList = append(bucketList, bucketName)
	}

	sort.Strings(bucketList)
	cc.cachedBucketConnections.removeOutdatedBuckets(SetOf(bucketList...))

	return bucketList, nil
}

func (cc *CouchbaseCluster) GetMetadataDocument(ctx context.Context, location, docID string, valuePtr any) (cas uint64, err error) {
	if cc == nil {
		return 0, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)

	if err != nil {
		return 0, err
	}

	defer teardown()

	cas, err = cc.configPersistence.loadConfig(ctx, b.DefaultCollection(), docID, valuePtr)
	SyncGatewayStats.GlobalStats.ResourceUtilizationStats().NumIdleKvOps.Add(1)
	return cas, err
}

func (cc *CouchbaseCluster) InsertMetadataDocument(ctx context.Context, location, key string, value any) (newCAS uint64, err error) {
	if cc == nil {
		return 0, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)
	if err != nil {
		return 0, err
	}
	defer teardown()

	return cc.configPersistence.insertConfig(b.DefaultCollection(), key, value)
}

// WriteMetadataDocument writes a metadata document, and fails on CAS mismatch
func (cc *CouchbaseCluster) WriteMetadataDocument(ctx context.Context, location, docID string, cas uint64, value any) (newCAS uint64, err error) {
	if cc == nil {
		return 0, errors.New("nil CouchbaseCluster")
	}
	if cas == 0 {
		return 0, RedactErrorf("CAS for %q in bucket %q must be non-zero to call WriteMetadataDocument, to add a new document use InsertMetadataDocument", MD(docID), MD(location))
	}

	b, teardown, err := cc.getBucket(ctx, location)
	if err != nil {
		return 0, err
	}
	defer teardown()

	rawDocument, err := JSONMarshal(value)
	if err != nil {
		return 0, err
	}

	casOut, err := cc.configPersistence.replaceRawConfig(b.DefaultCollection(), docID, rawDocument, gocb.Cas(cas))
	return uint64(casOut), err
}

func (cc *CouchbaseCluster) TouchMetadataDocument(ctx context.Context, location, docID string, property, value string, cas uint64) (newCAS uint64, err error) {

	if cc == nil {
		return 0, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)
	if err != nil {
		return 0, err
	}
	defer teardown()

	casOut, err := cc.configPersistence.touchConfigRollback(b.DefaultCollection(), docID, property, value, gocb.Cas(cas))
	return uint64(casOut), err

}

func (cc *CouchbaseCluster) DeleteMetadataDocument(ctx context.Context, location, key string, cas uint64) (err error) {
	if cc == nil {
		return errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)
	if err != nil {
		return err
	}
	defer teardown()

	_, removeErr := cc.configPersistence.removeRawConfig(b.DefaultCollection(), key, gocb.Cas(cas))
	return removeErr
}

// UpdateMetadataDocument retries on CAS mismatch
func (cc *CouchbaseCluster) UpdateMetadataDocument(ctx context.Context, location, docID string, updateCallback func(bucketConfig []byte, rawBucketConfigCas uint64) (newConfig []byte, err error)) (newCAS uint64, err error) {
	if cc == nil {
		return 0, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)
	if err != nil {
		return 0, err
	}
	defer teardown()

	collection := b.DefaultCollection()

	for {
		bucketValue, cas, err := cc.configPersistence.loadRawConfig(ctx, collection, docID)
		if err != nil {
			return 0, err
		}
		newConfig, err := updateCallback(bucketValue, uint64(cas))
		if err != nil {
			return 0, err
		}

		// handle delete when updateCallback returns nil
		if newConfig == nil {
			removeCasOut, err := cc.configPersistence.removeRawConfig(collection, docID, cas)
			if err != nil {
				// retry on cas failure
				if errors.Is(err, gocb.ErrCasMismatch) {
					continue
				}
				return 0, err
			}
			return uint64(removeCasOut), nil
		}

		replaceCfgCasOut, err := cc.configPersistence.replaceRawConfig(collection, docID, newConfig, cas)
		if err != nil {
			if errors.Is(err, gocb.ErrCasMismatch) {
				// retry on cas failure
				continue
			}
			return 0, err
		}

		return uint64(replaceCfgCasOut), nil
	}

}

// KeyExists checks whether a key exists in the default collection for the specified bucket
func (cc *CouchbaseCluster) KeyExists(ctx context.Context, location, docID string) (exists bool, err error) {
	if cc == nil {
		return false, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, location)

	if err != nil {
		return false, err
	}

	defer teardown()

	return cc.configPersistence.keyExists(b.DefaultCollection(), docID)
}

// GetDocument fetches a document from the default collection.  Does not use configPersistence - callers
// requiring configPersistence handling should use GetMetadataDocument.
func (cc *CouchbaseCluster) GetDocument(ctx context.Context, bucketName, docID string, rv any) (exists bool, err error) {
	if cc == nil {
		return false, errors.New("nil CouchbaseCluster")
	}

	b, teardown, err := cc.getBucket(ctx, bucketName)
	if err != nil {
		return false, err
	}

	defer teardown()

	getOptions := &gocb.GetOptions{
		Transcoder: NewSGJSONTranscoder(),
	}
	getResult, err := b.DefaultCollection().Get(docID, getOptions)
	if err != nil {
		if errors.Is(err, gocb.ErrDocumentNotFound) {
			return false, nil
		}
		return false, err
	}
	err = getResult.Content(rv)
	return true, err
}

// GetRawDocument fetches a document from the default collection as raw bytes.  Does not use configPersistence - callers
// requiring configPersistence handling should use GetMetadataDocument.
func (cc *CouchbaseCluster) GetRawDocument(ctx context.Context, bucketName, docID string) (value []byte, exists bool, err error) {
	if cc == nil {
		return nil, false, errors.New("nil CouchbaseCluster")
	}
	b, teardown, err := cc.getBucket(ctx, bucketName)
	if err != nil {
		return nil, false, err
	}

	defer teardown()
	getOptions := &gocb.GetOptions{
		Transcoder: NewSGRawTranscoder(),
	}
	getResult, err := b.DefaultCollection().Get(docID, getOptions)
	if err != nil {
		if errors.Is(err, gocb.ErrDocumentNotFound) {
			return nil, false, nil
		}
		return nil, false, err
	}
	err = getResult.Content(&value)
	return value, true, err
}

// Close calls teardown for any cached buckets and removes from cachedBucketConnections
func (cc *CouchbaseCluster) Close() {

	cc.cachedBucketConnections.closeAll()

	cc.cachedConnectionLock.Lock()
	defer cc.cachedConnectionLock.Unlock()

	if cc.cachedClusterConnection != nil {
		_ = cc.cachedClusterConnection.Close(nil)
		cc.cachedClusterConnection = nil
	}
}

func (cc *CouchbaseCluster) getBucket(ctx context.Context, bucketName string) (b *gocb.Bucket, teardownFn func(), err error) {

	if cc.bucketConnectionMode != CachedClusterConnections {
		return cc.connectToBucket(ctx, bucketName)
	}

	teardownFn = func() {
		cc.cachedBucketConnections.teardown(bucketName)
	}
	cc.cachedBucketConnections.lock.Lock()
	defer cc.cachedBucketConnections.lock.Unlock()
	bucket := cc.cachedBucketConnections._get(bucketName)
	if bucket != nil {
		return bucket.bucket, teardownFn, nil
	}

	// cached bucket not found, connect and add
	newBucket, bucketCloseFn, err := cc.connectToBucket(ctx, bucketName)
	if err != nil {
		return nil, nil, err
	}
	cc.cachedBucketConnections._set(bucketName, &cachedBucket{
		bucket:        newBucket,
		bucketCloseFn: bucketCloseFn,
		refcount:      1,
	})

	return newBucket, teardownFn, nil
}

func (cc *CouchbaseCluster) GetClusterConnectionForBucket(ctx context.Context, bucketName string) (connection *gocb.Cluster, teardownFn func(), err error) {
	if bucketAuth, set := cc.perBucketAuth[bucketName]; set {
		connection, err = cc.connect(bucketAuth)
	} else if cc.forcePerBucketAuth {
		return nil, nil, fmt.Errorf("unable to get bucket %q since credentials are not defined in bucket_credentials", MD(bucketName).Redact())
	} else {
		connection, err = cc.connect(nil)
	}
	if err != nil {
		return nil, nil, err
	}

	teardownFn = func() {
		err := connection.Close(&gocb.ClusterCloseOptions{})
		if err != nil {
			WarnfCtx(ctx, "Failed to close cluster connection: %v", err)
		}
	}
	return connection, teardownFn, nil
}

// connectToBucket establishes a new connection to a bucket, and returns the bucket after waiting for it to be ready.
func (cc *CouchbaseCluster) connectToBucket(ctx context.Context, bucketName string) (b *gocb.Bucket, teardownFn func(), err error) {
	connection, teardownFn, err := cc.GetClusterConnectionForBucket(ctx, bucketName)
	if err != nil {
		return nil, nil, err
	}

	b = connection.Bucket(bucketName)
	err = b.WaitUntilReady(time.Second*10, &gocb.WaitUntilReadyOptions{
		DesiredState:  gocb.ClusterStateOnline,
		RetryStrategy: goCBRetryStrategy(cc.useGOCBFastFailRetry),
		ServiceTypes:  []gocb.ServiceType{gocb.ServiceTypeKeyValue},
	})
	if err != nil {
		teardownFn()

		if errors.Is(err, gocb.ErrAuthenticationFailure) {
			return nil, nil, ErrAuthError
		}

		// In best-effort retry mode a missing/unreachable bucket surfaces as a timeout rather than an
		// auth failure. Classify it as a connection error so callers translate it to a 502 instead of a
		// generic 500, mirroring the per-database connection path (see db.connectToBucketErrorHandling).
		if !cc.useGOCBFastFailRetry {
			return nil, nil, HTTPErrorf(http.StatusBadGateway,
				"Unable to connect to Couchbase Server. Please ensure it is running and reachable, and that bucket %q exists. Error: %s", MD(bucketName).Redact(), err)
		}
		return nil, nil, err
	}

	return b, teardownFn, nil
}

type PerBucketCredentialsConfig map[string]*CredentialsConfig

type CredentialsConfig struct {
	Username string `json:"username,omitempty"       help:"Username for authenticating to the bucket"`
	Password string `json:"password,omitempty"       help:"Password for authenticating to the bucket"`
	CredentialsConfigX509
}

type CredentialsConfigX509 struct {
	X509CertPath string `json:"x509_cert_path,omitempty" help:"Cert path (public key) for X.509 bucket auth"`
	X509KeyPath  string `json:"x509_key_path,omitempty"  help:"Key path (private key) for X.509 bucket auth"`
}

// ConfigMerge applies non-empty fields from b onto non-empty fields on a
func ConfigMerge(a, b any) error {
	return mergo.Merge(a, b, mergo.WithTransformers(&mergoNilTransformer{}), mergo.WithOverride)
}

// mergoNilTransformer is a mergo.Transformers implementation that treats non-nil zero values as non-empty when merging.
type mergoNilTransformer struct{}

var _ mergo.Transformers = &mergoNilTransformer{}

func (t *mergoNilTransformer) Transformer(typ reflect.Type) func(dst, src reflect.Value) error {
	if typ.Kind() == reflect.Pointer {
		if typ.Elem().Kind() == reflect.Struct {
			// skip nilTransformer for structs, to allow recursion
			return nil
		}
		return func(dst, src reflect.Value) error {
			if dst.CanSet() && !src.IsNil() {
				dst.Set(src)
			}
			return nil
		}
	}
	return nil
}