// 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.

#pragma once

#include <gen_cpp/olap_file.pb.h>

#include <future>

#include <memory>

#include <ranges>

#include <string>

#include <tuple>

#include <unordered_map>

#include <utility>

#include <variant>

#include <vector>

#include "cloud/cloud_tablet.h"

#include "common/status.h"

#include "storage/rowset/rowset_fwd.h"

#include "storage/rowset/rowset_meta.h"

#include "util/s3_util.h"

namespace doris {

class DeleteBitmap;

class StreamLoadContext;

class CloudTablet;

class CloudTabletMgr;

class TabletMeta;

class TabletSchema;

class TabletMetaPB;

class RowsetMeta;

namespace cloud {

class FinishTabletJobResponse;

class CommitTxnResponse;

class ClusterPB;

class GetClusterStatusResponse;

class StartTabletJobResponse;

class TabletJobInfoPB;

class TabletStatsPB;

class TabletIndexPB;

class HostLevelMSRpcRateLimiters;

class MSBackpressureHandler;

using StorageVaultInfos = std::vector<

        std::tuple<std::string, std::variant<S3Conf, HdfsVaultInfo>, StorageVaultPB_PathFormat>>;

// run tasks in bthread with concurrency and wait until all tasks done

// it stops running tasks if there are any tasks return !ok, leaving some tasks untouched

// return OK if all tasks successfully done, otherwise return the result of the failed task

Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int concurrency);

// An async wrap of `bthread_fork_join` declared previously using promise-future

// return OK if fut successfully created, otherwise return error

Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concurrency,

                         std::future<Status>* fut);

// Returns the exact actual_code when recognized. An unknown actual_code uses an explicit non-OK

// legacy fallback and otherwise fails closed. Responses from a legacy Meta Service use code.

// Exposed for unit tests.

MetaServiceCode get_response_code(const MetaServiceResponseStatus& status);

// Decodes the raw status and generic cluster generations in milliseconds, plus whether the status

// generation can initialize a takeover timer. Raw values are retained for exact token comparison.

std::tuple<int64_t, int64_t, bool> get_cluster_status_mtime(const ClusterPB& cluster);

// Validates and decodes a cluster-status response. When requester identity was requested, an empty

// identity is an incomplete response and leaves both outputs unchanged so callers keep their last

// known-good cache.

Status decode_cluster_status_response(

        const GetClusterStatusResponse& response,

        std::unordered_map<std::string, std::tuple<int32_t, int64_t, int64_t, bool>>* result,

        std::string* my_cluster_id);

// Applies owner updates returned by a successful commit to tablets already cached on this BE.

// Exposed as a separate operation so direct-BE commit handling can be tested without an RPC.

void consume_commit_owner_updates(CloudTabletMgr& tablet_mgr, const CommitTxnResponse& response);

// Applies owner updates for successful commits and for a 2PC duplicate-visible replay whose first

// response was lost, while preserving the existing RPC error returned to the caller.

void consume_commit_owner_updates_after_commit(CloudTabletMgr& tablet_mgr,

                                               const Status& commit_status, bool is_2pc,

                                               const CommitTxnResponse& response);

class CloudMetaMgr {

public:

    CloudMetaMgr() = default;

    ~CloudMetaMgr() = default;

    CloudMetaMgr(const CloudMetaMgr&) = delete;

    CloudMetaMgr& operator=(const CloudMetaMgr&) = delete;

    Status get_tablet_meta(int64_t tablet_id, std::shared_ptr<TabletMeta>* tablet_meta);

    Status sync_tablet_rowsets(CloudTablet* tablet, const SyncOptions& options = {},

                               SyncRowsetStats* sync_stats = nullptr);

    Status sync_tablet_rowsets_unlocked(

            CloudTablet* tablet, std::unique_lock<bthread::Mutex>& lock /* _sync_meta_lock */,

            const SyncOptions& options = {}, SyncRowsetStats* sync_stats = nullptr);

    Status prepare_rowset(const RowsetMeta& rs_meta, const std::string& job_id, int64_t table_id,

                          std::shared_ptr<RowsetMeta>* existed_rs_meta = nullptr);

    Status commit_rowset(RowsetMeta& rs_meta, const std::string& job_id, int64_t table_id,

                         std::shared_ptr<RowsetMeta>* existed_rs_meta = nullptr,

                         RowsetMeta* attach_row_binlog = nullptr,

                         std::shared_ptr<RowsetMeta>* existed_attach_row_binlog = nullptr);

    void cache_committed_rowset(RowsetMetaSharedPtr rs_meta, int64_t expiration_time);

    Status update_tmp_rowset(const RowsetMeta& rs_meta, int64_t table_id,

                             const RowsetMeta* attach_row_binlog = nullptr);

    Status update_packed_file_info(const std::string& packed_file_path,

                                   const cloud::PackedFileInfoPB& packed_file_info,

                                   int64_t table_id);

    Status commit_txn(const StreamLoadContext& ctx, bool is_2pc);

    Status abort_txn(const StreamLoadContext& ctx);

    Status precommit_txn(const StreamLoadContext& ctx);

    /**

     * Prepares a restore job for a tablet to meta-service

     * Change the state to PREPARED

     * PREPARED state means the meta of tablet has been uploaded but not finalized.

     */

    Status prepare_restore_job(const TabletMetaPB& tablet_meta);

    /**

     * Commits a restore job for a tablet to meta-service

     * Change the state from PREPARED to COMMITTED

     * COMMITTED state means the meta of tablet has been finalized.

     */

    Status commit_restore_job(const int64_t tablet_id);

    /**

     * Finish a restore job for a tablet from meta-service

     * Change the state to final state.

     * If is_completed = true, change the state from COMMITTED to COMPLETED

     * If is_completed = false, change the state to from PREPARED/COMMITTED to DROPPED

     * COMPLETED state means the job is finished, the restored data should be visible.

     * DROPPED state means the job is aborted.

     * COMPLETED/DROPPED are the final states, jobs with final states will be recycled.

     */

    Status finish_restore_job(const int64_t tablet_id, bool is_completed);

    /**

     * Gets storage vault (storage backends) from meta-service

     * 

     * @param vault_info output param, all storage backends

     * @param is_vault_mode output param, true for pure vault mode, false for legacy mode

     * @return status

     */

    Status get_storage_vault_info(StorageVaultInfos* vault_infos, bool* is_vault_mode);

    Status prepare_tablet_job(const TabletJobInfoPB& job, StartTabletJobResponse* res);

    Status commit_tablet_job(const TabletJobInfoPB& job, FinishTabletJobResponse* res);

    Status abort_tablet_job(const TabletJobInfoPB& job);

    Status lease_tablet_job(const TabletJobInfoPB& job);

    Status update_delete_bitmap(const CloudTablet& tablet, int64_t lock_id, int64_t initiator,

                                DeleteBitmap* delete_bitmap, DeleteBitmap* delete_bitmap_v2,

                                std::string rowset_id,

                                std::optional<StorageResource> storage_resource,

                                int64_t store_version, int64_t table_id, int64_t txn_id = -1,

                                bool is_explicit_txn = false, int64_t next_visible_version = -1);

    Status cloud_update_delete_bitmap_without_lock(

            const CloudTablet& tablet, DeleteBitmap* delete_bitmap,

            std::map<std::string, int64_t>& rowset_to_versions, int64_t table_id,

            int64_t pre_rowset_agg_start_version = 0, int64_t pre_rowset_agg_end_version = 0);

    Status get_delete_bitmap_update_lock(const CloudTablet& tablet, int64_t lock_id,

                                         int64_t initiator);

    void remove_delete_bitmap_update_lock(int64_t table_id, int64_t lock_id, int64_t initiator,

                                          int64_t tablet_id);

    // Fill version holes by creating empty rowsets for missing versions

    Status fill_version_holes(CloudTablet* tablet, int64_t max_version,

                              std::unique_lock<BthreadSharedMutex>& wlock);

    // Create an empty rowset to fill a version hole

    Status create_empty_rowset_for_hole(CloudTablet* tablet, int64_t version,

                                        RowsetMetaSharedPtr prev_rowset_meta,

                                        RowsetSharedPtr* rowset);

    Status list_snapshot(std::vector<SnapshotInfoPB>& snapshots);

    Status get_snapshot_properties(SnapshotSwitchStatus& switch_status,

                                   int64_t& max_reserved_snapshots,

                                   int64_t& snapshot_interval_seconds);

    // Get all cluster status for the instance

    // Returns cluster_id -> (status, status_mtime_ms, generic_mtime_ms, status_mtime_trusted)

    // If my_cluster_id is not null, also returns the requesting node's cluster_id

    Status get_cluster_status(

            std::unordered_map<std::string, std::tuple<int32_t, int64_t, int64_t, bool>>* result,

            std::string* my_cluster_id = nullptr);

    void set_host_level_ms_rpc_rate_limiters(HostLevelMSRpcRateLimiters* limiters) {

        host_level_ms_rpc_rate_limiters_ = limiters;

    }

    void set_ms_backpressure_handler(MSBackpressureHandler* handler) {

        ms_backpressure_handler_ = handler;

    }

private:

    bool sync_tablet_delete_bitmap_by_cache(CloudTablet* tablet, std::ranges::range auto&& rs_metas,

                                            DeleteBitmap* delete_bitmap);

    Status sync_tablet_delete_bitmap(CloudTablet* tablet, int64_t old_max_version,

                                     std::ranges::range auto&& rs_metas, const TabletStatsPB& stats,

                                     const TabletIndexPB& idx, DeleteBitmap* delete_bitmap,

                                     bool full_sync, SyncRowsetStats* sync_stats,

                                     int32_t read_version, bool full_sync_v2);

    Status _read_tablet_delete_bitmap_v2(CloudTablet* tablet, int64_t old_max_version,

                                         std::ranges::range auto&& rs_metas,

                                         DeleteBitmap* delete_bitmap, GetDeleteBitmapResponse& res,

                                         int64_t& remote_delete_bitmap_bytes, bool full_sync_v2);

    Status _log_mow_delete_bitmap(CloudTablet* tablet, GetRowsetResponse& resp,

                                  DeleteBitmap& delete_bitmap, int64_t old_max_version,

                                  bool full_sync, int32_t read_version);

    Status _check_delete_bitmap_v2_correctness(CloudTablet* tablet, GetRowsetRequest& req,

                                               GetRowsetResponse& resp, int64_t old_max_version);

    Status _get_delete_bitmap_from_ms(GetDeleteBitmapRequest& req, GetDeleteBitmapResponse& res);

    Status _get_delete_bitmap_from_ms_by_batch(GetDeleteBitmapRequest& req,

                                               GetDeleteBitmapResponse& res,

                                               int64_t bytes_threadhold);

    void check_table_size_correctness(RowsetMeta& rs_meta);

    int64_t get_segment_file_size(RowsetMeta& rs_meta);

    int64_t get_inverted_index_file_size(RowsetMeta& rs_meta);

    Status do_commit_rowset(RowsetMeta& rs_meta, const std::string& job_id, int64_t table_id,

                            std::shared_ptr<RowsetMeta>* existed_rs_meta = nullptr);

    Status do_update_tmp_rowset(const RowsetMeta& rs_meta, int64_t table_id);

    HostLevelMSRpcRateLimiters* host_level_ms_rpc_rate_limiters_ {nullptr};

    MSBackpressureHandler* ms_backpressure_handler_ {nullptr};

};

} // namespace cloud

} // namespace doris