// Copyright (C) 2013 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.index;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.collect.Lists;
import com.google.gerrit.extensions.registration.DynamicSet;
import com.google.gerrit.index.Index;
import com.google.gerrit.index.IndexCollection;
import com.google.gerrit.index.IndexDefinition;
import com.google.gerrit.index.SiteIndexer;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OnlineReindexer<K, V, I extends Index<K, V>> {
  private static final Logger log = LoggerFactory.getLogger(OnlineReindexer.class);

  private final String name;
  private final IndexCollection<K, V, I> indexes;
  private final SiteIndexer<K, V, I> batchIndexer;
  private final int oldVersion;
  private final int newVersion;
  private final DynamicSet<OnlineUpgradeListener> listeners;
  private I index;
  private final AtomicBoolean running = new AtomicBoolean();

  public OnlineReindexer(
      IndexDefinition<K, V, I> def,
      int oldVersion,
      int newVersion,
      DynamicSet<OnlineUpgradeListener> listeners) {
    this.name = def.getName();
    this.indexes = def.getIndexCollection();
    this.batchIndexer = def.getSiteIndexer();
    this.oldVersion = oldVersion;
    this.newVersion = newVersion;
    this.listeners = listeners;
  }

  public void start() {
    if (running.compareAndSet(false, true)) {
      Thread t =
          new Thread() {
            @Override
            public void run() {
              boolean ok = false;
              try {
                reindex();
                ok = true;
              } finally {
                running.set(false);
                if (!ok) {
                  for (OnlineUpgradeListener listener : listeners) {
                    listener.onFailure(name, oldVersion, newVersion);
                  }
                }
              }
            }
          };
      t.setName(String.format("Reindex v%d-v%d", version(indexes.getSearchIndex()), newVersion));
      t.start();
    }
  }

  public boolean isRunning() {
    return running.get();
  }

  public int getVersion() {
    return newVersion;
  }

  private static int version(Index<?, ?> i) {
    return i.getSchema().getVersion();
  }

  private void reindex() {
    for (OnlineUpgradeListener listener : listeners) {
      listener.onStart(name, oldVersion, newVersion);
    }
    index =
        checkNotNull(
            indexes.getWriteIndex(newVersion),
            "not an active write schema version: %s",
            newVersion);
    log.info(
        "Starting online reindex from schema version {} to {}",
        version(indexes.getSearchIndex()),
        version(index));
    SiteIndexer.Result result = batchIndexer.indexAll(index);
    if (!result.success()) {
      log.error(
          "Online reindex of schema version {} failed. Successfully"
              + " indexed {} changes, failed to index {} changes",
          version(index),
          result.doneCount(),
          result.failedCount());
      return;
    }
    log.info("Reindex to version {} complete", version(index));
    activateIndex();
    for (OnlineUpgradeListener listener : listeners) {
      listener.onSuccess(name, oldVersion, newVersion);
    }
  }

  public void activateIndex() {
    indexes.setSearchIndex(index);
    log.info("Using schema version {}", version(index));
    try {
      index.markReady(true);
    } catch (IOException e) {
      log.warn("Error activating new schema version {}", version(index));
    }

    List<I> toRemove = Lists.newArrayListWithExpectedSize(1);
    for (I i : indexes.getWriteIndexes()) {
      if (version(i) != version(index)) {
        toRemove.add(i);
      }
    }
    for (I i : toRemove) {
      try {
        i.markReady(false);
        indexes.removeWriteIndex(version(i));
      } catch (IOException e) {
        log.warn("Error deactivating old schema version {}", version(i));
      }
    }
  }
}

@@ -2613,16 +2613,27 @@
 setName(
+%0A          
 String.f
@@ -2647,16 +2647,19 @@
 Reindex 
+%25s 
 v%25d-v%25d%22
@@ -2659,16 +2659,22 @@
 %25d-v%25d%22,
+ name,
  version
@@ -3225,17 +3225,38 @@
 sion: %25s
-%22
+ %25s%22,%0A            name
 ,%0A      
@@ -3320,16 +3320,22 @@
  reindex
+ of %7B%7D
  from sc
@@ -3358,16 +3358,30 @@
 to %7B%7D%22,%0A
+        name,%0A
         
@@ -3576,16 +3576,19 @@
 index of
+ %7B%7D
  schema 
@@ -3648,24 +3648,16 @@
 dexed %7B%7D
- changes
 , failed
@@ -3672,17 +3672,25 @@
 x %7B%7D
- changes%22
+%22,%0A          name
 ,%0A  
@@ -3817,16 +3817,19 @@
 %22Reindex
+ %7B%7D
  to vers
@@ -3845,16 +3845,22 @@
 mplete%22,
+ name,
  version
@@ -4098,16 +4098,19 @@
 o(%22Using
+ %7B%7D
  schema 
@@ -4113,32 +4113,38 @@
 ema version %7B%7D%22,
+ name,
  version(index))
@@ -4250,16 +4250,19 @@
 ting new
+ %7B%7D
  schema 
@@ -4265,32 +4265,38 @@
 ema version %7B%7D%22,
+ name,
  version(index))
@@ -4676,16 +4676,19 @@
 ting old
+ %7B%7D
  schema 
@@ -4699,16 +4699,22 @@
 ion %7B%7D%22,
+ name,
  version

Include index name in task name and log messages