package couchctl

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"time"

	"edge-infra.dev/pkg/k8s/runtime/inventory"
	unstructuredutil "edge-infra.dev/pkg/k8s/unstructured"

	corev1 "k8s.io/api/core/v1"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/dynamic"
	kuberecorder "k8s.io/client-go/tools/record"

	"sigs.k8s.io/cli-utils/pkg/kstatus/watcher"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"edge-infra.dev/pkg/edge/api/utils"
	dsapi "edge-infra.dev/pkg/edge/datasync/apis/v1alpha1"
	"edge-infra.dev/pkg/edge/datasync/couchdb"
	"edge-infra.dev/pkg/k8s/meta/status"
	"edge-infra.dev/pkg/k8s/runtime/conditions"
	"edge-infra.dev/pkg/k8s/runtime/controller/metrics"
	"edge-infra.dev/pkg/k8s/runtime/controller/reconcile"
	"edge-infra.dev/pkg/k8s/runtime/controller/reconcile/recerr"
	"edge-infra.dev/pkg/k8s/runtime/patch"
	"edge-infra.dev/pkg/k8s/runtime/sap"

	"github.com/go-logr/logr"
)

type CouchUserReconciler struct {
	client.Client
	NodeResourcePredicate
	kuberecorder.EventRecorder
	SecretManager   secretManager
	Name            string
	Config          *Config
	Metrics         metrics.Metrics
	patchOptions    []patch.Option
	ResourceManager *sap.ResourceManager
	replicationDB   string
}

var (
	ErrNewUserNotFoundInDB = errors.New("new user not found in users table")

	userConditions = reconcile.Conditions{
		Target: status.ReadyCondition,
		Owned: []string{
			dsapi.UserSetupSucceededReason,
			status.ReadyCondition,
			status.ReconcilingCondition,
			status.StalledCondition,
		},
		Summarize: []string{
			dsapi.UserSetupSucceededReason,
			status.StalledCondition,
		},
		NegativePolarity: []string{
			status.ReconcilingCondition,
			status.StalledCondition,
		},
	}
	edgeAdminNamespaces = []string{couchdb.Namespace, "cushion", "edge-iam"}
)

// SetupWithManager sets up ComputeAddressReconciler with the manager
func (r *CouchUserReconciler) SetupWithManager(mgr ctrl.Manager) error {
	r.replicationDB = r.Config.ReplicationDB()
	r.patchOptions = getPatchOptions(serverConditions.Owned, r.Name)
	d, err := dynamic.NewForConfig(mgr.GetConfig())
	if err != nil {
		return fmt.Errorf("fail to create dynamic client: %w", err)
	}
	r.ResourceManager = sap.NewResourceManager(
		r.Client,
		watcher.NewDefaultStatusWatcher(d, mgr.GetRESTMapper()),
		sap.Owner{Field: r.Name},
	)
	return ctrl.NewControllerManagedBy(mgr).
		For(&dsapi.CouchDBUser{}, r.userPredicates()).
		Owns(&corev1.Secret{}, r.secretPredicates()).
		Complete(r)
}

func (r *CouchUserReconciler) userPredicates() builder.Predicates {
	return builder.WithPredicates(
		predicate.GenerationChangedPredicate{},
		predicate.NewPredicateFuncs(func(obj client.Object) bool {
			if r.Config.IsDSDS() {
				return r.ShouldReconcile(r.Config, obj)
			}
			return true
		}))
}

func (r *CouchUserReconciler) secretPredicates() builder.Predicates {
	return builder.WithPredicates(
		predicate.GenerationChangedPredicate{},
		predicate.NewPredicateFuncs(func(obj client.Object) bool {
			if r.Config.IsDSDS() {
				list := dsapi.CouchDBUserList{}
				if err := r.Client.List(context.Background(), &list,
					client.InNamespace(obj.GetNamespace()), client.MatchingLabels{
						couchdb.NodeUIDLabel: r.Config.NodeUID,
					}); err != nil {
					return false
				}
				if len(list.Items) == 0 {
					return false
				}
				return isChild(obj, list.Items)
			}
			return true
		}))
}

func isChild(obj client.Object, items []dsapi.CouchDBUser) bool {
	// ID: '<namespace>_<name>_<group>_<kind>'
	secretID := fmt.Sprintf("%s_%s__Secret", obj.GetNamespace(), obj.GetName())
	for _, item := range items {
		if item.Status.Inventory == nil || item.Status.Inventory.Entries == nil {
			continue
		}
		for _, entry := range item.Status.Inventory.Entries {
			if entry.ID == secretID {
				return true
			}
		}
	}
	return false
}

func (r *CouchUserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, err error) {
	reconcileStart := time.Now()
	log := ctrl.LoggerFrom(ctx)

	user := &dsapi.CouchDBUser{}
	if err := r.Client.Get(ctx, req.NamespacedName, user); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	user.WithRetry(r.Config.RequeueTime)
	user.WithInterval(r.Config.PollingInterval)
	withUserType(user)

	ctx = logr.NewContext(ctx, log)

	patcher := patch.NewSerialPatcher(user, r.Client)
	if err := reconcile.Progressing(ctx, user, patcher, r.patchOptions...); err != nil {
		log.Error(err, "unable to update status")
		return ctrl.Result{}, err
	}

	recResult := reconcile.ResultEmpty
	var recErr recerr.Error

	defer func() {
		summarizer := reconcile.NewSummarizer(patcher)
		res, err = summarizer.SummarizeAndPatch(ctx, user, []reconcile.SummarizeOption{
			reconcile.WithConditions(userConditions),
			reconcile.WithResult(recResult),
			reconcile.WithError(recErr),
			reconcile.WithIgnoreNotFound(),
			reconcile.WithProcessors(
				reconcile.RecordResult,
			),
			reconcile.WithFieldOwner(r.Name),
			reconcile.WithEventRecorder(r.EventRecorder),
		}...)
		r.Metrics.RecordDuration(ctx, user, reconcileStart)
		r.Metrics.RecordReadiness(ctx, user)
	}()

	// Check if finalizer exists
	if !controllerutil.ContainsFinalizer(user, DatasyncFinalizer) {
		controllerutil.AddFinalizer(user, DatasyncFinalizer)
		// Return immediately so that we requeue and reconcile object with finalizer
		// added.
		recResult = reconcile.ResultRequeue
		return
	}

	if !user.ObjectMeta.DeletionTimestamp.IsZero() {
		recErr = r.finalize(ctx, user)
		return
	}

	if recErr = r.reconcile(ctx, user); recErr != nil {
		// CouchDBServer controller is responsible for surfacing couchdb Errors
		if !couchDBNotReadyOrNotFound(recErr) {
			recErr.ToCondition(user, dsapi.UserSetupSucceededReason)
			err = recErr
			return
		}
	}
	recResult = reconcile.ResultSuccess
	conditions.MarkTrue(user, dsapi.UserSetupSucceededReason, status.SucceededReason, "Successfully created CouchDB User")
	log.Info("Successfully created CouchDB User")

	return
}

// finalize checks is couchdb node exists removing deleting
func (r *CouchUserReconciler) finalize(ctx context.Context, user *dsapi.CouchDBUser) (recErr recerr.Error) {
	log := logr.FromContextOrDiscard(ctx)
	log.Info("running finalizer")

	var prune bool
	defer func() {
		if prune {
			log.Info("pruning user's resources")
			err := pruneInventory(ctx, r.ResourceManager, user)
			if err != nil {
				recErr = recerr.New(err, status.DependencyInvalidReason)
			}
			controllerutil.RemoveFinalizer(user, DatasyncFinalizer)
			log.Info("finalizer ran successfully")
		}
	}()

	nodeFound, err := r.nodeExistsAndSchedulable(ctx, user)
	if err != nil {
		return recerr.New(err, status.DependencyInvalidReason)
	}

	// node not found, assume couchdb server is gone
	if !nodeFound {
		prune = true
		return
	}

	// get couchdb client to be able to delete username
	cc, err := couchDBClient(ctx, r.Client, r.Config, user)
	switch {
	case err != nil && serverNotFound(err):
		prune = true
		return
	case err != nil:
		return recerr.New(err, status.DependencyInvalidReason)
	}

	// Get optional username to delete from couchdb
	username := user.Spec.User.Name
	if username == "" {
		up := &couchdb.UsernamePassword{}
		_, err = up.FromSecret(ctx, r.Client, user.SecretNamespacedName())
		switch {
		case err != nil && kerrors.IsNotFound(err):
			prune = true
			return
		case err != nil:
			return recerr.New(err, status.DependencyInvalidReason)
		}
		username = string(up.Username)
	}

	err = cc.DeleteUser(ctx, username)
	if err != nil && !couchdb.IsNotFound(err) {
		logError(ctx, err, "fail to delete user from couchdb", "username", username)
		return recerr.New(err, status.DependencyInvalidReason)
	}
	prune = true
	return
}

// nodeExistsAndSchedulable check for given user if the couchdb server node exists and schedulable
// we could check if the sts/pvc exist instead
func (r *CouchUserReconciler) nodeExistsAndSchedulable(ctx context.Context, user *dsapi.CouchDBUser) (bool, error) {
	var nodeUID string
	_, server, err := checkIfServerIsReady(ctx, r.Client, user)
	switch {
	case err != nil && !kerrors.IsNotFound(err):
		return false, err
	case server != nil:
		nodeUID = labelOrEmpty(server, couchdb.NodeUIDLabel)
	default:
		nodeUID = labelOrEmpty(user, couchdb.NodeUIDLabel)
	}

	if nodeUID == "" {
		return false, nil
	}

	nodes := &corev1.NodeList{}
	err = r.Client.List(ctx, nodes, &client.ListOptions{})
	if err != nil {
		return false, err
	}
	for _, node := range nodes.Items {
		if nodeUID == string(node.UID) {
			return !node.Spec.Unschedulable, nil
		}
	}
	return false, nil
}

func (r *CouchUserReconciler) reconcile(ctx context.Context, user *dsapi.CouchDBUser) recerr.Error {
	log := logr.FromContextOrDiscard(ctx)

	// attempt to get the couchdbserver
	ready, server, err := checkIfServerIsReady(ctx, r.Client, user)
	if err != nil {
		return recerr.NewWait(err, status.DependencyNotReadyReason, r.Config.ServerNotReady)
	}

	if !ready {
		err := fmt.Errorf("%w", ErrServerNotReady)
		return recerr.NewWait(err, status.DependencyNotReadyReason, r.Config.ServerNotReady)
	}

	log = log.WithValues("server", client.ObjectKeyFromObject(server))
	ctx = logr.NewContext(ctx, log)

	// Errors are always returned
	return r.reconcileUser(ctx, user, server)
}

func (r *CouchUserReconciler) authorizeUser(user *dsapi.CouchDBUser) (bool, error) {
	if user.IsAdminCredentials() && !utils.Contains(edgeAdminNamespaces, user.Namespace) {
		return false, fmt.Errorf("admin users can only be created in edge admin namespaces: %v", edgeAdminNamespaces)
	}
	return true, nil
}

func (r *CouchUserReconciler) reconcileUser(ctx context.Context, user *dsapi.CouchDBUser, server *dsapi.CouchDBServer) recerr.Error {
	log := logr.FromContextOrDiscard(ctx)
	// TODO: if the user passed in a secret get that and use it to set the new users username and password

	changeSet := sap.NewChangeSet()

	// get / check if the admin creds exist
	cc, err := couchDBClient(ctx, r.Client, r.Config, user)
	switch {
	case err != nil && serverNotFound(err):
		return recerr.NewWait(err, status.DependencyNotReadyReason, r.Config.ServerNotReady)
	case err != nil:
		logError(ctx, err, "error creating couchdb client", "NamespacedName", server.AdminCredentials())
		return recerr.NewWait(err, status.DependencyInvalidReason, r.Config.ServerNotReady)
	}

	// defer closing the client
	defer cc.Close(ctx)

	authorized, err := r.authorizeUser(user)
	if !authorized {
		return recerr.NewStalled(err, dsapi.UserCreationFailedReason)
	}

	// TODO: check if a user already exists and verify the secret is valid
	// and that the creds in the secret actually map to a user with the correct permissions
	// actually maybe this doesnt matter due to the GIGO nature

	userCreds := r.credentials(user, server)

	// try to get the user credentials secret
	// if it doesn't exist create it
	var cs *sap.ChangeSetEntry
	secretNN := user.SecretNamespacedName()
	secret, err := userCreds.FromSecret(ctx, r.Client, secretNN)
	switch {
	case err != nil && canMigrate(err):
		var ownerRefs []v1.OwnerReference
		if user.IsUserCredentials() {
			ownerRefs = couchDBUserOwnerReference(user)
		}
		secret, err := userCreds.ToSecret(ctx, r.Client, secretNN, ownerRefs...)
		if err != nil {
			logError(ctx, err, "fail to create couchdb k8 secret", "NamespacedName", secretNN)
			return recerr.New(err, dsapi.UserCreationFailedReason)
		}
		un, err := unstructuredutil.ToUnstructured(secret)
		if err != nil {
			return recerr.New(err, dsapi.UserCreationFailedReason)
		}
		cs, err = r.ResourceManager.Apply(ctx, un, sap.ApplyOptions{})
		if err != nil {
			return recerr.New(err, dsapi.UserCreationFailedReason)
		}
		changeSet.Add(*cs)
	case err != nil:
		log.Error(err, "fail to get couchdb k8 secret", "NamespacedName", secretNN)
		return recerr.New(err, status.DependencyInvalidReason)
	default:
		cs, err = ExistingChangeSetEntry(secret)
		if err != nil {
			return recerr.New(err, status.DependencyInvalidReason)
		}
		changeSet.Add(*cs)
	}

	usernamePassword := usernamePassword(userCreds, user)
	userName := string(usernamePassword.Username)
	log = log.WithValues("username", userName)

	// TODO: check if the users creds are defined in the spec via secret
	_, err = cc.CreateNewUser(ctx, userName, string(usernamePassword.Password), user.Spec.User.Roles)
	if err != nil && !errors.Is(err, couchdb.ErrConflict) {
		logError(ctx, err, "fail to create couchdb user", "NamespacedName", secretNN)
		return recerr.NewWait(err, dsapi.UserFailedToAddUserReason, r.Config.ServerNotReady)
	}
	var roles []string
	if !user.IsUserCredentials() {
		roles = user.Spec.User.Roles
	}
	// ensure the user and its roles have been successfully added to couchdb
	userExists, err := cc.CheckUserAndRolesExists(ctx, userName, roles)
	if !userExists || errors.Is(err, couchdb.ErrNotFound) {
		err := fmt.Errorf("%w", ErrNewUserNotFoundInDB)
		logError(ctx, err, "fail to check roles for couchdb user, user or role foes not exists", "NamespacedName", secretNN)
		return recerr.NewWait(err, dsapi.UserFailedToAddUserReason, r.Config.ServerNotReady)
	}
	if err != nil {
		log.Error(err, "fail to check roles for couchdb user", "NamespacedName", secretNN)
		return recerr.NewWait(err, dsapi.UserFailedToAddUserReason, r.Config.ServerNotReady)
	}

	// TODO: if the user isn't successfully created in couch should we
	// remove the new secret and try again next recon?

	// Add user to database security if provider is provided or has secure role
	if hasDBSecurityRole(user) { //nolint: nestif
		recErr := r.reconcileRoleBasedUser(ctx, user, cc, userName)
		if recErr != nil {
			return recErr
		}
	}

	// store secret in secret manager if label exists
	if server.IsCloud() && user.AddToSecretManager() {
		userRef := corev1.SecretReference{Namespace: user.Namespace, Name: user.Name}
		couchReplicationCred := server.ReplicationCredentials() == userRef
		err = r.CredsToSecretManager(ctx, user, userCreds, couchReplicationCred)
		if err != nil {
			return recerr.New(err, string(dsapi.ReplicationCreationFailedStatus))
		}
	}

	// TODO create service account to read secret

	i := inventory.New(inventory.FromSapChangeSet(changeSet))
	if err := r.prune(ctx, user, i); err != nil {
		return recerr.New(err, dsapi.PruneFailed)
	}
	user.Status.Inventory = i

	return nil
}

func (r *CouchUserReconciler) prune(ctx context.Context, user *dsapi.CouchDBUser, i *inventory.ResourceInventory) error {
	log := logr.FromContextOrDiscard(ctx)
	if user.Status.Inventory != nil {
		diff, err := inventory.Diff(user.Status.Inventory, i)
		if err != nil {
			return nil
		}
		if len(diff) > 0 {
			changeSet, err := r.ResourceManager.DeleteAll(ctx, diff, sap.DefaultDeleteOptions())
			if err != nil {
				return err
			}
			log.Info("pruned objects", "changeset", changeSet.ToMap())
		}
	}
	return nil
}

// reconcileRoleBasedUser adds the user's roles to replicated databases
func (r *CouchUserReconciler) reconcileRoleBasedUser(
	ctx context.Context,
	user *dsapi.CouchDBUser,
	cc *couchdb.CouchDB,
	userName string) recerr.Error {
	log := logr.FromContextOrDiscard(ctx)

	replDB := r.Config.ReplicationDB()
	replSet := &dsapi.ReplicationSet{}
	if err := cc.GetReplicationSetDoc(ctx, replDB, replSet); err != nil {
		log.Info("replication doc not found, skipping creation of replication creds", "dbname", replDB)
		return nil
	}

	// we need to add username to replication db's security, so it can be replicated
	if user.IsReplicationCredentials() {
		if recErr := r.addUsernameToSecurity(ctx, user, cc, userName, replDB); recErr != nil {
			return recErr
		}
	}

	for _, d := range replSet.Datasets {
		if !shouldReplicate(r.Config, d, r.replicationDB) {
			// if the database is not replicated, it will not exist.
			// TODO add test for user controller
			continue
		}
		if !updateDBSecurityForUser(user, d) {
			continue
		}
		if recErr := r.addUsernameToSecurity(ctx, user, cc, userName, d.Name); recErr != nil {
			return recErr
		}
	}
	return nil
}

func canMigrate(err error) bool {
	return kerrors.IsNotFound(err) ||
		errors.Is(err, couchdb.ErrCouchDBURIMissing) ||
		errors.Is(err, couchdb.ErrDBNameMissing) ||
		errors.Is(err, couchdb.ErrSecretDataMissing) ||
		errors.Is(err, couchdb.ErrInvalidCookieSecret)
}

func (r *CouchUserReconciler) addUsernameToSecurity(
	ctx context.Context,
	user *dsapi.CouchDBUser,
	cc *couchdb.CouchDB,
	userName, dbName string) recerr.Error {
	// check that the DB actually exists in couchdb
	exists, err := cc.CheckIfDBExists(ctx, dbName)
	if err != nil {
		// TODO add error to error list and continue
		return recerr.NewWait(err, dsapi.DatabaseCreationFailedReason, r.Config.ServerNotReady)
	}
	if !exists {
		err := fmt.Errorf("%w", ErrNewDatabaseNotFound)
		// TODO add error to error list and continue
		return recerr.NewWait(err, dsapi.DatabaseCreationFailedReason, r.Config.DatabaseNotFound)
	}
	security := userToSecurity(user, userName)

	// verify that the members / roles exist
	exists, err = cc.CheckDBUsersAndRoles(ctx, security, dbName)
	if err != nil {
		err = fmt.Errorf("error checking for couchdb users and roles")
		return recerr.NewWait(err, dsapi.UserCreationFailedReason, r.Config.DatabaseNotFound)
	}
	if exists { // no need to update security doc
		return nil
	}

	// add the user defined members / roles to the new db
	err = cc.AddMemberUserAndRolesToDB(ctx, security, dbName)
	if err != nil {
		return recerr.NewWait(err, dsapi.DatabaseAddRolesFailedReason, r.Config.DatabaseNotFound)
	}
	return nil
}

// CredsToSecretManager takes in user and attempts to add the fields as secrets to secretmanager
func (r *CouchUserReconciler) CredsToSecretManager(ctx context.Context, user *dsapi.CouchDBUser, userCreds couchdb.CredentialsManager, couchReplicationCred bool) error {
	sms, err := r.toSecretManagerSecret(userCreds)
	if err != nil {
		return fmt.Errorf("failed to convert secret to secret manager: %w", err)
	}

	value, err := json.Marshal(sms)
	if err != nil {
		return fmt.Errorf("failed to convert ReplicationCredentials to json string for Secret Manager: %w", err)
	}

	// create a new instance of secretmanager
	writer, err := r.SecretManager.NewWithOptions(ctx, r.Config.ProjectID)
	if err != nil {
		return err
	}

	var secretName string // Maximum 255 characters
	if couchReplicationCred {
		secretName = couchdb.ReplicationSMgrSecretName
	} else {
		secretName = fmt.Sprintf("%s-%s", user.Namespace, user.Name)
	}
	return writer.AddSecret(ctx, secretName, value, nil, false, nil, "")
}

func (r *CouchUserReconciler) toSecretManagerSecret(creds couchdb.CredentialsManager) (*couchdb.SecretManagerSecret, error) {
	sms := &couchdb.SecretManagerSecret{}
	switch s := creds.(type) {
	case *couchdb.AdminCredentials:
		sms.Username = string(s.Username)
		sms.Password = string(s.Password)
	case *couchdb.ReplicationCredentials:
		sms.Username = string(s.Username)
		sms.Password = string(s.Password)
		sms.URI = string(s.URI)
		sms.DBName = string(s.DBName)
	case *couchdb.UserCredentials:
		sms.Username = string(s.Username)
		sms.Password = string(s.Password)
		sms.URI = string(s.URI)
	default:
		return nil, fmt.Errorf("credentials of type %T not implemented for secret manager secret", s)
	}
	return sms, nil
}

func userToSecurity(user *dsapi.CouchDBUser, userName string) couchdb.Security {
	userRoles := user.Spec.User.Roles
	security := couchdb.Security{}
	for _, role := range userRoles {
		if role == couchdb.CreateViewUser {
			addNameToSecurity(&security, userName, true)
		} else if role == couchdb.ReadOnlyUser {
			addNameToSecurity(&security, userName, false)
		} else {
			addNameToSecurity(&security, userName, false)
			addRoleToSecurity(&security, role, false)
		}
	}
	return security
}

func addNameToSecurity(security *couchdb.Security, name string, admin bool) {
	if admin {
		if !utils.Contains(security.Admins.Names, name) {
			security.Admins.Names = append(security.Admins.Names, name)
		}
	} else {
		if !utils.Contains(security.Members.Names, name) {
			security.Members.Names = append(security.Members.Names, name)
		}
	}
}

func addRoleToSecurity(security *couchdb.Security, role string, admin bool) {
	if admin {
		if !utils.Contains(security.Admins.Roles, role) {
			security.Admins.Roles = append(security.Admins.Roles, role)
		}
	} else {
		if !utils.Contains(security.Members.Roles, role) {
			security.Members.Roles = append(security.Members.Roles, role)
		}
	}
}

func hasDBSecurityRole(user *dsapi.CouchDBUser) bool {
	return !user.Spec.Provider.Empty() || user.IsUserCredentials() || user.IsReplicationCredentials()
}

// withUserType ensure backward compatibility by explicitly setting the user type
func withUserType(user *dsapi.CouchDBUser) {
	if user.Spec.Type == "" {
		switch {
		case utils.Contains(user.Spec.User.Roles, couchdb.ReplicationUser):
			user.Spec.Type = dsapi.ReplicationCredentials
		case utils.Contains(user.Spec.User.Roles, couchdb.ReadOnlyUser) ||
			utils.Contains(user.Spec.User.Roles, couchdb.CreateViewUser):
			user.Spec.Type = dsapi.UserCredentials
		default:
			user.Spec.Type = dsapi.AdminCredentials
		}
	}
}

func updateDBSecurityForUser(user *dsapi.CouchDBUser, d dsapi.Dataset) bool {
	if user.Spec.Type == dsapi.ReplicationCredentials {
		return true
	}
	userProvider := user.Spec.Provider
	dbProvider := d.Provider
	if userProvider.Empty() && dbProvider.Empty() {
		return true
	}
	if !userProvider.Empty() && !dbProvider.Empty() {
		return userProvider.Name == dbProvider.Name
	}
	if dbProvider.Empty() {
		return true
	}
	return false
}

func couchDBUserOwnerReference(user *dsapi.CouchDBUser) []v1.OwnerReference {
	return []v1.OwnerReference{
		*v1.NewControllerRef(user,
			dsapi.GroupVersion.WithKind(reflect.TypeOf(dsapi.CouchDBUser{}).Name())),
	}
}

func (r *CouchUserReconciler) credentials(user *dsapi.CouchDBUser, server *dsapi.CouchDBServer) couchdb.CredentialsManager {
	up := couchdb.UsernamePassword{Username: []byte(user.Spec.User.Name)}
	switch user.Spec.Type {
	case dsapi.UserCredentials:
		return &couchdb.UserCredentials{
			UsernamePassword: up,
			URI:              []byte(getCouchDBUserURL(r.Config, server)),
		}
	case dsapi.ReplicationCredentials:
		return &couchdb.ReplicationCredentials{
			UserCredentials: couchdb.UserCredentials{
				UsernamePassword: up,
				URI:              []byte(getServerURL(r.Config, server)),
			},
			DBName: []byte(r.Config.ReplicationDB()),
		}
	default:
		return &couchdb.AdminCredentials{UsernamePassword: up}
	}
}

func usernamePassword(creds couchdb.CredentialsManager, user *dsapi.CouchDBUser) couchdb.UsernamePassword {
	switch user.Spec.Type {
	case dsapi.UserCredentials:
		return creds.(*couchdb.UserCredentials).UsernamePassword
	case dsapi.ReplicationCredentials:
		return creds.(*couchdb.ReplicationCredentials).UsernamePassword
	default:
		return creds.(*couchdb.AdminCredentials).UsernamePassword
	}
}