/*
 * Minio Cloud Storage, (C) 2015 Minio, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package xl

import (
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"encoding/xml"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/minio/minio/pkg/crypto/sha256"
	"github.com/minio/minio/pkg/crypto/sha512"
	"github.com/minio/minio/pkg/probe"
	signV4 "github.com/minio/minio/pkg/signature"
	"github.com/minio/minio/pkg/xl/block"
)

// config files used inside XL
const (
	// bucket, object metadata
	bucketMetadataConfig = "bucketMetadata.json"
	objectMetadataConfig = "objectMetadata.json"

	// versions
	objectMetadataVersion = "1.0.0"
	bucketMetadataVersion = "1.0.0"
)

/// v1 API functions

// makeBucket - make a new bucket
func (xl API) makeBucket(bucket string, acl BucketACL) *probe.Error {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return probe.NewError(InvalidArgument{})
	}
	return xl.makeXLBucket(bucket, acl.String())
}

// getBucketMetadata - get bucket metadata
func (xl API) getBucketMetadata(bucketName string) (BucketMetadata, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return BucketMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucketName]; !ok {
		return BucketMetadata{}, probe.NewError(BucketNotFound{Bucket: bucketName})
	}
	metadata, err := xl.getXLBucketMetadata()
	if err != nil {
		return BucketMetadata{}, err.Trace()
	}
	return metadata.Buckets[bucketName], nil
}

// setBucketMetadata - set bucket metadata
func (xl API) setBucketMetadata(bucketName string, bucketMetadata map[string]string) *probe.Error {
	if err := xl.listXLBuckets(); err != nil {
		return err.Trace()
	}
	metadata, err := xl.getXLBucketMetadata()
	if err != nil {
		return err.Trace()
	}
	oldBucketMetadata := metadata.Buckets[bucketName]
	acl, ok := bucketMetadata["acl"]
	if !ok {
		return probe.NewError(InvalidArgument{})
	}
	oldBucketMetadata.ACL = BucketACL(acl)
	metadata.Buckets[bucketName] = oldBucketMetadata
	return xl.setXLBucketMetadata(metadata)
}

// listBuckets - return list of buckets
func (xl API) listBuckets() (map[string]BucketMetadata, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return nil, err.Trace()
	}
	metadata, err := xl.getXLBucketMetadata()
	if err != nil {
		// intentionally left out the error when XL is empty
		// but we need to revisit this area in future - since we need
		// to figure out between acceptable and unacceptable errors
		return make(map[string]BucketMetadata), nil
	}
	if metadata == nil {
		return make(map[string]BucketMetadata), nil
	}
	return metadata.Buckets, nil
}

// listObjects - return list of objects
func (xl API) listObjects(bucket, prefix, marker, delimiter string, maxkeys int) (ListObjectsResults, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return ListObjectsResults{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return ListObjectsResults{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	listObjects, err := xl.buckets[bucket].ListObjects(prefix, marker, delimiter, maxkeys)
	if err != nil {
		return ListObjectsResults{}, err.Trace()
	}
	return listObjects, nil
}

// putObject - put object
func (xl API) putObject(bucket, object, expectedMD5Sum string, reader io.Reader, size int64, metadata map[string]string, signature *signV4.Signature) (ObjectMetadata, *probe.Error) {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return ObjectMetadata{}, probe.NewError(InvalidArgument{})
	}
	if object == "" || strings.TrimSpace(object) == "" {
		return ObjectMetadata{}, probe.NewError(InvalidArgument{})
	}
	if err := xl.listXLBuckets(); err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return ObjectMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	bucketMeta, err := xl.getXLBucketMetadata()
	if err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	if _, ok := bucketMeta.Buckets[bucket].BucketObjects[object]; ok {
		return ObjectMetadata{}, probe.NewError(ObjectExists{Object: object})
	}
	objMetadata, err := xl.buckets[bucket].WriteObject(object, reader, size, expectedMD5Sum, metadata, signature)
	if err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	bucketMeta.Buckets[bucket].BucketObjects[object] = struct{}{}
	if err := xl.setXLBucketMetadata(bucketMeta); err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	return objMetadata, nil
}

// putObject - put object
func (xl API) putObjectPart(bucket, object, expectedMD5Sum, uploadID string, partID int, reader io.Reader, size int64, metadata map[string]string, signature *signV4.Signature) (PartMetadata, *probe.Error) {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return PartMetadata{}, probe.NewError(InvalidArgument{})
	}
	if object == "" || strings.TrimSpace(object) == "" {
		return PartMetadata{}, probe.NewError(InvalidArgument{})
	}
	if err := xl.listXLBuckets(); err != nil {
		return PartMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return PartMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	bucketMeta, err := xl.getXLBucketMetadata()
	if err != nil {
		return PartMetadata{}, err.Trace()
	}
	if _, ok := bucketMeta.Buckets[bucket].Multiparts[object]; !ok {
		return PartMetadata{}, probe.NewError(InvalidUploadID{UploadID: uploadID})
	}
	if _, ok := bucketMeta.Buckets[bucket].BucketObjects[object]; ok {
		return PartMetadata{}, probe.NewError(ObjectExists{Object: object})
	}
	objectPart := object + "/" + "multipart" + "/" + strconv.Itoa(partID)
	objmetadata, err := xl.buckets[bucket].WriteObject(objectPart, reader, size, expectedMD5Sum, metadata, signature)
	if err != nil {
		return PartMetadata{}, err.Trace()
	}
	partMetadata := PartMetadata{
		PartNumber:   partID,
		LastModified: objmetadata.Created,
		ETag:         objmetadata.MD5Sum,
		Size:         objmetadata.Size,
	}
	multipartSession := bucketMeta.Buckets[bucket].Multiparts[object]
	multipartSession.Parts[strconv.Itoa(partID)] = partMetadata
	bucketMeta.Buckets[bucket].Multiparts[object] = multipartSession
	if err := xl.setXLBucketMetadata(bucketMeta); err != nil {
		return PartMetadata{}, err.Trace()
	}
	return partMetadata, nil
}

// getObject - get object
func (xl API) getObject(bucket, object string) (reader io.ReadCloser, size int64, err *probe.Error) {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return nil, 0, probe.NewError(InvalidArgument{})
	}
	if object == "" || strings.TrimSpace(object) == "" {
		return nil, 0, probe.NewError(InvalidArgument{})
	}
	if err := xl.listXLBuckets(); err != nil {
		return nil, 0, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return nil, 0, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	return xl.buckets[bucket].ReadObject(object)
}

// getObjectMetadata - get object metadata
func (xl API) getObjectMetadata(bucket, object string) (ObjectMetadata, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return ObjectMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	bucketMeta, err := xl.getXLBucketMetadata()
	if err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	if _, ok := bucketMeta.Buckets[bucket].BucketObjects[object]; !ok {
		return ObjectMetadata{}, probe.NewError(ObjectNotFound{Object: object})
	}
	objectMetadata, err := xl.buckets[bucket].GetObjectMetadata(object)
	if err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	return objectMetadata, nil
}

// newMultipartUpload - new multipart upload request
func (xl API) newMultipartUpload(bucket, object, contentType string) (string, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return "", err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return "", probe.NewError(BucketNotFound{Bucket: bucket})
	}
	allbuckets, err := xl.getXLBucketMetadata()
	if err != nil {
		return "", err.Trace()
	}
	bucketMetadata := allbuckets.Buckets[bucket]
	multiparts := make(map[string]MultiPartSession)
	if len(bucketMetadata.Multiparts) > 0 {
		multiparts = bucketMetadata.Multiparts
	}

	id := []byte(strconv.Itoa(rand.Int()) + bucket + object + time.Now().String())
	uploadIDSum := sha512.Sum512(id)
	uploadID := base64.URLEncoding.EncodeToString(uploadIDSum[:])[:47]

	multipartSession := MultiPartSession{
		UploadID:   uploadID,
		Initiated:  time.Now().UTC(),
		Parts:      make(map[string]PartMetadata),
		TotalParts: 0,
	}
	multiparts[object] = multipartSession
	bucketMetadata.Multiparts = multiparts
	allbuckets.Buckets[bucket] = bucketMetadata

	if err := xl.setXLBucketMetadata(allbuckets); err != nil {
		return "", err.Trace()
	}

	return uploadID, nil
}

// listObjectParts list all object parts
func (xl API) listObjectParts(bucket, object string, resources ObjectResourcesMetadata) (ObjectResourcesMetadata, *probe.Error) {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return ObjectResourcesMetadata{}, probe.NewError(InvalidArgument{})
	}
	if object == "" || strings.TrimSpace(object) == "" {
		return ObjectResourcesMetadata{}, probe.NewError(InvalidArgument{})
	}
	if err := xl.listXLBuckets(); err != nil {
		return ObjectResourcesMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return ObjectResourcesMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	allBuckets, err := xl.getXLBucketMetadata()
	if err != nil {
		return ObjectResourcesMetadata{}, err.Trace()
	}
	bucketMetadata := allBuckets.Buckets[bucket]
	if _, ok := bucketMetadata.Multiparts[object]; !ok {
		return ObjectResourcesMetadata{}, probe.NewError(InvalidUploadID{UploadID: resources.UploadID})
	}
	if bucketMetadata.Multiparts[object].UploadID != resources.UploadID {
		return ObjectResourcesMetadata{}, probe.NewError(InvalidUploadID{UploadID: resources.UploadID})
	}
	objectResourcesMetadata := resources
	objectResourcesMetadata.Bucket = bucket
	objectResourcesMetadata.Key = object
	var parts []*PartMetadata
	var startPartNumber int
	switch {
	case objectResourcesMetadata.PartNumberMarker == 0:
		startPartNumber = 1
	default:
		startPartNumber = objectResourcesMetadata.PartNumberMarker
	}
	for i := startPartNumber; i <= bucketMetadata.Multiparts[object].TotalParts; i++ {
		if len(parts) > objectResourcesMetadata.MaxParts {
			sort.Sort(partNumber(parts))
			objectResourcesMetadata.IsTruncated = true
			objectResourcesMetadata.Part = parts
			objectResourcesMetadata.NextPartNumberMarker = i
			return objectResourcesMetadata, nil
		}
		part, ok := bucketMetadata.Multiparts[object].Parts[strconv.Itoa(i)]
		if !ok {
			return ObjectResourcesMetadata{}, probe.NewError(InvalidPart{})
		}
		parts = append(parts, &part)
	}
	sort.Sort(partNumber(parts))
	objectResourcesMetadata.Part = parts
	return objectResourcesMetadata, nil
}

// completeMultipartUpload complete an incomplete multipart upload
func (xl API) completeMultipartUpload(bucket, object, uploadID string, data io.Reader, signature *signV4.Signature) (ObjectMetadata, *probe.Error) {
	if bucket == "" || strings.TrimSpace(bucket) == "" {
		return ObjectMetadata{}, probe.NewError(InvalidArgument{})
	}
	if object == "" || strings.TrimSpace(object) == "" {
		return ObjectMetadata{}, probe.NewError(InvalidArgument{})
	}
	if err := xl.listXLBuckets(); err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return ObjectMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	allBuckets, err := xl.getXLBucketMetadata()
	if err != nil {
		return ObjectMetadata{}, err.Trace()
	}
	bucketMetadata := allBuckets.Buckets[bucket]
	if _, ok := bucketMetadata.Multiparts[object]; !ok {
		return ObjectMetadata{}, probe.NewError(InvalidUploadID{UploadID: uploadID})
	}
	if bucketMetadata.Multiparts[object].UploadID != uploadID {
		return ObjectMetadata{}, probe.NewError(InvalidUploadID{UploadID: uploadID})
	}
	var partBytes []byte
	{
		var err error
		partBytes, err = ioutil.ReadAll(data)
		if err != nil {
			return ObjectMetadata{}, probe.NewError(err)
		}
	}
	if signature != nil {
		partHashBytes := sha256.Sum256(partBytes)
		ok, err := signature.DoesSignatureMatch(hex.EncodeToString(partHashBytes[:]))
		if err != nil {
			return ObjectMetadata{}, err.Trace()
		}
		if !ok {
			return ObjectMetadata{}, probe.NewError(signV4.SigDoesNotMatch{})
		}
	}
	parts := &CompleteMultipartUpload{}
	if err := xml.Unmarshal(partBytes, parts); err != nil {
		return ObjectMetadata{}, probe.NewError(MalformedXML{})
	}
	if !sort.IsSorted(completedParts(parts.Part)) {
		return ObjectMetadata{}, probe.NewError(InvalidPartOrder{})
	}
	for _, part := range parts.Part {
		if strings.Trim(part.ETag, "\"") != bucketMetadata.Multiparts[object].Parts[strconv.Itoa(part.PartNumber)].ETag {
			return ObjectMetadata{}, probe.NewError(InvalidPart{})
		}
	}
	var finalETagBytes []byte
	var finalSize int64
	totalParts := strconv.Itoa(bucketMetadata.Multiparts[object].TotalParts)
	for _, part := range bucketMetadata.Multiparts[object].Parts {
		partETagBytes, err := hex.DecodeString(part.ETag)
		if err != nil {
			return ObjectMetadata{}, probe.NewError(err)
		}
		finalETagBytes = append(finalETagBytes, partETagBytes...)
		finalSize += part.Size
	}
	finalETag := hex.EncodeToString(finalETagBytes)
	objMetadata := ObjectMetadata{}
	objMetadata.MD5Sum = finalETag + "-" + totalParts
	objMetadata.Object = object
	objMetadata.Bucket = bucket
	objMetadata.Size = finalSize
	objMetadata.Created = bucketMetadata.Multiparts[object].Parts[totalParts].LastModified
	return objMetadata, nil
}

// listMultipartUploads list all multipart uploads
func (xl API) listMultipartUploads(bucket string, resources BucketMultipartResourcesMetadata) (BucketMultipartResourcesMetadata, *probe.Error) {
	if err := xl.listXLBuckets(); err != nil {
		return BucketMultipartResourcesMetadata{}, err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return BucketMultipartResourcesMetadata{}, probe.NewError(BucketNotFound{Bucket: bucket})
	}
	allbuckets, err := xl.getXLBucketMetadata()
	if err != nil {
		return BucketMultipartResourcesMetadata{}, err.Trace()
	}
	bucketMetadata := allbuckets.Buckets[bucket]
	var uploads []*UploadMetadata
	for key, session := range bucketMetadata.Multiparts {
		if strings.HasPrefix(key, resources.Prefix) {
			if len(uploads) > resources.MaxUploads {
				sort.Sort(byKey(uploads))
				resources.Upload = uploads
				resources.NextKeyMarker = key
				resources.NextUploadIDMarker = session.UploadID
				resources.IsTruncated = true
				return resources, nil
			}
			// uploadIDMarker is ignored if KeyMarker is empty
			switch {
			case resources.KeyMarker != "" && resources.UploadIDMarker == "":
				if key > resources.KeyMarker {
					upload := new(UploadMetadata)
					upload.Key = key
					upload.UploadID = session.UploadID
					upload.Initiated = session.Initiated
					uploads = append(uploads, upload)
				}
			case resources.KeyMarker != "" && resources.UploadIDMarker != "":
				if session.UploadID > resources.UploadIDMarker {
					if key >= resources.KeyMarker {
						upload := new(UploadMetadata)
						upload.Key = key
						upload.UploadID = session.UploadID
						upload.Initiated = session.Initiated
						uploads = append(uploads, upload)
					}
				}
			default:
				upload := new(UploadMetadata)
				upload.Key = key
				upload.UploadID = session.UploadID
				upload.Initiated = session.Initiated
				uploads = append(uploads, upload)
			}
		}
	}
	sort.Sort(byKey(uploads))
	resources.Upload = uploads
	return resources, nil
}

// abortMultipartUpload - abort a incomplete multipart upload
func (xl API) abortMultipartUpload(bucket, object, uploadID string) *probe.Error {
	if err := xl.listXLBuckets(); err != nil {
		return err.Trace()
	}
	if _, ok := xl.buckets[bucket]; !ok {
		return probe.NewError(BucketNotFound{Bucket: bucket})
	}
	allbuckets, err := xl.getXLBucketMetadata()
	if err != nil {
		return err.Trace()
	}
	bucketMetadata := allbuckets.Buckets[bucket]
	if _, ok := bucketMetadata.Multiparts[object]; !ok {
		return probe.NewError(InvalidUploadID{UploadID: uploadID})
	}
	if bucketMetadata.Multiparts[object].UploadID != uploadID {
		return probe.NewError(InvalidUploadID{UploadID: uploadID})
	}
	delete(bucketMetadata.Multiparts, object)

	allbuckets.Buckets[bucket] = bucketMetadata
	if err := xl.setXLBucketMetadata(allbuckets); err != nil {
		return err.Trace()
	}

	return nil
}

//// internal functions

// getBucketMetadataWriters -
func (xl API) getBucketMetadataWriters() ([]io.WriteCloser, *probe.Error) {
	var writers []io.WriteCloser
	for _, node := range xl.nodes {
		disks, err := node.ListDisks()
		if err != nil {
			return nil, err.Trace()
		}
		writers = make([]io.WriteCloser, len(disks))
		for order, disk := range disks {
			bucketMetaDataWriter, err := disk.CreateFile(filepath.Join(xl.config.XLName, bucketMetadataConfig))
			if err != nil {
				return nil, err.Trace()
			}
			writers[order] = bucketMetaDataWriter
		}
	}
	return writers, nil
}

// getBucketMetadataReaders - readers are returned in map rather than slice
func (xl API) getBucketMetadataReaders() (map[int]io.ReadCloser, *probe.Error) {
	readers := make(map[int]io.ReadCloser)
	disks := make(map[int]block.Block)
	var err *probe.Error
	for _, node := range xl.nodes {
		nDisks := make(map[int]block.Block)
		nDisks, err = node.ListDisks()
		if err != nil {
			return nil, err.Trace()
		}
		for k, v := range nDisks {
			disks[k] = v
		}
	}
	var bucketMetaDataReader io.ReadCloser
	for order, disk := range disks {
		bucketMetaDataReader, err = disk.Open(filepath.Join(xl.config.XLName, bucketMetadataConfig))
		if err != nil {
			continue
		}
		readers[order] = bucketMetaDataReader
	}
	if err != nil {
		return nil, err.Trace()
	}
	return readers, nil
}

// setXLBucketMetadata -
func (xl API) setXLBucketMetadata(metadata *AllBuckets) *probe.Error {
	writers, err := xl.getBucketMetadataWriters()
	if err != nil {
		return err.Trace()
	}
	for _, writer := range writers {
		jenc := json.NewEncoder(writer)
		if err := jenc.Encode(metadata); err != nil {
			CleanupWritersOnError(writers)
			return probe.NewError(err)
		}
	}
	for _, writer := range writers {
		writer.Close()
	}
	return nil
}

// getXLBucketMetadata -
func (xl API) getXLBucketMetadata() (*AllBuckets, *probe.Error) {
	metadata := &AllBuckets{}
	readers, err := xl.getBucketMetadataReaders()
	if err != nil {
		return nil, err.Trace()
	}
	for _, reader := range readers {
		defer reader.Close()
	}
	{
		var err error
		for _, reader := range readers {
			jenc := json.NewDecoder(reader)
			if err = jenc.Decode(metadata); err == nil {
				return metadata, nil
			}
		}
		return nil, probe.NewError(err)
	}
}

// makeXLBucket -
func (xl API) makeXLBucket(bucketName, acl string) *probe.Error {
	if err := xl.listXLBuckets(); err != nil {
		return err.Trace()
	}
	if _, ok := xl.buckets[bucketName]; ok {
		return probe.NewError(BucketExists{Bucket: bucketName})
	}
	bkt, bucketMetadata, err := newBucket(bucketName, acl, xl.config.XLName, xl.nodes)
	if err != nil {
		return err.Trace()
	}
	nodeNumber := 0
	xl.buckets[bucketName] = bkt
	for _, node := range xl.nodes {
		disks := make(map[int]block.Block)
		disks, err = node.ListDisks()
		if err != nil {
			return err.Trace()
		}
		for order, disk := range disks {
			bucketSlice := fmt.Sprintf("%s$%d$%d", bucketName, nodeNumber, order)
			err := disk.MakeDir(filepath.Join(xl.config.XLName, bucketSlice))
			if err != nil {
				return err.Trace()
			}
		}
		nodeNumber = nodeNumber + 1
	}
	var metadata *AllBuckets
	metadata, err = xl.getXLBucketMetadata()
	if err != nil {
		if os.IsNotExist(err.ToGoError()) {
			metadata = new(AllBuckets)
			metadata.Buckets = make(map[string]BucketMetadata)
			metadata.Buckets[bucketName] = bucketMetadata
			err = xl.setXLBucketMetadata(metadata)
			if err != nil {
				return err.Trace()
			}
			return nil
		}
		return err.Trace()
	}
	metadata.Buckets[bucketName] = bucketMetadata
	err = xl.setXLBucketMetadata(metadata)
	if err != nil {
		return err.Trace()
	}
	return nil
}

// listXLBuckets -
func (xl API) listXLBuckets() *probe.Error {
	var disks map[int]block.Block
	var err *probe.Error
	for _, node := range xl.nodes {
		disks, err = node.ListDisks()
		if err != nil {
			return err.Trace()
		}
	}
	var dirs []os.FileInfo
	for _, disk := range disks {
		dirs, err = disk.ListDir(xl.config.XLName)
		if err == nil {
			break
		}
	}
	// if all disks are missing then return error
	if err != nil {
		return err.Trace()
	}
	for _, dir := range dirs {
		splitDir := strings.Split(dir.Name(), "$")
		if len(splitDir) < 3 {
			return probe.NewError(CorruptedBackend{Backend: dir.Name()})
		}
		bucketName := splitDir[0]
		// we dont need this once we cache from makeXLBucket()
		bkt, _, err := newBucket(bucketName, "private", xl.config.XLName, xl.nodes)
		if err != nil {
			return err.Trace()
		}
		xl.buckets[bucketName] = bkt
	}
	return nil
}