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

import static nz.org.riskscape.engine.Matchers.*;
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.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.locationtech.jts.geom.LineString;
import org.locationtech.jts.geom.MultiLineString;
import org.locationtech.jts.geom.MultiPoint;
import org.locationtech.jts.geom.MultiPolygon;
import org.locationtech.jts.geom.Point;
import org.locationtech.jts.geom.Polygon;

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.Tuple;
import nz.org.riskscape.engine.TupleMatchers;
import nz.org.riskscape.engine.TypeMatchers;
import nz.org.riskscape.engine.cli.PostgisIntegrationTestRunner;
import nz.org.riskscape.engine.coverage.TypedCoverage;
import nz.org.riskscape.engine.data.Bookmark;
import nz.org.riskscape.engine.defaults.data.jdbc.BaseJdbcOutputStore;
import nz.org.riskscape.engine.gt.GeometryHelper;
import nz.org.riskscape.engine.output.BaseJdbcPipelineOutputContainer;
import nz.org.riskscape.engine.output.BasePipelineOutputStoreTest;
import nz.org.riskscape.engine.output.CsvFormat;
import nz.org.riskscape.engine.output.Format;
import nz.org.riskscape.engine.output.PipelineOutputContainer;
import nz.org.riskscape.engine.output.PipelineOutputOptions;
import nz.org.riskscape.engine.output.SinkParameters;
import nz.org.riskscape.engine.pipeline.Sink;
import nz.org.riskscape.engine.pipeline.sink.SaveSink;
import nz.org.riskscape.engine.relation.Relation;
import nz.org.riskscape.engine.types.CoverageType;
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;
import nz.org.riskscape.problem.ResultOrProblems;
import nz.org.riskscape.rl.ast.StructDeclaration;


@Category(PostGISIntegrationTestMarker.class)
@RunWith(PostgisIntegrationTestRunner.class)
public class PostGISPipelineOutputStoreTest extends BasePipelineOutputStoreTest {

  public static final int PORT =  PostgisIntegrationTestRunner.PUBLISHED_PORT;

  GeometryHelper nztmHelper = GeometryHelper.nzTransverseMercatorWithOriginShift(project.getSridSet());

  URI locationWithPassword = URI.create("postgis://riskscape:riskscape@localhost:" + PORT + "/riskscape");
  String jdbcUrl = "jdbc:postgresql://localhost:55432/riskscape?user=riskscape&password=riskscape";

  PostGISResolver resolver = new PostGISResolver(engine);

  public PostGISPipelineOutputStoreTest() {
    super(new PostGISPipelineOutputStore());
  }

  @Test
  public void isApplicableForURIWithPostgisScheme() {
    assertThat(subject.isApplicable(URI.create("postgis://host")), is(PRIORITY_HIGH));
    assertThat(subject.isApplicable(URI.create("postgis://host:5432/db")), is(PRIORITY_HIGH));
    assertThat(subject.isApplicable(URI.create("postgis://host:5432/db?username=bob")), is(PRIORITY_HIGH));
    assertThat(subject.isApplicable(URI.create("postgis://host:5432/db?username=bob&password=mary")),
        is(PRIORITY_HIGH));
    assertThat(subject.isApplicable(URI.create("postgis://host")), is(PRIORITY_HIGH));

    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 canWriteNonSpatialData() throws Exception {
    try (PipelineOutputContainer container = getOutputContainer()) {
      writeNonSpatialData(container, "canWriteNonSpatialData", null);
    }

    // The PostGISResolver doesn't work for non-spatial data
//    Relation relation = getWrittenAsRelation("canWriteNonSpatialData", false);
//
//    assertThat(relation.getType(), is(inputType));
//    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
//    assertThat(tuples, contains(
//        Tuple.ofValues(inputType, "foo1", 100L, 12.5D),
//        Tuple.ofValues(inputType, "foo2", 200L, 22.3D)
//    ));
  }

  @Test
  public void canWriteSpatialData() throws Exception {
    try (PipelineOutputContainer container = getOutputContainer()) {
      writeSpatialData(container, "canWriteSpatialData", null);
    }

    assertSpatialData("canWriteSpatialData");
  }

  @Test
  public void incrementsTableNameOnDuplicates() {
    try (PipelineOutputContainer container = getOutputContainer()) {
      writeSpatialData(container, "incrementsFileOnDuplicateFiles", null);
      writeSpatialData(container, "incrementsFileOnDuplicateFiles", null);
    }
    assertSpatialData("incrementsFileOnDuplicateFiles");
    assertSpatialData("incrementsFileOnDuplicateFiles_1");
  }

  @Test
  public void willReplaceTablesWithReplaceTrue() {
    PipelineOutputOptions options = defaultOptions();
    options.setReplace(true);
    try (PipelineOutputContainer container = getOutputContainer(locationWithPassword, options)
        .getWithProblemsIgnored()) {
      writeSpatialData(container, "willReplaceTablesWithReplaceTrue", null);
      writeSpatialData(container, "willReplaceTablesWithReplaceTrue", null);
      writeData(container, "willReplaceTablesWithReplaceTrue", null, Tuple.ofValues(spatialType, point1, "replaced"));
    }
    // assert tha the replaced table only has the content from the last write
    assertSpatialData("willReplaceTablesWithReplaceTrue", "replaced");
  }

  @Test
  public void willWriteTableWithSpacesInName() {
    PipelineOutputOptions options = defaultOptions();
    options.setReplace(true);
    try (PipelineOutputContainer container = getOutputContainer(locationWithPassword, options)
        .getWithProblemsIgnored()) {
      SaveSink sink = writeSpatialData(container, "will write table with spaces", null);
      // the stored at URI contains the written table name and it is URL encoded (spaces -> +)
      assertThat(sink.getStoredAt().toString(), endsWith("?layer=will+write+table+with+spaces"));
    }
    assertSpatialData("will write table with spaces");
  }

  @Test
  public void writesAllGeometriesAsGeometry() {
    Struct inputType = Struct.of("geom1", Referenced.of(Types.POINT, latLongHelper.getCrs()),
        "geom2", Referenced.of(Types.POINT, crs84Helper.getCrs()));
    Point crs84Point = crs84Helper.point(30, 20);
    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("writesAllGeometriesAsGeometry", inputType)).get();
      sink.accept(Tuple.ofValues(inputType, point1, crs84Point));
      sink.finish();
    }

    Relation relation = getWrittenAsRelation("writesAllGeometriesAsGeometry", false);
    Struct relationType = relation.getType();

    assertThat(relationType, StructMatchers.isStruct(Lists.newArrayList(
        StructMatchers.isStructMember("geom1", TypeMatchers.isReferencedType(SRIDSet.EPSG4326_LONLAT, Types.POINT)),
        StructMatchers.isStructMember("geom2", Types.GEOMETRY)  // We are missing the referencing, but this is from the
                                                                // reading side. Not the writing.
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relation.getType(), crs84Helper.reproject(point1), crs84Point)
    ));
  }

  @Test
  public void flattensNestedData() {
    Struct inputType = Struct.of("c1", spatialType, "c2", spatialType);

    Point point2 = latLongHelper.point(178, -40);
    int latLongSRID = point2.getSRID();
    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("flattensNestedData", inputType)).get();
      sink.accept(Tuple.ofValues(inputType,
          Tuple.ofValues(spatialType, point1, "c1"),
          Tuple.ofValues(spatialType, point2, "c2")
      ));
      sink.finish();
    }

    // Ensure that the writing has not changed the input geom
    assertThat(point2.getX(), is(178D));
    assertThat(point2.getY(), is(-40D));
    assertThat(point2.getSRID(), is(latLongSRID));

    Relation relation = getWrittenAsRelation("flattensNestedData", false);
    Struct relationType = relation.getType();

    assertThat(relationType, StructMatchers.isStruct(Lists.newArrayList(
        StructMatchers.isStructMember("geom", TypeMatchers.isReferencedType(SRIDSet.EPSG4326_LONLAT, Types.POINT)),
        StructMatchers.isStructMember("foo", Types.TEXT),
        StructMatchers.isStructMember("geom_1", Types.GEOMETRY),
        StructMatchers.isStructMember("foo_1", Types.TEXT)
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relation.getType(), crs84Helper.reproject(point1), "c1",
            crs84Helper.reproject(point2), "c2")
    ));
  }

  @Test
  public void doesNotFlipAxisIfAlreadyXY() {
    Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, crs84Helper.getCrs()), "foo", Types.TEXT);
    Point crs84Point = crs84Helper.point(20, 60);
    int crs84SRID = crs84Point.getSRID();
    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(
          new SinkParameters("doesNotFlipAxisIfAlreadyXY", inputType)).get();
      sink.accept(Tuple.ofValues(inputType, crs84Point, "bar"));
      sink.finish();
    }

    // ensure that the writing has not changed the input geom
    assertThat(crs84Point.getX(), is(20D));
    assertThat(crs84Point.getY(), is(60D));
    assertThat(crs84Point.getSRID(), is(crs84SRID));

    Relation relation = getWrittenAsRelation("doesNotFlipAxisIfAlreadyXY", false);
    Struct relationType = relation.getType();

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

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relation.getType(), crs84Point, "bar")
    ));
  }

  @Test
  public void writesAllSupportedTypes() {
    Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
        "bool", Types.BOOLEAN,
        "float", Types.FLOATING)
        .and("integer", Types.INTEGER)
        .and("date", Types.DATE)
        .and("text", Types.TEXT);

    Date testDate = new Date();
    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("writesAllSupportedTypes", inputType)).get();
      sink.accept(Tuple.ofValues(inputType, point1, true, 12.32D, 25L, testDate, "foo"));
      sink.finish();
    }
    Relation relation = getWrittenAsRelation("writesAllSupportedTypes", false);
    Struct relationType = relation.getType();

    assertThat(relationType, StructMatchers.isStruct(Lists.newArrayList(
        StructMatchers.isStructMember("geom", TypeMatchers.isReferencedType(SRIDSet.EPSG4326_LONLAT, Types.POINT)),
        StructMatchers.isStructMember("bool", Types.BOOLEAN),
        StructMatchers.isStructMember("float", Types.FLOATING),
        StructMatchers.isStructMember("integer", Types.INTEGER),
        StructMatchers.isStructMember("date", Types.DATE),
        StructMatchers.isStructMember("text", Types.TEXT)
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relation.getType(), crs84Helper.reproject(point1), true, 12.32D, 25L, testDate, "foo")
    ));
  }

  @Test
  public void canWriteAMixedBagOfGeometries() throws Exception {
    // in this test we are showing that GeoPackages let you mix and match geometry types in the same layer.
    // we are also testing that each kind of geometry can be written and read back in as well.
    Struct inputType = Struct.of("geom", Referenced.of(Types.GEOMETRY, latLongHelper.getCrs()), "text", Types.TEXT);

    Point point = latLongHelper.point(10, 10);
    LineString line = latLongHelper.line(10, 10, 20, 20);
    Polygon box = latLongHelper.box(10, 10, 20, 20);

    MultiPoint mpoint = latLongHelper.multiPoint(
        latLongHelper.point(10, 20),
        latLongHelper.point(20, 20)
    );
    MultiLineString mline = latLongHelper.multiLine(
        latLongHelper.line(0, 0, 0, 20),
        latLongHelper.line(10, 10, 10, 20)
    );
    MultiPolygon mbox = latLongHelper.multiBox(
        latLongHelper.box(15, 15, 20, 20),
        latLongHelper.box(0,0, 5, 5)
    );

    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("canWriteAMixedBagOfGeometries", inputType)).get();
      sink.accept(Tuple.ofValues(inputType, point, "point"));
      sink.accept(Tuple.ofValues(inputType, line, "line"));
      sink.accept(Tuple.ofValues(inputType, box, "box"));
      sink.accept(Tuple.ofValues(inputType, mpoint, "mpoint"));
      sink.accept(Tuple.ofValues(inputType, mline, "mline"));
      sink.accept(Tuple.ofValues(inputType, mbox, "mbox"));
      sink.finish();
    }

    Relation relation = getWrittenAsRelation("canWriteAMixedBagOfGeometries", false);
    Struct relationType = relation.getType();

    assertThat(relationType, StructMatchers.isStruct(Lists.newArrayList(
        StructMatchers.isStructMember("geom", TypeMatchers.isReferencedType(SRIDSet.EPSG4326_LONLAT, Types.GEOMETRY)),
        StructMatchers.isStructMember("text", Types.TEXT)
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(point))),
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(line))),
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(box))),
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(mpoint))),
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(mline))),
        TupleMatchers.tupleWithValue("geom", GeometryMatchers.isGeometry(crs84Helper.reproject(mbox)))
    ));
  }

  @Test
  public void supportsMultipleWritersConcurrently() throws Exception {
    Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
        "foo", Types.TEXT, "bar", Types.INTEGER);

    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink1 = container.createSinkForStep(new SinkParameters("concurrent1", inputType)).get();
      Sink sink2 = container.createSinkForStep(new SinkParameters("concurrent2", inputType)).get();

      for (int i = 0; i < 3; i++) {
        sink1.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo" + i, i));
        sink2.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo" + i, i));
      }
      sink1.finish();
      sink2.finish();
    }

    List<Tuple> tuples1 = getWrittenAsRelation("concurrent1", false).iterator().collect(Collectors.toList());
    List<Tuple> tuples2 = getWrittenAsRelation("concurrent2", false).iterator().collect(Collectors.toList());

    assertThat(tuples1, hasSize(3));
    assertThat(tuples2, hasSize(3));
  }

  @Test
  public void canWriteFullRangeOfNumericValues() {
    Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
        "integer", Types.INTEGER,
        "floating", Types.FLOATING);

    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("canWriteFullRangeOfNumericValues", inputType)).get();
      sink.accept(Tuple.ofValues(inputType, point1, Long.MAX_VALUE, Double.MAX_VALUE));
      sink.accept(Tuple.ofValues(inputType, point1, Long.MIN_VALUE, Double.MIN_VALUE));
      sink.finish();
    }

    Relation relation = getWrittenAsRelation("canWriteFullRangeOfNumericValues", false);
    Struct relationType = relation.getType();

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relationType, crs84Helper.reproject(point1), Long.MAX_VALUE, Double.MAX_VALUE),
        Tuple.ofValues(relationType, crs84Helper.reproject(point1), Long.MIN_VALUE, Double.MIN_VALUE)
    ));
  }

  @Test
  public void writesUnsupportedTypesAsText() throws Exception {
    Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
        "foo", CoverageType.WILD);
    TypedCoverage coverage = TypedCoverage.empty(inputType);

    try (PipelineOutputContainer container = getOutputContainer()) {

      Sink sink = container.createSinkForStep(new SinkParameters("writesUnsupportedTypesAsText", inputType))
          .drainWarnings(warning -> sunkProblems.add(warning))
          .get();
      sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), coverage));
      sink.finish();
    }
    assertThat(sunkProblems, contains(
        OutputProblems.get().outputTypeAsText("foo", CoverageType.WILD, subject)
    ));

    Relation relation = getWrittenAsRelation("writesUnsupportedTypesAsText", false);
    Struct relationType = relation.getType();

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

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        TupleMatchers.tupleWithValue("foo", is(coverage.toString()))
    ));
  }

  @Test @Ignore("Example code, not really behaviour we want to assert")
  public void writtingToTheSameTableConcurrently() throws Exception {
    // Databases can be accessed by many concurrent users. So what happens if there are two RiskScapes
    // writting to the same table.

    // Make a second store with replace=true
    PipelineOutputOptions options2 = defaultOptions();
    options2.setReplace(true);
    try (PipelineOutputContainer container1 = getOutputContainer();
        PipelineOutputContainer container2 = getOutputContainer(locationWithPassword, options2)
            .getWithProblemsIgnored()) {

      Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
          "foo", Types.TEXT, "bar", Types.INTEGER);

      Sink sink1 = container1.createSinkForStep(new SinkParameters("write-same-table-concurrently", inputType)).get();
      Sink sink2 = container2.createSinkForStep(new SinkParameters("write-same-table-concurrently", inputType)).get();

      sink1.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo1", 100L));
      sink2.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo1", 100L));
      sink1.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 20), "foo2", 200L));
      sink2.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 20), "foo2", 200L));
      sink1.finish();
      sink2.finish();
    }

    Relation relation = getWrittenAsRelation("write-same-table-concurrently", false);

    // So in this test both writers have written to the same table which now contains both sets of results.
    // This example is very contrived. It only works because both writers are set up before the writing starts.
    // If first writer has already written then store2.writerFor() will hang when tring to delete the
    // existing table.
    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        TupleMatchers.tupleWithValue("foo", is("foo1")),
        TupleMatchers.tupleWithValue("foo", is("foo1")),
        TupleMatchers.tupleWithValue("foo", is("foo2")),
        TupleMatchers.tupleWithValue("foo", is("foo2"))
    ));
  }

  @Test @Ignore("Example code, not really behaviour we want to assert")
  public void writtingToTheSameTableConcurrently1_SecondWriterBlocksOnTableDelete() throws Exception {
    // Databases can be accessed by many concurrent users. So what happens if there are two RiskScapes
    // writting to the same table.
    // This tests has a second task that creates a writer while the first task is writing tuples.
    // The second task is blocked until the first finishes, but then it will replace the table.

    // Make a second store with replace=true
    PipelineOutputOptions options2 = defaultOptions();
    options2.setReplace(true);
    try (PipelineOutputContainer container1 = getOutputContainer();
        PipelineOutputContainer container2 = getOutputContainer(locationWithPassword, options2)
            .getWithProblemsIgnored()) {

      Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
          "foo", Types.TEXT, "bar", Types.INTEGER);

      List<Callable<Void>> tasks = new ArrayList<>();
      tasks.add(() -> {
        Sink sink = container1.createSinkForStep(new SinkParameters("write-same-table-concurrently1", inputType)).get();

        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo1-1", 100L));
        Thread.sleep(1000);
        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 20), "foo2-1", 200L));
        sink.finish();
        return null;
      });
      tasks.add(() -> {
        Thread.sleep(500);  // give the other task a head start, enougth that is will have started writing
        // tuples before we start. We expect store2.writerFor() to block on the table
        // delete until store1 has finished writing, before carrying on.
        Sink sink = container2.createSinkForStep(new SinkParameters("write-same-table-concurrently1", inputType)).get();

        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo1-2", 100L));
        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 20), "foo2-2", 200L));
        sink.finish();
        return null;
      });

      executeTasks(tasks, 2);
    }

    Relation relation = getWrittenAsRelation("write-same-table-concurrently1", false);

    // we get the results from the second writer task only.
    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        TupleMatchers.tupleWithValue("foo", is("foo1-2")),
        TupleMatchers.tupleWithValue("foo", is("foo2-2"))
    ));
  }

  @Test @Ignore("Example code, not really behaviour we want to assert")
  public void writtingToTheSameTableConcurrently2_SecondWriterGetsSQLException() throws Exception {
    // Databases can be accessed by many concurrent users. So what happens if there are two RiskScapes
    // writting to the same table.
    // This test has two tasks that are trying to create the same table concurrently.

    // Make a second store with replace=true
    PipelineOutputOptions options2 = defaultOptions();
    options2.setReplace(true);
    try (PipelineOutputContainer container1 = getOutputContainer();
        PipelineOutputContainer container2 = getOutputContainer(locationWithPassword, options2)
            .getWithProblemsIgnored()) {

      Struct inputType = Struct.of("geom", Referenced.of(Types.POINT, latLongHelper.getCrs()),
          "foo", Types.TEXT, "bar", Types.INTEGER);

      List<Callable<Void>> tasks = new ArrayList<>();
      AtomicReference<ResultOrProblems<Sink>> failedWriter = new AtomicReference<>();
      tasks.add(() -> {
        Sink sink = container1.createSinkForStep(new SinkParameters("write-same-table-concurrently2", inputType)).get();

        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 10), "foo1-1", 100L));
        Thread.sleep(1000);
        sink.accept(Tuple.ofValues(inputType, latLongHelper.point(10, 20), "foo2-1", 200L));
        sink.finish();
        return null;
      });
      tasks.add(() -> {
        Thread.sleep(20);   // give the other task a head start, but not enough to start writing in.
        // This should result in both task trying to create the table at the same time
        // so this task should fail.
        ResultOrProblems<Sink> sink
            = container2.createSinkForStep(new SinkParameters("write-same-table-concurrently2", inputType));
        failedWriter.set(sink);

        return null;
      });

      executeTasks(tasks, 2);

      // we expect the task2 writer to have failed, assert that now
      assertTrue(failedWriter.get().hasErrors());
      String problemText = render(failedWriter.get().getAsSingleProblem());
      assertThat(problemText,
          either(containsString("relation \"write-same-table-concurrently2\" already exists"))
              .or(containsString("duplicate key value violates unique constraint")
              ));
    }

    Relation relation = getWrittenAsRelation("write-same-table-concurrently2", false);

    // we get the results from the first writer task only (the second task failed)
    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        TupleMatchers.tupleWithValue("foo", is("foo1-1")),
        TupleMatchers.tupleWithValue("foo", is("foo2-1"))
    ));
  }

  @Test
  public void savesManifestWithPipelineAndStats() {
    // before we start lets delete any pre-existing manifest tables. we need to do this because the DB
    // is started before all of the tests. so there could be a table with existing records before we do
    // this test.
    try (JdbcHelper jdbcHelper = JdbcHelper.fromJdbcUrl(jdbcUrl)) {
      jdbcHelper.dropTable(BaseJdbcPipelineOutputContainer.MANIFEST_TABLE);
    }

    // we need to populate some metrics
    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(locationWithPassword, options)
        .getWithProblemsIgnored()) {
      writeSpatialData(container, "savesManifestWithPipelineAndStats1", null);
    }
    options.setReference("manifest-test2");
    try (PipelineOutputContainer container = getOutputContainer(locationWithPassword, options)
        .getWithProblemsIgnored()) {
      writeSpatialData(container, "savesManifestWithPipelineAndStats2", null);
      writeSpatialData(container, "savesManifestWithPipelineAndStats3", null);
    }

    try (JdbcHelper jdbcHelper = JdbcHelper.fromJdbcUrl(jdbcUrl)) {
      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}.*"), // looks like a datestamp
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}.*"), // 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("{savesManifestWithPipelineAndStats1}")
              ),
              contains(
                  is("manifest-test2"),
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}.*"), // looks like a datestamp
                  matchesPattern("^[0-9]{4}-[0-9]{2}-[0-9]{2}.*"), // 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("{savesManifestWithPipelineAndStats2,savesManifestWithPipelineAndStats3}")
              )
          ));
    }
  }

  @Test
  public void manifestWritingFailsIfExistingTableHasDifferentStructure() {
    // before we start lets delete any pre-existing manifest tables. we need to do this because the DB
    // is started before all of the tests. so there could be a table with existing records before we do
    // this test.
    try (JdbcHelper jdbcHelper = JdbcHelper.fromJdbcUrl(jdbcUrl)) {
      jdbcHelper.dropTable(BaseJdbcPipelineOutputContainer.MANIFEST_TABLE);
    }

    try (PipelineOutputContainer container = getOutputContainer()) {
      // 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
      writeSpatialData(container, BaseJdbcPipelineOutputContainer.MANIFEST_TABLE, null);
    }
    assertThat(sunkProblems, contains(Matchers.hasAncestorProblem(is(
        BaseJdbcOutputStore.PROBLEMS.cannotAppendTableStructureMismatch("riskscape_manifest",
            "riskscape_manifest(version TEXT, reference TEXT, started TIMESTAMP, finished TIMESTAMP, manifest TEXT, "
                + "pipeline TEXT, stats TEXT, output_tables TEXT[])",
            "riskscape_manifest(geom geometry, foo text)"
        ).withChildren(BaseJdbcOutputStore.PROBLEMS.cannotAppendTableHint())
    ))));

    // we need to clean up the bogus manifest table else other tests may fail because the database is
    // up for all of the tests
    try (JdbcHelper jdbcHelper = JdbcHelper.fromJdbcUrl(jdbcUrl)) {
      jdbcHelper.dropTable(BaseJdbcPipelineOutputContainer.MANIFEST_TABLE);
    }
  }

  @Test
  public void warningOnIgnoredDefaultFormat() {
    List<Problem> warnings = new ArrayList<>();
    PipelineOutputOptions options = defaultOptions();
    Format csvFormat = new CsvFormat();
    options.setFormat(csvFormat);
    try (PipelineOutputContainer container1 = getOutputContainer(locationWithPassword, options)
        .drainWarnings(p -> warnings.add(p))
        .get()) {
      assertThat(warnings, contains(
          OutputProblems.get().userSpecifiedFormatIgnored(csvFormat, subject),
          PostGISProblems.get().passwordInClearText()
      ));

    }
  }

  @Test
  public void warningIfSaveFormatIgnored() {
    List<Problem> warnings = new ArrayList<>();
    Format csvFormat = new CsvFormat();
    try (PipelineOutputContainer container = getOutputContainer()) {
      Sink sink = container.createSinkForStep(
          new SinkParameters("warningIfSaveFormatIgnored", spatialType, Optional.of(csvFormat),
              StructDeclaration.EMPTY))
          .drainWarnings(p -> warnings.add(p))
          .get();
      assertThat(warnings, contains(
          OutputProblems.get().userSpecifiedFormatIgnored(csvFormat, subject)
      ));

    }
  }

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

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

  @Test
  public void errorIfDbHostNotReachable() {
    assertThat(getOutputContainer(URI.create("postgis://riskscape@unknown-host:" + PORT + "/my-db")),
        Matchers.failedResult(Matchers.hasAncestorProblem(
            isProblem(Problem.Severity.ERROR, PostGISProblems.class, "connectionFailure")
        )));
  }

  @Test
  public void errorIfDbDoesNotExist() {
    assertThat(getOutputContainer(URI.create("postgis://riskscape@localhost:" + PORT + "/bad-db")),
        Matchers.failedResult(Matchers.hasAncestorProblem(
            isProblem(Problem.Severity.ERROR, PostGISProblems.class, "connectionFailure")
        )));
  }

  @Test
  public void errorIfDbUserDoesNotExist() {
    assertThat(getOutputContainer(URI.create("postgis://bad-user@localhost:" + PORT + "/riskscape")),
        Matchers.failedResult(Matchers.hasAncestorProblem(
            isProblem(Problem.Severity.ERROR, PostGISProblems.class, "connectionFailure")
        )));
  }

  void assertSpatialData(String named) {
    assertSpatialData(named, "foo");
  }

  void assertSpatialData(String named, String expectedFooValue) {
    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("foo", Types.TEXT)
    )));

    List<Tuple> tuples = relation.iterator().collect(Collectors.toList());
    assertThat(tuples, contains(
        Tuple.ofValues(relation.getType(), crs84Helper.reproject(point1), expectedFooValue)
    ));
  }

  Relation getWrittenAsRelation(String layer, boolean skipInvalid) {
    Bookmark bookmark = new Bookmark("test", "", "", locationWithPassword,
        ImmutableMap.of("layer", Arrays.asList(layer)));

    return resolver.resolve(bookmark, executionContext)
        .map(resolved -> resolved.getData(Relation.class).get())
        .orElse(null);
  }

  private PipelineOutputContainer getOutputContainer() {
    // this produces a warning because the password is in plain-text, which we ignore
    return getOutputContainer(locationWithPassword).getWithProblemsIgnored();
  }
}
