diff --git a/drivers/halalcloud_open/driver.go b/drivers/halalcloud_open/driver.go index 6a74538d47..a6087e99b1 100644 --- a/drivers/halalcloud_open/driver.go +++ b/drivers/halalcloud_open/driver.go @@ -15,6 +15,7 @@ type HalalCloudOpen struct { sdkClient *sdkClient.Client sdkUserFileService *sdkUserFile.UserFileService sdkUserService *sdkUser.UserService + offlineTaskService offlineTaskService uploadThread int } diff --git a/drivers/halalcloud_open/driver_init.go b/drivers/halalcloud_open/driver_init.go index 9f70263809..a908d6289f 100644 --- a/drivers/halalcloud_open/driver_init.go +++ b/drivers/halalcloud_open/driver_init.go @@ -2,10 +2,12 @@ package halalcloudopen import ( "context" + "net/http" "time" "github.com/OpenListTeam/OpenList/v4/internal/op" "github.com/halalcloud/golang-sdk-lite/halalcloud/apiclient" + sdkOffline "github.com/halalcloud/golang-sdk-lite/halalcloud/services/offline" sdkUser "github.com/halalcloud/golang-sdk-lite/halalcloud/services/user" sdkUserFile "github.com/halalcloud/golang-sdk-lite/halalcloud/services/userfile" ) @@ -36,10 +38,18 @@ func (d *HalalCloudOpen) Init(ctx context.Context) error { host = "openapi.2dland.cn" } - client := apiclient.NewClient(nil, host, d.Addition.ClientID, d.Addition.ClientSecret, d.halalCommon, apiclient.WithTimeout(time.Second*time.Duration(timeout))) + // All SDK services share the quota associated with these credentials. + httpClient := &http.Client{ + Transport: &rateLimitedTransport{ + base: http.DefaultTransport, + limiter: halalCloudAPILimiter(host, d.Addition.ClientID), + }, + } + client := apiclient.NewClient(httpClient, host, d.Addition.ClientID, d.Addition.ClientSecret, d.halalCommon, apiclient.WithTimeout(time.Second*time.Duration(timeout))) d.sdkClient = client d.sdkUserFileService = sdkUserFile.NewUserFileService(client) d.sdkUserService = sdkUser.NewUserService(client) + d.offlineTaskService = sdkOffline.NewOfflineTaskService(client) userInfo, err := d.sdkUserService.Get(ctx, &sdkUser.User{}) if err != nil { return err diff --git a/drivers/halalcloud_open/offline.go b/drivers/halalcloud_open/offline.go new file mode 100644 index 0000000000..168959dba6 --- /dev/null +++ b/drivers/halalcloud_open/offline.go @@ -0,0 +1,118 @@ +package halalcloudopen + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/OpenListTeam/OpenList/v4/internal/model" + sdkModel "github.com/halalcloud/golang-sdk-lite/halalcloud/model" + sdkOffline "github.com/halalcloud/golang-sdk-lite/halalcloud/services/offline" +) + +// A larger page reduces requests during status polling. +const offlineTaskListPageSize int64 = 200 + +type offlineTaskService interface { + Add(ctx context.Context, req *sdkOffline.UserTask) (*sdkOffline.UserTask, error) + List(ctx context.Context, req *sdkOffline.OfflineTaskListRequest) (*sdkOffline.OfflineTaskListResponse, error) + Delete(ctx context.Context, req *sdkOffline.OfflineTaskDeleteRequest) (*sdkOffline.OfflineTaskDeleteResponse, error) +} + +// OfflineDownload creates a URL-based task that writes directly into parentDir. +func (d *HalalCloudOpen) OfflineDownload(ctx context.Context, fileURL string, parentDir model.Obj) (*sdkOffline.UserTask, error) { + if d.offlineTaskService == nil { + return nil, errors.New("HalalCloudOpen offline task service is not initialized") + } + + fileURL = strings.TrimSpace(fileURL) + if fileURL == "" { + return nil, errors.New("HalalCloudOpen offline download URL is empty") + } + + // Let HalalCloud dispatch the URL to its supported task type. + task, err := d.offlineTaskService.Add(ctx, &sdkOffline.UserTask{ + Url: fileURL, + SavePath: parentDir.GetPath(), + }) + if err != nil { + return nil, fmt.Errorf("failed to create HalalCloudOpen offline task: %w", err) + } + if task == nil || strings.TrimSpace(task.Identity) == "" { + return nil, errors.New("failed to create HalalCloudOpen offline task: empty task identity") + } + // Identity is the user-task handle shared by Add, List, and Delete. + return task, nil +} + +// OfflineList returns every user task, following the opaque pagination token +// until the API reports that traversal is complete. +func (d *HalalCloudOpen) OfflineList(ctx context.Context) ([]*sdkOffline.UserTask, error) { + if d.offlineTaskService == nil { + return nil, errors.New("HalalCloudOpen offline task service is not initialized") + } + + tasks := make([]*sdkOffline.UserTask, 0) + token := "" + seenTokens := make(map[string]struct{}) + for { + resp, err := d.offlineTaskService.List(ctx, &sdkOffline.OfflineTaskListRequest{ + ListInfo: &sdkModel.ScanListRequest{ + Limit: offlineTaskListPageSize, + Token: token, + }, + }) + if err != nil { + return nil, fmt.Errorf("failed to list HalalCloudOpen offline tasks: %w", err) + } + if resp == nil { + return nil, errors.New("failed to list HalalCloudOpen offline tasks: empty response") + } + tasks = append(tasks, resp.Tasks...) + + if resp.ListInfo == nil || resp.ListInfo.Token == "" { + break + } + // ListInfo.Token is opaque. Guard against a malformed response repeating + // a token so one status poll cannot loop forever. + nextToken := resp.ListInfo.Token + if nextToken == token { + return nil, errors.New("failed to list HalalCloudOpen offline tasks: pagination token did not advance") + } + if _, ok := seenTokens[nextToken]; ok { + return nil, errors.New("failed to list HalalCloudOpen offline tasks: pagination token repeated") + } + seenTokens[nextToken] = struct{}{} + token = nextToken + } + return tasks, nil +} + +// DeleteOfflineTasks removes task records and optionally their downloaded files. +func (d *HalalCloudOpen) DeleteOfflineTasks(ctx context.Context, taskIDs []string, deleteFiles bool) error { + if d.offlineTaskService == nil { + return errors.New("HalalCloudOpen offline task service is not initialized") + } + + identities := make([]string, 0, len(taskIDs)) + for _, taskID := range taskIDs { + if taskID = strings.TrimSpace(taskID); taskID != "" { + identities = append(identities, taskID) + } + } + if len(identities) == 0 { + return nil + } + + _, err := d.offlineTaskService.Delete(ctx, &sdkOffline.OfflineTaskDeleteRequest{ + Identity: identities, + DeleteFiles: deleteFiles, + }) + if err != nil { + return fmt.Errorf("failed to delete HalalCloudOpen offline tasks: %w", err) + } + return nil +} + +var _ offlineTaskService = (*sdkOffline.OfflineTaskService)(nil) diff --git a/drivers/halalcloud_open/offline_test.go b/drivers/halalcloud_open/offline_test.go new file mode 100644 index 0000000000..8b79bb2e67 --- /dev/null +++ b/drivers/halalcloud_open/offline_test.go @@ -0,0 +1,188 @@ +package halalcloudopen + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/model" + sdkModel "github.com/halalcloud/golang-sdk-lite/halalcloud/model" + sdkOffline "github.com/halalcloud/golang-sdk-lite/halalcloud/services/offline" +) + +type fakeOfflineTaskService struct { + add func(context.Context, *sdkOffline.UserTask) (*sdkOffline.UserTask, error) + list func(context.Context, *sdkOffline.OfflineTaskListRequest) (*sdkOffline.OfflineTaskListResponse, error) + delete func(context.Context, *sdkOffline.OfflineTaskDeleteRequest) (*sdkOffline.OfflineTaskDeleteResponse, error) +} + +func (f *fakeOfflineTaskService) Add(ctx context.Context, req *sdkOffline.UserTask) (*sdkOffline.UserTask, error) { + if f.add == nil { + return nil, errors.New("unexpected Add call") + } + return f.add(ctx, req) +} + +func (f *fakeOfflineTaskService) List(ctx context.Context, req *sdkOffline.OfflineTaskListRequest) (*sdkOffline.OfflineTaskListResponse, error) { + if f.list == nil { + return nil, errors.New("unexpected List call") + } + return f.list(ctx, req) +} + +func (f *fakeOfflineTaskService) Delete(ctx context.Context, req *sdkOffline.OfflineTaskDeleteRequest) (*sdkOffline.OfflineTaskDeleteResponse, error) { + if f.delete == nil { + return nil, errors.New("unexpected Delete call") + } + return f.delete(ctx, req) +} + +func TestOfflineDownload(t *testing.T) { + var got *sdkOffline.UserTask + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + add: func(_ context.Context, req *sdkOffline.UserTask) (*sdkOffline.UserTask, error) { + got = req + return &sdkOffline.UserTask{Identity: "task-id"}, nil + }, + }, + } + parent := &model.Object{Path: "/downloads", IsFolder: true} + + task, err := driver.OfflineDownload(context.Background(), " https://example.com/file ", parent) + if err != nil { + t.Fatalf("OfflineDownload() error = %v", err) + } + if task.Identity != "task-id" { + t.Fatalf("OfflineDownload() identity = %q, want %q", task.Identity, "task-id") + } + if got == nil { + t.Fatal("OfflineDownload() did not call the SDK service") + } + if got.Url != "https://example.com/file" { + t.Errorf("Add request URL = %q, want trimmed task URL", got.Url) + } + if got.SavePath != "/downloads" { + t.Errorf("Add request SavePath = %q, want %q", got.SavePath, "/downloads") + } +} + +func TestOfflineDownloadRejectsEmptyURL(t *testing.T) { + called := false + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + add: func(_ context.Context, _ *sdkOffline.UserTask) (*sdkOffline.UserTask, error) { + called = true + return nil, nil + }, + }, + } + parent := &model.Object{Path: "/downloads", IsFolder: true} + + _, err := driver.OfflineDownload(context.Background(), " ", parent) + if err == nil || !strings.Contains(err.Error(), "URL is empty") { + t.Fatalf("OfflineDownload() error = %v, want empty URL error", err) + } + if called { + t.Fatal("OfflineDownload() called the SDK service for an empty URL") + } +} + +func TestOfflineDownloadRequiresTaskIdentity(t *testing.T) { + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + add: func(_ context.Context, _ *sdkOffline.UserTask) (*sdkOffline.UserTask, error) { + return &sdkOffline.UserTask{}, nil + }, + }, + } + + _, err := driver.OfflineDownload(context.Background(), "https://example.com/file", &model.Object{Path: "/"}) + if err == nil || !strings.Contains(err.Error(), "empty task identity") { + t.Fatalf("OfflineDownload() error = %v, want empty identity error", err) + } +} + +func TestOfflineListPaginates(t *testing.T) { + var requests []*sdkOffline.OfflineTaskListRequest + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + list: func(_ context.Context, req *sdkOffline.OfflineTaskListRequest) (*sdkOffline.OfflineTaskListResponse, error) { + requests = append(requests, req) + if req.ListInfo.Token == "" { + return &sdkOffline.OfflineTaskListResponse{ + Tasks: []*sdkOffline.UserTask{{Identity: "first"}}, + ListInfo: &sdkModel.ScanListRequest{Token: "next"}, + }, nil + } + return &sdkOffline.OfflineTaskListResponse{ + Tasks: []*sdkOffline.UserTask{{Identity: "second"}}, + ListInfo: &sdkModel.ScanListRequest{}, + }, nil + }, + }, + } + + tasks, err := driver.OfflineList(context.Background()) + if err != nil { + t.Fatalf("OfflineList() error = %v", err) + } + if len(tasks) != 2 || tasks[0].Identity != "first" || tasks[1].Identity != "second" { + t.Fatalf("OfflineList() tasks = %#v, want first and second tasks", tasks) + } + if len(requests) != 2 { + t.Fatalf("OfflineList() request count = %d, want 2", len(requests)) + } + if requests[0].ListInfo.Limit != offlineTaskListPageSize || requests[0].ListInfo.Token != "" { + t.Errorf("first List request = %#v, want initial page", requests[0].ListInfo) + } + if requests[1].ListInfo.Limit != offlineTaskListPageSize || requests[1].ListInfo.Token != "next" { + t.Errorf("second List request = %#v, want next page", requests[1].ListInfo) + } +} + +func TestOfflineListRejectsRepeatedToken(t *testing.T) { + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + list: func(_ context.Context, req *sdkOffline.OfflineTaskListRequest) (*sdkOffline.OfflineTaskListResponse, error) { + if req.ListInfo.Token == "" { + return &sdkOffline.OfflineTaskListResponse{ListInfo: &sdkModel.ScanListRequest{Token: "same"}}, nil + } + return &sdkOffline.OfflineTaskListResponse{ListInfo: &sdkModel.ScanListRequest{Token: "same"}}, nil + }, + }, + } + + _, err := driver.OfflineList(context.Background()) + if err == nil || !strings.Contains(err.Error(), "pagination token did not advance") { + t.Fatalf("OfflineList() error = %v, want repeated token error", err) + } +} + +func TestDeleteOfflineTasks(t *testing.T) { + var got *sdkOffline.OfflineTaskDeleteRequest + driver := &HalalCloudOpen{ + offlineTaskService: &fakeOfflineTaskService{ + delete: func(_ context.Context, req *sdkOffline.OfflineTaskDeleteRequest) (*sdkOffline.OfflineTaskDeleteResponse, error) { + got = req + return &sdkOffline.OfflineTaskDeleteResponse{Count: 2}, nil + }, + }, + } + + if err := driver.DeleteOfflineTasks(context.Background(), []string{" first ", "", "second"}, false); err != nil { + t.Fatalf("DeleteOfflineTasks() error = %v", err) + } + if got == nil { + t.Fatal("DeleteOfflineTasks() did not call the SDK service") + } + if len(got.Identity) != 2 || got.Identity[0] != "first" || got.Identity[1] != "second" { + t.Errorf("Delete request identities = %#v, want trimmed non-empty identities", got.Identity) + } + if got.DeleteFiles { + t.Error("Delete request DeleteFiles = true, want false") + } +} + +var _ offlineTaskService = (*fakeOfflineTaskService)(nil) diff --git a/drivers/halalcloud_open/rate_limit.go b/drivers/halalcloud_open/rate_limit.go new file mode 100644 index 0000000000..9b78b3d4ec --- /dev/null +++ b/drivers/halalcloud_open/rate_limit.go @@ -0,0 +1,37 @@ +package halalcloudopen + +import ( + "net/http" + "strings" + "sync" + "time" + + "golang.org/x/time/rate" +) + +const halalCloudAPIRequestInterval = time.Second + +// API quotas are shared by mounts that use the same credentials. +var halalCloudAPILimiters sync.Map + +func halalCloudAPILimiter(host, clientID string) *rate.Limiter { + key := strings.ToLower(strings.TrimSpace(host)) + "\x00" + strings.TrimSpace(clientID) + limiter, _ := halalCloudAPILimiters.LoadOrStore( + key, + rate.NewLimiter(rate.Every(halalCloudAPIRequestInterval), 1), + ) + return limiter.(*rate.Limiter) +} + +// rateLimitedTransport applies the credential quota to every SDK service. +type rateLimitedTransport struct { + base http.RoundTripper + limiter *rate.Limiter +} + +func (t *rateLimitedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if err := t.limiter.Wait(req.Context()); err != nil { + return nil, err + } + return t.base.RoundTrip(req) +} diff --git a/drivers/halalcloud_open/rate_limit_test.go b/drivers/halalcloud_open/rate_limit_test.go new file mode 100644 index 0000000000..4cb7ce9649 --- /dev/null +++ b/drivers/halalcloud_open/rate_limit_test.go @@ -0,0 +1,58 @@ +package halalcloudopen + +import ( + "context" + "net/http" + "testing" + "time" + + "golang.org/x/time/rate" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +func TestHalalCloudAPILimiterSharedByCredentials(t *testing.T) { + first := halalCloudAPILimiter("RATE-LIMIT-TEST.INVALID", t.Name()) + second := halalCloudAPILimiter("rate-limit-test.invalid", t.Name()) + if first != second { + t.Fatal("same host and client ID did not share an API limiter") + } + + differentClient := halalCloudAPILimiter("rate-limit-test.invalid", t.Name()+"-other") + if first == differentClient { + t.Fatal("different client IDs unexpectedly shared an API limiter") + } +} + +func TestRateLimitedTransportHonorsCanceledContext(t *testing.T) { + limiter := rate.NewLimiter(rate.Every(time.Hour), 1) + if !limiter.Allow() { + t.Fatal("failed to consume the limiter's initial token") + } + + baseCalled := false + transport := &rateLimitedTransport{ + base: roundTripFunc(func(*http.Request) (*http.Response, error) { + baseCalled = true + return &http.Response{StatusCode: http.StatusOK}, nil + }), + limiter: limiter, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://rate-limit-test.invalid", nil) + if err != nil { + t.Fatalf("NewRequestWithContext() error = %v", err) + } + + if _, err := transport.RoundTrip(req); err == nil { + t.Fatal("RoundTrip() error = nil, want canceled-context error") + } + if baseCalled { + t.Fatal("RoundTrip() called the base transport after limiter wait failed") + } +} diff --git a/internal/cache/keyed_cache.go b/internal/cache/keyed_cache.go index 07bf8cd9f9..b9723cde89 100644 --- a/internal/cache/keyed_cache.go +++ b/internal/cache/keyed_cache.go @@ -1,6 +1,7 @@ package cache import ( + "strings" "sync" "time" ) @@ -70,6 +71,29 @@ func (c *KeyedCache[T]) Delete(key string) { delete(c.entries, key) } +// DeletePrefix removes the key and all keys below it as a slash-delimited +// path. A path boundary prevents /foo from matching /foobar. +func (c *KeyedCache[T]) DeletePrefix(prefix string) { + c.mu.Lock() + defer c.mu.Unlock() + + for key := range c.entries { + if pathPrefixMatch(key, prefix) { + delete(c.entries, key) + } + } +} + +func pathPrefixMatch(key, prefix string) bool { + if key == prefix { + return true + } + if prefix == "/" { + return strings.HasPrefix(key, "/") + } + return strings.HasPrefix(key, prefix+"/") +} + func (c *KeyedCache[T]) Pop(key string) (T, bool) { c.mu.Lock() defer c.mu.Unlock() diff --git a/internal/cache/prefix_test.go b/internal/cache/prefix_test.go new file mode 100644 index 0000000000..505ef028d9 --- /dev/null +++ b/internal/cache/prefix_test.go @@ -0,0 +1,59 @@ +package cache + +import ( + "testing" + "time" +) + +func TestKeyedCacheDeletePrefix(t *testing.T) { + c := NewKeyedCache[int](time.Hour) + c.Set("/foo", 1) + c.Set("/foo/bar", 2) + c.Set("/foobar", 3) + + c.DeletePrefix("/foo") + + if _, ok := c.Get("/foo"); ok { + t.Fatal("DeletePrefix() kept the exact key") + } + if _, ok := c.Get("/foo/bar"); ok { + t.Fatal("DeletePrefix() kept a descendant key") + } + if value, ok := c.Get("/foobar"); !ok || value != 3 { + t.Fatalf("DeletePrefix() removed a sibling key: value=%d, ok=%v", value, ok) + } +} + +func TestKeyedCacheDeletePrefixRoot(t *testing.T) { + c := NewKeyedCache[int](time.Hour) + c.Set("/", 1) + c.Set("/foo", 2) + + c.DeletePrefix("/") + + if _, ok := c.Get("/"); ok { + t.Fatal("DeletePrefix(/) kept the root key") + } + if _, ok := c.Get("/foo"); ok { + t.Fatal("DeletePrefix(/) kept a descendant key") + } +} + +func TestTypedCacheDeleteKeyPrefix(t *testing.T) { + c := NewTypedCache[int](time.Hour) + c.SetType("/foo", "link", 1) + c.SetType("/foo/bar", "link", 2) + c.SetType("/foobar", "link", 3) + + c.DeleteKeyPrefix("/foo") + + if _, ok := c.GetType("/foo", "link"); ok { + t.Fatal("DeleteKeyPrefix() kept the exact key") + } + if _, ok := c.GetType("/foo/bar", "link"); ok { + t.Fatal("DeleteKeyPrefix() kept a descendant key") + } + if value, ok := c.GetType("/foobar", "link"); !ok || value != 3 { + t.Fatalf("DeleteKeyPrefix() removed a sibling key: value=%d, ok=%v", value, ok) + } +} diff --git a/internal/cache/typed_cache.go b/internal/cache/typed_cache.go index 7ba126be83..7dd2bfb39d 100644 --- a/internal/cache/typed_cache.go +++ b/internal/cache/typed_cache.go @@ -81,6 +81,18 @@ func (c *TypedCache[T]) DeleteKey(key string) { delete(c.entries, key) } +// DeleteKeyPrefix removes a key and all descendant path keys. +func (c *TypedCache[T]) DeleteKeyPrefix(prefix string) { + c.mu.Lock() + defer c.mu.Unlock() + + for key := range c.entries { + if pathPrefixMatch(key, prefix) { + delete(c.entries, key) + } + } +} + func (c *TypedCache[T]) Clear() { c.mu.Lock() defer c.mu.Unlock() diff --git a/internal/offline_download/all.go b/internal/offline_download/all.go index 7c9b9dcac8..b0996b2c16 100644 --- a/internal/offline_download/all.go +++ b/internal/offline_download/all.go @@ -7,6 +7,7 @@ import ( _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/123_open" _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/aria2" _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/guangyapan" + _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/halalcloud_open" _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/http" _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/pikpak" _ "github.com/OpenListTeam/OpenList/v4/internal/offline_download/qbit" diff --git a/internal/offline_download/halalcloud_open/cache_test.go b/internal/offline_download/halalcloud_open/cache_test.go new file mode 100644 index 0000000000..dc5aef5a55 --- /dev/null +++ b/internal/offline_download/halalcloud_open/cache_test.go @@ -0,0 +1,191 @@ +package halalcloudopen + +import ( + "context" + "path" + "sync" + "testing" + "time" + + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/op" +) + +type cacheTestDriver struct { + model.Storage + + mu sync.Mutex + listings map[string][]model.Obj + listCalls map[string]int + blockPath string + listStarted chan struct{} + listRelease chan struct{} + blockOnce sync.Once +} + +func (d *cacheTestDriver) Config() driver.Config { + return driver.Config{} +} + +func (d *cacheTestDriver) GetAddition() driver.Additional { + return nil +} + +func (d *cacheTestDriver) Init(context.Context) error { + return nil +} + +func (d *cacheTestDriver) Drop(context.Context) error { + return nil +} + +func (d *cacheTestDriver) Get(_ context.Context, objectPath string) (model.Obj, error) { + return &model.Object{ + Path: objectPath, + Name: path.Base(objectPath), + IsFolder: true, + }, nil +} + +func (d *cacheTestDriver) List(_ context.Context, dir model.Obj, _ model.ListArgs) ([]model.Obj, error) { + d.mu.Lock() + objectPath := dir.GetPath() + d.listCalls[objectPath]++ + objects := append([]model.Obj(nil), d.listings[objectPath]...) + d.mu.Unlock() + + if objectPath == d.blockPath { + d.blockOnce.Do(func() { + close(d.listStarted) + <-d.listRelease + }) + } + return objects, nil +} + +func (d *cacheTestDriver) Link(context.Context, model.Obj, model.LinkArgs) (*model.Link, error) { + return nil, nil +} + +func TestInvalidateDestinationCacheClearsDescendants(t *testing.T) { + driver := &cacheTestDriver{ + Storage: model.Storage{ + MountPath: "/halalcloud-cache-test-" + t.Name(), + CacheExpiration: 60, + }, + listings: map[string][]model.Obj{ + "/downloads": { + &model.Object{Path: "/downloads/bundle", Name: "bundle", IsFolder: true}, + }, + "/downloads/bundle": { + &model.Object{Path: "/downloads/bundle/old.txt", Name: "old.txt"}, + }, + }, + listCalls: make(map[string]int), + } + defer op.Cache.DeleteDirectoryTree(driver, "/downloads") + + ctx := context.Background() + listArgs := model.ListArgs{SkipHook: true} + if _, err := op.List(ctx, driver, "/downloads", listArgs); err != nil { + t.Fatalf("List(destination) error = %v", err) + } + if _, err := op.List(ctx, driver, "/downloads/bundle", listArgs); err != nil { + t.Fatalf("List(descendant) error = %v", err) + } + + driver.mu.Lock() + driver.listings["/downloads/bundle"] = []model.Obj{ + &model.Object{Path: "/downloads/bundle/new.txt", Name: "new.txt"}, + } + driver.mu.Unlock() + + // The parent listing may expire independently while a descendant remains + // cached; invalidation must still remove that descendant. + op.Cache.DeleteDirectory(driver, "/downloads") + invalidateDestinationCache(driver, "/downloads") + + objects, err := op.List(ctx, driver, "/downloads/bundle", listArgs) + if err != nil { + t.Fatalf("List(descendant after invalidation) error = %v", err) + } + if len(objects) != 1 || objects[0].GetName() != "new.txt" { + t.Fatalf("List(descendant after invalidation) = %#v, want new.txt", objects) + } + + driver.mu.Lock() + calls := driver.listCalls["/downloads/bundle"] + driver.mu.Unlock() + if calls != 2 { + t.Fatalf("descendant List call count = %d, want 2", calls) + } +} + +func TestInvalidateDestinationCachePreventsStaleInFlightWrite(t *testing.T) { + driver := &cacheTestDriver{ + Storage: model.Storage{ + MountPath: "/halalcloud-cache-race-" + t.Name(), + CacheExpiration: 60, + }, + listings: map[string][]model.Obj{ + "/downloads": { + &model.Object{Path: "/downloads/old.txt", Name: "old.txt"}, + }, + }, + listCalls: make(map[string]int), + blockPath: "/downloads", + listStarted: make(chan struct{}), + listRelease: make(chan struct{}), + } + defer op.Cache.DeleteDirectoryTree(driver, "/downloads") + + resultCh := make(chan []model.Obj, 1) + errCh := make(chan error, 1) + go func() { + objects, err := op.List(context.Background(), driver, "/downloads", model.ListArgs{SkipHook: true}) + resultCh <- objects + errCh <- err + }() + select { + case <-driver.listStarted: + case <-time.After(time.Second): + t.Fatal("List() did not start") + } + + driver.mu.Lock() + driver.listings["/downloads"] = []model.Obj{ + &model.Object{Path: "/downloads/new.txt", Name: "new.txt"}, + } + driver.mu.Unlock() + invalidateDestinationCache(driver, "/downloads") + close(driver.listRelease) + + select { + case err := <-errCh: + if err != nil { + t.Fatalf("initial List() error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("initial List() did not finish") + } + <-resultCh + + objects, err := op.List(context.Background(), driver, "/downloads", model.ListArgs{SkipHook: true}) + if err != nil { + t.Fatalf("List() after invalidation error = %v", err) + } + if len(objects) != 1 || objects[0].GetName() != "new.txt" { + t.Fatalf("List() after invalidation = %#v, want new.txt", objects) + } + + driver.mu.Lock() + calls := driver.listCalls["/downloads"] + driver.mu.Unlock() + if calls != 2 { + t.Fatalf("List() call count = %d, want 2", calls) + } +} + +var _ driver.Driver = (*cacheTestDriver)(nil) +var _ driver.Getter = (*cacheTestDriver)(nil) diff --git a/internal/offline_download/halalcloud_open/halalcloud_open.go b/internal/offline_download/halalcloud_open/halalcloud_open.go new file mode 100644 index 0000000000..0a61cd1926 --- /dev/null +++ b/internal/offline_download/halalcloud_open/halalcloud_open.go @@ -0,0 +1,146 @@ +package halalcloudopen + +import ( + "context" + "errors" + "fmt" + "time" + + halalcloudopendriver "github.com/OpenListTeam/OpenList/v4/drivers/halalcloud_open" + "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/internal/offline_download/tool" + "github.com/OpenListTeam/OpenList/v4/internal/op" +) + +// HalalCloudOpen adapts the storage driver's native offline-task API to the +// generic OpenList download-task lifecycle. +type HalalCloudOpen struct{} + +const halalCloudOfflineCleanupTimeout = 15 * time.Second + +func (*HalalCloudOpen) Name() string { + return "HalalCloudOpen" +} + +func (*HalalCloudOpen) Items() []model.SettingItem { + return nil +} + +func (*HalalCloudOpen) Run(_ *tool.DownloadTask) error { + return errs.NotSupport +} + +func (*HalalCloudOpen) Init() (string, error) { + return "ok", nil +} + +func (*HalalCloudOpen) IsReady() bool { + // Availability is resolved from the destination storage by NamesForPath. + return false +} + +func (h *HalalCloudOpen) AddURL(args *tool.AddUrlArgs) (string, error) { + storage, actualPath, err := storageAndActualPath(args.TempDir, args.StorageMountPath) + if err != nil { + return "", err + } + driver, ok := storage.(*halalcloudopendriver.HalalCloudOpen) + if !ok { + return "", errors.New("HalalCloudOpen offline download only supports HalalCloudOpen destination storage") + } + + if err := op.MakeDir(args.Ctx, storage, actualPath); err != nil { + return "", err + } + // The provider object carries the native destination path expected by the API. + parentDir, err := op.GetUnwrap(args.Ctx, storage, actualPath) + if err != nil { + return "", err + } + + task, err := driver.OfflineDownload(args.Ctx, args.Url, parentDir) + if err != nil { + return "", fmt.Errorf("failed to add HalalCloudOpen offline download task: %w", err) + } + h.invalidateTaskCache(driver) + return task.Identity, nil +} + +func (h *HalalCloudOpen) Remove(task *tool.DownloadTask) error { + storage, actualPath, err := storageAndActualPath(task.TempDir, task.StorageMountPath) + if err != nil { + return err + } + driver, ok := storage.(*halalcloudopendriver.HalalCloudOpen) + if !ok { + return errors.New("HalalCloudOpen offline download only supports HalalCloudOpen destination storage") + } + + // Provider-side writes can leave cached task and directory-tree data stale. + defer invalidateDestinationCache(storage, actualPath) + defer h.invalidateTaskCache(driver) + // Cleanup runs independently after the download task context is canceled, + // while the timeout keeps a stalled provider from holding the worker. + cleanupCtx, cancel := context.WithTimeout(detachedContext(task.Ctx()), halalCloudOfflineCleanupTimeout) + defer cancel() + if err := driver.DeleteOfflineTasks(cleanupCtx, []string{task.GID}, false); err != nil { + return err + } + return nil +} + +func (h *HalalCloudOpen) Status(task *tool.DownloadTask) (*tool.Status, error) { + storage, actualPath, err := storageAndActualPath(task.TempDir, task.StorageMountPath) + if err != nil { + return nil, err + } + driver, ok := storage.(*halalcloudopendriver.HalalCloudOpen) + if !ok { + return nil, errors.New("HalalCloudOpen offline download only supports HalalCloudOpen destination storage") + } + + tasks, err := h.getTasks(task.Ctx(), driver) + if err != nil { + return nil, err + } + for _, providerTask := range tasks { + if providerTask != nil && providerTask.Identity == task.GID { + status := statusFromTask(providerTask) + if status.Completed || status.Err != nil { + // Terminal provider writes invalidate the destination tree. + invalidateDestinationCache(storage, actualPath) + } + return status, nil + } + } + // Refresh provider and destination data before retrying a missing task. + h.invalidateTaskCache(driver) + invalidateDestinationCache(storage, actualPath) + return nil, fmt.Errorf("HalalCloudOpen offline task %s not found", task.GID) +} + +func invalidateDestinationCache(storage driver.Driver, actualPath string) { + op.Cache.DeleteDirectoryTree(storage, actualPath) +} + +func storageAndActualPath(rawPath, mountPath string) (driver.Driver, string, error) { + if mountPath != "" { + return op.GetStorageAndActualPathByMountPath(rawPath, mountPath) + } + return op.GetStorageAndActualPath(rawPath) +} + +func detachedContext(ctx context.Context) context.Context { + if ctx == nil { + return context.Background() + } + return context.WithoutCancel(ctx) +} + +var _ tool.Tool = (*HalalCloudOpen)(nil) + +func init() { + tool.Tools.Add(&HalalCloudOpen{}) +} diff --git a/internal/offline_download/halalcloud_open/util.go b/internal/offline_download/halalcloud_open/util.go new file mode 100644 index 0000000000..f07edd9fa2 --- /dev/null +++ b/internal/offline_download/halalcloud_open/util.go @@ -0,0 +1,145 @@ +package halalcloudopen + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + halalcloudopendriver "github.com/OpenListTeam/OpenList/v4/drivers/halalcloud_open" + "github.com/OpenListTeam/OpenList/v4/internal/offline_download/tool" + "github.com/OpenListTeam/OpenList/v4/pkg/singleflight" + "github.com/OpenListTeam/go-cache" + sdkOffline "github.com/halalcloud/golang-sdk-lite/halalcloud/services/offline" +) + +// HalalCloud offline task status contract: +// - 0: waiting to be added +// - 10: waiting for download +// - 1000: completed +// - negative: failed +// - every other non-negative value: downloading +const ( + offlineStatusWaitingToAdd = 0 + offlineStatusWaitingToDownload = 10 + offlineStatusComplete = 1000 + statusCacheExpiration = 10 * time.Second + offlineTaskListTimeout = 30 * time.Second +) + +// The manager polls every three seconds and may track several tasks from the +// same account. Cache one paginated list briefly so those tasks share a single +// provider request. +var taskCache = cache.NewMemCache(cache.WithShards[[]*sdkOffline.UserTask](16)) +var taskGroup singleflight.Group[[]*sdkOffline.UserTask] + +func taskCacheKey(driver *halalcloudopendriver.HalalCloudOpen) string { + // op.Key intentionally removes balance suffixes; retain the exact mount so + // two balanced backends never share a task list cache. + host := strings.ToLower(strings.TrimSpace(driver.Addition.Host)) + if host == "" { + host = "openapi.2dland.cn" + } + return driver.GetStorage().MountPath + "\x00" + host + "\x00" + driver.Addition.ClientID + "\x00/v6/offline_task/list" +} + +func (h *HalalCloudOpen) getTasks(ctx context.Context, driver *halalcloudopendriver.HalalCloudOpen) ([]*sdkOffline.UserTask, error) { + if ctx == nil { + ctx = context.Background() + } + if err := ctx.Err(); err != nil { + return nil, err + } + key := taskCacheKey(driver) + if tasks, ok := taskCache.Get(key); ok { + return tasks, nil + } + + resultCh := taskGroup.DoChan(key, func() ([]*sdkOffline.UserTask, error) { + // Keep the shared fetch independent from one canceled waiter, but bound + // it so a stalled provider cannot remain in flight forever. + requestCtx, cancel := context.WithTimeout(context.Background(), offlineTaskListTimeout) + defer cancel() + tasks, err := driver.OfflineList(requestCtx) + if err != nil { + return nil, err + } + taskCache.Set(key, tasks, cache.WithEx[[]*sdkOffline.UserTask](statusCacheExpiration)) + return tasks, nil + }) + select { + case result := <-resultCh: + return result.Val, result.Err + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func (*HalalCloudOpen) invalidateTaskCache(driver *halalcloudopendriver.HalalCloudOpen) { + taskCache.Del(taskCacheKey(driver)) +} + +// statusFromTask translates the provider's numeric contract into the generic +// status shape consumed by OpenList's download-task manager. +func statusFromTask(task *sdkOffline.UserTask) *tool.Status { + // Some List responses omit Size while still returning BytesTotal. + totalBytes := task.Size + if totalBytes <= 0 { + totalBytes = task.BytesTotal + } + + progress := float64(task.Progress) + if task.Status == offlineStatusComplete { + progress = 100 + } else { + if progress < 0 { + progress = 0 + } + if progress > 100 { + progress = 100 + } + } + + status := &tool.Status{ + TotalBytes: totalBytes, + Progress: progress, + Completed: task.Status == offlineStatusComplete, + Status: taskStatusText(task, progress), + } + if task.Status < 0 { + status.Err = taskStatusError(task) + } + return status +} + +func taskStatusText(task *sdkOffline.UserTask, normalizedProgress float64) string { + message := strings.TrimSpace(task.Message) + switch { + case task.Status == offlineStatusComplete: + return "completed" + case task.Status < 0: + if message != "" { + return message + } + return fmt.Sprintf("error (status %d)", task.Status) + case task.Status == offlineStatusWaitingToAdd: + return "waiting to be added" + case task.Status == offlineStatusWaitingToDownload: + return "waiting for download" + default: + // All other non-negative statuses represent downloading. + return fmt.Sprintf("downloading (%.0f%%)", normalizedProgress) + } +} + +func taskStatusError(task *sdkOffline.UserTask) error { + message := strings.TrimSpace(task.Message) + if message != "" { + return errors.New(message) + } + if task.Code != 0 { + return fmt.Errorf("HalalCloudOpen offline task failed with status %d and code %d", task.Status, task.Code) + } + return fmt.Errorf("HalalCloudOpen offline task failed with status %d", task.Status) +} diff --git a/internal/offline_download/halalcloud_open/util_test.go b/internal/offline_download/halalcloud_open/util_test.go new file mode 100644 index 0000000000..19d3a57178 --- /dev/null +++ b/internal/offline_download/halalcloud_open/util_test.go @@ -0,0 +1,156 @@ +package halalcloudopen + +import ( + "strings" + "testing" + + halalcloudopendriver "github.com/OpenListTeam/OpenList/v4/drivers/halalcloud_open" + "github.com/OpenListTeam/OpenList/v4/internal/model" + sdkOffline "github.com/halalcloud/golang-sdk-lite/halalcloud/services/offline" +) + +func TestTaskCacheKeyDistinguishesBalanceMounts(t *testing.T) { + first := &halalcloudopendriver.HalalCloudOpen{ + Storage: model.Storage{MountPath: "/downloads"}, + } + balanced := &halalcloudopendriver.HalalCloudOpen{ + Storage: model.Storage{MountPath: "/downloads.balance"}, + } + + if taskCacheKey(first) == taskCacheKey(balanced) { + t.Fatal("task cache key collapsed distinct balance mounts") + } +} + +func TestTaskCacheKeyIncludesEndpointAndCredentials(t *testing.T) { + base := &halalcloudopendriver.HalalCloudOpen{ + Storage: model.Storage{MountPath: "/downloads"}, + Addition: halalcloudopendriver.Addition{ClientID: "client-a", Host: "api.example"}, + } + differentHost := &halalcloudopendriver.HalalCloudOpen{ + Storage: model.Storage{MountPath: "/downloads"}, + Addition: halalcloudopendriver.Addition{ClientID: "client-a", Host: "other.example"}, + } + differentClient := &halalcloudopendriver.HalalCloudOpen{ + Storage: model.Storage{MountPath: "/downloads"}, + Addition: halalcloudopendriver.Addition{ClientID: "client-b", Host: "api.example"}, + } + + if taskCacheKey(base) == taskCacheKey(differentHost) { + t.Fatal("task cache key collapsed distinct API hosts") + } + if taskCacheKey(base) == taskCacheKey(differentClient) { + t.Fatal("task cache key collapsed distinct credentials") + } +} + +func TestStatusFromTask(t *testing.T) { + tests := []struct { + name string + task *sdkOffline.UserTask + wantProgress float64 + wantCompleted bool + wantError bool + wantStatus string + wantTotal int64 + }{ + { + name: "waiting to add", + task: &sdkOffline.UserTask{Status: 0, Size: 10}, + wantStatus: "waiting to be added", + wantTotal: 10, + }, + { + name: "waiting to download", + task: &sdkOffline.UserTask{Status: 10, Size: 10}, + wantStatus: "waiting for download", + wantTotal: 10, + }, + { + name: "downloading status 20", + task: &sdkOffline.UserTask{Status: 20, Progress: 5}, + wantProgress: 5, + wantStatus: "downloading (5%)", + }, + { + name: "downloading status 100", + task: &sdkOffline.UserTask{Status: 100, Progress: 42, Size: 100}, + wantProgress: 42, + wantStatus: "downloading (42%)", + wantTotal: 100, + }, + { + name: "downloading status 200", + task: &sdkOffline.UserTask{Status: 200, Progress: 50, Message: "temporary failure"}, + wantProgress: 50, + wantStatus: "downloading (50%)", + }, + { + name: "downloading status 710", + task: &sdkOffline.UserTask{Status: 710, Progress: 75}, + wantProgress: 75, + wantStatus: "downloading (75%)", + }, + { + name: "completed", + task: &sdkOffline.UserTask{Status: 1000, Progress: 0, BytesTotal: 200}, + wantProgress: 100, + wantCompleted: true, + wantStatus: "completed", + wantTotal: 200, + }, + { + name: "negative error", + task: &sdkOffline.UserTask{Status: -1, Code: 7, Message: "provider rejected task"}, + wantError: true, + wantStatus: "provider rejected task", + }, + { + name: "other non-negative status is downloading", + task: &sdkOffline.UserTask{Status: 1001, Code: 8}, + wantStatus: "downloading (0%)", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := statusFromTask(tt.task) + if got.Progress != tt.wantProgress { + t.Errorf("Progress = %v, want %v", got.Progress, tt.wantProgress) + } + if got.Completed != tt.wantCompleted { + t.Errorf("Completed = %v, want %v", got.Completed, tt.wantCompleted) + } + if (got.Err != nil) != tt.wantError { + t.Errorf("Err = %v, wantError %v", got.Err, tt.wantError) + } + if got.Status != tt.wantStatus { + t.Errorf("Status = %q, want %q", got.Status, tt.wantStatus) + } + if got.TotalBytes != tt.wantTotal { + t.Errorf("TotalBytes = %d, want %d", got.TotalBytes, tt.wantTotal) + } + if tt.wantError && tt.task.Code != 0 && !strings.Contains(got.Err.Error(), tt.task.Message) && !strings.Contains(got.Err.Error(), "code") { + t.Errorf("Err = %q, want provider message or code", got.Err) + } + }) + } +} + +func TestStatusFromTaskClampsProgress(t *testing.T) { + negative := statusFromTask(&sdkOffline.UserTask{Status: 100, Progress: -1}) + if negative.Progress != 0 { + t.Errorf("negative Progress = %v, want 0", negative.Progress) + } + if negative.Status != "downloading (0%)" { + t.Errorf("negative status text = %q, want normalized progress", negative.Status) + } + + oversized := statusFromTask(&sdkOffline.UserTask{Status: 100, Progress: 101}) + if oversized.Progress != 100 { + t.Errorf("oversized Progress = %v, want 100", oversized.Progress) + } + if oversized.Status != "downloading (100%)" { + t.Errorf("oversized status text = %q, want normalized progress", oversized.Status) + } +} diff --git a/internal/offline_download/tool/add.go b/internal/offline_download/tool/add.go index 12160dcfee..e7a0d5baaf 100644 --- a/internal/offline_download/tool/add.go +++ b/internal/offline_download/tool/add.go @@ -13,6 +13,7 @@ import ( _123 "github.com/OpenListTeam/OpenList/v4/drivers/123" _123_open "github.com/OpenListTeam/OpenList/v4/drivers/123_open" "github.com/OpenListTeam/OpenList/v4/drivers/guangyapan" + halalcloudopen "github.com/OpenListTeam/OpenList/v4/drivers/halalcloud_open" "github.com/OpenListTeam/OpenList/v4/drivers/pikpak" "github.com/OpenListTeam/OpenList/v4/drivers/thunder" "github.com/OpenListTeam/OpenList/v4/drivers/thunder_browser" @@ -115,6 +116,7 @@ func AddURL(ctx context.Context, args *AddURLArgs) (task.TaskExtensionInfo, erro uid := uuid.NewString() tempDir := filepath.Join(conf.Conf.TempDir, args.Tool, uid) deletePolicy := args.DeletePolicy + storageMountPath := "" // 如果当前 storage 是对应网盘，则直接下载到目标路径，无需转存 switch args.Tool { @@ -178,6 +180,13 @@ func AddURL(ctx context.Context, args *AddURLArgs) (task.TaskExtensionInfo, erro } tempDir = filepath.Join(tempBase, uid) } + case "HalalCloudOpen": + // Native tasks write directly to the selected HalalCloud directory. + if _, ok := storage.(*halalcloudopen.HalalCloudOpen); !ok { + return nil, errors.New("HalalCloudOpen offline download only supports HalalCloudOpen destination storage") + } + tempDir = args.DstDirPath + storageMountPath = storage.GetStorage().MountPath } taskCreator, _ := ctx.Value(conf.UserKey).(*model.User) // taskCreator is nil when convert failed @@ -186,12 +195,13 @@ func AddURL(ctx context.Context, args *AddURLArgs) (task.TaskExtensionInfo, erro Creator: taskCreator, ApiUrl: common.GetApiUrl(ctx), }, - Url: args.URL, - DstDirPath: args.DstDirPath, - TempDir: tempDir, - DeletePolicy: deletePolicy, - Toolname: args.Tool, - tool: tool, + Url: args.URL, + DstDirPath: args.DstDirPath, + TempDir: tempDir, + StorageMountPath: storageMountPath, + DeletePolicy: deletePolicy, + Toolname: args.Tool, + tool: tool, } DownloadTaskManager.Add(t) return t, nil @@ -243,6 +253,8 @@ func toolNameForStorage(storage driver.Driver) string { return "123 Open" case *guangyapan.GuangYaPan: return "GuangYaPan" + case *halalcloudopen.HalalCloudOpen: + return "HalalCloudOpen" case *pikpak.PikPak: return "PikPak" case *thunder.Thunder: diff --git a/internal/offline_download/tool/add_test.go b/internal/offline_download/tool/add_test.go index c2782130fd..50752c308a 100644 --- a/internal/offline_download/tool/add_test.go +++ b/internal/offline_download/tool/add_test.go @@ -8,6 +8,7 @@ import ( _123 "github.com/OpenListTeam/OpenList/v4/drivers/123" _123_open "github.com/OpenListTeam/OpenList/v4/drivers/123_open" "github.com/OpenListTeam/OpenList/v4/drivers/guangyapan" + halalcloudopen "github.com/OpenListTeam/OpenList/v4/drivers/halalcloud_open" "github.com/OpenListTeam/OpenList/v4/drivers/pikpak" "github.com/OpenListTeam/OpenList/v4/drivers/thunder" "github.com/OpenListTeam/OpenList/v4/drivers/thunder_browser" @@ -69,6 +70,7 @@ func TestToolNameForStorage(t *testing.T) { {name: "123Pan", storage: &_123.Pan123{}, want: "123Pan"}, {name: "123 Open", storage: &_123_open.Open123{}, want: "123 Open"}, {name: "GuangYaPan", storage: &guangyapan.GuangYaPan{}, want: "GuangYaPan"}, + {name: "HalalCloudOpen", storage: &halalcloudopen.HalalCloudOpen{}, want: "HalalCloudOpen"}, {name: "PikPak", storage: &pikpak.PikPak{}, want: "PikPak"}, {name: "Thunder", storage: &thunder.Thunder{}, want: "Thunder"}, {name: "ThunderX", storage: &thunderx.ThunderX{}, want: "ThunderX"}, diff --git a/internal/offline_download/tool/base.go b/internal/offline_download/tool/base.go index 823bac5266..b72b8796da 100644 --- a/internal/offline_download/tool/base.go +++ b/internal/offline_download/tool/base.go @@ -7,11 +7,12 @@ import ( ) type AddUrlArgs struct { - Url string - UID string - TempDir string - Signal chan int - Ctx context.Context + Url string + UID string + TempDir string + StorageMountPath string + Signal chan int + Ctx context.Context } type Status struct { diff --git a/internal/offline_download/tool/download.go b/internal/offline_download/tool/download.go index ca87768397..7f57b6ece9 100644 --- a/internal/offline_download/tool/download.go +++ b/internal/offline_download/tool/download.go @@ -20,9 +20,12 @@ import ( type DownloadTask struct { task.TaskExtension - Url string `json:"url"` - DstDirPath string `json:"dst_dir_path"` - TempDir string `json:"temp_dir"` + Url string `json:"url"` + DstDirPath string `json:"dst_dir_path"` + TempDir string `json:"temp_dir"` + // StorageMountPath pins destination-bound native tools to the storage + // selected when the task was created, including balance mounts. + StorageMountPath string `json:"storage_mount_path,omitempty"` DeletePolicy DeletePolicy `json:"delete_policy"` Toolname string `json:"toolname"` Status string `json:"-"` @@ -54,11 +57,12 @@ func (t *DownloadTask) Run() error { t.Signal = nil }() gid, err := t.tool.AddURL(&AddUrlArgs{ - Ctx: t.Ctx(), - Url: t.Url, - UID: t.ID, - TempDir: t.TempDir, - Signal: t.Signal, + Ctx: t.Ctx(), + Url: t.Url, + UID: t.ID, + TempDir: t.TempDir, + StorageMountPath: t.StorageMountPath, + Signal: t.Signal, }) if err != nil { return err @@ -101,6 +105,10 @@ outer: if t.tool.Name() == "GuangYaPan" { return nil } + if t.tool.Name() == "HalalCloudOpen" { + // The provider task completed directly in DstDirPath. + return nil + } if t.tool.Name() == "115 Cloud" { // hack for 115 <-time.After(time.Second * 1) @@ -179,7 +187,7 @@ func (t *DownloadTask) Update() (bool, error) { func (t *DownloadTask) Transfer() error { toolName := t.tool.Name() - if toolName == "115 Cloud" || toolName == "115 Open" || toolName == "123 Open" || toolName == "123Pan" || toolName == "PikPak" || toolName == "Thunder" || toolName == "ThunderX" || toolName == "ThunderBrowser" || toolName == "GuangYaPan" { + if toolName == "115 Cloud" || toolName == "115 Open" || toolName == "123 Open" || toolName == "123Pan" || toolName == "PikPak" || toolName == "Thunder" || toolName == "ThunderX" || toolName == "ThunderBrowser" || toolName == "GuangYaPan" || toolName == "HalalCloudOpen" { // 如果不是直接下载到目标路径，则进行转存 if t.TempDir != t.DstDirPath { return transferObj(t.Ctx(), t.TempDir, t.DstDirPath, t.DeletePolicy) diff --git a/internal/op/cache.go b/internal/op/cache.go index d8d32a74b6..0d36f80367 100644 --- a/internal/op/cache.go +++ b/internal/op/cache.go @@ -17,6 +17,8 @@ type CacheManager struct { userCache *cache.KeyedCache[*model.User] // Cache for user data settingCache *cache.KeyedCache[any] // Cache for settings detailCache *cache.KeyedCache[*model.StorageDetails] // Cache for storage details + dirVersionMu sync.RWMutex + dirVersions map[string]uint64 } func NewCacheManager() *CacheManager { @@ -26,6 +28,7 @@ func NewCacheManager() *CacheManager { userCache: cache.NewKeyedCache[*model.User](time.Hour), settingCache: cache.NewKeyedCache[any](time.Hour), detailCache: cache.NewKeyedCache[*model.StorageDetails](time.Minute * 30), + dirVersions: make(map[string]uint64), } } @@ -36,23 +39,44 @@ func Key(storage driver.Driver, path string) string { return utils.GetFullPath(storage.GetStorage().MountPath, path) } -// recursively delete directory and its children from dirCache +// DeleteDirectoryTree removes a directory and all cached descendants. func (cm *CacheManager) DeleteDirectoryTree(storage driver.Driver, dirPath string) { if storage.Config().NoCache { return } + cm.dirVersionMu.Lock() + defer cm.dirVersionMu.Unlock() + cm.dirVersions[directoryVersionKey(storage)]++ cm.deleteDirectoryTree(Key(storage, dirPath)) } + func (cm *CacheManager) deleteDirectoryTree(key string) { - if dirCache, exists := cm.dirCache.Pop(key); exists { - for _, obj := range dirCache.objs { - if obj.IsDir() { - cm.deleteDirectoryTree(stdpath.Join(key, obj.GetName())) - } else { - cm.linkCache.DeleteKey(stdpath.Join(key, obj.GetName())) - } - } + cm.dirCache.DeletePrefix(key) + cm.linkCache.DeleteKeyPrefix(key) +} + +func (cm *CacheManager) directoryVersion(storage driver.Driver) uint64 { + cm.dirVersionMu.RLock() + defer cm.dirVersionMu.RUnlock() + return cm.dirVersions[directoryVersionKey(storage)] +} + +func (cm *CacheManager) updateDirectoryCache(storage driver.Driver, key string, version uint64, value *directoryCache, ttl time.Duration) bool { + cm.dirVersionMu.RLock() + defer cm.dirVersionMu.RUnlock() + if cm.dirVersions[directoryVersionKey(storage)] != version { + return false + } + if value == nil { + cm.deleteDirectoryTree(key) + } else { + cm.dirCache.SetWithTTL(key, value, ttl) } + return true +} + +func directoryVersionKey(storage driver.Driver) string { + return utils.GetActualMountPath(storage.GetStorage().MountPath) } // remove directory from dirCache @@ -60,6 +84,9 @@ func (cm *CacheManager) DeleteDirectory(storage driver.Driver, dirPath string) { if storage.Config().NoCache { return } + cm.dirVersionMu.Lock() + cm.dirVersions[directoryVersionKey(storage)]++ + cm.dirVersionMu.Unlock() cm.dirCache.Delete(Key(storage, dirPath)) } @@ -77,7 +104,7 @@ func (cm *CacheManager) removeDirectoryObject(storage driver.Driver, dirPath str } if cache, exist := cm.dirCache.Get(key); exist { if obj.IsDir() { - cm.deleteDirectoryTree(stdpath.Join(key, obj.GetName())) + cm.DeleteDirectoryTree(storage, stdpath.Join(dirPath, obj.GetName())) } cache.RemoveObject(obj.GetName()) } @@ -146,11 +173,14 @@ func (cm *CacheManager) InvalidateStorageDetails(storage driver.Driver) { // clears all caches func (cm *CacheManager) ClearAll() { + cm.dirVersionMu.Lock() + defer cm.dirVersionMu.Unlock() cm.dirCache.Clear() cm.linkCache.Clear() cm.userCache.Clear() cm.settingCache.Clear() cm.detailCache.Clear() + cm.dirVersions = make(map[string]uint64) } type directoryCache struct { diff --git a/internal/op/fs.go b/internal/op/fs.go index f82a3ca8f8..77f7ed45fb 100644 --- a/internal/op/fs.go +++ b/internal/op/fs.go @@ -49,6 +49,7 @@ func list(ctx context.Context, storage driver.Driver, path string, args model.Li } objs, err, _ := listG.Do(key, func() ([]model.Obj, error) { + cacheVersion := Cache.directoryVersion(storage) dir, err := GetUnwrap(ctx, storage, path) if err != nil { return nil, errors.WithMessage(err, "failed get dir") @@ -105,10 +106,12 @@ func list(ctx context.Context, storage driver.Driver, path string, args model.Li } duration := time.Minute * time.Duration(ttl) - Cache.dirCache.SetWithTTL(key, newDirectoryCache(files), duration) + if !Cache.updateDirectoryCache(storage, key, cacheVersion, newDirectoryCache(files), duration) { + log.Debugf("skip stale cache update: %s", key) + } } else { log.Debugf("del cache: %s", key) - Cache.deleteDirectoryTree(key) + Cache.updateDirectoryCache(storage, key, cacheVersion, nil, 0) } } return files, nil @@ -420,7 +423,7 @@ func Move(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string if !storage.Config().NoCache { if cache, exist := Cache.dirCache.Get(srcKey); exist { if srcRawObj.IsDir() { - Cache.deleteDirectoryTree(stdpath.Join(srcKey, srcRawObj.GetName())) + Cache.DeleteDirectoryTree(storage, stdpath.Join(srcDirPath, srcRawObj.GetName())) } cache.RemoveObject(srcRawObj.GetName()) } @@ -484,7 +487,7 @@ func Rename(ctx context.Context, storage driver.Driver, srcPath, dstName string) if !storage.Config().NoCache { if cache, exist := Cache.dirCache.Get(dirKey); exist { if srcRawObj.IsDir() { - Cache.deleteDirectoryTree(stdpath.Join(dirKey, oldName)) + Cache.DeleteDirectoryTree(storage, stdpath.Join(stdpath.Dir(srcPath), oldName)) } if newObj == nil { newObj = &model.ObjWrapMask{Obj: &model.ObjWrapName{Name: dstName, Obj: srcObj}, Mask: model.Temp} diff --git a/internal/op/path.go b/internal/op/path.go index 9157e2f221..9dc5a674ec 100644 --- a/internal/op/path.go +++ b/internal/op/path.go @@ -30,6 +30,26 @@ func GetStorageAndActualPath(rawPath string) (storage driver.Driver, actualPath return } +// GetStorageAndActualPathByMountPath resolves rawPath against a specific +// storage mount. This is needed for balance mounts, where resolving the same +// virtual path repeatedly may intentionally select different backends. +func GetStorageAndActualPathByMountPath(rawPath, mountPath string) (storage driver.Driver, actualPath string, err error) { + rawPath = utils.FixAndCleanPath(rawPath) + mountPath = utils.FixAndCleanPath(mountPath) + storage, err = GetStorageByMountPath(mountPath) + if err != nil { + return nil, "", err + } + + actualMountPath := utils.GetActualMountPath(storage.GetStorage().MountPath) + if !utils.IsSubPath(actualMountPath, rawPath) { + return nil, "", errs.NewErr(errs.StorageNotFound, "rawPath %q is outside storage mount %q", rawPath, mountPath) + } + actualPath = strings.TrimPrefix(rawPath, actualMountPath) + if actualPath == "" { + actualPath = "/" + } + return storage, utils.FixAndCleanPath(actualPath), nil // GetStorageVirtualMountPath returns the deterministic virtual mount path // without advancing the balanced-storage counter. func GetStorageVirtualMountPath(rawPath string) (string, error) { diff --git a/internal/op/path_test.go b/internal/op/path_test.go new file mode 100644 index 0000000000..2d3b6a1040 --- /dev/null +++ b/internal/op/path_test.go @@ -0,0 +1,67 @@ +package op + +import ( + "context" + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/model" +) + +type pathResolutionDriver struct { + model.Storage +} + +func (d *pathResolutionDriver) Config() driver.Config { + return driver.Config{} +} + +func (d *pathResolutionDriver) GetAddition() driver.Additional { + return nil +} + +func (*pathResolutionDriver) Init(context.Context) error { + return nil +} + +func (*pathResolutionDriver) Drop(context.Context) error { + return nil +} + +func (*pathResolutionDriver) List(context.Context, model.Obj, model.ListArgs) ([]model.Obj, error) { + return nil, nil +} + +func (*pathResolutionDriver) Link(context.Context, model.Obj, model.LinkArgs) (*model.Link, error) { + return nil, nil +} + +func TestGetStorageAndActualPathByMountPathPinsBalanceMount(t *testing.T) { + mountPath := "/path-resolution-test" + balancedMountPath := mountPath + ".balance" + storage := &pathResolutionDriver{Storage: model.Storage{MountPath: balancedMountPath}} + storagesMap.Store(balancedMountPath, storage) + defer storagesMap.Delete(balancedMountPath) + + gotStorage, gotPath, err := GetStorageAndActualPathByMountPath(mountPath+"/downloads", balancedMountPath) + if err != nil { + t.Fatalf("GetStorageAndActualPathByMountPath() error = %v", err) + } + if gotStorage != storage { + t.Fatalf("resolved storage = %p, want pinned storage %p", gotStorage, storage) + } + if gotPath != "/downloads" { + t.Fatalf("resolved actual path = %q, want %q", gotPath, "/downloads") + } +} + +func TestGetStorageAndActualPathByMountPathRejectsOutsidePath(t *testing.T) { + mountPath := "/path-resolution-outside-test" + storage := &pathResolutionDriver{Storage: model.Storage{MountPath: mountPath}} + storagesMap.Store(mountPath, storage) + defer storagesMap.Delete(mountPath) + + if _, _, err := GetStorageAndActualPathByMountPath(mountPath+"-other/file", mountPath); err == nil { + t.Fatal("GetStorageAndActualPathByMountPath() accepted a path outside the pinned mount") + } +}