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

import java.util.ArrayList;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import nz.org.riskscape.engine.sched.Page.ReadOnlyPage;
import nz.org.riskscape.engine.task.PageBuffer;
import nz.org.riskscape.engine.task.ReadPageBuffer;

@RequiredArgsConstructor
public class LinkedListBuffer implements PageBuffer {

  private static final Page COMPLETE = new Page(0);
  private static final Page EMPTY = new Page(0);

  private static class Node {
    Page page = EMPTY;
    Node next;
  }

  /**
   * The start is always the next page to read out of the buffer.
   * The end is always where a new page gets added (write case).
   * The list is always terminated by an Node with an EMPTY page.
   */
  private volatile Node start = new Node();
  private volatile Node end = start;

  public static final int DEFAULT_PAGE_SIZE = 1024;
  public static final int DEFAULT_CAPACITY = 8;

  /**
   * Controls the number of Tuples per page (i.e. allocated by newPage())
   */
  private final PageAllocator allocator;

  /**
   * the maximum number of tuples we can have present in the buffer at once before it is considered
   * full. Note that this capacity can be overshot slightly, as pages can be different sizes.
   */
  @Getter
  private final int tupleCapacity;

  private volatile long tuplesAllocated = 0;
  private volatile long tuplesWritten = 0;
  private volatile long tuplesRead = 0;

  private final ArrayList<Clone> clones = new ArrayList<>();

  protected LinkedListBuffer() {
    this(DEFAULT_PAGE_SIZE, DEFAULT_CAPACITY);
  }

  /**
   * Create a buffer with a fixed page-size and a maximum number of pages it can hold at any one time.
   * This is mostly used for testing.
   */
  protected LinkedListBuffer(int pageSize, int pageCapacity) {
    this(new PageAllocator(pageSize), pageSize * pageCapacity);
  }

  /**
   * @return a new buffer with all the same settings, except for a new maximum page size.
   */
  public LinkedListBuffer withMaxPageSize(int pageSize) {
    return new LinkedListBuffer(allocator.withMaxPageSize(pageSize), tupleCapacity);
  }

  /**
   * @return the max number of tuples that could potentially fill in one page
   */
  public int getMaxPageSize() {
    return allocator.getMaxPageSize();
  }

  @Override
  public Page newPage() {
    // we always provide a new page, if asked. However, we still need to count
    // these toward the memory overhead of this buffer, even though they haven't
    // actually been added to the buffer yet
    synchronized (this) {
      Page page = allocator.newPage();
      tuplesAllocated += page.getMaxSize();
      return page;
    }
  }

  @Override
  public void add(Page page) {
    int numTuplesInPage = page.getTupleCount();

    synchronized (this) {
      end.next = new Node();
      end.page = page;
      end = end.next;
      tuplesWritten += numTuplesInPage;
      if (tuplesAllocated < page.getMaxSize()) {
        throw new IllegalStateException("Error calculating tuples allocated to buffer");
      }
      tuplesAllocated -= page.getMaxSize();
    }
  }

  @Override
  public boolean isFull() {
    int tuplesInBuffer = size();
    synchronized (this) {
      return (tuplesAllocated + tuplesInBuffer) >= getTupleCapacity();
    }
  }

  @Override
  public boolean isEmpty() {
    synchronized (this) {
      // a 'complete' buffer is still regarded as empty, as there are
      // no Tuples to read
      return start.page == EMPTY || start.page == COMPLETE;
    }
  }

  @Override
  public ReadOnlyPage read() {
    ReadOnlyPage page = null;
    synchronized (this) {
      if (!isEmpty()) {
        page = start.page.getReadOnlyCopy();
        start = start.next;
      }
    }

    if (page != null) {
      int numTuplesInPage = page.getTupleCount();
      synchronized (this) {
        tuplesRead += numTuplesInPage;
      }
    }

    return page;
  }

  @Override
  public void markComplete() {
    add(COMPLETE);
  }

  @Override
  public boolean isComplete() {
    synchronized (this) {
      return start.page == COMPLETE;
    }
  }

  /**
   * @return the current number of unread tuples in the buffer.
   */
  @Override
  public int size() {
    long minTuplesRead = getMinTuplesRead();
    synchronized (this) {
      return (int) (tuplesWritten - minTuplesRead);
    }
  }

  @Override
  public String toString() {
    if (isComplete()) {
      return String.format("%s[complete!, tuples-in=%d tuples-out=%d]",
          getClass().getSimpleName(), tuplesWritten, tuplesRead);
    } else {
      return String.format("%s[size=%d, tuples-in=%d tuples-out=%d]",
          getClass().getSimpleName(), size(), tuplesWritten, tuplesRead);
    }
  }

  /**
   * If multiple downstream steps use the same input (i.e. fan-out), then they
   * need to use clones of the ReadBuffer. This means that each step can manage
   * its own input (without inadvertently stealing Tuples from its fan-out peers).
   */
  private class Clone implements ReadPageBuffer {
    volatile Node cloneStart = LinkedListBuffer.this.start;
    volatile long cloneTuplesRead;

    @Override
    public ReadOnlyPage read() {
      ReadOnlyPage page = null;

      // there are two different aspects of synchronization here:
      // 1. Coping with multiple threads reading from this clone.
      //    We lock the Clone to address this.
      // 2. Coping with any number of writers updating the original buffer.
      //    We don't lock on LinkedListTupleBuffer.this here - it's not worth the
      //    complexity. There's always a Node there, so the worst case is that the
      //    clone will get a null page (i.e. one that's not quite available yet).
      //    This is the same as trying to read an empty buffer, so the task should
      //    just get rescheduled for later.
      //    NB we could use finer grained locking to avoid this, but it's gonna be complex
      synchronized (this) {
        if (!isEmpty()) {
          page = cloneStart.page.getReadOnlyCopy();
          cloneStart = cloneStart.next;
          cloneTuplesRead += page.getTupleCount();
        }
      }

      return page;
    }

    @Override
    public boolean isComplete() {
      synchronized (this) {
        return cloneStart.page == COMPLETE;
      }
    }

    @Override
    public boolean isEmpty() {
      synchronized (this) {
        return cloneStart.page == EMPTY || cloneStart.page == COMPLETE;
      }
    }

    @Override
    public int size() {
      return (int) (tuplesWritten - cloneTuplesRead);
    }

    @Override
    public long numTuplesRead() {
      return cloneTuplesRead;
    }
  }

  @Override
  public ReadPageBuffer newReaderClone() {
    Clone clone = new Clone();
    this.clones.add(clone);
    return clone;
  }

  /**
   * Returns the minimum point that the ReadBuffer, and any cloned ReadBuffers,
   * have all read up to. This is used to determine the effective size() of the
   * buffer, i.e. the number of unread pages it holds.
   */
  private long getMinTuplesRead() {
    // note that we don't bother grabbing a lock here. The Read/WriteBuffers
    // will keep polling for more work, so as long as we eventually return
    // the correct answer, it should be OK
    long min = tuplesRead;
    for (Clone clone : clones) {
      min = Math.min(min, clone.cloneTuplesRead);
    }
    return min;
  }

  @Override
  public long numTuplesWritten() {
    return tuplesWritten;
  }

  @Override
  public long numTuplesRead() {
    return tuplesRead;
  }
}
