view pkg/imports/st.go @ 5560:f2204f91d286

Join the log lines of imports to the log exports to recover data from them. Used in SR export to extract information that where in the meta json but now are only found in the log.
author Sascha L. Teichmann <sascha.teichmann@intevation.de>
date Wed, 09 Feb 2022 18:34:40 +0100
parents 59a99655f34d
children 6270951dda28
line wrap: on
line source

// 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) 2018 by via donau
//   – Österreichische Wasserstraßen-Gesellschaft mbH
// Software engineering by Intevation GmbH
//
// Author(s):
//  * Sascha L. Teichmann <sascha.teichmann@intevation.de>

package imports

import (
	"context"
	"database/sql"
	"errors"
	"time"

	"gemma.intevation.de/gemma/pkg/models"
)

// Stretch is a Job to create a stretch in the database.
type Stretch struct {
	Name      string                 `json:"name"`
	From      models.Isrs            `json:"from"`
	To        models.Isrs            `json:"to"`
	Tolerance float32                `json:"tolerance"`
	ObjNam    string                 `json:"objnam"`
	NObjNam   *string                `json:"nobjnam"`
	Source    string                 `json:"source-organization"`
	Date      models.Date            `json:"date-info"`
	Countries models.UniqueCountries `json:"countries"`
}

// STJobKind is the import queue type identifier.
const STJobKind JobKind = "st"

type stJobCreator struct{}

func init() { RegisterJobCreator(STJobKind, stJobCreator{}) }

func (stJobCreator) Description() string { return "stretch" }

func (stJobCreator) AutoAccept() bool { return false }

func (stJobCreator) Create() Job { return new(Stretch) }

func (stJobCreator) Depends() [2][]string {
	return [2][]string{
		{"stretches", "stretch_countries"},
		{"distance_marks_virtual", "waterway_axis", "waterway_area"},
	}
}

const (
	stDeleteSQL = `
DELETE FROM users.stretches WHERE
staging_done AND name IN (
  SELECT name
  FROM users.stretches WHERE
  id IN (
    SELECT key from import.track_imports
    WHERE import_id = $1 AND
      relation = 'users.stretches'::regclass)
  AND NOT staging_done
)`

	stStageDoneSQL = `
UPDATE users.stretches SET staging_done = true
WHERE id IN (
  SELECT key from import.track_imports
  WHERE import_id = $1 AND
        relation = 'users.stretches'::regclass)`

	stInsertSQL = `
WITH
bounds (b) AS (VALUES (
    ($1::char(2),
     $2::char(3),
     $3::char(5),
     $4::char(5),
     $5::int)::isrs
  ), (
    ($6::char(2),
     $7::char(3),
     $8::char(5),
     $9::char(5),
     $10::int)::isrs)),
r AS (SELECT isrsrange(
    (SELECT b FROM bounds ORDER BY b USING <~ FETCH FIRST ROW ONLY),
    (SELECT b FROM bounds ORDER BY b USING >~ FETCH FIRST ROW ONLY)) AS r),
axs AS (
  SELECT ISRSrange_axis((SELECT r FROM r), $16::double precision) AS axs)
INSERT INTO users.stretches (
  name,
  stretch,
  area,
  objnam,
  nobjnam,
  date_info,
  source_organization
) VALUES (
  $11,
  (SELECT r FROM r),
  ST_Transform(ISRSrange_area(
      (SELECT axs FROM axs),
      (SELECT ST_Buffer(axs, 10000) FROM axs)),
    4326),
  $12,
  $13,
  $14,
  $15)
RETURNING id`

	stInsertCountrySQL = `
INSERT INTO users.stretch_countries (
  stretch_id,
  country
) VALUES (
  $1,
  $2
)`
)

// StageDone moves the imported stretch out of the staging area.
func (stJobCreator) StageDone(
	ctx context.Context,
	tx *sql.Tx,
	id int64,
	_ Feedback,
) error {
	if _, err := tx.ExecContext(ctx, stDeleteSQL, id); err != nil {
		return err
	}
	_, err := tx.ExecContext(ctx, stStageDoneSQL, id)
	return err
}

// CleanUp of a stretch import is a NOP.
func (*Stretch) CleanUp() error { return nil }

// Do executes the actual stretch import.
func (st *Stretch) Do(
	ctx context.Context,
	importID int64,
	conn *sql.Conn,
	feedback Feedback,
) (interface{}, error) {

	start := time.Now()

	if st.Date.Time.IsZero() {
		st.Date = models.Date{Time: start}
	}

	feedback.Info("Storing stretch '%s'", st.Name)

	if len(st.Countries) == 0 {
		return nil, errors.New("list of countries is empty")
	}

	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	insertCountryStmt, err := tx.PrepareContext(ctx, stInsertCountrySQL)
	if err != nil {
		return nil, err
	}
	defer insertCountryStmt.Close()

	var nobjnm sql.NullString
	if st.NObjNam != nil {
		nobjnm = sql.NullString{String: *st.NObjNam, Valid: true}
	}

	feedback.Info("Stretch from %s to %s.", st.From.String(), st.To.String())
	feedback.Info("Tolerance used to snap waterway axis: %g", st.Tolerance)

	var id int64
	if err := tx.QueryRowContext(
		ctx,
		stInsertSQL,
		st.From.CountryCode,
		st.From.LoCode,
		st.From.FairwaySection,
		st.From.Orc,
		st.From.Hectometre,
		st.To.CountryCode,
		st.To.LoCode,
		st.To.FairwaySection,
		st.To.Orc,
		st.To.Hectometre,
		st.Name,
		st.ObjNam,
		nobjnm,
		st.Date.Time,
		st.Source,
		st.Tolerance,
	).Scan(&id); err != nil {
		return nil, err
	}

	// store the associated countries.

	feedback.Info("Countries associated with stretch: %s.", st.Countries)
	for _, c := range st.Countries {
		if _, err := insertCountryStmt.ExecContext(ctx, id, c); err != nil {
			return nil, err
		}
	}

	if err := track(ctx, tx, importID, "users.stretches", id); err != nil {
		return nil, err
	}

	feedback.Info("Storing stretch '%s' took %s", st.Name, time.Since(start))
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	feedback.Info("Import of stretch was successful")

	summary := st // provide full information for summary

	return summary, nil
}