package storage import ( "context" "crypto/sha256" "encoding/hex" "encoding/xml" "errors" "fmt" "io" "math/rand" "net" "net/http" "net/url" "os" "path" "strconv" "strings" "sync" "time" ) // WebDAVStorage 基于 net/http 的 WebDAV 引擎(本次重写的重点优化对象)。 // // 相比参考实现(WebDAVFileStorage,aiohttp)的改进: // - 单例 http.Client + 连接池化 Transport(参考实现每个操作新建 ClientSession,无复用); // - Basic 与 Digest(RFC 2617,qop=auth,MD5/SHA-256)双认证自动协商(参考实现仅 Basic); // - GET 下载透传 Range 头(参考实现全量 GET,无法断点/分段); // - 5xx/429/网络错误指数退避重试,可配次数(参考实现无重试); // - 下载经 io.Pipe 流式转发,全程不落盘; // - 目录存在性内存缓存,按需逐级 MKCOL,避免每次保存都发 PROPFIND; // - 非流式操作带可配超时;流式传输由调用方 ctx 管控(可取消)。 type WebDAVStorage struct { base *url.URL // 服务基址(含可能的路径前缀),以 / 结尾 root string // 远端根目录(webdav_root_path) username string password string client *http.Client transport *http.Transport auth *authState maxRetries int // 5xx/网络错误最大重试次数 baseBackoff time.Duration // 退避基数 opTimeout time.Duration // 非流式操作超时 dirMu sync.RWMutex knownDirs map[string]struct{} // 已确认存在的远端目录(含根前缀) spacesPool sync.Pool // 256KB 复用缓冲 } // NewWebDAVStorage 构造 WebDAV 引擎。 func NewWebDAVStorage(opts WebDAVOptions) (*WebDAVStorage, error) { opts.applyDefaults() raw := strings.TrimSpace(opts.BaseURL) if raw == "" { return nil, fmt.Errorf("storage/webdav: 缺少 webdav_url 配置") } if !strings.Contains(raw, "://") { raw = "http://" + raw } u, err := url.Parse(raw) if err != nil { return nil, fmt.Errorf("storage/webdav: webdav_url 非法: %w", err) } if u.Scheme != "http" && u.Scheme != "https" { return nil, fmt.Errorf("storage/webdav: webdav_url 仅支持 http/https,收到 %q", u.Scheme) } if !strings.HasSuffix(u.Path, "/") { u.Path += "/" } root := strings.Trim(opts.RootPath, "/") if root == "" { root = "filebox_storage" } root = strings.ReplaceAll(root, "\\", "/") transport := newPooledTransport(opts.MaxIdleConnsPerHost) return &WebDAVStorage{ base: u, root: root, username: opts.Username, password: opts.Password, client: &http.Client{Transport: transport}, transport: transport, auth: newAuthState(opts.Username, opts.Password), maxRetries: opts.MaxRetries, baseBackoff: time.Duration(opts.BaseBackoff) * time.Millisecond, opTimeout: time.Duration(opts.Timeout) * time.Second, knownDirs: map[string]struct{}{}, spacesPool: sync.Pool{New: func() any { b := make([]byte, localChunkSize) return &b }}, }, nil } func init() { RegisterEngine("webdav", func(ctx context.Context) (Storage, error) { return NewWebDAVStorage(engineOptions.WebDAV) }) } // newPooledTransport 连接池化 Transport:Keep-Alive 连接复用是 WebDAV 优化的核心。 func newPooledTransport(maxIdlePerHost int) *http.Transport { return &http.Transport{ Proxy: http.ProxyFromEnvironment, DialContext: (&net.Dialer{ Timeout: 10 * time.Second, KeepAlive: 30 * time.Second, }).DialContext, ForceAttemptHTTP2: true, MaxIdleConns: 100, MaxIdleConnsPerHost: maxIdlePerHost, IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, ExpectContinueTimeout: time.Second, ResponseHeaderTimeout: 60 * time.Second, } } // requestOpts 单次 WebDAV 请求参数。 type requestOpts struct { // body 请求体工厂:每次尝试调用一次(重试时重新获取,可重放)。 body func() (io.Reader, int64, error) // retryBody 请求体是否可重放(seekable);false 时 PUT 类请求失败不重试。 retryBody bool // headers 附加请求头。 headers map[string]string // streaming 流式传输(GET/PUT 大 body):不套 opTimeout,由调用方 ctx 管控。 streaming bool } // do 执行一次 WebDAV 请求:认证自动协商 + 指数退避重试。 // 返回的响应由调用方负责关闭(drainClose / readErrorBody)。 // // 重要:非流式操作的可配超时通过 ctx 实现,cancel 不随 do() 返回而调用, // 而是挂在 davResponse 上、待响应体读完后再触发——否则取消会提前杀掉 // Keep-Alive 连接,破坏连接复用。 func (w *WebDAVStorage) do(ctx context.Context, method, rawURL string, opts requestOpts) (*davResponse, error) { // 非流式操作套可配超时(流式由调用方 ctx 管控)。 var cancel context.CancelFunc if !opts.streaming { if _, hasDeadline := ctx.Deadline(); !hasDeadline { ctx, cancel = context.WithTimeout(ctx, w.opTimeout) } } fail := func(err error) (*davResponse, error) { if cancel != nil { cancel() } return nil, err } // 幂等方法或可重放 body 才允许整体重试。 idempotent := method == http.MethodGet || method == http.MethodHead || method == "PROPFIND" || method == "MKCOL" || method == http.MethodDelete || method == http.MethodOptions retryable := idempotent || opts.retryBody const maxAuthRetries = 2 budget := w.maxRetries + maxAuthRetries // 认证挑战重试不消耗退避预算 authRetries := 0 for attempt := 0; attempt < budget; attempt++ { var body io.Reader var length int64 = -1 if opts.body != nil { var err error body, length, err = opts.body() if err != nil { return fail(fmt.Errorf("%w: 构造请求体失败: %v", ErrUnavailable, err)) } } req, err := http.NewRequestWithContext(ctx, method, rawURL, body) if err != nil { return fail(fmt.Errorf("%w: 构造请求失败: %v", ErrInvalidPath, err)) } if length >= 0 { req.ContentLength = length } for k, v := range opts.headers { req.Header.Set(k, v) } w.auth.apply(req) resp, err := w.client.Do(req) if err != nil { if ctx.Err() != nil { // 调用方取消/超时优先 return fail(ctx.Err()) } if retryable && attempt+1 < budget { if sleepErr := w.backoff(ctx, attempt, 0); sleepErr != nil { return fail(sleepErr) } continue } return fail(fmt.Errorf("%w: %s %s: %v", ErrUnavailable, method, rawURL, err)) } // 401 认证挑战:切换 Basic/Digest 后立即重试(不退避、不额外计数)。 if resp.StatusCode == http.StatusUnauthorized && authRetries < maxAuthRetries { challenge := resp.Header.Get("WWW-Authenticate") drainClose(&davResponse{Response: resp}) if challenge != "" && w.auth.challenge(challenge) { authRetries++ continue } return fail(fmt.Errorf("%w: WebDAV 认证失败(401,%s)", ErrUnavailable, rawURL)) } // 5xx/429/408:幂等或可重放 body 时指数退避重试。 if retryable && isRetryStatus(resp.StatusCode) && attempt+1 < budget { retryAfter := retryAfterSeconds(resp.Header.Get("Retry-After")) drainClose(&davResponse{Response: resp}) if sleepErr := w.backoff(ctx, attempt, retryAfter); sleepErr != nil { return fail(sleepErr) } continue } return &davResponse{Response: resp, cancel: cancel}, nil } return fail(fmt.Errorf("%w: WebDAV 重试耗尽(%s %s)", ErrUnavailable, method, rawURL)) } // davResponse WebDAV 响应 + 关联的超时取消函数。 // 非流式操作读完响应体后必须经 drainClose/readErrorBody 释放(触发 cancel)。 type davResponse struct { *http.Response cancel context.CancelFunc } // isRetryStatus 判断状态码是否值得重试。 func isRetryStatus(code int) bool { switch code { case http.StatusRequestTimeout, http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout: return true } return false } // retryAfterSeconds 解析 Retry-After(秒);非法或负值返回 0。 func retryAfterSeconds(v string) time.Duration { if v == "" { return 0 } n, err := strconv.Atoi(strings.TrimSpace(v)) if err != nil || n <= 0 { return 0 } if n > 5 { n = 5 // 上限 5s,避免异常服务端拖死请求 } return time.Duration(n) * time.Second } // backoff 指数退避:base * 2^attempt,封顶 2s,带 ±20% 抖动;retryAfter 优先。 func (w *WebDAVStorage) backoff(ctx context.Context, attempt int, retryAfter time.Duration) error { d := retryAfter if d <= 0 { d = w.baseBackoff << attempt if d > 2*time.Second { d = 2 * time.Second } // ±20% 抖动 jitter := time.Duration(int64(d) / 5) if jitter > 0 { d -= time.Duration(rand.Int63n(int64(jitter))) } } if d <= 0 { d = time.Millisecond } timer := time.NewTimer(d) defer timer.Stop() select { case <-ctx.Done(): return ctx.Err() case <-timer.C: return nil } } // drainClose 读取少量残余并关闭响应体,保证连接可复用;随后触发超时清理。 func drainClose(resp *davResponse) { if resp == nil || resp.Body == nil { return } _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 8<<10)) _ = resp.Body.Close() if resp.cancel != nil { resp.cancel() } } // joinRemote 校验 savePath 并拼接远端完整路径(含根目录前缀)。 func (w *WebDAVStorage) joinRemote(savePath string) (string, error) { cleaned, ok := SanitizePath(savePath) if !ok { return "", fmt.Errorf("%w: %q", ErrInvalidPath, savePath) } return path.Join(w.root, cleaned), nil } // urlFor 将远端路径转为完整 URL(URL.String 自动按段转义)。 func (w *WebDAVStorage) urlFor(remotePath string) string { u := *w.base p := strings.TrimSuffix(u.Path, "/") remotePath = strings.Trim(remotePath, "/") if remotePath != "" && remotePath != "." { p += "/" + remotePath } u.Path = p return u.String() } // propfindBody PROPFIND 请求体:只取需要的属性。 const propfindBody = `` + `` + `` + `` // davMultistatus 207 Multi-Status XML 解析结构(标签名与命名空间无关匹配)。 type davMultistatus struct { Responses []struct { Href string `xml:"href"` Propstat []struct { Status string `xml:"status"` Prop struct { ContentLength int64 `xml:"getcontentlength"` ContentType string `xml:"getcontenttype"` ResourceType struct { Collection *struct{} `xml:"collection"` } `xml:"resourcetype"` } `xml:"prop"` } `xml:"propstat"` } `xml:"response"` } // firstProp 取第一个 HTTP 2xx 状态的属性块。 func (m *davMultistatus) firstProp() (length int64, ctype string, isDir bool, ok bool) { for _, r := range m.Responses { for _, ps := range r.Propstat { if !strings.Contains(ps.Status, " 200 ") { continue } return ps.Prop.ContentLength, ps.Prop.ContentType, ps.Prop.ResourceType.Collection != nil, true } } return 0, "", false, false } // propfind 执行 PROPFIND 并解析 207 响应;404 时返回 (nil, nil)。 func (w *WebDAVStorage) propfind(ctx context.Context, rawURL string, depth string) (*davMultistatus, error) { resp, err := w.do(ctx, "PROPFIND", rawURL, requestOpts{ body: func() (io.Reader, int64, error) { return strings.NewReader(propfindBody), int64(len(propfindBody)), nil }, headers: map[string]string{"Depth": depth, "Content-Type": "application/xml"}, }) if err != nil { return nil, err } defer drainClose(resp) switch resp.StatusCode { case http.StatusMultiStatus, http.StatusOK: body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) if err != nil { return nil, fmt.Errorf("%w: PROPFIND 读取失败: %v", ErrUnavailable, err) } var ms davMultistatus if err := xml.Unmarshal(body, &ms); err != nil { return nil, fmt.Errorf("%w: PROPFIND XML 解析失败: %v", ErrUnavailable, err) } return &ms, nil case http.StatusNotFound: return nil, nil default: return nil, fmt.Errorf("%w: PROPFIND %s → %d", ErrUnavailable, rawURL, resp.StatusCode) } } // remoteExists PROPFIND 探测远端路径存在性。 func (w *WebDAVStorage) remoteExists(ctx context.Context, remotePath string) (bool, error) { ms, err := w.propfind(ctx, w.urlFor(remotePath), "0") if err != nil { return false, err } return ms != nil, nil } // markDir 记录已确认存在的目录(避免重复 PROPFIND/MKCOL 往返)。 func (w *WebDAVStorage) markDir(remotePath string) { w.dirMu.Lock() defer w.dirMu.Unlock() w.knownDirs[remotePath] = struct{}{} } // unmarkDir 目录被删除时移除缓存。 func (w *WebDAVStorage) unmarkDir(remotePath string) { w.dirMu.Lock() defer w.dirMu.Unlock() delete(w.knownDirs, remotePath) } // isMarkedDir 查询目录缓存。 func (w *WebDAVStorage) isMarkedDir(remotePath string) bool { w.dirMu.RLock() defer w.dirMu.RUnlock() _, ok := w.knownDirs[remotePath] return ok } // ensureDirs 按需逐级创建远端目录(含根前缀;MKCOL 级联,成功后写缓存)。 func (w *WebDAVStorage) ensureDirs(ctx context.Context, remotePath string) error { segments := splitRemoteSegments(remotePath) cur := "" for _, seg := range segments { cur = path.Join(cur, seg) if w.isMarkedDir(cur) { continue } exists, err := w.remoteExists(ctx, cur) if err == nil && exists { w.markDir(cur) continue } if err != nil && !errors.Is(err, ErrNotFound) { return err } resp, err := w.do(ctx, "MKCOL", w.urlFor(cur), requestOpts{}) if err != nil { return err } status := resp.StatusCode drainClose(resp) // 201 创建成功;405 已存在;其余视为失败(409 通常因父目录缺失,理论上不会出现)。 if status == http.StatusCreated || status == http.StatusOK || status == http.StatusNoContent || status == http.StatusMethodNotAllowed { w.markDir(cur) continue } return fmt.Errorf("%w: MKCOL %s → %d", ErrUnavailable, cur, status) } return nil } // splitRemoteSegments 拆分远端路径段。 func splitRemoteSegments(p string) []string { p = strings.Trim(strings.ReplaceAll(p, "\\", "/"), "/") if p == "" { return nil } return strings.Split(p, "/") } // deleteEmptyParents 删除空父目录(含根前缀,但不删根目录本身);尽力而为。 func (w *WebDAVStorage) deleteEmptyParents(ctx context.Context, remotePath string) { dir := path.Dir(remotePath) for dir != "" && dir != "." && dir != w.root && strings.HasPrefix(dir+"/", w.root+"/") { ms, err := w.propfind(ctx, w.urlFor(dir), "1") if err != nil || ms == nil { return } if len(ms.Responses) > 1 { // 非空(自身 + 子项) return } resp, err := w.do(ctx, http.MethodDelete, w.urlFor(dir), requestOpts{}) if err != nil { return } ok := resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNoContent drainClose(resp) if !ok { return } w.unmarkDir(dir) dir = path.Dir(dir) } } // putFile PUT 上传:body 工厂每次尝试返回可重放的读取器。 func (w *WebDAVStorage) putFile(ctx context.Context, rawURL string, body func() (io.Reader, int64, error), retryBody bool) (*davResponse, error) { return w.do(ctx, http.MethodPut, rawURL, requestOpts{ body: body, retryBody: retryBody, headers: map[string]string{"Content-Type": "application/octet-stream"}, streaming: true, }) } // checkPutStatus 校验 PUT 响应状态。 func checkPutStatus(resp *davResponse, op string) error { switch resp.StatusCode { case http.StatusOK, http.StatusCreated, http.StatusNoContent: drainClose(resp) return nil default: msg := readErrorBody(resp) return fmt.Errorf("%w: %s → %d %s", ErrUnavailable, op, resp.StatusCode, msg) } } // readErrorBody 读取错误响应前 200 字节并释放连接。 func readErrorBody(resp *davResponse) string { if resp == nil || resp.Body == nil { return "" } b, _ := io.ReadAll(io.LimitReader(resp.Body, 200)) _ = resp.Body.Close() if resp.cancel != nil { resp.cancel() } return strings.TrimSpace(string(b)) } // SaveFile 流式保存(PUT):按需建目录,seekable 源可安全重试。 func (w *WebDAVStorage) SaveFile(ctx context.Context, r io.Reader, savePath string) (int64, error) { remote, err := w.joinRemote(savePath) if err != nil { return 0, err } if err := w.ensureDirs(ctx, path.Dir(remote)); err != nil { return 0, err } // 可重放判定:seekable 源失败后可从头重传(PUT 覆盖语义保证最终一致)。 seeker, seekable := r.(io.Seeker) var knownLen int64 = -1 if seekable { if cur, err := seeker.Seek(0, io.SeekCurrent); err == nil { if end, err := seeker.Seek(0, io.SeekEnd); err == nil { knownLen = end - cur _, _ = seeker.Seek(cur, io.SeekStart) } } } src := &countingReader{r: r} body := func() (io.Reader, int64, error) { if seekable { if _, err := seeker.Seek(0, io.SeekStart); err != nil { return nil, -1, err } src.reset() } return src, knownLen, nil } resp, err := w.putFile(ctx, w.urlFor(remote), body, seekable) if err != nil { return src.count(), err } if err := checkPutStatus(resp, fmt.Sprintf("PUT %s", remote)); err != nil { return src.count(), err } return src.count(), nil } // DeleteFile DELETE 文件 + 尽力清理空父目录。 func (w *WebDAVStorage) DeleteFile(ctx context.Context, savePath string) error { remote, err := w.joinRemote(savePath) if err != nil { return err } resp, err := w.do(ctx, http.MethodDelete, w.urlFor(remote), requestOpts{}) if err != nil { return err } switch resp.StatusCode { case http.StatusOK, http.StatusNoContent, http.StatusNotFound: drainClose(resp) default: msg := readErrorBody(resp) return fmt.Errorf("%w: DELETE %s → %d %s", ErrUnavailable, remote, resp.StatusCode, msg) } w.deleteEmptyParents(ctx, remote) return nil } // Open 打开下载流:Range 透传,io.Pipe 流式转发不落盘,ctx 可取消。 func (w *WebDAVStorage) Open(ctx context.Context, savePath string, rng *Range) (*Download, error) { remote, err := w.joinRemote(savePath) if err != nil { return nil, err } opts := requestOpts{streaming: true} if rng != nil { opts.headers = map[string]string{"Range": rangeHeaderValue(rng)} } resp, err := w.do(ctx, http.MethodGet, w.urlFor(remote), opts) if err != nil { return nil, err } switch resp.StatusCode { case http.StatusOK, http.StatusPartialContent: // 正常,继续 case http.StatusNotFound: drainClose(resp) return nil, ErrNotFound case http.StatusRequestedRangeNotSatisfiable: drainClose(resp) return nil, ErrRangeNotSatisfiable default: msg := readErrorBody(resp) return nil, fmt.Errorf("%w: GET %s → %d %s", ErrUnavailable, remote, resp.StatusCode, msg) } total := resp.ContentLength start, end := int64(0), total-1 if resp.StatusCode == http.StatusPartialContent { if cr := resp.Header.Get("Content-Range"); cr != "" { if s0, e0, t0, ok := parseContentRange(cr); ok { start, end = s0, e0 if t0 >= 0 { total = t0 } } } } if total < 0 { // 服务端未给出长度(chunked):按未知大小处理 start, end, total = 0, -1, -1 } if rng == nil { // 对齐契约:完整文件 Start=0、End=Total-1 start, end = 0, total-1 } if end < 0 { // 空文件或未知大小:End 未知语义 end = -1 } // io.Pipe 流式桥接:HTTP 响应体 → 管道 → 调用方,全程不落盘; // 调用方提前 Close 或 ctx 取消都会终止拷贝并释放连接。 body := resp.Body pr, pw := io.Pipe() go func() { bufp, _ := w.spacesPool.Get().(*[]byte) _, copyErr := io.CopyBuffer(pw, body, *bufp) w.spacesPool.Put(bufp) _ = body.Close() pw.CloseWithError(copyErr) // copyErr 为 nil 时写入 EOF }() context.AfterFunc(ctx, func() { _ = pw.CloseWithError(ctx.Err()) }) contentType := resp.Header.Get("Content-Type") return &Download{ ReadCloser: pr, Start: start, End: end, Total: total, Meta: FileMeta{ Size: total, ContentType: contentType, AcceptRanges: true, }, }, nil } // Stat PROPFIND Depth 0 获取元信息。 func (w *WebDAVStorage) Stat(ctx context.Context, savePath string) (*FileMeta, error) { remote, err := w.joinRemote(savePath) if err != nil { return nil, err } ms, err := w.propfind(ctx, w.urlFor(remote), "0") if err != nil { if errors.Is(err, ErrNotFound) { return nil, ErrNotFound } return nil, err } if ms == nil { return nil, ErrNotFound } length, ctype, _, ok := ms.firstProp() if !ok { return nil, ErrNotFound } return &FileMeta{Size: length, ContentType: ctype, AcceptRanges: true}, nil } // HeadMeta 读取文件元信息与前 n 字节(WebDAV 实现:PROPFIND + Range GET)。 func (w *WebDAVStorage) HeadMeta(ctx context.Context, savePath string, headBytes int64) (*FileMeta, []byte, error) { meta, err := w.Stat(ctx, savePath) if err != nil { return nil, nil, err } if headBytes <= 0 { return meta, nil, nil } dl, err := w.Open(ctx, savePath, &Range{Start: 0, End: headBytes - 1}) if err != nil { if errors.Is(err, ErrRangeNotSatisfiable) { // 空文件等边界:返回空头 return meta, nil, nil } return nil, nil, err } defer func() { _ = dl.Close() }() head := make([]byte, headBytes) n, _ := io.ReadFull(dl.ReadCloser, head) return meta, head[:n], nil } // SaveChunk 保存分片:落临时文件获得精确长度与可重放 body,PUT 到分片路径。 func (w *WebDAVStorage) SaveChunk(ctx context.Context, uploadID string, chunkIndex int, r io.Reader, savePath string) (int64, error) { if _, err := w.joinRemote(savePath); err != nil { return 0, err } chunkRel, ok := SanitizePath(ChunkPartPath(savePath, uploadID, chunkIndex)) if !ok { return 0, fmt.Errorf("%w: 分片路径非法 %q", ErrInvalidPath, savePath) } remote := path.Join(w.root, chunkRel) if err := w.ensureDirs(ctx, path.Dir(remote)); err != nil { return 0, err } // 分片体积有限(默认 ≤8MB):落临时文件换取精确 Content-Length 与可重试性。 tmp, err := os.CreateTemp("", "fcb-webdav-chunk-*") if err != nil { return 0, fmt.Errorf("storage/webdav: 创建临时文件失败: %w", err) } tmpName := tmp.Name() defer func() { _ = tmp.Close(); _ = os.Remove(tmpName) }() size, err := io.CopyBuffer(tmp, r, make([]byte, localChunkSize)) if err != nil { return 0, fmt.Errorf("storage/webdav: 缓存分片失败: %w", err) } if _, err := tmp.Seek(0, io.SeekStart); err != nil { return 0, fmt.Errorf("storage/webdav: 回卷分片失败: %w", err) } body := func() (io.Reader, int64, error) { if _, err := tmp.Seek(0, io.SeekStart); err != nil { return nil, -1, err } return tmp, size, nil } resp, err := w.putFile(ctx, w.urlFor(remote), body, true) if err != nil { return size, err } if err := checkPutStatus(resp, fmt.Sprintf("PUT 分片 %s", remote)); err != nil { return size, err } return size, nil } // MergeChunks 合并 WebDAV 分片: // 逐分片 GET 流式拼入本地临时文件(边拷贝边校验哈希)→ PUT 上传目标 → 清理远端分片与本地临时文件。 // 说明:WebDAV 无服务端聚合能力,合并必须经服务端中转;临时文件仅用于拼接与重试,最终 PUT 可重放。 func (w *WebDAVStorage) MergeChunks(ctx context.Context, uploadID string, total int, verifyHash func(index int) (string, error), savePath string) (int64, string, error) { if total <= 0 { return 0, "", fmt.Errorf("storage/webdav: 非法分片总数 %d", total) } remote, err := w.joinRemote(savePath) if err != nil { return 0, "", err } if err := w.ensureDirs(ctx, path.Dir(remote)); err != nil { return 0, "", err } tmp, err := os.CreateTemp("", "fcb-webdav-merge-*") if err != nil { return 0, "", fmt.Errorf("storage/webdav: 创建合并临时文件失败: %w", err) } tmpName := tmp.Name() defer func() { _ = tmp.Close(); _ = os.Remove(tmpName) }() totalHash := sha256.New() var size int64 for i := 0; i < total; i++ { if err := ctx.Err(); err != nil { return 0, "", err } chunkRel, ok := SanitizePath(ChunkPartPath(savePath, uploadID, i)) if !ok { return 0, "", fmt.Errorf("%w: 分片 %d 路径非法", ErrInvalidPath, i) } resp, err := w.do(ctx, http.MethodGet, w.urlFor(path.Join(w.root, chunkRel)), requestOpts{streaming: true}) if err != nil { return 0, "", fmt.Errorf("storage/webdav: 读取分片 %d 失败: %w", i, err) } if resp.StatusCode == http.StatusNotFound { drainClose(resp) return 0, "", fmt.Errorf("storage/webdav: 分片 %d 不存在", i) } if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent { msg := readErrorBody(resp) return 0, "", fmt.Errorf("storage/webdav: 读取分片 %d → %d %s", i, resp.StatusCode, msg) } chunkHash := sha256.New() n, err := io.CopyBuffer(io.MultiWriter(tmp, totalHash, chunkHash), resp.Body, make([]byte, localChunkSize)) _ = resp.Body.Close() if err != nil { return 0, "", fmt.Errorf("storage/webdav: 拼接分片 %d 失败: %w", i, err) } if verifyHash != nil { expected, err := verifyHash(i) if err != nil { return 0, "", err } if expected != "" && expected != hex.EncodeToString(chunkHash.Sum(nil)) { return 0, "", fmt.Errorf("%w: 分片 %d 期望 %s", ErrHashMismatch, i, expected) } } size += n } if _, err := tmp.Seek(0, io.SeekStart); err != nil { return 0, "", fmt.Errorf("storage/webdav: 回卷合并文件失败: %w", err) } body := func() (io.Reader, int64, error) { if _, err := tmp.Seek(0, io.SeekStart); err != nil { return nil, -1, err } return tmp, size, nil } resp, err := w.putFile(ctx, w.urlFor(remote), body, true) if err != nil { return size, "", err } if err := checkPutStatus(resp, fmt.Sprintf("PUT 合并 %s", remote)); err != nil { return size, "", err } // 合并成功后清理远端分片目录与本地临时文件(defer 兜底删除本地文件)。 _ = w.CleanChunks(ctx, uploadID, savePath) return size, hex.EncodeToString(totalHash.Sum(nil)), nil } // CleanChunks 递归删除远端分片目录(RFC 4918 DELETE 对 collection 递归)。 func (w *WebDAVStorage) CleanChunks(ctx context.Context, uploadID string, savePath string) error { dirRel, ok := SanitizePath(chunkDirOf(savePath, uploadID)) if !ok { return fmt.Errorf("%w: 分片目录非法 %q", ErrInvalidPath, savePath) } remote := path.Join(w.root, dirRel) resp, err := w.do(ctx, http.MethodDelete, w.urlFor(remote), requestOpts{}) if err != nil { return err } switch resp.StatusCode { case http.StatusOK, http.StatusNoContent, http.StatusNotFound: drainClose(resp) w.unmarkDir(remote) default: msg := readErrorBody(resp) return fmt.Errorf("%w: 清理分片目录 %s → %d %s", ErrUnavailable, remote, resp.StatusCode, msg) } return nil } // FileExists PROPFIND 探测存在性;非法路径按不存在处理。 func (w *WebDAVStorage) FileExists(ctx context.Context, savePath string) (bool, error) { remote, err := w.joinRemote(savePath) if err != nil { return false, nil } return w.remoteExists(ctx, remote) } // PresignGetURL WebDAV 无预签名直链能力。 func (w *WebDAVStorage) PresignGetURL(ctx context.Context, savePath string, expires int64) (string, error) { return "", ErrNotSupported } // PresignPutURL WebDAV 无预签名直传能力。 func (w *WebDAVStorage) PresignPutURL(ctx context.Context, savePath string, expires int64) (string, error) { return "", ErrNotSupported } // HealthCheck 健康检查:PROPFIND 根目录;不存在时 MKCOL 创建(启动自愈)。 // 同时完成凭据与连通性验证(do 内 401 协商)。 func (w *WebDAVStorage) HealthCheck(ctx context.Context) error { exists, err := w.remoteExists(ctx, w.root) if err == nil && exists { w.markDir(w.root) return nil } if err != nil && !errors.Is(err, ErrNotFound) { return fmt.Errorf("%w: WebDAV 健康检查失败: %v", ErrUnavailable, err) } resp, err := w.do(ctx, "MKCOL", w.urlFor(w.root), requestOpts{}) if err != nil { return err } switch resp.StatusCode { case http.StatusCreated, http.StatusOK, http.StatusNoContent, http.StatusMethodNotAllowed: drainClose(resp) w.markDir(w.root) return nil default: msg := readErrorBody(resp) return fmt.Errorf("%w: WebDAV 根目录创建失败 → %d %s", ErrUnavailable, resp.StatusCode, msg) } } // 接口编译期断言。 var _ Storage = (*WebDAVStorage)(nil)