// Copyright (C) 2017 The Android Open Source Project
//
// Licensed 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.

package com.google.gerrit.server.notedb.rebuild;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.gerrit.reviewdb.server.ReviewDbUtil.unwrapDb;
import static com.google.gerrit.server.notedb.NotesMigration.SECTION_NOTE_DB;
import static com.google.gerrit.server.notedb.NotesMigrationState.NOTE_DB;
import static com.google.gerrit.server.notedb.NotesMigrationState.READ_WRITE_NO_SEQUENCE;
import static com.google.gerrit.server.notedb.NotesMigrationState.READ_WRITE_WITH_SEQUENCE_NOTE_DB_PRIMARY;
import static com.google.gerrit.server.notedb.NotesMigrationState.READ_WRITE_WITH_SEQUENCE_REVIEW_DB_PRIMARY;
import static com.google.gerrit.server.notedb.NotesMigrationState.WRITE;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Comparator.comparing;
import static java.util.stream.Collectors.joining;
import static java.util.stream.Collectors.toList;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.MultimapBuilder;
import com.google.common.collect.Ordering;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Streams;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gerrit.common.FormatUtil;
import com.google.gerrit.common.Nullable;
import com.google.gerrit.extensions.registration.DynamicSet;
import com.google.gerrit.reviewdb.client.Change;
import com.google.gerrit.reviewdb.client.Project;
import com.google.gerrit.reviewdb.server.ReviewDb;
import com.google.gerrit.reviewdb.server.ReviewDbWrapper;
import com.google.gerrit.server.GerritPersonIdent;
import com.google.gerrit.server.InternalUser;
import com.google.gerrit.server.Sequences;
import com.google.gerrit.server.config.AllProjectsName;
import com.google.gerrit.server.config.AllUsersName;
import com.google.gerrit.server.config.GerritServerConfigProvider;
import com.google.gerrit.server.config.SitePaths;
import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
import com.google.gerrit.server.git.GitRepositoryManager;
import com.google.gerrit.server.git.LockFailureException;
import com.google.gerrit.server.git.WorkQueue;
import com.google.gerrit.server.notedb.ChangeBundleReader;
import com.google.gerrit.server.notedb.ChangeNotes;
import com.google.gerrit.server.notedb.MutableNotesMigration;
import com.google.gerrit.server.notedb.NoteDbTable;
import com.google.gerrit.server.notedb.NoteDbUpdateManager;
import com.google.gerrit.server.notedb.NotesMigrationState;
import com.google.gerrit.server.notedb.PrimaryStorageMigrator;
import com.google.gerrit.server.notedb.PrimaryStorageMigrator.NoNoteDbStateException;
import com.google.gerrit.server.notedb.RepoSequence;
import com.google.gerrit.server.notedb.rebuild.ChangeRebuilder.NoPatchSetsException;
import com.google.gerrit.server.project.NoSuchChangeException;
import com.google.gerrit.server.update.ChainedReceiveCommands;
import com.google.gerrit.server.update.RefUpdateUtil;
import com.google.gerrit.server.util.ManualRequestContext;
import com.google.gerrit.server.util.ThreadLocalRequestContext;
import com.google.gwtorm.server.OrmException;
import com.google.gwtorm.server.SchemaFactory;
import com.google.inject.Inject;
import com.google.inject.Provider;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Predicate;
import org.eclipse.jgit.errors.ConfigInvalidException;
import org.eclipse.jgit.errors.RepositoryNotFoundException;
import org.eclipse.jgit.internal.storage.file.FileRepository;
import org.eclipse.jgit.internal.storage.file.PackInserter;
import org.eclipse.jgit.lib.BatchRefUpdate;
import org.eclipse.jgit.lib.Config;
import org.eclipse.jgit.lib.ObjectInserter;
import org.eclipse.jgit.lib.ObjectReader;
import org.eclipse.jgit.lib.PersonIdent;
import org.eclipse.jgit.lib.ProgressMonitor;
import org.eclipse.jgit.lib.Repository;
import org.eclipse.jgit.lib.TextProgressMonitor;
import org.eclipse.jgit.revwalk.RevWalk;
import org.eclipse.jgit.storage.file.FileBasedConfig;
import org.eclipse.jgit.transport.ReceiveCommand;
import org.eclipse.jgit.util.FS;
import org.eclipse.jgit.util.io.NullOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** One stop shop for migrating a site's change storage from ReviewDb to NoteDb. */
public class NoteDbMigrator implements AutoCloseable {
  private static final Logger log = LoggerFactory.getLogger(NoteDbMigrator.class);

  private static final String AUTO_MIGRATE = "autoMigrate";
  private static final String TRIAL = "trial";

  public static boolean getAutoMigrate(Config cfg) {
    return cfg.getBoolean(SECTION_NOTE_DB, NoteDbTable.CHANGES.key(), AUTO_MIGRATE, false);
  }

  private static void setAutoMigrate(Config cfg, boolean autoMigrate) {
    cfg.setBoolean(SECTION_NOTE_DB, NoteDbTable.CHANGES.key(), AUTO_MIGRATE, autoMigrate);
  }

  public static boolean getTrialMode(Config cfg) {
    return cfg.getBoolean(SECTION_NOTE_DB, NoteDbTable.CHANGES.key(), TRIAL, false);
  }

  public static void setTrialMode(Config cfg, boolean trial) {
    cfg.setBoolean(SECTION_NOTE_DB, NoteDbTable.CHANGES.key(), TRIAL, trial);
  }

  public static class Builder {
    private final Config cfg;
    private final SitePaths sitePaths;
    private final Provider<PersonIdent> serverIdent;
    private final AllUsersName allUsers;
    private final SchemaFactory<ReviewDb> schemaFactory;
    private final GitRepositoryManager repoManager;
    private final NoteDbUpdateManager.Factory updateManagerFactory;
    private final ChangeBundleReader bundleReader;
    private final AllProjectsName allProjects;
    private final InternalUser.Factory userFactory;
    private final ThreadLocalRequestContext requestContext;
    private final ChangeRebuilderImpl rebuilder;
    private final WorkQueue workQueue;
    private final MutableNotesMigration globalNotesMigration;
    private final PrimaryStorageMigrator primaryStorageMigrator;
    private final DynamicSet<NotesMigrationStateListener> listeners;

    private int threads;
    private ImmutableList<Project.NameKey> projects = ImmutableList.of();
    private ImmutableList<Project.NameKey> skipProjects = ImmutableList.of();
    private ImmutableList<Change.Id> changes = ImmutableList.of();
    private OutputStream progressOut = NullOutputStream.INSTANCE;
    private NotesMigrationState stopAtState;
    private boolean trial;
    private boolean forceRebuild;
    private int sequenceGap = -1;
    private boolean autoMigrate;

    @Inject
    Builder(
        GerritServerConfigProvider configProvider,
        SitePaths sitePaths,
        @GerritPersonIdent Provider<PersonIdent> serverIdent,
        AllUsersName allUsers,
        SchemaFactory<ReviewDb> schemaFactory,
        GitRepositoryManager repoManager,
        NoteDbUpdateManager.Factory updateManagerFactory,
        ChangeBundleReader bundleReader,
        AllProjectsName allProjects,
        ThreadLocalRequestContext requestContext,
        InternalUser.Factory userFactory,
        ChangeRebuilderImpl rebuilder,
        WorkQueue workQueue,
        MutableNotesMigration globalNotesMigration,
        PrimaryStorageMigrator primaryStorageMigrator,
        DynamicSet<NotesMigrationStateListener> listeners) {
      // Reload gerrit.config/notedb.config on each migrator invocation, in case a previous
      // migration in the same process modified the on-disk contents. This ensures the defaults for
      // trial/autoMigrate get set correctly below.
      this.cfg = configProvider.get();
      this.sitePaths = sitePaths;
      this.serverIdent = serverIdent;
      this.allUsers = allUsers;
      this.schemaFactory = schemaFactory;
      this.repoManager = repoManager;
      this.updateManagerFactory = updateManagerFactory;
      this.bundleReader = bundleReader;
      this.allProjects = allProjects;
      this.requestContext = requestContext;
      this.userFactory = userFactory;
      this.rebuilder = rebuilder;
      this.workQueue = workQueue;
      this.globalNotesMigration = globalNotesMigration;
      this.primaryStorageMigrator = primaryStorageMigrator;
      this.listeners = listeners;
      this.trial = getTrialMode(cfg);
      this.autoMigrate = getAutoMigrate(cfg);
    }

    /**
     * Set the number of threads used by parallelizable phases of the migration, such as rebuilding
     * all changes.
     *
     * <p>Not all phases are parallelizable, and calling {@link #rebuild()} directly will do
     * substantial work in the calling thread regardless of the number of threads configured.
     *
     * <p>By default, all work is done in the calling thread.
     *
     * @param threads thread count; if less than 2, all work happens in the calling thread.
     * @return this.
     */
    public Builder setThreads(int threads) {
      this.threads = threads;
      return this;
    }

    /**
     * Limit the set of projects that are processed.
     *
     * <p>Incompatible with {@link #setChanges(Collection)}.
     *
     * <p>By default, all projects will be processed.
     *
     * @param projects set of projects; if null or empty, all projects will be processed.
     * @return this.
     */
    public Builder setProjects(@Nullable Collection<Project.NameKey> projects) {
      this.projects = projects != null ? ImmutableList.copyOf(projects) : ImmutableList.of();
      return this;
    }

    /**
     * Process all projects except these
     *
     * <p>Incompatible with {@link #setProjects(Collection)} and {@link #setChanges(Collection)}
     *
     * <p>By default, all projects will be processed.
     *
     * @param skipProjects set of projects; if null or empty all project will be processed
     * @return this.
     */
    public Builder setSkipProjects(@Nullable Collection<Project.NameKey> skipProjects) {
      this.skipProjects =
          skipProjects != null ? ImmutableList.copyOf(skipProjects) : ImmutableList.of();
      return this;
    }

    /**
     * Limit the set of changes that are processed.
     *
     * <p>Incompatible with {@link #setProjects(Collection)}.
     *
     * <p>By default, all changes will be processed.
     *
     * @param changes set of changes; if null or empty, all changes will be processed.
     * @return this.
     */
    public Builder setChanges(@Nullable Collection<Change.Id> changes) {
      this.changes = changes != null ? ImmutableList.copyOf(changes) : ImmutableList.of();
      return this;
    }

    /**
     * Set output stream for progress monitors.
     *
     * <p>By default, there is no progress monitor output (although there may be other logs).
     *
     * @param progressOut output stream.
     * @return this.
     */
    public Builder setProgressOut(OutputStream progressOut) {
      this.progressOut = checkNotNull(progressOut);
      return this;
    }

    /**
     * Stop at a specific migration state, for testing only.
     *
     * @param stopAtState state to stop at.
     * @return this.
     */
    @VisibleForTesting
    public Builder setStopAtStateForTesting(NotesMigrationState stopAtState) {
      this.stopAtState = stopAtState;
      return this;
    }

    /**
     * Rebuild in "trial mode": configure Gerrit to write to and read from NoteDb, but leave
     * ReviewDb as the source of truth for all changes.
     *
     * <p>By default, trial mode is off, and NoteDb is the source of truth for all changes following
     * the migration.
     *
     * @param trial whether to rebuild in trial mode.
     * @return this.
     */
    public Builder setTrialMode(boolean trial) {
      this.trial = trial;
      return this;
    }

    /**
     * Rebuild all changes in NoteDb from ReviewDb, even if Gerrit is currently configured to read
     * from NoteDb.
     *
     * <p>Only supported if ReviewDb is still the source of truth for all changes.
     *
     * <p>By default, force rebuilding is off.
     *
     * @param forceRebuild whether to force rebuilding.
     * @return this.
     */
    public Builder setForceRebuild(boolean forceRebuild) {
      this.forceRebuild = forceRebuild;
      return this;
    }

    /**
     * Gap between ReviewDb change sequence numbers and NoteDb.
     *
     * <p>If NoteDb sequences are enabled in a running server, there is a race between the migration
     * step that calls {@code nextChangeId()} to seed the ref, and other threads that call {@code
     * nextChangeId()} to create new changes. In order to prevent these operations stepping on one
     * another, we use this value to skip some predefined sequence numbers. This is strongly
     * recommended in a running server.
     *
     * <p>If the migration takes place offline, there is no race with other threads, and this option
     * may be set to 0. However, admins may still choose to use a gap, for example to make it easier
     * to distinguish changes that were created before and after the NoteDb migration.
     *
     * <p>By default, uses the value from {@code noteDb.changes.initialSequenceGap} in {@code
     * gerrit.config}, which defaults to 1000.
     *
     * @param sequenceGap sequence gap size; if negative, use the default.
     * @return this.
     */
    public Builder setSequenceGap(int sequenceGap) {
      this.sequenceGap = sequenceGap;
      return this;
    }

    /**
     * Enable auto-migration on subsequent daemon launches.
     *
     * <p>If true, prior to running any migration steps, sets the necessary configuration in {@code
     * gerrit.config} to make {@code gerrit.war daemon} retry the migration on next startup, if it
     * fails.
     *
     * @param autoMigrate whether to set auto-migration config.
     * @return this.
     */
    public Builder setAutoMigrate(boolean autoMigrate) {
      this.autoMigrate = autoMigrate;
      return this;
    }

    public NoteDbMigrator build() throws MigrationException {
      return new NoteDbMigrator(
          sitePaths,
          schemaFactory,
          serverIdent,
          allUsers,
          repoManager,
          updateManagerFactory,
          bundleReader,
          allProjects,
          requestContext,
          userFactory,
          rebuilder,
          globalNotesMigration,
          primaryStorageMigrator,
          listeners,
          threads > 1
              ? MoreExecutors.listeningDecorator(
                  workQueue.createQueue(threads, "RebuildChange", true))
              : MoreExecutors.newDirectExecutorService(),
          projects,
          skipProjects,
          changes,
          progressOut,
          stopAtState,
          trial,
          forceRebuild,
          sequenceGap >= 0 ? sequenceGap : Sequences.getChangeSequenceGap(cfg),
          autoMigrate);
    }
  }

  private final FileBasedConfig gerritConfig;
  private final FileBasedConfig noteDbConfig;
  private final SchemaFactory<ReviewDb> schemaFactory;
  private final Provider<PersonIdent> serverIdent;
  private final AllUsersName allUsers;
  private final GitRepositoryManager repoManager;
  private final NoteDbUpdateManager.Factory updateManagerFactory;
  private final ChangeBundleReader bundleReader;
  private final AllProjectsName allProjects;
  private final ThreadLocalRequestContext requestContext;
  private final InternalUser.Factory userFactory;
  private final ChangeRebuilderImpl rebuilder;
  private final MutableNotesMigration globalNotesMigration;
  private final PrimaryStorageMigrator primaryStorageMigrator;
  private final DynamicSet<NotesMigrationStateListener> listeners;

  private final ListeningExecutorService executor;
  private final ImmutableList<Project.NameKey> projects;
  private final ImmutableList<Project.NameKey> skipProjects;
  private final ImmutableList<Change.Id> changes;
  private final OutputStream progressOut;
  private final NotesMigrationState stopAtState;
  private final boolean trial;
  private final boolean forceRebuild;
  private final int sequenceGap;
  private final boolean autoMigrate;

  private NoteDbMigrator(
      SitePaths sitePaths,
      SchemaFactory<ReviewDb> schemaFactory,
      Provider<PersonIdent> serverIdent,
      AllUsersName allUsers,
      GitRepositoryManager repoManager,
      NoteDbUpdateManager.Factory updateManagerFactory,
      ChangeBundleReader bundleReader,
      AllProjectsName allProjects,
      ThreadLocalRequestContext requestContext,
      InternalUser.Factory userFactory,
      ChangeRebuilderImpl rebuilder,
      MutableNotesMigration globalNotesMigration,
      PrimaryStorageMigrator primaryStorageMigrator,
      DynamicSet<NotesMigrationStateListener> listeners,
      ListeningExecutorService executor,
      ImmutableList<Project.NameKey> projects,
      ImmutableList<Project.NameKey> skipProjects,
      ImmutableList<Change.Id> changes,
      OutputStream progressOut,
      NotesMigrationState stopAtState,
      boolean trial,
      boolean forceRebuild,
      int sequenceGap,
      boolean autoMigrate)
      throws MigrationException {
    if (ImmutableList.of(!changes.isEmpty(), !projects.isEmpty(), !skipProjects.isEmpty())
            .stream()
            .filter(e -> e)
            .count()
        > 1) {
      throw new MigrationException("Cannot combine changes, projects and skipProjects");
    }
    if (sequenceGap < 0) {
      throw new MigrationException("Sequence gap must be non-negative: " + sequenceGap);
    }

    this.schemaFactory = schemaFactory;
    this.serverIdent = serverIdent;
    this.allUsers = allUsers;
    this.rebuilder = rebuilder;
    this.repoManager = repoManager;
    this.updateManagerFactory = updateManagerFactory;
    this.bundleReader = bundleReader;
    this.allProjects = allProjects;
    this.requestContext = requestContext;
    this.userFactory = userFactory;
    this.globalNotesMigration = globalNotesMigration;
    this.primaryStorageMigrator = primaryStorageMigrator;
    this.listeners = listeners;
    this.executor = executor;
    this.projects = projects;
    this.skipProjects = skipProjects;
    this.changes = changes;
    this.progressOut = progressOut;
    this.stopAtState = stopAtState;
    this.trial = trial;
    this.forceRebuild = forceRebuild;
    this.sequenceGap = sequenceGap;
    this.autoMigrate = autoMigrate;

    // Stack notedb.config over gerrit.config, in the same way as GerritServerConfigProvider.
    this.gerritConfig = new FileBasedConfig(sitePaths.gerrit_config.toFile(), FS.detect());
    this.noteDbConfig =
        new FileBasedConfig(gerritConfig, sitePaths.notedb_config.toFile(), FS.detect());
  }

  @Override
  public void close() {
    executor.shutdownNow();
  }

  public void migrate() throws OrmException, IOException {
    if (!changes.isEmpty() || !projects.isEmpty() || !skipProjects.isEmpty()) {
      throw new MigrationException(
          "Cannot set changes or projects or skipProjects during full migration; call rebuild() instead");
    }
    Optional<NotesMigrationState> maybeState = loadState();
    if (!maybeState.isPresent()) {
      throw new MigrationException("Could not determine initial migration state");
    }

    NotesMigrationState state = maybeState.get();
    if (trial && state.compareTo(READ_WRITE_NO_SEQUENCE) > 0) {
      throw new MigrationException(
          "Migration has already progressed past the endpoint of the \"trial mode\" state;"
              + " NoteDb is already the primary storage for some changes");
    }
    if (forceRebuild && state.compareTo(READ_WRITE_WITH_SEQUENCE_REVIEW_DB_PRIMARY) > 0) {
      throw new MigrationException(
          "Cannot force rebuild changes; NoteDb is already the primary storage for some changes");
    }
    setControlFlags();

    boolean rebuilt = false;
    while (state.compareTo(NOTE_DB) < 0) {
      if (state.equals(stopAtState)) {
        return;
      }
      boolean stillNeedsRebuild = forceRebuild && !rebuilt;
      if (trial && state.compareTo(READ_WRITE_NO_SEQUENCE) >= 0) {
        if (stillNeedsRebuild && state == READ_WRITE_NO_SEQUENCE) {
          // We're at the end state of trial mode, but still need a rebuild due to forceRebuild. Let
          // the loop go one more time.
        } else {
          return;
        }
      }
      switch (state) {
        case REVIEW_DB:
          state = turnOnWrites(state);
          break;
        case WRITE:
          state = rebuildAndEnableReads(state);
          rebuilt = true;
          break;
        case READ_WRITE_NO_SEQUENCE:
          if (stillNeedsRebuild) {
            state = rebuildAndEnableReads(state);
            rebuilt = true;
          } else {
            state = enableSequences(state);
          }
          break;
        case READ_WRITE_WITH_SEQUENCE_REVIEW_DB_PRIMARY:
          if (stillNeedsRebuild) {
            state = rebuildAndEnableReads(state);
            rebuilt = true;
          } else {
            state = setNoteDbPrimary(state);
          }
          break;
        case READ_WRITE_WITH_SEQUENCE_NOTE_DB_PRIMARY:
          // The only way we can get here is if there was a failure on a previous run of
          // setNoteDbPrimary, since that method moves to NOTE_DB if it completes
          // successfully. Assume that not all changes were converted and re-run the step.
          // migrateToNoteDbPrimary is a relatively fast no-op for already-migrated changes, so this
          // isn't actually repeating work.
          state = setNoteDbPrimary(state);
          break;
        case NOTE_DB:
          // Done!
          break;
        default:
          throw new MigrationException(
              "Migration out of the following state is not supported:\n" + state.toText());
      }
    }
  }

  private NotesMigrationState turnOnWrites(NotesMigrationState prev) throws IOException {
    return saveState(prev, WRITE);
  }

  private NotesMigrationState rebuildAndEnableReads(NotesMigrationState prev)
      throws OrmException, IOException {
    rebuild();
    return saveState(prev, READ_WRITE_NO_SEQUENCE);
  }

  private NotesMigrationState enableSequences(NotesMigrationState prev)
      throws OrmException, IOException {
    try (ReviewDb db = schemaFactory.open()) {
      @SuppressWarnings("deprecation")
      final int nextChangeId = db.nextChangeId();

      RepoSequence seq =
          new RepoSequence(
              repoManager,
              GitReferenceUpdated.DISABLED,
              allProjects,
              Sequences.NAME_CHANGES,
              // If sequenceGap is 0, this writes into the sequence ref the same ID that is returned
              // by the call to seq.next() below. If we actually used this as a change ID, that
              // would be a problem, but we just discard it, so this is safe.
              () -> nextChangeId + sequenceGap - 1,
              1,
              nextChangeId);
      seq.next();
    }
    return saveState(prev, READ_WRITE_WITH_SEQUENCE_REVIEW_DB_PRIMARY);
  }

  private NotesMigrationState setNoteDbPrimary(NotesMigrationState prev)
      throws MigrationException, OrmException, IOException {
    checkState(
        projects.isEmpty() && changes.isEmpty() && skipProjects.isEmpty(),
        "Should not have attempted setNoteDbPrimary with a subset of changes");
    checkState(
        prev == READ_WRITE_WITH_SEQUENCE_REVIEW_DB_PRIMARY
            || prev == READ_WRITE_WITH_SEQUENCE_NOTE_DB_PRIMARY,
        "Unexpected start state for setNoteDbPrimary: %s",
        prev);

    // Before changing the primary storage of old changes, ensure new changes are created with
    // NoteDb primary.
    prev = saveState(prev, READ_WRITE_WITH_SEQUENCE_NOTE_DB_PRIMARY);

    Stopwatch sw = Stopwatch.createStarted();
    log.info("Setting primary storage to NoteDb");
    List<Change.Id> allChanges;
    try (ReviewDb db = unwrapDb(schemaFactory.open())) {
      allChanges = Streams.stream(db.changes().all()).map(Change::getId).collect(toList());
    }

    try (ContextHelper contextHelper = new ContextHelper()) {
      List<ListenableFuture<Boolean>> futures =
          allChanges
              .stream()
              .map(
                  id ->
                      executor.submit(
                          () -> {
                            try (ManualRequestContext ctx = contextHelper.open()) {
                              try {
                                primaryStorageMigrator.migrateToNoteDbPrimary(id);
                              } catch (NoNoteDbStateException e) {
                                if (canSkipPrimaryStorageMigration(
                                    ctx.getReviewDbProvider().get(), id)) {
                                  log.warn(
                                      "Change {} previously failed to rebuild;"
                                          + " skipping primary storage migration",
                                      id,
                                      e);
                                } else {
                                  throw e;
                                }
                              }
                              return true;
                            } catch (Exception e) {
                              log.error("Error migrating primary storage for " + id, e);
                              return false;
                            }
                          }))
              .collect(toList());

      boolean ok = futuresToBoolean(futures, "Error migrating primary storage");
      double t = sw.elapsed(TimeUnit.MILLISECONDS) / 1000d;
      log.info(
          String.format(
              "Migrated primary storage of %d changes in %.01fs (%.01f/s)\n",
              allChanges.size(), t, allChanges.size() / t));
      if (!ok) {
        throw new MigrationException("Migrating primary storage for some changes failed, see log");
      }
    }

    return disableReviewDb(prev);
  }

  /**
   * Checks whether a change is so corrupt that it can be completely skipped by the primary storage
   * migration step.
   *
   * <p>To get to the point where this method is called from {@link #setNoteDbPrimary}, it means we
   * attempted to rebuild it, and encountered an error that was then caught in {@link
   * #rebuildProject} and skipped. As a result, there is no {@code noteDbState} field in the change
   * by the time we get to {@link #setNoteDbPrimary}, so {@code migrateToNoteDbPrimary} throws an
   * exception.
   *
   * <p>We have to do this hacky double-checking because we don't have a way for the rebuilding
   * phase to communicate to the primary storage migration phase that the change is skippable. It
   * would be possible to store this info in some field in this class, but there is no guarantee
   * that the rebuild and primary storage migration phases are run in the same JVM invocation.
   *
   * <p>In an ideal world, we could do this through the {@link
   * com.google.gerrit.server.notedb.NoteDbChangeState.PrimaryStorage} enum, having a separate value
   * for errors. However, that would be an invasive change touching many non-migration-related parts
   * of the NoteDb migration code, which is too risky to attempt in the stable branch where this bug
   * had to be fixed.
   *
   * <p>As of this writing, the only case where this happens is when a change has no patch sets.
   */
  private static boolean canSkipPrimaryStorageMigration(ReviewDb db, Change.Id id) {
    try {
      return Iterables.isEmpty(unwrapDb(db).patchSets().byChange(id));
    } catch (Exception e) {
      log.error("Error checking if change " + id + " can be skipped, assuming no", e);
      return false;
    }
  }

  private NotesMigrationState disableReviewDb(NotesMigrationState prev) throws IOException {
    return saveState(prev, NOTE_DB, c -> setAutoMigrate(c, false));
  }

  private Optional<NotesMigrationState> loadState() throws IOException {
    try {
      gerritConfig.load();
      noteDbConfig.load();
      return NotesMigrationState.forConfig(noteDbConfig);
    } catch (ConfigInvalidException | IllegalArgumentException e) {
      log.warn("error reading NoteDb migration options from " + noteDbConfig.getFile(), e);
      return Optional.empty();
    }
  }

  private NotesMigrationState saveState(
      NotesMigrationState expectedOldState, NotesMigrationState newState) throws IOException {
    return saveState(expectedOldState, newState, c -> {});
  }

  private NotesMigrationState saveState(
      NotesMigrationState expectedOldState,
      NotesMigrationState newState,
      Consumer<Config> additionalUpdates)
      throws IOException {
    synchronized (globalNotesMigration) {
      // This read-modify-write is racy. We're counting on the fact that no other Gerrit operation
      // modifies gerrit.config, and hoping that admins don't either.
      Optional<NotesMigrationState> actualOldState = loadState();
      if (!actualOldState.equals(Optional.of(expectedOldState))) {
        throw new MigrationException(
            "Cannot move to new state:\n"
                + newState.toText()
                + "\n\n"
                + "Expected this state in gerrit.config:\n"
                + expectedOldState.toText()
                + "\n\n"
                + (actualOldState.isPresent()
                    ? "But found this state:\n" + actualOldState.get().toText()
                    : "But could not parse the current state"));
      }

      preStateChange(expectedOldState, newState);

      newState.setConfigValues(noteDbConfig);
      additionalUpdates.accept(noteDbConfig);
      noteDbConfig.save();

      // Only set in-memory state once it's been persisted to storage.
      globalNotesMigration.setFrom(newState);
      log.info("Migration state: {} => {}", expectedOldState, newState);

      return newState;
    }
  }

  private void preStateChange(NotesMigrationState oldState, NotesMigrationState newState)
      throws IOException {
    for (NotesMigrationStateListener listener : listeners) {
      listener.preStateChange(oldState, newState);
    }
  }

  private void setControlFlags() throws MigrationException {
    synchronized (globalNotesMigration) {
      try {
        noteDbConfig.load();
        setAutoMigrate(noteDbConfig, autoMigrate);
        setTrialMode(noteDbConfig, trial);
        noteDbConfig.save();
      } catch (ConfigInvalidException | IOException e) {
        throw new MigrationException("Error saving auto-migration config", e);
      }
    }
  }

  public void rebuild() throws MigrationException, OrmException {
    if (!globalNotesMigration.commitChangeWrites()) {
      throw new MigrationException("Cannot rebuild without noteDb.changes.write=true");
    }
    Stopwatch sw = Stopwatch.createStarted();
    log.info("Rebuilding changes in NoteDb");

    ImmutableListMultimap<Project.NameKey, Change.Id> changesByProject = getChangesByProject();
    List<ListenableFuture<Boolean>> futures = new ArrayList<>();
    try (ContextHelper contextHelper = new ContextHelper()) {
      List<Project.NameKey> projectNames =
          Ordering.usingToString().sortedCopy(changesByProject.keySet());
      for (Project.NameKey project : projectNames) {
        ListenableFuture<Boolean> future =
            executor.submit(
                () -> {
                  try {
                    return rebuildProject(contextHelper.getReviewDb(), changesByProject, project);
                  } catch (Exception e) {
                    log.error("Error rebuilding project " + project, e);
                    return false;
                  }
                });
        futures.add(future);
      }

      boolean ok = futuresToBoolean(futures, "Error rebuilding projects");
      double t = sw.elapsed(TimeUnit.MILLISECONDS) / 1000d;
      log.info(
          String.format(
              "Rebuilt %d changes in %.01fs (%.01f/s)\n",
              changesByProject.size(), t, changesByProject.size() / t));
      if (!ok) {
        throw new MigrationException("Rebuilding some changes failed, see log");
      }
    }
  }

  private ImmutableListMultimap<Project.NameKey, Change.Id> getChangesByProject()
      throws OrmException {
    // Memoize all changes so we can close the db connection and allow other threads to use the full
    // connection pool.
    SetMultimap<Project.NameKey, Change.Id> out =
        MultimapBuilder.treeKeys(comparing(Project.NameKey::get))
            .treeSetValues(comparing(Change.Id::get))
            .build();
    try (ReviewDb db = unwrapDb(schemaFactory.open())) {
      if (!projects.isEmpty()) {
        return byProject(db.changes().all(), c -> projects.contains(c.getProject()), out);
      }
      if (!skipProjects.isEmpty()) {
        return byProject(db.changes().all(), c -> !skipProjects.contains(c.getProject()), out);
      }
      if (!changes.isEmpty()) {
        return byProject(db.changes().get(changes), c -> true, out);
      }
      return byProject(db.changes().all(), c -> true, out);
    }
  }

  private static ImmutableListMultimap<Project.NameKey, Change.Id> byProject(
      Iterable<Change> changes,
      Predicate<Change> pred,
      SetMultimap<Project.NameKey, Change.Id> out) {
    Streams.stream(changes).filter(pred).forEach(c -> out.put(c.getProject(), c.getId()));
    return ImmutableListMultimap.copyOf(out);
  }

  private static ObjectInserter newPackInserter(Repository repo) {
    if (!(repo instanceof FileRepository)) {
      return repo.newObjectInserter();
    }
    PackInserter ins = ((FileRepository) repo).getObjectDatabase().newPackInserter();
    ins.checkExisting(false);
    return ins;
  }

  private boolean rebuildProject(
      ReviewDb db,
      ImmutableListMultimap<Project.NameKey, Change.Id> allChanges,
      Project.NameKey project) {
    checkArgument(allChanges.containsKey(project));
    boolean ok = true;
    ProgressMonitor pm =
        new TextProgressMonitor(
            new PrintWriter(new BufferedWriter(new OutputStreamWriter(progressOut, UTF_8))));
    try (Repository changeRepo = repoManager.openRepository(project);
        // Only use a PackInserter for the change repo, not All-Users.
        //
        // It's not possible to share a single inserter for All-Users across all project tasks, and
        // we don't want to add one pack per project to All-Users. Adding many loose objects is
        // preferable to many packs.
        //
        // Anyway, the number of objects inserted into All-Users is proportional to the number
        // of pending draft comments, which should not be high (relative to the total number of
        // changes), so the number of loose objects shouldn't be too unreasonable.
        ObjectInserter changeIns = newPackInserter(changeRepo);
        ObjectReader changeReader = changeIns.newReader();
        RevWalk changeRw = new RevWalk(changeReader);
        Repository allUsersRepo = repoManager.openRepository(allUsers);
        ObjectInserter allUsersIns = allUsersRepo.newObjectInserter();
        ObjectReader allUsersReader = allUsersIns.newReader();
        RevWalk allUsersRw = new RevWalk(allUsersReader)) {
      ChainedReceiveCommands changeCmds = new ChainedReceiveCommands(changeRepo);
      ChainedReceiveCommands allUsersCmds = new ChainedReceiveCommands(allUsersRepo);

      Collection<Change.Id> changes = allChanges.get(project);
      pm.beginTask(FormatUtil.elide("Rebuilding " + project.get(), 50), changes.size());
      int toSave = 0;
      try {
        for (Change.Id changeId : changes) {
          // NoteDbUpdateManager assumes that all commands in its OpenRepo were added by itself, so
          // we can't share the top-level ChainedReceiveCommands. Use a new set of commands sharing
          // the same underlying repo, and copy commands back to the top-level
          // ChainedReceiveCommands later. This also assumes that each ref in the final list of
          // commands was only modified by a single NoteDbUpdateManager; since we use one manager
          // per change, and each ref corresponds to exactly one change, this assumption should be
          // safe.
          ChainedReceiveCommands tmpChangeCmds =
              new ChainedReceiveCommands(changeCmds.getRepoRefCache());
          ChainedReceiveCommands tmpAllUsersCmds =
              new ChainedReceiveCommands(allUsersCmds.getRepoRefCache());

          try (NoteDbUpdateManager manager =
              updateManagerFactory
                  .create(project)
                  .setAtomicRefUpdates(false)
                  .setSaveObjects(false)
                  .setChangeRepo(changeRepo, changeRw, changeIns, tmpChangeCmds)
                  .setAllUsersRepo(allUsersRepo, allUsersRw, allUsersIns, tmpAllUsersCmds)) {
            rebuild(db, changeId, manager);

            // Executing with dryRun=true writes all objects to the underlying inserters and adds
            // commands to the ChainedReceiveCommands. Afterwards, we can discard the manager, so we
            // don't keep using any memory beyond what may be buffered in the PackInserter.
            manager.execute(true);

            tmpChangeCmds.getCommands().values().forEach(c -> addCommand(changeCmds, c));
            tmpAllUsersCmds.getCommands().values().forEach(c -> addCommand(allUsersCmds, c));

            toSave++;
          } catch (NoPatchSetsException e) {
            log.warn(e.getMessage());
          } catch (ConflictingUpdateException ex) {
            log.warn(
                "Rebuilding detected a conflicting ReviewDb update for change {};"
                    + " will be auto-rebuilt at runtime",
                changeId);
          } catch (Throwable t) {
            log.error("Failed to rebuild change " + changeId, t);
            ok = false;
          }
          pm.update(1);
        }
      } finally {
        pm.endTask();
      }

      pm.beginTask(FormatUtil.elide("Saving " + project.get(), 50), ProgressMonitor.UNKNOWN);
      try {
        save(changeRepo, changeRw, changeIns, changeCmds);
        save(allUsersRepo, allUsersRw, allUsersIns, allUsersCmds);
        // This isn't really useful progress. If we passed a real ProgressMonitor to
        // BatchRefUpdate#execute we might get something more incremental, but that doesn't allow us
        // to specify the repo name in the task text.
        pm.update(toSave);
      } catch (LockFailureException e) {
        log.warn(
            "Rebuilding detected a conflicting NoteDb update for the following refs, which will"
                + " be auto-rebuilt at runtime: {}",
            e.getFailedRefs().stream().distinct().sorted().collect(joining(", ")));
      } catch (IOException e) {
        log.error("Failed to save NoteDb state for " + project, e);
      } finally {
        pm.endTask();
      }
    } catch (RepositoryNotFoundException e) {
      log.warn("Repository {} not found", project);
    } catch (IOException e) {
      log.error("Failed to rebuild project " + project, e);
    }
    return ok;
  }

  private void rebuild(ReviewDb db, Change.Id changeId, NoteDbUpdateManager manager)
      throws OrmException, IOException {
    // Match ChangeRebuilderImpl#stage, but without calling manager.stage(), since that can only be
    // called after building updates for all changes.
    Change change =
        ChangeRebuilderImpl.checkNoteDbState(ChangeNotes.readOneReviewDbChange(db, changeId));
    if (change == null) {
      // Could log here instead, but this matches the behavior of ChangeRebuilderImpl#stage.
      throw new NoSuchChangeException(changeId);
    }
    rebuilder.buildUpdates(manager, bundleReader.fromReviewDb(db, changeId));

    rebuilder.execute(db, changeId, manager, true, false);
  }

  private static void addCommand(ChainedReceiveCommands cmds, ReceiveCommand cmd) {
    // ChainedReceiveCommands doesn't allow no-ops, but these occur when rebuilding a
    // previously-rebuilt change.
    if (!cmd.getOldId().equals(cmd.getNewId())) {
      cmds.add(cmd);
    }
  }

  private void save(Repository repo, RevWalk rw, ObjectInserter ins, ChainedReceiveCommands cmds)
      throws IOException {
    if (cmds.isEmpty()) {
      return;
    }
    ins.flush();
    BatchRefUpdate bru = repo.getRefDatabase().newBatchUpdate();
    bru.setRefLogMessage("Migrate changes to NoteDb", false);
    bru.setRefLogIdent(serverIdent.get());
    bru.setAtomic(false);
    bru.setAllowNonFastForwards(true);
    cmds.addTo(bru);
    RefUpdateUtil.executeChecked(bru, rw);
  }

  private static boolean futuresToBoolean(List<ListenableFuture<Boolean>> futures, String errMsg) {
    try {
      return Futures.allAsList(futures).get().stream().allMatch(b -> b);
    } catch (InterruptedException | ExecutionException e) {
      log.error(errMsg, e);
      return false;
    }
  }

  private class ContextHelper implements AutoCloseable {
    private final Thread callingThread;
    private ReviewDb db;
    private Runnable closeDb;

    ContextHelper() {
      callingThread = Thread.currentThread();
    }

    ManualRequestContext open() throws OrmException {
      return new ManualRequestContext(
          userFactory.create(),
          // Reuse the same lazily-opened ReviewDb on the original calling thread, otherwise open
          // SchemaFactory in the normal way.
          Thread.currentThread().equals(callingThread) ? this::getReviewDb : schemaFactory,
          requestContext);
    }

    synchronized ReviewDb getReviewDb() throws OrmException {
      if (db == null) {
        ReviewDb actual = schemaFactory.open();
        closeDb = actual::close;
        db =
            new ReviewDbWrapper(unwrapDb(actual)) {
              @Override
              public void close() {
                // Closed by ContextHelper#close.
              }
            };
      }
      return db;
    }

    @Override
    public synchronized void close() {
      if (db != null) {
        closeDb.run();
        db = null;
        closeDb = null;
      }
    }
  }
}