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

import static nz.org.riskscape.engine.output.PipelineOutputStore.*;

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

import java.io.File;
import java.net.URI;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.locationtech.jts.geom.Geometry;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import nz.org.riskscape.engine.GeometryMatchers;
import nz.org.riskscape.engine.JdbcHelper;
import nz.org.riskscape.engine.Matchers;
import nz.org.riskscape.engine.OutputProblems;
import nz.org.riskscape.engine.SRIDSet;
import nz.org.riskscape.engine.StructMatchers;
import nz.org.riskscape.engine.TemporaryDirectoryTestHelper;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.TypeMatchers;
import nz.org.riskscape.engine.data.Bookmark;
import nz.org.riskscape.engine.defaults.data.GeoPackageRelationResolver;
import nz.org.riskscape.engine.defaults.data.jdbc.BaseJdbcOutputStore;
import nz.org.riskscape.engine.pipeline.sink.SaveSink;
import nz.org.riskscape.engine.relation.Relation;
import nz.org.riskscape.engine.types.Referenced;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Types;
import nz.org.riskscape.problem.Problem;

public class GeoPackagePipelineOutputStoreTest extends BasePipelineOutputStoreTest
    implements TemporaryDirectoryTestHelper {

  Path outputDirectory;
  URI outputPath;

  GeoPackageRelationResolver resolver = new GeoPackageRelationResolver(engine);

  Struct inputType = Struct.of(
      "geom", Referenced.of(Types.POINT, crs84Helper.getCrs()),
      "index", Types.INTEGER,
      "name", Types.TEXT);

  Tuple inputValue = Tuple.ofValues(inputType, point1Crs84, 1L, "foo");

  public GeoPackagePipelineOutputStoreTest() {
    super(new GeoPackagePipelineOutputStore());
  }

  @Before
  public void setup() throws Exception {
    this.outputDirectory = createTempDirectory(GeoPackagePipelineOutputStoreTest.class.getSimpleName());
    this.outputPath = outputDirectory.resolve("test.gpkg").toUri();
  }

  @After
  public void cleanup() throws Exception {
    remove(outputDirectory);
  }

  @Test
  public void isApplicableForFileURIEndingWithGpgkSuffix() {
    assertThat(subject.isApplicable(URI.create("file://tmp/file.gpkg")), is(PRIORITY_HIGH));

    // some negative tests
    assertThat(subject.isApplicable(URI.create("file.gpkg")), is(PRIORITY_NA)); // missing the file scheme
    assertThat(subject.isApplicable(URI.create("file://tmp/file.gpkg/")), is(PRIORITY_NA)); // it's a directory
    assertThat(subject.isApplicable(URI.create("my-db")), is(PRIORITY_NA));
    assertThat(subject.isApplicable(URI.create("test")), is(PRIORITY_NA));
    assertThat(subject.isApplicable(new File("test").toURI()), is(PRIORITY_NA));
    assertThat(subject.isApplicable(new File("wfs://some-host/some-path").toURI()), is(PRIORITY_NA));
    assertThat(subject.isApplicable(new File("test").toURI()), is(PRIORITY_NA));
    assertThat(subject.isApplicable(new File("wfs://some-host/some-path").toURI()), is(PRIORITY_NA));
  }

  @Test
  public void canWriteSpatialData() {
    try (PipelineOutputContainer container = getOutputContainer(outputPath).get()) {
      writeData(container, "canWriteSpatialData", null, inputValue);
    }

    assertData("canWriteSpatialData", spatialTypeCrs84, Lists.newArrayList(inputValue));
  }

  @Test
  public void canWriteMultipleRelations() {
    try (PipelineOutputContainer container = getOutputContainer(outputPath).get()) {
      writeData(container, "canWriteMultipleRelations1", null, inputValue);
      writeData(container, "canWriteMultipleRelations2", null, inputValue);
      writeData(container, "canWriteMultipleRelations3", null, inputValue);
    }

    assertData("canWriteMultipleRelations1", spatialTypeCrs84, Lists.newArrayList(inputValue));
    assertData("canWriteMultipleRelations2", spatialTypeCrs84, Lists.newArrayList(inputValue));
    assertData("canWriteMultipleRelations3", spatialTypeCrs84, Lists.newArrayList(inputValue));
  }

  @Test
  public void canWriteMultipleRelationsFromDifferentThreads() {
    // exercise writing layers from different threads concurrently
    try (PipelineOutputContainer container = getOutputContainer(outputPath).get()) {
      List<Callable<Void>> tasks = new ArrayList<>();
      tasks.add(buildWriteTask("canWriteMultipleRelationsFromDifferentThreads1", container));
      tasks.add(buildWriteTask("canWriteMultipleRelationsFromDifferentThreads2", container));
      tasks.add(buildWriteTask("canWriteMultipleRelationsFromDifferentThreads3", container));

      executeTasks(tasks, 3);
    }

    assertData("canWriteMultipleRelationsFromDifferentThreads1", inputType,
        getMultiThreadedTestData("canWriteMultipleRelationsFromDifferentThreads1"));
    assertData("canWriteMultipleRelationsFromDifferentThreads2", inputType,
        getMultiThreadedTestData("canWriteMultipleRelationsFromDifferentThreads2"));
    assertData("canWriteMultipleRelationsFromDifferentThreads3", inputType,
        getMultiThreadedTestData("canWriteMultipleRelationsFromDifferentThreads3"));
  }

  @Test
  public void savesManifestWithPipelineAndStats() {
    // we need to seed some metrics to they can be written out later on
    Meter tuplesIn = pipeline.getContext().getMetricRegistry().meter("test.tuples-in");
    tuplesIn.mark();
    Counter runtime = pipeline.getContext().getMetricRegistry().counter("test.runtime");
    runtime.inc(2000);

    PipelineOutputOptions options = defaultOptions();
    options.setReference("manifest-test1");
    try (PipelineOutputContainer container = getOutputContainer(outputPath, options).get()) {
      writeData(container, "canWriteSpatialData", null, inputValue);
    }
    options.setReference("manifest-test2");
    try (PipelineOutputContainer container = getOutputContainer(outputPath, options).get()) {
      writeData(container, "canWriteSpatialData1", null, inputValue);
      writeData(container, "canWriteSpatialData2", null, inputValue);
    }

    try (JdbcHelper jdbcHelper = JdbcHelper.fromGeoPackage(new File(outputPath))) {
      assertThat(jdbcHelper.readTable(BaseJdbcPipelineOutputContainer.MANIFEST_TABLE,
          "reference", "started", "finished", "manifest", "pipeline", "stats", "output_tables"), contains(
              contains(
                  is("manifest-test1"),
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}T.*"), // looks like a datestamp
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}T.*"), // looks like a datestamp
                  startsWith("Pipeline-ID: "),
                  is("input(value: 10)\n-> filter(value > 8)"),
                  is(
"""
﻿name,runtime-ms,runtime-average-ms,tuples-in,tuples-in-per-sec,tuples-out,tuples-out-per-sec,context-switches
test,2000,,1,,,,
"""
                  ),
                  is("[canWriteSpatialData]")
              ),
              contains(
                  is("manifest-test2"),
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}T.*"), // looks like a datestamp
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}T.*"), // looks like a datestamp
                  startsWith("Pipeline-ID: "),
                  is("input(value: 10)\n-> filter(value > 8)"),
                  is(
"""
﻿name,runtime-ms,runtime-average-ms,tuples-in,tuples-in-per-sec,tuples-out,tuples-out-per-sec,context-switches
test,2000,,1,,,,
"""
                  ),
                  is("[canWriteSpatialData1, canWriteSpatialData2]")
              )
          ));
    }
  }

  @Test
  public void writesNewManifestEntryEvenWithReplaceTrue() {
    PipelineOutputOptions options = defaultOptions();
    options.setReference("writesNewManifestEntryEvenWithReplaceTrue");
    options.setReplace(true);

    // write 1, should get replaced
    try (PipelineOutputContainer container = getOutputContainer(outputPath, options).get()) {
      writeData(container, "writesNewManifestEntryEvenWithReplaceTrue", null, inputValue, inputValue, inputValue);
    }

    // write 2, should replace write 1 content
    try (PipelineOutputContainer container = getOutputContainer(outputPath, options).get()) {
      writeData(container, "writesNewManifestEntryEvenWithReplaceTrue", null, inputValue);
    }

    // table only has data from write 2
    assertData("writesNewManifestEntryEvenWithReplaceTrue", inputType, Lists.newArrayList(inputValue));

    // confirm both model runs have created a manifest entry
    try (JdbcHelper jdbcHelper = JdbcHelper.fromGeoPackage(new File(outputPath))) {
      assertThat(jdbcHelper.readTable(BaseJdbcPipelineOutputContainer.MANIFEST_TABLE, "reference"), contains(
              contains(
                  is("writesNewManifestEntryEvenWithReplaceTrue")
              ),
              contains(
                  is("writesNewManifestEntryEvenWithReplaceTrue")
              )
          ));
    }
  }

  @Test
  public void manifestWritingFailsIfExistingTableHasDifferentStructure() {
    try (PipelineOutputContainer container = getOutputContainer(outputPath).get()) {
      // we write our test data to the manifest table, then when the manifest is written it will need
      // to write it to an incremented table
      writeData(container, BaseJdbcPipelineOutputContainer.MANIFEST_TABLE, null, inputValue);
    }

    assertThat(sunkProblems, contains(Matchers.hasAncestorProblem(is(
        BaseJdbcOutputStore.PROBLEMS.cannotAppendTableStructureMismatch("riskscape_manifest",
            "riskscape_manifest(version TEXT, reference TEXT, started TEXT, finished TEXT, manifest TEXT, "
            + "pipeline TEXT, stats TEXT, output_tables TEXT)",
            "riskscape_manifest(geom POINT, index INTEGER, name TEXT)"
        ).withChildren(BaseJdbcOutputStore.PROBLEMS.cannotAppendTableHint())
    ))));
  }

  Callable<Void> buildWriteTask(String name, PipelineOutputContainer container) {
    return () -> {
      SaveSink sink = (SaveSink) container.createSinkForStep(new SinkParameters(name, inputType)
      ).get();
      for (Tuple data: getMultiThreadedTestData(name)) {
        sink.accept(data);
        try {
          Thread.sleep(10);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
      sink.finish();
      return null;
    };
  }

  List<Tuple> getMultiThreadedTestData(String name) {
    List<Tuple> data = new ArrayList<>();
    for (long i = 0L; i < 1000L; i++) {
      data.add(Tuple.ofValues(inputType, point1Crs84, i, name));
    }
    return data;
  }

  protected Relation getWrittenAsRelation(String layer, boolean skipInvalid) {
    Bookmark bookmark = new Bookmark("test", "", "", outputPath,
        ImmutableMap.<String, List<String>>builder()
            .put("layer", Arrays.asList(layer))
            //            .put("skip-invalid", String.valueOf(skipInvalid))
            .build());
    return resolver.resolve(bookmark, executionContext)
        .map(resolved -> resolved.getData(Relation.class).get())
        .orElse(null);
  }

  void assertData(String named, Struct expectedType, List<Tuple> expectedValues) {
    Relation relation = getWrittenAsRelation(named, false);
    Struct relationType = relation.getType();

    assertThat(relationType, StructMatchers.isStruct(Lists.newArrayList(
        StructMatchers.isStructMember("geom", TypeMatchers.isReferencedType(SRIDSet.EPSG4326_LONLAT, Types.POINT)),
        StructMatchers.isStructMember("index", Types.INTEGER),
        StructMatchers.isStructMember("name", Types.TEXT)
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples.size(), is(expectedValues.size()));

    for (int i = 0; i < expectedValues.size(); i++) {
      Tuple expected = expectedValues.get(i);
      Tuple found = tuples.get(i);
      assertThat((Geometry)found.fetch("geom"), GeometryMatchers.isGeometry((Geometry)expected.fetch("geom")));
      assertThat(found.toMap().get("index"), is(expected.toMap().get("index")));
      assertThat(found.fetch("name"), is(expected.toMap().get("name")));
    }
    relation.close();
  }

  @Test
  public void warnsWhenFormatIgnored() {
    PipelineOutputOptions options = defaultOptions();
    Format csv = new CsvFormat();
    options.setFormat(csv);

    List<Problem> warnings = new ArrayList<>();
    try (PipelineOutputContainer container1 = getOutputContainer(outputPath, options)
        .drainWarnings(p -> warnings.add(p))
        .get()) {
      assertThat(warnings, contains(
          is(OutputProblems.get().userSpecifiedFormatIgnored(csv, subject))
      ));
    }
  }

  @Test
  public void warnsWhenCheckSumIsIgnored() {
    PipelineOutputOptions options = defaultOptions();
    options.setChecksum(true);

    List<Problem> warnings = new ArrayList<>();
    try (PipelineOutputContainer container1 = getOutputContainer(outputPath, options)
        .drainWarnings(p -> warnings.add(p))
        .get()) {
      assertThat(warnings, contains(
          is(OutputProblems.get().checksumNotSupported(subject))
      ));
    }
  }

}
