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

#include "src/buildtool/execution_api/execution_service/cas_utils.hpp"

#include <utility>

#include "fmt/core.h"
#include "src/buildtool/file_system/file_system_manager.hpp"
#include "src/buildtool/storage/large_object_cas.hpp"
#include "src/buildtool/storage/local_cas.hpp"

static auto ToGrpc(LargeObjectError&& error) noexcept -> grpc::Status {
    switch (error.Code()) {
        case LargeObjectErrorCode::Internal:
            return grpc::Status{grpc::StatusCode::INTERNAL,
                                std::move(error).Message()};
        case LargeObjectErrorCode::FileNotFound:
            return grpc::Status{grpc::StatusCode::NOT_FOUND,
                                std::move(error).Message()};
        case LargeObjectErrorCode::InvalidResult:
        case LargeObjectErrorCode::InvalidTree:
            return grpc::Status{grpc::StatusCode::FAILED_PRECONDITION,
                                std::move(error).Message()};
    }
    return grpc::Status{grpc::StatusCode::INTERNAL, "an unknown error"};
}

auto CASUtils::EnsureTreeInvariant(ArtifactDigest const& digest,
                                   std::string const& tree_data,
                                   Storage const& storage) noexcept
    -> std::optional<std::string> {
    auto error = storage.CAS().CheckTreeInvariant(digest, tree_data);
    if (error) {
        return std::move(*error).Message();
    }
    return std::nullopt;
}

auto CASUtils::SplitBlobIdentity(ArtifactDigest const& blob_digest,
                                 Storage const& storage) noexcept
    -> expected<std::vector<ArtifactDigest>, grpc::Status> {
    // Check blob existence.
    auto path = blob_digest.IsTree()
                    ? storage.CAS().TreePath(blob_digest)
                    : storage.CAS().BlobPath(blob_digest, false);
    if (not path) {
        return unexpected{
            grpc::Status{grpc::StatusCode::NOT_FOUND,
                         fmt::format("blob not found {}", blob_digest.hash())}};
    }

    // The split protocol states that each chunk that is returned by the
    // operation is stored in (file) CAS. This means for the native mode, if we
    // return the identity of a tree, we need to put the tree data in file CAS
    // and return the resulting digest.
    auto chunk_digests = std::vector<ArtifactDigest>{};
    if (blob_digest.IsTree()) {
        auto tree_data = FileSystemManager::ReadFile(*path);
        if (not tree_data) {
            return unexpected{grpc::Status{
                grpc::StatusCode::INTERNAL,
                fmt::format("could read tree data {}", blob_digest.hash())}};
        }
        auto digest = storage.CAS().StoreBlob(*tree_data, false);
        if (not digest) {
            return unexpected{
                grpc::Status{grpc::StatusCode::INTERNAL,
                             fmt::format("could not store tree as blob {}",
                                         blob_digest.hash())}};
        }
        chunk_digests.emplace_back(*digest);
        return chunk_digests;
    }
    chunk_digests.emplace_back(blob_digest);
    return chunk_digests;
}

auto CASUtils::SplitBlobFastCDC(ArtifactDigest const& blob_digest,
                                Storage const& storage) noexcept
    -> expected<std::vector<ArtifactDigest>, grpc::Status> {
    // Split blob into chunks:
    auto split = blob_digest.IsTree() ? storage.CAS().SplitTree(blob_digest)
                                      : storage.CAS().SplitBlob(blob_digest);

    if (not split) {
        return unexpected{ToGrpc(std::move(split).error())};
    }
    return *std::move(split);
}

auto CASUtils::SpliceBlob(ArtifactDigest const& blob_digest,
                          std::vector<ArtifactDigest> const& chunk_digests,
                          Storage const& storage) noexcept
    -> expected<ArtifactDigest, grpc::Status> {
    // Splice blob from chunks:
    auto splice =
        blob_digest.IsTree()
            ? storage.CAS().SpliceTree(blob_digest, chunk_digests)
            : storage.CAS().SpliceBlob(blob_digest, chunk_digests, false);

    if (not splice) {
        return unexpected{ToGrpc(std::move(splice).error())};
    }
    return *std::move(splice);
}