user_event_sync_bridge.cc 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. // Copyright 2017 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "components/sync_user_events/user_event_sync_bridge.h"
  5. #include <set>
  6. #include <utility>
  7. #include <vector>
  8. #include "base/big_endian.h"
  9. #include "base/bind.h"
  10. #include "base/callback_helpers.h"
  11. #include "base/check_op.h"
  12. #include "base/containers/contains.h"
  13. #include "base/containers/cxx20_erase.h"
  14. #include "base/location.h"
  15. #include "base/strings/string_number_conversions.h"
  16. #include "build/build_config.h"
  17. #include "components/sync/model/data_type_activation_request.h"
  18. #include "components/sync/model/entity_change.h"
  19. #include "components/sync/model/metadata_batch.h"
  20. #include "components/sync/model/mutable_data_batch.h"
  21. #include "components/sync/protocol/entity_specifics.pb.h"
  22. #include "components/sync/protocol/user_event_specifics.pb.h"
  23. namespace syncer {
  24. using sync_pb::UserEventSpecifics;
  25. using IdList = ModelTypeStore::IdList;
  26. using Record = ModelTypeStore::Record;
  27. using RecordList = ModelTypeStore::RecordList;
  28. using WriteBatch = ModelTypeStore::WriteBatch;
  29. namespace {
  30. std::string GetStorageKeyFromSpecifics(const UserEventSpecifics& specifics) {
  31. // Force Big Endian, this means newly created keys are last in sort order,
  32. // which allows leveldb to append new writes, which it is best at.
  33. // TODO(skym): Until we force |event_time_usec| to never conflict, this has
  34. // the potential for errors.
  35. std::string key(8, 0);
  36. base::WriteBigEndian(&key[0], specifics.event_time_usec());
  37. return key;
  38. }
  39. int64_t GetEventTimeFromStorageKey(const std::string& storage_key) {
  40. int64_t event_time;
  41. base::ReadBigEndian(reinterpret_cast<const uint8_t*>(&storage_key[0]),
  42. &event_time);
  43. return event_time;
  44. }
  45. std::unique_ptr<EntityData> MoveToEntityData(
  46. std::unique_ptr<UserEventSpecifics> specifics) {
  47. auto entity_data = std::make_unique<EntityData>();
  48. entity_data->name = base::NumberToString(specifics->event_time_usec());
  49. entity_data->specifics.set_allocated_user_event(specifics.release());
  50. return entity_data;
  51. }
  52. } // namespace
  53. UserEventSyncBridge::UserEventSyncBridge(
  54. OnceModelTypeStoreFactory store_factory,
  55. std::unique_ptr<ModelTypeChangeProcessor> change_processor,
  56. GlobalIdMapper* global_id_mapper)
  57. : ModelTypeSyncBridge(std::move(change_processor)),
  58. global_id_mapper_(global_id_mapper) {
  59. DCHECK(global_id_mapper_);
  60. std::move(store_factory)
  61. .Run(USER_EVENTS, base::BindOnce(&UserEventSyncBridge::OnStoreCreated,
  62. weak_ptr_factory_.GetWeakPtr()));
  63. global_id_mapper_->AddGlobalIdChangeObserver(
  64. base::BindRepeating(&UserEventSyncBridge::HandleGlobalIdChange,
  65. weak_ptr_factory_.GetWeakPtr()));
  66. }
  67. UserEventSyncBridge::~UserEventSyncBridge() = default;
  68. std::unique_ptr<MetadataChangeList>
  69. UserEventSyncBridge::CreateMetadataChangeList() {
  70. return WriteBatch::CreateMetadataChangeList();
  71. }
  72. absl::optional<ModelError> UserEventSyncBridge::MergeSyncData(
  73. std::unique_ptr<MetadataChangeList> metadata_change_list,
  74. EntityChangeList entity_data) {
  75. DCHECK(entity_data.empty());
  76. DCHECK(change_processor()->IsTrackingMetadata());
  77. DCHECK(!change_processor()->TrackedAccountId().empty());
  78. return ApplySyncChanges(std::move(metadata_change_list),
  79. std::move(entity_data));
  80. }
  81. absl::optional<ModelError> UserEventSyncBridge::ApplySyncChanges(
  82. std::unique_ptr<MetadataChangeList> metadata_change_list,
  83. EntityChangeList entity_changes) {
  84. std::unique_ptr<WriteBatch> batch = store_->CreateWriteBatch();
  85. std::set<int64_t> deleted_event_times;
  86. for (const std::unique_ptr<EntityChange>& change : entity_changes) {
  87. DCHECK_EQ(EntityChange::ACTION_DELETE, change->type());
  88. batch->DeleteData(change->storage_key());
  89. deleted_event_times.insert(
  90. GetEventTimeFromStorageKey(change->storage_key()));
  91. }
  92. // Because we receive ApplySyncChanges with deletions when our commits are
  93. // confirmed, this is the perfect time to cleanup our in flight objects which
  94. // are no longer in flight.
  95. base::EraseIf(in_flight_nav_linked_events_,
  96. [&deleted_event_times](
  97. const std::pair<int64_t, sync_pb::UserEventSpecifics> kv) {
  98. return base::Contains(deleted_event_times,
  99. kv.second.event_time_usec());
  100. });
  101. batch->TakeMetadataChangesFrom(std::move(metadata_change_list));
  102. store_->CommitWriteBatch(std::move(batch),
  103. base::BindOnce(&UserEventSyncBridge::OnCommit,
  104. weak_ptr_factory_.GetWeakPtr()));
  105. return {};
  106. }
  107. void UserEventSyncBridge::GetData(StorageKeyList storage_keys,
  108. DataCallback callback) {
  109. store_->ReadData(
  110. storage_keys,
  111. base::BindOnce(&UserEventSyncBridge::OnReadData,
  112. weak_ptr_factory_.GetWeakPtr(), std::move(callback)));
  113. }
  114. void UserEventSyncBridge::GetAllDataForDebugging(DataCallback callback) {
  115. store_->ReadAllData(base::BindOnce(&UserEventSyncBridge::OnReadAllData,
  116. weak_ptr_factory_.GetWeakPtr(),
  117. std::move(callback)));
  118. }
  119. std::string UserEventSyncBridge::GetClientTag(const EntityData& entity_data) {
  120. return GetStorageKey(entity_data);
  121. }
  122. std::string UserEventSyncBridge::GetStorageKey(const EntityData& entity_data) {
  123. return GetStorageKeyFromSpecifics(entity_data.specifics.user_event());
  124. }
  125. void UserEventSyncBridge::ApplyStopSyncChanges(
  126. std::unique_ptr<MetadataChangeList> delete_metadata_change_list) {
  127. if (delete_metadata_change_list) {
  128. store_->DeleteAllDataAndMetadata(base::BindOnce(
  129. &UserEventSyncBridge::OnCommit, weak_ptr_factory_.GetWeakPtr()));
  130. }
  131. }
  132. void UserEventSyncBridge::RecordUserEvent(
  133. std::unique_ptr<UserEventSpecifics> specifics) {
  134. DCHECK(!specifics->has_user_consent());
  135. if (store_) {
  136. RecordUserEventImpl(std::move(specifics));
  137. return;
  138. }
  139. }
  140. // static
  141. std::string UserEventSyncBridge::GetStorageKeyFromSpecificsForTest(
  142. const UserEventSpecifics& specifics) {
  143. return GetStorageKeyFromSpecifics(specifics);
  144. }
  145. std::unique_ptr<ModelTypeStore> UserEventSyncBridge::StealStoreForTest() {
  146. return std::move(store_);
  147. }
  148. void UserEventSyncBridge::RecordUserEventImpl(
  149. std::unique_ptr<UserEventSpecifics> specifics) {
  150. DCHECK(store_);
  151. if (!change_processor()->IsTrackingMetadata()) {
  152. return;
  153. }
  154. std::string storage_key = GetStorageKeyFromSpecifics(*specifics);
  155. // There are two scenarios we need to guard against here. First, the given
  156. // user even may have been read from an old global_id timestamp off of a
  157. // navigation, which has already been re-written. In this case, we should be
  158. // able to look up the latest/best global_id to use right now, and update as
  159. // such. The other scenario is that the navigation is going to be updated in
  160. // the future, and the current global_id, while valid for now, is never going
  161. // to make it to the server, and will need to be fixed. To handle this
  162. // scenario, we store a specifics copy in |in in_flight_nav_linked_events_|,
  163. // and will re-record in HandleGlobalIdChange.
  164. if (specifics->has_navigation_id()) {
  165. int64_t latest_global_id =
  166. global_id_mapper_->GetLatestGlobalId(specifics->navigation_id());
  167. specifics->set_navigation_id(latest_global_id);
  168. in_flight_nav_linked_events_.insert(
  169. std::make_pair(latest_global_id, *specifics));
  170. }
  171. std::unique_ptr<WriteBatch> batch = store_->CreateWriteBatch();
  172. batch->WriteData(storage_key, specifics->SerializeAsString());
  173. DCHECK(change_processor()->IsTrackingMetadata());
  174. change_processor()->Put(storage_key, MoveToEntityData(std::move(specifics)),
  175. batch->GetMetadataChangeList());
  176. store_->CommitWriteBatch(std::move(batch),
  177. base::BindOnce(&UserEventSyncBridge::OnCommit,
  178. weak_ptr_factory_.GetWeakPtr()));
  179. }
  180. void UserEventSyncBridge::OnStoreCreated(
  181. const absl::optional<ModelError>& error,
  182. std::unique_ptr<ModelTypeStore> store) {
  183. if (error) {
  184. change_processor()->ReportError(*error);
  185. return;
  186. }
  187. store_ = std::move(store);
  188. store_->ReadAllMetadata(base::BindOnce(
  189. &UserEventSyncBridge::OnReadAllMetadata, weak_ptr_factory_.GetWeakPtr()));
  190. }
  191. void UserEventSyncBridge::OnReadAllMetadata(
  192. const absl::optional<ModelError>& error,
  193. std::unique_ptr<MetadataBatch> metadata_batch) {
  194. if (error) {
  195. change_processor()->ReportError(*error);
  196. } else {
  197. change_processor()->ModelReadyToSync(std::move(metadata_batch));
  198. }
  199. }
  200. void UserEventSyncBridge::OnCommit(const absl::optional<ModelError>& error) {
  201. if (error) {
  202. change_processor()->ReportError(*error);
  203. }
  204. }
  205. void UserEventSyncBridge::OnReadData(DataCallback callback,
  206. const absl::optional<ModelError>& error,
  207. std::unique_ptr<RecordList> data_records,
  208. std::unique_ptr<IdList> missing_id_list) {
  209. OnReadAllData(std::move(callback), error, std::move(data_records));
  210. }
  211. void UserEventSyncBridge::OnReadAllData(
  212. DataCallback callback,
  213. const absl::optional<ModelError>& error,
  214. std::unique_ptr<RecordList> data_records) {
  215. if (error) {
  216. change_processor()->ReportError(*error);
  217. return;
  218. }
  219. auto batch = std::make_unique<MutableDataBatch>();
  220. for (const Record& r : *data_records) {
  221. auto specifics = std::make_unique<UserEventSpecifics>();
  222. if (specifics->ParseFromString(r.value)) {
  223. DCHECK_EQ(r.id, GetStorageKeyFromSpecifics(*specifics));
  224. batch->Put(r.id, MoveToEntityData(std::move(specifics)));
  225. } else {
  226. change_processor()->ReportError(
  227. {FROM_HERE, "Failed deserializing user events."});
  228. return;
  229. }
  230. }
  231. std::move(callback).Run(std::move(batch));
  232. }
  233. void UserEventSyncBridge::HandleGlobalIdChange(int64_t old_global_id,
  234. int64_t new_global_id) {
  235. DCHECK_NE(old_global_id, new_global_id);
  236. // Store specifics in a temp vector while erasing, as |RecordUserEvent()| will
  237. // insert new values into |in_flight_nav_linked_events_|. While insert should
  238. // not invalidate a std::multimap's iterator, and the updated global_id should
  239. // not be within our given range, this approach seems less error prone.
  240. std::vector<std::unique_ptr<UserEventSpecifics>> affected;
  241. auto [begin, end] = in_flight_nav_linked_events_.equal_range(old_global_id);
  242. for (auto iter = begin; iter != end;) {
  243. DCHECK_EQ(old_global_id, iter->second.navigation_id());
  244. affected.emplace_back(std::make_unique<UserEventSpecifics>(iter->second));
  245. iter = in_flight_nav_linked_events_.erase(iter);
  246. }
  247. for (std::unique_ptr<UserEventSpecifics>& specifics : affected) {
  248. specifics->set_navigation_id(new_global_id);
  249. RecordUserEvent(std::move(specifics));
  250. }
  251. }
  252. } // namespace syncer