/*
 * 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.List;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import com.google.common.collect.Lists;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.data.InputDataProblems;
import nz.org.riskscape.engine.projection.Projection;
import nz.org.riskscape.engine.projection.Projector;
import nz.org.riskscape.engine.restriction.Restriction;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.problem.Problem;
import nz.org.riskscape.problem.ProblemSink;
import nz.org.riskscape.problem.ResultOrProblems;

@RequiredArgsConstructor
public abstract class BaseRelation implements Relation {

  /**
   * Contains a mutable set of the final fields that live on {@link BaseRelation} to support a nice functional cloning
   * API between this class and its subclasses.
   */
  protected final class Fields {
    /**
     * Construct a new set of Fields, copying them from the enclosing instance.
     */
    public Fields() {
      this.type = BaseRelation.this.type;
      this.transformers = Lists.newArrayList(BaseRelation.this.transformers);
      this.skipOnInvalid = BaseRelation.this.skipOnInvalid;
      this.spatialMetadata = BaseRelation.this.spatialMetadata;
      this.limit = BaseRelation.this.limit;
      this.offset = BaseRelation.this.offset;
    }

    public Fields tap(Consumer<Fields> consumer) {
      consumer.accept(this);
      return this;
    }

    Struct type;
    List<Transformer> transformers;
    ProblemSink skipOnInvalid;
    SpatialMetadata spatialMetadata;
    long limit;
    long offset;

    Fields append(Transformer transformer) {
      this.transformers.add(transformer);
      return this;
    }
  }

  /**
   * Wraps either a {@link Projection} or {@link Restriction} that should be applied to a {@link Relation}.  Used to
   * aid transformation of tuples that come from this relation, according to the various projections and restrictions
   * that have been applied to it
   */
  protected final class Transformer {
    Projection projection;
    Projector projectionFunction;

    Restriction restriction;
    Struct transformedType;

    Transformer(Struct previousType, Projection projection, Projector projectionFunction) {
      this.projection = projection;
      this.projectionFunction = projectionFunction;
      this.transformedType = projectionFunction.getProjectedType();
    }

    Transformer(Struct previousType, Restriction restriction) {
      this.restriction = restriction;
      this.transformedType = previousType;
    }

    Struct getType() {
      return transformedType;
    }

    Tuple apply(Tuple tuple) {
      if (projection != null) {
        return projectionFunction.apply(tuple);
      } else {
        return restriction.getPredicate().test(tuple) ? tuple : null;
      }
    }
  }

  /**
   * Constructor that assigns fields based on the state of a different {@link BaseRelation}, meant for use in
   * cloning operations
   */
  public BaseRelation(Fields fields) {
    this.type = fields.type;
    this.transformers = fields.transformers.toArray(new Transformer[0]);
    this.skipOnInvalid = fields.skipOnInvalid;
    this.spatialMetadata = fields.spatialMetadata;
    this.limit = fields.limit;
    this.offset = fields.offset;
  }

  public BaseRelation(Struct type) {
    this(type, null);
  }

  public BaseRelation(Struct type, ProblemSink skipOnInvalid) {
    this.type = type;
    this.transformers = new Transformer[0];
    this.skipOnInvalid = skipOnInvalid;
    this.spatialMetadata = null;
    this.limit = Long.MAX_VALUE;
    this.offset = 0;
  }

  public BaseRelation(Struct type, ProblemSink skipOnInvalid, SpatialMetadata spatialMetadata) {
    this.type = type;
    this.transformers = new Transformer[0];
    this.skipOnInvalid = skipOnInvalid;
    this.spatialMetadata = spatialMetadata;
    this.limit = Long.MAX_VALUE;
    this.offset = 0;
  }

  private final Struct type;

  protected final Transformer[] transformers;

  public final ProblemSink skipOnInvalid;

  private final SpatialMetadata spatialMetadata;

  @Getter
  private final long limit;

  @Getter
  private final long offset;

  /**
   * @return a {@link TupleIterator} that will apply any given transformations to each {@link Tuple} that emerges from
   * the source iterator as returned from {@link #rawIterator()}.  If no transformations are assigned, the value of
   * {@link #rawIterator()} is returned.
   */
  @Override
  public final TupleIterator iterator() {
    TupleIterator rawIterator = rawIterator();

    if (transformers.length == 0 && skipOnInvalid == null) {
      if (offset != 0 || limit != Long.MAX_VALUE) {
        return TupleIterators.peeker(new TransformationSupplier(rawIterator), Optional.of(() -> rawIterator.close()));
      } else {
        return rawIterator;
      }
    } else {
      return TupleIterators.peeker(new TransformationSupplier(rawIterator), Optional.of(() -> rawIterator.close()));
   }
  }



  @RequiredArgsConstructor
  private class TransformationSupplier implements Supplier<Tuple> {

    final TupleIterator rawIterator;
    long skipCounter = offset;
    long stopAtZero = limit;

    @Override
    public Tuple get() {

      if (stopAtZero == 0) {
        return null;
      }

      iteration: while (rawIterator.hasNext()) {

        Tuple raw;
        try {
          raw = rawIterator.next();
        } catch (InvalidTupleException ex) {
          skipOrThrow(ex);
          continue;
        }

        for (int i = 0; i < transformers.length; i++) {
          Transformer t = transformers[i];

          try {
            raw = t.apply(raw);
          } catch (InvalidTupleException ex) {
            skipOrThrow(ex);
            continue iteration;
          }

          if (raw == null) {
            break;
          }
        }

        if (raw != null) {

          if (skipCounter != 0) {
            skipCounter--;
            continue;
          }

          stopAtZero--;
          return raw;
        }
      }

      return null;
    }
  }

  /**
   * Skip (and log) the invalid tuple, if specified by the bookmark. Otherwise throw the error
   */
  protected void skipOrThrow(InvalidTupleException ex) {
    if (skipOnInvalid != null) {
      skipOnInvalid.log(InputDataProblems.get().invalidTupleSkipped().withChildren(ex.getReason()));
    } else {
      throw ex;
    }
  }

  /**
   * @return a {@link TupleIterator} that yields untransformed {@link Tuple}s from the underlying data source.
   */
  protected abstract TupleIterator rawIterator();

  /**
   * @return a clone of yourself using the given, potentially modified, set of fields.
   */
  protected abstract BaseRelation clone(BaseRelation.Fields fields);

  @Override
  public final ResultOrProblems<Relation> project(Projection projection) {

    ResultOrProblems<Projector> projector = projection.getProjectionFunction(getType());

    if (projector.hasErrors()) {
      return ResultOrProblems.failed(projector.getProblems());
    }

    BaseRelation clonedNatively = cloneWithProjectionIfSupported(projection);

    if (clonedNatively != null) {
      return ResultOrProblems.of(clonedNatively);
    } else {
      return ResultOrProblems.of(cloneWithTransformer(new Transformer(getType(), projection, projector.get())),
          projector.getProblems());
    }
  }

  protected BaseRelation cloneWithTransformer(Transformer transformer) {
    BaseRelation.Fields fields = new BaseRelation.Fields();

    if (transformer.projectionFunction != null
        && transformer.projectionFunction.getSpatialMetadataMapper().isPresent()) {

      Function<SpatialMetadata, SpatialMetadata> smMapper =
          transformer.projectionFunction.getSpatialMetadataMapper().get();

      fields.spatialMetadata = smMapper.apply(fields.spatialMetadata);
    }

    fields.append(transformer);
    return clone(fields);
  }

  /**
   * Subclasses should return a cloned version of self with the given projection applied, assuming it supports the
   * application of the given projection 'natively', that is, using something like a select clause before the data
   * is marshalled in to the JVM.
   * @return a cloned version of self with the projection applied, or null of it isn't possible
   */
  protected BaseRelation cloneWithProjectionIfSupported(Projection projection) {
    return null;
  }

  /**
   * Subclasses should return a cloned version of self with the given restriction applied, assuming it supports the
   * application of the given restriction 'natively', that is, using something like a where clause before the data
   * is marshalled in to the JVM.
   * @return a cloned version of self with the restriction applied, or null of it isn't possible
   */
  protected BaseRelation cloneWithRestrictionIfSupported(Restriction restriction) {
    return null;
  }

  @Override
  public final ResultOrProblems<Relation> restrict(Restriction restriction) {
    List<Problem> problems = restriction.validate(getType());

    if (!problems.isEmpty()) {
      return ResultOrProblems.failed(problems);
    }

    BaseRelation clonedNatively = cloneWithRestrictionIfSupported(restriction);

    if (clonedNatively != null) {
      return ResultOrProblems.of(clonedNatively);
    } else {
      return ResultOrProblems.of(cloneWithTransformer(new Transformer(getType(), restriction)));
    }
  }

  @Override
  public final Optional<SpatialMetadata> getSpatialMetadata() {
    return Optional.ofNullable(spatialMetadata);
  }

  /**
   * @return the original type this {@link Relation} would have returned if no transformations had been applied to it.
   */
  public Struct getRawType() {
    return this.type;
  }

  @Override
  public Struct getProducedType() {
    if (transformers.length == 0) {
      return type;
    } else {
      return transformers[transformers.length -1].getType();
    }
  }

  @Override
  public Relation limitAndOffset(long newLimit, long newOffset) {
    Fields fields = new Fields();
    fields.limit = newLimit;
    fields.offset = newOffset;
    return clone(fields);
  }

  @Override
  public Relation skipInvalid(ProblemSink sink) {
    return clone(new Fields().tap(f -> f.skipOnInvalid = sink));
  }

  @Override
  public String toString() {
    return String.format("%s", getClass().getSimpleName());
  }

  @Override
  public boolean hasSkipOnInvalid() {
    return skipOnInvalid != null;
  }
}
