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

import static nz.org.riskscape.engine.Matchers.*;
import static nz.org.riskscape.engine.TupleMatchers.*;

import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import org.junit.Test;

import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.pipeline.Realized;
import nz.org.riskscape.engine.projection.FlatProjector;
import nz.org.riskscape.engine.relation.EmptyRelation;
import nz.org.riskscape.engine.relation.ListRelation;
import nz.org.riskscape.engine.relation.Relation;
import nz.org.riskscape.engine.types.Nullable;
import nz.org.riskscape.engine.types.RSList;
import nz.org.riskscape.engine.types.RelationType;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Types;
import nz.org.riskscape.problem.Problem;
import nz.org.riskscape.problem.Problems;
import nz.org.riskscape.problem.ResultOrProblems;
import nz.org.riskscape.problem.Problem.Severity;
import nz.org.riskscape.rl.ast.ExpressionProblems;
import nz.org.riskscape.rl.ast.PropertyAccess;

@SuppressWarnings("unchecked")
public class UnnestStepTest extends BaseStepTest<UnnestStep> {

  FlatProjector projector;
  Relation input;
  private ResultOrProblems<? extends Realized> result;
  private List<Problem> warnings = new ArrayList<>();

  @Test
  public void canUnnestAList() throws Exception {
    addParam("to-unnest", toExpression("value"));

    assertThat(
      apply(tuple("{value: [1, 3, 5]}")),
      contains(
        tupleWithValue("value", equalTo(1L)),
        tupleWithValue("value", equalTo(3L)),
        tupleWithValue("value", equalTo(5L))
      )
    );

    assertThat(projector.getProducedType(), equalTo(Struct.of("value", Types.INTEGER)));
  }

  @Test
  public void canUnnestARelation() throws Exception {
    Struct relationType = Struct.of("name", Types.TEXT);
    Relation relation = ListRelation.ofTypedValues(relationType, "foo", "bar", "baz");

    addParam("to-unnest", toExpression("value"));

    assertThat(
      apply(Tuple.ofValues(Struct.of("value", new RelationType(relationType)), relation)),
      contains(
        tupleWithValue("value", equalTo(Tuple.ofValues(relationType, "foo"))),
        tupleWithValue("value", equalTo(Tuple.ofValues(relationType, "bar"))),
        tupleWithValue("value", equalTo(Tuple.ofValues(relationType, "baz")))
      )
    );

    assertThat(projector.getProducedType(), equalTo(Struct.of("value", relationType)));
  }

  @Test
  public void canUnnestAListOldSyntax() throws Exception {
    addParam("to-unnest", toExpression("'value'"));

    assertThat(
      apply(tuple("{value: [1, 3, 5]}")),
      contains(
        tupleWithValue("value", equalTo(1L)),
        tupleWithValue("value", equalTo(3L)),
        tupleWithValue("value", equalTo(5L))
      )
    );

    assertThat(projector.getProducedType(), equalTo(Struct.of("value", Types.INTEGER)));
    assertThat(warnings, contains(UnnestStep.PROBLEMS.deprecatedSyntax("unnest(to-unnest: 'value')",
        "unnest(to-unnest: value)")));
  }

  @Test
  public void canUnnestAnEmptyList() throws Exception {
    addParam("to-unnest", toExpression("value"));

    // NB we can also unnest a tuple with an empty list, but that yields a struct with a single nothing member
    assertThat(
      apply(Tuple.ofValues(RSList.create(Types.INTEGER).asStruct(), Collections.emptyList())),
      empty()
    );

    assertEquals(projector.getProducedType(), Struct.of("value", Types.INTEGER));
  }

  @Test
  public void canUnnestAnEmptyListWithEmitEmpty() throws Exception {
    addParam("to-unnest", toExpression("value"));
    addParam("emit-empty", true);

 // NB we can also unnest a tuple with an empty list, but that yields a struct with a single nothing member
    assertThat(
      apply(Tuple.ofValues(RSList.create(Types.INTEGER).asStruct(), Collections.emptyList())),
      contains(
        tupleWithValue("value", nullValue())
      )
    );

    assertEquals(projector.getProducedType(), Struct.of("value", Nullable.INTEGER));
  }

  @Test
  public void canUnnestAnEmptyRelationWithEmitEmpty() throws Exception {
    Struct relationType = Struct.of("value", Types.INTEGER);
    Relation emptyRelation = new EmptyRelation(relationType);
    addParam("to-unnest", toExpression("value"));
    addParam("emit-empty", true);

    assertThat(
      apply(Tuple.ofValues(Struct.of("value", new RelationType(relationType)), emptyRelation)),
      contains(
        tupleWithValue("value", nullValue())
      )
    );

    assertEquals(projector.getProducedType(), Struct.of("value", Nullable.of(relationType)));
  }

  @Test
  public void canUnnestAListOfStructs() throws Exception {
    addParam("to-unnest", toExpression("bar"));

    Struct listElementType = Types.TEXT.asStruct();
    Struct inputType = Struct.of("foo", Types.INTEGER, "bar", RSList.create(listElementType));
    Tuple toUnnest = Tuple.ofValues(inputType, 1L, Arrays.asList(
        Tuple.ofValues(listElementType, "a"),
        Tuple.ofValues(listElementType, "b"),
        Tuple.ofValues(listElementType, "c")
    ));

 // NB we can also unnest a tuple with an empty list, but that yields a struct with a single nothing member
    assertThat(
      apply(toUnnest),
      contains(
        allOf(tupleWithValue("foo", equalTo(1L)), tupleWithValue("bar", tupleWithValue("value", equalTo("a")))),
        allOf(tupleWithValue("foo", equalTo(1L)), tupleWithValue("bar", tupleWithValue("value", equalTo("b")))),
        allOf(tupleWithValue("foo", equalTo(1L)), tupleWithValue("bar", tupleWithValue("value", equalTo("c"))))
      )
    );

    assertEquals(projector.getProducedType(), Struct.of("foo", Types.INTEGER, "bar", listElementType));
  }

  @Test
  public void canUnnestAListOfStructsWithEmitEmpty() throws Exception {
    addParam("to-unnest", toExpression("bar"));
    addParam("emit-empty", true);

    Struct listElementType = Types.TEXT.asStruct();
    Struct inputType = Struct.of("foo", Types.INTEGER, "bar", RSList.create(listElementType));
    Tuple toUnnest = Tuple.ofValues(inputType, 1L, Arrays.asList());

 // NB we can also unnest a tuple with an empty list, but that yields a struct with a single nothing member
    assertThat(
      apply(toUnnest),
      contains(
        allOf(tupleWithValue("foo", equalTo(1L)), tupleWithValue("bar", nullValue()))
      )
    );

    assertEquals(projector.getProducedType(), Struct.of("foo", Types.INTEGER, "bar", Nullable.of(listElementType)));
  }

  @Test
  public void canUnnestMultipleEmptyListsWithIdentityValue() throws Exception {
    Struct inputType = Struct.of(
      "list1", RSList.create(Types.INTEGER),
      "list2", RSList.create(Types.TEXT)
    );

    Tuple bothEmpty = Tuple.ofValues(inputType, Collections.emptyList(), Collections.emptyList());

    addParam("to-unnest", toExpression("[list1,list2]"));
    addParam("emit-empty", true);

    assertThat(
      apply(bothEmpty),
      contains(
        allOf(tupleWithValue("list1", nullValue()), tupleWithValue("list2", nullValue()))
      )
    );
  }

  @Test
  public void canUnnestWithFirstListEmptyWithIdentityValue() throws Exception {
    Struct inputType = Struct.of(
      "list1", RSList.create(Types.INTEGER),
      "list2", RSList.create(Types.TEXT)
    );

    Tuple list1Empty = Tuple.ofValues(inputType, Collections.emptyList(), Arrays.asList("foo", "bar", "baz"));

    addParam("to-unnest", toExpression("[list1,list2]"));
    // note that emit-empty is not needed as long as one list is non-empty
    // addParam("emit-empty", true);

    assertThat(
      apply(list1Empty),
      contains(
        allOf(tupleWithValue("list1", nullValue()), tupleWithValue("list2", equalTo("foo"))),
        allOf(tupleWithValue("list1", nullValue()), tupleWithValue("list2", equalTo("bar"))),
        allOf(tupleWithValue("list1", nullValue()), tupleWithValue("list2", equalTo("baz")))
      )
    );
  }

  @Test
  public void canUnnestWithSecondListEmptyWithIdentityValue() throws Exception {
    Struct inputType = Struct.of(
      "list1", RSList.create(Types.INTEGER),
      "list2", RSList.create(Types.TEXT)
    );

    Tuple list2Empty = Tuple.ofValues(inputType, Arrays.asList(1L, 2L, 3L), Collections.emptyList());

    addParam("to-unnest", toExpression("[list1,list2]"));
    // emit-empty is not strictly needed here, but enable it this time for kicks
    addParam("emit-empty", true);

    assertThat(
      apply(list2Empty),
      contains(
        allOf(tupleWithValue("list1", equalTo(1L)), tupleWithValue("list2", nullValue())),
        allOf(tupleWithValue("list1", equalTo(2L)), tupleWithValue("list2", nullValue())),
        allOf(tupleWithValue("list1", equalTo(3L)), tupleWithValue("list2", nullValue()))
      )
    );
  }

  @Test
  public void canUnnestWithFirstRelationEmptyWithIdentityValue() throws Exception {
    Struct relationType = Struct.of("name", Types.TEXT);
    Relation rel1 = new EmptyRelation(relationType);
    Relation rel2 = ListRelation.ofTypedValues(relationType, "foo", "bar", "baz");
    Struct inputType = Struct.of(
      "rel1", rel1.getScalarDataType(),
      "rel2", rel2.getScalarDataType()
    );

    Tuple list1Empty = Tuple.ofValues(inputType, rel1, rel2);

    addParam("to-unnest", toExpression("[rel1,rel2]"));
    // emit-empty is not strictly needed here, but enable it for kicks
    addParam("emit-empty", true);

    assertThat(
      apply(list1Empty),
      contains(
        allOf(
            tupleWithValue("rel1", nullValue()),
            tupleWithValue("rel2", equalTo(Tuple.ofValues(relationType, "foo")))
        ),
        allOf(
            tupleWithValue("rel1", nullValue()),
            tupleWithValue("rel2", equalTo(Tuple.ofValues(relationType, "bar")))
        ),
        allOf(
            tupleWithValue("rel1", nullValue()),
            tupleWithValue("rel2", equalTo(Tuple.ofValues(relationType, "baz")))
        )
      )
    );
  }

  @Test
  public void canUnnestWithSecondRelationEmptyWithIdentityValue() throws Exception {
    Struct relationType = Struct.of("name", Types.TEXT);
    Relation rel2 = new EmptyRelation(relationType);
    Relation rel1 = ListRelation.ofTypedValues(relationType, "foo", "bar", "baz");
    Struct inputType = Struct.of(
      "rel1", rel1.getScalarDataType(),
      "rel2", rel2.getScalarDataType()
    );

    Tuple list1Empty = Tuple.ofValues(inputType, rel1, rel2);

    addParam("to-unnest", toExpression("[rel1,rel2]"));
    // note that emit-empty is not needed as long as one relation is non-empty
    // addParam("emit-empty", true);

    assertThat(
      apply(list1Empty),
      contains(
        allOf(
            tupleWithValue("rel2", nullValue()),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "foo")))
        ),
        allOf(
            tupleWithValue("rel2", nullValue()),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "bar")))
        ),
        allOf(
            tupleWithValue("rel2", nullValue()),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "baz")))
        )
      )
    );
  }

  @Test
  public void canUnnestListAndRelation() throws Exception {
    Struct relationType = Struct.of("name", Types.TEXT);
    Relation rel1 = ListRelation.ofTypedValues(relationType, "foo", "bar", "baz");
    Struct inputType = Struct.of(
      "rel1", rel1.getScalarDataType(),
      "list1", RSList.create(Types.INTEGER)
    );

    Tuple list1Empty = Tuple.ofValues(inputType, rel1, List.of(10L, 20L));

    addParam("to-unnest", toExpression("[rel1,list1]"));

    assertThat(
      apply(list1Empty),
      contains(
        allOf(
            tupleWithValue("list1", is(10L)),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "foo")))
        ),
        allOf(
            tupleWithValue("list1", is(20L)),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "bar")))
        ),
        allOf(
            tupleWithValue("list1", nullValue()),
            tupleWithValue("rel1", equalTo(Tuple.ofValues(relationType, "baz")))
        )
      )
    );
  }

  @Test
  public void canUnnestMultipleListsWithSingleElementInEach() throws Exception {
    Struct inputType = Struct.of(
        "foo", RSList.create(Types.INTEGER),
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("[foo,bar]"));

    Struct expectedType = Struct.of(
        "foo", Nullable.of(Types.INTEGER),
        "bar", Nullable.of(Types.DATE),
        "other", Types.TEXT
    );

    Date date = new Date();

    assertThat(
      apply(Tuple.ofValues(inputType, Arrays.asList(1L), Arrays.asList(date), "ok")),
      contains(
        Tuple.ofValues(expectedType, 1L, date, "ok")
      )
    );
  }

  @Test
  public void canUnnestMultipleListsWithDifferentNumberOfElements() throws Exception {
    Struct inputType = Struct.of(
        "foo", RSList.create(Types.INTEGER),
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("[foo,bar]"));

    Struct expectedType = Struct.of(
        "foo", Nullable.of(Types.INTEGER),
        "bar", Nullable.of(Types.DATE),
        "other", Types.TEXT
    );

    Date date = new Date();

    // foo list has 2, bar list has 1
    assertThat(
      apply(Tuple.ofValues(inputType, Arrays.asList(1L, 2L), Arrays.asList(date), "ok")),
      contains(
        Tuple.ofValues(expectedType, 1L, date, "ok"),
        Tuple.ofValues(expectedType, 2L, null, "ok")
      )
    );
  }

  @Test
  public void canUnnestMultipleListsWithOneListEmpty() throws Exception {
    Struct inputType = Struct.of(
        "foo", RSList.create(Types.INTEGER),
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("[foo,bar]"));

    Struct expectedType = Struct.of(
        "foo", Nullable.of(Types.INTEGER),
        "bar", Nullable.of(Types.DATE),
        "other", Types.TEXT
    );

    // bar list is empty
    assertThat(
      apply(Tuple.ofValues(inputType, Arrays.asList(1L, 2L), Arrays.asList(), "ok")),
      contains(
        Tuple.ofValues(expectedType, 1L, null, "ok"),
        Tuple.ofValues(expectedType, 2L, null, "ok")
      )
    );
  }

  @Test
  public void canUnnestMultipleListsOldSyntax() throws Exception {
    Struct inputType = Struct.of(
        "foo", RSList.create(Types.INTEGER),
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("[foo,'bar']"));

    Struct expectedType = Struct.of(
        "foo", Nullable.of(Types.INTEGER),
        "bar", Nullable.of(Types.DATE),
        "other", Types.TEXT
    );

    Date date = new Date();

    assertThat(
      apply(Tuple.ofValues(inputType, Arrays.asList(1L), Arrays.asList(date), "ok")),
      contains(
        Tuple.ofValues(expectedType, 1L, date, "ok")
      )
    );

    assertThat(warnings, contains(UnnestStep.PROBLEMS.deprecatedSyntax("unnest(to-unnest: [foo, 'bar'])",
        "unnest(to-unnest: [foo, bar])")));
  }

  @Test
  public void canUnnestAListAddingIndex() throws Exception {
    addParam("to-unnest", toExpression("value"));
    addParam("index-key", "index");

    assertThat(
      apply(tuple("{value: [1, 3, 5]}")),
      contains(
        allOf(tupleWithValue("value", equalTo(1L)), tupleWithValue("index", equalTo(1L))),
        allOf(tupleWithValue("value", equalTo(3L)), tupleWithValue("index", equalTo(2L))),
        allOf(tupleWithValue("value", equalTo(5L)), tupleWithValue("index", equalTo(3L)))
      )
    );

    assertThat(projector.getProducedType(), equalTo(Struct.of("value", Types.INTEGER, "index", Types.INTEGER)));
  }

  @Test
  public void errorIfIndexNameAlreadyPresent() throws Exception {
    addParam("to-unnest", toExpression("value"));
    addParam("index-key", "index");

    realize(Struct.of("index", Types.INTEGER, "value", RSList.create(Types.INTEGER)));

    assertThat(result, failedResult(isProblem(Severity.ERROR,
        is("Cannot create index-key 'index', member already present in {index=>Integer, value=>List[Integer]}"))));
  }

  @Test
  public void canUnnestANullableList() throws Exception {
    addParam("to-unnest", toExpression("list"));

    assertThat(
      apply(Tuple.ofValues(Struct.of("list", Nullable.of(RSList.create(Types.INTEGER))), (Object) null)),
      empty()
    );
  }

  @Test
  public void canUnnestANullableRelation() throws Exception {
    addParam("to-unnest", toExpression("rel"));

    assertThat(
      apply(Tuple.ofValues(Struct.of("rel", Nullable.of(new RelationType(Types.INTEGER.asStruct()))), (Object) null)),
      empty()
    );
  }

  @Test
  public void failsIfToUnnestNotProperty() {
    Struct inputType = Struct.of("foo", Types.TEXT);

    addParam("to-unnest", toExpression("[[1,2,3]]"));
    realize(inputType);
    assertThat(result, failedResult(
        is(Problems.foundWith("to-unnest",
            ExpressionProblems.get().mismatch(toExpression("[1,2,3]"), PropertyAccess.class, "list")))
    ));
  }

  @Test
  public void failsIfGivenMemberDoesNotExist() throws Exception {
    Struct inputType = Struct.of(
        "foo", RSList.create(Types.INTEGER),
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("[foos,bars]"));

    realize(inputType);
    assertThat(result, failedResult(isProblem(Severity.ERROR,
        containsString(
            "Could not find members to unnest [foos, bars] among [bar, foo, other]"
        )
    )));
  }

  @Test
  public void failsIfGivenMemberIsNotAList() throws Exception {
    Struct inputType = Struct.of(
        "foo", Types.INTEGER,
        "bar", RSList.create(Types.DATE),
        "other", Types.TEXT
    );

    addParam("to-unnest", toExpression("foo"));
    realize(inputType);
    assertThat(result, failedResult(
        isProblem(Severity.ERROR, containsString("Member 'foo' is not a list type, was 'Integer'")))
    );
  }

  private List<Tuple> apply(Tuple applyTo) {
    realize(applyTo.getStruct());
    result = result.drainWarnings(p -> warnings.add(p));
    assertTrue(result.get() instanceof FlatProjector);
    projector = (FlatProjector) result.get();

    return projector.apply(applyTo).collect(Collectors.toList());
  }

  private void realize(Struct inputType) {
    input = new EmptyRelation(inputType);

    addStubInput(inputType);

    this.result = step.realizeSimple(input());
  }

  @Override
  protected UnnestStep createStep() {
    return new UnnestStep(engine);
  }

}
