// Copyright (c) 2025 OpenBao a Series of LF Projects, LLC
// SPDX-License-Identifier: MPL-2.0

package core

import (
	"context"
	"crypto/subtle"
	"errors"
	"fmt"
	"net/http"
	"path"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	metrics "github.com/hashicorp/go-metrics/compat"
	"github.com/hashicorp/go-multierror"
	"github.com/hashicorp/go-secure-stdlib/base62"
	uuid "github.com/hashicorp/go-uuid"
	"github.com/openbao/openbao/sdk/v2/logical"
	"github.com/stephnangue/warden/internal/fairshare"
	"github.com/stephnangue/warden/internal/namespace"
	"github.com/stephnangue/warden/logger"
	wlogical "github.com/stephnangue/warden/logical"
)

const namespaceIdLength = 6

const (
	// Namespace storage location.
	namespaceStoreSubPath = "core/namespaces/"

	// namespaceBarrierPrefix is the prefix to the UUID of a namespaces
	// used in the barrier view for the namespace-owned backends.
	namespaceBarrierPrefix = "namespaces/"

	// nsDispatcherName is the name of the jobmanager instance for metrics.
	nsDispatcherName = "namespace-deletion"

	// nsMaxWorkers is the number of parallel namespace deletion workers. This
	// is set conservatively low due to different lock acquisitions; see
	// clearNamespaceResources for exact details. Add two to the number of
	// locks to handle the global namespace store lock acquisition and
	// overhead.
	nsMaxWorkers = 2 + /* namespace and overhead */
		1 + /* rotation (sync.Map) */
		1 + /* expiration (sync.Map) */
		1 + /* tokens (ristretto + storage) */
		1 + /* credential configs (RLock) */
		1 + /* policies */
		2 /* auth + provider */
)

// NamespaceStore is used to provide durable storage of namespace. It is
// a singleton store across the Core and contains all child namespaces.
type NamespaceStore struct {
	core    *Core
	storage logical.Storage

	// This lock ensures we don't concurrently modify the store while using
	// a namespace entry. We also store an atomic to check if we need to
	// reload all namespaces.
	lock        sync.RWMutex
	invalidated atomic.Bool

	// List of all namespaces within the store. This is loaded at store
	// initialization time and persisted throughout the lifetime of the
	// instance. Entries should not be returned directly but instead be
	// copied to prevent modification.
	namespacesByPath     *namespaceTree
	namespacesByUUID     map[string]*namespace.Namespace
	namespacesByAccessor map[string]*namespace.Namespace

	// creationDeletionMap tracks actively created or deleted namespaces,
	// enabling us to retry the deletion process if the namespace is tainted,
	// but isn't present in the said map.
	//
	// A namespace may not be marked tainted (in memory) if it is in this
	// list, so results need to be combined when namespaces are externally
	// exposed.
	creationDeletionMap           map[string]bool
	deletionDispatcher            *fairshare.JobManager
	creationDeletionJobContext    context.Context
	creationDeletionJobCancelFunc context.CancelFunc

	// logger is the server logger copied over from core
	logger *logger.GatedLogger
}

// NewNamespaceStore creates a new NamespaceStore that is backed
// using a given view. It used used to durable store and manage named namespace.
func NewNamespaceStore(ctx context.Context, core *Core, log *logger.GatedLogger) (*NamespaceStore, error) {
	ns := &NamespaceStore{
		core:                 core,
		storage:              core.barrier,
		logger:               log,
		namespacesByPath:     newNamespaceTree(nil),
		namespacesByUUID:     make(map[string]*namespace.Namespace),
		namespacesByAccessor: make(map[string]*namespace.Namespace),
		creationDeletionMap:  make(map[string]bool),
	}

	hclogLogger := logger.NewHCLogAdapter(log)

	ns.creationDeletionJobContext, ns.creationDeletionJobCancelFunc = context.WithCancel(core.activeContext)
	ns.deletionDispatcher = fairshare.NewJobManager(nsDispatcherName, nsMaxWorkers, hclogLogger, nil)
	ns.deletionDispatcher.Start()

	// Add namespaces from storage to our table. We can do this without
	// holding a lock as we've not returned ns to anyone yet.
	if err := ns.loadNamespacesLocked(ctx); err != nil {
		return nil, fmt.Errorf("error loading initial namespaces: %w", err)
	}

	return ns, nil
}

// NamespaceView uses given barrier and namespace to return back a view scoped to that namespace.
func NamespaceView(barrier logical.Storage, ns *namespace.Namespace) BarrierView {
	return NewBarrierView(barrier, NamespaceBarrierPrefix(ns))
}

// NamespaceBarrierPrefix uses given namespace to return back the common prefix
// used for all keys that belong to that namespace.
func NamespaceBarrierPrefix(ns *namespace.Namespace) string {
	if ns.ID == namespace.RootNamespaceID {
		return ""
	}

	return path.Join(namespaceBarrierPrefix, ns.UUID) + "/"
}

// cancelNamespaceDeletion cancels goroutine that runs namespace deletion.
func (c *Core) cancelNamespaceDeletion() {
	if c.namespaceStore == nil {
		return
	}

	c.namespaceStore.CancelNamespaceDeletion()
}

func (ns *NamespaceStore) CancelNamespaceDeletion() {
	// Cancel pending operations.
	ns.creationDeletionJobCancelFunc()

	// Stop jobs.
	ns.deletionDispatcher.Stop()
}

// checkInvalidation checks if the store has been marked as invalidated, and if
// so, reloads namespaces from disk.
// checkInvalidation returns true if it acquired a write-lock as part of the
// store reload and is handing the lock over to the caller.
// If checkInvalidation returns an error, it never keeps a write lock for the
// caller, so there is no need to check the bool before propagating an error.
func (ns *NamespaceStore) checkInvalidation(ctx context.Context) (bool, error) {
	if ctx.Err() != nil {
		return false, ctx.Err()
	}

	if !ns.invalidated.Load() {
		return false, nil
	}

	ns.lock.Lock()

	// Status might have changed
	if !ns.invalidated.Load() {
		return true, nil
	}

	if err := ns.loadNamespacesLocked(ctx); err != nil {
		// Caller likely just wants to propagate the error,
		// so don't keep the lock for them.
		ns.lock.Unlock()
		return false, fmt.Errorf("error handling invalidation: %w", err)
	}

	ns.invalidated.Store(false)
	return true, nil
}

// lockWithInvalidation is a helper calls [checkInvalidation] and acquires the
// desired type of lock, potentially carrying it over from [checkInvalidation].
// This is useful for most namespace store operations that initially revalidate
// the store and then need a lock to perform the main read and/or write
// operation.
//
// lockWithInvalidation in write = false mode may actually yield a write lock if
// one was acquired by [checkInvalidation].
// lockWithInvalidation in write = true mode will always yield a write lock.
//
// The returned unlock function will be the correct one to unlock the respective
// type of lock acquired under the hood. The caller must not call the unlock
// function if a non-nil error is returned.
func (ns *NamespaceStore) lockWithInvalidation(ctx context.Context, write bool) (func(), error) {
	locked, err := ns.checkInvalidation(ctx)
	if err != nil {
		return nil, err
	}

	switch {
	case locked:
		return ns.lock.Unlock, nil
	case write:
		ns.lock.Lock()
		return ns.lock.Unlock, nil
	default:
		ns.lock.RLock()
		return ns.lock.RUnlock, nil
	}
}

// loadNamespaces loads all stored namespaces from disk. It assumes the lock
// is held when required.
func (ns *NamespaceStore) loadNamespacesLocked(ctx context.Context) error {
	// Assume we roughly have as many namespaces as we have presently. During
	// invalidation this will pre-allocate enough space to reload everything
	// as we'll likely be essentially in sync already. However, at startup, this
	// will mostly just give us space for the root namespace.
	namespacesByPath := newNamespaceTree(namespace.RootNamespace)
	namespacesByUUID := make(map[string]*namespace.Namespace, len(ns.namespacesByUUID)+1)
	namespacesByAccessor := make(map[string]*namespace.Namespace, len(ns.namespacesByAccessor)+1)
	namespacesByUUID[namespace.RootNamespaceUUID] = namespace.RootNamespace
	namespacesByAccessor[namespace.RootNamespaceID] = namespace.RootNamespace

	loadingCallback := func(namespace *namespace.Namespace) error {
		if _, ok := namespacesByUUID[namespace.UUID]; ok {
			return fmt.Errorf("namespace with UUID %q is not unique in storage", namespace.UUID)
		}
		if err := namespacesByPath.Insert(namespace); err != nil {
			return err
		}
		namespacesByUUID[namespace.UUID] = namespace
		namespacesByAccessor[namespace.ID] = namespace
		return nil
	}

	if err := logical.WithTransaction(ctx, ns.storage, func(s logical.Storage) error {
		rootStoreView := NewBarrierView(s, namespaceStoreSubPath)
		return ns.loadNamespacesRecursive(ctx, s, rootStoreView, loadingCallback)
	}); err != nil {
		return err
	}

	err := namespacesByPath.validate()
	if err != nil {
		return err
	}

	ns.namespacesByPath = namespacesByPath
	ns.namespacesByUUID = namespacesByUUID
	ns.namespacesByAccessor = namespacesByAccessor

	return nil
}

// loadNamespacesRecursive reads all namespaces from a given namespace store view,
// recursing into the respective namespace store views of any discovered namespaces
// to load an entire namespace tree.
func (ns *NamespaceStore) loadNamespacesRecursive(
	ctx context.Context, barrier, view logical.Storage,
	callback func(*namespace.Namespace) error,
) error {
	return logical.HandleListPage(ctx, view, "", 100, func(page int, index int, entry string) (bool, error) {
		item, err := view.Get(ctx, entry)
		if err != nil {
			return false, fmt.Errorf("failed to fetch namespace %v (page %v / index %v): %w", entry, page, index, err)
		}

		if item == nil {
			return false, fmt.Errorf("%v has an empty namespace definition (page %v / index %v)", entry, page, index)
		}

		var namespace namespace.Namespace
		if err := item.DecodeJSON(&namespace); err != nil {
			return false, fmt.Errorf("failed to decode namespace %v (page %v / index %v): %w", entry, page, index, err)
		}

		namespace.Locked = namespace.UnlockKey != ""

		if err := callback(&namespace); err != nil {
			return false, err
		}

		childView := NamespaceView(barrier, &namespace).SubView(namespaceStoreSubPath)
		if err := ns.loadNamespacesRecursive(ctx, barrier, childView, callback); err != nil {
			return false, err
		}

		return true, nil
	}, nil)
}

// setupNamespaceStore is used to initialize the namespace store
// when the vault is being unsealed.
func (c *Core) setupNamespaceStore(ctx context.Context) error {
	// Create the Namespace store
	var err error
	nsLogger := c.logger.WithSystem("namespace")
	c.namespaceStore, err = NewNamespaceStore(ctx, c, nsLogger)
	return err
}

// teardownNamespaceStore is used to reverse setupNamespaceStore
// when the vault is being sealed.
func (c *Core) teardownNamespaceStore() error {
	c.namespaceStore = nil
	return nil
}

func (ns *NamespaceStore) invalidate(ctx context.Context, path string) {
	// Defer invalidation to the next load for performance.
	//
	// TODO(ascheel): handle individual entry invalidation correctly. We'll
	// need to handle child namespace invalidation as well. sync.Map could be
	// used instead in the future alongside the actual boolean.
	ns.invalidated.Store(true)
}

// SetNamespace is used to create or update a given namespace
func (ns *NamespaceStore) SetNamespace(ctx context.Context, namespace *namespace.Namespace) error {
	//defer metrics.MeasureSince([]string{"namespace", "set_namespace"}, time.Now())

	if _, err := ns.lockWithInvalidation(ctx, true); err != nil {
		return err
	}

	if err := ns.setNamespaceLocked(ctx, namespace); err != nil {
		ns.logger.Error("set namespace failed",
			logger.Err(err),
		)
		return err
	}

	return nil
}

// setNamespaceLocked must be called while holding a write lock over the
// NamespaceStore. This function unlocks the lock once finished.
func (ns *NamespaceStore) setNamespaceLocked(ctx context.Context, nsEntry *namespace.Namespace) error {
	// If we are creating a net-new namespace, we have to unlock before
	// creating required mounts as the mount type will call
	// GetNamespaceByAccessor. In that case, we will manually call
	// ns.lock.Unlock() but in all other cases (including early exit)
	// we _also_ need to unlock. So using a defer is preferable. Calling
	// unlock twice may fail (either incorrectly releasing the write lock
	// if someone else grabbed it or panicing if it was double unlocked)
	// so we need a boolean here to determine whether or not our deferred
	// unlock should actually be run.
	var unlocked bool
	defer func() {
		if !unlocked {
			ns.lock.Unlock()
			unlocked = true
		}
	}()

	// Copy the entry before validating and potentially mutating it.
	entry := nsEntry.Clone(true /* preserve unlock */)
	if err := entry.Validate(); err != nil {
		return wlogical.ErrBadRequest(err.Error())
	}

	// Validate that we have a parent namespace.
	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return fmt.Errorf("error loading parent namespace from context: %w", err)
	}

	var exists bool
	if entry.UUID == "" {
		id, err := ns.assignIdentifier(entry.Path)
		if err != nil {
			return err
		}

		entry.ID = id
		entry.UUID, err = uuid.GenerateUUID()
		if err != nil {
			return err
		}
	} else {
		var existing *namespace.Namespace
		existing, exists = ns.namespacesByUUID[entry.UUID]
		if !exists {
			return errors.New("trying to update a non-existent namespace")
		}

		if existing.ID != entry.ID {
			return errors.New("accessor ID does not match")
		}

		if existing.Path != entry.Path {
			return errors.New("unable to remount namespace at new path")
		}
	}

	if !exists {
		// Before attempting to create it, ensure we don't have a mount table
		// entry that conflicts with this new namespace. We assume we only
		// need to look at our parent's namespace's mount table for the last
		// path component of this new namespace; while mount paths can have
		// any number of components, our namespace only has one and is relative
		// to some parent path.
		path := entry.Path
		if parent.ID != namespace.RootNamespaceID {
			if !entry.HasParent(parent) {
				return errors.New("namespace path lacks parent as a prefix")
			}

			path = namespace.Canonicalize(parent.TrimmedPath(entry.Path))
		}

		ns.core.router.mu.RLock()
		conflict := ns.core.router.matchingPrefixInternal(ctx, path)
		ns.core.router.mu.RUnlock()
		if conflict != "" {
			return fmt.Errorf("new namespace conflicts with existing mount: %v", conflict)
		}
	}

	// Order of storage operations is important:
	//
	// 1. Write the durable storage entry as tainted. This helps signal
	//    standby nodes to ignore the update.
	// 2. Update the in-mem namespace store, marking the namespace as
	//    undergoing an asynchronous modification.
	// 3. Write the in-namespace entries.
	// 4. Finalize the namespace by re-writing the durable storage entry
	//    as not tainted.
	//
	// Cleanup thus needs to handle partial deletion and remove any storage
	// using the background context, not our primary context. We delete the
	// storage entry while holding the lock, but Tainted status ensures
	// nobody else attempts to mutate until we're done.
	failed := true

	cleanupFailed := func() {
		if !failed {
			return
		}

		if exists {
			return
		}

		// Do mount cleanup without holding the namespace lock to prevent
		// re-entrant locking.
		if !unlocked {
			ns.lock.Unlock()
			unlocked = true
		}

		// Queue partially created namespace deletion for the background
		// workers rather than doing it synchronously.
		ns.deletionDispatcher.AddJob(ns.newNamespaceCreationFailureJob(parent, entry), parent.UUID)
	}

	defer cleanupFailed()

	if !exists {
		// This initial write sets up the namespace as tainted, preventing
		// a lot of subsystems from using it if it is reloaded from storage.
		entry.Tainted = true
		if err := ns.writeNamespace(ctx, ns.storage, entry); err != nil {
			return fmt.Errorf("failed to persist initial tainted namespace: %w", err)
		}

		// But we don't mark our in-memory version as being tainted so that
		// initial mounts can succeed.
		entry.Tainted = false

		ns.creationDeletionMap[entry.UUID] = true
	}

	if err := ns.namespacesByPath.Insert(entry); err != nil {
		return fmt.Errorf("failed to insert namespace into tree: %w", err)
	}
	ns.namespacesByUUID[entry.UUID] = entry
	ns.namespacesByAccessor[entry.ID] = entry

	if !exists {
		// unlock before initializeNamespace since that will re-acquire the lock
		ns.lock.Unlock()
		unlocked = true

		// Create sys/, token/ mounts and policies for the new namespace.
		if err := ns.initializeNamespace(ctx, ns.storage, entry); err != nil {
			return err
		}

		// Reacquire the lock to undo tainting in storage. We need the lock
		// here to ensure we don't race to update storage across multiple
		// callers.
		ns.lock.Lock()
		unlocked = false
	}

	// Finally, write the non-tainted namespace or perform our only write if
	// we are modifying an existing entry.
	if err := ns.writeNamespace(ctx, ns.storage, entry); err != nil {
		return fmt.Errorf("failed to persist namespace: %w", err)
	}

	// Since the write succeeded, copy back any potentially changed values.
	nsEntry.UUID = entry.UUID
	nsEntry.ID = entry.ID
	nsEntry.Path = entry.Path
	delete(ns.creationDeletionMap, entry.UUID)
	failed = false

	// Lastly, push the change to all mounts.
	return ns.pushToMounts(ctx, entry)
}

func (ns *NamespaceStore) writeNamespace(ctx context.Context, storage logical.Storage, entry *namespace.Namespace) error {
	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return err
	}

	view := NamespaceView(storage, parent).SubView(namespaceStoreSubPath)
	item, err := logical.StorageEntryJSON(entry.UUID, &entry)
	if err != nil {
		return fmt.Errorf("error marshalling storage entry: %w", err)
	}

	if err := view.Put(ctx, item); err != nil {
		return err
	}

	return nil
}

// assignIdentifier assumes the lock is held.
func (ns *NamespaceStore) assignIdentifier(path string) (string, error) {
	if ns := ns.namespacesByPath.Get(path); ns != nil {
		return "", errors.New("unable to update when a namespace with this path already exists")
	}

	for {
		id, err := base62.Random(namespaceIdLength)
		if err != nil {
			return "", fmt.Errorf("unable to generate namespace identifier: %w", err)
		}

		// accessor id already exists
		if _, ok := ns.namespacesByAccessor[id]; ok {
			continue
		}

		return id, nil
	}
}

// initializeNamespace initializes default policies and  sys/ and token/ mounts for a new namespace
func (ns *NamespaceStore) initializeNamespace(ctx context.Context, storage logical.Storage, entry *namespace.Namespace) error {
	// ctx may have a namespace of the parent of our newly created namespace,
	// so create a new context with the newly created child namespace.
	nsCtx := namespace.ContextWithNamespace(ctx, entry.Clone(false))

	if err := ns.initializeNamespacePolicies(nsCtx); err != nil {
		return err
	}

	if err := ns.createMounts(nsCtx, storage); err != nil {
		return err
	}

	return nil
}

// initializeNamespacePolicies loads the default policies for the namespace store.
func (ns *NamespaceStore) initializeNamespacePolicies(ctx context.Context) error {
	if err := ns.core.policyStore.loadDefaultPolicies(ctx); err != nil {
		return fmt.Errorf("error creating default policies: %w", err)
	}
	return nil
}

// createMounts handles creation of sys/ mounts for this new
// namespace.
//
// This is a two-step process:
//
// 1. Create in-memory versions of the mount.
// 2. Persist into storage using the passed transaction.
func (ns *NamespaceStore) createMounts(ctx context.Context, storage logical.Storage) error {
	// Do not persist mounts yet.
	mounts, err := ns.core.requiredMountTable(ctx)
	if err != nil {
		return fmt.Errorf("for new namespace: %w", err)
	}

	// We hold the lock over updating
	// both the in-memory mount table and the transaction.
	ns.core.mountsLock.Lock()
	defer ns.core.mountsLock.Unlock()

	for _, mount := range mounts.Entries {
		if err := ns.core.mountInternalLocked(ctx, mount, MountTableNoUpdateStorage); err != nil {
			return err
		}
	}

	// Persist the mounts using the above storage entry.
	return logical.WithTransaction(ctx, storage, func(txn logical.Storage) error {
		for _, mount := range mounts.Entries {
			if err := ns.core.persistMounts(ctx, storage, ns.core.mounts, mount.UUID); err != nil {
				return fmt.Errorf("failed to persist secret mount (path=%v): %w", mount.Path, err)
			}
		}

		return nil
	})
}

// undoCreateMounts handles commit failure for in-memory resources. Note
// that we do not modify storage here as our context has (usually) been
// canceled and so we'd need activeContext or similar.
func (ns *NamespaceStore) undoCreateMounts(nsCtx context.Context, namespaceToDelete *namespace.Namespace) bool {
	success := true

	// clear mounts
	ns.core.mountsLock.Lock()
	mountEntries, err := ns.core.mounts.findAllNamespaceMounts(nsCtx)
	ns.core.mountsLock.Unlock()
	if err != nil {
		ns.logger.Error("failed to retrieve namespace mounts",
			logger.String("namespace", namespaceToDelete.Path),
			logger.Err(err))
		success = false
	} else {
		for _, me := range mountEntries {
			err := ns.core.unmountInternal(nsCtx, me.Path, false)
			if err != nil {
				if errors.Is(err, errNoMatchingMount) {
					continue
				}
				ns.logger.Error(fmt.Sprintf("failed to unmount %q", me.Path),
					logger.String("namespace", namespaceToDelete.Path),
					logger.Err(err))
				success = false
				continue
			}
		}
	}
	return success
}

func (ns *NamespaceStore) pushToMounts(ctx context.Context, entry *namespace.Namespace) error {
	ns.core.mountsLock.Lock()
	defer ns.core.mountsLock.Unlock()

	for _, mount := range ns.core.mounts.Entries {
		if mount.NamespaceID != entry.ID {
			continue
		}

		mount.namespace = entry
	}

	return nil
}

// GetNamespace is used to fetch the namespace with the given uuid.
func (ns *NamespaceStore) GetNamespace(ctx context.Context, uuid string) (*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "get_namespace"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, false)
	if err != nil {
		return nil, err
	}
	defer unlock()

	item, ok := ns.namespacesByUUID[uuid]
	if !ok {
		return nil, nil
	}

	entry := item.Clone(false)
	entry.Tainted = entry.Tainted || ns.creationDeletionMap[entry.UUID]

	return entry, nil
}

// GetNamespaceByAccessor is used to fetch the namespace with the given accessor.
func (ns *NamespaceStore) GetNamespaceByAccessor(ctx context.Context, id string) (*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "get_namespace_by_accessor"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, false)
	if err != nil {
		return nil, err
	}
	defer unlock()

	item, ok := ns.namespacesByAccessor[id]
	if !ok {
		return nil, nil
	}

	entry := item.Clone(false)
	entry.Tainted = entry.Tainted || ns.creationDeletionMap[entry.UUID]

	return entry, nil
}

func (ns *NamespaceStore) GetNamespaceByLongestPrefix(ctx context.Context, path string) (*namespace.Namespace, string) {
	ctxNs, err := namespace.FromContext(ctx)
	if err != nil {
		ctxNs = namespace.RootNamespace
	}

	combinedPath := ctxNs.Path + path
	ns.lock.RLock()
	prefix, resolved, _ := ns.namespacesByPath.LongestPrefix(combinedPath)
	entry := resolved.Clone(false)
	entry.Tainted = entry.Tainted || ns.creationDeletionMap[entry.UUID]
	ns.lock.RUnlock()
	return entry, strings.TrimPrefix(combinedPath, prefix)
}

// GetNamespaceByPath is used to fetch the namespace with the given full path.
func (ns *NamespaceStore) GetNamespaceByPath(ctx context.Context, path string) (*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "get_namespace_by_path"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, false)
	if err != nil {
		return nil, err
	}
	defer unlock()

	return ns.getNamespaceByPathLocked(ctx, path, false)
}

func (ns *NamespaceStore) getNamespaceByPathLocked(
	ctx context.Context, path string, withUnlockKey bool,
) (*namespace.Namespace, error) {
	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return nil, err
	}
	path = namespace.Canonicalize(parent.Path + path)
	item := ns.namespacesByPath.Get(path)
	if item == nil {
		return nil, nil
	}

	entry := item.Clone(withUnlockKey)
	entry.Tainted = entry.Tainted || ns.creationDeletionMap[entry.UUID]

	return entry, nil
}

// ModifyNamespace is used to perform modifications to a namespace while
// holding a write lock to prevent other changes to namespaces from occurring
// at the same time.
func (ns *NamespaceStore) ModifyNamespaceByPath(ctx context.Context, path string, callback func(context.Context, *namespace.Namespace) (*namespace.Namespace, error)) (*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "modify_namespace"}, time.Now())

	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return nil, err
	}

	path = namespace.Canonicalize(parent.Path + path)
	if path == "" {
		return nil, logical.CodedError(http.StatusBadRequest, "refusing to modify root namespace")
	}

	unlock, err := ns.lockWithInvalidation(ctx, true)
	if err != nil {
		return nil, err
	}

	entry := ns.namespacesByPath.Get(path)
	if entry != nil {
		if entry.Tainted {
			unlock()
			return nil, errors.New("namespace with that name exists and is currently tainted")
		}
		if value := ns.creationDeletionMap[entry.UUID]; value {
			unlock()
			return nil, errors.New("namespace with that name exists and is currently being created or deleted")
		}

		entry = entry.Clone(true /* preserve unlock key so we can copy it */)
	} else {
		entry = &namespace.Namespace{
			Path:           path,
			CustomMetadata: make(map[string]string),
		}
	}

	unlockKey := entry.UnlockKey

	if callback != nil {
		entry.UnlockKey = ""
		entry, err = callback(ctx, entry)
		if err != nil {
			unlock()
			return nil, err
		}

		// ModifyNamespaceByPath can never modify lock status.
		entry.UnlockKey = unlockKey
	}

	// setNamespaceLocked will unlock ns.lock
	if err := ns.setNamespaceLocked(ctx, entry); err != nil {
		ns.logger.Error("set namespace failed", logger.Err(err))
		return nil, err
	}

	return entry.Clone(false), nil
}

// ListAllNamespaces lists all available namespaces, optionally including the
// root namespace.
func (ns *NamespaceStore) ListAllNamespaces(ctx context.Context, includeRoot bool) ([]*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "list_all_namespaces"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, false)
	if err != nil {
		return nil, err
	}
	defer unlock()

	namespaces := make([]*namespace.Namespace, 0, len(ns.namespacesByUUID))
	for _, entry := range ns.namespacesByUUID {
		if !includeRoot && entry.ID == namespace.RootNamespaceID {
			continue
		}
		namespaces = append(namespaces, entry.Clone(false))
	}

	return namespaces, nil
}

// ListNamespaces is used to list namespaces below a parent namespace.
// Optionally it can include the parent namespace itself and/or include all
// descendants of the child namespaces.
func (ns *NamespaceStore) ListNamespaces(ctx context.Context, includeParent bool, recursive bool) ([]*namespace.Namespace, error) {
	//defer metrics.MeasureSince([]string{"namespace", "list_namespace_entries"}, time.Now())

	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return nil, err
	}

	unlock, err := ns.lockWithInvalidation(ctx, false)
	if err != nil {
		return nil, err
	}
	defer unlock()

	return ns.namespacesByPath.List(parent.Path, includeParent, recursive, ns.creationDeletionMap)
}

// taintNamespace is used to taint the namespace designated to be deleted
func (ns *NamespaceStore) taintNamespace(ctx context.Context, namespaceToTaint *namespace.Namespace) error {
	// to be extra safe
	if namespaceToTaint.ID == namespace.RootNamespaceID {
		return errors.New("cannot taint root namespace")
	}

	ns.namespacesByUUID[namespaceToTaint.UUID].Tainted = true
	ns.namespacesByAccessor[namespaceToTaint.ID].Tainted = true
	namespaceToTaint.Tainted = true
	err := ns.namespacesByPath.Insert(namespaceToTaint)
	if err != nil {
		return fmt.Errorf("failed to modify namespace tree: %w", err)
	}

	nsCopy := namespaceToTaint.Clone(true /* preserve unlock */)
	if err := ns.writeNamespace(ctx, ns.storage, nsCopy); err != nil {
		return fmt.Errorf("failed to persist namespace taint: %w", err)
	}

	// Push the update to all mounts.
	return ns.pushToMounts(ctx, namespaceToTaint.Clone(false))
}

// DeleteNamespace is used to delete the named namespace
func (ns *NamespaceStore) DeleteNamespace(ctx context.Context, path string) (string, error) {
	//defer metrics.MeasureSince([]string{"namespace", "delete_namespace"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, true)
	if err != nil {
		return "", err
	}
	defer unlock()

	namespaceToDelete, err := ns.getNamespaceByPathLocked(ctx, path, false)
	if err != nil {
		return "", err
	}
	if namespaceToDelete == nil {
		return "", nil
	}

	isNamespaceDeleting := ns.creationDeletionMap[namespaceToDelete.UUID]
	if namespaceToDelete.Tainted && isNamespaceDeleting {
		return "in-progress", nil
	}

	if namespaceToDelete.ID == namespace.RootNamespaceID {
		return "", errors.New("unable to delete root namespace")
	}

	// checking whether namespace has child namespaces
	childNS, err := ns.namespacesByPath.List(namespaceToDelete.Path, false, false, map[string]bool{})
	if err != nil {
		return "", err
	}

	if len(childNS) > 0 {
		return "", fmt.Errorf("cannot delete namespace (%q) containing child namespaces", namespaceToDelete.Path)
	}

	if !namespaceToDelete.Tainted {
		// taint the namespace
		err = ns.taintNamespace(ctx, namespaceToDelete)
	}

	parent, err := namespace.FromContext(ctx)
	if err != nil {
		return "", fmt.Errorf("error loading parent namespace from context: %w", err)
	}

	ns.creationDeletionMap[namespaceToDelete.UUID] = true

	ns.deletionDispatcher.AddJob(ns.newNamespaceDeletionJob(parent, namespaceToDelete), parent.UUID)

	return "in-progress", nil
}

func (ns *NamespaceStore) clearNamespaceResources(nsCtx context.Context, namespaceToDelete *namespace.Namespace) error {
	// clear rotation entries (stop active rotation jobs before touching other resources)
	if ns.core.rotationManager != nil {
		if err := ns.core.rotationManager.UnregisterByNamespace(namespaceToDelete.UUID); err != nil {
			return fmt.Errorf("failed to clear rotation entries: %w", err)
		}
	}

	// clear expiration entries (stop TTL timers to prevent revocation during cleanup)
	if ns.core.expirationManager != nil {
		ns.core.expirationManager.UnregisterByNamespace(namespaceToDelete.ID)
	}

	// clear tokens (remove auth tokens from cache and storage)
	if ns.core.tokenStore != nil {
		if err := ns.core.tokenStore.RevokeByNamespace(namespaceToDelete.ID); err != nil {
			return fmt.Errorf("failed to clear tokens: %w", err)
		}
	}

	// clear credential specs and sources
	if ns.core.credConfigStore != nil {
		if err := ns.core.credConfigStore.ClearNamespace(nsCtx); err != nil {
			return fmt.Errorf("failed to clear credential configs: %w", err)
		}
	}

	// clear CBP policies
	policiesToClear, err := ns.core.policyStore.ListPolicies(nsCtx, PolicyTypeCBP, false)
	if err != nil {
		return fmt.Errorf("failed to retrieve namespace policies: %w", err)
	}
	for _, policy := range policiesToClear {
		err := ns.core.policyStore.deletePolicyForce(nsCtx, policy, PolicyTypeCBP)
		if err != nil {
			return fmt.Errorf("failed to delete policy: %w", err)
		}
	}

	// clear mounts
	ns.core.mountsLock.Lock()
	mountEntries, err := ns.core.mounts.findAllNamespaceMounts(nsCtx)
	ns.core.mountsLock.Unlock()
	if err != nil {
		return fmt.Errorf("failed to retrieve namespace mounts: %w", err)
	}

	for _, me := range mountEntries {
		err := ns.core.unmountInternal(nsCtx, me.Path, true)
		if err != nil {
			if wlogical.IsNotFound(err) {
				continue
			}
			return fmt.Errorf("failed to disable mount (%v) in namespace (%v): %w", me.Path, namespaceToDelete.Path, err)
		}
	}

	return nil
}

// ResolveNamespaceFromRequest resolves a namespace from the 'X-Warden-Namespace'
// header combined with the request path, returning the namespace and the
// "trimmed" request path devoid of any namespace components.
func (ns *NamespaceStore) ResolveNamespaceFromRequest(nsHeader, reqPath string) (*namespace.Namespace, string) {
	nsHeader = namespace.Canonicalize(nsHeader)
	// Naively stack header ahead of request path.
	reqPath = nsHeader + reqPath

	// Find namespace that matches the longest prefix of reqPath.
	ns.lock.RLock()
	_, resolvedNs, trimmedPath := ns.namespacesByPath.LongestPrefix(reqPath)
	// Clone before setting Tainted to avoid mutating the shared namespace
	// pointer — the trie stores a single *Namespace per path, so concurrent
	// requests would race on the Tainted field write/read without a copy.
	// This matches the pattern used by GetNamespaceByAccessor/GetNamespaceByUUID.
	entry := resolvedNs.Clone(false)
	entry.Tainted = entry.Tainted || ns.creationDeletionMap[entry.UUID]
	ns.lock.RUnlock()

	// Ensure that entire header was matched, so unmatched paths don't leak
	// into the request path.
	if !strings.HasPrefix(entry.Path, nsHeader) {
		return nil, ""
	}

	return entry, trimmedPath
}

// GetLockingNamespace walks the namespace tree structure looking for
// a locked namespace, starting from one of the root children,
// ending at the namespace provided as a argument to the function.
func (ns *NamespaceStore) GetLockingNamespace(n *namespace.Namespace) *namespace.Namespace {
	ns.lock.RLock()
	defer ns.lock.RUnlock()

	return ns.getLockingNamespace(n)
}

func (ns *NamespaceStore) getLockingNamespace(n *namespace.Namespace) *namespace.Namespace {
	var lockedNS *namespace.Namespace
	ns.namespacesByPath.WalkPath(n.Path, func(curNS *namespace.Namespace) bool {
		if curNS.Locked {
			lockedNS = curNS
			return true
		}
		return false
	})
	if lockedNS != nil {
		return lockedNS.Clone(false)
	}

	return nil
}

// UnlockNamespace attempts to unlock the namespace with provided namespace path.
func (ns *NamespaceStore) UnlockNamespace(ctx context.Context, unlockKey, path string) error {
	//defer metrics.MeasureSince([]string{"namespace", "unlock_namespace"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, true)
	if err != nil {
		return err
	}

	defer func() {
		if unlock != nil {
			unlock()
		}
	}()

	namespaceToUnlock, err := ns.getNamespaceByPathLocked(ctx, path, true)
	if err != nil {
		return err
	}

	if namespaceToUnlock == nil {
		return errors.New("requested namespace does not exist")
	}

	if namespaceToUnlock.ID == namespace.RootNamespaceID {
		return errors.New("root namespace cannot be locked/unlocked")
	}

	if !namespaceToUnlock.Locked {
		return fmt.Errorf("namespace %q is not locked", namespaceToUnlock.Path)
	}

	lockingNamespace := ns.getLockingNamespace(namespaceToUnlock)
	if lockingNamespace.ID != namespaceToUnlock.ID {
		return fmt.Errorf("cannot unlock %q with namespace %q being locked", namespaceToUnlock.Path, lockingNamespace.Path)
	}

	// verify lock or skip when provided unlock key is empty
	// (meaning namespace is unlocked using root capability)
	if unlockKey != "" &&
		subtle.ConstantTimeCompare([]byte(namespaceToUnlock.UnlockKey), []byte(unlockKey)) != 1 {
		return errors.New("incorrect unlock key")
	}

	namespaceToUnlock.Locked = false
	namespaceToUnlock.UnlockKey = ""

	parentPath, _ := namespaceToUnlock.ParentPath()
	parentNs := ns.namespacesByPath.Get(parentPath)
	if parentNs == nil {
		return fmt.Errorf("namespace %q has no parent", namespaceToUnlock.Path)
	}
	parentCtx := namespace.ContextWithNamespace(ctx, parentNs)

	// setNamespaceLocked now handles unlocking.
	unlock = nil
	return ns.setNamespaceLocked(parentCtx, namespaceToUnlock)
}

// LockNamespace attempts to lock the namespace with provided path.
func (ns *NamespaceStore) LockNamespace(ctx context.Context, path string) (string, error) {
	defer metrics.MeasureSince([]string{"namespace", "lock_namespace"}, time.Now())

	unlock, err := ns.lockWithInvalidation(ctx, true)
	if err != nil {
		return "", err
	}

	defer func() {
		if unlock != nil {
			unlock()
		}
	}()

	namespaceToLock, err := ns.getNamespaceByPathLocked(ctx, path, false)
	if err != nil {
		return "", err
	}

	if namespaceToLock == nil {
		return "", errors.New("requested namespace does not exist")
	}

	if namespaceToLock.ID == namespace.RootNamespaceID {
		return "", errors.New("root namespace cannot be locked/unlocked")
	}

	lockedNamespace := ns.getLockingNamespace(namespaceToLock)
	if lockedNamespace != nil && lockedNamespace.ID == namespaceToLock.ID {
		return "", fmt.Errorf("cannot lock namespace %q: is already locked", namespaceToLock.Path)
	} else if lockedNamespace != nil {
		return "", fmt.Errorf("cannot lock namespace %q: ancestor namespace %q is already locked", namespaceToLock.Path, lockedNamespace.Path)
	}

	// create lock
	lockKey, err := base62.Random(24)
	if err != nil {
		return "", fmt.Errorf("unable to generate namespace lock key: %w", err)
	}

	namespaceToLock.Locked = true
	namespaceToLock.UnlockKey = lockKey

	parentPath, _ := namespaceToLock.ParentPath()
	parentNs := ns.namespacesByPath.Get(parentPath)
	if parentNs == nil {
		return "", fmt.Errorf("namespace %q has no parent", namespaceToLock.Path)
	}
	parentCtx := namespace.ContextWithNamespace(ctx, parentNs)

	// setNamespaceLocked now handles unlocking.
	unlock = nil
	if err := ns.setNamespaceLocked(parentCtx, namespaceToLock); err != nil {
		return "", fmt.Errorf("unable to save locked namespace %q", namespaceToLock.Path)
	}

	return lockKey, nil
}

// namespaceDeletionJob is used with NamespaceStore.deletionDispatcher to
// gradually remove items from the namespace store.
type namespaceDeletionJob struct {
	store  *NamespaceStore
	parent *namespace.Namespace
	target *namespace.Namespace
}

func (ns *NamespaceStore) newNamespaceDeletionJob(parent *namespace.Namespace, target *namespace.Namespace) fairshare.Job {
	return &namespaceDeletionJob{
		store:  ns,
		parent: parent,
		target: target,
	}
}

func (j *namespaceDeletionJob) Execute() error {
	// Clearing needs to happen without holding the namespace lock.
	ctx := namespace.ContextWithNamespace(j.store.creationDeletionJobContext, j.target)
	err := j.store.clearNamespaceResources(ctx, j.target)

	j.store.lock.Lock()
	defer j.store.lock.Unlock()

	// Make sure this happens _before_ the unlock.
	defer delete(j.store.creationDeletionMap, j.target.UUID)

	// If we failed to clear any resources, stop here and don't delete the namespace.
	if err != nil {
		return fmt.Errorf("failed clearing namespace resources: %w", err)
	}

	if err := j.store.namespacesByPath.Delete(j.target.Path); err != nil {
		return fmt.Errorf("failed to delete namespace entry in namespace tree: %w", err)
	}

	delete(j.store.namespacesByUUID, j.target.UUID)
	delete(j.store.namespacesByAccessor, j.target.ID)

	view := NamespaceView(j.store.storage, j.parent).SubView(namespaceStoreSubPath)
	if err := view.Delete(ctx, j.target.UUID); err != nil {
		return fmt.Errorf("failed to delete namespace storage entry: %w", err)
	}

	return nil
}

func (j *namespaceDeletionJob) OnFailure(err error) {
	j.store.logger.Error("failed to handle namespace deletion; job may be retried",
		logger.String("namespace", j.target.Path),
		logger.String("ns_uuid", j.target.UUID),
		logger.Err(err))
}

// namespaceCreationFailureJob is used with NamespaceStore.setNamespaceLocked
// to gracefully handle long-lived deletion.
type namespaceCreationFailureJob struct {
	store  *NamespaceStore
	parent *namespace.Namespace
	target *namespace.Namespace
}

func (ns *NamespaceStore) newNamespaceCreationFailureJob(parent *namespace.Namespace, target *namespace.Namespace) fairshare.Job {
	return &namespaceCreationFailureJob{
		store:  ns,
		parent: parent,
		target: target,
	}
}

func (j *namespaceCreationFailureJob) Execute() error {
	// Handle in-memory mount table entries that we should also clean
	// up.
	nsCtx := namespace.ContextWithNamespace(j.store.creationDeletionJobContext, j.target)
	cleanupSuccess := j.store.undoCreateMounts(nsCtx, j.target)

	var retErr error

	// Clear the view corresponding with the namespace for
	// completeness.
	view := NamespaceView(j.store.core.barrier, j.target)
	if err := logical.ClearViewWithLogging(j.store.creationDeletionJobContext, view, logger.NewHCLogAdapter(j.store.logger)); err != nil {
		retErr = fmt.Errorf("failed to remove remaining namespace storage: %w", err)
		cleanupSuccess = false
	}

	// Now grab the lock again to handle updating the namespace store.
	j.store.lock.Lock()
	defer j.store.lock.Unlock()

	// Cleanup finished either way.
	delete(j.store.creationDeletionMap, j.target.UUID)

	// Only perform cleanup from in-memory and stored contexts if we had
	// success removing entries. Otherwise, we'll treat this as a failed
	// deletion.
	j.target.Tainted = true

	if cleanupSuccess {
		// When cleanup succeeds and the namespace did not exist, we should back
		// out the entry from our in-memory and stored versions.
		if err := j.store.namespacesByPath.Delete(j.target.Path); err != nil {
			err = fmt.Errorf("failed to remove namespace from path manager: %w", err)
			retErr = multierror.Append(retErr, err)
		}

		delete(j.store.namespacesByUUID, j.target.UUID)
		delete(j.store.namespacesByAccessor, j.target.ID)

		nsView := NamespaceView(j.store.storage, j.parent).SubView(namespaceStoreSubPath)
		if err := nsView.Delete(j.store.creationDeletionJobContext, j.target.UUID); err != nil {
			err = fmt.Errorf("failed to remove created namespace storage entry on failure: %w", err)
			retErr = multierror.Append(retErr, err)
		}
	}

	return retErr
}

func (j *namespaceCreationFailureJob) OnFailure(err error) {
	j.store.logger.Error("failed to handle namespace deletion following failed creation; job may be retried via deletion of tainted namespace",
		logger.String("namespace", j.target.Path),
		logger.String("ns_uuid", j.target.UUID),
		logger.Err(err))
}
