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

import java.util.List;

import lombok.extern.slf4j.Slf4j;
import nz.org.riskscape.engine.pipeline.Collector;
import nz.org.riskscape.engine.relation.TupleIterator;
import nz.org.riskscape.engine.sched.AccumulatorSink;
import nz.org.riskscape.engine.util.PercentGauge;

@Slf4j
public class AccumulatorProcessorTask<T> extends WorkerTask {

  /*
   * I had a look at seeing if we could replace this with a TupleInputTask by adding an adapter that shifts the
   * accumulator specific logic in here in to the adapter.  This is made harder by the AccumulatorSink's ability
   * to 'feed' multiple branches of execution, i.e. when multiple group steps hang off the same select.  It's probably
   * still possible, but I don't want to do anything without some benchmarks.
   */

  private TupleIterator iterator;

  private Collector<T> collector;

  private final PercentGauge completionGuage;

  private long totalTuples = 0;

  public AccumulatorProcessorTask(TaskSpec spec) {
    super(spec);
    this.collector = spec.getFirstStep().getRealized(Collector.class).get();
    // we need to setup the gauge now, event though we don't know the total tuples until later
    this.completionGuage = spec.newMetric("complete", "complete",
        () -> new PercentGauge(() -> out.getCount(), () -> totalTuples));
  }

  @Override
  public boolean isReadyToRun() {
    return !spec.hasOutstandingDependencies() && isOutputReady();
  }

  @Override
  public ReturnState run() {
    if (iterator == null) {
      List<AccumulatorSink> sinks = spec.getProcessingResultsFrom(WorkerTask.class, AccumulatorSink.class, 1);
      if (sinks.size() > 1) {
        log.info("About to reduce {} accumulators...", sinks.size());
      }
      // XXX We could tweak the dependency code and allow this task to run as soon as there's more than one accumulator
      // in our list - that way we can process the results as they come in
      T accumulated = sinks.stream()
          .map(sink -> collector.getAccumulatorClass().cast(sink.getAndClearAccumulator(collector)))
          .reduce((a, b) -> collector.combine(a, b))
          .orElse(null);

      if (sinks.size() > 1) {
        log.info("   ...done");
      }

      totalTuples = collector.size(accumulated).orElse(0L);
      iterator = collector.process(accumulated);
    }

    while (iterator.hasNext()) {
      if (pageWriter.isFull()) {
        return ReturnState.OUTPUT_FULL;
      }
      out.mark();
      pageWriter.add(iterator.next());
      // we only process the underlying aggregated data once, so we can free the tuples as we go
      iterator.remove();
    }

    return taskComplete();
  }

  @Override
  public boolean producesResult() {
    return false;
  }

  @Override
  protected ReturnState taskComplete() {
    // let the iterator (and any tuples still hanging off it) be garbage collected now
    iterator = null;
    return super.taskComplete();
  }
}
