1 /* 2 * Copyright (c) 2023-2024 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef OHOS_DISTRIBUTED_DATA_SERVICES_CLOUD_SYNC_MANAGER_H 17 #define OHOS_DISTRIBUTED_DATA_SERVICES_CLOUD_SYNC_MANAGER_H 18 19 #include "cloud/cloud_event.h" 20 #include "cloud/cloud_info.h" 21 #include "cloud/sync_strategy.h" 22 #include "cloud_types.h" 23 #include "concurrent_map.h" 24 #include "eventcenter/event.h" 25 #include "executor_pool.h" 26 #include "metadata/store_meta_data_local.h" 27 #include "store/auto_cache.h" 28 #include "store/general_store.h" 29 #include "store/general_value.h" 30 #include "utils/ref_count.h" 31 32 namespace OHOS::CloudData { 33 class SyncManager { 34 public: 35 using GenAsync = DistributedData::GenAsync; 36 using GenStore = DistributedData::GeneralStore; 37 using GenQuery = DistributedData::GenQuery; 38 using RefCount = DistributedData::RefCount; 39 using AutoCache = DistributedData::AutoCache; 40 using StoreMetaData = DistributedData::StoreMetaData; 41 using SchemaMeta = DistributedData::SchemaMeta; 42 using TraceIds = std::map<std::string, std::string>; 43 using SyncStage = DistributedData::SyncStage; 44 using ReportParam = DistributedData::ReportParam; 45 static AutoCache::Store GetStore(const StoreMetaData &meta, int32_t user, bool mustBind = true); 46 class SyncInfo final { 47 public: 48 using Store = std::string; 49 using Stores = std::vector<Store>; 50 using Tables = std::vector<std::string>; 51 struct Param { 52 int32_t user; 53 std::string bundleName; 54 Store store; 55 Tables tables; 56 int32_t triggerMode = 0; 57 std::string prepareTraceId; 58 }; 59 using MutliStoreTables = std::map<Store, Tables>; 60 explicit SyncInfo(int32_t user, const std::string &bundleName = "", const Store &store = "", 61 const Tables &tables = {}, int32_t triggerMode = 0); 62 SyncInfo(int32_t user, const std::string &bundleName, const Stores &stores); 63 SyncInfo(int32_t user, const std::string &bundleName, const MutliStoreTables &tables); 64 explicit SyncInfo(const Param ¶m); 65 void SetMode(int32_t mode); 66 void SetWait(int32_t wait); 67 void SetAsyncDetail(GenAsync asyncDetail); 68 void SetQuery(std::shared_ptr<GenQuery> query); 69 void SetError(int32_t code) const; 70 void SetCompensation(bool isCompensation); 71 void SetTriggerMode(int32_t triggerMode); 72 void SetPrepareTraceId(const std::string &prepareTraceId); 73 std::shared_ptr<GenQuery> GenerateQuery(const std::string &store, const Tables &tables); 74 bool Contains(const std::string &storeName); 75 inline static constexpr const char *DEFAULT_ID = "default"; 76 77 private: 78 friend SyncManager; 79 uint64_t syncId_ = 0; 80 int32_t mode_ = GenStore::MixMode(GenStore::CLOUD_TIME_FIRST, GenStore::AUTO_SYNC_MODE); 81 int32_t user_ = 0; 82 int32_t wait_ = 0; 83 std::string id_ = DEFAULT_ID; 84 std::string bundleName_; 85 std::map<std::string, std::vector<std::string>> tables_; 86 GenAsync async_; 87 std::shared_ptr<GenQuery> query_; 88 bool isCompensation_ = false; 89 int32_t triggerMode_ = 0; 90 std::string prepareTraceId_; 91 }; 92 SyncManager(); 93 ~SyncManager(); 94 int32_t Bind(std::shared_ptr<ExecutorPool> executor); 95 int32_t DoCloudSync(SyncInfo syncInfo); 96 int32_t StopCloudSync(int32_t user = 0); 97 int32_t QueryLastSyncInfo(const std::vector<QueryKey> &queryKeys, QueryLastResults &results); 98 void Report(const ReportParam &reportParam); 99 100 private: 101 using Event = DistributedData::Event; 102 using Task = ExecutorPool::Task; 103 using TaskId = ExecutorPool::TaskId; 104 using Duration = ExecutorPool::Duration; 105 using Retryer = 106 std::function<bool(Duration interval, int32_t status, int32_t dbCode, const std::string &prepareTraceId)>; 107 using CloudInfo = DistributedData::CloudInfo; 108 using StoreInfo = DistributedData::StoreInfo; 109 using SyncStrategy = DistributedData::SyncStrategy; 110 using SyncId = uint64_t; 111 using GeneralError = DistributedData::GeneralError; 112 using GenProgress = DistributedData::GenProgress; 113 using GenDetails = DistributedData::GenDetails; 114 115 static constexpr ExecutorPool::Duration RETRY_INTERVAL = std::chrono::seconds(10); // second 116 static constexpr ExecutorPool::Duration LOCKED_INTERVAL = std::chrono::seconds(30); // second 117 static constexpr ExecutorPool::Duration BUSY_INTERVAL = std::chrono::seconds(180); // second 118 static constexpr int32_t RETRY_TIMES = 6; // normal retry 119 static constexpr int32_t CLIENT_RETRY_TIMES = 3; // normal retry 120 static constexpr uint64_t USER_MARK = 0xFFFFFFFF00000000; // high 32 bit 121 static constexpr int32_t MV_BIT = 32; 122 static constexpr int32_t EXPIRATION_TIME = 6 * 60 * 60 * 1000; // 6 hours 123 124 static uint64_t GenerateId(int32_t user); 125 static ExecutorPool::Duration GetInterval(int32_t code); 126 static std::map<uint32_t, GenStore::BindInfo> GetBindInfos(const StoreMetaData &meta, 127 const std::vector<int32_t> &users, const DistributedData::Database &schemaDatabase); 128 static std::string GetAccountId(int32_t user); 129 static std::vector<std::tuple<QueryKey, uint64_t>> GetCloudSyncInfo(const SyncInfo &info, CloudInfo &cloud); 130 static std::vector<SchemaMeta> GetSchemaMeta(const CloudInfo &cloud, const std::string &bundleName); 131 static bool NeedGetCloudInfo(CloudInfo &cloud); 132 Task GetSyncTask(int32_t times, bool retry, RefCount ref, SyncInfo &&syncInfo); 133 void UpdateSchema(const SyncInfo &syncInfo); 134 std::function<void(const Event &)> GetSyncHandler(Retryer retryer); 135 std::function<void(const Event &)> GetClientChangeHandler(); 136 Retryer GetRetryer(int32_t times, const SyncInfo &syncInfo, int32_t user); 137 RefCount GenSyncRef(uint64_t syncId); 138 int32_t Compare(uint64_t syncId, int32_t user); 139 GeneralError IsValid(SyncInfo &info, CloudInfo &cloud); 140 void UpdateStartSyncInfo(const std::vector<std::tuple<QueryKey, uint64_t>> &cloudSyncInfos); 141 void UpdateFinishSyncInfo(const QueryKey &queryKey, uint64_t syncId, int32_t code); 142 std::function<void(const DistributedData::GenDetails &result)> GetCallback(const GenAsync &async, 143 const StoreInfo &storeInfo, int32_t triggerMode, const std::string &prepareTraceId, int32_t user); 144 std::function<void()> GetPostEventTask(const std::vector<SchemaMeta> &schemas, CloudInfo &cloud, SyncInfo &info, 145 bool retry, const TraceIds &traceIds); 146 void DoExceptionalCallback(const GenAsync &async, GenDetails &details, const StoreInfo &storeInfo, 147 const std::string &prepareTraceId); 148 bool InitDefaultUser(int32_t &user); 149 std::function<void(const DistributedData::GenDetails &result)> RetryCallback(const StoreInfo &storeInfo, 150 Retryer retryer, int32_t triggerMode, const std::string &prepareTraceId, int32_t user); 151 static void GetLastResults( 152 const std::string &storeId, std::map<SyncId, CloudSyncInfo> &infos, QueryLastResults &results); 153 void BatchUpdateFinishState(const std::vector<std::tuple<QueryKey, uint64_t>> &cloudSyncInfos, int32_t code); 154 bool NeedSaveSyncInfo(const QueryKey &queryKey); 155 std::function<void(const Event &)> GetLockChangeHandler(); 156 void BatchReport(int32_t userId, const TraceIds &traceIds, SyncStage syncStage, int32_t errCode); 157 TraceIds GetPrepareTraceId(const SyncInfo &info, const CloudInfo &cloud); 158 std::pair<bool, StoreMetaData> GetMetaData(const StoreInfo &storeInfo); 159 160 static std::atomic<uint32_t> genId_; 161 std::shared_ptr<ExecutorPool> executor_; 162 ConcurrentMap<uint64_t, TaskId> actives_; 163 ConcurrentMap<uint64_t, uint64_t> activeInfos_; 164 std::shared_ptr<SyncStrategy> syncStrategy_; 165 ConcurrentMap<QueryKey, std::map<SyncId, CloudSyncInfo>> lastSyncInfos_; 166 std::set<std::string> kvApps_; 167 }; 168 } // namespace OHOS::CloudData 169 #endif // OHOS_DISTRIBUTED_DATA_SERVICES_CLOUD_SYNC_MANAGER_H