/*
 * RiskScape™ Copyright New Zealand Institute for Earth Science Limited
 * (Earth Sciences New Zealand) is distributed for research purposes only
 * under the terms of AGPLv3.
 *
 * RiskScape™ Copyright 2026 New Zealand Institute for Earth Science
 * Limited (Earth Sciences New Zealand). All rights reserved. Source code
 * available under the AGPLv3.
 * 
 * This program is free software: you can redistribute it and/or modify it under
 *  the terms of the GNU Affero General Public License as published by the Free
 *  Software Foundation, either version 3 of the License, or (at your option) any
 *  later version.
 * 
 * This program is distributed for RESEARCH PURPOSES ONLY, in the hope that it will
 * be useful for research and education initiatives.
 * 
 * If you are not a researcher, or you are a researcher who wishes to use this
 * program on terms other than AGPLv3 (including those who wish to restrict the
 * distribution of any source code created using this program), please contact:
 * https://riskscape.org.nz
 * 
 * This program is distributed WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Affero General Public License for more details.  You should have received a copy
 * of the GNU Affero General Public License along with this program.  If not, see
 * <http://www.gnu.org/licenses/>.
 * 
 * By way of summary only, under the AGPLv3:
 *     • Permissions of this strongest copyleft license are conditioned
 *       on making available complete source code of licensed works and
 *       modifications, which include larger works using a licensed work,
 *       under the same license.
 *     • Copyright and license notices must be preserved.
 *     • Contributors provide an express grant of patent rights.
 *     • When a modified version is used to provide a service over a
 *       network, the complete source code of the modified version must be made
 *       available.
 */


package nz.org.riskscape.engine.query;

import java.util.Set;
import java.util.function.Consumer;

import org.geotools.api.data.FeatureSource;
import org.geotools.factory.CommonFactoryFinder;
import org.geotools.filter.spatial.AbstractPreparedGeometryFilter;
import org.geotools.api.filter.Filter;
import org.geotools.api.filter.FilterFactory;
import org.geotools.api.filter.expression.Add;
import org.geotools.api.filter.expression.Divide;
import org.geotools.api.filter.expression.Expression;
import org.geotools.api.filter.expression.ExpressionVisitor;
import org.geotools.api.filter.expression.Function;
import org.geotools.api.filter.expression.Literal;
import org.geotools.api.filter.expression.Multiply;
import org.geotools.api.filter.expression.NilExpression;
import org.geotools.api.filter.expression.PropertyName;
import org.geotools.api.filter.expression.Subtract;
import org.geotools.api.filter.spatial.BinarySpatialOperator;
import org.geotools.api.filter.spatial.DistanceBufferOperator;

import com.google.common.collect.Sets;
import org.locationtech.jts.geom.Geometry;
import nz.org.riskscape.engine.SRIDSet;


import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.gt.TupleToGeoFilterConverter;
import nz.org.riskscape.engine.relation.FeatureSourceRelation;
import nz.org.riskscape.engine.relation.Relation;
import nz.org.riskscape.engine.relation.SpatialMetadata;


/**
 * Example Code
 *
 * This code used to be used to speed up the old join algorithm by making use of an underlying index on the upstream
 * feature source.  It has been superseded by using an in-memory index, but the approach is still valid in some
 * situations, so keeping this code in the source tree for reference.
 *
 * {@link Joiner} that will take advantage of a potential spatial index on the rhs by attempting to convert the given
 * join condition in a {@link Filter} that works with a {@link FeatureSource}.
 */
public class SpatiallyIndexedJoin implements Joiner {

  private Mode mode;
  private BruteForceJoin bruteForce;
  private final TupleToGeoFilterConverter convertFilter;
  private SRIDSet sridSet;

  /**
   * Designates a particular mode for the join.
   */
  public enum Mode {
    /**
     * No use of spatial indexing/filtering.  This will do a cartesian join and pass each joined tuple to the join
     * condition, so will be impossibly slow for large/complex datasets.
     */
    BRUTE_FORCE,

    /**
     * Convert the {@link Filter} in to one that can be used with a {@link FeatureSource} to reduce the size of the
     * {@link FeatureSourceRelation} before passing results to the {@link JoinCondition}.  This last step is possibly
     * unneccessary...
     */
    CONVERTED_FILTER,

    /**
     * Like {@link #CONVERTED_FILTER}, except the {@link Filter} is transformed in to a basic a interesects b
     * expression, for possibly the simplest/easiest or most efficient use of the spatial index. Results have been
     * inconclusive as to whether this helps or hinders.  The {@link JoinCondition} is the ultimate arbiter of the join.
     */
    INTERSECTS
  }


  public SpatiallyIndexedJoin(SRIDSet sridSet) {
    this(Mode.CONVERTED_FILTER, sridSet);
  }

  /**
   * Create a {@link Joiner} that operates at best at the given mode.
   */
  public SpatiallyIndexedJoin(Mode mode, SRIDSet sridSet) {
    this.mode = mode;
    this.sridSet = sridSet;
    this.bruteForce = new BruteForceJoin();
    this.convertFilter = new TupleToGeoFilterConverter(sridSet);
  }

  public FilterFactory factory = CommonFactoryFinder.getFilterFactory();

  @Override
  public void on(Tuple row, Join join, Consumer<Tuple> consumer) {

    boolean needsBruteForce = (this.mode == Mode.BRUTE_FORCE)
        ||
        !(
            (join.getJoinTo().canCloneWithFilter())
            &&
            (join.getOn() instanceof FilterJoinCondition)
        );

    Relation other = join.getJoinTo();
    SpatialMetadata otherSpatialMetadata = other.getSpatialMetadata().orElse(null);
    if (needsBruteForce || otherSpatialMetadata == null) {
      bruteForce.on(row, join, consumer);
      return;
    }
    FilterJoinCondition filterJoin = (FilterJoinCondition) join.getOn();


    // Convert the tuple based filter to one suitable for inserting into the relation. As part of this we convert
    // any literal geometries to the projection of the target relation if they are different. This is required as
    // geometries for the same location are unlikely to intersect if they are in different projections.
    // TODO see if we can cache some of the join condition specific parts of this
    Filter convertedFilter = convertFilter.convert(filterJoin.getFilter(), join.getAlias(), row,
        sridSet.get(otherSpatialMetadata.getCrs()));
    Filter gtFilter;

    if (this.mode == Mode.CONVERTED_FILTER) {
      gtFilter = convertedFilter;
    } else if (convertedFilter instanceof AbstractPreparedGeometryFilter) {
      AbstractPreparedGeometryFilter castFilter = (AbstractPreparedGeometryFilter) convertedFilter;

      // TODO check castFilter against a set of known safe functions (all except the d ones)
      Expression expression1 = castFilter.getExpression1();
      Expression expression2 = castFilter.getExpression2();

      Expression containsOnlyLiterals = findLiteral(expression1, expression2);
      Expression presumablyHasPropertyName = containsOnlyLiterals == expression1 ? expression2 : expression1;
      Geometry evaluated = containsOnlyLiterals.evaluate(null, Geometry.class);

      Expression propertyName = isIntersectSafe(presumablyHasPropertyName, convertedFilter);
      if (propertyName == null) {
        gtFilter = factory.intersects(propertyName, factory.literal(evaluated));
      } else {
        gtFilter = convertedFilter;
      }
    } else {
      gtFilter = convertedFilter;
    }

    // hopefully we now have a much reduced rhs to trawl through
    Relation filtered = other.cloneWithFilter(gtFilter);
    Join spatialJoin = new Join(filtered, join.getOn(), join.getJoinType());
    //defer to brute force with our reduced rhs.
    bruteForce.on(row, spatialJoin, consumer);
  }

  /**
   * @return `true` if the given expression and filter can be reduced to a simple intersection of the constant to the
   * property on the rhs, on the basis that this is a broader filter than the correct one
   */
  private PropertyName isIntersectSafe(Expression presumablyHasPropertyName, Filter convertedFilter) {
    boolean filterIsOk =
        (convertedFilter instanceof BinarySpatialOperator)
        &&
        !(convertedFilter instanceof DistanceBufferOperator);

    // TODO also check for centroid
    boolean expressionIsOk = presumablyHasPropertyName instanceof PropertyName;

    if (filterIsOk && expressionIsOk) {
      return (PropertyName) presumablyHasPropertyName;
    } else {
      return null;
    }
  }

  private Expression findLiteral(Expression expression1, Expression expression2) {
    if (containsNoPropertyNames(expression1)) {
      return expression1;
    }
    if (containsNoPropertyNames(expression2)) {
      return expression2;
    }

    return null;
  }

  /**
   * @return true if the given expression has no {@link PropertyName} expressions inside it
   */
  private boolean containsNoPropertyNames(Expression expression) {
    Set<PropertyName> found = Sets.newHashSetWithExpectedSize(1);
    expression.accept(new ExpressionVisitor() {

      @Override
      public Object visit(NilExpression expression, Object extraData) {
        return null;
      }

      @Override
      public Object visit(Add expression, Object extraData) {
        expression.getExpression1().accept(this, null);
        expression.getExpression2().accept(this, null);
        return null;
      }

      @Override
      public Object visit(Divide expression, Object extraData) {
        expression.getExpression1().accept(this, null);
        expression.getExpression2().accept(this, null);
        return null;
      }

      @Override
      public Object visit(Function expression, Object extraData) {
        for (Expression parameter : expression.getParameters()) {
          parameter.accept(this, extraData);
        }
        return null;
      }

      @Override
      public Object visit(Literal expression, Object extraData) {
        return null;
      }

      @Override
      public Object visit(Multiply expression, Object extraData) {
        expression.getExpression1().accept(this, null);
        expression.getExpression2().accept(this, null);
        return null;
      }

      @Override
      public Object visit(PropertyName expression, Object extraData) {
        found.add(expression);
        return null;
      }

      @Override
      public Object visit(Subtract expression, Object extraData) {
        expression.getExpression1().accept(this, null);
        expression.getExpression2().accept(this, null);
        return null;
      }

    }, null);

    return found.size() == 0;
  }

}
