feat(fs): support manually trigger objs update hook (#1620)

* feat(fs): support manually trigger objs update hook

* fix: support driver internal copy & move case

* fix

* fix: apply suggestions of Copilot

internal/bootstrap/data/setting.go

@@ -177,6 +177,8 @@ func InitialSettings() []model.SettingItem {
 		{Key: conf.ShareArchivePreview, Value: "false", Type: conf.TypeBool, Group: model.GLOBAL, Flag: model.PUBLIC},
 		{Key: conf.ShareForceProxy, Value: "true", Type: conf.TypeBool, Group: model.GLOBAL, Flag: model.PRIVATE},
 		{Key: conf.ShareSummaryContent, Value: "@{{creator}} shared {{#each files}}{{#if @first}}\"{{filename this}}\"{{/if}}{{#if @last}}{{#unless (eq @index 0)}} and {{@index}} more files{{/unless}}{{/if}}{{/each}} from {{site_title}}: {{base_url}}/@s/{{id}}{{#if pwd}} , the share code is {{pwd}}{{/if}}{{#if expires}}, please access before {{dateLocaleString expires}}.{{/if}}", Type: conf.TypeText, Group: model.GLOBAL, Flag: model.PUBLIC},
+		{Key: conf.HandleHookAfterWriting, Value: "false", Type: conf.TypeBool, Group: model.GLOBAL, Flag: model.PRIVATE},
+		{Key: conf.HandleHookRateLimit, Value: "0", Type: conf.TypeNumber, Group: model.GLOBAL, Flag: model.PRIVATE},
 		{Key: conf.IgnoreSystemFiles, Value: "false", Type: conf.TypeBool, Group: model.GLOBAL, Flag: model.PRIVATE, Help: `When enabled, ignores common system files during upload (.DS_Store, desktop.ini, Thumbs.db, and files starting with ._)`},
 
 		// single settings

internal/conf/const.go

@@ -56,6 +56,8 @@ const (
 	ShareArchivePreview     = "share_archive_preview"
 	ShareForceProxy         = "share_force_proxy"
 	ShareSummaryContent     = "share_summary_content"
+	HandleHookAfterWriting  = "handle_hook_after_writing"
+	HandleHookRateLimit     = "handle_hook_rate_limit"
 	IgnoreSystemFiles       = "ignore_system_files"
 
 	// index

internal/op/archive.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/OpenListTeam/OpenList/v4/internal/archive/tool"
 	"github.com/OpenListTeam/OpenList/v4/internal/cache"
+	"github.com/OpenListTeam/OpenList/v4/internal/conf"
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/errs"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
@@ -20,10 +21,13 @@ import (
 	gocache "github.com/OpenListTeam/go-cache"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 )
 
-var archiveMetaCache = gocache.NewMemCache(gocache.WithShards[*model.ArchiveMetaProvider](64))
-var archiveMetaG singleflight.Group[*model.ArchiveMetaProvider]
+var (
+	archiveMetaCache = gocache.NewMemCache(gocache.WithShards[*model.ArchiveMetaProvider](64))
+	archiveMetaG     singleflight.Group[*model.ArchiveMetaProvider]
+)
 
 func GetArchiveMeta(ctx context.Context, storage driver.Driver, path string, args model.ArchiveMetaArgs) (*model.ArchiveMetaProvider, error) {
 	if storage.Config().CheckStatus && storage.GetStorage().Status != WORK {
@@ -196,8 +200,10 @@ func getArchiveMeta(ctx context.Context, storage driver.Driver, path string, arg
 	return obj, archiveMetaProvider, err
 }
 
-var archiveListCache = gocache.NewMemCache(gocache.WithShards[[]model.Obj](64))
-var archiveListG singleflight.Group[[]model.Obj]
+var (
+	archiveListCache = gocache.NewMemCache(gocache.WithShards[[]model.Obj](64))
+	archiveListG     singleflight.Group[[]model.Obj]
+)
 
 func ListArchive(ctx context.Context, storage driver.Driver, path string, args model.ArchiveListArgs) ([]model.Obj, error) {
 	if storage.Config().CheckStatus && storage.GetStorage().Status != WORK {
@@ -397,8 +403,10 @@ type objWithLink struct {
 	obj  model.Obj
 }
 
-var extractCache = cache.NewKeyedCache[*objWithLink](5 * time.Minute)
-var extractG = singleflight.Group[*objWithLink]{}
+var (
+	extractCache = cache.NewKeyedCache[*objWithLink](5 * time.Minute)
+	extractG     = singleflight.Group[*objWithLink]{}
+)
 
 func DriverExtract(ctx context.Context, storage driver.Driver, path string, args model.ArchiveInnerArgs) (*model.Link, model.Obj, error) {
 	if storage.Config().CheckStatus && storage.GetStorage().Status != WORK {
@@ -506,9 +514,9 @@ func ArchiveDecompress(ctx context.Context, storage driver.Driver, srcPath, dstD
 		return errors.WithMessage(err, "failed to get dst dir")
 	}
 
+	var newObjs []model.Obj
 	switch s := storage.(type) {
 	case driver.ArchiveDecompressResult:
-		var newObjs []model.Obj
 		newObjs, err = s.ArchiveDecompress(ctx, srcObj, dstDir, args)
 		if err == nil {
 			if len(newObjs) > 0 {
@@ -527,5 +535,31 @@ func ArchiveDecompress(ctx context.Context, storage driver.Driver, srcPath, dstD
 	default:
 		return errs.NotImplement
 	}
+	if !utils.IsBool(lazyCache...) && err == nil && needHandleObjsUpdateHook() {
+		onlyList := false
+		targetPath := dstDirPath
+		if newObjs != nil && len(newObjs) == 1 && newObjs[0].IsDir() {
+			targetPath = stdpath.Join(dstDirPath, newObjs[0].GetName())
+		} else if newObjs != nil && len(newObjs) == 1 && !newObjs[0].IsDir() {
+			onlyList = true
+		} else if args.PutIntoNewDir {
+			targetPath = stdpath.Join(dstDirPath, strings.TrimSuffix(srcObj.GetName(), stdpath.Ext(srcObj.GetName())))
+		} else if innerBase := stdpath.Base(args.InnerPath); innerBase != "." && innerBase != "/" {
+			targetPath = stdpath.Join(dstDirPath, innerBase)
+			dstObj, e := GetUnwrap(ctx, storage, targetPath)
+			onlyList = e != nil || !dstObj.IsDir()
+		}
+		if onlyList {
+			go List(context.Background(), storage, dstDirPath, model.ListArgs{Refresh: true})
+		} else {
+			var limiter *rate.Limiter
+			if l, _ := GetSettingItemByKey(conf.HandleHookRateLimit); l != nil {
+				if f, e := strconv.ParseFloat(l.Value, 64); e == nil && f > .0 {
+					limiter = rate.NewLimiter(rate.Limit(f), 1)
+				}
+			}
+			go RecursivelyListStorage(context.Background(), storage, targetPath, limiter, nil)
+		}
+	}
 	return errors.WithStack(err)
 }

internal/op/fs.go

@@ -2,10 +2,11 @@ package op
 
 import (
 	"context"
-	stderrors "errors"
 	stdpath "path"
+	"strconv"
 	"time"
 
+	"github.com/OpenListTeam/OpenList/v4/internal/conf"
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/errs"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
@@ -14,6 +15,7 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 )
 
 var listG singleflight.Group[[]model.Obj]
@@ -310,7 +312,7 @@ func Move(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string
 	srcDirPath := stdpath.Dir(srcPath)
 	dstDirPath = utils.FixAndCleanPath(dstDirPath)
 	if dstDirPath == srcDirPath {
-		return stderrors.New("move in place")
+		return errors.New("move in place")
 	}
 	srcRawObj, err := Get(ctx, storage, srcPath)
 	if err != nil {
@@ -343,8 +345,24 @@ func Move(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string
 			}
 		}
 	default:
-		return errs.NotImplement
+		err = errs.NotImplement
 	}
+
+	if !utils.IsBool(lazyCache...) && err == nil && needHandleObjsUpdateHook() {
+		if !srcObj.IsDir() {
+			go List(context.Background(), storage, dstDirPath, model.ListArgs{Refresh: true})
+		} else {
+			targetPath := stdpath.Join(dstDirPath, srcObj.GetName())
+			var limiter *rate.Limiter
+			if l, _ := GetSettingItemByKey(conf.HandleHookRateLimit); l != nil {
+				if f, e := strconv.ParseFloat(l.Value, 64); e == nil && f > .0 {
+					limiter = rate.NewLimiter(rate.Limit(f), 1)
+				}
+			}
+			go RecursivelyListStorage(context.Background(), storage, targetPath, limiter, nil)
+		}
+	}
+
 	return errors.WithStack(err)
 }
 
@@ -397,7 +415,7 @@ func Copy(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string
 	srcPath = utils.FixAndCleanPath(srcPath)
 	dstDirPath = utils.FixAndCleanPath(dstDirPath)
 	if dstDirPath == stdpath.Dir(srcPath) {
-		return stderrors.New("copy in place")
+		return errors.New("copy in place")
 	}
 	srcRawObj, err := Get(ctx, storage, srcPath)
 	if err != nil {
@@ -428,8 +446,24 @@ func Copy(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string
 			}
 		}
 	default:
-		return errs.NotImplement
+		err = errs.NotImplement
 	}
+
+	if !utils.IsBool(lazyCache...) && err == nil && needHandleObjsUpdateHook() {
+		if !srcObj.IsDir() {
+			go List(context.Background(), storage, dstDirPath, model.ListArgs{Refresh: true})
+		} else {
+			targetPath := stdpath.Join(dstDirPath, srcObj.GetName())
+			var limiter *rate.Limiter
+			if l, _ := GetSettingItemByKey(conf.HandleHookRateLimit); l != nil {
+				if f, e := strconv.ParseFloat(l.Value, 64); e == nil && f > .0 {
+					limiter = rate.NewLimiter(rate.Limit(f), 1)
+				}
+			}
+			go RecursivelyListStorage(context.Background(), storage, targetPath, limiter, nil)
+		}
+	}
+
 	return errors.WithStack(err)
 }
 
@@ -557,6 +591,9 @@ func Put(ctx context.Context, storage driver.Driver, dstDirPath string, file mod
 			err = Remove(ctx, storage, tempPath)
 		}
 	}
+	if !utils.IsBool(lazyCache...) && err == nil && needHandleObjsUpdateHook() {
+		go List(context.Background(), storage, dstDirPath, model.ListArgs{Refresh: true})
+	}
 	return errors.WithStack(err)
 }
 
@@ -601,6 +638,9 @@ func PutURL(ctx context.Context, storage driver.Driver, dstDirPath, dstName, url
 	default:
 		return errors.WithStack(errs.NotImplement)
 	}
+	if !utils.IsBool(lazyCache...) && err == nil && needHandleObjsUpdateHook() {
+		go List(context.Background(), storage, dstDirPath, model.ListArgs{Refresh: true})
+	}
 	log.Debugf("put url [%s](%s) done", dstName, url)
 	return errors.WithStack(err)
 }
@@ -644,3 +684,8 @@ func GetDirectUploadInfo(ctx context.Context, tool string, storage driver.Driver
 	}
 	return info, nil
 }
+
+func needHandleObjsUpdateHook() bool {
+	needHandle, _ := GetSettingItemByKey(conf.HandleHookAfterWriting)
+	return needHandle != nil && (needHandle.Value == "true" || needHandle.Value == "1")
+}

internal/op/recursive_list.go

@@ -0,0 +1,125 @@
+package op
+
+import (
+	"context"
+	stdpath "path"
+	"sync"
+	"sync/atomic"
+
+	"github.com/OpenListTeam/OpenList/v4/internal/driver"
+	"github.com/OpenListTeam/OpenList/v4/internal/errs"
+	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
+)
+
+var (
+	ManualScanCancel = atomic.Pointer[context.CancelFunc]{}
+	ScannedCount     = atomic.Uint64{}
+)
+
+func ManualScanRunning() bool {
+	return ManualScanCancel.Load() != nil
+}
+
+func BeginManualScan(rawPath string, limit float64) error {
+	rawPath = utils.FixAndCleanPath(rawPath)
+	ctx, cancel := context.WithCancel(context.Background())
+	if !ManualScanCancel.CompareAndSwap(nil, &cancel) {
+		cancel()
+		return errors.New("manual scan is running, please try later")
+	}
+	ScannedCount.Store(0)
+	go func() {
+		defer func() { (*ManualScanCancel.Swap(nil))() }()
+		err := RecursivelyList(ctx, rawPath, rate.Limit(limit), &ScannedCount)
+		if err != nil {
+			log.Errorf("failed recursively list: %v", err)
+		}
+	}()
+	return nil
+}
+
+func StopManualScan() {
+	c := ManualScanCancel.Load()
+	if c != nil {
+		(*c)()
+	}
+}
+
+func RecursivelyList(ctx context.Context, rawPath string, limit rate.Limit, counter *atomic.Uint64) error {
+	storage, actualPath, err := GetStorageAndActualPath(rawPath)
+	if err != nil && !errors.Is(err, errs.StorageNotFound) {
+		return err
+	} else if err == nil {
+		var limiter *rate.Limiter
+		if limit > .0 {
+			limiter = rate.NewLimiter(limit, 1)
+		}
+		RecursivelyListStorage(ctx, storage, actualPath, limiter, counter)
+	} else {
+		var wg sync.WaitGroup
+		recursivelyListVirtual(ctx, rawPath, limit, counter, &wg)
+		wg.Wait()
+	}
+	return nil
+}
+
+func recursivelyListVirtual(ctx context.Context, rawPath string, limit rate.Limit, counter *atomic.Uint64, wg *sync.WaitGroup) {
+	objs := GetStorageVirtualFilesByPath(rawPath)
+	if counter != nil {
+		counter.Add(uint64(len(objs)))
+	}
+	for _, obj := range objs {
+		if utils.IsCanceled(ctx) {
+			return
+		}
+		nextPath := stdpath.Join(rawPath, obj.GetName())
+		storage, actualPath, err := GetStorageAndActualPath(nextPath)
+		if err != nil && !errors.Is(err, errs.StorageNotFound) {
+			log.Errorf("error recursively list: failed get storage [%s]: %v", nextPath, err)
+		} else if err == nil {
+			var limiter *rate.Limiter
+			if limit > .0 {
+				limiter = rate.NewLimiter(limit, 1)
+			}
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				RecursivelyListStorage(ctx, storage, actualPath, limiter, counter)
+			}()
+		} else {
+			recursivelyListVirtual(ctx, nextPath, limit, counter, wg)
+		}
+	}
+}
+
+func RecursivelyListStorage(ctx context.Context, storage driver.Driver, actualPath string, limiter *rate.Limiter, counter *atomic.Uint64) {
+	objs, err := List(ctx, storage, actualPath, model.ListArgs{Refresh: true})
+	if err != nil {
+		if !errors.Is(err, context.Canceled) {
+			log.Errorf("error recursively list: failed list (%s)[%s]: %v", storage.GetStorage().MountPath, actualPath, err)
+		}
+		return
+	}
+	if counter != nil {
+		counter.Add(uint64(len(objs)))
+	}
+	for _, obj := range objs {
+		if utils.IsCanceled(ctx) {
+			return
+		}
+		if !obj.IsDir() {
+			continue
+		}
+		if limiter != nil {
+			if err = limiter.Wait(ctx); err != nil {
+				return
+			}
+		}
+		nextPath := stdpath.Join(actualPath, obj.GetName())
+		RecursivelyListStorage(ctx, storage, nextPath, limiter, counter)
+	}
+}

internal/setting/setting.go

@@ -28,3 +28,11 @@ func GetInt(key string, defaultVal int) int {
 func GetBool(key string) bool {
 	return GetStr(key) == "true" || GetStr(key) == "1"
 }
+
+func GetFloat(key string, defaultVal float64) float64 {
+	f, err := strconv.ParseFloat(GetStr(key), 64)
+	if err != nil {
+		return defaultVal
+	}
+	return f
+}

internal/task_group/transfer.go

@@ -5,11 +5,14 @@ import (
 	"fmt"
 	"path"
 
+	"github.com/OpenListTeam/OpenList/v4/internal/conf"
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
 	"github.com/OpenListTeam/OpenList/v4/internal/op"
+	"github.com/OpenListTeam/OpenList/v4/internal/setting"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 )
 
 type SrcPathToRemove string
@@ -27,12 +30,31 @@ func RefreshAndRemove(dstPath string, payloads ...any) {
 	if dstNeedRefresh {
 		op.Cache.DeleteDirectory(dstStorage, dstActualPath)
 	}
+	dstNeedHandleHook := setting.GetBool(conf.HandleHookAfterWriting)
+	dstHandleHookLimit := setting.GetFloat(conf.HandleHookRateLimit, .0)
+	var listLimiter *rate.Limiter
+	if dstNeedRefresh && dstNeedHandleHook && dstHandleHookLimit > .0 {
+		listLimiter = rate.NewLimiter(rate.Limit(dstHandleHookLimit), 1)
+	}
 	var ctx context.Context
 	for _, payload := range payloads {
 		switch p := payload.(type) {
 		case DstPathToRefresh:
 			if dstNeedRefresh {
-				op.Cache.DeleteDirectory(dstStorage, string(p))
+				if dstNeedHandleHook {
+					if ctx == nil {
+						ctx = context.Background()
+					}
+					if listLimiter != nil {
+						_ = listLimiter.Wait(ctx)
+					}
+					_, e := op.List(ctx, dstStorage, string(p), model.ListArgs{Refresh: true})
+					if e != nil {
+						log.Errorf("failed handle objs update hook: %v", e)
+					}
+				} else {
+					op.Cache.DeleteDirectory(dstStorage, string(p))
+				}
 			}
 		case SrcPathToRemove:
 			if ctx == nil {

server/handles/scan.go

@@ -0,0 +1,47 @@
+package handles
+
+import (
+	"github.com/OpenListTeam/OpenList/v4/internal/op"
+	"github.com/OpenListTeam/OpenList/v4/server/common"
+	"github.com/gin-gonic/gin"
+)
+
+type ManualScanReq struct {
+	Path  string  `json:"path"`
+	Limit float64 `json:"limit"`
+}
+
+func StartManualScan(c *gin.Context) {
+	var req ManualScanReq
+	if err := c.ShouldBind(&req); err != nil {
+		common.ErrorResp(c, err, 400)
+		return
+	}
+	if err := op.BeginManualScan(req.Path, req.Limit); err != nil {
+		common.ErrorResp(c, err, 400)
+		return
+	}
+	common.SuccessResp(c)
+}
+
+func StopManualScan(c *gin.Context) {
+	if !op.ManualScanRunning() {
+		common.ErrorStrResp(c, "manual scan is not running", 400)
+		return
+	}
+	op.StopManualScan()
+	common.SuccessResp(c)
+}
+
+type ManualScanResp struct {
+	ObjCount uint64 `json:"obj_count"`
+	IsDone   bool   `json:"is_done"`
+}
+
+func GetManualScanProgress(c *gin.Context) {
+	ret := ManualScanResp{
+		ObjCount: op.ScannedCount.Load(),
+		IsDone:   !op.ManualScanRunning(),
+	}
+	common.SuccessResp(c, ret)
+}

server/middlewares/search.go

@@ -11,7 +11,7 @@ import (
 func SearchIndex(c *gin.Context) {
 	mode := setting.GetStr(conf.SearchIndex)
 	if mode == "none" {
-		common.ErrorResp(c, errs.SearchNotAvailable, 500)
+		common.ErrorResp(c, errs.SearchNotAvailable, 404)
 		c.Abort()
 	} else {
 		c.Next()

server/router.go

@@ -179,6 +179,11 @@ func admin(g *gin.RouterGroup) {
 	index.POST("/stop", middlewares.SearchIndex, handles.StopIndex)
 	index.POST("/clear", middlewares.SearchIndex, handles.ClearIndex)
 	index.GET("/progress", middlewares.SearchIndex, handles.GetProgress)
+
+	scan := g.Group("/scan")
+	scan.POST("/start", handles.StartManualScan)
+	scan.POST("/stop", handles.StopManualScan)
+	scan.GET("/progress", handles.GetManualScanProgress)
 }
 
 func fsAndShare(g *gin.RouterGroup) {