/*
 * 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 lombok.Getter;
import lombok.RequiredArgsConstructor;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.relation.TupleIterator;
import nz.org.riskscape.engine.sched.Page.ReadOnlyPage;

/**
 * Reads tuples from a buffer 'page'.
 * The {@link ReadPageBuffer} is potentially shared between multiple {@link WorkerTask}
 * threads, and so is thread-safe.
 * The PageReader is not thread-safe, however, it should only ever belong to
 * a single {@link WorkerTask} (so there's no potential contention).
 * Each {@link WorkerTask} deals with its own page, which they get one at a time
 * from the {@link ReadPageBuffer}. The PageReader manages the page for a
 * {@link WorkerTask} - it tracks where in the current page the task is up to
 * and whether it needs a new one.
 * Note that because the pipeline can potentially consume a lot of memory, we want
 * to drop any references to tuple pages as soon as possible. Centralizing the page
 * reading logic helps with that.
 */
@RequiredArgsConstructor
public class PageReader implements TupleIterator {

  private final ReadPageBuffer input;
  private ReadOnlyPage page = null;
  @Getter
  private long tuplesRead = 0;
  @Getter
  private long pagesRead = 0;

  /**
   * @return true if there is currently a page we're in the middle of reading.
   * Note that this is particularly important as a worker task should never
   * complete while it still has a page in progress.
   */
  public boolean hasPageInProgress() {
    return page != null;
  }

  /**
   * Returns true if there is something for a worker task to do, i.e. there are
   * tuples available to read.
   */
  public boolean hasInput() {
    return hasPageInProgress() || !input.isEmpty();
  }

  /**
   * @return true if there are no more tuples left to read
   */
  public boolean isComplete() {
    return !hasPageInProgress() && input.isComplete();
  }

  private boolean hasNextTuple() {
    return hasPageInProgress() && page.hasNext();
  }

  private boolean readNextPage() {
    page = input.read();
    if (page != null) {
      pagesRead++;
    }
    return (page != null);
  }

  private boolean checkTupleAvailable() {
    // If we haven't got a next tuple, try reading another page.
    // There may not be another page available if the buffer is empty
    return hasNextTuple() || readNextPage();
  }

  /**
   * Returns the Tuple that will be returned by next() without advancing where we
   * are currently up to in the page. Use this if you might have to stop halfway
   * through processing a tuple and come back to it later, e.g. when the output
   * buffer is full. (If it happens to be the last tuple in the page, then this
   * prevents the PageWriter from incorrectly thinking the task worker is finished
   * with the page.
   *
   * @return the next tuple to read, or null if there's no input available
   */
  public Tuple peek() {
    if (!checkTupleAvailable()) {
      return null;
    }
    return page.peek();
  }

  /**
   * Returns the next tuple in the page. This handles reading a new page from the
   * ReadPageBuffer if needed.
   *
   * @return the next tuple to read, or null if there's no input available
   */
  @Override
  public Tuple next() {
    if (!checkTupleAvailable()) {
      return null;
    }

    Tuple tuple = page.next();
    tuplesRead++;

    // If this was the last tuple, then clear the page in progress.
    // We'll try to read another page next time around
    if (!hasNextTuple()) {
      page = null;
    }

    return tuple;
  }

  /**
   * @return whether there is a next tuple available to read.
   * Returning false means there's no tuple *currently* available to read (but
   * there may be later, as more input gets added to the underlying {@link PageBuffer}.
   */
  @Override
  public boolean hasNext() {
    return checkTupleAvailable();
  }

  @Override
  public String toString() {
    String info = String.format("[Tuples: total=%d, this-task=%d, %d pages",
        input.numTuplesRead(), tuplesRead, pagesRead);
    if (page != null) {
      info += ", current=" + page.toString();
    }
    if (input.isComplete()) {
      info += " (complete)";
    } else if (input.isEmpty()) {
      info += " (empty)";
    }

    return info + "]";
  }
}
