/*
 * 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 static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.sched.Page.ReadOnlyPage;
import nz.org.riskscape.engine.task.ReadPageBuffer;
import nz.org.riskscape.engine.task.WritePageBuffer;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Types;

public class LinkedListBufferTest {

  Struct struct = Types.INTEGER.asStruct();
  int pageCapacity = 10;
  LinkedListBuffer buffer = new LinkedListBuffer(1, pageCapacity);
  List<Thread> threads = new ArrayList<Thread>();
  int limit = 1000;
  int numReaders = 3;
  int numWriters = 3;
  AtomicInteger counter = new AtomicInteger();
  int[] read = new int[limit];

  @Test
  public void canWriteAndThenReadABufferInSerial() {
    // check empty buffer
    assertTrue(buffer.isEmpty());
    assertFalse(buffer.isFull());
    assertEquals(0, buffer.size());
    assertFalse(buffer.isComplete());

    // check non-empty buffer
    buffer.add(pageOf(0));
    assertFalse(buffer.isEmpty());
    assertFalse(buffer.isFull());
    assertEquals(1, buffer.size());
    assertFalse(buffer.isComplete());

    // check almost full
    buffer.add(pageOf(1));
    buffer.add(pageOf(2));
    buffer.add(pageOf(3));
    buffer.add(pageOf(4));
    buffer.add(pageOf(5));
    buffer.add(pageOf(6));
    buffer.add(pageOf(7));
    buffer.add(pageOf(8));
    assertFalse(buffer.isFull());
    assertEquals(9, buffer.size());

    // check full
    buffer.add(pageOf(9));
    assertEquals(10, buffer.size());
    assertTrue(buffer.isFull());

    // check reading values out of buffer
    List<Long> values = new ArrayList<>();
    while (!buffer.isEmpty()) {
      values.add((Long) buffer.read().next().fetch("value"));
    }

    assertEquals(
        Arrays.asList(0L, 1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L),
        values);

    // check buffer empty again
    assertTrue(buffer.isEmpty());
    assertNull(buffer.read());
    assertNull(buffer.read());

    // check buffer complete
    assertFalse(buffer.isComplete());
    buffer.markComplete();
    assertTrue(buffer.isComplete());
    assertNull(buffer.read());
    assertNull(buffer.read());
  }

  @Test
  public void canDetermineBufferFullBasedOnTuples() {
    // the tests use a fixed page-size for simplicity, but the overall limit
    // is actually the total tuple count, so 100 x 2 = 200 tuples
    buffer = new LinkedListBuffer(100, 2);
    assertFalse(buffer.isFull());

    // so we can exhaust the limit with two x 100 tuple pages
    buffer.add(pageOf(1, 100));
    buffer.add(pageOf(1, 100));
    assertTrue(buffer.isFull());
    assertEquals(200, buffer.size());
    buffer.read();
    assertFalse(buffer.isFull());
    assertEquals(100, buffer.size());
    buffer.read();

    // or split it across different-sized pages
    buffer.add(pageOf(2, 100));
    buffer.add(pageOf(2, 99));
    assertFalse(buffer.isFull());
    buffer.add(pageOf(2, 1));
    assertTrue(buffer.isFull());
    buffer.read();
    buffer.read();
    buffer.read();
    assertFalse(buffer.isFull());
    assertTrue(buffer.isEmpty());

    // note the queue limit is an honesty policy, so we can still overfill it
    buffer.add(pageOf(3, 100));
    buffer.add(pageOf(3, 100));
    buffer.add(pageOf(3, 100));
    assertTrue(buffer.isFull());
    assertEquals(300, buffer.size());
  }

  @Test
  public void canCountAllocatedPagesTowardsBufferFull() {
    // create a buffer that holds five 1-tuple pages
    buffer = new LinkedListBuffer(1, 5);

    assertFalse(buffer.isFull());
    Page allocated1 = buffer.newPage();
    Page allocated2 = buffer.newPage();

    // check almost full
    buffer.add(pageOf(1));
    buffer.add(pageOf(2));
    assertFalse(buffer.isFull());
    assertEquals(2, buffer.size());

    // added 3 pages plus allocated 2
    buffer.add(pageOf(3));
    assertTrue(buffer.isFull());
    assertEquals(3, buffer.size());

    // add the allocated pages to the queue
    allocated1.add(Tuple.ofValues(struct, 4));
    buffer.add(allocated1);
    allocated2.add(Tuple.ofValues(struct, 5));
    buffer.add(allocated2);
    assertTrue(buffer.isFull());
    assertEquals(5, buffer.size());

    buffer.read();
    assertFalse(buffer.isFull());
    assertEquals(4, buffer.size());
  }

  @Test
  public void canReadAndWriteToBufferToAndFro() throws Exception {
    // write and read 1 page
    buffer.add(pageOf(0));
    assertReadNextEquals(0);
    assertBufferEmpty();

    // write and read 2 pages
    buffer.add(pageOf(1));
    buffer.add(pageOf(2));
    assertReadNextEquals(1);
    assertReadNextEquals(2);
    assertBufferEmpty();
    assertBufferEmpty();

    // write and read interleaved
    buffer.add(pageOf(3));
    assertReadNextEquals(3);
    buffer.add(pageOf(4));
    assertReadNextEquals(4);
    assertBufferEmpty();

    // fill the buffer
    for (int i = 0; i < pageCapacity; i++) {
      buffer.add(pageOf(i));
    }
    assertTrue(buffer.isFull());

    // read one page and check we can write more
    assertReadNextEquals(0);
    assertFalse(buffer.isFull());
    buffer.add(pageOf(10));
    assertTrue(buffer.isFull());

    buffer.markComplete();
    // note it's not considered complete until we've read all the output
    assertFalse(buffer.isComplete());

    // read all bar one of the pages
    for (int i = 1; i < pageCapacity; i++) {
      assertReadNextEquals(i);
    }

    // read the last page
    assertReadNextEquals(10);
    assertTrue(buffer.isComplete());
    assertNull(buffer.read());
    assertTrue(buffer.isEmpty());
  }

  @Test
  public void canReadAndWriteToClonedBuffer() throws Exception {
    // separate out the interfaces to make it a bit clearer what this test is doing
    ReadPageBuffer readPageBuffer = buffer;
    ReadPageBuffer cloneBuffer = buffer.newReaderClone();
    WritePageBuffer writePageBuffer = buffer;

    // write 1 page and read it from both buffers
    assertTrue(cloneBuffer.isEmpty());
    writePageBuffer.add(pageOf(0));
    assertFalse(cloneBuffer.isEmpty());
    assertFalse(readPageBuffer.isEmpty());
    assertReadNextEquals(cloneBuffer, 0);
    assertTrue(cloneBuffer.isEmpty());
    assertReadNextEquals(readPageBuffer, 0);
    assertTrue(readPageBuffer.isEmpty());

    // fill the write buffer
    for (int i = 0; i < pageCapacity; i++) {
      assertFalse(writePageBuffer.isFull());
      writePageBuffer.add(pageOf(i));
    }
    assertTrue(writePageBuffer.isFull());

    // empty the uncloned buffer
    for (int i = 0; i < pageCapacity; i++) {
      assertReadNextEquals(readPageBuffer, i);
    }

    // from a read perspective, the uncloned buffer is empty
    assertTrue(readPageBuffer.isEmpty());
    assertFalse(cloneBuffer.isEmpty());
    // the write buffer is still full because the clone hasn't read anything yet
    assertTrue(writePageBuffer.isFull());

    // when the clone reads one it should free up space
    assertReadNextEquals(cloneBuffer, 0);
    assertFalse(writePageBuffer.isFull());
    writePageBuffer.add(pageOf(10));
    assertTrue(writePageBuffer.isFull());

    writePageBuffer.markComplete();
    // note reading is not considered complete until we've gotten all the output
    assertFalse(readPageBuffer.isComplete());
    assertFalse(cloneBuffer.isComplete());

    // read the last page for the uncloned buffer
    assertReadNextEquals(readPageBuffer, 10);
    assertTrue(readPageBuffer.isComplete());
    assertNull(buffer.read());

    // read all bar one from the cloned buffer
    for (int i = 1; i < pageCapacity; i++) {
      assertReadNextEquals(cloneBuffer, i);
    }
    assertFalse(cloneBuffer.isComplete());
    assertFalse(cloneBuffer.isEmpty());

    // read the last page
    assertReadNextEquals(cloneBuffer, 10);
    assertTrue(cloneBuffer.isComplete());
    assertNull(cloneBuffer.read());
  }

  @Test
  public void canCountBufferTuplesCorrectly() throws Exception {
    // sanity-check that counters work correctly for debugging
    buffer = new LinkedListBuffer(1000, pageCapacity);
    ReadPageBuffer readPageBuffer = buffer;
    ReadPageBuffer cloneBuffer = buffer.newReaderClone();
    WritePageBuffer writePageBuffer = buffer;

    writePageBuffer.add(pageOf(1, 100));
    assertEquals(100, writePageBuffer.numTuplesWritten());
    assertEquals(0, readPageBuffer.numTuplesRead());
    assertEquals(0, cloneBuffer.numTuplesRead());

    writePageBuffer.add(pageOf(1, 250));
    assertEquals(350, writePageBuffer.numTuplesWritten());
    assertEquals(0, readPageBuffer.numTuplesRead());
    assertEquals(0, cloneBuffer.numTuplesRead());

    cloneBuffer.read();
    assertEquals(100, cloneBuffer.numTuplesRead());
    assertEquals(0, readPageBuffer.numTuplesRead());

    readPageBuffer.read();
    assertEquals(100, cloneBuffer.numTuplesRead());
    assertEquals(100, readPageBuffer.numTuplesRead());

    readPageBuffer.read();
    assertEquals(100, cloneBuffer.numTuplesRead());
    assertEquals(350, readPageBuffer.numTuplesRead());

    cloneBuffer.read();
    assertEquals(350, cloneBuffer.numTuplesRead());
    assertEquals(350, readPageBuffer.numTuplesRead());
    assertEquals(350, writePageBuffer.numTuplesWritten());
  }

  @Test(timeout = 30000)
  public void canReadAndWriteFromSeparateThreads() throws Exception {
    numReaders = 1;
    numWriters = 1;

    testThreadedReadingAndWriting();
  }

  @Test(timeout = 30000)
  public void canHaveMultipleThreadsWriting() throws Exception {
    numWriters = 3;
    numReaders = 1;

    testThreadedReadingAndWriting();
  }

  @Test(timeout = 30000)
  public void canHaveMultipleThreadsReading() throws Exception {
    numWriters = 1;
    numReaders = 3;

    testThreadedReadingAndWriting();
  }

  @Test(timeout = 30000)
  public void canHaveMultipleThreadsReadingAndWriting() throws Exception {
    numReaders = 3;
    numWriters = 3;

    testThreadedReadingAndWriting();
  }

  @Test
  public void aSingleWriterCanSupplyFannedOutReaders() throws Exception {
    numWriters = 1;
    addWriter();

    addReader(buffer);
    addReader(buffer.newReaderClone());
    addReader(buffer.newReaderClone());

    startAll();
    joinAll();

    for (int j = 0; j < read.length; j++) {
      // three counts - one for each fan out
      assertEquals(3, read[j]);
    }

    assertEquals(limit, buffer.numTuplesWritten());
  }

  @Test
  public void multipleWritersCanSupplyFannedOutReaders() throws Exception {
    numWriters = 3;
    addWriter();
    addWriter();
    addWriter();

    addReader(buffer);
    addReader(buffer.newReaderClone());
    addReader(buffer.newReaderClone());

    startAll();
    joinAll();

    for (int j = 0; j < read.length; j++) {
      // nine counts - one for each fan out and one for each writer
      assertEquals(9, read[j]);
    }

    assertEquals(limit * numWriters, buffer.numTuplesWritten());
  }


  @Test
  public void aSingleWriterCanSupplyMultipleThreadedFannedOutReaders() throws Exception {
    numWriters = 1;
    addWriter();

    addReader(buffer);
    addReader(buffer);
    addReader(buffer);

    ReadPageBuffer clone1 = buffer.newReaderClone();
    addReader(clone1);
    addReader(clone1);
    addReader(clone1);

    ReadPageBuffer clone2 = buffer.newReaderClone();
    addReader(clone2);
    addReader(clone2);
    addReader(clone2);

    startAll();
    joinAll();

    for (int j = 0; j < read.length; j++) {
      assertEquals(3, read[j]);
    }
  }

  @Test
  public void multipleWritersCanSupplyMultipleThreadedFannedOutReaders() throws Exception {
    numWriters = 3;
    addWriter();
    addWriter();
    addWriter();

    addReader(buffer);
    addReader(buffer);
    addReader(buffer);

    ReadPageBuffer clone1 = buffer.newReaderClone();
    addReader(clone1);
    addReader(clone1);
    addReader(clone1);

    ReadPageBuffer clone2 = buffer.newReaderClone();
    addReader(clone2);
    addReader(clone2);
    addReader(clone2);

    startAll();
    joinAll();

    for (int j = 0; j < read.length; j++) {
      // nine counts - one for each fan out and one for each writer
      assertEquals(9, read[j]);
    }
  }

  private void addWriter() {
    threads.add(new Thread(() -> {
      int i = 0;
      for (; i < limit;) {
        if (!buffer.isFull()) {
          buffer.add(pageOf(i++));
          counter.incrementAndGet();
        } else {
          try {
            Thread.sleep(10);
          } catch (InterruptedException e) {
          }
          continue;
        }
      }

      // if we're the last Writer thread to finish, mark the buffer as complete
      if (counter.get() == limit * numWriters) {
        buffer.markComplete();
      }
    }));
  }

  private void addReader(ReadPageBuffer readPageBuffer) {
    threads.add(new Thread(() -> {
      while (!readPageBuffer.isComplete()) {
        ReadOnlyPage page = readPageBuffer.read();

        if (page == null) {
          try {
            Thread.sleep(10);
          } catch (InterruptedException e) {
          }
          continue;
        }

        // keep track of how many times we've seen each Tuple value
        int value = ((Long) page.peek().fetch("value")).intValue();
        synchronized (read) {
          read[value]++;
        }
      }
    }));
  }

  private void testThreadedReadingAndWriting() {
    for (int threadCount = 0; threadCount < numWriters; threadCount++) {
      addWriter();
    }

    for (int readerIdx = 0; readerIdx < numReaders; readerIdx++) {
      addReader(buffer);
    }

    startAll();
    joinAll();

    // reading pages is distributed amongst the Reader threads (we don't know which
    // thread will see which page). So the number of times we've seen a Tuple value
    // should match the number of Writer threads
    for (int j = 0; j < read.length; j++) {
      assertEquals(numWriters, read[j]);
    }

    assertEquals(limit * numWriters, buffer.numTuplesWritten());
    assertEquals(limit * numWriters, buffer.numTuplesRead());
  }

  private void assertReadNextEquals(ReadPageBuffer readPageBuffer, Integer i) {
    if (i == null) {
      assertNull(readPageBuffer.read());
    } else {
      // if we've got a page to read, then buffer should not be empty nor complete
      assertFalse(readPageBuffer.isEmpty());
      assertFalse(readPageBuffer.isComplete());

      ReadOnlyPage page = readPageBuffer.read();
      assertEquals(Long.valueOf(i.longValue()), page.next().fetch("value"));
    }
  }

  private void assertReadNextEquals(Integer i) {
    assertReadNextEquals(buffer, i);
  }

  private void assertBufferEmpty() {
    assertTrue(buffer.isEmpty());
    assertReadNextEquals(null);
    assertFalse(buffer.isComplete());
  }

  private void startAll() {
    threads.stream().forEach(Thread::start);
  }

  private void joinAll() {
    threads.stream().forEach(t -> {
      try {
        t.join();
      } catch (InterruptedException e) {
      }
    });
  }

  public Integer read() {
    return ((Long) buffer.read().next().fetch("value")).intValue();
  }

  public Page pageOf(long value) {
    Page page = buffer.newPage();
    page.add(Tuple.ofValues(struct, value));
    return page;
  }

  public Page pageOf(long value, int repeated) {
    Page page = buffer.newPage();
    int i;
    for (i = 0; i < repeated; i++) {
      page.add(Tuple.ofValues(struct, value));
    }
    return page;
  }
}
