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/cloud_last_sync_info.h" 22 #include "cloud/sync_strategy.h" 23 #include "cloud_types.h" 24 #include "cloud/sync_event.h" 25 #include "concurrent_map.h" 26 #include "eventcenter/event.h" 27 #include "executor_pool.h" 28 #include "metadata/store_meta_data_local.h" 29 #include "store/auto_cache.h" 30 #include "store/general_store.h" 31 #include "store/general_value.h" 32 #include "utils/ref_count.h" 33 #include "radar_reporter.h" 34 35 namespace OHOS::CloudData { 36 class SyncManager { 37 public: 38 using CloudLastSyncInfo = DistributedData::CloudLastSyncInfo; 39 using GenAsync = DistributedData::GenAsync; 40 using GenStore = DistributedData::GeneralStore; 41 using GenQuery = DistributedData::GenQuery; 42 using RefCount = DistributedData::RefCount; 43 using AutoCache = DistributedData::AutoCache; 44 using StoreMetaData = DistributedData::StoreMetaData; 45 using SchemaMeta = DistributedData::SchemaMeta; 46 using TraceIds = std::map<std::string, std::string>; 47 using SyncStage = DistributedData::SyncStage; 48 using ReportParam = DistributedData::ReportParam; 49 static std::pair<int32_t, AutoCache::Store> GetStore(const StoreMetaData &meta, int32_t user, bool mustBind = true); 50 class SyncInfo final { 51 public: 52 using Store = std::string; 53 using Stores = std::vector<Store>; 54 using Tables = std::vector<std::string>; 55 struct Param { 56 int32_t user; 57 std::string bundleName; 58 Store store; 59 Tables tables; 60 int32_t triggerMode = 0; 61 std::string prepareTraceId; 62 }; 63 using MutliStoreTables = std::map<Store, Tables>; 64 explicit SyncInfo(int32_t user, const std::string &bundleName = "", const Store &store = "", 65 const Tables &tables = {}, int32_t triggerMode = 0); 66 SyncInfo(int32_t user, const std::string &bundleName, const Stores &stores); 67 SyncInfo(int32_t user, const std::string &bundleName, const MutliStoreTables &tables); 68 explicit SyncInfo(const Param ¶m); 69 void SetMode(int32_t mode); 70 void SetWait(int32_t wait); 71 void SetAsyncDetail(GenAsync asyncDetail); 72 void SetQuery(std::shared_ptr<GenQuery> query); 73 void SetError(int32_t code) const; 74 void SetCompensation(bool isCompensation); 75 void SetTriggerMode(int32_t triggerMode); 76 void SetPrepareTraceId(const std::string &prepareTraceId); 77 std::shared_ptr<GenQuery> GenerateQuery(const std::string &store, const Tables &tables); 78 bool Contains(const std::string &storeName); 79 inline static constexpr const char *DEFAULT_ID = "default"; 80 81 private: 82 friend SyncManager; 83 uint64_t syncId_ = 0; 84 int32_t mode_ = GenStore::MixMode(GenStore::CLOUD_TIME_FIRST, GenStore::AUTO_SYNC_MODE); 85 int32_t user_ = 0; 86 int32_t wait_ = 0; 87 std::string id_ = DEFAULT_ID; 88 std::string bundleName_; 89 std::map<std::string, std::vector<std::string>> tables_; 90 GenAsync async_; 91 std::shared_ptr<GenQuery> query_; 92 bool isCompensation_ = false; 93 int32_t triggerMode_ = 0; 94 std::string prepareTraceId_; 95 }; 96 SyncManager(); 97 ~SyncManager(); 98 int32_t Bind(std::shared_ptr<ExecutorPool> executor); 99 int32_t DoCloudSync(SyncInfo syncInfo); 100 int32_t StopCloudSync(int32_t user = 0); 101 std::pair<int32_t, std::map<std::string, CloudLastSyncInfo>> QueryLastSyncInfo( 102 const std::vector<QueryKey> &queryKeys); 103 void Report(const ReportParam &reportParam); 104 void OnScreenUnlocked(int32_t user); 105 void CleanCompensateSync(int32_t userId); 106 107 private: 108 using Event = DistributedData::Event; 109 using Task = ExecutorPool::Task; 110 using TaskId = ExecutorPool::TaskId; 111 using Duration = ExecutorPool::Duration; 112 using Retryer = 113 std::function<bool(Duration interval, int32_t status, int32_t dbCode, const std::string &prepareTraceId)>; 114 using CloudInfo = DistributedData::CloudInfo; 115 using StoreInfo = DistributedData::StoreInfo; 116 using SyncStrategy = DistributedData::SyncStrategy; 117 using SyncId = uint64_t; 118 using GeneralError = DistributedData::GeneralError; 119 using GenProgress = DistributedData::GenProgress; 120 using GenDetails = DistributedData::GenDetails; 121 122 static constexpr ExecutorPool::Duration RETRY_INTERVAL = std::chrono::seconds(10); // second 123 static constexpr ExecutorPool::Duration LOCKED_INTERVAL = std::chrono::seconds(30); // second 124 static constexpr ExecutorPool::Duration BUSY_INTERVAL = std::chrono::seconds(180); // second 125 static constexpr int32_t RETRY_TIMES = 6; // normal retry 126 static constexpr int32_t CLIENT_RETRY_TIMES = 3; // normal retry 127 static constexpr uint64_t USER_MARK = 0xFFFFFFFF00000000; // high 32 bit 128 static constexpr int32_t MV_BIT = 32; 129 static constexpr int32_t EXPIRATION_TIME = 6 * 60 * 60 * 1000; // 6 hours 130 131 static uint64_t GenerateId(int32_t user); 132 static ExecutorPool::Duration GetInterval(int32_t code); 133 static std::map<uint32_t, GenStore::BindInfo> GetBindInfos( 134 const StoreMetaData &meta, const std::vector<int32_t> &users, const DistributedData::Database &schemaDatabase); 135 static std::string GetAccountId(int32_t user); 136 static std::vector<std::tuple<QueryKey, uint64_t>> GetCloudSyncInfo(const SyncInfo &info, CloudInfo &cloud); 137 static std::vector<SchemaMeta> GetSchemaMeta(const CloudInfo &cloud, const std::string &bundleName); 138 static bool NeedGetCloudInfo(CloudInfo &cloud); 139 static GeneralError IsValid(SyncInfo &info, CloudInfo &cloud); 140 Task GetSyncTask(int32_t times, bool retry, RefCount ref, SyncInfo &&syncInfo); 141 void UpdateSchema(const SyncInfo &syncInfo); 142 std::function<void(const Event &)> GetSyncHandler(Retryer retryer); 143 std::function<void(const Event &)> GetClientChangeHandler(); 144 Retryer GetRetryer(int32_t times, const SyncInfo &syncInfo, int32_t user); 145 RefCount GenSyncRef(uint64_t syncId); 146 int32_t Compare(uint64_t syncId, int32_t user); 147 void UpdateStartSyncInfo(const std::vector<std::tuple<QueryKey, uint64_t>> &cloudSyncInfos); 148 void UpdateFinishSyncInfo(const QueryKey &queryKey, uint64_t syncId, int32_t code); 149 std::function<void(const DistributedData::GenDetails &result)> GetCallback(const GenAsync &async, 150 const StoreInfo &storeInfo, int32_t triggerMode, const std::string &prepareTraceId, int32_t user); 151 std::function<void()> GetPostEventTask(const std::vector<SchemaMeta> &schemas, CloudInfo &cloud, SyncInfo &info, 152 bool retry, const TraceIds &traceIds); 153 void DoExceptionalCallback(const GenAsync &async, GenDetails &details, const StoreInfo &storeInfo, 154 const std::string &prepareTraceId, int32_t code = GeneralError::E_ERROR); 155 bool InitDefaultUser(int32_t &user); 156 std::function<void(const DistributedData::GenDetails &result)> RetryCallback(const StoreInfo &storeInfo, 157 Retryer retryer, int32_t triggerMode, const std::string &prepareTraceId, int32_t user); 158 static std::pair<int32_t, CloudLastSyncInfo> GetLastResults(std::map<SyncId, CloudLastSyncInfo> &infos); 159 void BatchUpdateFinishState(const std::vector<std::tuple<QueryKey, uint64_t>> &cloudSyncInfos, int32_t code); 160 bool NeedSaveSyncInfo(const QueryKey &queryKey); 161 std::function<void(const Event &)> GetLockChangeHandler(); 162 void BatchReport(int32_t userId, const TraceIds &traceIds, SyncStage syncStage, int32_t errCode); 163 TraceIds GetPrepareTraceId(const SyncInfo &info, const CloudInfo &cloud); 164 std::pair<bool, StoreMetaData> GetMetaData(const StoreInfo &storeInfo); 165 void AddCompensateSync(const StoreMetaData &meta); 166 static void Report( 167 const std::string &faultType, const std::string &bundleName, int32_t errCode, const std::string &appendix); 168 void ReportSyncEvent(const DistributedData::SyncEvent &evt, DistributedDataDfx::BizState bizState, int32_t code); 169 std::pair<int32_t, CloudLastSyncInfo> GetLastSyncInfoFromMeta(const QueryKey &queryKey); 170 static void SaveLastSyncInfo(const QueryKey &queryKey, CloudLastSyncInfo &&info); 171 172 static std::atomic<uint32_t> genId_; 173 std::shared_ptr<ExecutorPool> executor_; 174 ConcurrentMap<uint64_t, TaskId> actives_; 175 ConcurrentMap<uint64_t, uint64_t> activeInfos_; 176 std::shared_ptr<SyncStrategy> syncStrategy_; 177 ConcurrentMap<QueryKey, std::map<SyncId, CloudLastSyncInfo>> lastSyncInfos_; 178 std::set<std::string> kvApps_; 179 ConcurrentMap<int32_t, std::map<std::string, std::set<std::string>>> compensateSyncInfos_; 180 }; 181 } // namespace OHOS::CloudData 182 #endif // OHOS_DISTRIBUTED_DATA_SERVICES_CLOUD_SYNC_MANAGER_H