/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.jackrabbit.oak.plugins.index.elastic.index;

import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Type;
import org.apache.jackrabbit.oak.commons.PathUtils;
import org.apache.jackrabbit.oak.commons.log.LogSilencer;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticPropertyDefinition;
import org.apache.jackrabbit.oak.plugins.index.elastic.query.inference.InferenceConfig;
import org.apache.jackrabbit.oak.plugins.index.elastic.query.inference.InferenceConstants;
import org.apache.jackrabbit.oak.plugins.index.elastic.util.ElasticIndexUtils;
import org.apache.jackrabbit.oak.plugins.index.search.Aggregate;
import org.apache.jackrabbit.oak.plugins.index.search.FieldNames;
import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.search.PropertyDefinition;
import org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextDocumentMaker;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.List;

public class ElasticDocumentMaker extends FulltextDocumentMaker<ElasticDocument> {

    private static final Logger LOG = LoggerFactory.getLogger(ElasticDocumentMaker.class);
    private static final int BLOB_LENGTH_DIVISOR = 4;

    private static final LogSilencer LOG_SILENCER = new LogSilencer(Duration.ofSeconds(10).toMillis(), 10);
    private static final String LOG_KEY_COULD_NOT_CONVERT_PROPERTY = "Could not convert property";
    private static final String LOG_KEY_SIMILARITY_BINARIES_WRONG_DIMENSION = "Similarity binaries wrong dimension";

    public ElasticDocumentMaker(@Nullable FulltextBinaryTextExtractor textExtractor,
                                @NotNull IndexDefinition definition,
                                IndexDefinition.IndexingRule indexingRule, @NotNull String path) {
        super(textExtractor, definition, indexingRule, path);
    }

    @Override
    protected ElasticDocument initDoc() {
        return new ElasticDocument(path, (int) definition.getDefinitionNodeState().getLong(ElasticIndexDefinition.PROP_INDEX_NAME_SEED));
    }

    @Override
    protected ElasticDocument finalizeDoc(ElasticDocument doc, boolean dirty, boolean facet) {
        // Evaluate path restrictions is enabled by default in elastic. Always index ancestors.
        // When specifically disabled, we will keep indexing it, but the field won't be used at query time
        doc.indexAncestors(path);
        doc.setLastUpdated(System.currentTimeMillis());
        if (InferenceConfig.getInstance().isInferenceEnabled()
            && InferenceConfig.getInstance().getInferenceIndexConfig(PathUtils.getName(definition.getIndexName())).isEnabled()) {
            doc.addProperty(InferenceConstants.ENRICH_NODE,
                InferenceConfig.getInstance().getEnricherStatus());
        }
        return doc;
    }

    // TODO: needed only for oak-lucene. Should be removed from oak-search
    @Override
    protected boolean isFacetingEnabled() {
        return false;
    }

    @Override
    protected boolean indexTypeOrderedFields(ElasticDocument doc, String pname, int tag, PropertyState property, PropertyDefinition pd) {
        // TODO: check the conjecture below
        // ES doesn't seem to require special mapping to sort so we don't need to add anything
        return false;
    }

    @Override
    protected boolean addBinary(ElasticDocument doc, String path, List<String> binaryValues) {
        boolean added = false;
        for (String binaryValue : binaryValues) {
            if (path != null) {
                doc.addFulltextRelative(path, binaryValue);
            } else {
                doc.addFulltext(binaryValue);
            }

            added = true;
        }

        return added;
    }

    @Override
    protected boolean indexFacetProperty(ElasticDocument doc, int tag, PropertyState property, String pname) {
        // faceting on ES works automatically for keyword (propertyIndex) and subsequently query params.
        // We don't need to add anything.
        // https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html
        return false;
    }

    @Override
    protected void indexAnalyzedProperty(ElasticDocument doc, String pname, String value, PropertyDefinition pd) {
        // no need to do anything here. Analyzed properties are persisted in Elastic
        // using multi fields. The analyzed properties are set calling #indexTypedProperty
    }

    @Override
    protected void indexSuggestValue(ElasticDocument doc, String value) {
        if (value != null) {
            String v = value.trim();
            if (!v.isEmpty()) {
                doc.addSuggest(v);
            }
        }
    }

    @Override
    protected void indexSpellcheckValue(ElasticDocument doc, String value) {
        if (value != null) {
            String v = value.trim();
            if (!v.isEmpty()) {
                doc.addSpellcheck(v);
            }
        }
    }

    @Override
    protected void indexFulltextValue(ElasticDocument doc, String value) {
        if (value != null) {
            String v = value.trim();
            if (!v.isEmpty()) {
                doc.addFulltext(v);
            }
        }
    }

    /**
     * We store the value in :fulltext only when the {@link PropertyDefinition} has a regular expression (that means we
     * were not able to create a ft property at mapping time) or the property is not analyzed or the type could be ignored
     * in case is malformed (eg: a date that cannot be parsed).
     */
    @Override
    protected boolean isFulltextValuePersistedAtNode(PropertyDefinition pd) {
        return pd.isRegexp || !pd.analyzed
                || pd.getType() == Type.DATE.tag()
                || pd.getType() == Type.BOOLEAN.tag()
                || pd.getType() == Type.LONG.tag()
                || pd.getType() == Type.DOUBLE.tag();
    }

    @Override
    protected void indexTypedProperty(ElasticDocument doc, PropertyState property, String propertyName, PropertyDefinition pd, int i) {
        // Get the Type tag from the defined index definition here - and not from the actual persisted property state - this way in case
        // If the actual property value is different from the property type defined in the index definition/mapping - this will try to convert the property if possible,
        // otherwise will log a warning and not try and add the property to index. If we try and index incompatible data types (like String to Date),
        // we would get an exception while indexing the node on elastic search and other properties for the node will also don't get indexed. (See OAK-9665).
        String fieldName = ElasticIndexUtils.fieldName(propertyName);
        if (pd.isRegexp) {
            ElasticPropertyDefinition epd = (ElasticPropertyDefinition) pd;
            if (epd.isFlattened()) {
                fieldName = FieldNames.FLATTENED_FIELD_PREFIX +
                        ElasticIndexUtils.fieldName(epd.nodeName) + "." + fieldName;
            }
        }
        int tag = pd.getType();
        Object f;
        try {
            if (tag == Type.LONG.tag()) {
                f = property.getValue(Type.LONG, i);
            } else if (tag == Type.DATE.tag()) {
                f = property.getValue(Type.DATE, i);
            } else if (tag == Type.DOUBLE.tag()) {
                f = property.getValue(Type.DOUBLE, i);
            } else if (tag == Type.BOOLEAN.tag()) {
                f = property.getValue(Type.BOOLEAN, i).toString();
            } else if (tag == Type.BINARY.tag()) {
                // ignore - never call getValue(Type.STRING) on a binary (see OAK-12133)
                f = null;
            } else {
                f = property.getValue(Type.STRING, i);
            }

            if (f != null) {
                doc.addProperty(fieldName, f);
            }
        } catch (Exception e) {
            if (!LOG_SILENCER.silence(LOG_KEY_COULD_NOT_CONVERT_PROPERTY)) {
                LOG.warn(
                        "[{}] Ignoring property. Could not convert property {} (field {}) of type {} to type {} for path {}. Error: {}",
                        getIndexName(), propertyName, fieldName,
                        Type.fromTag(property.getType().tag(), false),
                        Type.fromTag(tag, false), path, e.toString());
            }
        }
    }

    /**
     * Empty method implementation. Ancestors are always indexed
     *
     * @see ElasticDocumentMaker#finalizeDoc
     */
    @Override
    protected void indexAncestors(ElasticDocument doc, String path) { /* empty */ }

    @Override
    protected void indexNotNullProperty(ElasticDocument doc, PropertyDefinition pd) {
        // Elastic efficiently supports exist queries thanks to the special _field_names metadata field
        // to determine which fields are present in each document
    }

    @Override
    protected void indexNullProperty(ElasticDocument doc, PropertyDefinition pd) {
        // Elasticsearch performs poorly with must_not + exists because it needs to identify documents that do not
        // have a field — and it has no inverted index for "non-existence"
        doc.addNullProperty(pd.name);
    }

    @Override
    protected void indexAggregateValue(ElasticDocument doc, Aggregate.NodeIncludeResult result, String value, PropertyDefinition pd) {
        if (result.isRelativeNode()) {
            doc.addFulltextRelative(result.rootIncludePath, value);
        } else {
            doc.addFulltext(value);
        }
    }

    @Override
    protected void indexNodeName(ElasticDocument doc, String value) {
        doc.addProperty(FieldNames.NODE_NAME, value);
    }

    @Override
    protected boolean indexSimilarityTag(ElasticDocument doc, String value) {
        if (!value.isEmpty()) {
            doc.addSimilarityTag(value);
            return true;
        }
        return false;
    }

    @Override
    protected void indexSimilarityBinaries(ElasticDocument doc, PropertyDefinition pd, Blob blob) throws IOException {
        // without this check, if the vector size is not correct, the entire document will be skipped
        if (pd.getSimilaritySearchDenseVectorSize() == blob.length() / BLOB_LENGTH_DIVISOR) {
            // see https://www.elastic.co/blog/text-similarity-search-with-vectors-in-elasticsearch
            // see https://www.elastic.co/guide/en/elasticsearch/reference/current/dense-vector.html
            doc.addSimilarityField(ElasticIndexUtils.fieldName(pd.name), blob);
        } else {
            if (!LOG_SILENCER.silence(LOG_KEY_SIMILARITY_BINARIES_WRONG_DIMENSION)) {
                LOG.warn("[{}] Ignoring binary property {} for path {}. Expected dimension is {} but got {}",
                        getIndexName(), pd.name, this.path, pd.getSimilaritySearchDenseVectorSize(), blob.length() / BLOB_LENGTH_DIVISOR);
            }
        }
    }

    @Override
    protected void indexSimilarityStrings(ElasticDocument doc, PropertyDefinition pd, String value) {
        // TODO : not implemented
    }

    @Override
    protected boolean augmentCustomFields(String path, ElasticDocument doc, NodeState document) {
        // TODO : not implemented
        return false;
    }

    @Override
    protected boolean indexDynamicBoost(ElasticDocument doc, String parent, String nodeName, String token, double boost) {
        if (!token.isEmpty()) {
            doc.addDynamicBoostField(ElasticIndexUtils.fieldName(nodeName), token, boost);
            return true;
        }
        return false;
    }

    /**
     * ElasticDocument can be updated. If a property gets deleted from the node, we need to add it to the list of properties to remove.
     * This is needed to remove the property from the index. See @{link ElasticBulkProcessorHandler#updateDocument} for more details.
     */
    @Override
    protected void removeProperty(ElasticDocument doc, String propertyName) {
        doc.removeProperty(propertyName);
    }
}