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

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;

import lombok.AccessLevel;
import lombok.RequiredArgsConstructor;
import nz.org.riskscape.engine.SRIDSet;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.data.BaseBookmarkResolver;
import nz.org.riskscape.engine.projection.Projection;
import nz.org.riskscape.engine.projection.Projector;
import nz.org.riskscape.engine.projection.TypeProjection;
import nz.org.riskscape.engine.query.TupleUtils;
import nz.org.riskscape.engine.rl.RealizationContext;
import nz.org.riskscape.engine.types.CoercionException;
import nz.org.riskscape.engine.types.Nullable;
import nz.org.riskscape.engine.types.Referenced;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Struct.StructMember;
import nz.org.riskscape.engine.types.Text;
import nz.org.riskscape.engine.types.Type;
import nz.org.riskscape.engine.types.Types;
import nz.org.riskscape.problem.ResultOrProblems;
import nz.org.riskscape.rl.ExpressionParser;
import nz.org.riskscape.rl.TokenTypes;
import nz.org.riskscape.rl.ast.Expression;

/**
 * Utility for inferring the best type for a {@link Relation} by inspecting the relation's content
 * and produce a {@link Projection} to return the best type.
 *
 * This inferrer can make text columns numeric and add a geometry column if lat/lon numeric values
 * are found.
 *
 * The inferrer needs to read the relation to do it's work. For this reason it should only be used
 * if the source data is on a local file.
 * {@link BaseBookmarkResolver#getBookmarkedPath(nz.org.riskscape.engine.data.BookmarkParameters)}
 * Should be used to ensure any remote data is localized.
 */
@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
public class RelationBestTypeInferrer {

  // Likely names for lat/long for inferring a WGS84 geometry (point) attribute . The first one can be anywhere in the
  // column name, whereas the others must match exactly.   This seems a little ... mad... but we've seen examples of
  // this in real data.
  public static final List<String> ALLOWED_LAT_NAMES = List.of("latitude", "lat");
  public static final List<String> ALLOWED_LON_NAMES = List.of("longitude", "lon", "long");
  public static final String GEOMETRY_ATTRIBUTE = "geometry";

  /**
   * Return a projected version of the given relation if it can be improved by best type detection,
   * or empty if no improvements can be made.
   */
  public static Optional<ResultOrProblems<Relation>> projectToBestTypes(Relation source, RealizationContext context) {
    RelationBestTypeInferrer inferrer = new RelationBestTypeInferrer(context);
    return inferrer.infer(source);
  }

  private final RealizationContext realizationContext;

  Optional<ResultOrProblems<Relation>> infer(Relation source) {
    AtomicBoolean hasInferred = new AtomicBoolean();
    ResultOrProblems<Relation> result = ResultOrProblems.of(source);

    // XXX add some kind of mode to select which projections get applied. could be useful to only do
    // geometry (without relation scan) for NetCDF/HDF5
    // First we scan the relation to see if text columns can be made numeric
    Optional<Projection> columnProjection = projectBestColumnTypes(source);
    if (columnProjection.isPresent()) {
      hasInferred.set(true);
      result = result.flatMap(r -> r.project(columnProjection.get()));
    }
    result = result.flatMap(rel -> {
      // now there may be some numeric columns that can produce geometry (no relation scan needed here)
      Optional<Projection> geomProjection = projectLatLongToGeometry(rel);
      if (geomProjection.isPresent()) {
        hasInferred.set(true);
        return rel.project(geomProjection.get());
      }
      return ResultOrProblems.of(rel);
    });
    if (hasInferred.get() && !result.hasErrors()) {
      return Optional.of(result);
    }

    return Optional.empty();
  }

  /**
   * Returns a {@link Projection} that give columns a floating type if all of the column values
   * allow.
   *
   * Note: this method iterates over all of the tuples in the relation to determine the best types.
   */
  private Optional<Projection> projectBestColumnTypes(Relation source) {
    // we only improve columns that are text so extract them first.
    List<StructMember> textMembers = source.getProducedType().getMembers().stream()
        .filter(member -> member.getType().findAllowNull(Text.class).isPresent())
        .toList();
    if (textMembers.isEmpty()) {
      // nothing to do here
      return Optional.empty();
    }

    Struct bestType = source.getProducedType();
    Type[] bestTypes = new Type[bestType.size()];
    boolean[] nullableBestTypes = new boolean[bestType.size()];
    TupleIterator it = source.iterator();
    while (it.hasNext()) {
      Tuple t = it.next();
      for (StructMember textMember: textMembers) {
        String value = t.fetch(textMember);
        if (value == null || "".equals(value)) {
          nullableBestTypes[textMember.getIndex()] = true;
        } else if (bestTypes[textMember.getIndex()] != Types.TEXT) {
          // we only work out the best type if it hasn't already been set to test
          // otherwise we need to test the values to make sure they are all numeric
          try {
            Types.FLOATING.coerce(value);
            bestTypes[textMember.getIndex()] = Types.FLOATING;
          } catch (CoercionException nfe) {
            bestTypes[textMember.getIndex()] = Types.TEXT;
          }
        }
      }
    }

    Map<String, Expression> attrMap = new HashMap<>();
    for (StructMember textMember: textMembers) {
      if (bestTypes[textMember.getIndex()] == Types.FLOATING) {
        String expr = String.format("float(%s)", TokenTypes.quoteIdent(textMember.getKey()));
        if (nullableBestTypes[textMember.getIndex()]) {
          expr = String.format(
              "if(is_null(%1$s) || %1$s = '', null_of('floating'), () -> %s)",
              TokenTypes.quoteIdent(textMember.getKey()), expr
          );
        }
        attrMap.put(textMember.getKey(), ExpressionParser.parseString(expr));
        bestType = bestType.replace(
            textMember.getKey(),
            Nullable.ifTrue(nullableBestTypes[textMember.getIndex()], Types.FLOATING)
        );
      }
    }
    if (!attrMap.isEmpty()) {
      return Optional.of(new TypeProjection(bestType, attrMap, realizationContext));
    }
    return Optional.empty();
  }

  /**
   * Returns a {@link Projection} to add a geometry column if suitable lat/lon columns are found in
   * the source relation and is a numeric type.
   */
  private Optional<Projection> projectLatLongToGeometry(Relation source) {
    if (source.getProducedType().hasMember(GEOMETRY_ATTRIBUTE)
        || TupleUtils.hasGeometryMember(source.getProducedType())) {
      // we don't want to overwrite a column that already exists
      // nor do we to make a geometry when one already exists
      return Optional.empty();
    }

    StructMember latMember = findLatLongMember(source.getProducedType(), ALLOWED_LAT_NAMES);
    StructMember lonMember = findLatLongMember(source.getProducedType(), ALLOWED_LON_NAMES);

    if (latMember == null || lonMember == null) {
      return Optional.empty();
    }
    boolean nullable = Nullable.is(latMember.getType()) || Nullable.is(lonMember.getType());

    Map<String, Expression> attrMap = Map.of(
        GEOMETRY_ATTRIBUTE,
        // we use geom_from_wkt as that will set the SRID correctly, but we need to handle nullable lat/lon ourselves.
        ExpressionParser.parseString(String.format(
            "if(is_null(%1$s) || is_null(%2$s), null_of('geometry'), "
            + "() -> geom_from_wkt('POINT(' + str(%1$s) + ' ' + str(%2$s) + ')', 'EPSG:4326'))",
            TokenTypes.quoteIdent(latMember.getKey()), TokenTypes.quoteIdent(lonMember.getKey())
        ))
    );
    Struct projectedType = source.getProducedType();
    projectedType = projectedType.add(
        GEOMETRY_ATTRIBUTE,
        Referenced.ofNullable(Nullable.ifTrue(nullable, Types.POINT), SRIDSet.EPSG4326_LATLON)
    );

    SpatialMetadata spatialMetadata = new SpatialMetadata(
        SRIDSet.EPSG4326_LATLON, projectedType.getEntry(GEOMETRY_ATTRIBUTE)
    );
    return Optional.of(new TypeProjection(projectedType, attrMap, realizationContext) {
       @Override
      public ResultOrProblems<Projector> getProjectionFunction(Struct sourceType) {
        return buildMemberList(sourceType).map(mapping -> {
          return new TypeProjection.TypeProjector(sourceType, mapping.targetType, mapping.memberList) {
            @Override
            public Optional<Function<SpatialMetadata, SpatialMetadata>> getSpatialMetadataMapper() {
              return Optional.of((ignored) -> {
                return spatialMetadata;
              });
            }
          };
        });
      }
    });
  }

  /**
   * Finds a lat/long member from source. That returned member will have a key from allowedNames and
   * a numeric type.
   *
   * If there are no matches on the allowed names then a final pass will look for any attribute key that
   * contains the best name (which is the first entry from allowedNames).
   *
   * @param source struct that may contain lat/long atttributes
   * @param allowedNames list of allowed names
   */
  private StructMember findLatLongMember(Struct source, List<String> allowedNames) {
    // lat/long attributes must be numeric
    List<StructMember> numericMembers = source.getMembers().stream()
        .filter(member -> member.getType().isNumeric())
        .toList();

    for (String allowedName : allowedNames) {
      StructMember found = numericMembers.stream()
          .filter(member -> allowedName.equalsIgnoreCase(member.getKey()))
          .findFirst()
          .orElse(null);
      if (found != null) {
        return found;
      }
    }
    if (! allowedNames.isEmpty()) {
      // if no attribute has been found yet we give it a final try by looking for any attribute that contains
      // the best name (case insensitive). In this way we will find an attribute with a name like
      // `exposure_latitude`
      String bestName = allowedNames.get(0);
      StructMember found = numericMembers.stream()
          .filter(member -> member.getKey().toLowerCase().contains(bestName.toLowerCase()))
          .findFirst()
          .orElse(null);
      if (found != null) {
        return found;
      }
    }
    return null;
  }

}
