/*
 * 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;

import ch.qos.logback.classic.Level;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.api.Tree;
import org.apache.jackrabbit.oak.commons.junit.LogCustomizer;
import org.apache.jackrabbit.oak.commons.junit.TemporarySystemProperty;
import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticBulkProcessorHandler;
import org.apache.jackrabbit.oak.plugins.index.elastic.util.ElasticIndexDefinitionBuilder;
import org.apache.jackrabbit.oak.plugins.index.elastic.util.ElasticIndexUtils;
import org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants;
import org.apache.jackrabbit.oak.plugins.index.search.util.IndexDefinitionBuilder;
import org.apache.jackrabbit.oak.plugins.index.search.util.IndexDefinitionBuilder.PropertyRule;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import java.util.List;

import static org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants.PROPDEF_PROP_NODE_NAME;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.MatcherAssert.assertThat;

public class ElasticPropertyIndexTest extends ElasticAbstractQueryTest {
    @Rule
    public TemporarySystemProperty temporarySystemProperty = new TemporarySystemProperty();

    // Tests are hardcoded for these values
    private final static int BULK_ACTIONS_TEST = 250;
    private final static int BULK_SIZE_BYTES_TEST = 1024 * 1024;

    @Before
    public void before() throws Exception {
        // Use a low value for the tests
        System.setProperty(ElasticBulkProcessorHandler.BULK_ACTIONS_PROP, Integer.toString(BULK_ACTIONS_TEST));
        System.setProperty(ElasticBulkProcessorHandler.BULK_SIZE_BYTES_PROP, Integer.toString(BULK_SIZE_BYTES_TEST));
        super.before();
    }

    @Test
    public void bulkProcessorEventsFlushLimit() throws Exception {
        setIndex("test1", createIndex("propa"));

        Tree test = root.getTree("/").addChild("test");
        for (int i = 1; i < 249; i++) {
            test.addChild("a" + i).setProperty("propa", "foo" + i);
        }
        root.commit();

        // 250 is the default flush limit for bulk processor, and we added just less than 250 nodes
        // So once the index writer is closed, bulk Processor would be closed and all the 248 entries should be flushed.
        // Make sure that the last entry is indexed correctly.
        String propaQuery = "select [jcr:path] from [nt:base] where [propa] = 'foo248'";
        assertEventually(() -> {
            assertThat(explain(propaQuery), containsString("elasticsearch:test1"));
            assertQuery(propaQuery, List.of("/test/a248"));
        });

        // Now we test for 250 < nodes < 500

        for (int i = 250; i < 300; i++) {
            test.addChild("a" + i).setProperty("propa", "foo" + i);
        }
        root.commit();
        String propaQuery2 = "select [jcr:path] from [nt:base] where [propa] = 'foo299'";
        assertEventually(() -> {
            assertThat(explain(propaQuery2), containsString("elasticsearch:test1"));
            assertQuery(propaQuery2, List.of("/test/a299"));
        });
    }

    @Test
    public void bulkProcessorSizeFlushLimit() throws Exception {
        LogCustomizer customLogger = LogCustomizer
                .forLogger("org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticBulkProcessorHandler")
                .enable(Level.DEBUG).create();
        try {
            customLogger.starting();
        /*
        Below are the conditions to flush data from bulk processor.
        1. Based on events reaching 250. (BULK_ACTIONS_TEST)
        2. Based on size of estimated bulk request size.
        3. When index writer is closed.
        To trigger flush on bulk request size, we will load large documents so that
         instead of event, flush is triggered because of bulk request size.
         */
            setIndex("test1", createIndex("propa", "propb"));
            int docSize = 1024 * 16;
            // +1 at end leads to bulk size breach, leading to two bulkIds.
            int docCountBreachingBulkSize = (BULK_SIZE_BYTES_TEST / docSize) + 1;
            // 250 is the default flush limit for bulk processor
            Assert.assertTrue(docCountBreachingBulkSize < 250);
            String random = RandomStringUtils.insecure().next(docSize, true, true);

            Tree test = root.getTree("/").addChild("test");
            for (int i = 1; i <= docCountBreachingBulkSize; i++) {
                test.addChild("a" + i).setProperty("propa", "foo" + i);
                test.addChild("a" + i).setProperty("propb", random + i);
            }
            root.commit();

            String propaQuery = "select [jcr:path] from [nt:base] where [propa] = 'foo" + docCountBreachingBulkSize + "'";
            assertEventually(() -> {
                assertThat(explain(propaQuery), containsString("elasticsearch:test1"));
                assertQuery(propaQuery, List.of("/test/a" + docCountBreachingBulkSize));
            });

            Assert.assertEquals(1, customLogger.getLogs().stream().filter(n -> n.contains("Bulk with id 2 processed in")).count());
            Assert.assertEquals(0, customLogger.getLogs().stream().filter(n -> n.contains("Bulk with id 3 processed in")).count());
        } finally {
            customLogger.finished();
        }
    }

    @Test
    public void indexSelection() throws Exception {
        setIndex("test1", createIndex("propa", "propb"));
        setIndex("test2", createIndex("propc"));

        Tree test = root.getTree("/").addChild("test");
        test.addChild("a").setProperty("propa", "foo");
        test.addChild("b").setProperty("propa", "foo");
        test.addChild("c").setProperty("propa", "foo2");
        test.addChild("d").setProperty("propc", "foo");
        test.addChild("e").setProperty("propd", "foo");
        root.commit();

        String propaQuery = "select [jcr:path] from [nt:base] where [propa] = 'foo'";

        assertEventually(() -> {
            assertThat(explain(propaQuery), containsString("elasticsearch:test1"));
            assertThat(explain("select [jcr:path] from [nt:base] where [propc] = 'foo'"), containsString("elasticsearch:test2"));

            assertQuery(propaQuery, List.of("/test/a", "/test/b"));
            assertQuery("select [jcr:path] from [nt:base] where [propa] = 'foo2'", List.of("/test/c"));
            assertQuery("select [jcr:path] from [nt:base] where [propc] = 'foo'", List.of("/test/d"));
        });
    }

    //OAK-3825
    @Test
    public void nodeNameViaPropDefinition() throws Exception {
        //make index
        IndexDefinitionBuilder builder = createIndex();
        builder.includedPaths("/test")
                .indexRule("nt:base")
                .property("nodeName", PROPDEF_PROP_NODE_NAME);
        setIndex("test1", builder);
        root.commit();

        //add content
        Tree test = root.getTree("/").addChild("test");
        test.addChild("foo");
        test.addChild("camelCase");
        test.addChild("sc").addChild("bar");
        root.commit();

        String queryPrefix = "select [jcr:path] from [nt:base] where ISDESCENDANTNODE('/test') AND ";
        //test
        String propabQuery = queryPrefix + "LOCALNAME() = 'foo'";

        assertEventually(() -> {
            String explanation = explain(propabQuery);
            assertThat(explanation, containsString("/oak:index/test1"));
            assertThat(explanation, containsString("{\"term\":{\":nodeName\":{\"value\":\"foo\""));
            assertQuery(propabQuery, List.of("/test/foo"));

            assertQuery(queryPrefix + "LOCALNAME() = 'bar'", List.of("/test/sc/bar"));
            assertQuery(queryPrefix + "LOCALNAME() LIKE 'foo'", List.of("/test/foo"));
            assertQuery(queryPrefix + "LOCALNAME() LIKE 'camel%'", List.of("/test/camelCase"));

            assertQuery(queryPrefix + "NAME() = 'bar'", List.of("/test/sc/bar"));
            assertQuery(queryPrefix + "NAME() LIKE 'foo'", List.of("/test/foo"));
            assertQuery(queryPrefix + "NAME() LIKE 'camel%'", List.of("/test/camelCase"));
        });
    }

    // OAK-11530
    @Test
    public void propertyWithDotPrefix() throws Exception {
        IndexDefinitionBuilder builder = createIndex();
        builder.includedPaths("/test")
                .indexRule("nt:base")
                .property("foo", "foo").propertyIndex()
                .property("test", "./test").propertyIndex();
        setIndex("test1", builder);
        root.commit();

        //add content
        root.getTree("/").addChild("test")
            .setProperty("test", "1");
        root.commit();

        String query = "select [jcr:path] from [nt:base] " +
                "where test = '1'";

        assertEventually(() -> {
            String explanation = explain(query);
            assertThat(explanation, containsString("no-index"));
        });

        String queryFoo = "select [jcr:path] from [nt:base] " +
                "where foo = '1'";
        assertEventually(() -> {
            String explanation = explain(queryFoo);
            assertThat(explanation, containsString("/oak:index/test1"));
            assertThat(explanation, containsString("{\"term\":{\"foo\":{\"value\":\"1\""));
            assertQuery(query, List.of());
        });
    }

    @Test
    public void propertyWithDot() throws Exception {
        IndexDefinitionBuilder builder = createIndex();
        builder.includedPaths("/test")
                .indexRule("nt:base")
                .property("firstName", "first.name").propertyIndex()
                .property("lowerFirstName", "first.name");
        PropertyRule lowerFirstName = builder.indexRule("nt:base").property("lowerFirstName");
        lowerFirstName.getBuilderTree().setProperty(
                FulltextIndexConstants.PROP_FUNCTION, "lower([first.name])");
        setIndex("test1", builder);
        root.commit();

        //add content
        root.getTree("/").addChild("test").setProperty("first.name", "Antonio");
        root.commit();

        String query = "select [jcr:path] from [nt:base] " +
                "where [first.name] = 'Antonio'";

        assertEventually(() -> {
            String explanation = explain(query);
            assertThat(explanation, containsString("/oak:index/test1"));
            assertThat(explanation, containsString(
                    "{\"term\":{\"" +
                            ElasticIndexUtils.fieldName("first.name") +
                            "\":{\"value\":\"Antonio\""));
            assertQuery(query, List.of("/test"));
        });

        String lowerQuery = "select [jcr:path] from [nt:base] " +
                "where lower([first.name]) = 'antonio'";

        assertEventually(() -> {
            String explanation = explain(lowerQuery);
            assertThat(explanation, containsString("/oak:index/test1"));
            assertThat(explanation, containsString(
                    "{\"term\":{\"" +
                            ElasticIndexUtils.fieldName("function*lower*@first.name") +
                            "\":{\"value\":\"antonio\""));
            assertQuery(lowerQuery, List.of("/test"));
        });
    }

    @Test
    public void emptyIndex() throws Exception {
        setIndex("test1", createIndex("propa", "propb"));
        root.commit();

        Tree test = root.getTree("/").addChild("test");
        test.addChild("a");
        test.addChild("b");
        root.commit();

        assertEventually(() -> assertThat(explain("select [jcr:path] from [nt:base] where [propa] = 'foo'"),
                containsString("elasticsearch:test1")));
    }

    @Test
    public void inOperandStringValues() throws Exception {
        String query = "select [jcr:path] from [nt:base] where [propa] in(\"a\", \"e\", \"i\")";
        createIndexOfType("String");

        Tree test = root.getTree("/").addChild("test");
        for (char ch = 'a'; ch <= 'z'; ch++) {
            test.addChild("node-" + ch).setProperty("propa", Character.toString(ch));
        }
        root.commit();
        assertEventually(() -> {
            assertThat(explain(query), containsString("{\"terms\":{\"propa\":[\"a\",\"e\",\"i\"]}}"));
            assertQuery(query, SQL2, List.of("/test/node-a", "/test/node-e", "/test/node-i"));
        });
    }

    @Test
    public void inOperandLongValues() throws Exception {
        String query = "select [jcr:path] from [nt:base] where [propa] in(2, 3, 5, 7)";
        createIndexOfType("Long");

        Tree test = root.getTree("/").addChild("test");
        for (int i = 0; i < 10; i++) {
            test.addChild("node-" + i).setProperty("propa", Integer.toString(i));
        }
        root.commit();

        assertEventually(() -> {
            assertThat(explain(query), containsString("{\"terms\":{\"propa\":[2,3,5,7]}}"));
            assertQuery(query, SQL2, List.of("/test/node-2", "/test/node-3", "/test/node-5", "/test/node-7"));
        });
    }

    @Test
    public void inOperandDoubleValues() throws Exception {
        String query = "select [jcr:path] from [nt:base] where [propa] in(2.0, 3.0, 5.0, 7.0)";

        createIndexOfType("Double");

        Tree test = root.getTree("/").addChild("test");
        for (int i = 0; i < 10; i++) {
            test.addChild("node-" + i).setProperty("propa", (double) i);
        }
        root.commit();

        assertEventually(() -> {
            assertThat(explain(query), containsString("{\"terms\":{\"propa\":[2.0,3.0,5.0,7.0]}}"));
            assertQuery(query, SQL2, List.of("/test/node-2", "/test/node-3", "/test/node-5", "/test/node-7"));
        });
    }

    private void createIndexOfType(String type) throws CommitFailedException {
        IndexDefinitionBuilder builder = new ElasticIndexDefinitionBuilder();
        builder.noAsync();
        builder.indexRule("nt:base")
                .property("propa")
                .propertyIndex()
                .type(type);

        setIndex("test", builder);
        root.commit();
    }
}
