--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/pkg/imports/fm_bcnlat.go	Thu Feb 06 19:09:14 2020 +0100
@@ -0,0 +1,237 @@
+// This is Free Software under GNU Affero General Public License v >= 3.0
+// without warranty, see README.md and license for details.
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+// License-Filename: LICENSES/AGPL-3.0.txt
+//
+// Copyright (C) 2020 by via donau
+//   – Österreichische Wasserstraßen-Gesellschaft mbH
+// Software engineering by Intevation GmbH
+//
+// Author(s):
+//  * Tom Gottfried <tom.gottfried@intevation.de>
+
+package imports
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"strconv"
+	"time"
+
+	"gemma.intevation.de/gemma/pkg/pgxutils"
+)
+
+// Bcnlat is an import job to import
+// fairway marks of type BCNLAT in form of point geometries
+// and attribute data from a WFS service.
+type Bcnlat struct {
+	FairwayMarks
+}
+
+// Description gives a short info about relevant facts of this import.
+func (bcnlat *Bcnlat) Description() (string, error) {
+	return bcnlat.URL + "|" + bcnlat.FeatureType, nil
+}
+
+// FMJobKind is the import queue type identifier.
+const FMJobKind JobKind = "fm"
+
+type bcnlatJobCreator struct{}
+
+func init() {
+	RegisterJobCreator(FMJobKind, bcnlatJobCreator{})
+}
+
+func (bcnlatJobCreator) Description() string { return "fairway marks bcnlat" }
+
+func (bcnlatJobCreator) AutoAccept() bool { return true }
+
+func (bcnlatJobCreator) Create() Job { return new(Bcnlat) }
+
+func (bcnlatJobCreator) Depends() [2][]string {
+	return [2][]string{
+		{"fairway_marks"},
+		{},
+	}
+}
+
+// StageDone is a NOP for fairway marks imports.
+func (bcnlatJobCreator) StageDone(context.Context, *sql.Tx, int64) error {
+	return nil
+}
+
+// CleanUp for fairway marks imports is a NOP.
+func (*Bcnlat) CleanUp() error { return nil }
+
+type bcnlatProperties struct {
+	fairwayMarksProperties
+	Colour      *string `json:"hydro_colour"`
+	Colpat      *string `json:"hydro_colpat"`
+	Condtn      *int    `json:"hydro_condtn"`
+	Bcnshp      *int    `json:"hydro_bcnshp"`
+	HydroCatlam *int64  `json:"hydro_catlam,omitempty"`
+	IENCCatlam  *int64  `json:"ienc_catlam,omitempty"`
+	Dirimp      *string `json:"ienc_dirimp,omitempty"`
+}
+
+const (
+	insertBCNLATSQL = `
+with a as (
+  select users.current_user_area_utm() AS a
+)
+INSERT INTO waterway.fairway_marks (
+  geom,
+  datsta,
+  datend,
+  persta,
+  perend,
+  objnam,
+  nobjnm,
+  inform,
+  ninfom,
+  scamin,
+  picrep,
+  txtdsc,
+  sordat,
+  sorind,
+  colour,
+  colpat,
+  condtn,
+  bcnshp,
+  catlam,
+  dirimp
+)
+SELECT newfm, $3, $4, $5, $6, $7, $8, $9,
+    $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21
+  FROM ST_Transform(ST_GeomFromWKB($1, $2::integer), 4326) AS newfm (newfm)
+  WHERE pg_has_role('sys_admin', 'MEMBER')
+    OR ST_Intersects((select a from a),
+      ST_Transform(newfm, (select ST_SRID(a) from a)))
+ON CONFLICT (
+    CAST((0, geom,
+      datsta, datend, persta, perend, objnam, nobjnm, inform, ninfom,
+      scamin, picrep, txtdsc, sordat, sorind, colour, colpat, condtn,
+      bcnshp, catlam, dirimp) AS waterway.fairway_marks)
+  )
+  DO NOTHING
+RETURNING id
+`
+)
+
+// Do executes the actual import.
+func (fm *Bcnlat) Do(
+	ctx context.Context,
+	importID int64,
+	conn *sql.Conn,
+	feedback Feedback,
+) (interface{}, error) {
+
+	start := time.Now()
+
+	feedback.Info("Import fairway marks of type BCNLAT/bcnlat")
+
+	var props bcnlatProperties
+	fms, epsg, err := getFMFeatures(
+		feedback,
+		fm.FairwayMarks,
+		&props,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	tx, err := conn.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	insertStmt, err := tx.PrepareContext(ctx, insertBCNLATSQL)
+	if err != nil {
+		return nil, err
+	}
+	defer insertStmt.Close()
+
+	savepoint := Savepoint(ctx, tx, "feature")
+
+	var outsideOrDup int
+
+	var features int
+	for _, fm := range fms {
+
+		p := fm[0].(pointSlice)
+		fp := fm[1].(*bcnlatProperties)
+
+		var catlam sql.NullInt64
+		if fp.HydroCatlam != nil {
+			catlam = sql.NullInt64{Int64: *fp.HydroCatlam, Valid: true}
+		} else if fp.IENCCatlam != nil {
+			catlam = sql.NullInt64{Int64: *fp.IENCCatlam, Valid: true}
+		}
+
+		var dirimp sql.NullInt64
+		if fp.Dirimp != nil {
+			if value, err := strconv.ParseInt(*fp.Dirimp, 10, 64); err == nil {
+				dirimp = sql.NullInt64{Int64: value, Valid: true}
+			}
+		}
+
+		var fmid int64
+		err := savepoint(func() error {
+			err := insertStmt.QueryRowContext(
+				ctx,
+				p.asWKB(),
+				epsg,
+				fp.Datsta,
+				fp.Datend,
+				fp.Persta,
+				fp.Perend,
+				fp.Objnam,
+				fp.Nobjnm,
+				fp.Inform,
+				fp.Ninfom,
+				fp.Scamin,
+				fp.Picrep,
+				fp.Txtdsc,
+				fp.Sordat,
+				fp.Sorind,
+				fp.Colour,
+				fp.Colpat,
+				fp.Condtn,
+				fp.Bcnshp,
+				catlam,
+				dirimp,
+			).Scan(&fmid)
+			return err
+		})
+		switch {
+		case err == sql.ErrNoRows:
+			outsideOrDup++
+			// ignore -> filtered by responsibility_areas
+		case err != nil:
+			feedback.Error(pgxutils.ReadableError{Err: err}.Error())
+		default:
+			features++
+		}
+	}
+
+	if outsideOrDup > 0 {
+		feedback.Info(
+			"Features outside responsibility area and duplicates: %d",
+			outsideOrDup)
+	}
+
+	if features == 0 {
+		err := errors.New("no valid new features found")
+		return nil, err
+	}
+
+	if err = tx.Commit(); err == nil {
+		feedback.Info("Storing %d features took %s",
+			features, time.Since(start))
+	}
+
+	return nil, err
+}