diff --git a/docs/consumers/pkg-model.md b/docs/consumers/pkg-model.md index cca1a89..6624dea 100644 --- a/docs/consumers/pkg-model.md +++ b/docs/consumers/pkg-model.md @@ -41,9 +41,6 @@ Related child types include: - `WeatherOutlookDiscussion` - `WMOCode` -`WeatherOutlookRun` includes `WeatherOutlookDiscussion` entries as run-level -SPC day discussions. - ## Wire And Compatibility Rules - JSON tags define canonical payload field names. diff --git a/docs/integrations/events.md b/docs/integrations/events.md index 9b9c9ff..9813e54 100644 --- a/docs/integrations/events.md +++ b/docs/integrations/events.md @@ -211,8 +211,9 @@ Payload type: `WeatherAlertRun`. | `instruction` | string | no | Alert instruction. | | `sent` | timestamp | no | Provider sent time. | | `effective` | timestamp | no | Effective time. | -| `onset` | timestamp | no | Onset time. | -| `expires` | timestamp | no | Expiration time. | +| `onset` | timestamp | no | Alert period start. | +| `ends` | timestamp | no | Alert period end. | +| `expires` | timestamp | no | Provider expiration metadata; not necessarily the alert period end. | | `areaDescription` | string | no | Affected area description. | | `senderName` | string | no | Provider sender name. | | `references` | array | no | Related alerts. | diff --git a/docs/integrations/nws.md b/docs/integrations/nws.md index ba669eb..3ad321b 100644 --- a/docs/integrations/nws.md +++ b/docs/integrations/nws.md @@ -44,6 +44,9 @@ normalizer uses fields under `properties` such as `stationId`, `stationName`, `nws_alerts` expects an alerts FeatureCollection. The normalizer uses the collection `updated` timestamp, `title`, each feature ID, alert classification fields, narrative fields, timing fields, sender fields, and references. +`properties.onset` and `properties.ends` map to the canonical alert period +start and end. `properties.expires` maps only to canonical `expires` provider +metadata and is not treated as the alert period end. `nws_forecast_hourly` and `nws_forecast_narrative` expect gridpoint forecast GeoJSON with `properties.generatedAt`, `properties.updateTime`, elevation, @@ -98,8 +101,9 @@ unset. Forecast temperatures are converted to Celsius when NWS supplies Fahrenheit, and wind speed strings are converted to kilometers per hour. Alert timing fields are parsed best-effort. Invalid per-alert timestamps are -left unset rather than failing the whole alert run. Missing alert IDs are -synthesized from the run snapshot time and array position. +left unset rather than failing the whole alert run. NWS `ends` is preserved +separately from `expires`; `expires` does not fall back to `ends`. Missing alert +IDs are synthesized from the run snapshot time and array position. Forecast discussion parsing requires an issue time. Weather story entries require start time, end time, and update time. diff --git a/docs/integrations/postgres.md b/docs/integrations/postgres.md index 6bf429a..b9b1540 100644 --- a/docs/integrations/postgres.md +++ b/docs/integrations/postgres.md @@ -353,6 +353,7 @@ Indexes: | `sent` | `TIMESTAMPTZ` | yes | `payload.alerts[].sent` | | `effective` | `TIMESTAMPTZ` | yes | `payload.alerts[].effective` | | `onset` | `TIMESTAMPTZ` | yes | `payload.alerts[].onset` | +| `ends` | `TIMESTAMPTZ` | yes | `payload.alerts[].ends` | | `expires` | `TIMESTAMPTZ` | yes | `payload.alerts[].expires` | | `area_description` | `TEXT` | yes | `payload.alerts[].areaDescription` | | `sender_name` | `TEXT` | yes | `payload.alerts[].senderName` | diff --git a/internal/normalizers/nws/alerts.go b/internal/normalizers/nws/alerts.go index 20d7c37..d500d8e 100644 --- a/internal/normalizers/nws/alerts.go +++ b/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), diff --git a/internal/normalizers/nws/alerts_test.go b/internal/normalizers/nws/alerts_test.go new file mode 100644 index 0000000..151f80e --- /dev/null +++ b/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 +} diff --git a/internal/sinks/postgres/doc.go b/internal/sinks/postgres/doc.go index 22bc38d..7627d30 100644 --- a/internal/sinks/postgres/doc.go +++ b/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 diff --git a/internal/sinks/postgres/map.go b/internal/sinks/postgres/map.go index 1a68830..7a89473 100644 --- a/internal/sinks/postgres/map.go +++ b/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), diff --git a/internal/sinks/postgres/map_test.go b/internal/sinks/postgres/map_test.go index 62fda91..98e64da 100644 --- a/internal/sinks/postgres/map_test.go +++ b/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() diff --git a/internal/sinks/postgres/schema.go b/internal/sinks/postgres/schema.go index f95e397..227bb9a 100644 --- a/internal/sinks/postgres/schema.go +++ b/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}, diff --git a/internal/sinks/postgres/schema_test.go b/internal/sinks/postgres/schema_test.go index c090f34..0e81ee8 100644 --- a/internal/sinks/postgres/schema_test.go +++ b/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"] { diff --git a/model/alert.go b/model/alert.go index 0f18cb8..b1bd1d4 100644 --- a/model/alert.go +++ b/model/alert.go @@ -55,9 +55,11 @@ type WeatherAlert struct { Instruction string `json:"instruction,omitempty"` // Timing (all optional; provider-dependent). + // Onset and Ends describe the alert period. Expires is provider expiration metadata. Sent *time.Time `json:"sent,omitempty"` Effective *time.Time `json:"effective,omitempty"` Onset *time.Time `json:"onset,omitempty"` + Ends *time.Time `json:"ends,omitempty"` Expires *time.Time `json:"expires,omitempty"` // Scope / affected area.