depot/go/trains/darwin/darwindb/ddbschedule.go

package darwindb

import (
	"context"
	"encoding/json"
	"fmt"
	"log"

	pgx "github.com/jackc/pgx/v4"
	"git.lukegb.com/lukegb/depot/go/trains/darwin"
)

type scheduleMode int

const (
	scheduleModeOverwrite scheduleMode = iota
	scheduleModeIgnoreExisting
)

// handleSchedule processes a Darwin Schedule.
func handleSchedule(ctx context.Context, tx pgx.Tx, s *darwin.Schedule, a *Affected, sm scheduleMode) error {
	startTS, err := startTime(s)
	if err != nil {
		return fmt.Errorf("finding start time for RID %v: %w", s.RID, err)
	}

	var cancelCode, cancelTIPLOC *string
	var cancelTIPLOCNear *bool
	if s.CancellationReason != nil {
		cancelCode = &s.CancellationReason.Code
		cancelTIPLOC = &s.CancellationReason.TIPLOC
		cancelTIPLOCNear = &s.CancellationReason.Near
	}
	sql := `
INSERT INTO train_services
(rid, uid, rsid, headcode, scheduled_start_date, effective_start_time,
 train_operator, status, train_category, is_passenger_svc, is_charter, is_q_train,
 cancellation_reason_code, cancellation_reason_tiploc, cancellation_reason_tiploc_near,
 active, deleted, cancelled)
VALUES
($1, $2, $3, $4, $5, $6,
 $7, COALESCE($8, 'P'), COALESCE($9, 'OO'), COALESCE($10, true), COALESCE($11, false), COALESCE($12, false),
 $13, $14, $15,
 COALESCE($16, true), $17, COALESCE($18, false))
ON CONFLICT (rid)`
	switch sm {
	case scheduleModeOverwrite:
		sql += ` DO UPDATE SET
uid=$2, rsid=$3, headcode=$4, scheduled_start_date=$5, effective_start_time=$6,
train_operator=$7, status=COALESCE($8, 'P'), train_category=COALESCE($9, 'OO'), is_passenger_svc=COALESCE($10, true), is_charter=COALESCE($11, false), is_q_train=COALESCE($12, train_services.is_q_train),
cancellation_reason_code=$13, cancellation_reason_tiploc=$14, cancellation_reason_tiploc_near=$15,
active=COALESCE($16, 'true'), deleted=$17, cancelled=COALESCE($18, 'false')
`
	case scheduleModeIgnoreExisting:
		sql += ` DO NOTHING
`
	}
	sql += `RETURNING id`
	row := tx.QueryRow(ctx, sql,
		s.RID, s.UID, s.RSID, s.TrainID, s.SSD, startTS,
		s.TOC, s.Status, s.TrainCat, s.IsPassengerSvc, s.IsCharter, s.IsQTrain,
		cancelCode, cancelTIPLOC, cancelTIPLOCNear,
		s.IsActive, s.Deleted, s.Cancelled)
	var serviceID int
	if err := row.Scan(&serviceID); err == pgx.ErrNoRows && sm == scheduleModeIgnoreExisting {
		// The row existed already.
		return nil
	} else if err != nil {
		log.Printf("inserting schedule for RID %v: %v", s.RID, err)
		return fmt.Errorf("inserting schedule for RID %v: %w", s.RID, err)
	}
	a.RID(s.RID)
	a.TSID(serviceID)

	// Insert/update all the locations.
	var activeLocations []int
	for _, cp := range s.CallingPoints {
		var locid int
		row := tx.QueryRow(ctx, `
SELECT train_locations_upsert_from_schedule(
p_tsid => $1, p_rid => $2, p_tiploc => $3, p_calling_point => $4,
p_schedule_public_arrival => $5, p_schedule_working_arrival => $6,
p_schedule_public_departure => $7, p_schedule_working_departure => $8,
p_schedule_working_pass => $9,
p_schedule_route_delay => $10, p_schedule_activity => $11, p_schedule_planned_activity => $12, p_schedule_cancelled => $13,
p_schedule_false_destination_tiploc => $14, p_schedule_platform => $15)
`,
			serviceID, s.RID, cp.TIPLOC, cp.XMLName.Local,
			timeToTimestamp(cp.PTA, startTS), timeToTimestamp(cp.WTA, startTS),
			timeToTimestamp(cp.PTD, startTS), timeToTimestamp(cp.WTD, startTS),
			timeToTimestamp(cp.WTP, startTS),
			cp.RDelay, cp.Act, cp.PlanAct, cp.Cancelled,
			cp.FD, cp.Platform,
		)
		if err := row.Scan(&locid); err != nil {
			log.Printf("updating location %v from RID %v/my ID %v: %v", cp.TIPLOC, s.RID, serviceID, err)
			zz, _ := json.MarshalIndent(s, "", "  ")
			log.Println(string(zz))
			zz, _ = json.MarshalIndent(cp, "", "  ")
			log.Println(string(zz))
			return fmt.Errorf("updating location %v from RID %v/my ID %v: %w", cp.TIPLOC, s.RID, serviceID, err)
		} else {
			activeLocations = append(activeLocations, locid)
		}
		a.TIPLOC(cp.TIPLOC)
	}
	if _, err := tx.Exec(ctx, "UPDATE train_locations SET active_in_schedule=false WHERE tsid=$1 AND NOT (id = ANY ($2))", serviceID, activeLocations); err != nil {
		return fmt.Errorf("marking old locations as removed from schedule for RID %v/my ID %v (active locations: %v): %w", s.RID, serviceID, activeLocations, err)
	}
	fmt.Printf("s")
	return nil
}