// Copyright 2022 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "components/saved_tab_groups/internal/saved_tab_group_sync_bridge.h"

#include <algorithm>
#include <iterator>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>

#include "base/check.h"
#include "base/functional/bind.h"
#include "base/functional/callback_helpers.h"
#include "base/strings/utf_string_conversions.h"
#include "base/time/time.h"
#include "base/trace_event/trace_event.h"
#include "base/uuid.h"
#include "build/build_config.h"
#include "components/prefs/pref_service.h"
#include "components/saved_tab_groups/internal/saved_tab_group_proto_conversions.h"
#include "components/saved_tab_groups/internal/stats.h"
#include "components/saved_tab_groups/internal/sync_bridge_tab_group_model_wrapper.h"
#include "components/saved_tab_groups/proto/saved_tab_group_data.pb.h"
#include "components/saved_tab_groups/public/features.h"
#include "components/saved_tab_groups/public/pref_names.h"
#include "components/saved_tab_groups/public/saved_tab_group.h"
#include "components/saved_tab_groups/public/saved_tab_group_tab.h"
#include "components/saved_tab_groups/public/types.h"
#include "components/saved_tab_groups/public/utils.h"
#include "components/sync/base/data_type.h"
#include "components/sync/base/deletion_origin.h"
#include "components/sync/model/conflict_resolution.h"
#include "components/sync/model/data_type_activation_request.h"
#include "components/sync/model/data_type_local_change_processor.h"
#include "components/sync/model/data_type_store.h"
#include "components/sync/model/entity_change.h"
#include "components/sync/model/metadata_batch.h"
#include "components/sync/model/metadata_change_list.h"
#include "components/sync/model/model_error.h"
#include "components/sync/model/mutable_data_batch.h"
#include "components/sync/protocol/entity_data.h"
#include "components/sync/protocol/saved_tab_group_specifics.pb.h"
#include "google_apis/gaia/gaia_id.h"
#include "url/gurl.h"

namespace tab_groups {
namespace {

BASE_FEATURE(kUpdatePostionsOnTabClose, base::FEATURE_ENABLED_BY_DEFAULT);

// Time period for orphaned tabs/groups to live till. once this threshold is
// passed, on the next merge, they will be deleted.
constexpr base::TimeDelta kOrphanedObjectDiscardThreshold = base::Days(30);

bool IsValidSpecifics(const sync_pb::SavedTabGroupSpecifics& specifics) {
  // A valid specifics should have at least a guid and be either a group or a
  // tab.
  return specifics.has_guid() && (specifics.has_tab() || specifics.has_group());
}

std::unique_ptr<syncer::EntityData> CreateEntityData(
    sync_pb::SavedTabGroupSpecifics specific) {
  if (specific.has_tab()) {
    // If the tab URL is not valid for syncing, change it to the Chrome
    // unsupported URL before sending it to sync server. The local db will still
    // store the original URL for session restoration.
    if (!IsURLValidForSavedTabGroups(GURL(specific.tab().url()))) {
      sync_pb::SavedTabGroupTab* tab = specific.mutable_tab();
      tab->set_url(kChromeSavedTabGroupUnsupportedURL);
      tab->clear_title();
    }
  }
  std::unique_ptr<syncer::EntityData> entity_data =
      std::make_unique<syncer::EntityData>();
  entity_data->name = specific.guid();
  entity_data->specifics.mutable_saved_tab_group()->Swap(&specific);
  return entity_data;
}

base::Time TimeFromWindowsEpochMicros(int64_t time_windows_epoch_micros) {
  return base::Time::FromDeltaSinceWindowsEpoch(
      base::Microseconds(time_windows_epoch_micros));
}

std::vector<proto::SavedTabGroupData> LoadStoredEntries(
    std::vector<proto::SavedTabGroupData> stored_entries,
    SyncBridgeTabGroupModelWrapper* model_wrapper) {
  std::vector<SavedTabGroup> groups;
  std::unordered_set<std::string> group_guids;

  // `stored_entries` is not ordered such that groups are guaranteed to be
  // at the front of the vector. As such, we can run into the case where we
  // try to add a tab to a group that does not exist for us yet.
  for (const proto::SavedTabGroupData& proto : stored_entries) {
    if (proto.specifics().has_group()) {
      groups.emplace_back(DataToSavedTabGroup(proto));
      group_guids.emplace(proto.specifics().guid());
    }
  }

  // Parse tabs and find tabs missing groups.
  std::vector<proto::SavedTabGroupData> tabs_missing_groups;
  std::vector<SavedTabGroupTab> tabs;
  for (const proto::SavedTabGroupData& proto : stored_entries) {
    if (proto.specifics().has_group()) {
      continue;
    }
    if (group_guids.contains(proto.specifics().tab().group_guid())) {
      tabs.emplace_back(DataToSavedTabGroupTab(proto));
      continue;
    }
    tabs_missing_groups.push_back(std::move(proto));
  }

  model_wrapper->Initialize(std::move(groups), std::move(tabs));
  return tabs_missing_groups;
}

// Returns the index of the group with `group_id`. The group must exist and be
// present in `groups`.
size_t CalculateIndexOfGroup(const std::vector<const SavedTabGroup*>& groups,
                             const base::Uuid& group_id) {
  auto iter =
      std::ranges::find_if(groups, [&group_id](const SavedTabGroup* group) {
        return group->saved_guid() == group_id;
      });
  CHECK(iter != groups.end());
  return std::distance(groups.begin(), iter);
}

}  // anonymous namespace

SavedTabGroupSyncBridge::SavedTabGroupSyncBridge(
    SyncBridgeTabGroupModelWrapper* model_wrapper,
    syncer::OnceDataTypeStoreFactory create_store_callback,
    std::unique_ptr<syncer::DataTypeLocalChangeProcessor> change_processor,
    PrefService* pref_service)
    : syncer::DataTypeSyncBridge(std::move(change_processor)),
      model_wrapper_(model_wrapper),
      pref_service_(pref_service) {
  CHECK(model_wrapper_);
  CHECK(pref_service_);
  std::move(create_store_callback)
      .Run(syncer::SAVED_TAB_GROUP,
           base::BindOnce(&SavedTabGroupSyncBridge::OnStoreCreated,
                          weak_ptr_factory_.GetWeakPtr()));
}

SavedTabGroupSyncBridge::~SavedTabGroupSyncBridge() = default;

void SavedTabGroupSyncBridge::OnSyncStarting(
    const syncer::DataTypeActivationRequest& request) {}

std::optional<syncer::ModelError> SavedTabGroupSyncBridge::MergeFullSyncData(
    std::unique_ptr<syncer::MetadataChangeList> metadata_change_list,
    syncer::EntityChangeList entity_changes) {
  // This method must not be called reentrantly.
  CHECK(!ongoing_write_batch_);

  // Do not commit the write batch by default because this method returns an
  // error after the scoped write batch is destroyed (otherwise, changes would
  // be committed in case of error).
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/false,
                                  std::move(metadata_change_list));
  CHECK(ongoing_write_batch_);

  std::set<std::string> synced_items;

  // MergeFullSyncData is the first command called when the user signs in/or
  // turns sync on. When this happens, a cache guid  will be added to the
  // metadata of the change processor. This cache guid should be used on all
  // groups and tabs that previously didnt have a cache guid attached.
  CHECK(change_processor()->IsTrackingMetadata());
  UpdateLocalCacheGuidForGroups(ongoing_write_batch_.get());

  // Merge sync to local data.
  for (const auto& change : entity_changes) {
    synced_items.insert(change->storage_key());
    AddDataToLocalStorage(std::move(change->data().specifics.saved_tab_group()),
                          ongoing_write_batch_->GetMetadataChangeList(),
                          ongoing_write_batch_.get(),
                          /*notify_sync=*/true);
  }

  ResolveTabsMissingGroups(ongoing_write_batch_.get());
  ResolveGroupsMissingTabs(ongoing_write_batch_.get());

  // Update sync with any locally stored data not currently stored in sync.
  for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
    for (const SavedTabGroupTab& tab : group->saved_tabs()) {
      if (synced_items.count(tab.saved_tab_guid().AsLowercaseString())) {
        continue;
      }
      SendToSync(SavedTabGroupTabToData(tab).specifics(),
                 ongoing_write_batch_->GetMetadataChangeList());
    }

    if (synced_items.count(group->saved_guid().AsLowercaseString())) {
      continue;
    }
    SendToSync(SavedTabGroupToData(*group).specifics(),
               ongoing_write_batch_->GetMetadataChangeList());
  }

  // Successfully applied all the changes. Explicitly commit the write batch to
  // store.
  CommitOngoingWriteBatch();
  return {};
}

std::optional<syncer::ModelError>
SavedTabGroupSyncBridge::ApplyIncrementalSyncChanges(
    std::unique_ptr<syncer::MetadataChangeList> metadata_change_list,
    syncer::EntityChangeList entity_changes) {
  // Do not store the write batch on destroy by default. This is only required
  // while processing remote updates because it reports an error after the
  // scoped write batch is destroyed. There must not be reentrant calls to
  // ApplyIncrementalSyncChanges().
  CHECK(!ongoing_write_batch_);
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/false,
                                  std::move(metadata_change_list));
  CHECK(ongoing_write_batch_);

  std::vector<std::string> deleted_entities;
  for (const std::unique_ptr<syncer::EntityChange>& change : entity_changes) {
    switch (change->type()) {
      case syncer::EntityChange::ACTION_DELETE: {
        deleted_entities.push_back(change->storage_key());
        break;
      }
      case syncer::EntityChange::ACTION_ADD:
      case syncer::EntityChange::ACTION_UPDATE: {
        AddDataToLocalStorage(
            std::move(change->data().specifics.saved_tab_group()),
            ongoing_write_batch_->GetMetadataChangeList(),
            ongoing_write_batch_.get(),
            /*notify_sync=*/false);
        break;
      }
    }
  }

  // Process deleted entities last. This is done for consistency. Since
  // `entity_changes` is not guaranteed to be in order, it is possible that a
  // user could add or remove tabs in a way that puts the group in an empty
  // state. This will unintentionally delete the group and drop any additional
  // add / update messages. By processing deletes last, we can give the groups
  // an opportunity to resolve themselves before they become empty.
  for (const std::string& entity : deleted_entities) {
    DeleteDataFromLocalStorage(base::Uuid::ParseLowercase(entity),
                               ongoing_write_batch_.get());
  }

  ResolveTabsMissingGroups(ongoing_write_batch_.get());
  ResolveGroupsMissingTabs(ongoing_write_batch_.get());

  // Successfully applied all the changes. Explicitly commit the write batch to
  // store.
  CommitOngoingWriteBatch();
  return {};
}

syncer::ConflictResolution SavedTabGroupSyncBridge::ResolveConflict(
    const std::string& storage_key,
    const syncer::EntityData& remote_data) const {
  if (remote_data.is_deleted()) {
    return syncer::ConflictResolution::kUseLocal;
  }

  CHECK(IsEntityDataValid(remote_data));
  const sync_pb::SavedTabGroupSpecifics& remote_specifics =
      remote_data.specifics.saved_tab_group();

  // Do a conflict resolution based on last update timestamp.
  base::Uuid guid = base::Uuid::ParseLowercase(remote_specifics.guid());
  base::Time local_timestamp;
  if (remote_specifics.has_group()) {
    if (const SavedTabGroup* group = model_wrapper_->GetGroup(guid)) {
      local_timestamp = group->update_time();
    }
  } else {
    CHECK(remote_specifics.has_tab());
    base::Uuid group_guid =
        base::Uuid::ParseLowercase(remote_specifics.tab().group_guid());
    const SavedTabGroup* group = model_wrapper_->GetGroup(group_guid);
    if (const SavedTabGroupTab* tab = group ? group->GetTab(guid) : nullptr) {
      local_timestamp = tab->update_time();
    }
  }

  base::Time remote_timestamp = TimeFromWindowsEpochMicros(
      remote_specifics.update_time_windows_epoch_micros());
  bool local_is_more_recent =
      !local_timestamp.is_null() && local_timestamp > remote_timestamp;

  return local_is_more_recent ? syncer::ConflictResolution::kUseLocal
                              : syncer::ConflictResolution::kUseRemote;
}

void SavedTabGroupSyncBridge::ApplyDisableSyncChanges(
    std::unique_ptr<syncer::MetadataChangeList> delete_metadata_change_list) {
  // Close the local groups that were created before sign-in.
  // They should still exist in sync server.
  CHECK(!ongoing_write_batch_);
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  std::move(delete_metadata_change_list));

  std::vector<base::Uuid> groups_to_close_locally;
  for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
    if (group->created_before_syncing_tab_groups()) {
      continue;
    }

    groups_to_close_locally.emplace_back(group->saved_guid());
  }

  for (const base::Uuid& group_id : groups_to_close_locally) {
    // This group should be closed locally. Remove it from model and storage and
    // close all the tabs.
    const SavedTabGroup* group = model_wrapper_->GetGroup(group_id);
    std::vector<base::Uuid> tabs_to_close_locally;
    for (const SavedTabGroupTab& tab : group->saved_tabs()) {
      tabs_to_close_locally.emplace_back(tab.saved_tab_guid());
    }

    // The group could have been deleted when the last tab was closed, hence
    // double check before calling RemoveGroup().
    if (model_wrapper_->GetGroup(group_id)) {
      model_wrapper_->RemoveGroup(group_id);
    }

    // Remove the tabs from storage.
    for (const base::Uuid& tab_id : tabs_to_close_locally) {
      ongoing_write_batch_->DeleteData(tab_id.AsLowercaseString());
    }

    ongoing_write_batch_->DeleteData(group_id.AsLowercaseString());
  }

  // Reset the cache guids for groups and tabs on sign-out.
  UpdateLocalCacheGuidForGroups(ongoing_write_batch_.get());
}

std::string SavedTabGroupSyncBridge::GetStorageKey(
    const syncer::EntityData& entity_data) const {
  return entity_data.specifics.saved_tab_group().guid();
}

std::string SavedTabGroupSyncBridge::GetClientTag(
    const syncer::EntityData& entity_data) const {
  return GetStorageKey(entity_data);
}

std::unique_ptr<syncer::DataBatch> SavedTabGroupSyncBridge::GetDataForCommit(
    StorageKeyList storage_keys) {
  auto batch = std::make_unique<syncer::MutableDataBatch>();

  for (const std::string& guid : storage_keys) {
    base::Uuid parsed_guid = base::Uuid::ParseLowercase(guid);
    for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
      if (group->saved_guid() == parsed_guid) {
        AddEntryToBatch(batch.get(), SavedTabGroupToData(*group));
        break;
      }

      if (const SavedTabGroupTab* tab = group->GetTab(parsed_guid)) {
        AddEntryToBatch(batch.get(), SavedTabGroupTabToData(*tab));
        break;
      }
    }
  }

  return batch;
}

std::unique_ptr<syncer::DataBatch>
SavedTabGroupSyncBridge::GetAllDataForDebugging() {
  auto batch = std::make_unique<syncer::MutableDataBatch>();
  for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
    AddEntryToBatch(batch.get(), SavedTabGroupToData(*group));
    for (const SavedTabGroupTab& tab : group->saved_tabs()) {
      AddEntryToBatch(batch.get(), SavedTabGroupTabToData(tab));
    }
  }

  return batch;
}

bool SavedTabGroupSyncBridge::IsEntityDataValid(
    const syncer::EntityData& entity_data) const {
  const sync_pb::SavedTabGroupSpecifics& specifics =
      entity_data.specifics.saved_tab_group();
  return specifics.has_group() || specifics.has_tab();
}

sync_pb::EntitySpecifics
SavedTabGroupSyncBridge::TrimAllSupportedFieldsFromRemoteSpecifics(
    const sync_pb::EntitySpecifics& entity_specifics) const {
  // LINT.IfChange(TrimAllSupportedFieldsFromRemoteSpecifics)
  sync_pb::SavedTabGroupSpecifics trimmed_specifics =
      entity_specifics.saved_tab_group();
  trimmed_specifics.clear_guid();
  trimmed_specifics.clear_creation_time_windows_epoch_micros();
  trimmed_specifics.clear_update_time_windows_epoch_micros();
  trimmed_specifics.clear_version();

  if (trimmed_specifics.has_tab()) {
    sync_pb::SavedTabGroupTab* tab = trimmed_specifics.mutable_tab();
    tab->clear_group_guid();
    tab->clear_position();
    tab->clear_url();
    tab->clear_title();

    if (tab->ByteSizeLong() == 0) {
      trimmed_specifics.clear_tab();
    }
  }

  if (trimmed_specifics.has_group()) {
    sync_pb::SavedTabGroup* tab_group = trimmed_specifics.mutable_group();
    tab_group->clear_position();
    tab_group->clear_title();
    tab_group->clear_color();
    tab_group->clear_pinned_position();
    tab_group->clear_bookmark_node_id();
    tab_group->clear_projects_position();

    if (tab_group->ByteSizeLong() == 0) {
      trimmed_specifics.clear_group();
    }
  }

  if (trimmed_specifics.has_attribution_metadata()) {
    sync_pb::AttributionMetadata* attribution_metadata =
        trimmed_specifics.mutable_attribution_metadata();
    if (attribution_metadata->has_created()) {
      sync_pb::AttributionMetadata::Attribution* created =
          attribution_metadata->mutable_created();
      if (created->has_device_info()) {
        created->mutable_device_info()->clear_cache_guid();
        if (created->mutable_device_info()->ByteSizeLong() == 0) {
          created->clear_device_info();
        }
      }
      if (created->ByteSizeLong() == 0) {
        attribution_metadata->clear_created();
      }
    }

    if (attribution_metadata->has_updated()) {
      sync_pb::AttributionMetadata::Attribution* updated =
          attribution_metadata->mutable_updated();
      if (updated->has_device_info()) {
        updated->mutable_device_info()->clear_cache_guid();
        if (updated->mutable_device_info()->ByteSizeLong() == 0) {
          updated->clear_device_info();
        }
      }
      if (updated->ByteSizeLong() == 0) {
        attribution_metadata->clear_updated();
      }
    }

    if (attribution_metadata->ByteSizeLong() == 0) {
      trimmed_specifics.clear_attribution_metadata();
    }
  }

  // LINT.ThenChange(//components/sync/protocol/saved_tab_group_specifics.proto:SavedTabGroupSpecifics)

  sync_pb::EntitySpecifics trimmed_entity_specifics;
  if (trimmed_specifics.ByteSizeLong() > 0) {
    *trimmed_entity_specifics.mutable_saved_tab_group() =
        std::move(trimmed_specifics);
  }
  return trimmed_entity_specifics;
}

proto::SavedTabGroupData SavedTabGroupSyncBridge::SavedTabGroupToData(
    const SavedTabGroup& group) const {
  sync_pb::SavedTabGroupSpecifics trimmed_specifics;
  if (change_processor()->IsTrackingMetadata()) {
    trimmed_specifics = change_processor()
                            ->GetPossiblyTrimmedRemoteSpecifics(
                                group.saved_guid().AsLowercaseString())
                            .saved_tab_group();
  }

  return tab_groups::SavedTabGroupToData(group, trimmed_specifics);
}

proto::SavedTabGroupData SavedTabGroupSyncBridge::SavedTabGroupTabToData(
    const SavedTabGroupTab& tab) const {
  sync_pb::SavedTabGroupSpecifics trimmed_specifics;
  if (change_processor()->IsTrackingMetadata()) {
    trimmed_specifics = change_processor()
                            ->GetPossiblyTrimmedRemoteSpecifics(
                                tab.saved_tab_guid().AsLowercaseString())
                            .saved_tab_group();
  }

  return tab_groups::SavedTabGroupTabToData(tab, trimmed_specifics);
}

// SavedTabGroupModelObserver
void SavedTabGroupSyncBridge::SavedTabGroupAddedLocally(
    const base::Uuid& guid) {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  const SavedTabGroup* group = model_wrapper_->GetGroup(guid);
  CHECK(group);

  int index = CalculateIndexOfGroup(model_wrapper_->GetTabGroups(), guid);
  proto::SavedTabGroupData group_data = SavedTabGroupToData(*group);
  group_data.mutable_specifics()->mutable_group()->set_position(index);

  UpsertEntitySpecific(std::move(group_data), ongoing_write_batch_.get());
  for (size_t i = 0; i < group->saved_tabs().size(); ++i) {
    // Pending NTP should never be created for locally added groups.
    CHECK(!group->saved_tabs()[i].is_pending_ntp());
    proto::SavedTabGroupData tab_data =
        SavedTabGroupTabToData(group->saved_tabs()[i]);
    tab_data.mutable_specifics()->mutable_tab()->set_position(i);
    UpsertEntitySpecific(std::move(tab_data), ongoing_write_batch_.get());
  }
}

void SavedTabGroupSyncBridge::SavedTabGroupRemovedLocally(
    const SavedTabGroup& removed_group) {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  // Intentionally only remove the group (creating orphaned tabs in the
  // process), so other devices with the group open in the Tabstrip can react to
  // the deletion appropriately (i.e. We do not have to determine if a tab
  // deletion was part of a group deletion).
  RemoveEntitySpecific(removed_group.saved_guid(), ongoing_write_batch_.get());

  // Keep track of the newly orphaned tabs since their group no longer exists.
  for (const SavedTabGroupTab& tab : removed_group.saved_tabs()) {
    tabs_missing_groups_.emplace_back(SavedTabGroupTabToData(tab));
  }

  // Update the DataTypeStore (local storage) and sync with the new positions
  // of all the groups after a remove has occurred so the positions are
  // preserved on browser restart. See crbug/1462443.
  SavedTabGroupReorderedLocally();
}

void SavedTabGroupSyncBridge::SavedTabGroupUpdatedLocally(
    const base::Uuid& group_guid,
    const std::optional<base::Uuid>& tab_guid) {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  const SavedTabGroup* const group = model_wrapper_->GetGroup(group_guid);
  CHECK(group);

  if (tab_guid.has_value()) {
    if (!group->ContainsTab(tab_guid.value())) {
      RemoveEntitySpecific(tab_guid.value(), ongoing_write_batch_.get());
      if (base::FeatureList::IsEnabled(kUpdatePostionsOnTabClose)) {
        // Remaining tabs in the group have their positions updated.
        SavedTabGroupTabsReorderedLocally(group_guid);
      }
    } else {
      int tab_index = group->GetIndexOfTab(tab_guid.value()).value();
      const SavedTabGroupTab& tab = group->saved_tabs()[tab_index];
      UpsertEntitySpecific(
          SavedTabGroupTabToData(group->saved_tabs()[tab_index]),
          ongoing_write_batch_.get(), /*send_to_sync=*/!tab.is_pending_ntp());
      // TODO(https://crbug.com/437033786): If a new tab is inserted in the
      // middle of the group we need to update the positions of the existing
      // tabs.
    }

    // There might be an updated user interaction time for the group. Hence
    // write the group to DB.
    auto group_data = SavedTabGroupToData(*group);
    ongoing_write_batch_->WriteData(group_data.specifics().guid(),
                                    group_data.SerializeAsString());
  } else {
    UpsertEntitySpecific(SavedTabGroupToData(*group),
                         ongoing_write_batch_.get());
  }
}

void SavedTabGroupSyncBridge::SavedTabGroupTabsReorderedLocally(
    const base::Uuid& group_guid) {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  const SavedTabGroup* const group = model_wrapper_->GetGroup(group_guid);
  DCHECK(group);

  for (const SavedTabGroupTab& tab : group->saved_tabs()) {
    // None of the tabs in the group can be pending NTP since pending NTP can be
    // the only tab in the group and those groups are never reordered.
    CHECK(!tab.is_pending_ntp());
    UpsertEntitySpecific(SavedTabGroupTabToData(tab),
                         ongoing_write_batch_.get());
  }
}

void SavedTabGroupSyncBridge::SavedTabGroupLocalIdChanged(
    const base::Uuid& group_guid) {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  const SavedTabGroup* const group = model_wrapper_->GetGroup(group_guid);
  CHECK(group);

  auto data = SavedTabGroupToData(*group);
  ongoing_write_batch_->WriteData(data.specifics().guid(),
                                  data.SerializeAsString());
}

void SavedTabGroupSyncBridge::SavedTabGroupLastUserInteractionTimeUpdated(
    const base::Uuid& group_guid) {
  const SavedTabGroup* const group = model_wrapper_->GetGroup(group_guid);
  CHECK(group);

  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);
  proto::SavedTabGroupData data = SavedTabGroupToData(*group);
  ongoing_write_batch_->WriteData(data.specifics().guid(),
                                  data.SerializeAsString());
}

void SavedTabGroupSyncBridge::SavedTabGroupReorderedLocally() {
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/true,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
    UpsertEntitySpecific(SavedTabGroupToData(*group),
                         ongoing_write_batch_.get());
  }
}

std::optional<std::string> SavedTabGroupSyncBridge::GetLocalCacheGuid() const {
  if (!change_processor()->IsTrackingMetadata()) {
    return std::nullopt;
  }
  return change_processor()->TrackedCacheGuid();
}

std::optional<GaiaId> SavedTabGroupSyncBridge::GetTrackedGaiaId() const {
  if (!change_processor()->IsTrackingMetadata()) {
    return std::nullopt;
  }
  return change_processor()->TrackedGaiaId();
}

bool SavedTabGroupSyncBridge::IsSyncing() const {
  return change_processor()->IsTrackingMetadata();
}

// static
SavedTabGroup SavedTabGroupSyncBridge::SpecificsToSavedTabGroupForTest(
    const sync_pb::SavedTabGroupSpecifics& specifics) {
  proto::SavedTabGroupData data;
  data.set_allocated_specifics(new sync_pb::SavedTabGroupSpecifics(specifics));
  return DataToSavedTabGroup(data);
}

// static
sync_pb::SavedTabGroupSpecifics
SavedTabGroupSyncBridge::SavedTabGroupToSpecificsForTest(
    const SavedTabGroup& group) {
  return tab_groups::SavedTabGroupToData(group,
                                         sync_pb::SavedTabGroupSpecifics())
      .specifics();
}

// static
SavedTabGroupTab SavedTabGroupSyncBridge::SpecificsToSavedTabGroupTabForTest(
    const sync_pb::SavedTabGroupSpecifics& specifics) {
  proto::SavedTabGroupData data;
  data.set_allocated_specifics(new sync_pb::SavedTabGroupSpecifics(specifics));
  return DataToSavedTabGroupTab(data);
}

// static
sync_pb::SavedTabGroupSpecifics
SavedTabGroupSyncBridge::SavedTabGroupTabToSpecificsForTest(
    const SavedTabGroupTab& tab) {
  return tab_groups::SavedTabGroupTabToData(tab,
                                            sync_pb::SavedTabGroupSpecifics())
      .specifics();
}

// static
SavedTabGroup SavedTabGroupSyncBridge::DataToSavedTabGroupForTest(
    const proto::SavedTabGroupData& data) {
  return DataToSavedTabGroup(data);
}

// static
proto::SavedTabGroupData SavedTabGroupSyncBridge::SavedTabGroupToDataForTest(
    const SavedTabGroup& group) {
  return tab_groups::SavedTabGroupToData(group,
                                         sync_pb::SavedTabGroupSpecifics());
}

// static
SavedTabGroupTab SavedTabGroupSyncBridge::DataToSavedTabGroupTabForTest(
    const proto::SavedTabGroupData& data) {
  return DataToSavedTabGroupTab(data);
}

// static
proto::SavedTabGroupData SavedTabGroupSyncBridge::SavedTabGroupTabToDataForTest(
    const SavedTabGroupTab& tab) {
  return tab_groups::SavedTabGroupTabToData(tab,
                                            sync_pb::SavedTabGroupSpecifics());
}

void SavedTabGroupSyncBridge::UpsertEntitySpecific(
    const proto::SavedTabGroupData& data,
    syncer::DataTypeStore::WriteBatch* write_batch) {
  UpsertEntitySpecific(data, write_batch, /*send_to_sync=*/true);
}

void SavedTabGroupSyncBridge::UpsertEntitySpecific(
    const proto::SavedTabGroupData& data,
    syncer::DataTypeStore::WriteBatch* write_batch,
    bool send_to_sync) {
  write_batch->WriteData(data.specifics().guid(), data.SerializeAsString());
  if (send_to_sync) {
    SendToSync(data.specifics(), write_batch->GetMetadataChangeList());
  }
}

void SavedTabGroupSyncBridge::RemoveEntitySpecific(
    const base::Uuid& guid,
    syncer::DataTypeStore::WriteBatch* write_batch) {
  write_batch->DeleteData(guid.AsLowercaseString());

  if (!change_processor()->IsTrackingMetadata()) {
    return;
  }

  change_processor()->Delete(guid.AsLowercaseString(),
                             syncer::DeletionOrigin::Unspecified(),
                             write_batch->GetMetadataChangeList());
}

void SavedTabGroupSyncBridge::AddDataToLocalStorage(
    const sync_pb::SavedTabGroupSpecifics& specifics,
    syncer::MetadataChangeList* metadata_change_list,
    syncer::DataTypeStore::WriteBatch* write_batch,
    bool notify_sync) {
  base::Uuid group_guid = base::Uuid::ParseLowercase(
      specifics.has_tab() ? specifics.tab().group_guid() : specifics.guid());
  const SavedTabGroup* existing_group = model_wrapper_->GetGroup(group_guid);

  proto::SavedTabGroupData data;
  data.set_allocated_specifics(new sync_pb::SavedTabGroupSpecifics(specifics));

  // Sanitize incoming remote URLs before writing to the local database. This
  // ensures we don't store unsafe URLs (e.g. from a compromised sync server)
  // while allowing locally created unsyncable URLs (like file://) to be
  // preserved.
  if (data.specifics().has_tab()) {
    if (!IsURLValidForSavedTabGroups(GURL(data.specifics().tab().url()))) {
      data.mutable_specifics()->mutable_tab()->set_url(
          kChromeSavedTabGroupUnsupportedURL);
      data.mutable_specifics()->mutable_tab()->clear_title();
    }
  }

  std::string guid = data.specifics().guid();

  // Cases where `specifics` is a group.
  if (specifics.has_group()) {
    if (existing_group) {
      // Resolve the conflict by merging the sync and local data. Once
      // finished, write the result to the store and update sync with the new
      // merged result if appropriate.
      existing_group = model_wrapper_->MergeRemoteGroupMetadata(
          group_guid, base::UTF8ToUTF16(specifics.group().title()),
          SyncColorToTabGroupColor(specifics.group().color()),
          GroupPositionFromSpecifics(specifics),
          GetCreatorCacheGuidFromSpecifics(specifics),
          GetLastUpdaterCacheGuidFromSpecifics(specifics),
          TimeFromWindowsEpochMicros(
              specifics.update_time_windows_epoch_micros()),
          /*updated_by=*/GaiaId());

      proto::SavedTabGroupData updated_data =
          SavedTabGroupToData(*existing_group);

      // Write result to the store.
      write_batch->WriteData(guid, updated_data.SerializeAsString());

      // Update sync with the new merged result.
      if (notify_sync) {
        SendToSync(std::move(updated_data.specifics()), metadata_change_list);
      }
    } else {
      // We do not have this group. Add the group from sync into local storage.
      SavedTabGroup new_group = DataToSavedTabGroup(data);
      write_batch->WriteData(guid, data.SerializeAsString());
      model_wrapper_->AddGroup(std::move(new_group));
    }

    return;
  }

  // Cases where `specifics` is a tab.
  if (specifics.has_tab()) {
    base::Uuid tab_guid = base::Uuid::ParseLowercase(data.specifics().guid());
    if (existing_group && existing_group->ContainsTab(tab_guid)) {
      // Resolve the conflict by merging the sync and local data. Once finished,
      // write the result to the store and update sync with the new merged
      // result if appropriate.
      const SavedTabGroupTab* merged_tab =
          model_wrapper_->MergeRemoteTab(DataToSavedTabGroupTab(data));
      data = SavedTabGroupTabToData(*merged_tab);

      // Write result to the store.
      write_batch->WriteData(guid, data.SerializeAsString());

      // Update sync with the new merged result.
      if (notify_sync) {
        SendToSync(std::move(data.specifics()), metadata_change_list);
      }

      return;
    }

    // We write tabs to local storage regardless of the existence of its group
    // in order to recover the tabs in the event the group was not received and
    // a crash / restart occurred.
    write_batch->WriteData(guid, data.SerializeAsString());

    if (existing_group) {
      // We do not have this tab. Add the tab from sync into local storage.
      model_wrapper_->AddTabToGroup(existing_group->saved_guid(),
                                    DataToSavedTabGroupTab(data));
    } else {
      // We reach this case if we were unable to find a group for this tab. This
      // can happen when sync sends the tab data before the group data. In this
      // case, we will store the tabs in case the group comes in later.
      tabs_missing_groups_.emplace_back(std::move(data));
    }
  }
}

void SavedTabGroupSyncBridge::DeleteDataFromLocalStorage(
    const base::Uuid& guid,
    syncer::DataTypeStore::WriteBatch* write_batch) {
  write_batch->DeleteData(guid.AsLowercaseString());
  // Check if the model contains the group guid. If so, remove that group and
  // all of its tabs.
  // TODO(b/336586617): Close tabs on desktop on receiving this event.
  if (model_wrapper_->GetGroup(guid)) {
    model_wrapper_->RemoveGroup(guid);
    return;
  }

  const SavedTabGroup* group_containing_tab =
      model_wrapper_->GetGroupContainingTab(guid);
  if (group_containing_tab) {
    model_wrapper_->RemoveTabFromGroup(group_containing_tab->saved_guid(),
                                       guid);
  }
}

void SavedTabGroupSyncBridge::ResolveTabsMissingGroups(
    syncer::DataTypeStore::WriteBatch* write_batch) {
  auto tab_iterator = tabs_missing_groups_.begin();
  while (tab_iterator != tabs_missing_groups_.end()) {
    const auto& specifics = tab_iterator->specifics();
    const SavedTabGroup* group = model_wrapper_->GetGroup(
        base::Uuid::ParseLowercase(specifics.tab().group_guid()));
    if (!group) {
      base::Time last_update_time = base::Time::FromDeltaSinceWindowsEpoch(
          base::Microseconds(specifics.update_time_windows_epoch_micros()));
      base::Time now = base::Time::Now();

      // Discard the tabs that do not have an associated group and have not been
      // updated within the discard threshold.
      if (now - last_update_time >= kOrphanedObjectDiscardThreshold) {
        RemoveEntitySpecific(base::Uuid::ParseLowercase(specifics.guid()),
                             write_batch);
        tab_iterator = tabs_missing_groups_.erase(tab_iterator);
      } else {
        ++tab_iterator;
      }
    } else {
      write_batch->WriteData(specifics.guid(),
                             tab_iterator->SerializeAsString());
      model_wrapper_->AddTabToGroup(group->saved_guid(),
                                    DataToSavedTabGroupTab(*tab_iterator));
      tab_iterator = tabs_missing_groups_.erase(tab_iterator);
    }
  }
}

void SavedTabGroupSyncBridge::ResolveGroupsMissingTabs(
    syncer::DataTypeStore::WriteBatch* write_batch) {
  std::vector<base::Uuid> orphaned_groups_to_destroy;
  for (const SavedTabGroup* group : model_wrapper_->GetTabGroups()) {
    if (!group->saved_tabs().empty()) {
      continue;
    }

    if ((base::Time::Now() - group->update_time()) <
        kOrphanedObjectDiscardThreshold) {
      continue;
    }

    orphaned_groups_to_destroy.push_back(group->saved_guid());
  }

  for (const base::Uuid& group_id : orphaned_groups_to_destroy) {
    model_wrapper_->RemoveGroup(group_id);
    write_batch->DeleteData(group_id.AsLowercaseString());
  }
}

void SavedTabGroupSyncBridge::AddEntryToBatch(syncer::MutableDataBatch* batch,
                                              proto::SavedTabGroupData data) {
  std::unique_ptr<syncer::EntityData> entity_data =
      CreateEntityData(std::move(data.specifics()));

  // Copy because our key is the name of `entity_data`.
  std::string name = entity_data->name;

  batch->Put(name, std::move(entity_data));
}

void SavedTabGroupSyncBridge::SendToSync(
    sync_pb::SavedTabGroupSpecifics specific,
    syncer::MetadataChangeList* metadata_change_list) {
  DCHECK(metadata_change_list);
  if (!change_processor()->IsTrackingMetadata()) {
    return;
  }

  auto entity_data = CreateEntityData(std::move(specific));

  // Copy because our key is the name of `entity_data`.
  std::string name = entity_data->name;
  change_processor()->Put(name, std::move(entity_data), metadata_change_list);
}

void SavedTabGroupSyncBridge::OnStoreCreated(
    const std::optional<syncer::ModelError>& error,
    std::unique_ptr<syncer::DataTypeStore> store) {
  if (error) {
    stats::RecordMigrationResult(stats::MigrationResult::kStoreCreateFailed);
    change_processor()->ReportError(*error);
    return;
  }

  store_ = std::move(store);
  stats::RecordMigrationResult(stats::MigrationResult::kStoreLoadStarted);
  store_->ReadAllData(base::BindOnce(&SavedTabGroupSyncBridge::OnDatabaseLoad,
                                     weak_ptr_factory_.GetWeakPtr()));
}

void SavedTabGroupSyncBridge::OnDatabaseLoad(
    const std::optional<syncer::ModelError>& error,
    std::unique_ptr<syncer::DataTypeStore::RecordList> entries) {
  // This function does a series of migrations and finally loads the metadata.
  // After each migration step, the DB is read again which invokes this callback
  // again. If a migration isn't required, it will be skipped to execute the
  // next steps.
  if (error) {
    stats::RecordMigrationResult(stats::MigrationResult::kStoreLoadFailed);
    change_processor()->ReportError(*error);
    return;
  }

  // 1. Check if we haven't yet migrated from SavedTabGroupSpecifics
  // to SavedTabGroupData.
  if (!pref_service_->GetBoolean(
          prefs::kSavedTabGroupSpecificsToDataMigration)) {
    stats::RecordMigrationResult(
        stats::MigrationResult::kSpecificsToDataMigrationStarted);
    MigrateSpecificsToSavedTabGroupData(std::move(entries));
    return;
  }

  if (!migration_already_complete_recorded_) {
    migration_already_complete_recorded_ = true;
    stats::RecordMigrationResult(
        stats::MigrationResult::kSpecificsToDataMigrationAlreadyComplete);
  }

  // 2. If we are done with all the migrations, proceed with regular metadata
  // loading.
  stats::RecordMigrationResult(stats::MigrationResult::kStoreLoadCompleted);
  store_->ReadAllMetadata(
      base::BindOnce(&SavedTabGroupSyncBridge::OnReadAllMetadata,
                     weak_ptr_factory_.GetWeakPtr(), std::move(entries)));
}

void SavedTabGroupSyncBridge::MigrateSpecificsToSavedTabGroupData(
    std::unique_ptr<syncer::DataTypeStore::RecordList> entries) {
  // The migration should never be a reentrant call. Note that it uses a custom
  // write batch commit logic and hence should be committed explicitly.
  CHECK(!ongoing_write_batch_);
  base::ScopedClosureRunner scoped_write_batch_destroy_runner =
      MaybeCreateScopedWriteBatch(/*commit_write_batch_on_destroy=*/false,
                                  /*metadata_change_list=*/nullptr);
  CHECK(ongoing_write_batch_);

  int parse_failure_count = 0;
  for (const syncer::DataTypeStore::Record& r : *entries) {
    sync_pb::SavedTabGroupSpecifics specifics;
    // We might potentially be parsing a SavedTabGroupData as a
    // SavedTabGroupSpecifics and vice versa. At times parsing succeeds, hence
    // we need to check for existence of required fields.
    if (!specifics.ParseFromString(r.value) || !IsValidSpecifics(specifics)) {
      DLOG(WARNING)
          << "Failed to parse SavedTabGroupSpecifics during migration";
      parse_failure_count++;
      continue;
    }

    proto::SavedTabGroupData new_data;
    new_data.set_allocated_specifics(
        new sync_pb::SavedTabGroupSpecifics(specifics));

    ongoing_write_batch_->WriteData(specifics.guid(),
                                    new_data.SerializeAsString());
  }

  if (parse_failure_count > 0) {
    stats::RecordMigrationResult(
        stats::MigrationResult::
            kSpecificsToDataMigrationParseFailedAtLeastOnce);
    stats::RecordSpecificsParseFailureCount(parse_failure_count,
                                            entries->size());
  }

  // Do not use DestroyOngoingWriteBatch() here because the migration requires
  // a custom logic after commit.
  store_->CommitWriteBatch(
      std::move(ongoing_write_batch_),
      base::BindOnce(
          &SavedTabGroupSyncBridge::OnSpecificsToDataMigrationComplete,
          weak_ptr_factory_.GetWeakPtr()));
}

void SavedTabGroupSyncBridge::OnSpecificsToDataMigrationComplete(
    const std::optional<syncer::ModelError>& error) {
  if (error) {
    stats::RecordMigrationResult(
        stats::MigrationResult::kSpecificsToDataMigrationWriteFailed);
    change_processor()->ReportError(*error);
    DLOG(WARNING) << "Failed to write migrated data";
    return;
  }

  // Migration successful, write it to prefs, and read the DB again.
  pref_service_->SetBoolean(prefs::kSavedTabGroupSpecificsToDataMigration,
                            true);
  stats::RecordMigrationResult(
      stats::MigrationResult::kSpecificsToDataMigrationSuccess);
  store_->ReadAllData(base::BindOnce(&SavedTabGroupSyncBridge::OnDatabaseLoad,
                                     weak_ptr_factory_.GetWeakPtr()));
}

void SavedTabGroupSyncBridge::OnReadAllMetadata(
    std::unique_ptr<syncer::DataTypeStore::RecordList> entries,
    const std::optional<syncer::ModelError>& error,
    std::unique_ptr<syncer::MetadataBatch> metadata_batch) {
  TRACE_EVENT0("ui", "SavedTabGroupSyncBridge::OnReadAllMetadata");
  if (error) {
    stats::RecordMigrationResult(
        stats::MigrationResult::kReadAllMetadataFailed);
    change_processor()->ReportError(*error);
    return;
  }

  stats::RecordMigrationResult(stats::MigrationResult::kReadAllMetadataSuccess);

  std::vector<proto::SavedTabGroupData> stored_entries;
  stored_entries.reserve(entries->size());

  for (const syncer::DataTypeStore::Record& r : *entries) {
    proto::SavedTabGroupData proto;
    if (!proto.ParseFromString(r.value)) {
      continue;
    }
    stored_entries.emplace_back(std::move(proto));
  }

  stats::RecordParsedSavedTabGroupDataCount(stored_entries.size(),
                                            entries->size());

  // Update `model_` with any data stored in local storage except for orphaned
  // tabs. Orphaned tabs will be returned and added to `tabs_missing_groups_` in
  // case their missing group ever arrives.
  tabs_missing_groups_ =
      LoadStoredEntries(std::move(stored_entries), model_wrapper_);

  // change_process() will ignore the following call in case of error.
  change_processor()->ModelReadyToSync(std::move(metadata_batch));
}

void SavedTabGroupSyncBridge::OnDatabaseSave(
    const std::optional<syncer::ModelError>& error) {
  if (error) {
    change_processor()->ReportError(*error);
    return;
  }

  // TODO(dljames): React to store failures when a save is not successful.
}

void SavedTabGroupSyncBridge::UpdateLocalCacheGuidForGroups(
    syncer::DataTypeStore::WriteBatch* write_batch) {
  std::pair<std::set<base::Uuid>, std::set<base::Uuid>> updated_ids =
      model_wrapper_->UpdateLocalCacheGuid(std::nullopt, GetLocalCacheGuid());
  const std::set<base::Uuid>& updated_group_ids = updated_ids.first;
  const std::set<base::Uuid>& updated_tab_ids = updated_ids.second;

  for (const base::Uuid& saved_guid : updated_group_ids) {
    proto::SavedTabGroupData data =
        SavedTabGroupToData(*model_wrapper_->GetGroup(saved_guid));
    write_batch->WriteData(data.specifics().guid(), data.SerializeAsString());
  }

  for (const base::Uuid& tab_guid : updated_tab_ids) {
    const SavedTabGroup* group =
        model_wrapper_->GetGroupContainingTab(tab_guid);
    CHECK(group);
    const SavedTabGroupTab* tab = group->GetTab(tab_guid);
    CHECK(tab);
    proto::SavedTabGroupData data = SavedTabGroupTabToData(*tab);
    write_batch->WriteData(data.specifics().guid(), data.SerializeAsString());
  }
}

bool SavedTabGroupSyncBridge::IsRemoteGroup(const SavedTabGroup& group) {
  std::optional<std::string> local_cache_guid = GetLocalCacheGuid();
  std::optional<std::string> group_cache_guid = group.creator_cache_guid();
  if (!local_cache_guid || !group_cache_guid) {
    return false;
  }

  return local_cache_guid.value() != group_cache_guid.value();
}

base::ScopedClosureRunner SavedTabGroupSyncBridge::MaybeCreateScopedWriteBatch(
    bool commit_write_batch_on_destroy,
    std::unique_ptr<syncer::MetadataChangeList> metadata_change_list) {
  if (ongoing_write_batch_) {
    // There is an ongoing write batch, hence do not create a new one and do not
    // destroy the existing one in the current scope.
    if (metadata_change_list) {
      ongoing_write_batch_->TakeMetadataChangesFrom(
          std::move(metadata_change_list));
    }
    return base::ScopedClosureRunner(base::DoNothing());
  }

  // This is not a reentrant call, create a new write batch and return a scoped
  // closure runner that will destroy it when it goes out of scope.
  ongoing_write_batch_ =
      store_->CreateWriteBatch(std::move(metadata_change_list));
  return base::ScopedClosureRunner(base::BindOnce(
      &SavedTabGroupSyncBridge::DestroyOngoingWriteBatch,
      weak_ptr_factory_.GetWeakPtr(), commit_write_batch_on_destroy));
}

void SavedTabGroupSyncBridge::CommitOngoingWriteBatch() {
  // If the bridge is in the error state, the write batch should not be
  // committed to avoid storing invalid state.
  if (change_processor()->GetError().has_value()) {
    return;
  }

  CHECK(ongoing_write_batch_);
  store_->CommitWriteBatch(
      std::move(ongoing_write_batch_),
      base::BindOnce(&SavedTabGroupSyncBridge::OnDatabaseSave,
                     weak_ptr_factory_.GetWeakPtr()));
}

void SavedTabGroupSyncBridge::DestroyOngoingWriteBatch(
    bool commit_write_batch_on_destroy) {
  // When `commit_write_batch_on_destroy` is true, `ongoing_write_batch_` must
  // exist (the write batch must not be committed explicitly in this case).
  if (commit_write_batch_on_destroy) {
    CommitOngoingWriteBatch();
  }
  ongoing_write_batch_ = nullptr;
}

}  // namespace tab_groups