From a992395d766c3cc5aff64e50f4b1ade3e724e47f Mon Sep 17 00:00:00 2001 From: totza2010 Date: Sat, 29 Aug 2026 17:20:57 +0700 Subject: [PATCH] feat(drivers): add Teldrive V2 driver TelDrive v2 rewrites its HTTP API: every path moves under /api/v1, the listing is cursor-paginated instead of page-numbered, objects are addressed by UUID rather than path, uploads go through durable server-side sessions, and every mutating endpoint requires an Idempotency-Key. None of it is reachable from the existing Teldrive driver, so this adds a separate one and leaves the v1 driver untouched. Notable differences from the v1 driver: - Copy is a single request; the server copies a whole subtree transactionally, so no client-side recursion is needed. - Uploads create a session, PUT each part, then complete. Parts are idempotent on (uploadId, partNo), and an interrupted upload resumes by reusing its session and skipping parts the server already stored. - The client mtime is preserved, which the v1 driver could not do. - Files carry a BLAKE3 tree hash. It is registered as blake3_tree in pkg/utils/hash and reported on every object; when hash_enabled is set, each part is also sent with a checksum for the server to verify. Part uploads use a dedicated resty client: the shared one caps requests at 30s, which a part cannot meet because the server only responds once it has relayed the part to Telegram, and it would buffer the whole part in memory to compute Content-Length. Requires a TelDrive v2 server at e3142b5 or newer, where the move endpoint began accepting a conflictPolicy other than "fail". Co-Authored-By: Claude Opus 5 --- drivers/all.go | 1 + drivers/teldrive_v2/driver.go | 434 +++++++++++++++++++++++++++++ drivers/teldrive_v2/driver_test.go | 194 +++++++++++++ drivers/teldrive_v2/meta.go | 50 ++++ drivers/teldrive_v2/types.go | 119 ++++++++ drivers/teldrive_v2/upload.go | 312 +++++++++++++++++++++ drivers/teldrive_v2/util.go | 218 +++++++++++++++ go.mod | 2 +- pkg/utils/hash/tdhash.go | 78 ++++++ pkg/utils/hash/tdhash_test.go | 93 +++++++ 10 files changed, 1500 insertions(+), 1 deletion(-) create mode 100644 drivers/teldrive_v2/driver.go create mode 100644 drivers/teldrive_v2/driver_test.go create mode 100644 drivers/teldrive_v2/meta.go create mode 100644 drivers/teldrive_v2/types.go create mode 100644 drivers/teldrive_v2/upload.go create mode 100644 drivers/teldrive_v2/util.go create mode 100644 pkg/utils/hash/tdhash.go create mode 100644 pkg/utils/hash/tdhash_test.go diff --git a/drivers/all.go b/drivers/all.go index 601c7bfc5c..77815d7ebc 100644 --- a/drivers/all.go +++ b/drivers/all.go @@ -75,6 +75,7 @@ import ( _ "github.com/OpenListTeam/OpenList/v4/drivers/strm" _ "github.com/OpenListTeam/OpenList/v4/drivers/teambition" _ "github.com/OpenListTeam/OpenList/v4/drivers/teldrive" + _ "github.com/OpenListTeam/OpenList/v4/drivers/teldrive_v2" _ "github.com/OpenListTeam/OpenList/v4/drivers/terabox" _ "github.com/OpenListTeam/OpenList/v4/drivers/thunder" _ "github.com/OpenListTeam/OpenList/v4/drivers/thunder_browser" diff --git a/drivers/teldrive_v2/driver.go b/drivers/teldrive_v2/driver.go new file mode 100644 index 0000000000..bc00570874 --- /dev/null +++ b/drivers/teldrive_v2/driver.go @@ -0,0 +1,434 @@ +package teldrive_v2 + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "path" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/OpenListTeam/OpenList/v4/drivers/base" + "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/op" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + hash_extend "github.com/OpenListTeam/OpenList/v4/pkg/utils/hash" + "github.com/go-resty/resty/v2" +) + +// listPageSize is the maximum the v2 API accepts for `limit`. +const listPageSize = 200 + +// treeHashBlockMiB is teldrive's fixed tree-hash block size (internal/treehash). +const treeHashBlockMiB = 16 + +// maxChunkMiB is the largest part Telegram will take as a single message. +const maxChunkMiB = 2000 + +// defaultChunkMiB matches the part size the server picks for itself. +const defaultChunkMiB = 512 + +// hashAlgorithm is the only value teldrive v2 reports in FileEntry.hash. +const hashAlgorithm = "blake3" + +// defaultShareExpireHours is used when the configured value cannot be read. +const defaultShareExpireHours = 1 + +// shareRenewMargin keeps a share from being handed out right before it lapses. +const shareRenewMargin = 5 * time.Minute + +// cachedShare holds the one-time token the API returns when a share is created. +type cachedShare struct { + token string + expiresAt time.Time +} + +type TeldriveV2 struct { + model.Storage + Addition + + // rootID is the resolved UUID of RootFolderPath; "" means the drive root. + rootID string + // pathCache memoises absolute drive path -> folder UUID. + pathCache *sync.Map + // shareCache memoises file UUID -> public share token. + shareCache *sync.Map + // lastShareSweep is the unix-nano time shareCache was last swept, so the + // sweep cost is amortised instead of paid on every mint. + lastShareSweep atomic.Int64 + // uploadClient carries part bodies: no client deadline, no implicit retry. + uploadClient *resty.Client + // shareExpire is Addition.ShareExpire parsed into a duration. + shareExpire time.Duration +} + +func (d *TeldriveV2) Config() driver.Config { return config } + +func (d *TeldriveV2) GetAddition() driver.Additional { return &d.Addition } + +func (d *TeldriveV2) Init(ctx context.Context) error { + d.Address = strings.TrimSuffix(strings.TrimSpace(d.Address), "/") + if d.Address == "" { + return fmt.Errorf("url is required") + } + if strings.TrimSpace(d.ApiKey) == "" { + return fmt.Errorf("api_key is required") + } + if d.UploadConcurrency <= 0 { + d.UploadConcurrency = 4 + } + if d.UploadConcurrency > 16 { + d.UploadConcurrency = 16 + } + d.ChunkSize = normalizeChunkSize(d.ChunkSize) + d.shareExpire = parseShareExpire(d.ShareExpire) + + d.pathCache = &sync.Map{} + d.shareCache = &sync.Map{} + // Inherits the shared proxy/TLS/UA setup, then drops the two settings that + // are wrong for a multi-hundred-MiB body. + // + // The pre-request hook is what keeps a part streaming. resty's own + // SetContentLength drains an io.Reader body into memory to measure it + // (middleware.go handleRequestBody), which would buffer a whole part and + // make progress meaningless - the bytes would all be "sent" long before + // they reach Telegram. Instead the caller sets a Content-Length header and + // this copies it onto the raw request, so net/http streams the body with a + // known length. + d.uploadClient = base.NewRestyClient(). + SetTimeout(0). + SetRetryCount(0). + SetPreRequestHook(func(_ *resty.Client, req *http.Request) error { + if req.ContentLength > 0 || req.Body == nil { + return nil + } + v := req.Header.Get("Content-Length") + if v == "" { + return nil + } + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + return fmt.Errorf("invalid Content-Length %q: %w", v, err) + } + req.ContentLength = n + return nil + }) + + rootID, err := d.resolveFolderID(ctx, d.RootFolderPath) + if err != nil { + return err + } + d.rootID = rootID + + op.MustSaveDriverStorage(d) + return nil +} + +func (d *TeldriveV2) Drop(ctx context.Context) error { + d.pathCache = &sync.Map{} + d.shareCache = &sync.Map{} + return nil +} + +// GetRoot hands OpenList a root object that already carries the resolved UUID, +// so every write operation below can pass a parentId without another lookup. +func (d *TeldriveV2) GetRoot(ctx context.Context) (model.Obj, error) { + return &model.Object{ + ID: d.rootID, + Path: normalizePath(d.RootFolderPath), + Name: "root", + IsFolder: true, + }, nil +} + +func (d *TeldriveV2) List(ctx context.Context, dir model.Obj, args model.ListArgs) ([]model.Obj, error) { + parentID, err := d.dirID(ctx, dir) + if err != nil { + return nil, err + } + dirPath := normalizePath(dir.GetPath()) + + // parentID == "" is the drive root, which needs no query parameter at all. + entries, err := d.listAll(ctx, parentID, "") + if err != nil { + return nil, err + } + + return utils.SliceConvert(entries, func(src FileEntry) (model.Obj, error) { + return d.toObj(&src, path.Join(dirPath, src.Name)), nil + }) +} + +func (d *TeldriveV2) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*model.Link, error) { + // One endpoint serves both streaming and downloading: it answers Range + // requests with 206, and its Content-Disposition already carries the stored + // filename either way. The trailing segment is for clients that take the + // name from the URL instead - players and command-line tools - which would + // otherwise see the UUID. The segment-less form remains as downloadFileLegacy. + name := url.PathEscape(file.GetName()) + if d.UseShareLink { + token, expiresAt, err := d.shareToken(ctx, file.GetID()) + if err != nil { + return nil, err + } + link := &model.Link{ + URL: fmt.Sprintf("%s%s/public/shares/%s/files/%s/content/%s", + d.Address, apiPrefix, url.PathEscape(token), url.PathEscape(file.GetID()), name), + } + // Without this OpenList caches the link against its SyncClosers, which + // never expire for a plain URL, so it would keep handing out the token + // long after the share behind it lapsed. Expiring the cache a little + // early sends the next request back through here for a fresh one. + if ttl := time.Until(expiresAt) - shareRenewMargin; ttl > 0 { + link.Expiration = &ttl + } + return link, nil + } + return &model.Link{ + URL: fmt.Sprintf("%s%s/files/%s/content/%s", d.Address, apiPrefix, url.PathEscape(file.GetID()), name), + Header: http.Header{ + "X-Api-Key": {d.ApiKey}, + }, + }, nil +} + +func (d *TeldriveV2) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) (model.Obj, error) { + parentID, err := d.dirID(ctx, parentDir) + if err != nil { + return nil, err + } + body := base.Json{ + "name": dirName, + "conflictPolicy": PolicyFail, + } + if parentID != "" { + body["parentId"] = parentID + } + + newPath := path.Join(normalizePath(parentDir.GetPath()), dirName) + + entry := &FileEntry{} + if err := d.request(ctx, http.MethodPost, "/folders", idempotent(func(req *resty.Request) { + req.SetBody(body) + }), entry); err != nil { + // conflictPolicy=fail returns 409 when the folder is already there. + // Treat that as success and hand back the existing folder. + var e *ErrResp + if errors.As(err, &e) && e.status == http.StatusConflict { + id, resolveErr := d.resolveFolderID(ctx, newPath) + if resolveErr != nil { + return nil, err + } + return &model.Object{ID: id, Path: newPath, Name: dirName, IsFolder: true}, nil + } + return nil, err + } + return d.toObj(entry, newPath), nil +} + +func (d *TeldriveV2) Move(ctx context.Context, srcObj, dstDir model.Obj) (model.Obj, error) { + dstID, err := d.dirID(ctx, dstDir) + if err != nil { + return nil, err + } + body := base.Json{"conflictPolicy": PolicyReplace} + if dstID != "" { + body["parentId"] = dstID + } + + entry := &FileEntry{} + if err := d.request(ctx, http.MethodPost, "/files/{id}/move", idempotent(func(req *resty.Request) { + req.SetPathParam("id", srcObj.GetID()) + req.SetBody(body) + }), entry); err != nil { + return nil, err + } + d.forgetPath(srcObj) + return d.toObj(entry, path.Join(normalizePath(dstDir.GetPath()), entry.Name)), nil +} + +func (d *TeldriveV2) Rename(ctx context.Context, srcObj model.Obj, newName string) (model.Obj, error) { + entry := &FileEntry{} + if err := d.request(ctx, http.MethodPatch, "/files/{id}", func(req *resty.Request) { + req.SetPathParam("id", srcObj.GetID()) + req.SetBody(base.Json{"name": newName}) + }, entry); err != nil { + return nil, err + } + d.forgetPath(srcObj) + parent := path.Dir(normalizePath(srcObj.GetPath())) + return d.toObj(entry, path.Join(parent, entry.Name)), nil +} + +// Copy is a single call in v2: the server copies the whole subtree +// transactionally, so the recursive CopyManager the v1 driver needs is gone. +func (d *TeldriveV2) Copy(ctx context.Context, srcObj, dstDir model.Obj) (model.Obj, error) { + dstID, err := d.dirID(ctx, dstDir) + if err != nil { + return nil, err + } + body := base.Json{ + "name": srcObj.GetName(), + "conflictPolicy": PolicyReplace, + } + if dstID != "" { + body["parentId"] = dstID + } + + entry := &FileEntry{} + if err := d.request(ctx, http.MethodPost, "/files/{id}/copy", idempotent(func(req *resty.Request) { + req.SetPathParam("id", srcObj.GetID()) + req.SetBody(body) + }), entry); err != nil { + return nil, err + } + return d.toObj(entry, path.Join(normalizePath(dstDir.GetPath()), entry.Name)), nil +} + +// Remove moves the object to trash. With HardDelete enabled it is then purged, +// which v2 only permits on an already-trashed object. +func (d *TeldriveV2) Remove(ctx context.Context, obj model.Obj) error { + if err := d.request(ctx, http.MethodDelete, "/files/{id}", func(req *resty.Request) { + req.SetPathParam("id", obj.GetID()) + }, nil); err != nil { + return err + } + d.forgetPath(obj) + d.shareCache.Delete(obj.GetID()) + if !d.HardDelete { + return nil + } + return d.request(ctx, http.MethodDelete, "/files/{id}/purge", func(req *resty.Request) { + req.SetPathParam("id", obj.GetID()) + }, nil) +} + +func (d *TeldriveV2) toObj(entry *FileEntry, objPath string) model.Obj { + isFolder := entry.Kind == KindFolder + size := entry.Size + if isFolder { + size = 0 + } + // The server hashes every part as it streams to Telegram, so a file entry + // always carries its tree hash - it costs us nothing to surface it. + var hashInfo utils.HashInfo + if entry.Hash != nil && strings.EqualFold(entry.Hash.Algorithm, hashAlgorithm) { + hashInfo = utils.NewHashInfo(hash_extend.BLAKE3Tree, entry.Hash.Value) + } + return &model.Object{ + ID: entry.ID, + Path: objPath, + Name: entry.Name, + Size: size, + IsFolder: isFolder, + HashInfo: hashInfo, + // v2 renamed updatedAt -> modTime and round-trips the client mtime, + // so this is the timestamp the file actually had on upload. + Modified: entry.ModTime, + Ctime: entry.CreatedAt, + } +} + +func (d *TeldriveV2) forgetPath(obj model.Obj) { + if obj.IsDir() { + d.pathCache.Delete(normalizePath(obj.GetPath())) + } +} + +// sweepShares drops entries whose share has lapsed. +// +// Every entry expires within shareExpire, so a sweep leaves behind only shares +// that are still usable: the map is bounded by the number of distinct files +// linked within one expiry window rather than by everything ever linked. It +// runs off the mint path - already an HTTP round trip - and no more than once +// per shareRenewMargin, so a busy storage does not pay the walk on every miss. +func (d *TeldriveV2) sweepShares(now time.Time) { + last := d.lastShareSweep.Load() + if now.UnixNano()-last < int64(shareRenewMargin) { + return + } + if !d.lastShareSweep.CompareAndSwap(last, now.UnixNano()) { + // Another goroutine is sweeping; one pass is enough. + return + } + d.shareCache.Range(func(k, v any) bool { + if s, ok := v.(*cachedShare); ok && !now.Before(s.expiresAt.Add(-shareRenewMargin)) { + d.shareCache.Delete(k) + } + return true + }) +} + +// shareToken returns a public token for the file, minting one only when needed. +// +// The token cannot be recovered after the fact: GET /files/{id}/shares lists a +// share's id, expiry and download count but never its token, which the API hands +// back exactly once at creation. So reusing a share means remembering the token +// ourselves - without this cache every Link() call would leave another live +// share behind on the server. +func (d *TeldriveV2) shareToken(ctx context.Context, fileID string) (string, time.Time, error) { + if v, ok := d.shareCache.Load(fileID); ok { + if s := v.(*cachedShare); time.Now().Before(s.expiresAt.Add(-shareRenewMargin)) { + return s.token, s.expiresAt, nil + } + d.shareCache.Delete(fileID) + } + + now := time.Now().UTC() + d.sweepShares(now) + + expires := now.Add(d.shareExpire) + share := &ShareCreated{} + if err := d.request(ctx, http.MethodPost, "/files/{id}/shares", idempotent(func(req *resty.Request) { + req.SetPathParam("id", fileID) + req.SetBody(base.Json{ + "permission": "read", + "expiresAt": expires.Format(time.RFC3339), + }) + }), share); err != nil { + return "", time.Time{}, err + } + if share.Token == "" { + return "", time.Time{}, fmt.Errorf("[TeldriveV2] share created without a token") + } + + if !share.ExpiresAt.IsZero() { + expires = share.ExpiresAt + } + d.shareCache.Store(fileID, &cachedShare{token: share.Token, expiresAt: expires}) + return share.Token, expires, nil +} + +func (d *TeldriveV2) GetArchiveMeta(ctx context.Context, obj model.Obj, args model.ArchiveArgs) (model.ArchiveMeta, error) { + return nil, errs.NotImplement +} + +func (d *TeldriveV2) ListArchive(ctx context.Context, obj model.Obj, args model.ArchiveInnerArgs) ([]model.Obj, error) { + return nil, errs.NotImplement +} + +func (d *TeldriveV2) Extract(ctx context.Context, obj model.Obj, args model.ArchiveInnerArgs) (*model.Link, error) { + return nil, errs.NotImplement +} + +func (d *TeldriveV2) ArchiveDecompress(ctx context.Context, srcObj, dstDir model.Obj, args model.ArchiveDecompressArgs) ([]model.Obj, error) { + return nil, errs.NotImplement +} + +var ( + _ driver.Driver = (*TeldriveV2)(nil) + _ driver.GetRooter = (*TeldriveV2)(nil) + _ driver.MkdirResult = (*TeldriveV2)(nil) + _ driver.MoveResult = (*TeldriveV2)(nil) + _ driver.RenameResult = (*TeldriveV2)(nil) + _ driver.CopyResult = (*TeldriveV2)(nil) + _ driver.Remove = (*TeldriveV2)(nil) + _ driver.PutResult = (*TeldriveV2)(nil) +) diff --git a/drivers/teldrive_v2/driver_test.go b/drivers/teldrive_v2/driver_test.go new file mode 100644 index 0000000000..090062052c --- /dev/null +++ b/drivers/teldrive_v2/driver_test.go @@ -0,0 +1,194 @@ +package teldrive_v2 + +import ( + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/go-resty/resty/v2" +) + +func TestNormalizePath(t *testing.T) { + for _, tc := range []struct{ in, want string }{ + {"", "/"}, + {"/", "/"}, + {" ", "/"}, + {"a", "/a"}, + {"/a/b", "/a/b"}, + {"/a/b/", "/a/b"}, + {"//a///b//", "/a/b"}, + {"/a/./b", "/a/b"}, + {"/a/../b", "/b"}, + {" /a/b ", "/a/b"}, + } { + if got := normalizePath(tc.in); got != tc.want { + t.Errorf("normalizePath(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +// The server accepts any preferredPartSize it is given, so these bounds are the +// only thing keeping parts hash-aligned and small enough for Telegram. +func TestNormalizeChunkSize(t *testing.T) { + for _, tc := range []struct{ in, want int64 }{ + {0, defaultChunkMiB}, + {-5, defaultChunkMiB}, + {16, 16}, + {512, 512}, + {1, 16}, // below one tree-hash block + {10, 16}, // rounded up to a block boundary + {17, 32}, // ditto + {100, 112}, // ditto + {2000, 2000}, // exactly the ceiling + {2001, maxChunkMiB}, + {5000, maxChunkMiB}, + } { + got := normalizeChunkSize(tc.in) + if got != tc.want { + t.Errorf("normalizeChunkSize(%d) = %d, want %d", tc.in, got, tc.want) + } + if got%treeHashBlockMiB != 0 { + t.Errorf("normalizeChunkSize(%d) = %d, which is not a multiple of %d", tc.in, got, treeHashBlockMiB) + } + if got > maxChunkMiB { + t.Errorf("normalizeChunkSize(%d) = %d, above the %d ceiling", tc.in, got, maxChunkMiB) + } + } +} + +// The value arrives from a select, but a hand-edited config or an older storage +// can still carry anything at all, and none of it may take the storage down. +func TestParseShareExpire(t *testing.T) { + for _, tc := range []struct { + in string + want time.Duration + }{ + {"1", time.Hour}, + {"24", 24 * time.Hour}, + {"168", 168 * time.Hour}, + {" 6 ", 6 * time.Hour}, + {"", defaultShareExpireHours * time.Hour}, + {"0", defaultShareExpireHours * time.Hour}, + {"-3", defaultShareExpireHours * time.Hour}, + {"abc", defaultShareExpireHours * time.Hour}, + {"1.5", defaultShareExpireHours * time.Hour}, + } { + if got := parseShareExpire(tc.in); got != tc.want { + t.Errorf("parseShareExpire(%q) = %v, want %v", tc.in, got, tc.want) + } + } +} + +// teldrive collapses several documented statuses onto 409, so the whole 4xx +// family has to count as final; only overload and server faults are worth +// replaying. +func TestRetryable(t *testing.T) { + for _, tc := range []struct { + name string + status int + want bool + }{ + {"bad request", http.StatusBadRequest, false}, + {"unauthorized", http.StatusUnauthorized, false}, + {"not found", http.StatusNotFound, false}, + {"part conflict", http.StatusConflict, false}, + {"gone", http.StatusGone, false}, + {"hash mismatch", http.StatusUnprocessableEntity, false}, + {"rate limited", http.StatusTooManyRequests, true}, + {"internal", http.StatusInternalServerError, true}, + {"bad gateway", http.StatusBadGateway, true}, + {"unavailable", http.StatusServiceUnavailable, true}, + } { + err := &ErrResp{status: tc.status} + if got := retryable(err); got != tc.want { + t.Errorf("%s (%d): retryable = %v, want %v", tc.name, tc.status, got, tc.want) + } + } + + // A transport failure never reached the server, so it says nothing about + // whether the request would be accepted. + if !retryable(http.ErrHandlerTimeout) { + t.Error("a non-API error should be retryable") + } +} + +// Every mutating v2 endpoint rejects a request without this header, with a +// message that points at the body rather than the header. +func TestIdempotentSetsAStableKey(t *testing.T) { + inner := false + cb := idempotent(func(*resty.Request) { inner = true }) + + first := resty.New().R() + cb(first) + key := first.Header.Get("Idempotency-Key") + if key == "" { + t.Fatal("Idempotency-Key was not set") + } + if !inner { + t.Error("the wrapped callback did not run") + } + + // A retry of the same logical operation must reuse the key. + second := resty.New().R() + cb(second) + if got := second.Header.Get("Idempotency-Key"); got != key { + t.Errorf("key changed between attempts: %q then %q", key, got) + } + + // A different operation must not. + other := resty.New().R() + idempotent(nil)(other) + if got := other.Header.Get("Idempotency-Key"); got == key { + t.Error("two operations shared an idempotency key") + } +} + +func TestErrRespMessage(t *testing.T) { + e := &ErrResp{status: http.StatusConflict} + e.Detail.Code = "conflict" + e.Detail.Message = "operation conflicts with current state" + + msg := e.Error() + for _, want := range []string{"409", "conflict", "operation conflicts"} { + if !strings.Contains(msg, want) { + t.Errorf("Error() = %q, missing %q", msg, want) + } + } + if e.IsNotFound() { + t.Error("a 409 must not report as not-found") + } + if !(&ErrResp{status: http.StatusNotFound}).IsNotFound() { + t.Error("a 404 must report as not-found") + } +} + +// TestSweepSharesDropsLapsedEntries pins the bound on shareCache: a share that +// has run out is not left in the map for the lifetime of the storage. +func TestSweepShares(t *testing.T) { + now := time.Now().UTC() + d := &TeldriveV2{shareCache: &sync.Map{}} + d.shareCache.Store("live", &cachedShare{token: "a", expiresAt: now.Add(time.Hour)}) + d.shareCache.Store("lapsed", &cachedShare{token: "b", expiresAt: now.Add(-time.Minute)}) + // Inside the renew margin: unusable, so it should go too. + d.shareCache.Store("lapsing", &cachedShare{token: "c", expiresAt: now.Add(shareRenewMargin / 2)}) + + d.sweepShares(now) + + for _, tc := range []struct { + key string + want bool + }{{"live", true}, {"lapsed", false}, {"lapsing", false}} { + if _, ok := d.shareCache.Load(tc.key); ok != tc.want { + t.Errorf("after sweep, %q present = %v, want %v", tc.key, ok, tc.want) + } + } + + // A second sweep inside the margin must be skipped, not repeated. + d.shareCache.Store("added", &cachedShare{token: "d", expiresAt: now.Add(-time.Minute)}) + d.sweepShares(now) + if _, ok := d.shareCache.Load("added"); !ok { + t.Error("sweep ran again within shareRenewMargin; it should be rate limited") + } +} diff --git a/drivers/teldrive_v2/meta.go b/drivers/teldrive_v2/meta.go new file mode 100644 index 0000000000..e9aebf90f4 --- /dev/null +++ b/drivers/teldrive_v2/meta.go @@ -0,0 +1,50 @@ +package teldrive_v2 + +import ( + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/op" +) + +type Addition struct { + driver.RootPath + Address string `json:"url" required:"true"` + ApiKey string `json:"api_key" type:"string" required:"true" help:"API key from TelDrive (Settings -> API keys). Sent as the X-Api-Key header."` + + UseShareLink bool `json:"use_share_link" type:"bool" default:"false" help:"Serve downloads as a direct 302 to TelDrive instead of streaming them through OpenList. Requires web proxy to be off. Each link carries a TelDrive share token, so anyone holding the URL can download that file without signing in, and OpenList can no longer count or revoke those accesses. Shares are left on your TelDrive until they expire, including after this storage is removed."` + // A select rather than a number: an emptied number field arrives as "" and + // fails to unmarshal into int64, which leaves the storage dead with a + // message about readUint64 that means nothing to the person reading it. + ShareExpire string `json:"share_expire" type:"select" options:"1,2,3,6,12,24,48,72,168" default:"1" help:"Only used when Use share link is on. How long each generated TelDrive link stays usable, in hours. This is not the expiry you set when sharing in OpenList; it only bounds how long a link that leaks keeps working. Raise it if long downloads are cut off."` + + // 512 MiB is what the server itself picks when preferredPartSize is omitted, + // and what the reference rclone backend defaults to. Note that OpenList + // buffers each in-flight part, so the working set is roughly + // chunk_size x upload_concurrency - lower this on memory-tight hosts. + ChunkSize int64 `json:"chunk_size" type:"number" default:"512" help:"Preferred upload part size in MiB, rounded up to a multiple of 16 and capped at 2000. Smaller parts use less memory and move the progress bar more often."` + UploadConcurrency int64 `json:"upload_concurrency" type:"number" default:"4" help:"Number of parts uploaded in parallel (max 16). Each one is buffered, so this multiplies chunk size."` + + // The server hashes every part regardless; this only controls whether we + // also hash locally and make the server compare, catching corruption in + // transit. Named after the reference rclone backend's option. + HashEnabled bool `json:"hash_enabled" type:"bool" default:"true" help:"Send a BLAKE3 checksum with each part so the server verifies it. Costs one extra read of each part."` + + HardDelete bool `json:"hard_delete" type:"bool" default:"false" help:"Purge files permanently instead of moving them to trash. Trashed files stay recoverable, and their Telegram storage is only reclaimed when TelDrive's cleanup job expires them."` +} + +var config = driver.Config{ + Name: "Teldrive V2", + DefaultRoot: "/", + // v2 always resolves conflicts server-side via conflictPolicy + NoOverwriteUpload: false, + // Without UseShareLink the download URL is authenticated by the X-Api-Key + // header, which a browser following a 302 cannot send - it just gets a 401. + // Default to proxying so a freshly added storage works; turning UseShareLink + // on is what makes direct 302 links viable. + PreferProxy: true, +} + +func init() { + op.RegisterDriver(func() driver.Driver { + return &TeldriveV2{} + }) +} diff --git a/drivers/teldrive_v2/types.go b/drivers/teldrive_v2/types.go new file mode 100644 index 0000000000..35fa232e65 --- /dev/null +++ b/drivers/teldrive_v2/types.go @@ -0,0 +1,119 @@ +package teldrive_v2 + +import ( + "fmt" + "time" +) + +// ErrResp is the v2 error envelope: {"error":{"code":"...","message":"..."}} +type ErrResp struct { + Detail struct { + Code string `json:"code"` + Message string `json:"message"` + Details map[string]any `json:"details,omitempty"` + } `json:"error"` + status int +} + +func (e *ErrResp) Error() string { + if e == nil { + return "" + } + return fmt.Sprintf("[TeldriveV2] %d %s: %s", e.status, e.Detail.Code, e.Detail.Message) +} + +func (e *ErrResp) IsNotFound() bool { return e != nil && e.status == 404 } + +// FileKind values +const ( + KindFile = "file" + KindFolder = "folder" +) + +// NameConflictPolicy values +const ( + PolicyFail = "fail" + PolicyReplace = "replace" + PolicyRename = "rename" +) + +type FileHash struct { + Algorithm string `json:"algorithm"` + Value string `json:"value"` +} + +// FileEntry is the v2 replacement for the v1 Object. +// Note the renames: type -> kind, updatedAt -> modTime. +type FileEntry struct { + ID string `json:"id"` + ParentID string `json:"parentId,omitempty"` + Name string `json:"name"` + Kind string `json:"kind"` + MimeType string `json:"mimeType,omitempty"` + Size int64 `json:"size,omitempty"` + Hash *FileHash `json:"hash,omitempty"` + Encryption bool `json:"encryption"` + Status string `json:"status"` + ModTime time.Time `json:"modTime"` + Generation int64 `json:"generation"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// ListResp is cursor-paginated; v1's meta.totalPages is gone. +type ListResp struct { + Items []FileEntry `json:"items"` + NextCursor string `json:"nextCursor,omitempty"` +} + +type UploadSession struct { + ID string `json:"id"` + ParentID string `json:"parentId,omitempty"` + Name string `json:"name"` + ExpectedSize int64 `json:"expectedSize"` + MimeType string `json:"mimeType,omitempty"` + ModTime time.Time `json:"modTime"` + Encryption bool `json:"encryption"` + ConflictPolicy string `json:"conflictPolicy"` + PartSize int64 `json:"partSize"` + State string `json:"state"` + ExpiresAt time.Time `json:"expiresAt"` + CreatedAt time.Time `json:"createdAt"` +} + +type UploadPart struct { + UploadID string `json:"uploadId"` + PartNo int `json:"partNo"` + State string `json:"state"` + PlainSize int64 `json:"plainSize"` + StoredSize int64 `json:"storedSize,omitempty"` + Checksum string `json:"checksum,omitempty"` +} + +type UploadPartList struct { + Items []UploadPart `json:"items"` + NextCursor string `json:"nextCursor,omitempty"` +} + +// ShareCreated is returned only once, at creation time. +type ShareCreated struct { + ID string `json:"id"` + FileID string `json:"fileId"` + Token string `json:"token"` + PublicURL string `json:"publicUrl"` + PasswordProtected bool `json:"passwordProtected"` + Permission string `json:"permission"` + ExpiresAt time.Time `json:"expiresAt,omitempty"` + CreatedAt time.Time `json:"createdAt"` +} + +type UploadSessionList struct { + Items []UploadSession `json:"items"` + NextCursor string `json:"nextCursor,omitempty"` +} + +// UploadState values +const ( + UploadOpen = "open" + UploadPartStored = "stored" +) diff --git a/drivers/teldrive_v2/upload.go b/drivers/teldrive_v2/upload.go new file mode 100644 index 0000000000..b65c1ac787 --- /dev/null +++ b/drivers/teldrive_v2/upload.go @@ -0,0 +1,312 @@ +package teldrive_v2 + +import ( + "context" + "fmt" + "io" + "math" + "net/http" + "strconv" + "strings" + "sync/atomic" + "time" + + "github.com/OpenListTeam/OpenList/v4/drivers/base" + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/stream" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + hash_extend "github.com/OpenListTeam/OpenList/v4/pkg/utils/hash" + "github.com/avast/retry-go" + "github.com/go-resty/resty/v2" + "golang.org/x/sync/errgroup" +) + +const uploadMaxRetries = 3 + +// Put uploads through the v2 durable upload session: +// +// POST /uploads -> server allocates the id and the part size +// PUT /uploads/{id}/parts/{n} -> idempotent on (uploadId, partNo) +// POST /uploads/{id}/complete -> transactional publish +// +// This replaces the v1 dance of minting a client-side uuid, naming every chunk, +// re-reading the remote part list and rebuilding the file from parts+salts. +func (d *TeldriveV2) Put(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress) (model.Obj, error) { + parentID, err := d.dirID(ctx, dstDir) + if err != nil { + return nil, err + } + + body := base.Json{ + "name": file.GetName(), + "size": file.GetSize(), + // v2 stores the client mtime, so uploads no longer lose it. + "modTime": file.ModTime().UTC().Format(time.RFC3339Nano), + // The server resolves the overwrite itself; no pre-delete round trip. + "conflictPolicy": PolicyReplace, + "preferredPartSize": d.ChunkSize * 1024 * 1024, + } + if parentID != "" { + body["parentId"] = parentID + } + if mime := file.GetMimetype(); mime != "" { + body["mimeType"] = mime + } + + session, resumed, err := d.getOrCreateSession(ctx, parentID, file, body) + if err != nil { + return nil, err + } + + // A failed session is deliberately left behind so the next attempt can pick + // up where this one stopped; teldrive expires abandoned sessions itself. + var stored map[int]UploadPart + if resumed { + stored = d.storedParts(ctx, session.ID) + } + + if err := d.uploadParts(ctx, session, file, up, stored); err != nil { + return nil, err + } + + entry := &FileEntry{} + if err := d.request(ctx, http.MethodPost, "/uploads/{id}/complete", idempotent(func(req *resty.Request) { + req.SetPathParam("id", session.ID) + }), entry); err != nil { + return nil, err + } + + return d.toObj(entry, utils.FixAndCleanPath(dstDir.GetPath()+"/"+entry.Name)), nil +} + +func (d *TeldriveV2) uploadParts(ctx context.Context, session *UploadSession, file model.FileStreamer, up driver.UpdateProgress, stored map[int]UploadPart) error { + totalSize := file.GetSize() + if totalSize <= 0 { + // Empty file: complete publishes it with no parts. + up(100) + return nil + } + + // The part size is the one the server allocated, not the configured + // preference - v2 aligns it to its own block size. + partSize := session.PartSize + if partSize <= 0 { + partSize = d.ChunkSize * 1024 * 1024 + } + totalParts := int(math.Ceil(float64(totalSize) / float64(partSize))) + + ss, err := stream.NewStreamSectionReader(file, int(min(partSize, totalSize)), &up) + if err != nil { + return err + } + + progress := &uploadProgress{total: totalSize, up: up} + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(int(d.UploadConcurrency)) + + for partNo := 1; partNo <= totalParts; partNo++ { + if utils.IsCanceled(gctx) { + break + } + offset := int64(partNo-1) * partSize + curSize := min(totalSize-offset, partSize) + done, resumable := stored[partNo] + resumable = resumable && done.PlainSize == curSize + + // Without a checksum to compare there is nothing to verify, so trust the + // recorded size and skip the bytes without ever buffering them. + if resumable && !d.HashEnabled { + if err := ss.DiscardSection(offset, curSize); err != nil { + return err + } + progress.commit(curSize) + continue + } + + reader, err := ss.GetSectionReader(offset, curSize) + if err != nil { + return err + } + + // Hashing here rather than inside the worker: the same value answers + // "may this part be skipped" and, if not, becomes its X-Part-Checksum. + var checksum string + if d.HashEnabled { + if _, err := reader.Seek(0, io.SeekStart); err != nil { + ss.FreeSectionReader(reader) + return err + } + if checksum, err = utils.HashReader(hash_extend.BLAKE3Tree, reader); err != nil { + ss.FreeSectionReader(reader) + return err + } + if resumable && strings.EqualFold(checksum, done.Checksum) { + ss.FreeSectionReader(reader) + progress.commit(curSize) + continue + } + } + + partNo, curSize, reader, checksum := partNo, curSize, reader, checksum + g.Go(func() error { + defer ss.FreeSectionReader(reader) + if err := d.putPart(gctx, session.ID, partNo, curSize, reader, checksum); err != nil { + return fmt.Errorf("upload part %d: %w", partNo, err) + } + progress.commit(curSize) + return nil + }) + } + + return g.Wait() +} + +// uploadProgress tracks how much of the file the server has actually committed. +// +// A part is credited when its PUT returns, not as its bytes leave us. teldrive +// drains a whole part body at local speed and only then relays it to Telegram - +// measured at 256 MiB absorbed in ~0.5s against a 26s request - so counting +// bytes on the way out would report ~100% while nearly all the work remained. +// Part granularity is the finest signal the API actually gives us; smaller +// chunk_size is what buys a smoother bar. +type uploadProgress struct { + total int64 + done atomic.Int64 + up driver.UpdateProgress +} + +func (p *uploadProgress) commit(n int64) { + p.up(float64(p.done.Add(n)) / float64(p.total) * 100) +} + +// putPart uploads one part. PUT is idempotent on (uploadId, partNo), so a retry +// after a mid-flight failure either replays cleanly or returns the stored part. +func (d *TeldriveV2) putPart(ctx context.Context, uploadID string, partNo int, size int64, reader io.ReadSeeker, checksum string) error { + return retry.Do(func() error { + if _, err := reader.Seek(0, io.SeekStart); err != nil { + return err + } + part := &UploadPart{} + return d.uploadRequest(ctx, http.MethodPut, "/uploads/{id}/parts/{partNo}", func(req *resty.Request) { + req.SetPathParam("id", uploadID) + req.SetPathParam("partNo", strconv.Itoa(partNo)) + req.SetHeader("Content-Type", "application/octet-stream") + // Deliberately not SetContentLength: the upload client's pre-request + // hook promotes this header onto the raw request instead, so the body + // streams rather than being buffered whole. + req.SetHeader("Content-Length", strconv.FormatInt(size, 10)) + if checksum != "" { + req.SetHeader("X-Part-Checksum", checksum) + } + req.SetBody(driver.NewLimitedUploadStream(ctx, reader)) + }, part) + }, + retry.Context(ctx), + retry.RetryIf(retryable), + retry.Attempts(uploadMaxRetries), + retry.DelayType(retry.BackOffDelay), + retry.Delay(time.Second), + ) +} + +// sessionMatchSlack absorbs the second-level rounding the server applies to +// modTime, so a resumed session still matches the file it was opened for. +const sessionMatchSlack = time.Second + +// getOrCreateSession reuses an interrupted upload instead of starting over. +// +// A session survives a failed transfer (the server keeps it until its own +// cleanup job expires it), and every part already stored under it can be +// skipped, so an upload that died on its last part does not re-send the ones +// before it. Both first-party clients do this; teldrive's own web UI calls the +// same two endpoints. +// +// Matching is on parent, name and size. That alone is not proof of identity, so +// safety comes from re-hashing each candidate part and comparing it with the +// checksum the server recorded before anything is skipped. When hashing is off +// there is nothing to compare, so mtime has to carry the identity instead and +// is required to match too. +// +// mtime is deliberately not part of the match otherwise: OpenList only knows it +// from a Last-Modified header (server/handles/fsup.go), and a client that omits +// one gets time.Now(), which differs on every attempt and would make resuming +// impossible for exactly the uploads that need it most. +func (d *TeldriveV2) getOrCreateSession(ctx context.Context, parentID string, file model.FileStreamer, body base.Json) (*UploadSession, bool, error) { + modTime := file.ModTime().UTC() + var cursor string + for { + resp := &UploadSessionList{} + if err := d.request(ctx, http.MethodGet, "/uploads", func(req *resty.Request) { + params := map[string]string{"state": UploadOpen, "limit": strconv.Itoa(listPageSize)} + if cursor != "" { + params["cursor"] = cursor + } + req.SetQueryParams(params) + }, resp); err != nil { + // Resuming is an optimisation; a listing failure must not stop the + // upload, so fall through to creating a fresh session. + break + } + for i := range resp.Items { + s := &resp.Items[i] + if s.State != UploadOpen || s.ParentID != parentID || s.Name != file.GetName() { + continue + } + if s.ExpectedSize != file.GetSize() || s.PartSize <= 0 { + continue + } + if !d.HashEnabled { + diff := s.ModTime.UTC().Sub(modTime) + if diff > sessionMatchSlack || diff < -sessionMatchSlack { + continue + } + } + if !s.ExpiresAt.IsZero() && !s.ExpiresAt.After(time.Now()) { + continue + } + return s, true, nil + } + if resp.NextCursor == "" { + break + } + cursor = resp.NextCursor + } + + session := &UploadSession{} + if err := d.request(ctx, http.MethodPost, "/uploads", idempotent(func(req *resty.Request) { + req.SetBody(body) + }), session); err != nil { + return nil, false, err + } + return session, false, nil +} + +// storedParts indexes the parts the server already holds for a session. +func (d *TeldriveV2) storedParts(ctx context.Context, uploadID string) map[int]UploadPart { + parts := make(map[int]UploadPart) + var cursor string + for { + resp := &UploadPartList{} + if err := d.request(ctx, http.MethodGet, "/uploads/{id}/parts", func(req *resty.Request) { + req.SetPathParam("id", uploadID) + params := map[string]string{"limit": strconv.Itoa(listPageSize)} + if cursor != "" { + params["cursor"] = cursor + } + req.SetQueryParams(params) + }, resp); err != nil { + // Same reasoning as above: worst case we re-upload everything. + return parts + } + for _, p := range resp.Items { + if p.State == UploadPartStored { + parts[p.PartNo] = p + } + } + if resp.NextCursor == "" { + return parts + } + cursor = resp.NextCursor + } +} diff --git a/drivers/teldrive_v2/util.go b/drivers/teldrive_v2/util.go new file mode 100644 index 0000000000..0f496ba4af --- /dev/null +++ b/drivers/teldrive_v2/util.go @@ -0,0 +1,218 @@ +package teldrive_v2 + +import ( + "context" + "errors" + "fmt" + "net/http" + "path" + "strconv" + "strings" + "time" + + "github.com/OpenListTeam/OpenList/v4/drivers/base" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/go-resty/resty/v2" + "github.com/google/uuid" +) + +const apiPrefix = "/api/v1" + +// request performs an authenticated metadata call against the v2 API. +// Unlike v1 (cookie `access_token`), v2 authenticates with the X-Api-Key header. +func (d *TeldriveV2) request(ctx context.Context, method, pathname string, callback base.ReqCallback, resp any) error { + return d.do(base.RestyClient, ctx, method, pathname, callback, resp) +} + +// uploadRequest sends a part over the driver's own client. +// +// base.RestyClient caps every request at base.DefaultTimeout (30s), which a part +// upload blows straight through: the server only answers once it has relayed the +// whole part to Telegram. The shared client also retries three times on its own, +// which would replay a part body behind the back of our retry policy. +func (d *TeldriveV2) uploadRequest(ctx context.Context, method, pathname string, callback base.ReqCallback, resp any) error { + return d.do(d.uploadClient, ctx, method, pathname, callback, resp) +} + +func (d *TeldriveV2) do(client *resty.Client, ctx context.Context, method, pathname string, callback base.ReqCallback, resp any) error { + req := client.R().SetContext(ctx) + req.SetHeader("X-Api-Key", d.ApiKey) + if callback != nil { + callback(req) + } + if resp != nil { + req.SetResult(resp) + } + e := &ErrResp{} + req.SetError(e) + + res, err := req.Execute(method, d.Address+apiPrefix+pathname) + if err != nil { + return err + } + if res.IsError() { + e.status = res.StatusCode() + if e.Detail.Message == "" { + e.Detail.Message = res.Status() + } + return e + } + return nil +} + +// idempotent stamps the Idempotency-Key header that v2 requires on every +// mutating endpoint (folders, move, copy, shares, uploads, uploads/complete). +// One fresh UUID per logical operation: the key only has to stay stable across +// retries of that same call, which resty replays with the header intact. +func idempotent(callback base.ReqCallback) base.ReqCallback { + key := uuid.NewString() + return func(req *resty.Request) { + req.SetHeader("Idempotency-Key", key) + if callback != nil { + callback(req) + } + } +} + +// retryable reports whether an error is worth another attempt. +// Anything the server answered with a 4xx is a decision, not a hiccup: a 409 on +// a part whose size or checksum disagrees, a 410 on an expired session and a +// 422 on a malformed body all reproduce exactly on replay. Only overload (429), +// server faults (5xx) and transport errors get retried. +func retryable(err error) bool { + var e *ErrResp + if !errors.As(err, &e) { + // Transport-level failure with no HTTP response - worth retrying. + return true + } + return e.status == http.StatusTooManyRequests || e.status >= http.StatusInternalServerError +} + +// normalizeChunkSize brings a configured part size, in MiB, into the range the +// server will not enforce for us. +// +// The server does not sanity-check preferredPartSize at all: it echoes back +// whatever is asked for, 1 MiB or 10 GiB alike. So both bounds are load-bearing. +// The lower one exists because the file hash is BLAKE3 over the concatenated +// 16 MiB block hashes of every part, and a part that is not a whole number of +// blocks shifts the block boundaries for every later part. The upper one exists +// because a part becomes a single Telegram message. +func normalizeChunkSize(mib int64) int64 { + if mib <= 0 { + return defaultChunkMiB + } + if rem := mib % treeHashBlockMiB; rem != 0 { + mib += treeHashBlockMiB - rem + } + if mib > maxChunkMiB { + return maxChunkMiB + } + return mib +} + +// parseShareExpire turns the configured hour count into a duration, falling +// back to the default for anything unusable so a bad value cannot take the +// storage down. +func parseShareExpire(v string) time.Duration { + hours, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64) + if err != nil || hours <= 0 { + hours = defaultShareExpireHours + } + return time.Duration(hours) * time.Hour +} + +// listPage fetches a single cursor page. Exactly one of parentID / dirPath is +// used: v2 rejects requests that carry both. +func (d *TeldriveV2) listPage(ctx context.Context, parentID, dirPath, cursor string, extra map[string]string) (*ListResp, error) { + resp := &ListResp{} + err := d.request(ctx, http.MethodGet, "/files", func(req *resty.Request) { + params := map[string]string{ + "limit": strconv.Itoa(listPageSize), + "status": "active", + } + if parentID != "" { + params["parentId"] = parentID + } else if dirPath != "" { + params["path"] = dirPath + } + if cursor != "" { + params["cursor"] = cursor + } + for k, v := range extra { + params[k] = v + } + req.SetQueryParams(params) + }, resp) + if err != nil { + return nil, err + } + return resp, nil +} + +// listAll walks every cursor page of a folder. +// v1 could fan out pages in parallel because it knew meta.totalPages up front; +// v2 is cursor-based, so pages must be fetched sequentially. +func (d *TeldriveV2) listAll(ctx context.Context, parentID, dirPath string) ([]FileEntry, error) { + var ( + items []FileEntry + cursor string + ) + for { + resp, err := d.listPage(ctx, parentID, dirPath, cursor, nil) + if err != nil { + return nil, err + } + items = append(items, resp.Items...) + if resp.NextCursor == "" { + return items, nil + } + cursor = resp.NextCursor + } +} + +// normalizePath turns any input into a clean absolute drive path ("" -> "/"). +func normalizePath(p string) string { + p = path.Clean("/" + strings.TrimSpace(p)) + if p == "." { + return "/" + } + return p +} + +// resolveFolderID maps an absolute drive path to a folder UUID. +// An empty string means the drive root: v2 treats a missing parentId as root, +// so callers pass it straight through and simply omit the field. +func (d *TeldriveV2) resolveFolderID(ctx context.Context, folderPath string) (string, error) { + folderPath = normalizePath(folderPath) + if folderPath == "/" { + return "", nil + } + if id, ok := d.pathCache.Load(folderPath); ok { + return id.(string), nil + } + + parent, name := path.Split(folderPath) + resp, err := d.listPage(ctx, "", normalizePath(parent), "", map[string]string{ + "search": name, + "kind": KindFolder, + }) + if err != nil { + return "", err + } + for _, it := range resp.Items { + if it.Kind == KindFolder && it.Name == name { + d.pathCache.Store(folderPath, it.ID) + return it.ID, nil + } + } + return "", fmt.Errorf("[TeldriveV2] folder not found: %s", folderPath) +} + +// dirID returns the folder UUID for a listed object, falling back to a path +// lookup for objects that reached us without an ID. +func (d *TeldriveV2) dirID(ctx context.Context, dir model.Obj) (string, error) { + if id := dir.GetID(); id != "" { + return id, nil + } + return d.resolveFolderID(ctx, dir.GetPath()) +} diff --git a/go.mod b/go.mod index 2eabb81d4c..7c9a42fd47 100644 --- a/go.mod +++ b/go.mod @@ -74,6 +74,7 @@ require ( github.com/tchap/go-patricia/v2 v2.3.3 github.com/u2takey/ffmpeg-go v0.5.0 github.com/upyun/go-sdk/v3 v3.0.4 + github.com/zeebo/blake3 v0.2.4 github.com/zzzhr1990/go-common-entity v0.0.0-20250202070650-1a200048f0d3 golang.org/x/crypto v0.55.0 golang.org/x/image v0.44.0 @@ -136,7 +137,6 @@ require ( github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/stangelandcl/ppmd v0.1.1 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect - github.com/zeebo/blake3 v0.2.4 // indirect github.com/zeebo/xxh3 v1.1.0 // indirect go.mongodb.org/mongo-driver v1.17.9 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect diff --git a/pkg/utils/hash/tdhash.go b/pkg/utils/hash/tdhash.go new file mode 100644 index 0000000000..9adf4b2a80 --- /dev/null +++ b/pkg/utils/hash/tdhash.go @@ -0,0 +1,78 @@ +package hash_extend + +import ( + "hash" + + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + "github.com/zeebo/blake3" +) + +// tdBlockSize is TelDrive's fixed tree-hash block size. +const tdBlockSize = 16 * 1024 * 1024 + +// BLAKE3Tree is the content hash used by TelDrive v2. +// +// It is not plain BLAKE3: the stream is cut into fixed 16 MiB blocks, each block +// is hashed with BLAKE3, and the concatenated block digests are hashed once more +// to give the root. The server computes it for every upload part and reports it +// on every file, which is why it gets its own type rather than reusing "blake3" +// - a value produced by a plain BLAKE3 of the same bytes would not match. +// +// Algorithm mirrors teldrive's internal/treehash and rclone's +// backend/teldrive/tdhash (both MIT). rclone exposes the same value under the +// name "teldrive", so `rclone lsjson --hash` output is directly comparable. +var BLAKE3Tree = utils.RegisterHash("blake3_tree", "BLAKE3-Tree", 64, NewBLAKE3Tree) + +func NewBLAKE3Tree() hash.Hash { + return &blake3Tree{block: blake3.New()} +} + +type blake3Tree struct { + block *blake3.Hasher + blocks [][]byte // digests of the completed blocks + n int // bytes written into the current block +} + +func (h *blake3Tree) Write(p []byte) (int, error) { + n := len(p) + for len(p) > 0 { + room := tdBlockSize - h.n + if room > len(p) { + room = len(p) + } + h.block.Write(p[:room]) + h.n += room + p = p[room:] + + if h.n == tdBlockSize { + h.blocks = append(h.blocks, h.block.Sum(nil)) + h.block.Reset() + h.n = 0 + } + } + return n, nil +} + +// Sum does not consume the hasher: a trailing partial block is folded into a +// copy so that writing can continue afterwards. +func (h *blake3Tree) Sum(b []byte) []byte { + digests := h.blocks + if h.n > 0 { + digests = append(append(make([][]byte, 0, len(h.blocks)+1), h.blocks...), h.block.Sum(nil)) + } + root := blake3.New() + for _, d := range digests { + root.Write(d) + } + return root.Sum(b) +} + +func (h *blake3Tree) Reset() { + h.block.Reset() + h.blocks = nil + h.n = 0 +} + +func (h *blake3Tree) Size() int { return 32 } + +func (h *blake3Tree) BlockSize() int { return tdBlockSize } diff --git a/pkg/utils/hash/tdhash_test.go b/pkg/utils/hash/tdhash_test.go new file mode 100644 index 0000000000..a8552db7e5 --- /dev/null +++ b/pkg/utils/hash/tdhash_test.go @@ -0,0 +1,93 @@ +package hash_extend + +import ( + "encoding/hex" + "testing" + + "github.com/OpenListTeam/OpenList/v4/pkg/utils" +) + +// fixture reproduces the byte pattern of the file these vectors were captured +// from: a 36 MiB stream where data[i] == byte(i % 256). off is the offset of the +// returned slice within that stream. +func fixture(off, n int) []byte { + b := make([]byte, n) + for i := range b { + b[i] = byte((off + i) % 256) + } + return b +} + +// Vectors captured from a live teldrive v2 server (1.8.3-SNAPSHOT-e3142b5) by +// uploading the fixture in 16 MiB parts and recording the checksum the server +// computed for each part plus the tree hash it published for the whole file. +// 36 MiB is deliberate: two whole 16 MiB blocks plus a 4 MiB remainder, so the +// vectors cover a full block, a partial block, and the concatenation of both. +func TestBLAKE3TreeServerVectors(t *testing.T) { + for _, tc := range []struct { + name string + off int + n int + want string + }{ + {"whole file: 2 full blocks + partial", 0, 37748736, "c2ccef62af6cd438401c9a547d031c01e07b066483420321265acde683a414f3"}, + {"part 1: exactly one block", 0, 16777216, "0d6739b729971512b1d02c2029eae45525db27ddcd5fdfb5010256a6c946c674"}, + {"part 3: partial block", 33554432, 4194304, "7cf2f855eaf2fc6863e64ce56879f61e3d91eb0e02ff0ae28f713807e3bd2ab6"}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := utils.HashData(BLAKE3Tree, fixture(tc.off, tc.n)); got != tc.want { + t.Errorf("hash = %s, want %s", got, tc.want) + } + }) + } +} + +// The block boundary is the only interesting state in Write, so feed the same +// bytes in sizes that straddle it and confirm the result never changes. +func TestBLAKE3TreeIsWriteChunkingIndependent(t *testing.T) { + data := fixture(0, tdBlockSize+4096) + want := utils.HashData(BLAKE3Tree, data) + + for _, step := range []int{1, 7, 4096, tdBlockSize - 1, tdBlockSize, tdBlockSize + 1} { + h := NewBLAKE3Tree() + for off := 0; off < len(data); off += step { + end := min(off+step, len(data)) + if _, err := h.Write(data[off:end]); err != nil { + t.Fatalf("Write: %v", err) + } + } + if got := hex.EncodeToString(h.Sum(nil)); got != want { + t.Errorf("step %d: hash = %s, want %s", step, got, want) + } + } +} + +// Sum must not consume the hasher: the driver hashes a part, then hashes the +// next one with a fresh hasher, and utils.HashReader calls Sum once at the end. +func TestBLAKE3TreeSumIsRepeatable(t *testing.T) { + h := NewBLAKE3Tree() + h.Write(fixture(0, 1000)) + + first := hex.EncodeToString(h.Sum(nil)) + if second := hex.EncodeToString(h.Sum(nil)); second != first { + t.Errorf("Sum is destructive: %s then %s", first, second) + } + + h.Write(fixture(1000, 1000)) + if after := hex.EncodeToString(h.Sum(nil)); after == first { + t.Error("Sum ignored data written after an earlier Sum") + } +} + +func TestBLAKE3TreeRegistered(t *testing.T) { + ht, ok := utils.GetHashByName("blake3_tree") + if !ok { + t.Fatal("blake3_tree is not registered") + } + if ht != BLAKE3Tree { + t.Error("blake3_tree resolves to a different hash type") + } + if BLAKE3Tree.Width != 64 { + t.Errorf("Width = %d, want 64", BLAKE3Tree.Width) + } +}