// 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.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.MultimapBuilder;
import com.google.common.collect.Sets;
import com.google.gerrit.elasticsearch.ElasticMapping.MappingProperties;
import com.google.gerrit.elasticsearch.builders.QueryBuilder;
import com.google.gerrit.elasticsearch.builders.SearchSourceBuilder;
import com.google.gerrit.elasticsearch.bulk.BulkRequest;
import com.google.gerrit.elasticsearch.bulk.DeleteRequest;
import com.google.gerrit.elasticsearch.bulk.IndexRequest;
import com.google.gerrit.elasticsearch.bulk.UpdateRequest;
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.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.gson.JsonParser;
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 java.io.IOException;
import java.io.UnsupportedEncodingException;
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.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.elasticsearch.client.Response;
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 {
    final MappingProperties changes;
    final MappingProperties openChanges;
    final MappingProperties closedChanges;

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

  private static final String CHANGES = "changes";
  private static final String OPEN_CHANGES = "open_" + CHANGES;
  private static final String CLOSED_CHANGES = "closed_" + CHANGES;
  private static final String ALL_CHANGES = OPEN_CHANGES + "," + CLOSED_CHANGES;
  private final ChangeMapping mapping;
  private final Provider<ReviewDb> db;
  private final ChangeData.Factory changeDataFactory;
  private final Schema<ChangeData> schema;

  @Inject
  ElasticChangeIndex(
      ElasticConfiguration cfg,
      Provider<ReviewDb> db,
      ChangeData.Factory changeDataFactory,
      SitePaths sitePaths,
      ElasticRestClientProvider client,
      @Assisted Schema<ChangeData> schema) {
    super(cfg, sitePaths, schema, client, CHANGES, ALL_CHANGES);
    this.db = db;
    this.changeDataFactory = changeDataFactory;
    this.schema = schema;
    this.mapping = new ChangeMapping(schema, client.adapter());
  }

  @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);
    }

    ElasticQueryAdapter adapter = client.adapter();
    BulkRequest bulk =
        new IndexRequest(getId(cd), indexName, adapter.getType(insertIndex), adapter)
            .add(new UpdateRequest<>(schema, cd));
    if (adapter.deleteToReplace()) {
      bulk.add(new DeleteRequest(cd.getId().toString(), indexName, deleteIndex, adapter));
    }

    String uri = getURI(type, BULK);
    Response response = postRequest(uri, bulk, getRefreshParam());
    int statusCode = response.getStatusLine().getStatusCode();
    if (statusCode != HttpStatus.SC_OK) {
      throw new IOException(
          String.format(
              "Failed to replace change %s in index %s: %s", cd.getId(), indexName, statusCode));
    }
  }

  @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 (!client.adapter().omitType()) {
      if (client.adapter().useV6Type()) {
        if (!Sets.intersection(statuses, OPEN_STATUSES).isEmpty()
            || !Sets.intersection(statuses, CLOSED_STATUSES).isEmpty()) {
          indexes.add(ElasticQueryAdapter.V6_TYPE);
        }
      } else {
        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 String getDeleteActions(Id c) {
    if (!client.adapter().useV5Type()) {
      return delete(client.adapter().getType(), c);
    }
    return delete(OPEN_CHANGES, c) + delete(CLOSED_CHANGES, c);
  }

  @Override
  protected String getMappings() {
    if (!client.adapter().useV5Type()) {
      return getMappingsFor(client.adapter().getType(), mapping.changes);
    }
    return gson.toJson(ImmutableMap.of(MAPPINGS, mapping));
  }

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

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

    QuerySource(List<String> types, Predicate<ChangeData> p, QueryOptions opts)
        throws QueryParseException {
      QueryBuilder qb = queryBuilder.toQueryBuilder(p);
      fields = IndexUtils.changeFields(opts);
      SearchSourceBuilder searchSource =
          new SearchSourceBuilder(client.adapter())
              .query(qb)
              .from(opts.start())
              .size(opts.limit())
              .fields(Lists.newArrayList(fields));

      search = getSearch(searchSource, getSortArray());
      this.types = types;
    }

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

    @Override
    public ResultSet<ChangeData> read() throws OrmException {
      try {
        List<ChangeData> results = Collections.emptyList();
        String uri = getURI(types);
        Response response = postRequest(uri, search);
        StatusLine statusLine = response.getStatusLine();
        if (statusLine.getStatusCode() == HttpStatus.SC_OK) {
          String content = getContent(response);
          JsonObject obj =
              new JsonParser().parse(content).getAsJsonObject().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(statusLine.getReasonPhrase());
        }
        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;
    }

    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();
        cd.setChangedLines(added, deleted);
      }

      // Star.
      JsonElement starredElement = source.get(ChangeField.STAR.getName());
      if (starredElement != null) {
        ListMultimap<Account.Id, String> stars =
            MultimapBuilder.hashKeys().arrayListValues().build();
        JsonArray starBy = starredElement.getAsJsonArray();
        if (starBy.size() > 0) {
          for (int i = 0; i < starBy.size(); i++) {
            String[] indexableFields = starBy.get(i).getAsString().split(":");
            Account.Id id = Account.Id.parse(indexableFields[0]);
            stars.put(id, indexableFields[1]);
          }
        }
        cd.setStars(stars);
      }

      // 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());
    }

    private JsonArray getSortArray() {
      JsonObject properties = new JsonObject();
      properties.addProperty(ORDER, "desc");

      JsonArray sortArray = new JsonArray();
      addNamedElement(ChangeField.UPDATED.getName(), properties, sortArray);
      addNamedElement(ChangeField.LEGACY_ID.getName(), properties, sortArray);
      return sortArray;
    }
  }

  private String getURI(List<String> types) throws UnsupportedEncodingException {
    String joinedTypes = String.join(",", types);
    return getURI(joinedTypes, SEARCH);
  }
}