// Copyright (C) 2014 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.elasticsearch;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.gerrit.server.index.change.ChangeField.APPROVAL_CODEC;
import static com.google.gerrit.server.index.change.ChangeField.CHANGE_CODEC;
import static com.google.gerrit.server.index.change.ChangeField.PATCH_SET_CODEC;
import static com.google.gerrit.server.index.change.ChangeIndexRewriter.CLOSED_STATUSES;
import static com.google.gerrit.server.index.change.ChangeIndexRewriter.OPEN_STATUSES;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.commons.codec.binary.Base64.decodeBase64;

import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.gerrit.elasticsearch.ElasticMapping.MappingProperties;
import com.google.gerrit.index.QueryOptions;
import com.google.gerrit.index.Schema;
import com.google.gerrit.index.query.Predicate;
import com.google.gerrit.index.query.QueryParseException;
import com.google.gerrit.reviewdb.client.Account;
import com.google.gerrit.reviewdb.client.Change;
import com.google.gerrit.reviewdb.client.Change.Id;
import com.google.gerrit.reviewdb.client.Project;
import com.google.gerrit.reviewdb.server.ReviewDb;
import com.google.gerrit.server.ReviewerByEmailSet;
import com.google.gerrit.server.ReviewerSet;
import com.google.gerrit.server.config.GerritServerConfig;
import com.google.gerrit.server.config.SitePaths;
import com.google.gerrit.server.index.IndexUtils;
import com.google.gerrit.server.index.change.ChangeField;
import com.google.gerrit.server.index.change.ChangeIndex;
import com.google.gerrit.server.index.change.ChangeIndexRewriter;
import com.google.gerrit.server.project.SubmitRuleOptions;
import com.google.gerrit.server.query.change.ChangeData;
import com.google.gerrit.server.query.change.ChangeDataSource;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gwtorm.server.OrmException;
import com.google.gwtorm.server.ResultSet;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.assistedinject.Assisted;
import io.searchbox.client.JestResult;
import io.searchbox.core.Bulk;
import io.searchbox.core.Bulk.Builder;
import io.searchbox.core.Search;
import io.searchbox.core.search.sort.Sort;
import io.searchbox.core.search.sort.Sort.Sorting;
import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import org.apache.commons.codec.binary.Base64;
import org.eclipse.jgit.lib.Config;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Secondary index implementation using Elasticsearch. */
class ElasticChangeIndex extends AbstractElasticIndex<Change.Id, ChangeData>
    implements ChangeIndex {
  private static final Logger log = LoggerFactory.getLogger(ElasticChangeIndex.class);

  static class ChangeMapping {
    MappingProperties openChanges;
    MappingProperties closedChanges;

    ChangeMapping(Schema<ChangeData> schema) {
      MappingProperties mapping = ElasticMapping.createMapping(schema);
      this.openChanges = mapping;
      this.closedChanges = mapping;
    }
  }

  static final String CHANGES_PREFIX = "changes_";
  static final String OPEN_CHANGES = "open_changes";
  static final String CLOSED_CHANGES = "closed_changes";

  private final ChangeMapping mapping;
  private final Provider<ReviewDb> db;
  private final ChangeData.Factory changeDataFactory;

  @Inject
  ElasticChangeIndex(
      @GerritServerConfig Config cfg,
      Provider<ReviewDb> db,
      ChangeData.Factory changeDataFactory,
      SitePaths sitePaths,
      JestClientBuilder clientBuilder,
      @Assisted Schema<ChangeData> schema) {
    super(cfg, sitePaths, schema, clientBuilder, CHANGES_PREFIX);
    this.db = db;
    this.changeDataFactory = changeDataFactory;
    mapping = new ChangeMapping(schema);
  }

  @Override
  public void replace(ChangeData cd) throws IOException {
    String deleteIndex;
    String insertIndex;

    try {
      if (cd.change().getStatus().isOpen()) {
        insertIndex = OPEN_CHANGES;
        deleteIndex = CLOSED_CHANGES;
      } else {
        insertIndex = CLOSED_CHANGES;
        deleteIndex = OPEN_CHANGES;
      }
    } catch (OrmException e) {
      throw new IOException(e);
    }

    Bulk bulk =
        new Bulk.Builder()
            .defaultIndex(indexName)
            .defaultType("changes")
            .addAction(insert(insertIndex, cd))
            .addAction(delete(deleteIndex, cd.getId()))
            .refresh(true)
            .build();
    JestResult result = client.execute(bulk);
    if (!result.isSucceeded()) {
      throw new IOException(
          String.format(
              "Failed to replace change %s in index %s: %s",
              cd.getId(), indexName, result.getErrorMessage()));
    }
  }

  @Override
  public ChangeDataSource getSource(Predicate<ChangeData> p, QueryOptions opts)
      throws QueryParseException {
    Set<Change.Status> statuses = ChangeIndexRewriter.getPossibleStatus(p);
    List<String> indexes = Lists.newArrayListWithCapacity(2);
    if (!Sets.intersection(statuses, OPEN_STATUSES).isEmpty()) {
      indexes.add(OPEN_CHANGES);
    }
    if (!Sets.intersection(statuses, CLOSED_STATUSES).isEmpty()) {
      indexes.add(CLOSED_CHANGES);
    }
    return new QuerySource(indexes, p, opts);
  }

  @Override
  protected Builder addActions(Builder builder, Id c) {
    return builder.addAction(delete(OPEN_CHANGES, c)).addAction(delete(OPEN_CHANGES, c));
  }

  @Override
  protected String getMappings() {
    return gson.toJson(ImmutableMap.of("mappings", mapping));
  }

  @Override
  protected String getId(ChangeData cd) {
    return cd.getId().toString();
  }

  private class QuerySource implements ChangeDataSource {
    private final Search search;
    private final Set<String> fields;

    QuerySource(List<String> types, Predicate<ChangeData> p, QueryOptions opts)
        throws QueryParseException {
      List<Sort> sorts =
          ImmutableList.of(
              new Sort(ChangeField.UPDATED.getName(), Sorting.DESC),
              new Sort(ChangeField.LEGACY_ID.getName(), Sorting.DESC));
      for (Sort sort : sorts) {
        sort.setIgnoreUnmapped();
      }
      QueryBuilder qb = queryBuilder.toQueryBuilder(p);
      fields = IndexUtils.changeFields(opts);
      SearchSourceBuilder searchSource =
          new SearchSourceBuilder()
              .query(qb)
              .from(opts.start())
              .size(opts.limit())
              .fields(Lists.newArrayList(fields));

      search =
          new Search.Builder(searchSource.toString())
              .addType(types)
              .addSort(sorts)
              .addIndex(indexName)
              .build();
    }

    @Override
    public int getCardinality() {
      return 10;
    }

    @Override
    public ResultSet<ChangeData> read() throws OrmException {
      try {
        List<ChangeData> results = Collections.emptyList();
        JestResult result = client.execute(search);
        if (result.isSucceeded()) {
          JsonObject obj = result.getJsonObject().getAsJsonObject("hits");
          if (obj.get("hits") != null) {
            JsonArray json = obj.getAsJsonArray("hits");
            results = Lists.newArrayListWithCapacity(json.size());
            for (int i = 0; i < json.size(); i++) {
              results.add(toChangeData(json.get(i)));
            }
          }
        } else {
          log.error(result.getErrorMessage());
        }
        final List<ChangeData> r = Collections.unmodifiableList(results);
        return new ResultSet<ChangeData>() {
          @Override
          public Iterator<ChangeData> iterator() {
            return r.iterator();
          }

          @Override
          public List<ChangeData> toList() {
            return r;
          }

          @Override
          public void close() {
            // Do nothing.
          }
        };
      } catch (IOException e) {
        throw new OrmException(e);
      }
    }

    @Override
    public boolean hasChange() {
      return false;
    }

    @Override
    public String toString() {
      return search.toString();
    }

    private ChangeData toChangeData(JsonElement json) {
      JsonElement sourceElement = json.getAsJsonObject().get("_source");
      if (sourceElement == null) {
        sourceElement = json.getAsJsonObject().get("fields");
      }
      JsonObject source = sourceElement.getAsJsonObject();
      JsonElement c = source.get(ChangeField.CHANGE.getName());

      if (c == null) {
        int id = source.get(ChangeField.LEGACY_ID.getName()).getAsInt();
        // IndexUtils#changeFields ensures either CHANGE or PROJECT is always present.
        String projectName = checkNotNull(source.get(ChangeField.PROJECT.getName()).getAsString());
        return changeDataFactory.create(
            db.get(), new Project.NameKey(projectName), new Change.Id(id));
      }

      ChangeData cd =
          changeDataFactory.create(
              db.get(), CHANGE_CODEC.decode(Base64.decodeBase64(c.getAsString())));

      // Patch sets.
      cd.setPatchSets(decodeProtos(source, ChangeField.PATCH_SET.getName(), PATCH_SET_CODEC));

      // Approvals.
      if (source.get(ChangeField.APPROVAL.getName()) != null) {
        cd.setCurrentApprovals(
            decodeProtos(source, ChangeField.APPROVAL.getName(), APPROVAL_CODEC));
      } else if (fields.contains(ChangeField.APPROVAL.getName())) {
        cd.setCurrentApprovals(Collections.emptyList());
      }

      JsonElement addedElement = source.get(ChangeField.ADDED.getName());
      JsonElement deletedElement = source.get(ChangeField.DELETED.getName());
      if (addedElement != null && deletedElement != null) {
        // Changed lines.
        int added = addedElement.getAsInt();
        int deleted = deletedElement.getAsInt();
        if (added != 0 && deleted != 0) {
          cd.setChangedLines(added, deleted);
        }
      }

      // Mergeable.
      JsonElement mergeableElement = source.get(ChangeField.MERGEABLE.getName());
      if (mergeableElement != null) {
        String mergeable = mergeableElement.getAsString();
        if ("1".equals(mergeable)) {
          cd.setMergeable(true);
        } else if ("0".equals(mergeable)) {
          cd.setMergeable(false);
        }
      }

      // Reviewed-by.
      if (source.get(ChangeField.REVIEWEDBY.getName()) != null) {
        JsonArray reviewedBy = source.get(ChangeField.REVIEWEDBY.getName()).getAsJsonArray();
        if (reviewedBy.size() > 0) {
          Set<Account.Id> accounts = Sets.newHashSetWithExpectedSize(reviewedBy.size());
          for (int i = 0; i < reviewedBy.size(); i++) {
            int aId = reviewedBy.get(i).getAsInt();
            if (reviewedBy.size() == 1 && aId == ChangeField.NOT_REVIEWED) {
              break;
            }
            accounts.add(new Account.Id(aId));
          }
          cd.setReviewedBy(accounts);
        }
      } else if (fields.contains(ChangeField.REVIEWEDBY.getName())) {
        cd.setReviewedBy(Collections.emptySet());
      }

      if (source.get(ChangeField.REVIEWER.getName()) != null) {
        cd.setReviewers(
            ChangeField.parseReviewerFieldValues(
                FluentIterable.from(source.get(ChangeField.REVIEWER.getName()).getAsJsonArray())
                    .transform(JsonElement::getAsString)));
      } else if (fields.contains(ChangeField.REVIEWER.getName())) {
        cd.setReviewers(ReviewerSet.empty());
      }

      if (source.get(ChangeField.REVIEWER_BY_EMAIL.getName()) != null) {
        cd.setReviewersByEmail(
            ChangeField.parseReviewerByEmailFieldValues(
                FluentIterable.from(
                        source.get(ChangeField.REVIEWER_BY_EMAIL.getName()).getAsJsonArray())
                    .transform(JsonElement::getAsString)));
      } else if (fields.contains(ChangeField.REVIEWER_BY_EMAIL.getName())) {
        cd.setReviewersByEmail(ReviewerByEmailSet.empty());
      }

      if (source.get(ChangeField.PENDING_REVIEWER.getName()) != null) {
        cd.setPendingReviewers(
            ChangeField.parseReviewerFieldValues(
                FluentIterable.from(
                        source.get(ChangeField.PENDING_REVIEWER.getName()).getAsJsonArray())
                    .transform(JsonElement::getAsString)));
      } else if (fields.contains(ChangeField.PENDING_REVIEWER.getName())) {
        cd.setPendingReviewers(ReviewerSet.empty());
      }

      if (source.get(ChangeField.PENDING_REVIEWER_BY_EMAIL.getName()) != null) {
        cd.setPendingReviewersByEmail(
            ChangeField.parseReviewerByEmailFieldValues(
                FluentIterable.from(
                        source
                            .get(ChangeField.PENDING_REVIEWER_BY_EMAIL.getName())
                            .getAsJsonArray())
                    .transform(JsonElement::getAsString)));
      } else if (fields.contains(ChangeField.PENDING_REVIEWER_BY_EMAIL.getName())) {
        cd.setPendingReviewersByEmail(ReviewerByEmailSet.empty());
      }
      decodeSubmitRecords(
          source,
          ChangeField.STORED_SUBMIT_RECORD_STRICT.getName(),
          ChangeField.SUBMIT_RULE_OPTIONS_STRICT,
          cd);
      decodeSubmitRecords(
          source,
          ChangeField.STORED_SUBMIT_RECORD_LENIENT.getName(),
          ChangeField.SUBMIT_RULE_OPTIONS_LENIENT,
          cd);
      decodeUnresolvedCommentCount(source, ChangeField.UNRESOLVED_COMMENT_COUNT.getName(), cd);

      if (fields.contains(ChangeField.REF_STATE.getName())) {
        cd.setRefStates(getByteArray(source, ChangeField.REF_STATE.getName()));
      }
      if (fields.contains(ChangeField.REF_STATE_PATTERN.getName())) {
        cd.setRefStatePatterns(getByteArray(source, ChangeField.REF_STATE_PATTERN.getName()));
      }

      return cd;
    }

    private Iterable<byte[]> getByteArray(JsonObject source, String name) {
      JsonElement element = source.get(name);
      return element != null
          ? Iterables.transform(element.getAsJsonArray(), e -> Base64.decodeBase64(e.getAsString()))
          : Collections.emptyList();
    }

    private void decodeSubmitRecords(
        JsonObject doc, String fieldName, SubmitRuleOptions opts, ChangeData out) {
      JsonArray records = doc.getAsJsonArray(fieldName);
      if (records == null) {
        return;
      }
      ChangeField.parseSubmitRecords(
          FluentIterable.from(records)
              .transform(i -> new String(decodeBase64(i.toString()), UTF_8))
              .toList(),
          opts,
          out);
    }

    private void decodeUnresolvedCommentCount(JsonObject doc, String fieldName, ChangeData out) {
      JsonElement count = doc.get(fieldName);
      if (count == null) {
        return;
      }
      out.setUnresolvedCommentCount(count.getAsInt());
    }
  }
}