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

import static nz.org.riskscape.engine.Matchers.*;
import static nz.org.riskscape.engine.problem.ProblemMatchers.*;

import java.net.URI;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Future;

import org.junit.Before;
import org.junit.Test;

import nz.org.riskscape.dsl.SourceLocation;
import nz.org.riskscape.dsl.Token;
import nz.org.riskscape.engine.Matchers;
import nz.org.riskscape.engine.ProjectTest;
import nz.org.riskscape.engine.bind.ParamProblems;
import nz.org.riskscape.engine.bind.Parameter;
import nz.org.riskscape.engine.pipeline.ExecutionResult;
import nz.org.riskscape.engine.pipeline.Pipeline;
import nz.org.riskscape.engine.pipeline.PipelineBuilder;
import nz.org.riskscape.engine.pipeline.PipelineProblems;
import nz.org.riskscape.engine.pipeline.RealizedPipeline;
import nz.org.riskscape.engine.pipeline.TestPipelineJobContext;
import nz.org.riskscape.engine.relation.ListRelation;
import nz.org.riskscape.engine.resource.ResourceProblems;
import nz.org.riskscape.problem.Problem;
import nz.org.riskscape.problem.Problems;
import nz.org.riskscape.rl.ExpressionParser;
import nz.org.riskscape.rl.TokenTypes;
import nz.org.riskscape.rl.ast.ExpressionProblems;

// BaseStepTest not suitable - it focuses on simple steps only - won't let me execute the whole pipeline
public class SubpipelineStepTest extends ProjectTest {

  RealizedPipeline realized;
  TestPipelineJobContext jobContext;

  @Before
  public void setup() {
    engine.getPipelineSteps().add(new FilterStep(engine));
    engine.getPipelineSteps().add(new SelectStep(engine));
    engine.getPipelineSteps().add(new SubpipelineStep(engine));
    engine.getPipelineSteps().add(new RelationInputStep(engine));
    engine.getPipelineSteps().add(new SaveStep(engine));
    engine.getPipelineSteps().add(new JoinStep(engine));
//    engine.getPipelineSteps().add(new SaveToList(engine));
    addPickledData("list", ListRelation.ofValues("foo", "bar", "baz"));
  }

  @Test
  public void canIncludeAnotherSimplePipeline() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({text: value})");

    realize("input('list') -> subpipeline(location: 'test:pipeline') as child");

    assertThat(realized.getFailures(), empty());
    execute();

    assertThat(
        // output is named after the subpipeline step, not the last step in the child
        jobContext.getSink("child").getTuples(),
        equalTo(tuples("[{text: 'foo'}, {text: 'bar'}, {text: 'baz'}]"))
    );
  }

  @Test
  public void failsToRealizeIfPipelineLocationIsBung() throws Exception {
    realize("input('list') -> subpipeline(location: 'test:missing')");

    assertThat(realized.getFailures(), not(empty()));
    assertThat(
        realized.getStep("subpipeline").get().getProblems(),
        contains(
            Problems.foundWith(
                Parameter.required("location", URI.class),
                ResourceProblems.get().notFound(URI.create("test:missing"))
            )
        )
    );
  }

  @Test
  public void canChainStepsFromTheSubpipeline() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({text: value})");

    realize("input('list') -> subpipeline(location: 'test:pipeline') -> select({foo: text}) as output");

    assertThat(realized.getFailures(), empty());
    execute();

    // should only be one output...
    assertThat(jobContext.getStepToSink().keySet(), hasSize(1));

    // and it is our select step after the subpipeline
    assertThat(
        jobContext.getSink("output").getTuples(),
        equalTo(tuples("[{foo: 'foo'}, {foo: 'bar'}, {foo: 'baz'}]"))
    );
  }

  @Test
  public void subpipelineCanAddItsOwnOutputs() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", """
        in -> select({text: value})

        in -> save(name: 'debug')
        """);

    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(realized.getFailures(), empty());
    execute();

    // two outputs
    assertThat(jobContext.getStepToSink().keySet(), containsInAnyOrder("subpipeline", "debug"));

    // check the select transformed the input
    assertThat(
        jobContext.getSink("subpipeline").getTuples(),
        equalTo(tuples("[{text: 'foo'}, {text: 'bar'}, {text: 'baz'}]"))
    );

    // the debug should be the original values we input
    assertThat(
        jobContext.getSink("debug").getTuples(),
        equalTo(tuples("[{value: 'foo'}, {value: 'bar'}, {value: 'baz'}]"))
    );
  }

  @Test
  public void subpipelineCanHaveNoOutputs() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", """
        in -> save(name: 'results')
        """);

    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(realized.getFailures(), empty());
    execute();

    // one output
    assertThat(jobContext.getStepToSink().keySet(), containsInAnyOrder("results"));

    // the results should be the original values we input
    assertThat(
        jobContext.getSink("results").getTuples(),
        equalTo(tuples("[{value: 'foo'}, {value: 'bar'}, {value: 'baz'}]"))
    );
  }

  @Test
  public void cannotChainFromSubpipelineThatProducesNoOutput() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", """
        in -> save(name: 'results')
        """);

    realize("input('list') -> subpipeline(location: 'test:pipeline') -> save('results1')");

    assertThat(realized.getFailures(), contains(
        Matchers.hasAncestorProblem(
            is(PipelineProblems.get().chainingFromStepWithNoOutput("subpipeline", List.of("save")))
        )
    ));
  }

  @Test(timeout = 5_000)
  public void recursiveSubpipelinesAreDetected() throws Exception {
    customProtocolTestingResourceLoader.addContent("recursive1", """
        in -> select({*, $i as "recursive1$i"})
        -> subpipeline(location: 'test:recursive2', parameters: {i: $i})
        """);
    customProtocolTestingResourceLoader.addContent("recursive2", """
        in -> select({*, $i as "recursive2$i"})
        -> subpipeline(location: 'test:recursive1', parameters: {i: $i})
        """);

    realize("input('list') -> subpipeline(location: 'test:recursive1', parameters: {i: 1})");
    assertThat(realized.getFailures(), contains(
        Matchers.hasAncestorProblem(is(
            Problems.foundWith(URI.create("test:recursive1"),
                SubpipelineStep.PROBLEMS.recursion(URI.create("test:recursive1"))
            )
        ))
    ));
  }

  @Test
  public void subpipelineCanNominateAStepAsOutputTheOthersGetCapped() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", """
        in -> select({from_out: value}) as out
        in -> select({from_select: value}) as select
        """);

    realize("input('list') -> subpipeline(location: 'test:pipeline') as child -> select({text: from_out})");

    assertThat(realized.getFailures(), empty());
    execute();

    // two outputs
    assertThat(jobContext.getStepToSink().keySet(), containsInAnyOrder("child.select", "select"));

    // check the select transformed the input
    assertThat(
        jobContext.getSink("select").getTuples(),
        equalTo(tuples("[{text: 'foo'}, {text: 'bar'}, {text: 'baz'}]"))
    );

    // the debug should be the original values we input
    assertThat(
        jobContext.getSink("child.select").getTuples(),
        equalTo(tuples("[{from_select: 'foo'}, {from_select: 'bar'}, {from_select: 'baz'}]"))
    );
  }

  @Test
  public void failsIfSubPipelineHasMultipleNonEmptyOutputs() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", """
        # neither of these are named 'out', both produce a non-empty struct - we don't know which one to pick, so fail
        in -> select({from1: value})
        in -> select({from1: value}) as named
        """);

    realize("input('list') -> subpipeline(location: 'test:pipeline') as child -> select({*})");

    Problem expectedProblem = SubpipelineStep.PROBLEMS.ambiguousOutput(List.of(
        // one had a name, the other didn't
        Token.token(TokenTypes.IDENTIFIER, "select"),
        Token.token(TokenTypes.IDENTIFIER, "named")
    ), SubpipelineStep.PROBLEMS.ambiguousOutputHint());

    assertThat(
        realized.getFailures(),
        contains(
            Matchers.hasAncestorProblem(equalTo(expectedProblem))
        )
    );
  }

  @Test
  public void canPassParameters() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({*, param: $foo})");

    realize("input('list') -> subpipeline(location: 'test:pipeline', parameters: {foo: 'value'})");

    assertThat(realized.getFailures(), empty());
    execute();

    // the debug should be the original values we input
    assertThat(
        jobContext.getSink("subpipeline").getTuples(),
        equalTo(tuples(
            "[{value: 'foo', param: 'value'}, {value: 'bar', param: 'value'}, {value: 'baz', param: 'value'}]"
        ))
    );
  }

  @Test
  public void failsIfMissingParameters() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({*, param: $foo})");

    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(
        realized.getFailures(),
        contains(
            Matchers.hasAncestorProblem(equalTo(
                SubpipelineStep.PROBLEMS.missingParameters(URI.create("test:pipeline"), Set.of("foo"))
                  .withChildren(SubpipelineStep.PROBLEMS.missingParameterHint("foo", new SourceLocation(24, 1, 25)))
            ))
        )
    );
  }

  @Test
  public void warnsIfSurplusParametersAreGiven() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({*})");

    realize("input('list') -> subpipeline(location: 'test:pipeline', parameters: {bonus: true})");

    assertThat(
        realized.getFailures(),
        empty()
    );

    assertThat(
        realized.getStep("subpipeline").get().getProblems(),
        contains(
            Problems.foundWith(
                URI.create("test:pipeline"), ParamProblems.get().ignored("bonus")
            )
        )
    );
  }

  @Test
  public void failsIfChildPipelineFailsToRealize() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({text: source})");

    realize("input('list') -> subpipeline(location: 'test:pipeline') as child");

    assertThat(
        realized.getFailures(),
        // NB we'll add some full error matching methods in an integration test later/elsewhere
        contains(Matchers.hasAncestorProblem(Matchers.equalIgnoringChildren(
            SubpipelineStep.PROBLEMS.childFailed(URI.create("test:pipeline"))
        )))
    );

    assertThat(realized.getStep("child").get(), hasProperty("directlyFailed", is(true)));
    // this should be the problem from the select, buried deep...
    Problem failureProblem = realized.getStep("child").get().getProblems().get(0).getChildren().get(0)
        .getChildren().get(0);

    assertThat(
        realized.getStep("child.select").get(),
        allOf(
            hasProperty("directlyFailed", is(false)),
            hasProperty("failed", is(true)),
            // assert the thing we pulled from child is also on child.select - the idea is that if you run the pipeline,
            // you'll see only the error on the subpipeline step, but if you debug it, you'll see everything ( as
            // --print shows all the errors)
            hasProperty("problems", contains(failureProblem))
        )
    );
  }

  @Test
  public void failsIfChildPipelineFailsToRealize2() throws Exception {
    // GL#1550 reduced test case. previously this pipeline caused a multiple end step error
    customProtocolTestingResourceLoader.addContent("pipeline", """
        input(value: 'foo', name: 'foo') as foo
        ->
        # but removing this filter(true) would produce the correct error
        filter(true)
        ->
        filter(x > y)
        ->
        join.rhs

        input(value: 'bar', name: 'bar') as bar
        -> join(on: true)

        foo -> select({*})
        bar -> select({*})
        """);

    realize("subpipeline(location: 'test:pipeline')");

    assertThat(
        realized.getFailures(),
        contains(
            Matchers.hasAncestorProblem(equalTo(ExpressionProblems.get().noSuchStructMember("x", List.of("foo"))))
        )
    );
  }

  @Test
  public void doesNotNeedInputIfInNotPresent() throws Exception {
    // this example is a flipped compared to the others, the input comes out of the subpipeline
    customProtocolTestingResourceLoader.addContent("pipeline", "input('list')");
    realize("subpipeline(location: 'test:pipeline') as input -> select({foo: value})");

    assertThat(realized.getFailures(), empty());

    execute();

    // check the select transformed the input
    assertThat(
        jobContext.getSink("select").getTuples(),
        equalTo(tuples("[{foo: 'foo'}, {foo: 'bar'}, {foo: 'baz'}]"))
    );
  }

  @Test
  public void failsIfGivenMultipleInputs() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({text: source})");

    realize("""
        subpipeline(location: 'test:pipeline') as child
        input('list') as input1 -> child
        input('list') as input2 -> child
        """);

    // this is actually coming out of the realizer, but I'm adding this test "just in case" that gets refactored
    Problem expectedProblem = Problem.error(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED, "input2", "child", "input1");

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(equalTo(expectedProblem)))
    );
  }

  @Test
  public void failsIfGivenBadParameterSyntax() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> filter($filter)");
    realize("input('list') -> subpipeline(location: 'test:pipeline', parameters: {filter = true})");

    assertThat(
        realized.getFailures(),
        contains(hasAncestorProblem(
            isProblem(SubpipelineStep.LocalProblems.class, (r, p) -> p.parametersNotNamed(
                r.match(hasProperty("location", is(SourceLocation.index(69))))
            ))
        ))
    );
  }

  @Test
  public void failsIfGivenInputButNoInStepReference() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "select({text: source})");
    realize("input('list') as foo_input -> subpipeline(location: 'test:pipeline')");

    Problem expectedProblem = SubpipelineStep.PROBLEMS.inStepRefMissing("foo_input");

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(equalTo(expectedProblem)))
    );
  }

  @Test
  public void failsIfNoInputButInStepIsReferredTo() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "in -> select({text: source})");
    realize("subpipeline(location: 'test:pipeline')");

    Problem expectedProblem = SubpipelineStep.PROBLEMS.inStepRefButNoInput();

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(equalTo(expectedProblem)))
    );
  }

  @Test
  public void failsWhenSubpipelineIsEmpty() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline", "");
    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(
            is(Problems.foundWith(URI.create("test:pipeline"), SubpipelineStep.PROBLEMS.empty()))
        ))
    );
  }

  @Test
  public void failsIfInIsRedeclared() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline",
        """
        in -> select({*}) -> out
        input(value: {foo: 'bar'}) as in
        """);
    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(isError(PipelineBuilder.ProblemCodes.STEP_REDEFINITION)))
    );
  }

  @Test
  public void failsIfSubpipelineRedirectsToIn() throws Exception {
    customProtocolTestingResourceLoader.addContent("pipeline",
        """
        in -> filter(true) -> select({*}) as out
        input(value: {foo: 'bar'}) -> in
        """);
    realize("input('list') -> subpipeline(location: 'test:pipeline')");

    assertThat(
        realized.getFailures(),
        contains(Matchers.hasAncestorProblem(
            isProblem(SubpipelineStep.LocalProblems.class, (r, p) -> p.chainingToIn(
                r.match(hasProperty("location", is(new SourceLocation(71, 2, 31))))
            ))
        ))
    );
  }

  private void execute() throws Exception {
    jobContext = new TestPipelineJobContext(realized);
    Future<ExecutionResult> future =
        executionContext.getPipelineExecutor().execute(jobContext);
    future.get();
  }

  private void realize(String pipelineSource) {
    this.realized = executionContext.realize(ExpressionParser.INSTANCE.parsePipeline(pipelineSource));
  }
}
