upstream-seaweedfs/weed/filer/filerstore_wrapper.go

package filer

import (
	"context"
	"fmt"
	"io"
	"math"
	"strings"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/viant/ptrie"

	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/stats"
	"github.com/seaweedfs/seaweedfs/weed/util"
)

var (
	_ = VirtualFilerStore(&FilerStoreWrapper{})
	_ = Debuggable(&FilerStoreWrapper{})
)

type VirtualFilerStore interface {
	FilerStore
	DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error
	DeleteOneEntry(ctx context.Context, entry *Entry) error
	InsertEntryKnownAbsent(ctx context.Context, entry *Entry) error
	AddPathSpecificStore(path string, storeId string, store FilerStore)
	OnBucketCreation(bucket string)
	OnBucketDeletion(bucket string)
	CanDropWholeBucket() bool
	SameActualStore(a, b util.FullPath) bool
}

type FilerStoreWrapper struct {
	defaultStore         FilerStore
	pathToStore          ptrie.Trie[string]
	storeIdToStore       map[string]FilerStore
	hasPathSpecificStore bool // fast check to skip MatchPrefix when no path-specific stores
}

func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
	if innerStore, ok := store.(*FilerStoreWrapper); ok {
		return innerStore
	}
	return &FilerStoreWrapper{
		defaultStore:   store,
		pathToStore:    ptrie.New[string](),
		storeIdToStore: make(map[string]FilerStore),
	}
}

func (fsw *FilerStoreWrapper) CanDropWholeBucket() bool {
	if ba, ok := fsw.defaultStore.(BucketAware); ok {
		return ba.CanDropWholeBucket()
	}
	return false
}

func (fsw *FilerStoreWrapper) OnBucketCreation(bucket string) {
	for _, store := range fsw.storeIdToStore {
		if ba, ok := store.(BucketAware); ok {
			ba.OnBucketCreation(bucket)
		}
	}
	if ba, ok := fsw.defaultStore.(BucketAware); ok {
		ba.OnBucketCreation(bucket)
	}
}
func (fsw *FilerStoreWrapper) OnBucketDeletion(bucket string) {
	for _, store := range fsw.storeIdToStore {
		if ba, ok := store.(BucketAware); ok {
			ba.OnBucketDeletion(bucket)
		}
	}
	if ba, ok := fsw.defaultStore.(BucketAware); ok {
		ba.OnBucketDeletion(bucket)
	}
}

func (fsw *FilerStoreWrapper) AddPathSpecificStore(path string, storeId string, store FilerStore) {
	fsw.storeIdToStore[storeId] = NewFilerStorePathTranslator(path, store)
	err := fsw.pathToStore.Put([]byte(path), storeId)
	if err != nil {
		glog.Fatalf("put path specific store: %v", err)
	}
	fsw.hasPathSpecificStore = true
}

func (fsw *FilerStoreWrapper) getActualStore(path util.FullPath) (store FilerStore) {
	store = fsw.defaultStore
	// Fast path: skip MatchPrefix if no path-specific stores are configured (common case)
	if !fsw.hasPathSpecificStore {
		return
	}
	if path == "/" || path == "//" {
		return
	}
	var storeId string
	fsw.pathToStore.MatchPrefix([]byte(path), func(key []byte, value string) bool {
		storeId = value
		return false
	})
	if storeId != "" {
		store = fsw.storeIdToStore[storeId]
	}
	return
}

// SameActualStore reports whether two paths resolve to the same underlying
// store. When path-specific stores are configured, different subtrees may
// be served by different backends.
func (fsw *FilerStoreWrapper) SameActualStore(a, b util.FullPath) bool {
	return fsw.getActualStore(a) == fsw.getActualStore(b)
}

func (fsw *FilerStoreWrapper) getDefaultStore() (store FilerStore) {
	return fsw.defaultStore
}

func (fsw *FilerStoreWrapper) GetName() string {
	return fsw.getDefaultStore().GetName()
}

func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
	return fsw.getDefaultStore().Initialize(configuration, prefix)
}

func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(entry.FullPath)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
	}()

	filer_pb.BeforeEntrySerialization(entry.GetChunks())
	normalizeEntryMimeForStore(entry)

	if len(entry.HardLinkId) > 0 {
		glog.V(4).InfofCtx(ctx, "InsertEntry %s has HardLinkId %x counter=%d",
			entry.FullPath, entry.HardLinkId, entry.HardLinkCounter)
	}

	if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
		return err
	}

	return actualStore.InsertEntry(ctx, entry)
}

// InsertEntryKnownAbsent skips the pre-insert FindEntry path when the caller has
// already established that the target path does not exist.
func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry *Entry) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(entry.FullPath)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
	}()

	filer_pb.BeforeEntrySerialization(entry.GetChunks())
	normalizeEntryMimeForStore(entry)

	if len(entry.HardLinkId) > 0 {
		glog.V(4).InfofCtx(ctx, "InsertEntryKnownAbsent %s has HardLinkId %x counter=%d",
			entry.FullPath, entry.HardLinkId, entry.HardLinkCounter)
		if err := fsw.setHardLink(ctx, entry); err != nil {
			return fmt.Errorf("setHardLink %x: %v", entry.HardLinkId, err)
		}
	}

	return actualStore.InsertEntry(ctx, entry)
}

func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(entry.FullPath)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "update").Observe(time.Since(start).Seconds())
	}()

	filer_pb.BeforeEntrySerialization(entry.GetChunks())
	normalizeEntryMimeForStore(entry)

	if len(entry.HardLinkId) > 0 {
		glog.V(4).InfofCtx(ctx, "UpdateEntry %s has HardLinkId %x counter=%d",
			entry.FullPath, entry.HardLinkId, entry.HardLinkCounter)
	}

	if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
		return err
	}

	return actualStore.UpdateEntry(ctx, entry)
}

func normalizeEntryMimeForStore(entry *Entry) {
	if entry == nil {
		return
	}
	if !entry.IsDirectory() && entry.Mime == "application/octet-stream" {
		entry.Mime = ""
	}
}

func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (entry *Entry, err error) {
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(fp)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "find").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "find").Observe(time.Since(start).Seconds())
	}()

	entry, err = actualStore.FindEntry(ctx, fp)
	// glog.V(4).Infof("FindEntry %s: %v", fp, err)
	if err != nil {
		if fsw.CanDropWholeBucket() && strings.Contains(err.Error(), "Table") && strings.Contains(err.Error(), "doesn't exist") {
			err = filer_pb.ErrNotFound
		}
		return nil, err
	}

	fsw.maybeReadHardLink(ctx, entry)

	filer_pb.AfterEntryDeserialization(entry.GetChunks())
	return
}

func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(fp)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
	}()

	existingEntry, findErr := fsw.FindEntry(ctx, fp)
	if findErr == filer_pb.ErrNotFound || existingEntry == nil {
		return nil
	}
	if len(existingEntry.HardLinkId) != 0 {
		// remove hard link
		op := ctx.Value("OP")
		if op != "MV" {
			glog.V(4).InfofCtx(ctx, "DeleteHardLink %s", existingEntry.FullPath)
			if hlErr := fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); hlErr != nil {
				// Log but continue: the directory entry must be removed
				// even if hard link counter cleanup fails, otherwise the
				// parent directory cannot be removed (rmdir ENOTEMPTY).
				glog.Warningf("DeleteHardLink %s (id %x): %v", existingEntry.FullPath, existingEntry.HardLinkId, hlErr)
			}
		}
	}

	return actualStore.DeleteEntry(ctx, fp)
}

func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(existingEntry.FullPath)
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
	}()

	if len(existingEntry.HardLinkId) != 0 {
		// remove hard link
		op := ctx.Value("OP")
		if op != "MV" {
			glog.V(4).InfofCtx(ctx, "DeleteHardLink %s", existingEntry.FullPath)
			if hlErr := fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); hlErr != nil {
				// Log the hard link cleanup error but continue to delete
				// the directory entry. If we return early here, the entry
				// remains in the store and the parent directory cannot be
				// removed (rmdir returns ENOTEMPTY).
				glog.Warningf("DeleteHardLink %s (id %x): %v", existingEntry.FullPath, existingEntry.HardLinkId, hlErr)
			}
		}
	}

	return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
}

func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(fp + "/")
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
	}()

	return actualStore.DeleteFolderChildren(ctx, fp)
}

func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(dirPath + "/")
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "list").Observe(time.Since(start).Seconds())
	}()

	// glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
	return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) {
		fsw.maybeReadHardLink(ctx, entry)
		filer_pb.AfterEntryDeserialization(entry.GetChunks())
		return eachEntryFunc(entry)
	})
}

func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
	ctx = context.WithoutCancel(ctx)
	actualStore := fsw.getActualStore(dirPath + "/")
	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
	start := time.Now()
	defer func() {
		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
	}()
	if limit > math.MaxInt32-1 {
		limit = math.MaxInt32 - 1
	}
	// glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
	adjustedEntryFunc := func(entry *Entry) (bool, error) {
		fsw.maybeReadHardLink(ctx, entry)
		filer_pb.AfterEntryDeserialization(entry.GetChunks())
		return eachEntryFunc(entry)
	}
	lastFileName, err = actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
	if err == ErrUnsupportedListDirectoryPrefixed {
		lastFileName, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
	}
	return lastFileName, err
}

func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
	actualStore := fsw.getActualStore(dirPath + "/")

	if prefix == "" {
		return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
	}

	var notPrefixed []*Entry
	lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) {
		notPrefixed = append(notPrefixed, entry)
		return true, nil
	})
	if err != nil {
		return
	}

	count := int64(0)
	for count < limit && len(notPrefixed) > 0 {
		for _, entry := range notPrefixed {
			if strings.HasPrefix(entry.Name(), prefix) {
				count++
				res, resErr := eachEntryFunc(entry)

				if resErr != nil {
					err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr)
					return
				}

				if !res {
					return
				}
				if count >= limit {
					break
				}
			}
		}
		if count < limit && lastFileName < prefix {
			notPrefixed = notPrefixed[:0]
			lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) (bool, error) {
				notPrefixed = append(notPrefixed, entry)
				return true, nil
			})
			if err != nil {
				return
			}
		} else {
			break
		}
	}
	return
}

func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().BeginTransaction(ctx)
}

func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().CommitTransaction(ctx)
}

func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().RollbackTransaction(ctx)
}

func (fsw *FilerStoreWrapper) Shutdown() {
	fsw.getDefaultStore().Shutdown()
}

func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().KvPut(ctx, key, value)
}
func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().KvGet(ctx, key)
}
func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	ctx = context.WithoutCancel(ctx)
	return fsw.getDefaultStore().KvDelete(ctx, key)
}

func (fsw *FilerStoreWrapper) Debug(writer io.Writer) {
	if debuggable, ok := fsw.getDefaultStore().(Debuggable); ok {
		debuggable.Debug(writer)
	}
}