weatherfeeder/internal/normalizers/spc/convective_outlook.go

package spc

import (
	"context"
	"encoding/json"
	"fmt"
	"math"
	"regexp"
	"sort"
	"strings"
	"time"

	"gitea.maximumdirect.net/ejr/feedkit/event"
	"gitea.maximumdirect.net/ejr/weatherfeeder/internal/geo"
	normcommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/common"
	spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc"
	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
)

const (
	providerSPC        = "spc"
	productConvective  = "convective"
	outlookNormalizer  = "spc convective outlook"
	outlookTypeUnknown = 99
)

var idTokenRE = regexp.MustCompile(`[^a-z0-9]+`)

// ConvectiveOutlookNormalizer converts:
//
//	standards.SchemaRawSPCConvectiveOutlookV1 -> standards.SchemaWeatherOutlookV1
//
// It maps SPC GeoJSON outlook features into canonical outlook polygons and
// enriches each day with the matching required print-page discussion.
type ConvectiveOutlookNormalizer struct{}

func (ConvectiveOutlookNormalizer) Match(e event.Event) bool {
	return strings.TrimSpace(e.Schema) == standards.SchemaRawSPCConvectiveOutlookV1
}

func (ConvectiveOutlookNormalizer) Normalize(ctx context.Context, in event.Event) (*event.Event, error) {
	_ = ctx

	fallbackAsOf := in.EmittedAt.UTC()
	if in.EffectiveAt != nil && !in.EffectiveAt.IsZero() {
		fallbackAsOf = in.EffectiveAt.UTC()
	}

	return normcommon.NormalizeJSON(
		in,
		outlookNormalizer,
		standards.SchemaWeatherOutlookV1,
		func(parsed spcprovider.RawConvectiveOutlookBundle) (model.WeatherOutlookRun, time.Time, error) {
			return buildConvectiveOutlook(parsed, fallbackAsOf)
		},
	)
}

func buildConvectiveOutlook(bundle spcprovider.RawConvectiveOutlookBundle, fallbackAsOf time.Time) (model.WeatherOutlookRun, time.Time, error) {
	if err := validateCoordinates(bundle.Latitude, bundle.Longitude); err != nil {
		return model.WeatherOutlookRun{}, time.Time{}, err
	}

	discussions, latestDiscussionUpdated, err := parseDiscussions(bundle.Discussions)
	if err != nil {
		return model.WeatherOutlookRun{}, time.Time{}, err
	}

	products := orderedProducts(bundle.Products)
	point := geo.Point{Latitude: bundle.Latitude, Longitude: bundle.Longitude}
	outlooks := make([]model.WeatherOutlook, 0)
	var latestIssue time.Time

	for _, product := range products {
		if err := validateProductMetadata(product); err != nil {
			return model.WeatherOutlookRun{}, time.Time{}, err
		}
		discussion, ok := discussions[product.Day]
		if !ok {
			return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: discussion for day %d is required", product.Key, product.Day)
		}

		collection, err := spcprovider.DecodeGeoJSON(product.Body)
		if err != nil {
			return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: %w", product.Key, err)
		}

		for i, feature := range collection.Features {
			if spcprovider.IsEmptyGeometryCollection(feature.Geometry) {
				issuedAt, err := parseRequiredSPCTime(feature.Properties.IssueISO, fmt.Sprintf("product %s feature %d.ISSUE_ISO", product.Key, i))
				if err != nil {
					return model.WeatherOutlookRun{}, time.Time{}, err
				}
				if latestIssue.IsZero() || issuedAt.After(latestIssue) {
					latestIssue = issuedAt
				}
				continue
			}

			outlook, err := mapFeature(product, feature, i, point, discussion)
			if err != nil {
				return model.WeatherOutlookRun{}, time.Time{}, err
			}
			if latestIssue.IsZero() || outlook.IssuedAt.After(latestIssue) {
				latestIssue = outlook.IssuedAt
			}
			outlooks = append(outlooks, outlook)
		}
	}

	asOf := latestIssue
	if asOf.IsZero() {
		asOf = latestDiscussionUpdated
	}
	if asOf.IsZero() {
		asOf = fallbackAsOf.UTC()
	}

	var issuedAt *time.Time
	if !latestIssue.IsZero() {
		t := latestIssue.UTC()
		issuedAt = &t
	}

	lat := bundle.Latitude
	lon := bundle.Longitude
	run := model.WeatherOutlookRun{
		LocationID:   strings.TrimSpace(bundle.LocationID),
		LocationName: strings.TrimSpace(bundle.LocationName),
		Latitude:     &lat,
		Longitude:    &lon,
		AsOf:         asOf.UTC(),
		IssuedAt:     issuedAt,
		Outlooks:     outlooks,
	}
	return run, run.AsOf, nil
}

type parsedDiscussion struct {
	Headline   string
	Summary    string
	Discussion string
	UpdatedAt  *time.Time
}

func parseDiscussions(pages []spcprovider.RawDiscussionPage) (map[int]parsedDiscussion, time.Time, error) {
	out := map[int]parsedDiscussion{}
	var latestUpdated time.Time
	for _, page := range pages {
		parsed, err := spcprovider.ParseDiscussionHTML(page.Body)
		if err != nil {
			return nil, time.Time{}, fmt.Errorf("discussion %s: %w", page.Key, err)
		}
		day := page.Day
		if day == 0 {
			if meta, ok := spcprovider.DiscussionProductByKey(page.Key); ok {
				day = meta.Day
			}
		}
		if day < 1 || day > 3 {
			return nil, time.Time{}, fmt.Errorf("discussion %s: day must be 1, 2, or 3, got %d", page.Key, page.Day)
		}
		disc := parsedDiscussion{
			Headline:   strings.TrimSpace(parsed.Headline),
			Summary:    strings.TrimSpace(parsed.Summary),
			Discussion: strings.TrimSpace(parsed.Discussion),
			UpdatedAt:  parsed.UpdatedAt,
		}
		out[day] = disc
		if parsed.UpdatedAt != nil && (latestUpdated.IsZero() || parsed.UpdatedAt.After(latestUpdated)) {
			latestUpdated = parsed.UpdatedAt.UTC()
		}
	}
	return out, latestUpdated, nil
}

func orderedProducts(products []spcprovider.RawOutlookProduct) []spcprovider.RawOutlookProduct {
	out := make([]spcprovider.RawOutlookProduct, len(products))
	copy(out, products)
	sort.SliceStable(out, func(i, j int) bool {
		if out[i].Day != out[j].Day {
			return out[i].Day < out[j].Day
		}
		left := outlookTypeOrder(out[i].OutlookType)
		right := outlookTypeOrder(out[j].OutlookType)
		if left != right {
			return left < right
		}
		return out[i].Key < out[j].Key
	})
	return out
}

func outlookTypeOrder(outlookType string) int {
	switch strings.TrimSpace(outlookType) {
	case spcprovider.OutlookTypeCategorical:
		return 0
	case spcprovider.OutlookTypeTornado:
		return 1
	case spcprovider.OutlookTypeHail:
		return 2
	case spcprovider.OutlookTypeWind:
		return 3
	default:
		return outlookTypeUnknown
	}
}

func validateProductMetadata(product spcprovider.RawOutlookProduct) error {
	if product.Day < 1 || product.Day > 3 {
		return fmt.Errorf("product %s: day must be 1, 2, or 3, got %d", product.Key, product.Day)
	}
	switch strings.TrimSpace(product.OutlookType) {
	case spcprovider.OutlookTypeCategorical, spcprovider.OutlookTypeTornado, spcprovider.OutlookTypeHail, spcprovider.OutlookTypeWind:
		return nil
	default:
		return fmt.Errorf("product %s: unsupported outlook type %q", product.Key, product.OutlookType)
	}
}

func mapFeature(product spcprovider.RawOutlookProduct, feature spcprovider.GeoJSONFeature, index int, point geo.Point, discussion parsedDiscussion) (model.WeatherOutlook, error) {
	fieldPrefix := fmt.Sprintf("product %s feature %d", product.Key, index)
	props := feature.Properties

	validFrom, err := parseRequiredSPCTime(props.ValidISO, fieldPrefix+".VALID_ISO")
	if err != nil {
		return model.WeatherOutlook{}, err
	}
	validTo, err := parseRequiredSPCTime(props.ExpireISO, fieldPrefix+".EXPIRE_ISO")
	if err != nil {
		return model.WeatherOutlook{}, err
	}
	issuedAt, err := parseRequiredSPCTime(props.IssueISO, fieldPrefix+".ISSUE_ISO")
	if err != nil {
		return model.WeatherOutlook{}, err
	}
	label := strings.TrimSpace(props.Label)
	if label == "" {
		return model.WeatherOutlook{}, fmt.Errorf("%s.LABEL is required", fieldPrefix)
	}
	if len(feature.Geometry) == 0 {
		return model.WeatherOutlook{}, fmt.Errorf("%s.geometry is required", fieldPrefix)
	}
	containsLocation, err := geo.ContainsPoint(feature.Geometry, point)
	if err != nil {
		return model.WeatherOutlook{}, fmt.Errorf("%s.geometry: %w", fieldPrefix, err)
	}

	geometry := make(json.RawMessage, len(feature.Geometry))
	copy(geometry, feature.Geometry)

	return model.WeatherOutlook{
		ID:               outlookID(product.Day, product.OutlookType, label, issuedAt, validFrom, index),
		Provider:         providerSPC,
		Product:          productConvective,
		Day:              product.Day,
		OutlookType:      strings.TrimSpace(product.OutlookType),
		Label:            label,
		LabelText:        strings.TrimSpace(props.Label2),
		SeverityRank:     props.DN,
		ValidFrom:        validFrom,
		ValidTo:          validTo,
		IssuedAt:         issuedAt,
		ExpiresAt:        validTo,
		Forecaster:       strings.TrimSpace(props.Forecaster),
		Headline:         discussion.Headline,
		Summary:          discussion.Summary,
		Discussion:       discussion.Discussion,
		SourceURL:        strings.TrimSpace(product.URL),
		ImageURL:         "",
		ContainsLocation: containsLocation,
		Geometry:         geometry,
	}, nil
}

func parseRequiredSPCTime(value, field string) (time.Time, error) {
	if strings.TrimSpace(value) == "" {
		return time.Time{}, fmt.Errorf("%s is required", field)
	}
	t, err := spcprovider.ParseISOTimestamp(value)
	if err != nil {
		return time.Time{}, fmt.Errorf("%s: %w", field, err)
	}
	return t.UTC(), nil
}

func outlookID(day int, outlookType, label string, issuedAt time.Time, validFrom time.Time, index int) string {
	return fmt.Sprintf(
		"spc-convective-day%d-%s-%s-%s-%s-%d",
		day,
		safeIDToken(outlookType),
		safeIDToken(label),
		issuedAt.UTC().Format(time.RFC3339),
		validFrom.UTC().Format(time.RFC3339),
		index,
	)
}

func safeIDToken(value string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	value = idTokenRE.ReplaceAllString(value, "-")
	value = strings.Trim(value, "-")
	if value == "" {
		return "unknown"
	}
	return value
}

func validateCoordinates(latitude, longitude float64) error {
	switch {
	case math.IsNaN(latitude) || math.IsInf(latitude, 0):
		return fmt.Errorf("latitude must be finite")
	case math.IsNaN(longitude) || math.IsInf(longitude, 0):
		return fmt.Errorf("longitude must be finite")
	case latitude < -90 || latitude > 90:
		return fmt.Errorf("latitude must be between -90 and 90, got %v", latitude)
	case longitude < -180 || longitude > 180:
		return fmt.Errorf("longitude must be between -180 and 180, got %v", longitude)
	default:
		return nil
	}
}