/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.camel.component.azure.storage.blob.client;

import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import com.azure.core.http.HttpHeaders;
import com.azure.core.http.rest.PagedIterable;
import com.azure.core.http.rest.Response;
import com.azure.core.http.rest.ResponseBase;
import com.azure.core.util.Context;
import com.azure.storage.blob.BlobClient;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClient;
import com.azure.storage.blob.BlobServiceClientBuilder;
import com.azure.storage.blob.models.AccessTier;
import com.azure.storage.blob.models.AppendBlobItem;
import com.azure.storage.blob.models.AppendBlobRequestConditions;
import com.azure.storage.blob.models.BlobDownloadHeaders;
import com.azure.storage.blob.models.BlobHttpHeaders;
import com.azure.storage.blob.models.BlobImmutabilityPolicy;
import com.azure.storage.blob.models.BlobLegalHoldResult;
import com.azure.storage.blob.models.BlobProperties;
import com.azure.storage.blob.models.BlobRange;
import com.azure.storage.blob.models.BlobRequestConditions;
import com.azure.storage.blob.models.BlockBlobItem;
import com.azure.storage.blob.models.BlockList;
import com.azure.storage.blob.models.BlockListType;
import com.azure.storage.blob.models.DeleteSnapshotsOptionType;
import com.azure.storage.blob.models.DownloadRetryOptions;
import com.azure.storage.blob.models.PageBlobItem;
import com.azure.storage.blob.models.PageBlobRequestConditions;
import com.azure.storage.blob.models.PageRange;
import com.azure.storage.blob.models.PageRangeItem;
import com.azure.storage.blob.models.ParallelTransferOptions;
import com.azure.storage.blob.models.RehydratePriority;
import com.azure.storage.blob.options.BlobGetTagsOptions;
import com.azure.storage.blob.options.BlobParallelUploadOptions;
import com.azure.storage.blob.options.BlobSetTagsOptions;
import com.azure.storage.blob.options.BlobUploadFromFileOptions;
import com.azure.storage.blob.options.BlockBlobSimpleUploadOptions;
import com.azure.storage.blob.options.ListPageRangesOptions;
import com.azure.storage.blob.sas.BlobSasPermission;
import com.azure.storage.blob.sas.BlobServiceSasSignatureValues;
import com.azure.storage.blob.specialized.AppendBlobClient;
import com.azure.storage.blob.specialized.BlobClientBase;
import com.azure.storage.blob.specialized.BlobInputStream;
import com.azure.storage.blob.specialized.BlobLeaseClient;
import com.azure.storage.blob.specialized.BlobLeaseClientBuilder;
import com.azure.storage.blob.specialized.BlockBlobClient;
import com.azure.storage.blob.specialized.PageBlobClient;
import com.azure.storage.common.StorageSharedKeyCredential;
import com.azure.storage.common.Utility;
import org.apache.camel.util.ObjectHelper;
import reactor.core.publisher.Flux;

public class BlobClientWrapper {
    private static final String SERVICE_URI_SEGMENT = ".blob.core.windows.net";
    private final BlobClient client;

    public BlobClientWrapper(final BlobClient client) {
        ObjectHelper.notNull(client, "client can not be null");

        this.client = client;
    }

    public String getBlobName() {
        return client.getBlobName();
    }

    public String getBlobUrl() {
        return client.getBlobUrl();
    }

    public Response<Void> delete(
            final DeleteSnapshotsOptionType deleteBlobSnapshotOptions,
            final BlobRequestConditions requestConditions, final Duration timeout) {
        return client.deleteWithResponse(deleteBlobSnapshotOptions, requestConditions, timeout, Context.NONE);
    }

    public Map<String, Object> openInputStream(final BlobRange blobRange, final BlobRequestConditions blobRequestConditions) {

        final BlobInputStream blobInputStream = client.openInputStream(blobRange, blobRequestConditions);

        // Hack to fold the response in order to ease the mocking in the unit tests
        final Map<String, Object> results = new HashMap<>();
        results.put("inputStream", blobInputStream);
        results.put("properties", blobInputStream.getProperties());

        return results;
    }

    public ResponseBase<BlobDownloadHeaders, Void> downloadWithResponse(
            final OutputStream stream, final BlobRange range,
            final DownloadRetryOptions options, final BlobRequestConditions requestConditions, final boolean getRangeContentMd5,
            final Duration timeout) {
        return client.downloadStreamWithResponse(stream, range, options, requestConditions, getRangeContentMd5, timeout,
                Context.NONE);
    }

    public Response<BlobProperties> downloadToFileWithResponse(
            final String filePath, final BlobRange range,
            final ParallelTransferOptions parallelTransferOptions, final DownloadRetryOptions downloadRetryOptions,
            final BlobRequestConditions requestConditions,
            final boolean rangeGetContentMd5, final Duration timeout) {
        return client.downloadToFileWithResponse(filePath, range, parallelTransferOptions, downloadRetryOptions,
                requestConditions, rangeGetContentMd5, timeout, Context.NONE);
    }

    public Response<BlockBlobItem> uploadBlockBlob(
            final InputStream data, final long length, final BlobHttpHeaders headers,
            final Map<String, String> metadata, AccessTier tier, final byte[] contentMd5,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {
        Flux<ByteBuffer> dataBuffer = Utility.convertStreamToByteBuffer(data, length, 4194304, data.markSupported());
        BlockBlobSimpleUploadOptions uploadOptions = new BlockBlobSimpleUploadOptions(dataBuffer, length).setHeaders(headers)
                .setMetadata(metadata).setTier(tier).setContentMd5(contentMd5).setRequestConditions(requestConditions);
        return getBlockBlobClient().uploadWithResponse(uploadOptions, timeout, Context.NONE);
    }

    /**
     * Uploads a file to block blob using the Azure SDK's optimized file upload. This method handles chunking
     * automatically for files larger than maxSingleUploadSize.
     *
     * @param  filePath                the local file path to upload
     * @param  parallelTransferOptions options for parallel upload (block size, concurrency, etc.)
     * @param  headers                 blob HTTP headers
     * @param  metadata                blob metadata
     * @param  tier                    access tier
     * @param  requestConditions       request conditions
     * @param  timeout                 operation timeout
     * @return                         response with upload result
     */
    public Response<BlockBlobItem> uploadBlockBlobChunked(
            final String filePath,
            final ParallelTransferOptions parallelTransferOptions,
            final BlobHttpHeaders headers,
            final Map<String, String> metadata,
            final AccessTier tier,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {

        BlobUploadFromFileOptions options = new BlobUploadFromFileOptions(filePath);

        if (parallelTransferOptions != null) {
            options.setParallelTransferOptions(parallelTransferOptions);
        }
        if (headers != null) {
            options.setHeaders(headers);
        }
        if (metadata != null) {
            options.setMetadata(metadata);
        }
        if (tier != null) {
            options.setTier(tier);
        }
        if (requestConditions != null) {
            options.setRequestConditions(requestConditions);
        }

        return client.uploadFromFileWithResponse(options, timeout, Context.NONE);
    }

    /**
     * Uploads an InputStream to block blob using the Azure SDK's parallel upload. This method handles chunking
     * automatically for streams larger than maxSingleUploadSize, supporting memory-efficient uploads of large files.
     *
     * @param  data                    the input stream to upload
     * @param  length                  the exact length of the data in the stream
     * @param  parallelTransferOptions options for parallel upload (block size, concurrency, etc.)
     * @param  headers                 blob HTTP headers
     * @param  metadata                blob metadata
     * @param  tier                    access tier
     * @param  requestConditions       request conditions
     * @param  timeout                 operation timeout
     * @return                         response with upload result
     */
    public Response<BlockBlobItem> uploadBlockBlobWithParallelOptions(
            final InputStream data,
            final long length,
            final ParallelTransferOptions parallelTransferOptions,
            final BlobHttpHeaders headers,
            final Map<String, String> metadata,
            final AccessTier tier,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {

        BlobParallelUploadOptions options = new BlobParallelUploadOptions(data, length);

        if (parallelTransferOptions != null) {
            options.setParallelTransferOptions(parallelTransferOptions);
        }
        if (headers != null) {
            options.setHeaders(headers);
        }
        if (metadata != null) {
            options.setMetadata(metadata);
        }
        if (tier != null) {
            options.setTier(tier);
        }
        if (requestConditions != null) {
            options.setRequestConditions(requestConditions);
        }

        return client.uploadWithResponse(options, timeout, Context.NONE);
    }

    public HttpHeaders stageBlockBlob(
            final String base64BlockId, final InputStream data, final long length, final byte[] contentMd5,
            final String leaseId, final Duration timeout) {
        return client.getBlockBlobClient()
                .stageBlockWithResponse(base64BlockId, data, length, contentMd5, leaseId, timeout, Context.NONE).getHeaders();
    }

    public Response<BlockBlobItem> commitBlockBlob(
            final List<String> base64BlockIds, final BlobHttpHeaders headers,
            final Map<String, String> metadata, final AccessTier tier, final BlobRequestConditions requestConditions,
            final Duration timeout) {
        return getBlockBlobClient().commitBlockListWithResponse(base64BlockIds, headers, metadata, tier, requestConditions,
                timeout, Context.NONE);
    }

    public Response<BlockList> listBlobBlocks(final BlockListType listType, final String leaseId, final Duration timeout) {
        return getBlockBlobClient().listBlocksWithResponse(listType, leaseId, timeout, Context.NONE);
    }

    public Response<AppendBlobItem> createAppendBlob(
            final BlobHttpHeaders headers, final Map<String, String> metadata,
            final BlobRequestConditions requestConditions, final Duration timeout) {

        return getAppendBlobClient().createWithResponse(headers, metadata, requestConditions, timeout, Context.NONE);
    }

    public Response<AppendBlobItem> appendBlobBlock(
            final InputStream data, final long length, final byte[] contentMd5,
            final AppendBlobRequestConditions appendBlobRequestConditions, final Duration timeout) {
        return getAppendBlobClient().appendBlockWithResponse(data, length, contentMd5, appendBlobRequestConditions, timeout,
                Context.NONE);
    }

    public String copyBlob(String sourceBlobName, String accountName, String containerName, String accessKey) {
        BlobServiceClient blobServiceClient = new BlobServiceClientBuilder()
                .endpoint(String.format(Locale.ROOT, "https://%s" + SERVICE_URI_SEGMENT, accountName))
                .credential(new StorageSharedKeyCredential(accountName, accessKey))
                .buildClient();
        BlobContainerClient sourceContainerClient = blobServiceClient.getBlobContainerClient(containerName);
        BlobClient sourceBlob = sourceContainerClient.getBlobClient(sourceBlobName);

        OffsetDateTime expiryTime = OffsetDateTime.now().plusSeconds(5L);
        BlobSasPermission permission = new BlobSasPermission().setReadPermission(true);
        BlobServiceSasSignatureValues values = new BlobServiceSasSignatureValues(expiryTime, permission)
                .setStartTime(OffsetDateTime.now());
        String sasToken = sourceBlob.generateSas(values);

        return client.copyFromUrl(sourceBlob.getBlobUrl() + "?" + sasToken);
    }

    public boolean appendBlobExists() {
        return getAppendBlobClient().exists();
    }

    public Response<PageBlobItem> createPageBlob(
            final long size, final Long sequenceNumber, final BlobHttpHeaders headers,
            final Map<String, String> metadata, final BlobRequestConditions requestConditions, final Duration timeout) {
        return getPageBlobClient().createWithResponse(size, sequenceNumber, headers, metadata, requestConditions, timeout,
                Context.NONE);
    }

    public Response<PageBlobItem> uploadPageBlob(
            final PageRange pageRange, final InputStream body, final byte[] contentMd5,
            final PageBlobRequestConditions pageBlobRequestConditions, final Duration timeout) {
        return getPageBlobClient().uploadPagesWithResponse(pageRange, body, contentMd5, pageBlobRequestConditions, timeout,
                Context.NONE);
    }

    public Response<PageBlobItem> resizePageBlob(
            final long size, final BlobRequestConditions requestConditions, final Duration timeout) {
        return getPageBlobClient().resizeWithResponse(size, requestConditions, timeout, Context.NONE);
    }

    public Response<PageBlobItem> clearPagesBlob(
            final PageRange pageRange, final PageBlobRequestConditions pageBlobRequestConditions, final Duration timeout) {
        return getPageBlobClient().clearPagesWithResponse(pageRange, pageBlobRequestConditions, timeout, Context.NONE);
    }

    public PagedIterable<PageRangeItem> getPageBlobRanges(
            final BlobRange blobRange, final BlobRequestConditions requestConditions,
            final Duration timeout) {
        final ListPageRangesOptions listPageRangesOptions = new ListPageRangesOptions(blobRange);
        listPageRangesOptions.setRequestConditions(requestConditions);
        return getPageBlobClient().listPageRanges(listPageRangesOptions, timeout, Context.NONE);
    }

    public boolean pageBlobExists() {
        return getPageBlobClient().exists();
    }

    public String generateSas(final BlobServiceSasSignatureValues blobServiceSasSignatureValues) {
        return client.generateSas(blobServiceSasSignatureValues);
    }

    private BlockBlobClient getBlockBlobClient() {
        return client.getBlockBlobClient();
    }

    private AppendBlobClient getAppendBlobClient() {
        return client.getAppendBlobClient();
    }

    private PageBlobClient getPageBlobClient() {
        return client.getPageBlobClient();
    }

    public Response<BlobClientBase> createSnapshot(
            final Map<String, String> metadata,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {
        return client.createSnapshotWithResponse(metadata, requestConditions, timeout, Context.NONE);
    }

    /**
     * Returns a wrapper scoped to the given blob snapshot, or {@code this} when the snapshot id is empty. Subsequent
     * read operations on the returned wrapper target the snapshot version of the blob instead of the live one.
     */
    public BlobClientWrapper withSnapshot(final String snapshotId) {
        if (ObjectHelper.isEmpty(snapshotId)) {
            return this;
        }
        return new BlobClientWrapper(client.getSnapshotClient(snapshotId));
    }

    /**
     * Returns a wrapper scoped to the given blob version, or {@code this} when the version id is empty. Subsequent read
     * operations on the returned wrapper target the specified version of the blob instead of the live one. Requires
     * blob versioning to be enabled on the storage account.
     */
    public BlobClientWrapper withVersion(final String versionId) {
        if (ObjectHelper.isEmpty(versionId)) {
            return this;
        }
        return new BlobClientWrapper(client.getVersionClient(versionId));
    }

    public Response<Void> setTags(
            final Map<String, String> tags,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {
        BlobSetTagsOptions options = new BlobSetTagsOptions(tags);
        if (requestConditions != null) {
            options.setRequestConditions(requestConditions);
        }
        return client.setTagsWithResponse(options, timeout, Context.NONE);
    }

    public Response<Map<String, String>> getTags(
            final BlobRequestConditions requestConditions,
            final Duration timeout) {
        BlobGetTagsOptions options = new BlobGetTagsOptions();
        if (requestConditions != null) {
            options.setRequestConditions(requestConditions);
        }
        return client.getTagsWithResponse(options, timeout, Context.NONE);
    }

    public Response<BlobLegalHoldResult> setLegalHold(
            final boolean legalHold,
            final Duration timeout) {
        return client.setLegalHoldWithResponse(legalHold, timeout, Context.NONE);
    }

    public Response<BlobImmutabilityPolicy> setImmutabilityPolicy(
            final BlobImmutabilityPolicy policy,
            final BlobRequestConditions requestConditions,
            final Duration timeout) {
        return client.setImmutabilityPolicyWithResponse(policy, requestConditions, timeout, Context.NONE);
    }

    public Response<Void> undelete(final Duration timeout) {
        return client.undeleteWithResponse(timeout, Context.NONE);
    }

    public Response<Void> setAccessTier(
            final AccessTier tier,
            final RehydratePriority priority,
            final String leaseId,
            final Duration timeout) {
        return client.setAccessTierWithResponse(tier, priority, leaseId, timeout, Context.NONE);
    }

    public BlobLeaseClient getLeaseClient() {
        return new BlobLeaseClientBuilder().blobClient(client).buildClient();
    }
}
