/*
    _____           _____   _____   ____          ______  _____  ------
   |     |  |      |     | |     | |     |     | |       |            |
   |     |  |      |     | |     | |     |     | |       |            |
   | --- |  |      |     | |-----| |---- |     | |-----| |-----  ------
   |     |  |      |     | |     | |     |     |       | |       |
   | ____|  |_____ | ____| | ____| |     |_____|  _____| |_____  |_____


   Licensed under the MIT License <http://opensource.org/licenses/MIT>.

   Copyright © 2020-2025 Microsoft Corporation. All rights reserved.
   Author : <blobfusedev@microsoft.com>

   Permission is hereby granted, free of charge, to any person obtaining a copy
   of this software and associated documentation files (the "Software"), to deal
   in the Software without restriction, including without limitation the rights
   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   copies of the Software, and to permit persons to whom the Software is
   furnished to do so, subject to the following conditions:

   The above copyright notice and this permission notice shall be included in all
   copies or substantial portions of the Software.

   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
   SOFTWARE
*/

package azstorage

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
	"github.com/Azure/azure-storage-fuse/v2/common"
	"github.com/Azure/azure-storage-fuse/v2/common/log"
	"github.com/Azure/azure-storage-fuse/v2/internal"
	"github.com/Azure/azure-storage-fuse/v2/internal/stats_manager"
	"github.com/vibhansa-msft/blobfilter"
)

const (
	folderKey           = "hdi_isfolder"
	symlinkKey          = "is_symlink"
	max_context_timeout = 5
)

type BlockBlob struct {
	AzStorageConnection
	Auth            azAuth
	Service         *service.Client
	Container       *container.Client
	blobCPKOpt      *blob.CPKInfo
	downloadOptions *blob.DownloadFileOptions
	listDetails     container.ListBlobsInclude
	blockLocks      common.KeyedMutex
}

// Verify that BlockBlob implements AzConnection interface
var _ AzConnection = &BlockBlob{}

const (
	MaxBlobSize = blockblob.MaxStageBlockBytes * blockblob.MaxBlocks
)

func (bb *BlockBlob) Configure(cfg AzStorageConfig) error {
	bb.Config = cfg

	if bb.Config.cpkEnabled {
		bb.blobCPKOpt = &blob.CPKInfo{
			EncryptionKey:       &bb.Config.cpkEncryptionKey,
			EncryptionKeySHA256: &bb.Config.cpkEncryptionKeySha256,
			EncryptionAlgorithm: to.Ptr(blob.EncryptionAlgorithmTypeAES256),
		}
	}

	bb.downloadOptions = &blob.DownloadFileOptions{
		BlockSize:   bb.Config.blockSize,
		Concurrency: bb.Config.maxConcurrency,
		CPKInfo:     bb.blobCPKOpt,
	}

	bb.listDetails = container.ListBlobsInclude{
		Metadata:    true,
		Deleted:     false,
		Snapshots:   false,
		Permissions: false, //Added to get permissions, acl, group, owner for HNS accounts
	}

	return nil
}

// For dynamic config update the config here
func (bb *BlockBlob) UpdateConfig(cfg AzStorageConfig) error {
	bb.Config.blockSize = cfg.blockSize
	bb.Config.maxConcurrency = cfg.maxConcurrency
	bb.Config.defaultTier = cfg.defaultTier
	bb.Config.ignoreAccessModifiers = cfg.ignoreAccessModifiers
	return nil
}

// UpdateServiceClient : Update the SAS specified by the user and create new service client
func (bb *BlockBlob) UpdateServiceClient(key, value string) (err error) {
	if key == "saskey" {
		bb.Auth.setOption(key, value)

		// get the service client with updated SAS
		svcClient, err := bb.Auth.getServiceClient(&bb.Config)
		if err != nil {
			log.Err("BlockBlob::UpdateServiceClient : Failed to get service client [%s]", err.Error())
			return err
		}

		// update the service client
		bb.Service = svcClient.(*service.Client)

		// Update the container client
		bb.Container = bb.Service.NewContainerClient(bb.Config.container)
	}
	return nil
}

// createServiceClient : Create the service client
func (bb *BlockBlob) createServiceClient() (*service.Client, error) {
	log.Trace("BlockBlob::createServiceClient : Getting service client")

	bb.Auth = getAzAuth(bb.Config.authConfig)
	if bb.Auth == nil {
		log.Err("BlockBlob::createServiceClient : Failed to retrieve auth object")
		return nil, fmt.Errorf("failed to retrieve auth object")
	}

	svcClient, err := bb.Auth.getServiceClient(&bb.Config)
	if err != nil {
		log.Err("BlockBlob::createServiceClient : Failed to get service client [%s]", err.Error())
		return nil, err
	}

	return svcClient.(*service.Client), nil
}

// SetupPipeline : Based on the config setup the ***URLs
func (bb *BlockBlob) SetupPipeline() error {
	log.Trace("BlockBlob::SetupPipeline : Setting up")
	var err error

	// create the service client
	bb.Service, err = bb.createServiceClient()
	if err != nil {
		log.Err("BlockBlob::SetupPipeline : Failed to get service client [%s]", err.Error())
		return err
	}

	// create the container client
	bb.Container = bb.Service.NewContainerClient(bb.Config.container)
	return nil
}

// TestPipeline : Validate the credentials specified in the auth config
func (bb *BlockBlob) TestPipeline() error {
	log.Trace("BlockBlob::TestPipeline : Validating")

	if bb.Config.mountAllContainers {
		return nil
	}

	if bb.Container == nil || bb.Container.URL() == "" {
		log.Err("BlockBlob::TestPipeline : Container Client is not built, check your credentials")
		return nil
	}

	includeFields := bb.listDetails
	if bb.listDetails.Permissions {
		// This flag is set to true if user has explicitly asked to mount a HNS account
		// Validate account is indeed HNS checking permissions field
		// If the account is FNS, the call will fail with InvalidQueryParameterValue and such mount shall fail
		includeFields.Permissions = true
	}

	listBlobPager := bb.Container.NewListBlobsHierarchyPager("/", &container.ListBlobsHierarchyOptions{
		MaxResults: to.Ptr((int32)(2)),
		Prefix:     &bb.Config.prefixPath,
		Include:    includeFields,
	})

	// we are just validating the auth mode used. So, no need to iterate over the pages
	_, err := listBlobPager.NextPage(context.Background())
	if err != nil {
		log.Err("BlockBlob::TestPipeline : Failed to validate account with given auth %s", err.Error())
		var respErr *azcore.ResponseError
		errors.As(err, &respErr)
		if respErr != nil {
			if respErr.ErrorCode == "InvalidQueryParameterValue" {
				// User explicitly mounting FNS account as HNS which is not supported
				return fmt.Errorf("BlockBlob::TestPipeline : Detected FNS account being mounted as HNS")
			}
			return fmt.Errorf("BlockBlob::TestPipeline : [%s]", respErr.ErrorCode)
		}
		return err
	}

	return nil
}

// IsAccountADLS : Check account is ADLS or not
func (bb *BlockBlob) IsAccountADLS() bool {
	includeFields := bb.listDetails
	includeFields.Permissions = true // for FNS account this property will return back error

	listBlobPager := bb.Container.NewListBlobsHierarchyPager("/", &container.ListBlobsHierarchyOptions{
		MaxResults: to.Ptr((int32)(2)),
		Prefix:     &bb.Config.prefixPath,
		Include:    includeFields,
	})

	// we are just validating the auth mode used. So, no need to iterate over the pages
	_, err := listBlobPager.NextPage(context.Background())

	if err == nil {
		// Call will be successful only when we are able to retrieve the permissions
		// Permissions will work only in case of HNS accounts
		log.Crit("BlockBlob::IsAccountADLS : Detected HNS account")
		return true
	}

	var respErr *azcore.ResponseError
	errors.As(err, &respErr)
	if respErr != nil {
		if respErr.ErrorCode == "InvalidQueryParameterValue" {
			log.Crit("BlockBlob::IsAccountADLS : Detected FNS account")
			return false
		}
	}

	log.Crit("BlockBlob::IsAccountADLS : Unable to detect account type, assuming FNS [%s]", err.Error())
	return false
}

func (bb *BlockBlob) ListContainers() ([]string, error) {
	log.Trace("BlockBlob::ListContainers : Listing containers")
	cntList := make([]string, 0)

	pager := bb.Service.NewListContainersPager(nil)
	for pager.More() {
		resp, err := pager.NextPage(context.Background())
		if err != nil {
			log.Err("BlockBlob::ListContainers : Failed to get container list [%s]", err.Error())
			return cntList, err
		}
		for _, v := range resp.ContainerItems {
			cntList = append(cntList, *v.Name)
		}
	}

	return cntList, nil
}

func (bb *BlockBlob) SetPrefixPath(path string) error {
	log.Trace("BlockBlob::SetPrefixPath : path %s", path)
	bb.Config.prefixPath = path
	return nil
}

// CreateFile : Create a new file in the container/virtual directory
func (bb *BlockBlob) CreateFile(name string, mode os.FileMode) error {
	log.Trace("BlockBlob::CreateFile : name %s", name)
	var data []byte
	return bb.WriteFromBuffer(name, nil, data)
}

// CreateDirectory : Create a new directory in the container/virtual directory
func (bb *BlockBlob) CreateDirectory(name string) error {
	log.Trace("BlockBlob::CreateDirectory : name %s", name)

	var data []byte
	metadata := make(map[string]*string)
	metadata[folderKey] = to.Ptr("true")

	return bb.WriteFromBuffer(name, metadata, data)
}

// CreateLink : Create a symlink in the container/virtual directory
func (bb *BlockBlob) CreateLink(source string, target string) error {
	log.Trace("BlockBlob::CreateLink : %s -> %s", source, target)
	data := []byte(target)
	metadata := make(map[string]*string)
	metadata[symlinkKey] = to.Ptr("true")
	return bb.WriteFromBuffer(source, metadata, data)
}

// DeleteFile : Delete a blob in the container/virtual directory
func (bb *BlockBlob) DeleteFile(name string) (err error) {
	log.Trace("BlockBlob::DeleteFile : name %s", name)

	blobClient := bb.Container.NewBlobClient(filepath.Join(bb.Config.prefixPath, name))
	_, err = blobClient.Delete(context.Background(), &blob.DeleteOptions{
		DeleteSnapshots: to.Ptr(blob.DeleteSnapshotsOptionTypeInclude),
	})
	if err != nil {
		serr := storeBlobErrToErr(err)
		switch serr {
		case ErrFileNotFound:
			log.Err("BlockBlob::DeleteFile : %s does not exist", name)
			return syscall.ENOENT
		case BlobIsUnderLease:
			log.Err("BlockBlob::DeleteFile : %s is under lease [%s]", name, err.Error())
			return syscall.EIO
		default:
			log.Err("BlockBlob::DeleteFile : Failed to delete blob %s [%s]", name, err.Error())
			return err
		}
	}

	return nil
}

// DeleteDirectory : Delete a virtual directory in the container/virtual directory
func (bb *BlockBlob) DeleteDirectory(name string) (err error) {
	log.Trace("BlockBlob::DeleteDirectory : name %s", name)
	err = bb.DeleteFile(name)
	// libfuse deletes the files in the directory before this method is called.
	// If the marker blob for directory is not present, ignore the ENOENT error.
	if err == syscall.ENOENT {
		err = nil
	}
	return err
}

// RenameFile : Rename the file
// Source file must exist in storage account before calling this method.
// When the rename is success, Data, metadata, of the blob will be copied to the destination.
// Creation time and LMT is not preserved for copyBlob API.
// Etag of the destination blob changes.
// Copy the LMT to the src attr if the copy is success.
// https://learn.microsoft.com/en-us/rest/api/storageservices/copy-blob?tabs=microsoft-entra-id
func (bb *BlockBlob) RenameFile(source string, target string, srcAttr *internal.ObjAttr) error {
	log.Trace("BlockBlob::RenameFile : %s -> %s", source, target)

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, source))
	newBlobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, target))

	// not specifying source blob metadata, since passing empty metadata headers copies
	// the source blob metadata to destination blob
	copyResponse, err := newBlobClient.StartCopyFromURL(context.Background(), blobClient.URL(), &blob.StartCopyFromURLOptions{
		Tier: bb.Config.defaultTier,
	})

	if err != nil {
		serr := storeBlobErrToErr(err)
		if serr == ErrFileNotFound {
			//Ideally this case doesn't hit as we are checking for the existence of src
			//before making the call for RenameFile
			log.Err("BlockBlob::RenameFile : Src Blob doesn't Exist %s [%s]", source, err.Error())
			return syscall.ENOENT
		}
		log.Err("BlockBlob::RenameFile : Failed to start copy of file %s [%s]", source, err.Error())
		return err
	}

	var dstLMT = copyResponse.LastModified
	var dstETag = sanitizeEtag(copyResponse.ETag)

	copyStatus := copyResponse.CopyStatus
	var prop blob.GetPropertiesResponse
	pollCnt := 0
	for copyStatus != nil && *copyStatus == blob.CopyStatusTypePending {
		time.Sleep(time.Second * 1)
		pollCnt++
		prop, err = newBlobClient.GetProperties(context.Background(), &blob.GetPropertiesOptions{
			CPKInfo: bb.blobCPKOpt,
		})
		if err != nil {
			log.Err("BlockBlob::RenameFile : CopyStats : Failed to get blob properties for %s [%s]", source, err.Error())
		}
		copyStatus = prop.CopyStatus
	}

	if pollCnt > 0 {
		dstLMT = prop.LastModified
		dstETag = sanitizeEtag(prop.ETag)
	}

	if copyStatus != nil && *copyStatus == blob.CopyStatusTypeSuccess {
		modifyLMTandEtag(srcAttr, dstLMT, dstETag)
	}

	log.Trace("BlockBlob::RenameFile : %s -> %s done", source, target)

	// Copy of the file is done so now delete the older file
	err = bb.DeleteFile(source)
	for retry := 0; retry < 3 && err == syscall.ENOENT; retry++ {
		// Sometimes backend is able to copy source file to destination but when we try to delete the
		// source files it returns back with ENOENT. If file was just created on backend it might happen
		// that it has not been synced yet at all layers and hence delete is not able to find the source file
		log.Trace("BlockBlob::RenameFile : %s -> %s, unable to find source. Retrying %d", source, target, retry)
		time.Sleep(1 * time.Second)
		err = bb.DeleteFile(source)
	}

	if err == syscall.ENOENT {
		// Even after 3 retries, 1 second apart if server returns 404 then source file no longer
		// exists on the backend and its safe to assume rename was successful
		err = nil
	}

	return err
}

// RenameDirectory : Rename the directory
func (bb *BlockBlob) RenameDirectory(source string, target string) error {
	log.Trace("BlockBlob::RenameDirectory : %s -> %s", source, target)

	srcDirPresent := false
	pager := bb.Container.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
		Prefix: to.Ptr(filepath.Join(bb.Config.prefixPath, source) + "/"),
	})
	for pager.More() {
		listBlobResp, err := pager.NextPage(context.Background())
		if err != nil {
			log.Err("BlockBlob::RenameDirectory : Failed to get list of blobs %s", err.Error())
			return err
		}

		// Process the blobs returned in this result segment (if the segment is empty, the loop body won't execute)
		for _, blobInfo := range listBlobResp.Segment.BlobItems {
			srcDirPresent = true
			srcPath := removePrefixPath(bb.Config.prefixPath, *blobInfo.Name)
			err = bb.RenameFile(srcPath, strings.Replace(srcPath, source, target, 1), nil)
			if err != nil {
				log.Err("BlockBlob::RenameDirectory : Failed to rename file %s [%s]", srcPath, err.Error)
			}
		}
	}

	// To rename source marker blob check its properties before calling rename on it.
	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, source))
	_, err := blobClient.GetProperties(context.Background(), &blob.GetPropertiesOptions{
		CPKInfo: bb.blobCPKOpt,
	})
	if err != nil {
		serr := storeBlobErrToErr(err)
		if serr == ErrFileNotFound { //marker blob doesn't exist for the directory
			if srcDirPresent { //Some files exist inside the directory
				return nil
			}
			log.Err("BlockBlob::RenameDirectory : %s marker blob does not exist and Src Directory doesn't Exist", source)
			return syscall.ENOENT
		} else {
			log.Err("BlockBlob::RenameDirectory : Failed to get source directory marker blob properties for %s [%s]", source, err.Error())
			return err
		}
	}

	return bb.RenameFile(source, target, nil)
}

func (bb *BlockBlob) getAttrUsingRest(name string) (attr *internal.ObjAttr, err error) {
	log.Trace("BlockBlob::getAttrUsingRest : name %s", name)

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	prop, err := blobClient.GetProperties(context.Background(), &blob.GetPropertiesOptions{
		CPKInfo: bb.blobCPKOpt,
	})

	if err != nil {
		serr := storeBlobErrToErr(err)
		switch serr {
		case ErrFileNotFound:
			return attr, syscall.ENOENT
		case InvalidPermission:
			log.Err("BlockBlob::getAttrUsingRest : Insufficient permissions for %s [%s]", name, err.Error())
			return attr, syscall.EACCES
		default:
			log.Err("BlockBlob::getAttrUsingRest : Failed to get blob properties for %s [%s]", name, err.Error())
			return attr, err
		}
	}

	// Since block blob does not support acls, we set mode to 0 and FlagModeDefault to true so the fuse layer can return the default permission.
	attr = &internal.ObjAttr{
		Path:   name, // We don't need to strip the prefixPath here since we pass the input name
		Name:   filepath.Base(name),
		Size:   *prop.ContentLength,
		Mode:   0,
		Mtime:  *prop.LastModified,
		Atime:  *prop.LastModified,
		Ctime:  *prop.LastModified,
		Crtime: *prop.CreationTime,
		Flags:  internal.NewFileBitMap(),
		MD5:    prop.ContentMD5,
		ETag:   sanitizeEtag(prop.ETag),
	}

	parseMetadata(attr, prop.Metadata)

	// We do not get permissions as part of this getAttr call hence setting the flag to true
	attr.Flags.Set(internal.PropFlagModeDefault)

	return attr, nil
}

func (bb *BlockBlob) getAttrUsingList(name string) (attr *internal.ObjAttr, err error) {
	log.Trace("BlockBlob::getAttrUsingList : name %s", name)

	iteration := 0
	var marker, new_marker *string
	var blobs []*internal.ObjAttr
	blobsRead := 0

	for marker != nil || iteration == 0 {
		blobs, new_marker, err = bb.List(name, marker, bb.Config.maxResultsForList)
		if err != nil {
			e := storeBlobErrToErr(err)
			switch e {
			case ErrFileNotFound:
				return attr, syscall.ENOENT
			case InvalidPermission:
				log.Err("BlockBlob::getAttrUsingList : Insufficient permissions for %s [%s]", name, err.Error())
				return attr, syscall.EACCES
			default:
				log.Warn("BlockBlob::getAttrUsingList : Failed to list blob properties for %s [%s]", name, err.Error())
			}
		}

		for i, blob := range blobs {
			log.Trace("BlockBlob::getAttrUsingList : Item %d Blob %s", i+blobsRead, blob.Name)
			if blob.Path == name {
				return blob, nil
			}
		}

		marker = new_marker
		iteration++
		blobsRead += len(blobs)

		log.Trace("BlockBlob::getAttrUsingList : So far retrieved %d objects in %d iterations", blobsRead, iteration)
		if new_marker == nil || *new_marker == "" {
			break
		}
	}

	if err == nil {
		log.Warn("BlockBlob::getAttrUsingList : blob %s does not exist", name)
		return nil, syscall.ENOENT
	}

	log.Err("BlockBlob::getAttrUsingList : Failed to list blob properties for %s [%s]", name, err.Error())
	return nil, err
}

// GetAttr : Retrieve attributes of the blob
func (bb *BlockBlob) GetAttr(name string) (attr *internal.ObjAttr, err error) {
	log.Trace("BlockBlob::GetAttr : name %s", name)

	// To support virtual directories with no marker blob, we call list instead of get properties since list will not return a 404
	if bb.Config.virtualDirectory {
		attr, err = bb.getAttrUsingList(name)
	} else {
		attr, err = bb.getAttrUsingRest(name)
	}

	if bb.Config.filter != nil && attr != nil {
		if !bb.Config.filter.IsAcceptable(&blobfilter.BlobAttr{
			Name:  attr.Name,
			Mtime: attr.Mtime,
			Size:  attr.Size,
		}) {
			log.Debug("BlockBlob::GetAttr : Filtered out %s", name)
			return nil, syscall.ENOENT
		}
	}

	return attr, err
}

// List : Get a list of blobs matching the given prefix
// This fetches the list using a marker so the caller code should handle marker logic
// If count=0 - fetch max entries
func (bb *BlockBlob) List(prefix string, marker *string, count int32) ([]*internal.ObjAttr, *string, error) {
	log.Trace("BlockBlob::List : prefix %s, marker %s", prefix, func(marker *string) string {
		if marker != nil {
			return *marker
		} else {
			return ""
		}
	}(marker))

	if count == 0 {
		count = common.MaxDirListCount
	}

	listPath := bb.getListPath(prefix)

	// Get a result segment starting with the blob indicated by the current Marker.
	pager := bb.Container.NewListBlobsHierarchyPager("/", &container.ListBlobsHierarchyOptions{
		Marker:     marker,
		MaxResults: &count,
		Prefix:     &listPath,
		Include:    bb.listDetails,
	})

	listBlob, err := pager.NextPage(context.Background())

	// Note: Since we make a list call with a prefix, we will not fail here for a non-existent directory.
	// The blob service will not validate for us whether or not the path exists.
	// This is different from ADLS Gen2 behavior.
	// APIs that may be affected include IsDirEmpty, ReadDir and StreamDir

	if err != nil {
		log.Err("BlockBlob::List : Failed to list the container with the prefix %s", err.Error)
		return nil, nil, err
	}

	// Process the blobs returned in this result segment (if the segment is empty, the loop body won't execute)
	// Since block blob does not support acls, we set mode to 0 and FlagModeDefault to true so the fuse layer can return the default permission.

	blobList, dirList, err := bb.processBlobItems(listBlob.Segment.BlobItems)
	if err != nil {
		return nil, nil, err
	}

	// In case virtual directory exists but its corresponding 0 byte marker file is not there holding hdi_isfolder then just iterating
	// over BlobItems will fail to identify that directory. In such cases BlobPrefixes help to list all directories
	// dirList contains all dirs for which we got 0 byte meta file in this iteration, so exclude those and add rest to the list
	// Note: Since listing is paginated, sometimes the marker file may come in a different iteration from the BlobPrefix. For such
	// cases we manually call GetAttr to check the existence of the marker file.
	err = bb.processBlobPrefixes(listBlob.Segment.BlobPrefixes, dirList, &blobList)
	if err != nil {
		return nil, nil, err
	}

	return blobList, listBlob.NextMarker, nil
}

func (bb *BlockBlob) getListPath(prefix string) string {
	listPath := filepath.Join(bb.Config.prefixPath, prefix)
	if (prefix != "" && prefix[len(prefix)-1] == '/') || (prefix == "" && bb.Config.prefixPath != "") {
		listPath += "/"
	}
	return listPath
}

func (bb *BlockBlob) processBlobItems(blobItems []*container.BlobItem) ([]*internal.ObjAttr, map[string]bool, error) {
	blobList := make([]*internal.ObjAttr, 0)
	// For some directories 0 byte meta file may not exists so just create a map to figure out such directories
	dirList := make(map[string]bool)
	filterAttr := blobfilter.BlobAttr{}

	for _, blobInfo := range blobItems {
		blobAttr, err := bb.getBlobAttr(blobInfo)
		if err != nil {
			return nil, nil, err
		}

		if blobAttr.IsDir() {
			// 0 byte meta found so mark this directory in map
			dirList[*blobInfo.Name+"/"] = true
			blobAttr.Size = 4096
		}

		if bb.Config.filter != nil && !blobAttr.IsDir() {
			filterAttr.Name = blobAttr.Name
			filterAttr.Mtime = blobAttr.Mtime
			filterAttr.Size = blobAttr.Size

			if bb.Config.filter.IsAcceptable(&filterAttr) {
				blobList = append(blobList, blobAttr)
			} else {
				log.Debug("BlockBlob::List : Filtered out blob %s", blobAttr.Name)
			}
		} else {
			blobList = append(blobList, blobAttr)
		}
	}

	return blobList, dirList, nil
}

func (bb *BlockBlob) getBlobAttr(blobInfo *container.BlobItem) (*internal.ObjAttr, error) {
	if blobInfo.Properties.CustomerProvidedKeySHA256 != nil && *blobInfo.Properties.CustomerProvidedKeySHA256 != "" {
		log.Trace("BlockBlob::List : blob is encrypted with customer provided key so fetching metadata explicitly using REST")
		return bb.getAttrUsingRest(*blobInfo.Name)
	}
	mode, err := bb.getFileMode(blobInfo.Properties.Permissions)
	if err != nil {
		mode = 0
		log.Warn("BlockBlob::getBlobAttr : Failed to get file mode for %s [%s]", *blobInfo.Name, err.Error())
	}

	attr := &internal.ObjAttr{
		Path:   removePrefixPath(bb.Config.prefixPath, *blobInfo.Name),
		Name:   filepath.Base(*blobInfo.Name),
		Size:   *blobInfo.Properties.ContentLength,
		Mode:   mode,
		Mtime:  *blobInfo.Properties.LastModified,
		Atime:  bb.dereferenceTime(blobInfo.Properties.LastAccessedOn, *blobInfo.Properties.LastModified),
		Ctime:  *blobInfo.Properties.LastModified,
		Crtime: bb.dereferenceTime(blobInfo.Properties.CreationTime, *blobInfo.Properties.LastModified),
		Flags:  internal.NewFileBitMap(),
		MD5:    blobInfo.Properties.ContentMD5,
		ETag:   sanitizeEtag(blobInfo.Properties.ETag),
	}

	parseMetadata(attr, blobInfo.Metadata)
	if !bb.listDetails.Permissions {
		// In case of HNS account do not set this flag
		attr.Flags.Set(internal.PropFlagModeDefault)
	}

	return attr, nil
}

func (bb *BlockBlob) getFileMode(permissions *string) (os.FileMode, error) {
	if permissions == nil {
		return 0, nil
	}
	return getFileMode(*permissions)
}

func (bb *BlockBlob) dereferenceTime(input *time.Time, defaultTime time.Time) time.Time {
	if input == nil {
		return defaultTime
	}
	return *input
}

func (bb *BlockBlob) processBlobPrefixes(blobPrefixes []*container.BlobPrefix, dirList map[string]bool, blobList *[]*internal.ObjAttr) error {
	for _, blobInfo := range blobPrefixes {
		if _, ok := dirList[*blobInfo.Name]; ok {
			// marker file found in current iteration, skip adding the directory
			continue
		} else {
			//Check to see if its a HNS account and we received properties in blob prefixes
			if bb.listDetails.Permissions {
				attr, err := bb.createDirAttrWithPermissions(blobInfo)
				if err != nil {
					return err
				}
				*blobList = append(*blobList, attr)
			} else {
				// marker file not found in current iteration, so we need to manually check attributes via REST
				_, err := bb.getAttrUsingRest(*blobInfo.Name)
				// marker file also not found via manual check, safe to add to list
				// For HNS accounts mounted as FNS we used to list directories and files in blobfusev1,
				// in blobfusev2 to replicate this behaviour the below check of blobInfo.Properties != nil is added.
				if err == syscall.ENOENT || blobInfo.Properties != nil {
					attr := bb.createDirAttr(*blobInfo.Name)
					*blobList = append(*blobList, attr)
				}
			}
		}
	}

	// Clean up the temp map as its no more needed
	for k := range dirList {
		delete(dirList, k)
	}

	return nil
}

func (bb *BlockBlob) createDirAttr(name string) *internal.ObjAttr {
	// For these dirs we get only the name and no other properties so hardcoding time to current time
	name = strings.TrimSuffix(name, "/")
	attr := &internal.ObjAttr{
		Path:  removePrefixPath(bb.Config.prefixPath, name),
		Name:  filepath.Base(name),
		Size:  4096,
		Mode:  os.ModeDir,
		Mtime: time.Now(),
		Flags: internal.NewDirBitMap(),
	}
	attr.Atime = attr.Mtime
	attr.Crtime = attr.Mtime
	attr.Ctime = attr.Mtime

	// This is called only in case of FNS when blobPrefix is there but the marker does not exists
	attr.Flags.Set(internal.PropFlagModeDefault)
	return attr
}

func (bb *BlockBlob) createDirAttrWithPermissions(blobInfo *container.BlobPrefix) (*internal.ObjAttr, error) {
	if blobInfo.Properties == nil {
		return nil, fmt.Errorf("failed to get properties of blobprefix %s", *blobInfo.Name)
	}

	mode, err := bb.getFileMode(blobInfo.Properties.Permissions)
	if err != nil {
		mode = 0
		log.Warn("BlockBlob::createDirAttrWithPermissions : Failed to get file mode for %s [%s]", *blobInfo.Name, err.Error())
	}

	name := strings.TrimSuffix(*blobInfo.Name, "/")
	attr := &internal.ObjAttr{
		Path:   removePrefixPath(bb.Config.prefixPath, name),
		Name:   filepath.Base(name),
		Size:   *blobInfo.Properties.ContentLength,
		Mode:   mode,
		Mtime:  *blobInfo.Properties.LastModified,
		Atime:  bb.dereferenceTime(blobInfo.Properties.LastAccessedOn, *blobInfo.Properties.LastModified),
		Ctime:  *blobInfo.Properties.LastModified,
		Crtime: bb.dereferenceTime(blobInfo.Properties.CreationTime, *blobInfo.Properties.LastModified),
		Flags:  internal.NewDirBitMap(),
	}

	return attr, nil
}

// track the progress of download of blobs where every 100MB of data downloaded is being tracked. It also tracks the completion of download
func trackDownload(name string, bytesTransferred int64, count int64, downloadPtr *int64) {
	if bytesTransferred >= (*downloadPtr)*100*common.MbToBytes || bytesTransferred == count {
		(*downloadPtr)++
		log.Debug("BlockBlob::trackDownload : Download: Blob = %v, Bytes transferred = %v, Size = %v", name, bytesTransferred, count)

		// send the download progress as an event
		azStatsCollector.PushEvents(downloadProgress, name, map[string]any{bytesTfrd: bytesTransferred, size: count})
	}
}

// ReadToFile : Download a blob to a local file
func (bb *BlockBlob) ReadToFile(name string, offset int64, count int64, fi *os.File) (err error) {
	log.Trace("BlockBlob::ReadToFile : name %s, offset : %d, count %d", name, offset, count)
	//defer exectime.StatTimeCurrentBlock("BlockBlob::ReadToFile")()

	blobClient := bb.Container.NewBlobClient(filepath.Join(bb.Config.prefixPath, name))

	downloadPtr := to.Ptr(int64(1))

	if common.MonitorBfs() {
		bb.downloadOptions.Progress = func(bytesTransferred int64) {
			trackDownload(name, bytesTransferred, count, downloadPtr)
		}
	}

	defer log.TimeTrack(time.Now(), "BlockBlob::ReadToFile", name)

	dlOpts := *bb.downloadOptions
	dlOpts.Range = blob.HTTPRange{
		Offset: offset,
		Count:  count,
	}

	_, err = blobClient.DownloadFile(context.Background(), fi, &dlOpts)

	if err != nil {
		e := storeBlobErrToErr(err)
		if e == ErrFileNotFound {
			return syscall.ENOENT
		} else {
			log.Err("BlockBlob::ReadToFile : Failed to download blob %s [%s]", name, err.Error())
			return err
		}
	} else {
		log.Debug("BlockBlob::ReadToFile : Download complete of blob %v", name)

		// store total bytes downloaded so far
		azStatsCollector.UpdateStats(stats_manager.Increment, bytesDownloaded, count)
	}

	if bb.Config.validateMD5 {
		// Compute md5 of local file
		fileMD5, err := common.GetMD5(fi)
		if err != nil {
			log.Warn("BlockBlob::ReadToFile : Failed to generate MD5 Sum for %s", name)
		} else {
			// Get latest properties from container to get the md5 of blob
			prop, err := blobClient.GetProperties(context.Background(), &blob.GetPropertiesOptions{
				CPKInfo: bb.blobCPKOpt,
			})
			if err != nil {
				log.Warn("BlockBlob::ReadToFile : Failed to get properties of blob %s [%s]", name, err.Error())
			} else {
				blobMD5 := prop.ContentMD5
				if blobMD5 == nil {
					log.Warn("BlockBlob::ReadToFile : Failed to get MD5 Sum for blob %s", name)
				} else {
					// compare md5 and fail is not match
					if !reflect.DeepEqual(fileMD5, blobMD5) {
						log.Err("BlockBlob::ReadToFile : MD5 Sum mismatch %s", name)
						return errors.New("md5 sum mismatch on download")
					}
				}
			}
		}
	}

	return nil
}

// ReadBuffer : Download a specific range from a blob to a buffer
func (bb *BlockBlob) ReadBuffer(name string, offset int64, length int64) ([]byte, error) {
	log.Trace("BlockBlob::ReadBuffer : name %s, offset %v, len %v", name, offset, length)
	var buff []byte
	if length == 0 {
		attr, err := bb.GetAttr(name)
		if err != nil {
			return buff, err
		}
		length = attr.Size - offset
	}

	buff = make([]byte, length)
	blobClient := bb.Container.NewBlobClient(filepath.Join(bb.Config.prefixPath, name))

	dlOpts := (blob.DownloadBufferOptions)(*bb.downloadOptions)
	dlOpts.Range = blob.HTTPRange{
		Offset: offset,
		Count:  length,
	}

	_, err := blobClient.DownloadBuffer(context.Background(), buff, &dlOpts)

	if err != nil {
		e := storeBlobErrToErr(err)
		switch e {
		case ErrFileNotFound:
			return buff, syscall.ENOENT
		case InvalidRange:
			return buff, syscall.ERANGE
		}

		log.Err("BlockBlob::ReadBuffer : Failed to download blob %s [%s]", name, err.Error())
		return buff, err
	}

	return buff, nil
}

// ReadInBuffer : Download specific range from a file to a user provided buffer
// Specifying "0" len will download the entire blob.
func (bb *BlockBlob) ReadInBuffer(name string, offset int64, length int64, data []byte, etag *string) error {
	// log.Trace("BlockBlob::ReadInBuffer : name %s", name)
	if etag != nil {
		*etag = ""
	}

	blobClient := bb.Container.NewBlobClient(filepath.Join(bb.Config.prefixPath, name))

	ctx, cancel := context.WithTimeout(context.Background(), max_context_timeout*time.Minute)
	defer cancel()

	opt := &blob.DownloadStreamOptions{
		Range: blob.HTTPRange{
			Offset: offset,
			Count:  length,
		},
		CPKInfo: bb.blobCPKOpt,
	}

	downloadResponse, err := blobClient.DownloadStream(ctx, opt)

	if err != nil {
		e := storeBlobErrToErr(err)
		switch e {
		case ErrFileNotFound:
			return syscall.ENOENT
		case InvalidRange:
			return syscall.ERANGE
		}

		log.Err("BlockBlob::ReadInBufferWithETag : Failed to download blob %s [%s]", name, err.Error())
		return err
	}

	var streamBody io.ReadCloser = downloadResponse.NewRetryReader(ctx, nil)
	dataRead, err := io.ReadFull(streamBody, data)

	if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
		log.Err("BlockBlob::ReadInBuffer : Failed to copy data from body to buffer for blob %s [%s]", name, err.Error())
		return err
	}

	if dataRead < 0 {
		log.Err("BlockBlob::ReadInBuffer : Failed to copy data from body to buffer for blob %s", name)
		return errors.New("failed to copy data from body to buffer")
	}

	err = streamBody.Close()
	if err != nil {
		log.Err("BlockBlob::ReadInBuffer : Failed to close body for blob %s [%s]", name, err.Error())
	}

	if etag != nil {
		*etag = sanitizeEtag(downloadResponse.ETag)
	}

	return nil
}

func (bb *BlockBlob) calculateBlockSize(name string, fileSize int64) (blockSize int64, err error) {
	// If bufferSize > (BlockBlobMaxStageBlockBytes * BlockBlobMaxBlocks), then error
	if fileSize > MaxBlobSize {
		log.Err("BlockBlob::calculateBlockSize : buffer is too large to upload to a block blob %s", name)
		err = errors.New("buffer is too large to upload to a block blob")
		return 0, err
	}

	// If bufferSize <= BlockBlobMaxUploadBlobBytes, then Upload should be used with just 1 I/O request
	if fileSize <= blockblob.MaxUploadBlobBytes {
		// Files up to 256MB can be uploaded as a single block
		blockSize = blockblob.MaxUploadBlobBytes
	} else {
		// buffer / max blocks = block size to use all 50,000 blocks
		blockSize = int64(math.Ceil(float64(fileSize) / blockblob.MaxBlocks))

		if blockSize < blob.DefaultDownloadBlockSize {
			// Block size is smaller then 4MB then consider 4MB as default
			blockSize = blob.DefaultDownloadBlockSize
		} else {
			if (blockSize & (-8)) != 0 {
				// EXTRA : round off the block size to next higher multiple of 8.
				// No reason to do so just the odd numbers in block size will not be good on server end is assumption
				blockSize = (blockSize + 7) & (-8)
			}

			if blockSize > blockblob.MaxStageBlockBytes {
				// After rounding off the blockSize has become bigger then max allowed blocks.
				log.Err("BlockBlob::calculateBlockSize : blockSize exceeds max allowed block size for %s", name)
				err = errors.New("block-size is too large to upload to a block blob")
				return 0, err
			}
		}
	}

	log.Info("BlockBlob::calculateBlockSize : %s size %v, blockSize %v", name, fileSize, blockSize)
	return blockSize, nil
}

// track the progress of upload of blobs where every 100MB of data uploaded is being tracked. It also tracks the completion of upload
func trackUpload(name string, bytesTransferred int64, count int64, uploadPtr *int64) {
	if bytesTransferred >= (*uploadPtr)*100*common.MbToBytes || bytesTransferred == count {
		(*uploadPtr)++
		log.Debug("BlockBlob::trackUpload : Upload: Blob = %v, Bytes transferred = %v, Size = %v", name, bytesTransferred, count)

		// send upload progress as event
		azStatsCollector.PushEvents(uploadProgress, name, map[string]any{bytesTfrd: bytesTransferred, size: count})
	}
}

// WriteFromFile : Upload local file to blob
func (bb *BlockBlob) WriteFromFile(name string, metadata map[string]*string, fi *os.File) (err error) {
	log.Trace("BlockBlob::WriteFromFile : name %s", name)
	//defer exectime.StatTimeCurrentBlock("WriteFromFile::WriteFromFile")()

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	defer log.TimeTrack(time.Now(), "BlockBlob::WriteFromFile", name)

	uploadPtr := to.Ptr(int64(1))

	blockSize := bb.Config.blockSize
	// get the size of the file
	stat, err := fi.Stat()
	if err != nil {
		log.Err("BlockBlob::WriteFromFile : Failed to get file size %s [%s]", name, err.Error())
		return err
	}

	// if the block size is not set then we configure it based on file size
	if blockSize == 0 {
		// based on file-size calculate block size
		blockSize, err = bb.calculateBlockSize(name, stat.Size())
		if err != nil {
			return err
		}
	}

	// Compute md5 of this file is requested by user
	// If file is uploaded in one shot (no blocks created) then server is populating md5 on upload automatically.
	// hence we take cost of calculating md5 only for files which are bigger in size and which will be converted to blocks.
	md5sum := []byte{}
	if bb.Config.updateMD5 && stat.Size() >= blockblob.MaxUploadBlobBytes {
		md5sum, err = common.GetMD5(fi)
		if err != nil {
			// Md5 sum generation failed so set nil while uploading
			log.Warn("BlockBlob::WriteFromFile : Failed to generate md5 of %s", name)
			md5sum = []byte{0}
		}
	}

	uploadOptions := &blockblob.UploadFileOptions{
		BlockSize:   blockSize,
		Concurrency: bb.Config.maxConcurrency,
		Metadata:    metadata,
		AccessTier:  bb.Config.defaultTier,
		HTTPHeaders: &blob.HTTPHeaders{
			BlobContentType: to.Ptr(getContentType(name)),
			BlobContentMD5:  md5sum,
		},
		CPKInfo: bb.blobCPKOpt,
	}
	if common.MonitorBfs() && stat.Size() > 0 {
		uploadOptions.Progress = func(bytesTransferred int64) {
			trackUpload(name, bytesTransferred, stat.Size(), uploadPtr)
		}
	}

	_, err = blobClient.UploadFile(context.Background(), fi, uploadOptions)

	if err != nil {
		serr := storeBlobErrToErr(err)
		switch serr {
		case BlobIsUnderLease:
			log.Err("BlockBlob::WriteFromFile : %s is under a lease, can not update file [%s]", name, err.Error())
			return syscall.EIO
		case InvalidPermission:
			log.Err("BlockBlob::WriteFromFile : Insufficient permissions for %s [%s]", name, err.Error())
			return syscall.EACCES
		default:
			log.Err("BlockBlob::WriteFromFile : Failed to upload blob %s [%s]", name, err.Error())
		}
		return err
	} else {
		log.Debug("BlockBlob::WriteFromFile : Upload complete of blob %v", name)

		// store total bytes uploaded so far
		if stat.Size() > 0 {
			azStatsCollector.UpdateStats(stats_manager.Increment, bytesUploaded, stat.Size())
		}
	}

	return nil
}

// WriteFromBuffer : Upload from a buffer to a blob
func (bb *BlockBlob) WriteFromBuffer(name string, metadata map[string]*string, data []byte) error {
	log.Trace("BlockBlob::WriteFromBuffer : name %s", name)
	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))

	defer log.TimeTrack(time.Now(), "BlockBlob::WriteFromBuffer", name)

	_, err := blobClient.UploadBuffer(context.Background(), data, &blockblob.UploadBufferOptions{
		BlockSize:   bb.Config.blockSize,
		Concurrency: bb.Config.maxConcurrency,
		Metadata:    metadata,
		AccessTier:  bb.Config.defaultTier,
		HTTPHeaders: &blob.HTTPHeaders{
			BlobContentType: to.Ptr(getContentType(name)),
		},
		CPKInfo: bb.blobCPKOpt,
	})

	if err != nil {
		log.Err("BlockBlob::WriteFromBuffer : Failed to upload blob %s [%s]", name, err.Error())
		return err
	}

	return nil
}

// GetFileBlockOffsets: store blocks ids and corresponding offsets
func (bb *BlockBlob) GetFileBlockOffsets(name string) (*common.BlockOffsetList, error) {
	var blockOffset int64 = 0
	blockList := common.BlockOffsetList{}
	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))

	storageBlockList, err := blobClient.GetBlockList(context.Background(), blockblob.BlockListTypeCommitted, nil)

	if err != nil {
		log.Err("BlockBlob::GetFileBlockOffsets : Failed to get block list %s ", name, err.Error())
		return &common.BlockOffsetList{}, err
	}

	// if block list empty its a small file
	if len(storageBlockList.CommittedBlocks) == 0 {
		blockList.BlockIdLength = common.BlockIDLength
		return &blockList, nil
	}

	blockList.BlockList = make([]*common.Block, 0, len(storageBlockList.CommittedBlocks))

	for _, block := range storageBlockList.CommittedBlocks {
		blk := &common.Block{
			Id:         *block.Name,
			StartIndex: int64(blockOffset),
			EndIndex:   int64(blockOffset) + *block.Size,
		}
		blockOffset += *block.Size
		blockList.BlockList = append(blockList.BlockList, blk)
	}
	// blockList.Etag = storageBlockList.ETag()
	blockList.BlockIdLength = common.GetIdLength(blockList.BlockList[0].Id)
	return &blockList, nil
}

func (bb *BlockBlob) createBlock(blockIdLength, startIndex, size int64) *common.Block {
	newBlockId := common.GetBlockID(blockIdLength)
	newBlock := &common.Block{
		Id:         newBlockId,
		StartIndex: startIndex,
		EndIndex:   startIndex + size,
	}
	// mark truncated since it is a new empty block
	newBlock.Flags.Set(common.TruncatedBlock)
	newBlock.Flags.Set(common.DirtyBlock)
	return newBlock
}

func (bb *BlockBlob) createBlockFromBuffer(blockIdLength, startIndex int64, data []byte) *common.Block {
	newBlockId := common.GetBlockID(blockIdLength)
	newBlock := &common.Block{
		Id:         newBlockId,
		StartIndex: startIndex,
		EndIndex:   startIndex + int64(len(data)),
		Data:       data,
	}
	// mark dirty since it is a new block with data
	newBlock.Flags.Set(common.DirtyBlock)
	return newBlock
}

// create new blocks based on the offset and total length we're adding to the file
func (bb *BlockBlob) createNewBlocks(blockList *common.BlockOffsetList, offset, length int64) (int64, error) {
	blockSize := bb.Config.blockSize
	prevIndex := blockList.BlockList[len(blockList.BlockList)-1].EndIndex
	numOfBlocks := int64(len(blockList.BlockList))
	if blockSize == 0 {
		blockSize = (16 * 1024 * 1024)
		if math.Ceil((float64)(numOfBlocks)+(float64)(length)/(float64)(blockSize)) > blockblob.MaxBlocks {
			blockSize = int64(math.Ceil((float64)(length) / (float64)(blockblob.MaxBlocks-numOfBlocks)))
			if blockSize > blockblob.MaxStageBlockBytes {
				return 0, errors.New("cannot accommodate data within the block limit")
			}
		}
	} else if math.Ceil((float64)(numOfBlocks)+(float64)(length)/(float64)(blockSize)) > blockblob.MaxBlocks {
		return 0, errors.New("cannot accommodate data within the block limit with configured block-size")
	}

	// BufferSize is the size of the buffer that will go beyond our current blob (appended)
	var bufferSize int64
	for i := prevIndex; i < offset+length; i += blockSize {
		blkSize := int64(math.Min(float64(blockSize), float64((offset+length)-i)))
		newBlock := bb.createBlock(blockList.BlockIdLength, i, blkSize)
		blockList.BlockList = append(blockList.BlockList, newBlock)
		// reset the counter to determine if there are leftovers at the end
		bufferSize += blkSize
	}
	return bufferSize, nil
}

// This function is called when the file is expanded using truncate operation and the new size is greater than the old
// size of the file.
func (bb *BlockBlob) createNewBlocksTruncate(blockList *common.BlockOffsetList, options *internal.TruncateFileOptions) error {
	log.Trace("BlockBlob::createNewBlocksTruncate : name: %s, old size: %d, new size: %d", options.Name, options.OldSize,
		options.NewSize)

	numOfBlocks := int64(len(blockList.BlockList))
	length := options.NewSize - options.OldSize

	blockSize := options.BlockSize
	if blockSize == 0 {
		blockSize = bb.Config.blockSize
		if blockSize == 0 {
			// TODO: This should be set in init of blobfuse
			blockSize = 16 * 1024 * 1024
		}

		blocksNeeded := (length + blockSize - 1) / blockSize

		if numOfBlocks+blocksNeeded > blockblob.MaxBlocks {
			// if we cannot accommodate the data with current block size then recalculate the block size
			availableBlocks := blockblob.MaxBlocks - numOfBlocks
			blockSize = (length + availableBlocks - 1) / availableBlocks
			if blockSize > blockblob.MaxStageBlockBytes {
				return errors.New("cannot accommodate data within the block limit")
			}
		}
	} else {
		// Case when user has specified the block size from previous component
		// Say if a 10MB file is created using block_cache with blockSize 8MB, say it's block list would be 1(8M), 2(2M).
		// Now say user is trying to truncate it to say 20MB. Then we should first download the second block and modify
		// it (i.e., New block list would be 1(8M), 2(8M), 3(4M)).

		// Calculate how many blocks we need with this block size
		blocksNeeded := (length + blockSize - 1) / blockSize
		// For user specified block size, validate if we can accommodate the data within the block limit
		if numOfBlocks+blocksNeeded > blockblob.MaxBlocks {
			return fmt.Errorf("cannot accommodate data within the block limit with configured block-size: %d", blockSize)
		}

		// if last block is not aligned to block size then adjust it
		if len(blockList.BlockList) > 0 {
			lastBlock := blockList.BlockList[len(blockList.BlockList)-1]

			if lastBlock.EndIndex-lastBlock.StartIndex < blockSize {
				// last block is smaller than block size, so adjust it
				lastBlock.EndIndex = min(lastBlock.StartIndex+blockSize, options.NewSize)

				lastBlock.Data = make([]byte, lastBlock.EndIndex-lastBlock.StartIndex)
				lastBlock.Flags.Set(common.DirtyBlock)

				// create the new block id for this block otherwise it would corrupt the state of the blob.
				lastBlock.Id = common.GetBlockID(blockList.BlockIdLength)

				err := bb.ReadInBuffer(options.Name, lastBlock.StartIndex, lastBlock.EndIndex-lastBlock.StartIndex, lastBlock.Data, nil)
				if err != nil {
					log.Err("BlockBlob::createNewBlocksTruncate : Failed to adjust last block %s [%v]", options.Name, err)
					return err
				}
			}

		}
	}

	var startIdx int64 = 0
	if len(blockList.BlockList) > 0 {
		startIdx = blockList.BlockList[len(blockList.BlockList)-1].EndIndex
	}

	for i := startIdx; i < options.NewSize; i += blockSize {
		curBlkSize := min(blockSize, options.NewSize-i)

		newBlock := bb.createBlock(blockList.BlockIdLength, i, curBlkSize)
		blockList.BlockList = append(blockList.BlockList, newBlock)
	}

	blockList.Flags.Set(common.BlobFlagBlockListModified)

	return nil
}

func (bb *BlockBlob) removeBlocksTruncate(blockList *common.BlockOffsetList, options *internal.TruncateFileOptions) error {
	log.Trace("BlockBlob::removeBlocksTruncate : name: %s, old size: %d, new size: %d", options.Name, options.OldSize,
		options.NewSize)

	if len(blockList.BlockList) == 0 {
		return errors.New("removeBlocksTruncate: block list is empty, cannot remove blocks")
	}

	size := options.NewSize

	idx := sort.Search(len(blockList.BlockList), func(i int) bool {
		return blockList.BlockList[i].StartIndex >= size
	})

	idx-- // move one index back to get the last block which is less than or equal to new size

	log.Debug("BlockBlob::removeBlocksTruncate : idx: %d, blockList length: %d, idx.start: %d, idx.end: %d, new size: %d",
		idx, len(blockList.BlockList), blockList.BlockList[idx].StartIndex, blockList.BlockList[idx].EndIndex, size)

	// if the file we're shrinking is in the middle of a block then shrink that block
	if blockList.BlockList[idx].EndIndex > size {
		blk := blockList.BlockList[idx]
		blk.EndIndex = size
		blk.Data = make([]byte, blk.EndIndex-blk.StartIndex)
		blk.Flags.Set(common.DirtyBlock)

		// create the new block id for this block otherwise it would corrupt the state of the blob.
		blk.Id = common.GetBlockID(blockList.BlockIdLength)

		err := bb.ReadInBuffer(options.Name, blk.StartIndex, blk.EndIndex-blk.StartIndex, blk.Data, nil)
		if err != nil {
			log.Err("BlockBlob::removeBlocksTruncate : Failed to remove blocks %s [%v]", options.Name, err)
			return err
		}
	}

	blockList.BlockList = blockList.BlockList[:idx+1]
	blockList.Flags.Set(common.BlobFlagBlockListModified)

	return nil
}

func (bb *BlockBlob) TruncateFile(options internal.TruncateFileOptions) error {
	log.Trace("BlockBlob::TruncateFile : name: %s, old size: %d, new size: %d",
		options.Name, options.OldSize, options.NewSize)

	// If old size is not specified, get it from the storage.
	if options.OldSize == -1 {
		attr, err := bb.GetAttr(options.Name)
		if err != nil {
			log.Err("BlockBlob::TruncateFile : Failed to get attributes of file %s [%v]", options.Name, err)
			return err
		}

		options.OldSize = attr.Size

		log.Trace("BlockBlob::TruncateFile : name: %s, old size: %d[Got from storage], new size: %d",
			options.Name, options.OldSize, options.NewSize)
	}

	if options.OldSize == options.NewSize {
		return nil
	}

	if options.NewSize == 0 {
		var buf []byte
		if err := bb.WriteFromBuffer(options.Name, nil, buf); err != nil {
			log.Err("BlockBlob::TruncateFile : Failed to truncate file %s to zero size [%v]", options.Name, err)
			return err
		}
		return nil
	}

	// Handle files whose new size <= 256MiB && when block size is not specified.
	if options.NewSize <= blockblob.MaxUploadBlobBytes && options.BlockSize == 0 {
		if err := bb.TruncateFileWithoutBlocks(&options); err != nil {
			log.Err("BlockBlob::TruncateFile : Failed to truncate file %s[%v]", options.Name, err)
			return err
		}
		return nil
	}

	// Truncate file by managing blocks
	if err := bb.TruncateFileUsingBlocks(&options); err != nil {
		log.Err("BlockBlob::TruncateFile : Failed to truncate file %s[%v]", options.Name, err)
		return err
	}

	return nil
}

func (bb *BlockBlob) TruncateFileWithoutBlocks(options *internal.TruncateFileOptions) error {
	log.Trace("BlockBlob::TruncateFileWithoutBlocks : name: %s, old size: %d, new size: %d", options.Name,
		options.OldSize, options.NewSize)
	var err error

	buf := make([]byte, options.NewSize)

	if options.OldSize > 0 {
		// Read the file
		err = bb.ReadInBuffer(options.Name, 0, min(options.NewSize, options.OldSize), buf, nil)
		if err != nil {
			log.Err("BlockBlob::TruncateFileWithoutBlocks : Failed to read small file %s[%v]", options.Name, err)
			return err
		}
	}

	// Write the file
	err = bb.WriteFromBuffer(options.Name, nil, buf)
	if err != nil {
		log.Err("BlockBlob::TruncateFileWithoutBlocks : Failed to write from buffer file %s[%v]", options.Name, err)
		return err
	}

	return nil
}

func (bb *BlockBlob) TruncateFileUsingBlocks(options *internal.TruncateFileOptions) error {
	log.Trace("BlockBlob::TruncateFileUsingBlocks : name: %s, old size: %d, new size: %d", options.Name,
		options.OldSize, options.NewSize)
	var err error

	blob, err := bb.GetFileBlockOffsets(options.Name)
	if err != nil {
		log.Err("BlockBlob::TruncateFileUsingBlocks : Failed to get block offsets for file %s[%v]", options.Name, err)
		return err
	}

	// If blob has no blocks, we should convert the blob into blocks first.
	if blob.HasNoBlocks() && options.OldSize > 0 {
		if options.OldSize > blockblob.MaxUploadBlobBytes {
			err = fmt.Errorf("Blob %v has size %d bytes greater than 256MB but has no blocks, inconsistent state",
				options.Name, options.OldSize)
			log.Err("BlockBlob::TruncateFileUsingBlocks : %v", err)
			return err
		}

		buf := make([]byte, options.OldSize)

		// Read the file
		err = bb.ReadInBuffer(options.Name, 0, options.OldSize, buf, nil)
		if err != nil {
			log.Err("BlockBlob::TruncateFileUsingBlocks : Failed to read small file %s[%v]", options.Name, err)
			return err
		}

		// Create blocks for the blob.
		blockSize := options.BlockSize
		if blockSize == 0 {
			blockSize = bb.Config.blockSize
			if blockSize == 0 {
				// TODO: This should be set in init of blobfuse
				blockSize = 16 * 1024 * 1024
			}

			for i := int64(0); i < options.OldSize; i += blockSize {
				blkSize := min(blockSize, options.OldSize-i)
				newBlock := bb.createBlockFromBuffer(blob.BlockIdLength, i, buf[i:i+blkSize])
				blob.BlockList = append(blob.BlockList, newBlock)
			}
		}
		blob.Flags.Set(common.BlobFlagBlockListModified)
	} else {
		// If blob has blocks, we should validate the block list against the old size.
		if ok := blob.ValidateBlockListAgainstFileSize(options.OldSize); !ok {
			err = fmt.Errorf("Blob %v has blocks that do not match the old size %d bytes, inconsistent state",
				options.Name, options.OldSize)
			log.Err("BlockBlob::TruncateFileUsingBlocks : %v", err)
			return err
		}
	}

	if options.NewSize < options.OldSize {
		err = bb.removeBlocksTruncate(blob, options)
		if err != nil {
			log.Err("BlockBlob::TruncateFileUsingBlocks : Failed to Remove Blocks from Blocklist for file %s[%v]",
				options.Name, err)
			return err
		}
	} else {
		err = bb.createNewBlocksTruncate(blob, options)
		if err != nil {
			log.Err("BlockBlob::TruncateFileUsingBlocks : Failed to Create New Blocks for file %s[%v]", options.Name, err)
			return err
		}
	}

	// Stage and commit the blocks.
	err = bb.StageAndCommit(options.Name, blob)
	if err != nil {
		log.Err("BlockBlob::TruncateFileUsingBlocks : Failed to Stage and Commit blocks for file %s[%v]", options.Name, err)
		return err
	}

	return nil
}

// Write : write data at given offset to a blob
func (bb *BlockBlob) Write(options *internal.WriteFileOptions) error {
	name := options.Handle.Path
	offset := options.Offset
	defer log.TimeTrack(time.Now(), "BlockBlob::Write", options.Handle.Path)
	log.Trace("BlockBlob::Write : name %s offset %v", name, offset)
	// tracks the case where our offset is great than our current file size (appending only - not modifying pre-existing data)
	var dataBuffer *[]byte
	// when the file offset mapping is cached we don't need to make a get block list call
	fileOffsets, err := bb.GetFileBlockOffsets(name)
	if err != nil {
		return err
	}
	length := int64(len(options.Data))
	data := options.Data
	// case 1: file consists of no blocks (small file)
	if fileOffsets.HasNoBlocks() {
		// get all the data
		oldData, _ := bb.ReadBuffer(name, 0, 0)
		// update the data with the new data
		// if we're only overwriting existing data
		if int64(len(oldData)) >= offset+length {
			copy(oldData[offset:], data)
			dataBuffer = &oldData
			// else appending and/or overwriting
		} else {
			// if the file is not empty then we need to combine the data
			if len(oldData) > 0 {
				// new data buffer with the size of old and new data
				newDataBuffer := make([]byte, offset+length)
				// copy the old data into it
				// TODO: better way to do this?
				if offset != 0 {
					copy(newDataBuffer, oldData)
					oldData = nil
				}
				// overwrite with the new data we want to add
				copy(newDataBuffer[offset:], data)
				dataBuffer = &newDataBuffer
			} else {
				dataBuffer = &data
			}
		}
		// WriteFromBuffer should be able to handle the case where now the block is too big and gets split into multiple blocks
		err := bb.WriteFromBuffer(name, options.Metadata, *dataBuffer)
		if err != nil {
			log.Err("BlockBlob::Write : Failed to upload to blob %s ", name, err.Error())
			return err
		}
		// case 2: given offset is within the size of the blob - and the blob consists of multiple blocks
		// case 3: new blocks need to be added
	} else {
		index, oldDataSize, exceedsFileBlocks, appendOnly := fileOffsets.FindBlocksToModify(offset, length)
		// keeps track of how much new data will be appended to the end of the file (applicable only to case 3)
		newBufferSize := int64(0)
		// case 3?
		if exceedsFileBlocks {
			newBufferSize, err = bb.createNewBlocks(fileOffsets, offset, length)
			if err != nil {
				log.Err("BlockBlob::Write : Failed to create new blocks for file %s", name, err.Error())
				return err
			}
		}
		// buffer that holds that pre-existing data in those blocks we're interested in
		oldDataBuffer := make([]byte, oldDataSize+newBufferSize)
		if !appendOnly {
			// fetch the blocks that will be impacted by the new changes so we can overwrite them
			err = bb.ReadInBuffer(name, fileOffsets.BlockList[index].StartIndex, oldDataSize, oldDataBuffer, nil)
			if err != nil {
				log.Err("BlockBlob::Write : Failed to read data in buffer %s [%s]", name, err.Error())
			}
		}
		// this gives us where the offset with respect to the buffer that holds our old data - so we can start writing the new data
		blockOffset := offset - fileOffsets.BlockList[index].StartIndex
		copy(oldDataBuffer[blockOffset:], data)
		err := bb.stageAndCommitModifiedBlocks(name, oldDataBuffer, fileOffsets)
		return err
	}
	return nil
}

// TODO: make a similar method facing stream that would enable us to write to cached blocks then stage and commit
func (bb *BlockBlob) stageAndCommitModifiedBlocks(name string, data []byte, offsetList *common.BlockOffsetList) error {
	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	blockOffset := int64(0)
	var blockIDList []string
	for _, blk := range offsetList.BlockList {
		blockIDList = append(blockIDList, blk.Id)
		if blk.Dirty() {
			_, err := blobClient.StageBlock(context.Background(),
				blk.Id,
				streaming.NopCloser(bytes.NewReader(data[blockOffset:(blk.EndIndex-blk.StartIndex)+blockOffset])),
				&blockblob.StageBlockOptions{
					CPKInfo: bb.blobCPKOpt,
				})

			if err != nil {
				log.Err("BlockBlob::stageAndCommitModifiedBlocks : Failed to stage to blob %s at block %v [%s]", name, blockOffset, err.Error())
				return err
			}
			blockOffset = (blk.EndIndex - blk.StartIndex) + blockOffset
		}
	}
	_, err := blobClient.CommitBlockList(context.Background(),
		blockIDList,
		&blockblob.CommitBlockListOptions{
			HTTPHeaders: &blob.HTTPHeaders{
				BlobContentType: to.Ptr(getContentType(name)),
			},
			Tier:    bb.Config.defaultTier,
			CPKInfo: bb.blobCPKOpt,
		})

	if err != nil {
		log.Err("BlockBlob::stageAndCommitModifiedBlocks : Failed to commit block list to blob %s [%s]", name, err.Error())
		return err
	}
	return nil
}

func (bb *BlockBlob) StageAndCommit(name string, bol *common.BlockOffsetList) error {
	// lock on the blob name so that no stage and commit race condition occur causing failure
	blobMtx := bb.blockLocks.GetLock(name)
	blobMtx.Lock()
	defer blobMtx.Unlock()

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	var blockIDList []string
	var truncBuf, data []byte
	var truncateBufAllocated = false
	staged := bol.IsBlockListModified()

	for i, blk := range bol.BlockList {
		blockIDList = append(blockIDList, blk.Id)

		if blk.Truncated() {
			if !truncateBufAllocated {
				truncBuf = make([]byte, blk.EndIndex-blk.StartIndex)
				truncateBufAllocated = true
			}

			// Match the size of the buffer to the size of the block
			if len(truncBuf) < int(blk.EndIndex-blk.StartIndex) {
				// This should not happen since we allocate the buffer once to the max size of block
				return fmt.Errorf("BlockBlob::StageAndCommit : Truncate buffer size %d is smaller than block size %d for idx: %d, id: %s for blob %s",
					len(truncBuf), blk.EndIndex-blk.StartIndex, i, blk.Id, name)
			}

			// Reslice the buffer to the size of the block, as the last block could be smaller than the rest
			data = truncBuf[:blk.EndIndex-blk.StartIndex]

			blk.Flags.Clear(common.TruncatedBlock)
		} else {
			data = blk.Data
		}

		if blk.Dirty() {
			_, err := blobClient.StageBlock(context.Background(),
				blk.Id,
				streaming.NopCloser(bytes.NewReader(data)),
				&blockblob.StageBlockOptions{
					CPKInfo: bb.blobCPKOpt,
				})
			if err != nil {
				log.Err("BlockBlob::StageAndCommit : Failed to stage to blob %s with ID %s at block %v [%s]", name, blk.Id, blk.StartIndex, err.Error())
				return err
			}
			staged = true
			blk.Flags.Clear(common.DirtyBlock)
		}
	}

	if staged {
		_, err := blobClient.CommitBlockList(context.Background(),
			blockIDList,
			&blockblob.CommitBlockListOptions{
				HTTPHeaders: &blob.HTTPHeaders{
					BlobContentType: to.Ptr(getContentType(name)),
				},
				Tier:    bb.Config.defaultTier,
				CPKInfo: bb.blobCPKOpt,
				// AccessConditions: &blob.AccessConditions{ModifiedAccessConditions: &blob.ModifiedAccessConditions{IfMatch: bol.Etag}},
			})
		if err != nil {
			log.Err("BlockBlob::StageAndCommit : Failed to commit block list to blob %s [%s]", name, err.Error())
			return err
		}
		// update the etag
		// bol.Etag = resp.ETag()
	}
	return nil
}

// ChangeMod : Change mode of a blob
func (bb *BlockBlob) ChangeMod(name string, _ os.FileMode) error {
	log.Trace("BlockBlob::ChangeMod : name %s", name)

	if bb.Config.ignoreAccessModifiers {
		// for operations like git clone where transaction fails if chmod is not successful
		// return success instead of ENOSYS
		return nil
	}

	// This is not currently supported for a flat namespace account
	return syscall.ENOTSUP
}

// ChangeOwner : Change owner of a blob
func (bb *BlockBlob) ChangeOwner(name string, _ int, _ int) error {
	log.Trace("BlockBlob::ChangeOwner : name %s", name)

	if bb.Config.ignoreAccessModifiers {
		// for operations like git clone where transaction fails if chown is not successful
		// return success instead of ENOSYS
		return nil
	}

	// This is not currently supported for a flat namespace account
	return syscall.ENOTSUP
}

// GetCommittedBlockList : Get the list of committed blocks
func (bb *BlockBlob) GetCommittedBlockList(name string) (*internal.CommittedBlockList, error) {
	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))

	storageBlockList, err := blobClient.GetBlockList(context.Background(), blockblob.BlockListTypeCommitted, nil)

	if err != nil {
		log.Err("BlockBlob::GetFileBlockOffsets : Failed to get block list %s ", name, err.Error())
		return nil, err
	}

	// if block list empty its a small file
	if len(storageBlockList.CommittedBlocks) == 0 {
		return nil, nil
	}

	blockList := make(internal.CommittedBlockList, 0)
	startOffset := int64(0)
	for _, block := range storageBlockList.CommittedBlocks {
		blk := internal.CommittedBlock{
			Id:     *block.Name,
			Offset: startOffset,
			Size:   uint64(*block.Size),
		}
		startOffset += *block.Size
		blockList = append(blockList, blk)
	}

	return &blockList, nil
}

// StageBlock : stages a block and returns its blockid
func (bb *BlockBlob) StageBlock(name string, data []byte, id string) error {
	log.Trace("BlockBlob::StageBlock : name %s, ID %v, length %v", name, id, len(data))

	ctx, cancel := context.WithTimeout(context.Background(), max_context_timeout*time.Minute)
	defer cancel()

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	_, err := blobClient.StageBlock(ctx,
		id,
		streaming.NopCloser(bytes.NewReader(data)),
		&blockblob.StageBlockOptions{
			CPKInfo: bb.blobCPKOpt,
		})

	if err != nil {
		log.Err("BlockBlob::StageBlock : Failed to stage to blob %s with ID %s [%s]", name, id, err.Error())
		return err
	}

	return nil
}

// CommitBlocks : persists the block list
func (bb *BlockBlob) CommitBlocks(name string, blockList []string, newEtag *string) error {
	log.Trace("BlockBlob::CommitBlocks : name %s", name)

	ctx, cancel := context.WithTimeout(context.Background(), max_context_timeout*time.Minute)
	defer cancel()

	blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name))
	resp, err := blobClient.CommitBlockList(ctx,
		blockList,
		&blockblob.CommitBlockListOptions{
			HTTPHeaders: &blob.HTTPHeaders{
				BlobContentType: to.Ptr(getContentType(name)),
			},
			Tier:    bb.Config.defaultTier,
			CPKInfo: bb.blobCPKOpt,
		})

	if err != nil {
		log.Err("BlockBlob::CommitBlocks : Failed to commit block list to blob %s [%s]", name, err.Error())
		return err
	}

	if newEtag != nil {
		*newEtag = sanitizeEtag(resp.ETag)
	}

	return nil
}

func (bb *BlockBlob) SetFilter(filter string) error {
	if filter == "" {
		bb.Config.filter = nil
		return nil
	}

	bb.Config.filter = &blobfilter.BlobFilter{}
	return bb.Config.filter.Configure(filter)
}
