// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

#include "azure/storage/files/datalake/datalake_path_client.hpp"

#include "private/datalake_constants.hpp"
#include "private/datalake_utilities.hpp"
#include "private/package_version.hpp"

#include <azure/core/http/policies/policy.hpp>
#include <azure/storage/common/crypt.hpp>
#include <azure/storage/common/internal/constants.hpp>
#include <azure/storage/common/internal/shared_key_policy.hpp>
#include <azure/storage/common/internal/storage_bearer_token_auth.hpp>
#include <azure/storage/common/internal/storage_pipeline.hpp>
#include <azure/storage/common/internal/storage_switch_to_secondary_policy.hpp>
#include <azure/storage/common/storage_common.hpp>
#include <azure/storage/common/storage_exception.hpp>

namespace Azure { namespace Storage { namespace Files { namespace DataLake {

  DataLakePathClient DataLakePathClient::CreateFromConnectionString(
      const std::string& connectionString,
      const std::string& fileSystemName,
      const std::string& path,
      const DataLakeClientOptions& options)
  {
    auto parsedConnectionString = _internal::ParseConnectionString(connectionString);
    auto pathUrl = std::move(parsedConnectionString.DataLakeServiceUrl);
    pathUrl.AppendPath(_internal::UrlEncodePath(fileSystemName));
    pathUrl.AppendPath(_internal::UrlEncodePath(path));

    if (parsedConnectionString.KeyCredential)
    {
      return DataLakePathClient(
          pathUrl.GetAbsoluteUrl(), parsedConnectionString.KeyCredential, options);
    }
    else
    {
      return DataLakePathClient(pathUrl.GetAbsoluteUrl(), options);
    }
  }

  DataLakePathClient::DataLakePathClient(
      const std::string& pathUrl,
      std::shared_ptr<StorageSharedKeyCredential> credential,
      const DataLakeClientOptions& options)
      : m_pathUrl(pathUrl), m_blobClient(
                                _detail::GetBlobUrlFromUrl(pathUrl),
                                credential,
                                _detail::GetBlobClientOptions(options))
  {
    m_clientConfiguration.ApiVersion
        = options.ApiVersion.empty() ? _detail::ApiVersion : options.ApiVersion;
    m_clientConfiguration.CustomerProvidedKey = options.CustomerProvidedKey;
    m_clientConfiguration.UploadValidationOptions = options.UploadValidationOptions;
    m_clientConfiguration.DownloadValidationOptions = options.DownloadValidationOptions;

    _internal::BuildStoragePipelineOptions pipelineOptions;
    pipelineOptions.PackageName = _internal::DatalakeServicePackageName;
    pipelineOptions.PackageVersion = _detail::PackageVersion::ToString();
    pipelineOptions.PrimaryHost = m_pathUrl.GetHost();
    pipelineOptions.SecondaryHost = options.SecondaryHostForRetryReads;
    pipelineOptions.ApiVersion = options.ApiVersion;
    pipelineOptions.SharedKeyAuthPolicy = std::make_unique<_internal::SharedKeyPolicy>(credential);

    m_pipeline = std::make_shared<Azure::Core::Http::_internal::HttpPipeline>(
        _internal::BuildHttpPipelinePolicies(options, std::move(pipelineOptions)));
  }

  DataLakePathClient::DataLakePathClient(
      const std::string& pathUrl,
      std::shared_ptr<const Core::Credentials::TokenCredential> credential,
      const DataLakeClientOptions& options)
      : m_pathUrl(pathUrl), m_blobClient(
                                _detail::GetBlobUrlFromUrl(pathUrl),
                                credential,
                                _detail::GetBlobClientOptions(options))
  {
    m_clientConfiguration.ApiVersion
        = options.ApiVersion.empty() ? _detail::ApiVersion : options.ApiVersion;
    m_clientConfiguration.TokenCredential = credential;
    m_clientConfiguration.CustomerProvidedKey = options.CustomerProvidedKey;
    m_clientConfiguration.UploadValidationOptions = options.UploadValidationOptions;
    m_clientConfiguration.DownloadValidationOptions = options.DownloadValidationOptions;

    _internal::BuildStoragePipelineOptions pipelineOptions;
    pipelineOptions.PackageName = _internal::DatalakeServicePackageName;
    pipelineOptions.PackageVersion = _detail::PackageVersion::ToString();
    pipelineOptions.PrimaryHost = m_pathUrl.GetHost();
    pipelineOptions.SecondaryHost = options.SecondaryHostForRetryReads;
    pipelineOptions.ApiVersion = options.ApiVersion;
    {
      Azure::Core::Credentials::TokenRequestContext tokenContext;
      tokenContext.Scopes.emplace_back(
          options.Audience.HasValue()
              ? _internal::GetDefaultScopeForAudience(options.Audience.Value().ToString())
              : _internal::StorageScope);
      pipelineOptions.TokenAuthPolicy
          = std::make_unique<_internal::StorageBearerTokenAuthenticationPolicy>(
              credential, tokenContext, options.EnableTenantDiscovery);
    }

    m_pipeline = std::make_shared<Azure::Core::Http::_internal::HttpPipeline>(
        _internal::BuildHttpPipelinePolicies(options, std::move(pipelineOptions)));
  }

  DataLakePathClient::DataLakePathClient(
      const std::string& pathUrl,
      const DataLakeClientOptions& options)
      : m_pathUrl(pathUrl),
        m_blobClient(_detail::GetBlobUrlFromUrl(pathUrl), _detail::GetBlobClientOptions(options))
  {
    m_clientConfiguration.ApiVersion
        = options.ApiVersion.empty() ? _detail::ApiVersion : options.ApiVersion;
    m_clientConfiguration.CustomerProvidedKey = options.CustomerProvidedKey;
    m_clientConfiguration.UploadValidationOptions = options.UploadValidationOptions;
    m_clientConfiguration.DownloadValidationOptions = options.DownloadValidationOptions;

    _internal::BuildStoragePipelineOptions pipelineOptions;
    pipelineOptions.PackageName = _internal::DatalakeServicePackageName;
    pipelineOptions.PackageVersion = _detail::PackageVersion::ToString();
    pipelineOptions.PrimaryHost = m_pathUrl.GetHost();
    pipelineOptions.SecondaryHost = options.SecondaryHostForRetryReads;
    pipelineOptions.ApiVersion = options.ApiVersion;

    m_pipeline = std::make_shared<Azure::Core::Http::_internal::HttpPipeline>(
        _internal::BuildHttpPipelinePolicies(options, std::move(pipelineOptions)));
  }

  Azure::Response<Models::SetPathAccessControlListResult> DataLakePathClient::SetAccessControlList(
      std::vector<Models::Acl> acls,
      const SetPathAccessControlListOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::SetPathAccessControlListOptions protocolLayerOptions;
    protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
    protocolLayerOptions.Owner = options.Owner;
    protocolLayerOptions.Group = options.Group;
    protocolLayerOptions.Acl = Models::Acl::SerializeAcls(acls);
    protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
    protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    return _detail::PathClient::SetAccessControlList(
        *m_pipeline, m_pathUrl, protocolLayerOptions, context);
  }

  Azure::Response<Models::SetPathPermissionsResult> DataLakePathClient::SetPermissions(
      std::string permissions,
      const SetPathPermissionsOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::SetPathAccessControlListOptions protocolLayerOptions;
    protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
    protocolLayerOptions.Owner = options.Owner;
    protocolLayerOptions.Group = options.Group;
    protocolLayerOptions.Permissions = permissions;
    protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
    protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    return _detail::PathClient::SetAccessControlList(
        *m_pipeline, m_pathUrl, protocolLayerOptions, context);
  }

  Azure::Response<Models::SetPathHttpHeadersResult> DataLakePathClient::SetHttpHeaders(
      Models::PathHttpHeaders httpHeaders,
      const SetPathHttpHeadersOptions& options,
      const Azure::Core::Context& context) const
  {
    Blobs::SetBlobHttpHeadersOptions blobOptions;
    Blobs::Models::BlobHttpHeaders blobHttpHeaders;
    blobHttpHeaders.CacheControl = httpHeaders.CacheControl;
    blobHttpHeaders.ContentType = httpHeaders.ContentType;
    blobHttpHeaders.ContentDisposition = httpHeaders.ContentDisposition;
    blobHttpHeaders.ContentEncoding = httpHeaders.ContentEncoding;
    blobHttpHeaders.ContentLanguage = httpHeaders.ContentLanguage;
    blobOptions.AccessConditions.IfMatch = options.AccessConditions.IfMatch;
    blobOptions.AccessConditions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    blobOptions.AccessConditions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    blobOptions.AccessConditions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    blobOptions.AccessConditions.LeaseId = options.AccessConditions.LeaseId;
    auto response = m_blobClient.SetHttpHeaders(blobHttpHeaders, blobOptions, context);
    Models::SetPathHttpHeadersResult ret;
    ret.ETag = std::move(response.Value.ETag);
    ret.LastModified = std::move(response.Value.LastModified);
    return Azure::Response<Models::SetPathHttpHeadersResult>(
        std::move(ret), std::move(response.RawResponse));
  }

  Azure::Response<Models::CreatePathResult> DataLakePathClient::Create(
      Models::PathResourceType type,
      const CreatePathOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::CreatePathOptions protocolLayerOptions;
    protocolLayerOptions.Resource = type;
    protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
    protocolLayerOptions.CacheControl = options.HttpHeaders.CacheControl;
    protocolLayerOptions.ContentType = options.HttpHeaders.ContentType;
    protocolLayerOptions.ContentDisposition = options.HttpHeaders.ContentDisposition;
    protocolLayerOptions.ContentEncoding = options.HttpHeaders.ContentEncoding;
    protocolLayerOptions.ContentLanguage = options.HttpHeaders.ContentLanguage;
    protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
    protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    protocolLayerOptions.Properties = _detail::SerializeMetadata(options.Metadata);
    protocolLayerOptions.Umask = options.Umask;
    protocolLayerOptions.Permissions = options.Permissions;
    protocolLayerOptions.Owner = options.Owner;
    protocolLayerOptions.Group = options.Group;
    protocolLayerOptions.ProposedLeaseId = options.LeaseId;
    protocolLayerOptions.EncryptionContext = options.EncryptionContext;
    if (options.Acls.HasValue())
    {
      protocolLayerOptions.Acl = Models::Acl::SerializeAcls(options.Acls.Value());
    }
    if (options.LeaseDuration.HasValue())
    {
      protocolLayerOptions.LeaseDuration = static_cast<int64_t>(options.LeaseDuration->count());
    }
    AZURE_ASSERT_MSG(
        !(options.ScheduleDeletionOptions.ExpiresOn.HasValue()
          && options.ScheduleDeletionOptions.TimeToExpire.HasValue()),
        "ExpiresOn and TimeToExpire are mutually exclusive.");
    if (options.ScheduleDeletionOptions.ExpiresOn.HasValue())
    {
      protocolLayerOptions.ExpiryOptions
          = Files::DataLake::ScheduleFileExpiryOriginType::Absolute.ToString();
      protocolLayerOptions.ExpiresOn = options.ScheduleDeletionOptions.ExpiresOn.Value().ToString(
          Azure::DateTime::DateFormat::Rfc1123);
    }
    else if (options.ScheduleDeletionOptions.TimeToExpire.HasValue())
    {
      protocolLayerOptions.ExpiryOptions
          = Files::DataLake::ScheduleFileExpiryOriginType::RelativeToNow.ToString();
      protocolLayerOptions.ExpiresOn
          = std::to_string(options.ScheduleDeletionOptions.TimeToExpire.Value().count());
    }
    if (m_clientConfiguration.CustomerProvidedKey.HasValue())
    {
      protocolLayerOptions.EncryptionKey = m_clientConfiguration.CustomerProvidedKey.Value().Key;
      protocolLayerOptions.EncryptionKeySha256
          = m_clientConfiguration.CustomerProvidedKey.Value().KeyHash;
      protocolLayerOptions.EncryptionAlgorithm
          = m_clientConfiguration.CustomerProvidedKey.Value().Algorithm.ToString();
    }
    return _detail::PathClient::Create(*m_pipeline, m_pathUrl, protocolLayerOptions, context);
  }

  Azure::Response<Models::CreatePathResult> DataLakePathClient::CreateIfNotExists(
      Models::PathResourceType type,
      const CreatePathOptions& options,
      const Azure::Core::Context& context) const
  {
    try
    {
      auto createOptions = options;
      createOptions.AccessConditions.IfNoneMatch = Azure::ETag::Any();
      return Create(type, createOptions, context);
    }
    catch (StorageException& e)
    {
      if (e.ErrorCode == _detail::DataLakePathAlreadyExists)
      {
        Models::CreatePathResult ret;
        ret.Created = false;
        return Azure::Response<Models::CreatePathResult>(std::move(ret), std::move(e.RawResponse));
      }
      throw;
    }
  }

  Azure::Response<Models::DeletePathResult> DataLakePathClient::Delete(
      const DeletePathOptions& options,
      const Azure::Core::Context& context) const
  {
    bool paginated = m_clientConfiguration.ApiVersion >= "2023-08-03"
        && m_clientConfiguration.TokenCredential && options.Recursive.HasValue()
        && options.Recursive.Value();
    std::string continuationToken;
    while (true)
    {
      _detail::PathClient::DeletePathOptions protocolLayerOptions;
      protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
      protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
      protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
      protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
      protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
      protocolLayerOptions.Recursive = options.Recursive;
      protocolLayerOptions.ContinuationToken = continuationToken;
      if (options.Recursive.HasValue())
      {
        protocolLayerOptions.Paginated = paginated;
      }
      auto response
          = _detail::PathClient::Delete(*m_pipeline, m_pathUrl, protocolLayerOptions, context);
      continuationToken = Azure::Core::Http::_internal::HttpShared::GetHeaderOrEmptyString(
          response.RawResponse->GetHeaders(), "x-ms-continuation");
      if (continuationToken.empty())
      {
        return response;
      }
    }
  }

  Azure::Response<Models::DeletePathResult> DataLakePathClient::DeleteIfExists(
      const DeletePathOptions& options,
      const Azure::Core::Context& context) const
  {
    try
    {
      return Delete(options, context);
    }
    catch (StorageException& e)
    {
      if (e.ErrorCode == _detail::DataLakeFilesystemNotFound
          || e.ErrorCode == _detail::DataLakePathNotFound)
      {
        Models::DeletePathResult ret;
        ret.Deleted = false;
        return Azure::Response<Models::DeletePathResult>(std::move(ret), std::move(e.RawResponse));
      }
      throw;
    }
  }

  Azure::Response<Models::PathProperties> DataLakePathClient::GetProperties(
      const GetPathPropertiesOptions& options,
      const Azure::Core::Context& context) const
  {
    Blobs::GetBlobPropertiesOptions blobOptions;
    blobOptions.AccessConditions.IfMatch = options.AccessConditions.IfMatch;
    blobOptions.AccessConditions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    blobOptions.AccessConditions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    blobOptions.AccessConditions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    blobOptions.AccessConditions.LeaseId = options.AccessConditions.LeaseId;
    auto response = m_blobClient.GetProperties(
        blobOptions,
        options.IncludeUserPrincipalName.HasValue() ? context.WithValue(
            Blobs::_detail::DataLakeInteroperabilityExtraOptionsKey,
            options.IncludeUserPrincipalName.Value())
                                                    : context);
    Models::PathProperties ret;
    ret.ETag = std::move(response.Value.ETag);
    ret.LastModified = std::move(response.Value.LastModified);
    ret.CreatedOn = std::move(response.Value.CreatedOn);
    ret.Metadata = std::move(response.Value.Metadata);
    ret.LeaseDuration = std::move(response.Value.LeaseDuration);
    ret.LeaseState = std::move(response.Value.LeaseState);
    ret.LeaseStatus = std::move(response.Value.LeaseStatus);
    ret.HttpHeaders = std::move(response.Value.HttpHeaders);
    ret.IsServerEncrypted = response.Value.IsServerEncrypted;
    ret.EncryptionKeySha256 = std::move(response.Value.EncryptionKeySha256);
    ret.EncryptionScope = std::move(response.Value.EncryptionScope);
    ret.CopyId = std::move(response.Value.CopyId);
    ret.CopySource = std::move(response.Value.CopySource);
    ret.CopyStatus = std::move(response.Value.CopyStatus);
    ret.CopyProgress = std::move(response.Value.CopyProgress);
    ret.CopyCompletedOn = std::move(response.Value.CopyCompletedOn);
    ret.ExpiresOn = std::move(response.Value.ExpiresOn);
    ret.LastAccessedOn = std::move(response.Value.LastAccessedOn);
    ret.FileSize = response.Value.BlobSize;
    ret.ArchiveStatus = std::move(response.Value.ArchiveStatus);
    ret.RehydratePriority = std::move(response.Value.RehydratePriority);
    ret.CopyStatusDescription = std::move(response.Value.CopyStatusDescription);
    ret.IsIncrementalCopy = std::move(response.Value.IsIncrementalCopy);
    ret.IncrementalCopyDestinationSnapshot
        = std::move(response.Value.IncrementalCopyDestinationSnapshot);
    ret.VersionId = std::move(response.Value.VersionId);
    ret.IsCurrentVersion = std::move(response.Value.IsCurrentVersion);
    ret.IsDirectory = _detail::MetadataIndicatesIsDirectory(ret.Metadata);
    auto& headers = response.RawResponse->GetHeaders();
    auto encryptionContext = headers.find(_detail::EncryptionContextHeaderName);
    if (encryptionContext != headers.end())
    {
      ret.EncryptionContext = encryptionContext->second;
    }
    auto acl = headers.find(_detail::AclHeaderName);
    if (acl != headers.end())
    {
      ret.Acls = Models::Acl::DeserializeAcls(acl->second);
    }
    auto owner = headers.find(_detail::OwnerHeaderName);
    if (owner != headers.end())
    {
      ret.Owner = owner->second;
    }
    auto group = headers.find(_detail::GroupHeaderName);
    if (group != headers.end())
    {
      ret.Group = group->second;
    }
    auto permissions = headers.find(_detail::PermissionsHeaderName);
    if (permissions != headers.end())
    {
      ret.Permissions = permissions->second;
    }
    return Azure::Response<Models::PathProperties>(std::move(ret), std::move(response.RawResponse));
  }

  Azure::Response<Models::PathAccessControlList> DataLakePathClient::GetAccessControlList(
      const GetPathAccessControlListOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::GetPathAccessControlListOptions protocolLayerOptions;
    protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
    protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
    protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    protocolLayerOptions.Upn = options.IncludeUserPrincipalName;
    auto response = _detail::PathClient::GetAccessControlList(
        *m_pipeline, m_pathUrl, protocolLayerOptions, _internal::WithReplicaStatus(context));
    Azure::Nullable<std::vector<Models::Acl>> acl;
    if (!response.Value.Acl.empty())
    {
      acl = Models::Acl::DeserializeAcls(response.Value.Acl);
    }
    Models::PathAccessControlList ret;
    if (!acl.HasValue())
    {
      throw Azure::Core::RequestFailedException(
          "Got null value returned when getting access control.");
    }
    ret.Acls = std::move(acl.Value());
    ret.Owner = std::move(response.Value.Owner);
    ret.Group = std::move(response.Value.Group);
    ret.Permissions = std::move(response.Value.Permissions);
    return Azure::Response<Models::PathAccessControlList>(
        std::move(ret), std::move(response.RawResponse));
  }

  Azure::Response<Models::PathSystemProperties> DataLakePathClient::GetSystemProperties(
      const GetPathSystemPropertiesOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::GetPathSystemPropertiesOptions protocolLayerOptions;
    protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
    protocolLayerOptions.IfMatch = options.AccessConditions.IfMatch;
    protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    protocolLayerOptions.Upn = options.IncludeUserPrincipalName;
    auto response = _detail::PathClient::GetSystemProperties(
        *m_pipeline, m_pathUrl, protocolLayerOptions, _internal::WithReplicaStatus(context));
    if (response.RawResponse->GetHeaders().at("x-ms-resource-type") == "file")
    {
      response.Value.IsDirectory = false;
    }
    else
    {
      response.Value.IsDirectory = true;
    }
    return response;
  }

  Azure::Response<Models::SetPathMetadataResult> DataLakePathClient::SetMetadata(
      Storage::Metadata metadata,
      const SetPathMetadataOptions& options,
      const Azure::Core::Context& context) const
  {
    Blobs::SetBlobMetadataOptions blobOptions;
    blobOptions.AccessConditions.IfMatch = options.AccessConditions.IfMatch;
    blobOptions.AccessConditions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
    blobOptions.AccessConditions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
    blobOptions.AccessConditions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
    blobOptions.AccessConditions.LeaseId = options.AccessConditions.LeaseId;
    auto response = m_blobClient.SetMetadata(std::move(metadata), blobOptions, context);
    Models::SetPathMetadataResult ret;
    ret.ETag = std::move(response.Value.ETag);
    ret.LastModified = std::move(response.Value.LastModified);
    return Azure::Response<Models::SetPathMetadataResult>(
        std::move(ret), std::move(response.RawResponse));
  }

  SetPathAccessControlListRecursivePagedResponse
  DataLakePathClient::SetAccessControlListRecursiveInternal(
      Models::_detail::PathSetAccessControlListRecursiveMode mode,
      const std::vector<Models::Acl>& acls,
      const SetPathAccessControlListRecursiveOptions& options,
      const Azure::Core::Context& context) const
  {
    _detail::PathClient::SetPathAccessControlListRecursiveOptions protocolLayerOptions;
    protocolLayerOptions.Mode = mode.ToString();
    protocolLayerOptions.ContinuationToken = options.ContinuationToken;
    protocolLayerOptions.MaxRecords = options.PageSizeHint;
    protocolLayerOptions.ForceFlag = options.ContinueOnFailure;
    protocolLayerOptions.Acl = Models::Acl::SerializeAcls(acls);
    auto response = _detail::PathClient::SetAccessControlListRecursive(
        *m_pipeline, m_pathUrl, protocolLayerOptions, context);

    SetPathAccessControlListRecursivePagedResponse pagedResponse;
    pagedResponse.NumberOfSuccessfulFiles = response.Value.NumberOfSuccessfulDirectories;
    pagedResponse.NumberOfSuccessfulFiles = response.Value.NumberOfSuccessfulFiles;
    pagedResponse.NumberOfFailures = response.Value.NumberOfFailures;
    pagedResponse.FailedEntries = std::move(response.Value.FailedEntries);
    pagedResponse.m_dataLakePathClient = std::make_shared<DataLakePathClient>(*this);
    pagedResponse.m_operationOptions = options;
    pagedResponse.m_acls = acls;
    pagedResponse.m_mode = mode;
    pagedResponse.CurrentPageToken = options.ContinuationToken.ValueOr(std::string());
    pagedResponse.NextPageToken = response.Value.ContinuationToken;
    pagedResponse.RawResponse = std::move(response.RawResponse);

    return pagedResponse;
  }

  Azure::Response<Models::SetPathTagsResult> DataLakePathClient::SetTags(
      std::map<std::string, std::string> tags,
      const SetPathTagsOptions& options,
      const Azure::Core::Context& context) const
  {
    return m_blobClient.SetTags(tags, options, context);
  }

  Azure::Response<std::map<std::string, std::string>> DataLakePathClient::GetTags(
      const GetPathTagsOptions& options,
      const Azure::Core::Context& context) const
  {
    return m_blobClient.GetTags(options, context);
  }

}}}} // namespace Azure::Storage::Files::DataLake
