Add a new field to the alert schema to fix a mismatch between the prior schema and the upstream NWS API

internal/normalizers/nws/alerts.go

@@ -93,12 +93,8 @@ func buildAlerts(parsed nwsAlertsResponse, fallbackAsOf time.Time) (model.Weathe
 		sent := nwscommon.ParseTimePtr(p.Sent)
 		effective := nwscommon.ParseTimePtr(p.Effective)
 		onset := nwscommon.ParseTimePtr(p.Onset)
-
-		// Expires: prefer "expires"; fall back to "ends" if present.
+		ends := nwscommon.ParseTimePtr(p.Ends)
 		expires := nwscommon.ParseTimePtr(p.Expires)
-		if expires == nil {
-			expires = nwscommon.ParseTimePtr(p.Ends)
-		}
 
 		refs := parseNWSAlertReferences(p.References)
 
@@ -123,6 +119,7 @@ func buildAlerts(parsed nwsAlertsResponse, fallbackAsOf time.Time) (model.Weathe
 			Sent:      sent,
 			Effective: effective,
 			Onset:     onset,
+			Ends:      ends,
 			Expires:   expires,
 
 			AreaDescription: strings.TrimSpace(p.AreaDesc),

internal/normalizers/nws/alerts_test.go

@@ -0,0 +1,136 @@
+package nws
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
+)
+
+func TestAlertsNormalizerMapsEndsSeparatelyFromExpires(t *testing.T) {
+	raw := []byte(`{
+		"updated":"2026-06-16T10:00:00+00:00",
+		"title":"Current watches, warnings, and advisories for St. Louis",
+		"features":[{
+			"id":"https://api.weather.gov/alerts/alert-1",
+			"properties":{
+				"event":"Flood Warning",
+				"headline":"Flood Warning issued",
+				"sent":"2026-06-16T09:55:00+00:00",
+				"effective":"2026-06-16T10:00:00+00:00",
+				"onset":"2026-06-16T10:15:00+00:00",
+				"ends":"2026-06-16T14:00:00+00:00",
+				"expires":"2026-06-16T11:00:00+00:00"
+			}
+		}]
+	}`)
+
+	out, err := AlertsNormalizer{}.Normalize(context.Background(), alertRawEvent(raw))
+	if err != nil {
+		t.Fatalf("Normalize() error = %v", err)
+	}
+
+	run := decodeAlertRun(t, out)
+	if len(run.Alerts) != 1 {
+		t.Fatalf("expected 1 alert, got %d", len(run.Alerts))
+	}
+	alert := run.Alerts[0]
+	wantEnds := time.Date(2026, 6, 16, 14, 0, 0, 0, time.UTC)
+	wantExpires := time.Date(2026, 6, 16, 11, 0, 0, 0, time.UTC)
+	if alert.Ends == nil || !alert.Ends.Equal(wantEnds) {
+		t.Fatalf("ends = %v, want %s", alert.Ends, wantEnds)
+	}
+	if alert.Expires == nil || !alert.Expires.Equal(wantExpires) {
+		t.Fatalf("expires = %v, want %s", alert.Expires, wantExpires)
+	}
+}
+
+func TestAlertsNormalizerDoesNotFallbackExpiresToEnds(t *testing.T) {
+	raw := []byte(`{
+		"updated":"2026-06-16T10:00:00+00:00",
+		"features":[{
+			"id":"alert-ends-only",
+			"properties":{
+				"event":"Heat Advisory",
+				"ends":"2026-06-16T22:00:00+00:00"
+			}
+		}]
+	}`)
+
+	out, err := AlertsNormalizer{}.Normalize(context.Background(), alertRawEvent(raw))
+	if err != nil {
+		t.Fatalf("Normalize() error = %v", err)
+	}
+
+	run := decodeAlertRun(t, out)
+	alert := run.Alerts[0]
+	if alert.Ends == nil {
+		t.Fatal("expected ends to be populated")
+	}
+	if alert.Expires != nil {
+		t.Fatalf("expected expires nil when upstream expires is absent, got %v", alert.Expires)
+	}
+}
+
+func TestAlertsNormalizerIgnoresInvalidEnds(t *testing.T) {
+	raw := []byte(`{
+		"updated":"2026-06-16T10:00:00+00:00",
+		"features":[{
+			"id":"alert-invalid-ends",
+			"properties":{
+				"event":"Special Weather Statement",
+				"ends":"not-a-time",
+				"expires":"2026-06-16T11:00:00+00:00"
+			}
+		}]
+	}`)
+
+	out, err := AlertsNormalizer{}.Normalize(context.Background(), alertRawEvent(raw))
+	if err != nil {
+		t.Fatalf("Normalize() error = %v", err)
+	}
+
+	run := decodeAlertRun(t, out)
+	alert := run.Alerts[0]
+	if alert.Ends != nil {
+		t.Fatalf("expected invalid ends to map nil, got %v", alert.Ends)
+	}
+	if alert.Expires == nil {
+		t.Fatal("expected expires to remain populated")
+	}
+}
+
+func alertRawEvent(raw []byte) event.Event {
+	emittedAt := time.Date(2026, 6, 16, 10, 5, 0, 0, time.UTC)
+	return event.Event{
+		ID:        "raw-alerts",
+		Kind:      event.Kind(standards.KindAlert),
+		Source:    "NWSAlerts",
+		Schema:    standards.SchemaRawNWSAlertsV1,
+		EmittedAt: emittedAt,
+		Payload:   json.RawMessage(raw),
+	}
+}
+
+func decodeAlertRun(t *testing.T, e *event.Event) model.WeatherAlertRun {
+	t.Helper()
+	if e == nil {
+		t.Fatal("expected normalized event")
+	}
+	if e.Schema != standards.SchemaWeatherAlertV1 {
+		t.Fatalf("schema = %q, want %q", e.Schema, standards.SchemaWeatherAlertV1)
+	}
+	var run model.WeatherAlertRun
+	raw, err := json.Marshal(e.Payload)
+	if err != nil {
+		t.Fatalf("marshal alert payload: %v", err)
+	}
+	if err := json.Unmarshal(raw, &run); err != nil {
+		t.Fatalf("decode alert payload: %v", err)
+	}
+	return run
+}

internal/sinks/postgres/doc.go

@@ -198,6 +198,7 @@
 //     - sent TIMESTAMPTZ NULL -> payload.alerts[i].sent
 //     - effective TIMESTAMPTZ NULL -> payload.alerts[i].effective
 //     - onset TIMESTAMPTZ NULL -> payload.alerts[i].onset
+//     - ends TIMESTAMPTZ NULL -> payload.alerts[i].ends
 //     - expires TIMESTAMPTZ NULL -> payload.alerts[i].expires
 //     - area_description TEXT NULL -> payload.alerts[i].areaDescription
 //     - sender_name TEXT NULL -> payload.alerts[i].senderName

internal/sinks/postgres/map.go

@@ -302,6 +302,7 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 				"sent":             nullableTime(a.Sent),
 				"effective":        nullableTime(a.Effective),
 				"onset":            nullableTime(a.Onset),
+				"ends":             nullableTime(a.Ends),
 				"expires":          nullableTime(a.Expires),
 				"area_description": nullableString(a.AreaDescription),
 				"sender_name":      nullableString(a.SenderName),

internal/sinks/postgres/map_test.go

@@ -103,6 +103,8 @@ func TestMapPostgresEventForecastStructPayload(t *testing.T) {
 
 func TestMapPostgresEventAlertStructPayload(t *testing.T) {
 	sent := time.Date(2026, 3, 16, 17, 0, 0, 0, time.UTC)
+	ends := time.Date(2026, 3, 16, 20, 0, 0, 0, time.UTC)
+	expires := time.Date(2026, 3, 16, 18, 30, 0, 0, time.UTC)
 	run := model.WeatherAlertRun{
 		AsOf: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
 		Alerts: []model.WeatherAlert{
@@ -110,6 +112,8 @@ func TestMapPostgresEventAlertStructPayload(t *testing.T) {
 				ID:       "urn:alert:1",
 				Headline: "Winter Weather Advisory",
 				Severity: "Moderate",
+				Ends:     &ends,
+				Expires:  &expires,
 				References: []model.AlertReference{
 					{ID: "urn:ref:1", Sent: &sent},
 					{Identifier: "ref-two"},
@@ -145,6 +149,20 @@ func TestMapPostgresEventAlertStructPayload(t *testing.T) {
 	if got := firstAlert.Values["reference_count"]; got != 2 {
 		t.Fatalf("alerts reference_count = %#v, want 2", got)
 	}
+	if got := firstAlert.Values["ends"]; got != ends {
+		t.Fatalf("alerts ends = %#v, want %#v", got, ends)
+	}
+	if got := firstAlert.Values["expires"]; got != expires {
+		t.Fatalf("alerts expires = %#v, want %#v", got, expires)
+	}
+
+	alertWrites := writesForTable(writes, tableAlerts)
+	if len(alertWrites) != 2 {
+		t.Fatalf("alert writes len = %d, want 2", len(alertWrites))
+	}
+	if got := alertWrites[1].Values["ends"]; got != nil {
+		t.Fatalf("second alert ends = %#v, want nil", got)
+	}
 
 	assertAllWritesIncludeAllColumns(t, writes)
 }
@@ -758,6 +776,16 @@ func firstWriteForTable(writes []fksinks.PostgresWrite, table string) (fksinks.P
 	return fksinks.PostgresWrite{}, false
 }
 
+func writesForTable(writes []fksinks.PostgresWrite, table string) []fksinks.PostgresWrite {
+	out := make([]fksinks.PostgresWrite, 0)
+	for _, w := range writes {
+		if w.Table == table {
+			out = append(out, w)
+		}
+	}
+	return out
+}
+
 func assertAllWritesIncludeAllColumns(t *testing.T, writes []fksinks.PostgresWrite) {
 	t.Helper()
 	colCounts := tableColumnCounts()

internal/sinks/postgres/schema.go

@@ -238,6 +238,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 					{Name: "sent", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "effective", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "onset", Type: "TIMESTAMPTZ", Nullable: true},
+					{Name: "ends", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "expires", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "area_description", Type: "TEXT", Nullable: true},
 					{Name: "sender_name", Type: "TEXT", Nullable: true},

internal/sinks/postgres/schema_test.go

@@ -91,6 +91,15 @@ func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
 	assertTableUniqueIndex(t, tableOutlookDiscussions, "idx_wf_outlook_discussions_run_day", []string{"run_event_id", "day"})
 }
 
+func TestWeatherPostgresSchemaIncludesAlertEndsColumn(t *testing.T) {
+	alertColumns := columnsForTable(t, tableAlerts)
+	for _, col := range []string{"run_event_id", "alert_index", "as_of", "alert_id", "onset", "ends", "expires"} {
+		if !alertColumns[col] {
+			t.Fatalf("%s missing %s column", tableAlerts, col)
+		}
+	}
+}
+
 func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
 	runColumns := columnsForTable(t, tableWeatherStoryRuns)
 	if !runColumns["as_of"] {