Add Postgres mapping for SPC outlooks

internal/sinks/postgres/map.go

@@ -1,6 +1,7 @@
 package postgres
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -26,6 +27,8 @@ func mapPostgresEvent(_ context.Context, e fkevent.Event) ([]fksinks.PostgresWri
 		return mapWeatherStoryEvent(e)
 	case standards.SchemaWeatherAlertV1:
 		return mapAlertEvent(e)
+	case standards.SchemaWeatherOutlookV1:
+		return mapOutlookEvent(e)
 	default:
 		return nil, nil
 	}
@@ -356,6 +359,107 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 	return writes, nil
 }
 
+func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
+	run, err := decodePayload[model.WeatherOutlookRun](e.Payload)
+	if err != nil {
+		return nil, fmt.Errorf("decode outlook payload: %w", err)
+	}
+	if run.AsOf.IsZero() {
+		return nil, fmt.Errorf("decode outlook payload: asOf is required")
+	}
+
+	asOf := run.AsOf.UTC()
+	writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks))
+	writes = append(writes, fksinks.PostgresWrite{
+		Table: tableOutlookRuns,
+		Values: map[string]any{
+			"event_id":           e.ID,
+			"event_kind":         string(e.Kind),
+			"event_source":       e.Source,
+			"event_schema":       e.Schema,
+			"event_emitted_at":   e.EmittedAt.UTC(),
+			"event_effective_at": nullableTime(e.EffectiveAt),
+			"location_id":        nullableString(run.LocationID),
+			"location_name":      nullableString(run.LocationName),
+			"latitude":           nullableFloat64(run.Latitude),
+			"longitude":          nullableFloat64(run.Longitude),
+			"as_of":              asOf,
+			"issued_at":          nullableTime(run.IssuedAt),
+			"outlook_count":      len(run.Outlooks),
+		},
+	})
+
+	for i, outlook := range run.Outlooks {
+		if err := validateOutlook(outlook, i); err != nil {
+			return nil, err
+		}
+		geometryJSON, err := requiredCompactJSONText(outlook.Geometry)
+		if err != nil {
+			return nil, fmt.Errorf("decode outlook payload: outlooks[%d].geometry: %w", i, err)
+		}
+
+		writes = append(writes, fksinks.PostgresWrite{
+			Table: tableOutlooks,
+			Values: map[string]any{
+				"run_event_id":      e.ID,
+				"outlook_index":     i,
+				"as_of":             asOf,
+				"product":           outlook.Product,
+				"day":               outlook.Day,
+				"outlook_type":      outlook.OutlookType,
+				"label":             outlook.Label,
+				"label_text":        nullableString(outlook.LabelText),
+				"severity_rank":     nullableInt(outlook.SeverityRank),
+				"valid_from":        outlook.ValidFrom.UTC(),
+				"valid_to":          outlook.ValidTo.UTC(),
+				"issued_at":         outlook.IssuedAt.UTC(),
+				"expires_at":        outlook.ExpiresAt.UTC(),
+				"forecaster":        nullableString(outlook.Forecaster),
+				"headline":          nullableString(outlook.Headline),
+				"summary":           nullableString(outlook.Summary),
+				"discussion":        nullableString(outlook.Discussion),
+				"source_url":        nullableString(outlook.SourceURL),
+				"image_url":         nullableString(outlook.ImageURL),
+				"contains_location": outlook.ContainsLocation,
+				"geometry_json":     geometryJSON,
+			},
+		})
+	}
+
+	return writes, nil
+}
+
+func validateOutlook(outlook model.WeatherOutlook, index int) error {
+	if strings.TrimSpace(outlook.ID) == "" {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].id is required", index)
+	}
+	if strings.TrimSpace(outlook.Provider) == "" {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].provider is required", index)
+	}
+	if strings.TrimSpace(outlook.Product) == "" {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].product is required", index)
+	}
+	if outlook.Day == 0 {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].day is required", index)
+	}
+	if strings.TrimSpace(outlook.OutlookType) == "" {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].outlookType is required", index)
+	}
+	if strings.TrimSpace(outlook.Label) == "" {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].label is required", index)
+	}
+	if outlook.ValidFrom.IsZero() || outlook.ValidTo.IsZero() {
+		return fmt.Errorf("decode outlook payload: outlooks[%d] validFrom/validTo are required", index)
+	}
+	if outlook.IssuedAt.IsZero() || outlook.ExpiresAt.IsZero() {
+		return fmt.Errorf("decode outlook payload: outlooks[%d] issuedAt/expiresAt are required", index)
+	}
+	if len(outlook.Geometry) == 0 {
+		return fmt.Errorf("decode outlook payload: outlooks[%d].geometry is required", index)
+	}
+	return nil
+}
+
 func decodePayload[T any](payload any) (T, error) {
 	var out T
 	if payload == nil {
@@ -418,6 +522,13 @@ func nullableBool(v *bool) any {
 	return *v
 }
 
+func nullableInt(v *int) any {
+	if v == nil {
+		return nil
+	}
+	return *v
+}
+
 func nullableTime(v *time.Time) any {
 	if v == nil || v.IsZero() {
 		return nil
@@ -445,3 +556,19 @@ func compactJSONText(v any) (any, error) {
 	}
 	return string(b), nil
 }
+
+func requiredCompactJSONText(v any) (string, error) {
+	compact, err := compactJSONText(v)
+	if err != nil {
+		return "", err
+	}
+	s, ok := compact.(string)
+	if !ok || strings.TrimSpace(s) == "" || strings.TrimSpace(s) == "null" {
+		return "", fmt.Errorf("is required")
+	}
+	var buf bytes.Buffer
+	if err := json.Compact(&buf, []byte(s)); err != nil {
+		return "", err
+	}
+	return buf.String(), nil
+}