/*
 * 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.rya.indexing.accumulo.geo;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

import org.apache.commons.lang3.math.NumberUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.rya.indexing.GeoConstants;
import org.apache.rya.indexing.GeoIndexer;
import org.apache.rya.indexing.IndexingExpr;
import org.apache.rya.indexing.IteratorFactory;
import org.apache.rya.indexing.SearchFunction;
import org.apache.rya.indexing.StatementConstraints;
import org.apache.rya.indexing.external.tupleSet.ExternalTupleSet;
import org.eclipse.rdf4j.common.iteration.CloseableIteration;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Statement;
import org.eclipse.rdf4j.model.Value;
import org.eclipse.rdf4j.query.BindingSet;
import org.eclipse.rdf4j.query.QueryEvaluationException;
import org.eclipse.rdf4j.query.algebra.Var;

import com.google.common.base.Joiner;
import com.google.common.collect.Maps;
import com.vividsolutions.jts.geom.Geometry;
import com.vividsolutions.jts.io.ParseException;
import com.vividsolutions.jts.io.WKTReader;

//Indexing Node for geo expressions to be inserted into execution plan
//to delegate geo portion of query to geo index
public class GeoTupleSet extends ExternalTupleSet {
    private static final String NEAR_DELIM = "::";
    private final Configuration conf;
    private final GeoIndexer geoIndexer;
    private final IndexingExpr filterInfo;


    public GeoTupleSet(final IndexingExpr filterInfo, final GeoIndexer geoIndexer) {
        this.filterInfo = filterInfo;
        this.geoIndexer = geoIndexer;
        conf = geoIndexer.getConf();
    }

    @Override
    public Set<String> getBindingNames() {
        return filterInfo.getBindingNames();
    }

    @Override
	public GeoTupleSet clone() {
        return new GeoTupleSet(filterInfo, geoIndexer);
    }

    @Override
    public double cardinality() {
        return 0.0; // No idea how the estimate cardinality here.
    }


    @Override
    public String getSignature() {
        return "(GeoTuple Projection) " + "variables: " + Joiner.on(", ").join(getBindingNames()).replaceAll("\\s+", " ");
    }


    @Override
    public boolean equals(final Object other) {
        if (other == this) {
            return true;
        }
        if (!(other instanceof GeoTupleSet)) {
            return false;
        }
        final GeoTupleSet arg = (GeoTupleSet) other;
        return filterInfo.equals(arg.filterInfo);
    }

    @Override
    public int hashCode() {
        int result = 17;
        result = 31*result + filterInfo.hashCode();

        return result;
    }


    /**
     * Returns an iterator over the result set of the contained IndexingExpr.
     * <p>
     * Should be thread-safe (concurrent invocation {@link OfflineIterable} this
     * method can be expected with some query evaluators.
     */
    @Override
    public CloseableIteration<BindingSet, QueryEvaluationException> evaluate(final BindingSet bindings)
            throws QueryEvaluationException {

        final IRI funcURI = filterInfo.getFunction();
        final SearchFunction searchFunction = new GeoSearchFunctionFactory(conf, geoIndexer).getSearchFunction(funcURI);

        String queryText;
        Object arg = filterInfo.getArguments()[0];
        if (arg instanceof Value) {
            queryText = ((Value) arg).stringValue();
        } else if (arg instanceof Var) {
            queryText = bindings.getBinding(((Var) arg).getName()).getValue().stringValue();
        } else {
            throw new IllegalArgumentException("Query text was not resolved");
        }

        if(funcURI.equals(GeoConstants.GEO_SF_NEAR)) {
            if (filterInfo.getArguments().length > 3) {
                throw new IllegalArgumentException("Near functions do not support more than four arguments.");
            }

            final List<String> valueList = new ArrayList<>();
            for (final Object val : filterInfo.getArguments()) {
                if (val instanceof Value) {
                    valueList.add(((Value)val).stringValue());
                } else if (val instanceof Var) {
                    valueList.add(bindings.getBinding(((Var) val).getName()).getValue().stringValue());
                } else {
                    throw new IllegalArgumentException("Query text was not resolved");
                }
            }
            queryText = String.join(NEAR_DELIM, valueList);
        } else if (filterInfo.getArguments().length > 1) {
            throw new IllegalArgumentException("Index functions do not support more than two arguments.");
        }

        try {
            final CloseableIteration<BindingSet, QueryEvaluationException> iterrez = IteratorFactory
                    .getIterator(filterInfo.getSpConstraint(), bindings,
                    queryText, searchFunction);
            return iterrez;
        } catch (final Exception e) {
            System.out.println(e.getMessage());
            throw e;
        }
    }

    //returns appropriate search function for a given URI
    //search functions used in GeoMesaGeoIndexer to access index
    public static class GeoSearchFunctionFactory {

        Configuration conf;

        private final Map<IRI, SearchFunction> SEARCH_FUNCTION_MAP = Maps.newHashMap();

        private final GeoIndexer geoIndexer;

        public GeoSearchFunctionFactory(final Configuration conf, final GeoIndexer geoIndexer) {
            this.conf = conf;
            this.geoIndexer = geoIndexer;
        }


        /**
         * Get a {@link GeoSearchFunction} for a given URI.
         *
         * @param searchFunction
         * @return
         */
        public SearchFunction getSearchFunction(final IRI searchFunction) {

            SearchFunction geoFunc = null;

            try {
                geoFunc = getSearchFunctionInternal(searchFunction);
            } catch (final QueryEvaluationException e) {
                e.printStackTrace();
            }

            return geoFunc;
        }

        private SearchFunction getSearchFunctionInternal(final IRI searchFunction) throws QueryEvaluationException {
            final SearchFunction sf = SEARCH_FUNCTION_MAP.get(searchFunction);

            if (sf != null) {
                return sf;
            } else {
                throw new QueryEvaluationException("Unknown Search Function: " + searchFunction.stringValue());
            }
        }

        private final SearchFunction GEO_EQUALS = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryEquals(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_EQUALS";
            };
        };

        private final SearchFunction GEO_DISJOINT = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryDisjoint(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_DISJOINT";
            };
        };

        private final SearchFunction GEO_INTERSECTS = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryIntersects(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_INTERSECTS";
            };
        };

        private final SearchFunction GEO_TOUCHES = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryTouches(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_TOUCHES";
            };
        };

        private final SearchFunction GEO_CONTAINS = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryContains(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_CONTAINS";
            };
        };

        private final SearchFunction GEO_OVERLAPS = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryOverlaps(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_OVERLAPS";
            };
        };

        private final SearchFunction GEO_CROSSES = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryCrosses(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_CROSSES";
            };
        };

        private final SearchFunction GEO_WITHIN = new SearchFunction() {

            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(queryText);
                    final CloseableIteration<Statement, QueryEvaluationException> statements = geoIndexer.queryWithin(
                            geometry, contraints);
                    return statements;
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public String toString() {
                return "GEO_WITHIN";
            };
        };

        private final SearchFunction GEO_NEAR = new SearchFunction() {
            @Override
            public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String queryText,
                    final StatementConstraints contraints) throws QueryEvaluationException {
                try {
                    final String[] args = queryText.split(NEAR_DELIM);
                    Optional<Double> maxDistanceOpt = Optional.empty();
                    Optional<Double> minDistanceOpt = Optional.empty();
                    final String query = args[0];

                    for (int ii = 1; ii < args.length; ii++) {
                        String numArg = args[ii];

                        // remove pre-padding 0's since NumberUtils.isNumber()
                        // will assume its octal if it starts with a 0.
                        while (numArg.startsWith("0")) {
                            numArg = numArg.substring(1);
                        }
                        // was 0
                        if (numArg.equals("")) {
                            // if max hasn't been set, set it to 0.
                            // Otherwise, min is just ignored.
                            if (!maxDistanceOpt.isPresent()) {
                                maxDistanceOpt = Optional.of(0.0);
                            }
                        } else {
                            if (!maxDistanceOpt.isPresent() && NumberUtils.isNumber(numArg)) {
                                // no variable identifier, going by order.
                                maxDistanceOpt = getDistanceOpt(numArg, "maxDistance");
                            } else if (NumberUtils.isNumber(numArg)) {
                                // no variable identifier, going by order.
                                minDistanceOpt = getDistanceOpt(numArg, "minDistance");
                            } else {
                                throw new IllegalArgumentException(numArg + " is not a valid Near function argument.");
                            }
                        }
                    }
                    final WKTReader reader = new WKTReader();
                    final Geometry geometry = reader.read(query);
                    final NearQuery nearQuery = new NearQuery(maxDistanceOpt, minDistanceOpt, geometry);
                    return geoIndexer.queryNear(nearQuery, contraints);
                } catch (final ParseException e) {
                    throw new QueryEvaluationException(e);
                }
            }

            private Optional<Double> getDistanceOpt(final String num, final String name) {
                try {
                    double dist = Double.parseDouble(num);
                    if(dist < 0) {
                        throw new IllegalArgumentException("Value for: " + name + " must be non-negative.");
                    }
                    return Optional.of(Double.parseDouble(num));
                } catch (final NumberFormatException nfe) {
                    throw new IllegalArgumentException("Value for: " + name + " must be a number.");
                }
            }

            @Override
            public String toString() {
                return "GEO_NEAR";
            }
        };

        /**
         *
         */
        public class NearQuery {
            private final Optional<Double> maxDistanceOpt;
            private final Optional<Double> minDistanceOpt;
            private final Geometry geo;

            /**
             *
             * @param maxDistance
             * @param minDistance
             * @param geo
             */
            public NearQuery(final Optional<Double> maxDistance, final Optional<Double> minDistance,
                    final Geometry geo) {
                maxDistanceOpt = maxDistance;
                minDistanceOpt = minDistance;
                this.geo = geo;
            }

            public Optional<Double> getMaxDistance() {
                return maxDistanceOpt;
            }

            public Optional<Double> getMinDistance() {
                return minDistanceOpt;
            }

            public Geometry getGeometry() {
                return geo;
            }
        }

        {
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_EQUALS, GEO_EQUALS);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_DISJOINT, GEO_DISJOINT);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_INTERSECTS, GEO_INTERSECTS);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_TOUCHES, GEO_TOUCHES);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_CONTAINS, GEO_CONTAINS);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_OVERLAPS, GEO_OVERLAPS);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_CROSSES, GEO_CROSSES);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_WITHIN, GEO_WITHIN);
            SEARCH_FUNCTION_MAP.put(GeoConstants.GEO_SF_NEAR, GEO_NEAR);
        }
    }
}