11 Commits

Author SHA1 Message Date
50215d2105 Add a new field to the alert schema to fix a mismatch between the prior schema and the upstream NWS API
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
2026-06-16 19:57:48 -05:00
74411e3f54 Clean up completed roadmap and documentation for the new outlook schema
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
2026-06-12 08:30:41 -05:00
4358a7cdce Verify outlook schema release readiness
Some checks failed
ci/woodpecker/push/build-image Pipeline failed
2026-06-12 04:40:33 +00:00
dcea5261ab Document current outlook schema behavior 2026-06-12 04:38:44 +00:00
97141c7a9b Add outlook schema transition guide 2026-06-12 04:32:52 +00:00
2e2d36024e Align normalizer docs with outlook v2 2026-06-12 04:31:15 +00:00
4d2cddf801 Update Postgres outlook storage for v2 2026-06-12 04:29:00 +00:00
21a35a5205 Emit location-filtered SPC outlook v2 runs 2026-06-12 04:24:12 +00:00
435d1ade07 Add outlook v2 model contract 2026-06-12 04:19:56 +00:00
819ac24aed Added an implementation plan for updates and revisions to the outlook code
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
2026-06-11 23:15:00 -05:00
2c472449e8 Fix a bug in the GeoJSON handling when there are no active polygons
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
2026-06-11 22:21:59 -05:00
31 changed files with 1182 additions and 188 deletions

View File

@@ -63,7 +63,7 @@ func Decode(payload []byte) (any, error) {
case standards.SchemaWeatherAlertV1: case standards.SchemaWeatherAlertV1:
var out model.WeatherAlertRun var out model.WeatherAlertRun
return &out, json.Unmarshal(evt.Payload, &out) return &out, json.Unmarshal(evt.Payload, &out)
case standards.SchemaWeatherOutlookV1: case standards.SchemaWeatherOutlookV2:
var out model.WeatherOutlookRun var out model.WeatherOutlookRun
return &out, json.Unmarshal(evt.Payload, &out) return &out, json.Unmarshal(evt.Payload, &out)
default: default:

View File

@@ -9,8 +9,8 @@ import "gitea.maximumdirect.net/ejr/weatherfeeder/model"
## Purpose ## Purpose
Package `model` defines `weatherfeeder`'s canonical weather payload structs. Package `model` defines `weatherfeeder`'s canonical weather payload structs.
These structs are emitted as the `payload` of canonical `weather.*.v1` events These structs are emitted as the `payload` of canonical `weather.*` events and
and are also the domain types consumed by downstream applications such as are also the domain types consumed by downstream applications such as
`weatherapi`. `weatherapi`.
The JSON field tags on these structs are part of the wire contract. For the full The JSON field tags on these structs are part of the wire contract. For the full
@@ -27,7 +27,7 @@ Current canonical schema families map to these public types:
| `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` | | `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` |
| `weather.weather_story.v1` | `WeatherStoryRun` | | `weather.weather_story.v1` | `WeatherStoryRun` |
| `weather.alert.v1` | `WeatherAlertRun` | | `weather.alert.v1` | `WeatherAlertRun` |
| `weather.outlook.v1` | `WeatherOutlookRun` | | `weather.outlook.v2` | `WeatherOutlookRun` |
Related child types include: Related child types include:
@@ -38,6 +38,7 @@ Related child types include:
- `WeatherAlert` - `WeatherAlert`
- `WeatherAlertReference` - `WeatherAlertReference`
- `WeatherOutlook` - `WeatherOutlook`
- `WeatherOutlookDiscussion`
- `WMOCode` - `WMOCode`
## Wire And Compatibility Rules ## Wire And Compatibility Rules

View File

@@ -41,6 +41,12 @@ Canonical schemas emitted after normalization:
| `SchemaWeatherForecastDiscussionV1` | `weather.forecast_discussion.v1` | | `SchemaWeatherForecastDiscussionV1` | `weather.forecast_discussion.v1` |
| `SchemaWeatherStoryV1` | `weather.weather_story.v1` | | `SchemaWeatherStoryV1` | `weather.weather_story.v1` |
| `SchemaWeatherAlertV1` | `weather.alert.v1` | | `SchemaWeatherAlertV1` | `weather.alert.v1` |
| `SchemaWeatherOutlookV2` | `weather.outlook.v2` |
Historical canonical schema constant:
| Constant | Value |
|---|---|
| `SchemaWeatherOutlookV1` | `weather.outlook.v1` | | `SchemaWeatherOutlookV1` | `weather.outlook.v1` |
## Raw Schema Constants ## Raw Schema Constants

View File

@@ -37,7 +37,7 @@ Canonical schemas emitted after normalization:
| `forecast_discussion` | `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` | | `forecast_discussion` | `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` |
| `weather_story` | `weather.weather_story.v1` | `WeatherStoryRun` | | `weather_story` | `weather.weather_story.v1` | `WeatherStoryRun` |
| `alert` | `weather.alert.v1` | `WeatherAlertRun` | | `alert` | `weather.alert.v1` | `WeatherAlertRun` |
| `outlook` | `weather.outlook.v1` | `WeatherOutlookRun` | | `outlook` | `weather.outlook.v2` | `WeatherOutlookRun` |
Raw upstream schemas emitted by current sources: Raw upstream schemas emitted by current sources:
@@ -211,8 +211,9 @@ Payload type: `WeatherAlertRun`.
| `instruction` | string | no | Alert instruction. | | `instruction` | string | no | Alert instruction. |
| `sent` | timestamp | no | Provider sent time. | | `sent` | timestamp | no | Provider sent time. |
| `effective` | timestamp | no | Effective time. | | `effective` | timestamp | no | Effective time. |
| `onset` | timestamp | no | Onset time. | | `onset` | timestamp | no | Alert period start. |
| `expires` | timestamp | no | Expiration time. | | `ends` | timestamp | no | Alert period end. |
| `expires` | timestamp | no | Provider expiration metadata; not necessarily the alert period end. |
| `areaDescription` | string | no | Affected area description. | | `areaDescription` | string | no | Affected area description. |
| `senderName` | string | no | Provider sender name. | | `senderName` | string | no | Provider sender name. |
| `references` | array | no | Related alerts. | | `references` | array | no | Related alerts. |
@@ -220,13 +221,15 @@ Payload type: `WeatherAlertRun`.
`references[]` entries contain optional `id`, `identifier`, `sender`, and `references[]` entries contain optional `id`, `identifier`, `sender`, and
`sent` fields. `sent` fields.
## `weather.outlook.v1` ## `weather.outlook.v2`
Payload type: `WeatherOutlookRun`. Payload type: `WeatherOutlookRun`.
The current producer is the SPC convective outlook normalizer. It emits Day 1-3 The current producer is the SPC convective outlook normalizer. It emits Day 1-3
convective outlook polygons for categorical, tornado, hail, and wind products. convective outlook snapshots for categorical, tornado, hail, and wind products
All timestamps are UTC. that apply to the configured forecast point. Raw SPC bundles remain complete;
canonical outlook payloads are filtered to local polygons. All timestamps are
UTC.
| Field | Type | Required | Notes | | Field | Type | Required | Notes |
|---|---|:---:|---| |---|---|:---:|---|
@@ -236,7 +239,8 @@ All timestamps are UTC.
| `longitude` | number | no | Configured point longitude in decimal degrees. | | `longitude` | number | no | Configured point longitude in decimal degrees. |
| `asOf` | timestamp | yes | Snapshot time. For SPC, this is the latest outlook issue time when available. | | `asOf` | timestamp | yes | Snapshot time. For SPC, this is the latest outlook issue time when available. |
| `issuedAt` | timestamp | no | Latest issue time across outlook features when any feature exists. | | `issuedAt` | timestamp | no | Latest issue time across outlook features when any feature exists. |
| `outlooks` | array | yes | Ordered outlook polygons. | | `outlooks` | array | yes | Ordered outlook polygons containing the configured point. |
| `discussions` | array | yes | Run-level day discussions for retained outlook days. |
`outlooks[]` entries: `outlooks[]` entries:
@@ -255,19 +259,53 @@ All timestamps are UTC.
| `issuedAt` | timestamp | yes | Feature issue time. | | `issuedAt` | timestamp | yes | Feature issue time. |
| `expiresAt` | timestamp | yes | Expiration time; currently equal to `validTo`. | | `expiresAt` | timestamp | yes | Expiration time; currently equal to `validTo`. |
| `forecaster` | string | no | SPC forecaster text, when present. | | `forecaster` | string | no | SPC forecaster text, when present. |
| `headline` | string | no | Matching Day 1-3 print-page product title. |
| `summary` | string | no | Text from the print-page `...SUMMARY...` section. |
| `discussion` | string | no | Cleaned full print-page product text. |
| `sourceUrl` | string | no | GeoJSON product URL for this outlook feature. | | `sourceUrl` | string | no | GeoJSON product URL for this outlook feature. |
| `imageUrl` | string | no | Reserved for provider image URLs; currently empty. | | `imageUrl` | string | no | Reserved for provider image URLs; currently empty. |
| `containsLocation` | boolean | yes | Whether the configured point is inside or on the boundary of the polygon. | | `containsLocation` | boolean | yes | Always `true` for emitted v2 outlooks. |
| `geometry` | object | yes | Compact GeoJSON `Polygon` or `MultiPolygon` geometry. | | `geometry` | object | yes | Compact GeoJSON `Polygon` or `MultiPolygon` geometry. |
`geometry` preserves the SPC feature geometry as compact GeoJSON using `geometry` preserves the SPC feature geometry as compact GeoJSON using
`[longitude, latitude]` coordinate order. `containsLocation` is computed with `[longitude, latitude]` coordinate order. `containsLocation` is computed with
that geometry and the configured source `latitude`/`longitude`; boundary points that geometry and the configured source `latitude`/`longitude`; boundary points
count as contained. All outlook polygons are emitted, including polygons that do count as contained. Polygons that do not contain the configured point are not
not contain the configured point. included in canonical v2 payloads.
`discussions[]` entries:
| Field | Type | Required | Notes |
|---|---|:---:|---|
| `day` | integer | yes | SPC outlook day, currently `1`, `2`, or `3`. |
| `headline` | string | no | Matching Day 1-3 print-page product title. |
| `summary` | string | no | Text from the print-page `...SUMMARY...` section. |
| `discussion` | string | no | Cleaned full print-page product text. |
| `updatedAt` | timestamp | no | Print-page update time, when present. |
When no SPC polygons apply locally, the run is still emitted with `outlooks: []`
and `discussions: []`. Discussions are included only for days represented by at
least one retained outlook, and multiple retained outlook types for the same day
share one discussion entry.
### SPC Outlook Supersession
Consumers should prefer latest-run semantics for current conditions: read the
latest `WeatherOutlookRun` for the configured location and use its `outlooks`
and `discussions` arrays together.
Historical SQL consumers that collapse older rows should identify superseded
outlooks by `provider`, `product`, `outlookType`, `validFrom`, and `validTo`,
then keep rows with the greatest `issuedAt`. `day` and `label` are not identity
fields. When multiple retained polygons share that latest `issuedAt`, preserve
the full group.
## Legacy `weather.outlook.v1`
`weather.outlook.v1` is a historical canonical schema retained as a standards
constant for older data and consumers. Current SPC normalization emits
`weather.outlook.v2`.
The v1 payload used `WeatherOutlookRun` and placed `headline`, `summary`, and
`discussion` on each `outlooks[]` polygon. It also represented the pre-v2 SPC
canonical behavior, where national polygons were preserved in canonical output.
## Compact Example ## Compact Example

View File

@@ -44,6 +44,9 @@ normalizer uses fields under `properties` such as `stationId`, `stationName`,
`nws_alerts` expects an alerts FeatureCollection. The normalizer uses the `nws_alerts` expects an alerts FeatureCollection. The normalizer uses the
collection `updated` timestamp, `title`, each feature ID, alert classification collection `updated` timestamp, `title`, each feature ID, alert classification
fields, narrative fields, timing fields, sender fields, and references. 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 `nws_forecast_hourly` and `nws_forecast_narrative` expect gridpoint forecast
GeoJSON with `properties.generatedAt`, `properties.updateTime`, elevation, 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. Fahrenheit, and wind speed strings are converted to kilometers per hour.
Alert timing fields are parsed best-effort. Invalid per-alert timestamps are 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 left unset rather than failing the whole alert run. NWS `ends` is preserved
synthesized from the run snapshot time and array position. 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 Forecast discussion parsing requires an issue time. Weather story entries require
start time, end time, and update time. start time, end time, and update time.

View File

@@ -23,7 +23,7 @@ Events are mapped only for canonical weather schemas:
- `weather.forecast_discussion.v1` - `weather.forecast_discussion.v1`
- `weather.weather_story.v1` - `weather.weather_story.v1`
- `weather.alert.v1` - `weather.alert.v1`
- `weather.outlook.v1` - `weather.outlook.v2`
Unsupported schemas produce no writes for this sink. Mapped events are inserted Unsupported schemas produce no writes for this sink. Mapped events are inserted
transactionally. Inserts use ordinary `INSERT`; duplicate primary keys fail the transactionally. Inserts use ordinary `INSERT`; duplicate primary keys fail the
@@ -59,6 +59,7 @@ Parent tables store the feed event envelope:
| `alert_references` | `run_event_id`, `alert_index`, `reference_index` | `as_of` | | `alert_references` | `run_event_id`, `alert_index`, `reference_index` | `as_of` |
| `outlook_runs` | `event_id` | `as_of` | | `outlook_runs` | `event_id` | `as_of` |
| `outlooks` | `run_event_id`, `outlook_index` | `as_of` | | `outlooks` | `run_event_id`, `outlook_index` | `as_of` |
| `outlook_discussions` | `run_event_id`, `discussion_index` | `as_of` |
## Table Contract ## Table Contract
@@ -352,6 +353,7 @@ Indexes:
| `sent` | `TIMESTAMPTZ` | yes | `payload.alerts[].sent` | | `sent` | `TIMESTAMPTZ` | yes | `payload.alerts[].sent` |
| `effective` | `TIMESTAMPTZ` | yes | `payload.alerts[].effective` | | `effective` | `TIMESTAMPTZ` | yes | `payload.alerts[].effective` |
| `onset` | `TIMESTAMPTZ` | yes | `payload.alerts[].onset` | | `onset` | `TIMESTAMPTZ` | yes | `payload.alerts[].onset` |
| `ends` | `TIMESTAMPTZ` | yes | `payload.alerts[].ends` |
| `expires` | `TIMESTAMPTZ` | yes | `payload.alerts[].expires` | | `expires` | `TIMESTAMPTZ` | yes | `payload.alerts[].expires` |
| `area_description` | `TEXT` | yes | `payload.alerts[].areaDescription` | | `area_description` | `TEXT` | yes | `payload.alerts[].areaDescription` |
| `sender_name` | `TEXT` | yes | `payload.alerts[].senderName` | | `sender_name` | `TEXT` | yes | `payload.alerts[].senderName` |
@@ -408,6 +410,7 @@ Indexes:
| `as_of` | `TIMESTAMPTZ` | no | `payload.asOf` | | `as_of` | `TIMESTAMPTZ` | no | `payload.asOf` |
| `issued_at` | `TIMESTAMPTZ` | yes | `payload.issuedAt` | | `issued_at` | `TIMESTAMPTZ` | yes | `payload.issuedAt` |
| `outlook_count` | `INTEGER` | no | `len(payload.outlooks)` | | `outlook_count` | `INTEGER` | no | `len(payload.outlooks)` |
| `discussion_count` | `INTEGER` | no | `len(payload.discussions)` |
### `outlooks` ### `outlooks`
@@ -442,14 +445,36 @@ Indexes:
| `issued_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].issuedAt` | | `issued_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].issuedAt` |
| `expires_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].expiresAt` | | `expires_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].expiresAt` |
| `forecaster` | `TEXT` | yes | `payload.outlooks[].forecaster` | | `forecaster` | `TEXT` | yes | `payload.outlooks[].forecaster` |
| `headline` | `TEXT` | yes | `payload.outlooks[].headline` |
| `summary` | `TEXT` | yes | `payload.outlooks[].summary` |
| `discussion` | `TEXT` | yes | `payload.outlooks[].discussion` |
| `source_url` | `TEXT` | yes | `payload.outlooks[].sourceUrl` | | `source_url` | `TEXT` | yes | `payload.outlooks[].sourceUrl` |
| `image_url` | `TEXT` | yes | `payload.outlooks[].imageUrl` | | `image_url` | `TEXT` | yes | `payload.outlooks[].imageUrl` |
| `contains_location` | `BOOLEAN` | no | `payload.outlooks[].containsLocation` | | `contains_location` | `BOOLEAN` | no | `payload.outlooks[].containsLocation` |
| `geometry_json` | `TEXT` | no | Compact JSON from `payload.outlooks[].geometry` | | `geometry_json` | `TEXT` | no | Compact JSON from `payload.outlooks[].geometry` |
### `outlook_discussions`
Primary key: `run_event_id`, `discussion_index`
Prune column: `as_of`
Foreign key: `run_event_id` references `outlook_runs(event_id)` with cascade
delete.
Indexes:
- `idx_wf_outlook_discussions_day_as_of` on `day`, `as_of`
- unique `idx_wf_outlook_discussions_run_day` on `run_event_id`, `day`
| Column | Type | Null | Source |
|---|---|:---:|---|
| `run_event_id` | `TEXT REFERENCES outlook_runs(event_id) ON DELETE CASCADE` | no | Parent event ID. |
| `discussion_index` | `INTEGER` | no | `payload.discussions[]` index. |
| `as_of` | `TIMESTAMPTZ` | no | Parent `payload.asOf` |
| `day` | `INTEGER` | no | `payload.discussions[].day` |
| `headline` | `TEXT` | yes | `payload.discussions[].headline` |
| `summary` | `TEXT` | yes | `payload.discussions[].summary` |
| `discussion` | `TEXT` | yes | `payload.discussions[].discussion` |
| `updated_at` | `TIMESTAMPTZ` | yes | `payload.discussions[].updatedAt` |
## Retention ## Retention
When sink param `prune` is set, every successful write transaction deletes rows When sink param `prune` is set, every successful write transaction deletes rows
@@ -472,5 +497,6 @@ binary does not provide CLI commands for them.
- `WeatherAlertRun`: read `alert_runs`, join `alerts` by `run_event_id` ordered - `WeatherAlertRun`: read `alert_runs`, join `alerts` by `run_event_id` ordered
by `alert_index`, then join `alert_references` by `run_event_id` and by `alert_index`, then join `alert_references` by `run_event_id` and
`alert_index` ordered by `reference_index`. `alert_index` ordered by `reference_index`.
- `WeatherOutlookRun`: read `outlook_runs`, then join `outlooks` by - `WeatherOutlookRun`: read `outlook_runs`, join `outlooks` by `run_event_id`
`run_event_id` ordered by `outlook_index`. ordered by `outlook_index`, then join `outlook_discussions` by
`run_event_id` ordered by `discussion_index`.

View File

@@ -13,7 +13,7 @@ events are documented in [event wire contract](events.md).
| Driver | Kind | Raw schema | Canonical schema | | Driver | Kind | Raw schema | Canonical schema |
| --- | --- | --- | --- | | --- | --- | --- | --- |
| `spc_convective_outlook` | `outlook` | `raw.spc.convective_outlook.v1` | `weather.outlook.v1` | | `spc_convective_outlook` | `outlook` | `raw.spc.convective_outlook.v1` | `weather.outlook.v2` |
## Config Requirements ## Config Requirements
@@ -52,8 +52,8 @@ It also fetches three required print pages:
- Day 3 convective outlook print page - Day 3 convective outlook print page
GeoJSON products are authoritative for outlook polygons, valid windows, issue GeoJSON products are authoritative for outlook polygons, valid windows, issue
times, labels, and severity rank. Print pages are authoritative for headline, times, labels, and severity rank. Print pages are authoritative for run-level
summary, and discussion text. day discussion headline, summary, and discussion text.
## Accept Headers ## Accept Headers
@@ -99,23 +99,33 @@ Raw source `effective_at` prefers:
4. fetch time. 4. fetch time.
The normalizer sets canonical `asOf` and normalized event `effective_at` from The normalizer sets canonical `asOf` and normalized event `effective_at` from
the latest valid outlook feature `issuedAt`, with fallback to print-page update the latest valid GeoJSON issue time across the complete raw bundle, including
time and then the incoming event envelope. empty no-risk placeholders, with fallback to print-page update time and then the
incoming event envelope.
## Mapping Notes ## Mapping Notes
Each GeoJSON feature becomes one canonical outlook. Products are ordered by day, The raw source fetches and envelopes the complete SPC bundle. The normalizer
then categorical, tornado, hail, and wind. Feature order is preserved within decodes every configured GeoJSON product, skips empty no-risk
each product. `GeometryCollection` placeholders, and emits canonical outlooks only when the
configured point is inside or on the boundary of a real feature. Products are
ordered by day, then categorical, tornado, hail, and wind. Retained feature order
is preserved within each product.
The normalizer computes `containsLocation` with the configured latitude and The normalizer computes `containsLocation` with the configured latitude and
longitude against compact GeoJSON `Polygon` or `MultiPolygon` geometry. longitude against compact GeoJSON `Polygon` or `MultiPolygon` geometry.
Coordinates use GeoJSON order, `[longitude, latitude]`, and boundary points Coordinates use GeoJSON order, `[longitude, latitude]`, and boundary points
count as contained. count as contained.
All outlook polygons are preserved, including polygons that do not contain the Canonical outlook runs are emitted even when no polygons apply locally. In that
configured point. Matching print-page headline, summary, and discussion text is case the payload contains empty `outlooks` and `discussions` arrays.
attached to every outlook for the same day.
Print-page prose is represented as run-level day discussions. Discussions are
included only for days represented by at least one retained outlook. Multiple
retained outlook types for the same day share one discussion entry.
For downstream current-state and historical supersession guidance, see the
[event wire contract](events.md#spc-outlook-supersession).
## Failure Behavior ## Failure Behavior

View File

@@ -27,7 +27,7 @@ Current mappings:
| `raw.nws.forecast_discussion.v1` | `weather.forecast_discussion.v1` | | `raw.nws.forecast_discussion.v1` | `weather.forecast_discussion.v1` |
| `raw.nws.weatherstories.v1` | `weather.weather_story.v1` | | `raw.nws.weatherstories.v1` | `weather.weather_story.v1` |
| `raw.nws.alerts.v1` | `weather.alert.v1` | | `raw.nws.alerts.v1` | `weather.alert.v1` |
| `raw.spc.convective_outlook.v1` | `weather.outlook.v1` | | `raw.spc.convective_outlook.v1` | `weather.outlook.v2` |
## Boundaries ## Boundaries
@@ -59,13 +59,13 @@ Weatherfeeder registers normalizers in a stable order:
The current normalizers avoid ambiguous matches by using schema equality. The current normalizers avoid ambiguous matches by using schema equality.
The SPC outlook normalizer decodes the raw multi-document bundle, maps each The SPC outlook normalizer decodes the raw multi-document bundle, maps
GeoJSON feature to a canonical outlook, and enriches all outlooks for a day with location-containing GeoJSON features to canonical outlooks, and adds one
the matching print-page headline, summary, and discussion. It preserves compact run-level print-page discussion per retained outlook day. It preserves compact
GeoJSON feature geometry and computes `containsLocation` with GeoJSON feature geometry and computes `containsLocation` with
`internal/geo.ContainsPoint` using the source-configured point. Boundary points `internal/geo.ContainsPoint` using the source-configured point. Boundary points
count as contained, and all polygons are preserved whether or not they contain count as contained. Polygons that do not contain the point are omitted from the
the point. canonical run.
## State ## State

View File

@@ -19,11 +19,16 @@ Inputs are canonical feed events. The mapper currently handles these schemas:
- `weather.forecast_discussion.v1` - `weather.forecast_discussion.v1`
- `weather.weather_story.v1` - `weather.weather_story.v1`
- `weather.alert.v1` - `weather.alert.v1`
- `weather.outlook.v1` - `weather.outlook.v2`
Outputs are feedkit `PostgresWrite` values for weatherfeeder-owned tables. Outputs are feedkit `PostgresWrite` values for weatherfeeder-owned tables.
Unsupported schemas produce no writes and no error. Unsupported schemas produce no writes and no error.
Outlook runs are written to `outlook_runs`, retained local polygons are written
to `outlooks`, and run-level day discussions are written to
`outlook_discussions`. The parent run row stores `outlook_count` and
`discussion_count`.
## Boundaries ## Boundaries
- Weatherfeeder owns table definitions in `schema.go`. - Weatherfeeder owns table definitions in `schema.go`.
@@ -82,6 +87,7 @@ Child rows use positional indexes to preserve canonical array order:
- `alert_index` - `alert_index`
- `reference_index` - `reference_index`
- `outlook_index` - `outlook_index`
- `discussion_index`
Required canonical fields are validated before writes are returned: Required canonical fields are validated before writes are returned:
@@ -94,7 +100,10 @@ Required canonical fields are validated before writes are returned:
- alert runs require `asOf`, and each alert requires `id`; - alert runs require `asOf`, and each alert requires `id`;
- outlook runs require `asOf`, and each outlook requires `id`, `provider`, - outlook runs require `asOf`, and each outlook requires `id`, `provider`,
`product`, `day`, `outlookType`, `label`, `validFrom`, `validTo`, `issuedAt`, `product`, `day`, `outlookType`, `label`, `validFrom`, `validTo`, `issuedAt`,
`expiresAt`, and `geometry`. `expiresAt`, `containsLocation: true`, and `geometry`;
- outlook discussions require day `1`, `2`, or `3`; at least one of
`headline`, `summary`, or `discussion`; and no duplicate discussion day in the
same run.
Nullable canonical values are converted to SQL nulls by mapper helpers. Nullable canonical values are converted to SQL nulls by mapper helpers.
Observation present-weather raw values and outlook geometry values are stored as Observation present-weather raw values and outlook geometry values are stored as

View File

@@ -82,7 +82,7 @@ document bodies as the previous successful poll.
Every event passes through normalization first and dedupe second. Every event passes through normalization first and dedupe second.
Normalizers match raw source schemas and produce canonical `weather.*.v1` Normalizers match raw source schemas and produce versioned canonical `weather.*`
payloads. If an event has no matching normalizer, the normalize processor passes payloads. If an event has no matching normalizer, the normalize processor passes
it through unchanged. it through unchanged.

View File

@@ -48,6 +48,7 @@ The implementation style is:
- Hexagonal boundaries: provider APIs, config loading, scheduling, dispatch, and sinks are external mechanisms around the weather domain model and normalization logic. - Hexagonal boundaries: provider APIs, config loading, scheduling, dispatch, and sinks are external mechanisms around the weather domain model and normalization logic.
- Raw-to-canonical flow: sources should fetch and envelope raw provider payloads; normalizers should own provider-to-canonical mapping. - Raw-to-canonical flow: sources should fetch and envelope raw provider payloads; normalizers should own provider-to-canonical mapping.
- Location-focused canonical data: canonical weather events represent data relevant to the configured forecast location or configured provider object; `weatherfeeder` is not a national provider-data archive.
- Schema-based routing: normalizers match on event schema, not source name or event kind. - Schema-based routing: normalizers match on event schema, not source name or event kind.
- Composable registries: source drivers, normalizers, processors, and sinks are assembled explicitly through registries. - Composable registries: source drivers, normalizers, processors, and sinks are assembled explicitly through registries.
- Bounded concurrency: scheduling and sink fanout are concurrent, but the application should keep queues, goroutine ownership, logging, and cancellation behavior visible. - Bounded concurrency: scheduling and sink fanout are concurrent, but the application should keep queues, goroutine ownership, logging, and cancellation behavior visible.

View File

@@ -31,7 +31,7 @@ Notes:
- Day 4-8 products have different semantics from Day 1-3 categorical/tornado/hail/wind products. - Day 4-8 products have different semantics from Day 1-3 categorical/tornado/hail/wind products.
- Avoid forcing Day 4-8 assumptions into the current Day 1-3 model until the source shapes and consumer needs are reviewed. - Avoid forcing Day 4-8 assumptions into the current Day 1-3 model until the source shapes and consumer needs are reviewed.
- Prefer reusing `weather.outlook.v1` if the fields remain accurate; otherwise write a separate roadmap before changing the canonical contract. - Prefer reusing `weather.outlook.v2` if the fields remain accurate; otherwise write a separate roadmap before changing the canonical contract.
### Degraded SPC Bundle Mode ### Degraded SPC Bundle Mode

View File

@@ -0,0 +1,87 @@
# SPC Outlook Postgres Schema Transition
## Purpose
This document describes how to reset existing Postgres outlook tables from the
`weather.outlook.v1` storage shape to the `weather.outlook.v2` compatible
storage shape.
Updated `weatherfeeder` versions create outlook tables with run-level
discussion storage. Existing databases that already contain the old outlook
table family need a manual reset because the Postgres sink creates tables with
`CREATE TABLE IF NOT EXISTS`.
## Scope
This reset drops only the outlook table family and lets updated `weatherfeeder`
recreate it:
- `outlook_discussions`
- `outlooks`
- `outlook_runs`
Other weather tables are not affected.
## Warning
These commands delete stored SPC outlook history. Existing `weather.outlook.v1`
outlook rows are intentionally removed. Downstream readers should be updated
intentionally for the new outlook shape.
## Deployment Order
1. Stop `weatherfeeder`.
2. Drop the existing outlook tables.
3. Deploy updated `weatherfeeder`.
4. Start `weatherfeeder` so the Postgres sink recreates the new outlook tables.
5. Deploy updated downstream consumers such as `weatherapi`.
## Reset SQL
```sql
DROP TABLE IF EXISTS outlook_discussions;
DROP TABLE IF EXISTS outlooks;
DROP TABLE IF EXISTS outlook_runs;
```
## Verification SQL
Before or after the updated daemon starts, this query shows which outlook tables
exist:
```sql
SELECT table_name
FROM information_schema.tables
WHERE table_name IN ('outlook_runs', 'outlooks', 'outlook_discussions')
ORDER BY table_name;
```
After the updated daemon has started and recreated the tables, verify the new
run column:
```sql
SELECT column_name, is_nullable, data_type
FROM information_schema.columns
WHERE table_name = 'outlook_runs'
AND column_name = 'discussion_count';
```
Verify the discussion table indexes:
```sql
SELECT indexname
FROM pg_indexes
WHERE tablename = 'outlook_discussions'
ORDER BY indexname;
```
Verify that legacy polygon-level prose columns are gone from `outlooks`:
```sql
SELECT column_name
FROM information_schema.columns
WHERE table_name = 'outlooks'
AND column_name IN ('headline', 'summary', 'discussion');
```
The final query should return zero rows.

View File

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

View File

@@ -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
}

View File

@@ -29,10 +29,10 @@ var idTokenRE = regexp.MustCompile(`[^a-z0-9]+`)
// ConvectiveOutlookNormalizer converts: // ConvectiveOutlookNormalizer converts:
// //
// standards.SchemaRawSPCConvectiveOutlookV1 -> standards.SchemaWeatherOutlookV1 // standards.SchemaRawSPCConvectiveOutlookV1 -> standards.SchemaWeatherOutlookV2
// //
// It maps SPC GeoJSON outlook features into canonical outlook polygons and // It maps SPC GeoJSON outlook features containing the configured location into
// enriches each day with the matching required print-page discussion. // canonical outlook polygons and adds matching day-level print-page discussions.
type ConvectiveOutlookNormalizer struct{} type ConvectiveOutlookNormalizer struct{}
func (ConvectiveOutlookNormalizer) Match(e event.Event) bool { func (ConvectiveOutlookNormalizer) Match(e event.Event) bool {
@@ -50,7 +50,7 @@ func (ConvectiveOutlookNormalizer) Normalize(ctx context.Context, in event.Event
return normcommon.NormalizeJSON( return normcommon.NormalizeJSON(
in, in,
outlookNormalizer, outlookNormalizer,
standards.SchemaWeatherOutlookV1, standards.SchemaWeatherOutlookV2,
func(parsed spcprovider.RawConvectiveOutlookBundle) (model.WeatherOutlookRun, time.Time, error) { func(parsed spcprovider.RawConvectiveOutlookBundle) (model.WeatherOutlookRun, time.Time, error) {
return buildConvectiveOutlook(parsed, fallbackAsOf) return buildConvectiveOutlook(parsed, fallbackAsOf)
}, },
@@ -76,8 +76,7 @@ func buildConvectiveOutlook(bundle spcprovider.RawConvectiveOutlookBundle, fallb
if err := validateProductMetadata(product); err != nil { if err := validateProductMetadata(product); err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err return model.WeatherOutlookRun{}, time.Time{}, err
} }
discussion, ok := discussions[product.Day] if _, ok := discussions[product.Day]; !ok {
if !ok {
return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: discussion for day %d is required", product.Key, product.Day) return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: discussion for day %d is required", product.Key, product.Day)
} }
@@ -87,17 +86,36 @@ func buildConvectiveOutlook(bundle spcprovider.RawConvectiveOutlookBundle, fallb
} }
for i, feature := range collection.Features { for i, feature := range collection.Features {
outlook, err := mapFeature(product, feature, i, point, discussion) 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)
if err != nil { if err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err return model.WeatherOutlookRun{}, time.Time{}, err
} }
if latestIssue.IsZero() || outlook.IssuedAt.After(latestIssue) { if latestIssue.IsZero() || outlook.IssuedAt.After(latestIssue) {
latestIssue = outlook.IssuedAt latestIssue = outlook.IssuedAt
} }
if !outlook.ContainsLocation {
continue
}
outlooks = append(outlooks, outlook) outlooks = append(outlooks, outlook)
} }
} }
runDiscussions, err := buildOutlookDiscussions(outlooks, discussions)
if err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err
}
asOf := latestIssue asOf := latestIssue
if asOf.IsZero() { if asOf.IsZero() {
asOf = latestDiscussionUpdated asOf = latestDiscussionUpdated
@@ -122,6 +140,7 @@ func buildConvectiveOutlook(bundle spcprovider.RawConvectiveOutlookBundle, fallb
AsOf: asOf.UTC(), AsOf: asOf.UTC(),
IssuedAt: issuedAt, IssuedAt: issuedAt,
Outlooks: outlooks, Outlooks: outlooks,
Discussions: runDiscussions,
} }
return run, run.AsOf, nil return run, run.AsOf, nil
} }
@@ -164,6 +183,43 @@ func parseDiscussions(pages []spcprovider.RawDiscussionPage) (map[int]parsedDisc
return out, latestUpdated, nil return out, latestUpdated, nil
} }
func buildOutlookDiscussions(outlooks []model.WeatherOutlook, discussions map[int]parsedDiscussion) ([]model.WeatherOutlookDiscussion, error) {
daysWithOutlooks := map[int]bool{}
for _, outlook := range outlooks {
daysWithOutlooks[outlook.Day] = true
}
days := make([]int, 0, len(daysWithOutlooks))
for day := range daysWithOutlooks {
days = append(days, day)
}
sort.Ints(days)
out := make([]model.WeatherOutlookDiscussion, 0, len(days))
for _, day := range days {
disc, ok := discussions[day]
if !ok {
return nil, fmt.Errorf("discussion for retained day %d is required", day)
}
out = append(out, model.WeatherOutlookDiscussion{
Day: day,
Headline: disc.Headline,
Summary: disc.Summary,
Discussion: disc.Discussion,
UpdatedAt: utcTimePtr(disc.UpdatedAt),
})
}
return out, nil
}
func utcTimePtr(t *time.Time) *time.Time {
if t == nil {
return nil
}
tt := t.UTC()
return &tt
}
func orderedProducts(products []spcprovider.RawOutlookProduct) []spcprovider.RawOutlookProduct { func orderedProducts(products []spcprovider.RawOutlookProduct) []spcprovider.RawOutlookProduct {
out := make([]spcprovider.RawOutlookProduct, len(products)) out := make([]spcprovider.RawOutlookProduct, len(products))
copy(out, products) copy(out, products)
@@ -208,7 +264,7 @@ func validateProductMetadata(product spcprovider.RawOutlookProduct) error {
} }
} }
func mapFeature(product spcprovider.RawOutlookProduct, feature spcprovider.GeoJSONFeature, index int, point geo.Point, discussion parsedDiscussion) (model.WeatherOutlook, error) { func mapFeature(product spcprovider.RawOutlookProduct, feature spcprovider.GeoJSONFeature, index int, point geo.Point) (model.WeatherOutlook, error) {
fieldPrefix := fmt.Sprintf("product %s feature %d", product.Key, index) fieldPrefix := fmt.Sprintf("product %s feature %d", product.Key, index)
props := feature.Properties props := feature.Properties
@@ -253,9 +309,6 @@ func mapFeature(product spcprovider.RawOutlookProduct, feature spcprovider.GeoJS
IssuedAt: issuedAt, IssuedAt: issuedAt,
ExpiresAt: validTo, ExpiresAt: validTo,
Forecaster: strings.TrimSpace(props.Forecaster), Forecaster: strings.TrimSpace(props.Forecaster),
Headline: discussion.Headline,
Summary: discussion.Summary,
Discussion: discussion.Discussion,
SourceURL: strings.TrimSpace(product.URL), SourceURL: strings.TrimSpace(product.URL),
ImageURL: "", ImageURL: "",
ContainsLocation: containsLocation, ContainsLocation: containsLocation,

View File

@@ -27,8 +27,8 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
if err != nil { if err != nil {
t.Fatalf("Normalize() error = %v", err) t.Fatalf("Normalize() error = %v", err)
} }
if out.Schema != standards.SchemaWeatherOutlookV1 { if out.Schema != standards.SchemaWeatherOutlookV2 {
t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV1) t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV2)
} }
if out.Kind != event.Kind(standards.KindOutlook) { if out.Kind != event.Kind(standards.KindOutlook) {
t.Fatalf("Kind = %q, want outlook", out.Kind) t.Fatalf("Kind = %q, want outlook", out.Kind)
@@ -54,8 +54,23 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
if run.Latitude == nil || *run.Latitude != 38.5 || run.Longitude == nil || *run.Longitude != -90.5 { if run.Latitude == nil || *run.Latitude != 38.5 || run.Longitude == nil || *run.Longitude != -90.5 {
t.Fatalf("coordinates = %v,%v", run.Latitude, run.Longitude) t.Fatalf("coordinates = %v,%v", run.Latitude, run.Longitude)
} }
if len(run.Outlooks) != 9 { if len(run.Outlooks) != 4 {
t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks)) t.Fatalf("Outlooks length = %d, want 4", len(run.Outlooks))
}
assertAllOutlooksContainLocation(t, run.Outlooks)
assertDiscussionDays(t, run.Discussions, 1)
day1Discussion := run.Discussions[0]
if day1Discussion.Headline != "Day 1 Convective Outlook" {
t.Fatalf("day 1 Headline = %q", day1Discussion.Headline)
}
if !strings.Contains(day1Discussion.Summary, "central Plains") {
t.Fatalf("day 1 Summary = %q", day1Discussion.Summary)
}
if !strings.Contains(day1Discussion.Discussion, "...DISCUSSION...") {
t.Fatalf("day 1 Discussion missing product text: %q", day1Discussion.Discussion)
}
if !strings.HasPrefix(day1Discussion.Discussion, "SPC AC 111234") {
t.Fatalf("day 1 Discussion = %q, want SPC product code prefix", day1Discussion.Discussion)
} }
got := run.Outlooks[0] got := run.Outlooks[0]
@@ -88,18 +103,6 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
if !got.ContainsLocation { if !got.ContainsLocation {
t.Fatalf("ContainsLocation = false, want true") t.Fatalf("ContainsLocation = false, want true")
} }
if got.Headline != "Day 1 Convective Outlook" {
t.Fatalf("Headline = %q", got.Headline)
}
if !strings.Contains(got.Summary, "central Plains") {
t.Fatalf("Summary = %q", got.Summary)
}
if !strings.Contains(got.Discussion, "...DISCUSSION...") {
t.Fatalf("Discussion missing product text: %q", got.Discussion)
}
if !strings.HasPrefix(got.Discussion, "SPC AC 111234") {
t.Fatalf("Discussion = %q, want SPC product code prefix", got.Discussion)
}
if got.ID != "spc-convective-day1-categorical-slgt-2026-06-11T12:34:56Z-2026-06-11T13:00:00Z-0" { if got.ID != "spc-convective-day1-categorical-slgt-2026-06-11T12:34:56Z-2026-06-11T13:00:00Z-0" {
t.Fatalf("ID = %q", got.ID) t.Fatalf("ID = %q", got.ID)
} }
@@ -115,13 +118,39 @@ func TestConvectiveOutlookNormalizerAcceptsTypedSourcePayload(t *testing.T) {
t.Fatalf("Normalize() error = %v", err) t.Fatalf("Normalize() error = %v", err)
} }
run := out.Payload.(model.WeatherOutlookRun) run := out.Payload.(model.WeatherOutlookRun)
if len(run.Outlooks) != 9 { if len(run.Outlooks) != 4 {
t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks)) t.Fatalf("Outlooks length = %d, want 4", len(run.Outlooks))
}
assertAllOutlooksContainLocation(t, run.Outlooks)
assertDiscussionDays(t, run.Discussions, 1)
}
func TestConvectiveOutlookNormalizerEmitsEmptyLocalRunOutsidePolygons(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if len(run.Outlooks) != 0 {
t.Fatalf("Outlooks length = %d, want 0", len(run.Outlooks))
}
if len(run.Discussions) != 0 {
t.Fatalf("Discussions length = %d, want 0", len(run.Discussions))
}
wantAsOf := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)
if !run.AsOf.Equal(wantAsOf) {
t.Fatalf("AsOf = %s, want latest product issue time %s", run.AsOf, wantAsOf)
}
if run.IssuedAt == nil || !run.IssuedAt.Equal(wantAsOf) {
t.Fatalf("IssuedAt = %v, want latest product issue time %s", run.IssuedAt, wantAsOf)
}
if out.EffectiveAt == nil || !out.EffectiveAt.Equal(run.AsOf) {
t.Fatalf("EffectiveAt = %v, want run AsOf %s", out.EffectiveAt, run.AsOf)
} }
} }
func TestConvectiveOutlookNormalizerOrdersProductsByDayAndType(t *testing.T) { func TestConvectiveOutlookNormalizerOrdersProductsByDayAndType(t *testing.T) {
bundle := spcBundle(t, 0, 0) bundle := spcBundle(t, 38.5, -90.5)
for i, j := 0, len(bundle.Products)-1; i < j; i, j = i+1, j-1 { for i, j := 0, len(bundle.Products)-1; i < j; i, j = i+1, j-1 {
bundle.Products[i], bundle.Products[j] = bundle.Products[j], bundle.Products[i] bundle.Products[i], bundle.Products[j] = bundle.Products[j], bundle.Products[i]
} }
@@ -151,11 +180,12 @@ func TestConvectiveOutlookNormalizerOrdersProductsByDayAndType(t *testing.T) {
} }
func TestConvectiveOutlookNormalizerMapsProbabilisticOutlookTypes(t *testing.T) { func TestConvectiveOutlookNormalizerMapsProbabilisticOutlookTypes(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0))) out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil { if err != nil {
t.Fatalf("Normalize() error = %v", err) t.Fatalf("Normalize() error = %v", err)
} }
run := out.Payload.(model.WeatherOutlookRun) run := out.Payload.(model.WeatherOutlookRun)
assertAllOutlooksContainLocation(t, run.Outlooks)
for _, outlookType := range []string{ for _, outlookType := range []string{
spcprovider.OutlookTypeTornado, spcprovider.OutlookTypeTornado,
spcprovider.OutlookTypeHail, spcprovider.OutlookTypeHail,
@@ -167,19 +197,75 @@ func TestConvectiveOutlookNormalizerMapsProbabilisticOutlookTypes(t *testing.T)
} }
} }
func TestConvectiveOutlookNormalizerContainsLocationFalseOutsidePolygon(t *testing.T) { func TestConvectiveOutlookNormalizerSkipsEmptyGeometryCollectionPlaceholder(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0))) bundle := spcBundle(t, 36, -99)
replaced := false
for i := range bundle.Products {
if bundle.Products[i].Day == 2 && bundle.Products[i].OutlookType == spcprovider.OutlookTypeTornado {
bundle.Products[i].Body = json.RawMessage(emptyGeometryCollectionGeoJSON())
replaced = true
}
}
if !replaced {
t.Fatalf("test setup did not find day 2 tornado product")
}
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err != nil { if err != nil {
t.Fatalf("Normalize() error = %v", err) t.Fatalf("Normalize() error = %v", err)
} }
run := out.Payload.(model.WeatherOutlookRun) run := out.Payload.(model.WeatherOutlookRun)
if run.Outlooks[0].ContainsLocation { if len(run.Outlooks) != 3 {
t.Fatalf("ContainsLocation = true, want false") t.Fatalf("Outlooks length = %d, want 3", len(run.Outlooks))
}
assertAllOutlooksContainLocation(t, run.Outlooks)
assertDiscussionDays(t, run.Discussions, 2)
if got := findOutlook(run.Outlooks, 2, spcprovider.OutlookTypeTornado); got != nil {
t.Fatalf("day 2 tornado outlook = %+v, want nil placeholder skipped", *got)
}
wantAsOf := time.Date(2026, 6, 12, 10, 0, 0, 0, time.UTC)
if !run.AsOf.Equal(wantAsOf) {
t.Fatalf("AsOf = %s, want placeholder ISSUE_ISO %s", run.AsOf, wantAsOf)
}
if out.EffectiveAt == nil || !out.EffectiveAt.Equal(wantAsOf) {
t.Fatalf("EffectiveAt = %v, want placeholder ISSUE_ISO %s", out.EffectiveAt, wantAsOf)
} }
} }
func TestConvectiveOutlookNormalizerIncludesOnlyDayWithContainingPolygons(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 36, -99)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if len(run.Outlooks) == 0 {
t.Fatalf("Outlooks length = 0, want retained day 2 outlooks")
}
assertAllOutlooksContainLocation(t, run.Outlooks)
for i, outlook := range run.Outlooks {
if outlook.Day != 2 {
t.Fatalf("outlook[%d].Day = %d, want 2", i, outlook.Day)
}
}
assertDiscussionDays(t, run.Discussions, 2)
}
func TestConvectiveOutlookNormalizerUsesOneDiscussionForMultipleSameDayOutlooks(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if got := countOutlooksByDay(run.Outlooks, 1); got < 2 {
t.Fatalf("day 1 outlook count = %d, want multiple", got)
}
assertAllOutlooksContainLocation(t, run.Outlooks)
assertDiscussionDays(t, run.Discussions, 1)
}
func TestConvectiveOutlookNormalizerPreservesCorrectionMarker(t *testing.T) { func TestConvectiveOutlookNormalizerPreservesCorrectionMarker(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0))) bundle := spcBundle(t, 36, -99)
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err != nil { if err != nil {
t.Fatalf("Normalize() error = %v", err) t.Fatalf("Normalize() error = %v", err)
} }
@@ -188,11 +274,12 @@ func TestConvectiveOutlookNormalizerPreservesCorrectionMarker(t *testing.T) {
if got == nil { if got == nil {
t.Fatalf("missing day 2 tornado outlook") t.Fatalf("missing day 2 tornado outlook")
} }
if !strings.Contains(got.Headline, "CORR 1") { assertDiscussionDays(t, run.Discussions, 2)
t.Fatalf("Headline = %q, want correction marker", got.Headline) if !strings.Contains(run.Discussions[0].Headline, "CORR 1") {
t.Fatalf("day 2 headline = %q, want correction marker", run.Discussions[0].Headline)
} }
if !strings.Contains(got.Discussion, "CORR 1") { if !strings.Contains(run.Discussions[0].Discussion, "CORR 1") {
t.Fatalf("Discussion = %q, want correction marker", got.Discussion) t.Fatalf("day 2 discussion = %q, want correction marker", run.Discussions[0].Discussion)
} }
} }
@@ -258,6 +345,19 @@ func TestConvectiveOutlookNormalizerRejectsMissingLabel(t *testing.T) {
} }
} }
func TestConvectiveOutlookNormalizerRejectsMissingDiscussion(t *testing.T) {
bundle := spcBundle(t, 38.5, -90.5)
bundle.Discussions = bundle.Discussions[1:]
_, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err == nil {
t.Fatalf("Normalize() error = nil, want error")
}
if !strings.Contains(err.Error(), "product day1_categorical: discussion for day 1 is required") {
t.Fatalf("error = %q, want missing discussion context", err)
}
}
func TestConvectiveOutlookNormalizerOutputJSONShape(t *testing.T) { func TestConvectiveOutlookNormalizerOutputJSONShape(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5))) out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil { if err != nil {
@@ -268,18 +368,53 @@ func TestConvectiveOutlookNormalizerOutputJSONShape(t *testing.T) {
t.Fatalf("Marshal(payload) error = %v", err) t.Fatalf("Marshal(payload) error = %v", err)
} }
got := string(raw) got := string(raw)
for _, want := range []string{`"asOf"`, `"outlooks"`, `"containsLocation"`, `"geometry"`} { for _, want := range []string{`"asOf"`, `"outlooks"`, `"discussions"`, `"containsLocation"`, `"geometry"`} {
if !strings.Contains(got, want) { if !strings.Contains(got, want) {
t.Fatalf("payload JSON missing %s: %s", want, got) t.Fatalf("payload JSON missing %s: %s", want, got)
} }
} }
for _, unwanted := range []string{`"products"`, `"discussions"`, `"fetchedAt"`, `"body"`} { outlookStart := strings.Index(got, `"outlooks"`)
discussionStart := strings.Index(got, `"discussions"`)
if outlookStart == -1 || discussionStart == -1 || discussionStart <= outlookStart {
t.Fatalf("payload JSON has unexpected outlook/discussion order: %s", got)
}
outlookJSON := got[outlookStart:discussionStart]
for _, unwanted := range []string{`"headline"`, `"summary"`, `"discussion"`} {
if strings.Contains(outlookJSON, unwanted) {
t.Fatalf("outlook JSON exposed polygon-level prose key %s: %s", unwanted, got)
}
}
for _, unwanted := range []string{`"products"`, `"fetchedAt"`, `"body"`} {
if strings.Contains(got, unwanted) { if strings.Contains(got, unwanted) {
t.Fatalf("payload JSON exposed raw key %s: %s", unwanted, got) t.Fatalf("payload JSON exposed raw key %s: %s", unwanted, got)
} }
} }
} }
func emptyGeometryCollectionGeoJSON() []byte {
return []byte(`{
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"properties": {
"VALID_ISO": "2026-06-12T12:00:00Z",
"EXPIRE_ISO": "2026-06-13T12:00:00Z",
"ISSUE_ISO": "2026-06-12T10:00:00Z",
"FORECASTER": "DOE",
"LABEL": "Less Than 2% All Areas",
"LABEL2": "",
"DN": 0
},
"geometry": {
"type": "GeometryCollection",
"geometries": []
}
}
]
}`)
}
func spcRawEvent(t *testing.T, bundle spcprovider.RawConvectiveOutlookBundle) event.Event { func spcRawEvent(t *testing.T, bundle spcprovider.RawConvectiveOutlookBundle) event.Event {
t.Helper() t.Helper()
raw, err := json.Marshal(bundle) raw, err := json.Marshal(bundle)
@@ -351,6 +486,37 @@ func findOutlook(outlooks []model.WeatherOutlook, day int, outlookType string) *
return nil return nil
} }
func countOutlooksByDay(outlooks []model.WeatherOutlook, day int) int {
count := 0
for _, outlook := range outlooks {
if outlook.Day == day {
count++
}
}
return count
}
func assertAllOutlooksContainLocation(t *testing.T, outlooks []model.WeatherOutlook) {
t.Helper()
for i, outlook := range outlooks {
if !outlook.ContainsLocation {
t.Fatalf("outlook[%d].ContainsLocation = false, want true", i)
}
}
}
func assertDiscussionDays(t *testing.T, discussions []model.WeatherOutlookDiscussion, want ...int) {
t.Helper()
if len(discussions) != len(want) {
t.Fatalf("Discussions length = %d, want %d", len(discussions), len(want))
}
for i, day := range want {
if discussions[i].Day != day {
t.Fatalf("Discussions[%d].Day = %d, want %d", i, discussions[i].Day, day)
}
}
}
func assertTime(t *testing.T, name string, got time.Time, year int, month time.Month, day int, hour int, minute int, second int) { func assertTime(t *testing.T, name string, got time.Time, year int, month time.Month, day int, hour int, minute int, second int) {
t.Helper() t.Helper()
want := time.Date(year, month, day, hour, minute, second, 0, time.UTC) want := time.Date(year, month, day, hour, minute, second, 0, time.UTC)

View File

@@ -33,6 +33,11 @@ type GeoJSONProperties struct {
DN *int `json:"DN"` DN *int `json:"DN"`
} }
type geometryMetadata struct {
Type string `json:"type"`
Geometries []json.RawMessage `json:"geometries"`
}
// DecodeGeoJSON decodes an SPC GeoJSON outlook product and compacts feature // DecodeGeoJSON decodes an SPC GeoJSON outlook product and compacts feature
// geometry JSON for stable downstream storage. // geometry JSON for stable downstream storage.
func DecodeGeoJSON(raw []byte) (GeoJSONFeatureCollection, error) { func DecodeGeoJSON(raw []byte) (GeoJSONFeatureCollection, error) {
@@ -50,6 +55,16 @@ func DecodeGeoJSON(raw []byte) (GeoJSONFeatureCollection, error) {
return collection, nil return collection, nil
} }
// IsEmptyGeometryCollection reports whether raw is SPC's no-polygon placeholder
// geometry shape: a GeometryCollection with no child geometries.
func IsEmptyGeometryCollection(raw json.RawMessage) bool {
var meta geometryMetadata
if err := json.Unmarshal(raw, &meta); err != nil {
return false
}
return meta.Type == "GeometryCollection" && len(meta.Geometries) == 0
}
func (p *GeoJSONProperties) UnmarshalJSON(raw []byte) error { func (p *GeoJSONProperties) UnmarshalJSON(raw []byte) error {
type alias GeoJSONProperties type alias GeoJSONProperties
var aux struct { var aux struct {

View File

@@ -66,6 +66,43 @@ func TestDecodeGeoJSONParsesSeverityRankString(t *testing.T) {
} }
} }
func TestIsEmptyGeometryCollection(t *testing.T) {
tests := []struct {
name string
raw string
want bool
}{
{
name: "empty geometry collection",
raw: `{"type":"GeometryCollection","geometries":[]}`,
want: true,
},
{
name: "non-empty geometry collection",
raw: `{"type":"GeometryCollection","geometries":[{"type":"Polygon","coordinates":[]} ]}`,
want: false,
},
{
name: "polygon",
raw: `{"type":"Polygon","coordinates":[]}`,
want: false,
},
{
name: "invalid json",
raw: `{`,
want: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := IsEmptyGeometryCollection([]byte(tt.raw)); got != tt.want {
t.Fatalf("IsEmptyGeometryCollection() = %v, want %v", got, tt.want)
}
})
}
}
func TestParseISOTimestampTrimsAndReturnsUTC(t *testing.T) { func TestParseISOTimestampTrimsAndReturnsUTC(t *testing.T) {
got, err := ParseISOTimestamp(" 2026-06-11T12:34:56Z ") got, err := ParseISOTimestamp(" 2026-06-11T12:34:56Z ")
if err != nil { if err != nil {

View File

@@ -11,7 +11,7 @@
// - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion // - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion
// - weather.weather_story.v1 -> model.WeatherStoryRun // - weather.weather_story.v1 -> model.WeatherStoryRun
// - weather.alert.v1 -> model.WeatherAlertRun // - weather.alert.v1 -> model.WeatherAlertRun
// - weather.outlook.v1 -> model.WeatherOutlookRun // - weather.outlook.v2 -> model.WeatherOutlookRun
// //
// Parent/child relationships: // Parent/child relationships:
// - observations.event_id -> observation_present_weather.event_id // - observations.event_id -> observation_present_weather.event_id
@@ -21,6 +21,7 @@
// - alert_runs.event_id -> alerts.run_event_id // - alert_runs.event_id -> alerts.run_event_id
// - alerts.(run_event_id, alert_index) -> alert_references.(run_event_id, alert_index) // - alerts.(run_event_id, alert_index) -> alert_references.(run_event_id, alert_index)
// - outlook_runs.event_id -> outlooks.run_event_id // - outlook_runs.event_id -> outlooks.run_event_id
// - outlook_runs.event_id -> outlook_discussions.run_event_id
// //
// Dedupe and retention behavior: // Dedupe and retention behavior:
// - Parent primary keys (event_id): observations, forecasts, alert_runs, outlook_runs. // - Parent primary keys (event_id): observations, forecasts, alert_runs, outlook_runs.
@@ -39,6 +40,7 @@
// - alert_references.as_of // - alert_references.as_of
// - outlook_runs.as_of // - outlook_runs.as_of
// - outlooks.as_of // - outlooks.as_of
// - outlook_discussions.as_of
// //
// Envelope field mapping (shared parent columns) // Envelope field mapping (shared parent columns)
// //
@@ -196,6 +198,7 @@
// - sent TIMESTAMPTZ NULL -> payload.alerts[i].sent // - sent TIMESTAMPTZ NULL -> payload.alerts[i].sent
// - effective TIMESTAMPTZ NULL -> payload.alerts[i].effective // - effective TIMESTAMPTZ NULL -> payload.alerts[i].effective
// - onset TIMESTAMPTZ NULL -> payload.alerts[i].onset // - onset TIMESTAMPTZ NULL -> payload.alerts[i].onset
// - ends TIMESTAMPTZ NULL -> payload.alerts[i].ends
// - expires TIMESTAMPTZ NULL -> payload.alerts[i].expires // - expires TIMESTAMPTZ NULL -> payload.alerts[i].expires
// - area_description TEXT NULL -> payload.alerts[i].areaDescription // - area_description TEXT NULL -> payload.alerts[i].areaDescription
// - sender_name TEXT NULL -> payload.alerts[i].senderName // - sender_name TEXT NULL -> payload.alerts[i].senderName
@@ -227,6 +230,7 @@
// - as_of TIMESTAMPTZ -> payload.asOf // - as_of TIMESTAMPTZ -> payload.asOf
// - issued_at TIMESTAMPTZ NULL -> payload.issuedAt // - issued_at TIMESTAMPTZ NULL -> payload.issuedAt
// - outlook_count INTEGER -> len(payload.outlooks) // - outlook_count INTEGER -> len(payload.outlooks)
// - discussion_count INTEGER -> len(payload.discussions)
// //
// 11. outlooks (PK: run_event_id, outlook_index) // 11. outlooks (PK: run_event_id, outlook_index)
// //
@@ -246,14 +250,22 @@
// - issued_at TIMESTAMPTZ -> payload.outlooks[i].issuedAt // - issued_at TIMESTAMPTZ -> payload.outlooks[i].issuedAt
// - expires_at TIMESTAMPTZ -> payload.outlooks[i].expiresAt // - expires_at TIMESTAMPTZ -> payload.outlooks[i].expiresAt
// - forecaster TEXT NULL -> payload.outlooks[i].forecaster // - forecaster TEXT NULL -> payload.outlooks[i].forecaster
// - headline TEXT NULL -> payload.outlooks[i].headline
// - summary TEXT NULL -> payload.outlooks[i].summary
// - discussion TEXT NULL -> payload.outlooks[i].discussion
// - source_url TEXT NULL -> payload.outlooks[i].sourceUrl // - source_url TEXT NULL -> payload.outlooks[i].sourceUrl
// - image_url TEXT NULL -> payload.outlooks[i].imageUrl // - image_url TEXT NULL -> payload.outlooks[i].imageUrl
// - contains_location BOOLEAN -> payload.outlooks[i].containsLocation // - contains_location BOOLEAN -> payload.outlooks[i].containsLocation
// - geometry_json TEXT -> compact JSON payload.outlooks[i].geometry // - geometry_json TEXT -> compact JSON payload.outlooks[i].geometry
// //
// 12. outlook_discussions (PK: run_event_id, discussion_index)
//
// - run_event_id TEXT -> outlook_runs.event_id / payload.discussions[i]
// - discussion_index INTEGER -> i (array position in payload.discussions)
// - as_of TIMESTAMPTZ -> payload.asOf (copied from parent)
// - day INTEGER -> payload.discussions[i].day
// - headline TEXT NULL -> payload.discussions[i].headline
// - summary TEXT NULL -> payload.discussions[i].summary
// - discussion TEXT NULL -> payload.discussions[i].discussion
// - updated_at TIMESTAMPTZ NULL -> payload.discussions[i].updatedAt
//
// Reconstructing canonical JSON payloads // Reconstructing canonical JSON payloads
// //
// - WeatherObservation: // - WeatherObservation:
@@ -274,6 +286,7 @@
// ordered by reference_index to rebuild references per alert. // ordered by reference_index to rebuild references per alert.
// //
// - WeatherOutlookRun: // - WeatherOutlookRun:
// read one row from outlook_runs, then join outlooks by run_event_id ordered // read one row from outlook_runs, join outlooks by run_event_id ordered by
// by outlook_index to rebuild outlooks. // outlook_index to rebuild outlooks, then join outlook_discussions by
// run_event_id ordered by discussion_index to rebuild discussions.
package postgres package postgres

View File

@@ -0,0 +1,33 @@
package postgres
import (
"os"
"strings"
"testing"
)
func TestDocumentedOutlookDiscussionStorage(t *testing.T) {
for _, path := range []string{
"../../../docs/integrations/postgres.md",
"../../../docs/internal/postgres-sink.md",
} {
t.Run(path, func(t *testing.T) {
raw, err := os.ReadFile(path)
if err != nil {
t.Fatalf("ReadFile(%s) error = %v", path, err)
}
doc := string(raw)
for _, want := range []string{
tableOutlookDiscussions,
"discussion_count",
"discussion_index",
"weather.outlook.v2",
} {
if !strings.Contains(doc, want) {
t.Fatalf("%s missing %q", path, want)
}
}
})
}
}

View File

@@ -27,7 +27,7 @@ func mapPostgresEvent(_ context.Context, e fkevent.Event) ([]fksinks.PostgresWri
return mapWeatherStoryEvent(e) return mapWeatherStoryEvent(e)
case standards.SchemaWeatherAlertV1: case standards.SchemaWeatherAlertV1:
return mapAlertEvent(e) return mapAlertEvent(e)
case standards.SchemaWeatherOutlookV1: case standards.SchemaWeatherOutlookV2:
return mapOutlookEvent(e) return mapOutlookEvent(e)
default: default:
return nil, nil return nil, nil
@@ -302,6 +302,7 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
"sent": nullableTime(a.Sent), "sent": nullableTime(a.Sent),
"effective": nullableTime(a.Effective), "effective": nullableTime(a.Effective),
"onset": nullableTime(a.Onset), "onset": nullableTime(a.Onset),
"ends": nullableTime(a.Ends),
"expires": nullableTime(a.Expires), "expires": nullableTime(a.Expires),
"area_description": nullableString(a.AreaDescription), "area_description": nullableString(a.AreaDescription),
"sender_name": nullableString(a.SenderName), "sender_name": nullableString(a.SenderName),
@@ -339,7 +340,11 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
} }
asOf := run.AsOf.UTC() asOf := run.AsOf.UTC()
writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks)) if err := validateOutlookDiscussions(run.Discussions); err != nil {
return nil, err
}
writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks)+len(run.Discussions))
writes = append(writes, fksinks.PostgresWrite{ writes = append(writes, fksinks.PostgresWrite{
Table: tableOutlookRuns, Table: tableOutlookRuns,
Values: parentEventValues(e, map[string]any{ Values: parentEventValues(e, map[string]any{
@@ -350,6 +355,7 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
"as_of": asOf, "as_of": asOf,
"issued_at": nullableTime(run.IssuedAt), "issued_at": nullableTime(run.IssuedAt),
"outlook_count": len(run.Outlooks), "outlook_count": len(run.Outlooks),
"discussion_count": len(run.Discussions),
}), }),
}) })
@@ -381,9 +387,6 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
"issued_at": outlook.IssuedAt.UTC(), "issued_at": outlook.IssuedAt.UTC(),
"expires_at": outlook.ExpiresAt.UTC(), "expires_at": outlook.ExpiresAt.UTC(),
"forecaster": nullableString(outlook.Forecaster), "forecaster": nullableString(outlook.Forecaster),
"headline": nullableString(outlook.Headline),
"summary": nullableString(outlook.Summary),
"discussion": nullableString(outlook.Discussion),
"source_url": nullableString(outlook.SourceURL), "source_url": nullableString(outlook.SourceURL),
"image_url": nullableString(outlook.ImageURL), "image_url": nullableString(outlook.ImageURL),
"contains_location": outlook.ContainsLocation, "contains_location": outlook.ContainsLocation,
@@ -392,6 +395,22 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
}) })
} }
for i, discussion := range run.Discussions {
writes = append(writes, fksinks.PostgresWrite{
Table: tableOutlookDiscussions,
Values: map[string]any{
"run_event_id": e.ID,
"discussion_index": i,
"as_of": asOf,
"day": discussion.Day,
"headline": nullableString(discussion.Headline),
"summary": nullableString(discussion.Summary),
"discussion": nullableString(discussion.Discussion),
"updated_at": nullableTime(discussion.UpdatedAt),
},
})
}
return writes, nil return writes, nil
} }
@@ -423,6 +442,28 @@ func validateOutlook(outlook model.WeatherOutlook, index int) error {
if len(outlook.Geometry) == 0 { if len(outlook.Geometry) == 0 {
return fmt.Errorf("decode outlook payload: outlooks[%d].geometry is required", index) return fmt.Errorf("decode outlook payload: outlooks[%d].geometry is required", index)
} }
if !outlook.ContainsLocation {
return fmt.Errorf("decode outlook payload: outlooks[%d].containsLocation must be true", index)
}
return nil
}
func validateOutlookDiscussions(discussions []model.WeatherOutlookDiscussion) error {
seenDays := map[int]int{}
for i, discussion := range discussions {
if discussion.Day < 1 || discussion.Day > 3 {
return fmt.Errorf("decode outlook payload: discussions[%d].day must be 1, 2, or 3", i)
}
if strings.TrimSpace(discussion.Headline) == "" &&
strings.TrimSpace(discussion.Summary) == "" &&
strings.TrimSpace(discussion.Discussion) == "" {
return fmt.Errorf("decode outlook payload: discussions[%d] headline, summary, or discussion is required", i)
}
if first, ok := seenDays[discussion.Day]; ok {
return fmt.Errorf("decode outlook payload: discussions[%d].day duplicates discussions[%d].day %d", i, first, discussion.Day)
}
seenDays[discussion.Day] = i
}
return nil return nil
} }

View File

@@ -103,6 +103,8 @@ func TestMapPostgresEventForecastStructPayload(t *testing.T) {
func TestMapPostgresEventAlertStructPayload(t *testing.T) { func TestMapPostgresEventAlertStructPayload(t *testing.T) {
sent := time.Date(2026, 3, 16, 17, 0, 0, 0, time.UTC) 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{ run := model.WeatherAlertRun{
AsOf: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC), AsOf: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
Alerts: []model.WeatherAlert{ Alerts: []model.WeatherAlert{
@@ -110,6 +112,8 @@ func TestMapPostgresEventAlertStructPayload(t *testing.T) {
ID: "urn:alert:1", ID: "urn:alert:1",
Headline: "Winter Weather Advisory", Headline: "Winter Weather Advisory",
Severity: "Moderate", Severity: "Moderate",
Ends: &ends,
Expires: &expires,
References: []model.AlertReference{ References: []model.AlertReference{
{ID: "urn:ref:1", Sent: &sent}, {ID: "urn:ref:1", Sent: &sent},
{Identifier: "ref-two"}, {Identifier: "ref-two"},
@@ -145,6 +149,20 @@ func TestMapPostgresEventAlertStructPayload(t *testing.T) {
if got := firstAlert.Values["reference_count"]; got != 2 { if got := firstAlert.Values["reference_count"]; got != 2 {
t.Fatalf("alerts reference_count = %#v, want 2", got) 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) assertAllWritesIncludeAllColumns(t, writes)
} }
@@ -243,6 +261,7 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
lat := 38.6239 lat := 38.6239
lon := -90.3571 lon := -90.3571
issuedAt := time.Date(2026, 6, 11, 19, 45, 0, 0, time.FixedZone("UTC-5", -5*60*60)) issuedAt := time.Date(2026, 6, 11, 19, 45, 0, 0, time.FixedZone("UTC-5", -5*60*60))
updatedAt := time.Date(2026, 6, 11, 21, 15, 0, 0, time.FixedZone("UTC-5", -5*60*60))
severity := 3 severity := 3
run := model.WeatherOutlookRun{ run := model.WeatherOutlookRun{
LocationID: "stl", LocationID: "stl",
@@ -266,9 +285,6 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
IssuedAt: issuedAt, IssuedAt: issuedAt,
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
Forecaster: "SMITH", Forecaster: "SMITH",
Headline: "Day 1 Convective Outlook",
Summary: "Severe thunderstorms are possible.",
Discussion: "Full discussion text.",
SourceURL: "https://example.invalid/day1.geojson", SourceURL: "https://example.invalid/day1.geojson",
ContainsLocation: true, ContainsLocation: true,
Geometry: json.RawMessage(`{ "type" : "Polygon", "coordinates" : [ [ [ -91.0, 38.0 ], [ -90.0, 38.0 ], [ -90.0, 39.0 ], [ -91.0, 39.0 ], [ -91.0, 38.0 ] ] ] }`), Geometry: json.RawMessage(`{ "type" : "Polygon", "coordinates" : [ [ [ -91.0, 38.0 ], [ -90.0, 38.0 ], [ -90.0, 39.0 ], [ -91.0, 39.0 ], [ -91.0, 38.0 ] ] ] }`),
@@ -284,18 +300,27 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC), IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: false, ContainsLocation: true,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-100,35],[-98,35],[-98,37],[-100,37],[-100,35]]]}`), Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-100,35],[-98,35],[-98,37],[-100,37],[-100,35]]]}`),
}, },
}, },
Discussions: []model.WeatherOutlookDiscussion{
{
Day: 1,
Headline: "Day 1 Convective Outlook",
Summary: "Severe thunderstorms are possible.",
Discussion: "Full discussion text.",
UpdatedAt: &updatedAt,
},
},
} }
writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run)) writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err != nil { if err != nil {
t.Fatalf("mapPostgresEvent() error = %v", err) t.Fatalf("mapPostgresEvent() error = %v", err)
} }
if len(writes) != 3 { if len(writes) != 4 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 3", len(writes)) t.Fatalf("mapPostgresEvent() writes len = %d, want 4", len(writes))
} }
if writes[0].Table != tableOutlookRuns { if writes[0].Table != tableOutlookRuns {
t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns) t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns)
@@ -303,6 +328,9 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
if got := writes[0].Values["outlook_count"]; got != 2 { if got := writes[0].Values["outlook_count"]; got != 2 {
t.Fatalf("outlook_runs outlook_count = %#v, want 2", got) t.Fatalf("outlook_runs outlook_count = %#v, want 2", got)
} }
if got := writes[0].Values["discussion_count"]; got != 1 {
t.Fatalf("outlook_runs discussion_count = %#v, want 1", got)
}
if got := writes[0].Values["issued_at"]; got != issuedAt.UTC() { if got := writes[0].Values["issued_at"]; got != issuedAt.UTC() {
t.Fatalf("outlook_runs issued_at = %#v, want UTC %s", got, issuedAt.UTC()) t.Fatalf("outlook_runs issued_at = %#v, want UTC %s", got, issuedAt.UTC())
} }
@@ -324,15 +352,64 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
if got := writes[1].Values["geometry_json"]; got != `{"type":"Polygon","coordinates":[[[-91.0,38.0],[-90.0,38.0],[-90.0,39.0],[-91.0,39.0],[-91.0,38.0]]]}` { if got := writes[1].Values["geometry_json"]; got != `{"type":"Polygon","coordinates":[[[-91.0,38.0],[-90.0,38.0],[-90.0,39.0],[-91.0,39.0],[-91.0,38.0]]]}` {
t.Fatalf("first geometry_json = %#v", got) t.Fatalf("first geometry_json = %#v", got)
} }
if got := writes[2].Values["contains_location"]; got != false { if got := writes[2].Values["contains_location"]; got != true {
t.Fatalf("second contains_location = %#v, want false", got) t.Fatalf("second contains_location = %#v, want true", got)
}
if writes[3].Table != tableOutlookDiscussions {
t.Fatalf("writes[3].Table = %q, want %q", writes[3].Table, tableOutlookDiscussions)
}
if got := writes[3].Values["discussion_index"]; got != 0 {
t.Fatalf("discussion_index = %#v, want 0", got)
}
if got := writes[3].Values["as_of"]; got != run.AsOf.UTC() {
t.Fatalf("discussion as_of = %#v, want %s", got, run.AsOf.UTC())
}
if got := writes[3].Values["day"]; got != 1 {
t.Fatalf("discussion day = %#v, want 1", got)
}
if got := writes[3].Values["headline"]; got != "Day 1 Convective Outlook" {
t.Fatalf("discussion headline = %#v", got)
}
if got := writes[3].Values["summary"]; got != "Severe thunderstorms are possible." {
t.Fatalf("discussion summary = %#v", got)
}
if got := writes[3].Values["discussion"]; got != "Full discussion text." {
t.Fatalf("discussion text = %#v", got)
}
if got := writes[3].Values["updated_at"]; got != updatedAt.UTC() {
t.Fatalf("discussion updated_at = %#v, want UTC %s", got, updatedAt.UTC())
}
assertAllWritesIncludeAllColumns(t, writes)
}
func TestMapPostgresEventOutlookEmptyLocalRun(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
}
writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err != nil {
t.Fatalf("mapPostgresEvent() error = %v", err)
}
if len(writes) != 1 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 1", len(writes))
}
if writes[0].Table != tableOutlookRuns {
t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns)
}
if got := writes[0].Values["outlook_count"]; got != 0 {
t.Fatalf("outlook_runs outlook_count = %#v, want 0", got)
}
if got := writes[0].Values["discussion_count"]; got != 0 {
t.Fatalf("outlook_runs discussion_count = %#v, want 0", got)
} }
assertAllWritesIncludeAllColumns(t, writes) assertAllWritesIncludeAllColumns(t, writes)
} }
func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) { func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, model.WeatherOutlookRun{})) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, model.WeatherOutlookRun{}))
if err == nil { if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error") t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error")
} }
@@ -353,6 +430,7 @@ func TestMapPostgresEventOutlookRejectsMissingIDAndProvider(t *testing.T) {
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC), IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: true,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`), Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
} }
@@ -381,7 +459,7 @@ func TestMapPostgresEventOutlookRejectsMissingIDAndProvider(t *testing.T) {
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC), AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []model.WeatherOutlook{outlook}, Outlooks: []model.WeatherOutlook{outlook},
} }
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run)) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil { if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want %q", tt.wantErr) t.Fatalf("mapPostgresEvent() error = nil, want %q", tt.wantErr)
} }
@@ -402,10 +480,11 @@ func TestMapPostgresEventOutlookRejectsMissingRequiredTimes(t *testing.T) {
Day: 1, Day: 1,
OutlookType: "categorical", OutlookType: "categorical",
Label: "SLGT", Label: "SLGT",
ContainsLocation: true,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`), Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
}}, }},
} }
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run)) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil { if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want missing time error") t.Fatalf("mapPostgresEvent() error = nil, want missing time error")
} }
@@ -428,9 +507,10 @@ func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC), IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: true,
}}, }},
} }
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run)) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil { if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want geometry error") t.Fatalf("mapPostgresEvent() error = nil, want geometry error")
} }
@@ -439,6 +519,67 @@ func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
} }
} }
func TestMapPostgresEventOutlookRejectsDuplicateDiscussionDay(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Discussions: []model.WeatherOutlookDiscussion{
{Day: 1, Discussion: "First day one discussion."},
{Day: 1, Discussion: "Duplicate day one discussion."},
},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want duplicate discussion day error")
}
if !strings.Contains(err.Error(), "discussions[1].day duplicates discussions[0].day 1") {
t.Fatalf("error = %q, want duplicate discussion day context", err)
}
}
func TestMapPostgresEventOutlookRejectsInvalidDiscussionDay(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Discussions: []model.WeatherOutlookDiscussion{{Day: 4, Discussion: "Invalid day."}},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want invalid discussion day error")
}
if !strings.Contains(err.Error(), "discussions[0].day must be 1, 2, or 3") {
t.Fatalf("error = %q, want invalid discussion day context", err)
}
}
func TestMapPostgresEventOutlookRejectsEmptyDiscussionContent(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Discussions: []model.WeatherOutlookDiscussion{{Day: 1}},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want empty discussion content error")
}
if !strings.Contains(err.Error(), "discussions[0] headline, summary, or discussion is required") {
t.Fatalf("error = %q, want empty discussion content context", err)
}
}
func TestMapPostgresEventOutlookRejectsContainsLocationFalse(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []model.WeatherOutlook{validTestOutlook()},
}
run.Outlooks[0].ContainsLocation = false
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want containsLocation error")
}
if !strings.Contains(err.Error(), "outlooks[0].containsLocation must be true") {
t.Fatalf("error = %q, want containsLocation context", err)
}
}
func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) { func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, model.WeatherStoryRun{})) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, model.WeatherStoryRun{}))
if err == nil { if err == nil {
@@ -508,6 +649,17 @@ func TestMapPostgresEventUnknownSchemaNoOp(t *testing.T) {
} }
} }
func TestMapPostgresEventLegacyOutlookSchemaNoOp(t *testing.T) {
run := model.WeatherOutlookRun{AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)}
writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
if err != nil {
t.Fatalf("mapPostgresEvent() error = %v", err)
}
if len(writes) != 0 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 0", len(writes))
}
}
func TestMapPostgresEventMalformedPayload(t *testing.T) { func TestMapPostgresEventMalformedPayload(t *testing.T) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, "bad")) _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, "bad"))
if err == nil { if err == nil {
@@ -624,6 +776,16 @@ func firstWriteForTable(writes []fksinks.PostgresWrite, table string) (fksinks.P
return fksinks.PostgresWrite{}, false 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) { func assertAllWritesIncludeAllColumns(t *testing.T, writes []fksinks.PostgresWrite) {
t.Helper() t.Helper()
colCounts := tableColumnCounts() colCounts := tableColumnCounts()
@@ -647,6 +809,23 @@ func tableColumnCounts() map[string]int {
return m return m
} }
func validTestOutlook() model.WeatherOutlook {
return model.WeatherOutlook{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.UTC),
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: true,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
}
}
func wmoCodePtr(v model.WMOCode) *model.WMOCode { func wmoCodePtr(v model.WMOCode) *model.WMOCode {
out := v out := v
return &out return &out

View File

@@ -18,6 +18,7 @@ const (
tableAlertReferences = "alert_references" tableAlertReferences = "alert_references"
tableOutlookRuns = "outlook_runs" tableOutlookRuns = "outlook_runs"
tableOutlooks = "outlooks" tableOutlooks = "outlooks"
tableOutlookDiscussions = "outlook_discussions"
) )
// PostgresSchema returns weatherfeeder's Postgres schema definition. // PostgresSchema returns weatherfeeder's Postgres schema definition.
@@ -237,6 +238,7 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "sent", Type: "TIMESTAMPTZ", Nullable: true}, {Name: "sent", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "effective", Type: "TIMESTAMPTZ", Nullable: true}, {Name: "effective", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "onset", Type: "TIMESTAMPTZ", Nullable: true}, {Name: "onset", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "ends", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "expires", Type: "TIMESTAMPTZ", Nullable: true}, {Name: "expires", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "area_description", Type: "TEXT", Nullable: true}, {Name: "area_description", Type: "TEXT", Nullable: true},
{Name: "sender_name", Type: "TEXT", Nullable: true}, {Name: "sender_name", Type: "TEXT", Nullable: true},
@@ -279,6 +281,7 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false}, {Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true}, {Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "outlook_count", Type: "INTEGER", Nullable: false}, {Name: "outlook_count", Type: "INTEGER", Nullable: false},
{Name: "discussion_count", Type: "INTEGER", Nullable: false},
}...), }...),
PrimaryKey: []string{"event_id"}, PrimaryKey: []string{"event_id"},
PruneColumn: "as_of", PruneColumn: "as_of",
@@ -306,9 +309,6 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false}, {Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "expires_at", Type: "TIMESTAMPTZ", Nullable: false}, {Name: "expires_at", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "forecaster", Type: "TEXT", Nullable: true}, {Name: "forecaster", Type: "TEXT", Nullable: true},
{Name: "headline", Type: "TEXT", Nullable: true},
{Name: "summary", Type: "TEXT", Nullable: true},
{Name: "discussion", Type: "TEXT", Nullable: true},
{Name: "source_url", Type: "TEXT", Nullable: true}, {Name: "source_url", Type: "TEXT", Nullable: true},
{Name: "image_url", Type: "TEXT", Nullable: true}, {Name: "image_url", Type: "TEXT", Nullable: true},
{Name: "contains_location", Type: "BOOLEAN", Nullable: false}, {Name: "contains_location", Type: "BOOLEAN", Nullable: false},
@@ -322,6 +322,25 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "idx_wf_outlooks_valid", Columns: []string{"valid_from", "valid_to"}}, {Name: "idx_wf_outlooks_valid", Columns: []string{"valid_from", "valid_to"}},
}, },
}, },
{
Name: tableOutlookDiscussions,
Columns: []fksinks.PostgresColumn{
{Name: "run_event_id", Type: "TEXT REFERENCES outlook_runs(event_id) ON DELETE CASCADE", Nullable: false},
{Name: "discussion_index", Type: "INTEGER", Nullable: false},
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "day", Type: "INTEGER", Nullable: false},
{Name: "headline", Type: "TEXT", Nullable: true},
{Name: "summary", Type: "TEXT", Nullable: true},
{Name: "discussion", Type: "TEXT", Nullable: true},
{Name: "updated_at", Type: "TIMESTAMPTZ", Nullable: true},
},
PrimaryKey: []string{"run_event_id", "discussion_index"},
PruneColumn: "as_of",
Indexes: []fksinks.PostgresIndex{
{Name: "idx_wf_outlook_discussions_day_as_of", Columns: []string{"day", "as_of"}},
{Name: "idx_wf_outlook_discussions_run_day", Columns: []string{"run_event_id", "day"}, Unique: true},
},
},
}, },
MapEvent: mapPostgresEvent, MapEvent: mapPostgresEvent,
} }

View File

@@ -28,6 +28,7 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
tableAlertReferences: true, tableAlertReferences: true,
tableOutlookRuns: true, tableOutlookRuns: true,
tableOutlooks: true, tableOutlooks: true,
tableOutlookDiscussions: true,
} }
if len(s.Tables) != len(wantTables) { if len(s.Tables) != len(wantTables) {
@@ -53,7 +54,7 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) { func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
runColumns := columnsForTable(t, tableOutlookRuns) runColumns := columnsForTable(t, tableOutlookRuns)
for _, col := range []string{"event_id", "event_kind", "event_source", "event_schema", "event_emitted_at", "event_effective_at", "location_id", "location_name", "latitude", "longitude", "as_of", "issued_at", "outlook_count"} { for _, col := range []string{"event_id", "event_kind", "event_source", "event_schema", "event_emitted_at", "event_effective_at", "location_id", "location_name", "latitude", "longitude", "as_of", "issued_at", "outlook_count", "discussion_count"} {
if !runColumns[col] { if !runColumns[col] {
t.Fatalf("%s missing %s column", tableOutlookRuns, col) t.Fatalf("%s missing %s column", tableOutlookRuns, col)
} }
@@ -63,15 +64,40 @@ func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_as_of", []string{"as_of"}) assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_as_of", []string{"as_of"})
outlookColumns := columnsForTable(t, tableOutlooks) outlookColumns := columnsForTable(t, tableOutlooks)
for _, col := range []string{"run_event_id", "outlook_index", "as_of", "outlook_id", "provider", "product", "day", "outlook_type", "label", "label_text", "severity_rank", "valid_from", "valid_to", "issued_at", "expires_at", "forecaster", "headline", "summary", "discussion", "source_url", "image_url", "contains_location", "geometry_json"} { for _, col := range []string{"run_event_id", "outlook_index", "as_of", "outlook_id", "provider", "product", "day", "outlook_type", "label", "label_text", "severity_rank", "valid_from", "valid_to", "issued_at", "expires_at", "forecaster", "source_url", "image_url", "contains_location", "geometry_json"} {
if !outlookColumns[col] { if !outlookColumns[col] {
t.Fatalf("%s missing %s column", tableOutlooks, col) t.Fatalf("%s missing %s column", tableOutlooks, col)
} }
} }
for _, col := range []string{"headline", "summary", "discussion"} {
if outlookColumns[col] {
t.Fatalf("%s still includes legacy %s column", tableOutlooks, col)
}
}
assertTablePrimaryKey(t, tableOutlooks, []string{"run_event_id", "outlook_index"}) assertTablePrimaryKey(t, tableOutlooks, []string{"run_event_id", "outlook_index"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_contains_valid", []string{"contains_location", "valid_from", "valid_to"}) assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_contains_valid", []string{"contains_location", "valid_from", "valid_to"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_day_type_label", []string{"day", "outlook_type", "label"}) assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_day_type_label", []string{"day", "outlook_type", "label"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_valid", []string{"valid_from", "valid_to"}) assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_valid", []string{"valid_from", "valid_to"})
discussionColumns := columnsForTable(t, tableOutlookDiscussions)
for _, col := range []string{"run_event_id", "discussion_index", "as_of", "day", "headline", "summary", "discussion", "updated_at"} {
if !discussionColumns[col] {
t.Fatalf("%s missing %s column", tableOutlookDiscussions, col)
}
}
assertTablePrimaryKey(t, tableOutlookDiscussions, []string{"run_event_id", "discussion_index"})
assertTablePruneColumn(t, tableOutlookDiscussions, "as_of")
assertTableIndex(t, tableOutlookDiscussions, "idx_wf_outlook_discussions_day_as_of", []string{"day", "as_of"})
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) { func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
@@ -115,35 +141,56 @@ func TestWeatherPostgresSchemaParentTablesStartWithEnvelopeColumns(t *testing.T)
func assertTablePrimaryKey(t *testing.T, table string, want []string) { func assertTablePrimaryKey(t *testing.T, table string, want []string) {
t.Helper() t.Helper()
for _, tbl := range PostgresSchema().Tables { tbl := tableByName(t, table)
if tbl.Name != table {
continue
}
if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") { if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") {
t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want) t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want)
} }
return }
func assertTablePruneColumn(t *testing.T, table string, want string) {
t.Helper()
tbl := tableByName(t, table)
if tbl.PruneColumn != want {
t.Fatalf("%s prune column = %q, want %q", table, tbl.PruneColumn, want)
} }
t.Fatalf("missing table %q", table)
} }
func assertTableIndex(t *testing.T, table string, name string, want []string) { func assertTableIndex(t *testing.T, table string, name string, want []string) {
t.Helper() t.Helper()
for _, tbl := range PostgresSchema().Tables { assertTableIndexWithUnique(t, table, name, want, false)
if tbl.Name != table { }
continue
} func assertTableUniqueIndex(t *testing.T, table string, name string, want []string) {
t.Helper()
assertTableIndexWithUnique(t, table, name, want, true)
}
func assertTableIndexWithUnique(t *testing.T, table string, name string, want []string, unique bool) {
t.Helper()
tbl := tableByName(t, table)
for _, idx := range tbl.Indexes { for _, idx := range tbl.Indexes {
if idx.Name == name { if idx.Name == name {
if strings.Join(idx.Columns, ",") != strings.Join(want, ",") { if strings.Join(idx.Columns, ",") != strings.Join(want, ",") {
t.Fatalf("%s index %s columns = %#v, want %#v", table, name, idx.Columns, want) t.Fatalf("%s index %s columns = %#v, want %#v", table, name, idx.Columns, want)
} }
if idx.Unique != unique {
t.Fatalf("%s index %s unique = %v, want %v", table, name, idx.Unique, unique)
}
return return
} }
} }
t.Fatalf("%s missing index %s", table, name) t.Fatalf("%s missing index %s", table, name)
}
func tableByName(t *testing.T, table string) fksinks.PostgresTable {
t.Helper()
for _, tbl := range PostgresSchema().Tables {
if tbl.Name == table {
return tbl
}
} }
t.Fatalf("missing table %q", table) t.Fatalf("missing table %q", table)
return fksinks.PostgresTable{}
} }
func orderedColumnsForTable(t *testing.T, table string) []fksinks.PostgresColumn { func orderedColumnsForTable(t *testing.T, table string) []fksinks.PostgresColumn {

View File

@@ -55,9 +55,11 @@ type WeatherAlert struct {
Instruction string `json:"instruction,omitempty"` Instruction string `json:"instruction,omitempty"`
// Timing (all optional; provider-dependent). // Timing (all optional; provider-dependent).
// Onset and Ends describe the alert period. Expires is provider expiration metadata.
Sent *time.Time `json:"sent,omitempty"` Sent *time.Time `json:"sent,omitempty"`
Effective *time.Time `json:"effective,omitempty"` Effective *time.Time `json:"effective,omitempty"`
Onset *time.Time `json:"onset,omitempty"` Onset *time.Time `json:"onset,omitempty"`
Ends *time.Time `json:"ends,omitempty"`
Expires *time.Time `json:"expires,omitempty"` Expires *time.Time `json:"expires,omitempty"`
// Scope / affected area. // Scope / affected area.

View File

@@ -26,6 +26,7 @@ func TestDocumentedConsumerModelTypes(t *testing.T) {
"WeatherAlert", "WeatherAlert",
"WeatherAlertReference", "WeatherAlertReference",
"WeatherOutlookRun", "WeatherOutlookRun",
"WeatherOutlookDiscussion",
"WeatherOutlook", "WeatherOutlook",
"WMOCode", "WMOCode",
} }

View File

@@ -15,6 +15,16 @@ type WeatherOutlookRun struct {
AsOf time.Time `json:"asOf"` AsOf time.Time `json:"asOf"`
IssuedAt *time.Time `json:"issuedAt,omitempty"` IssuedAt *time.Time `json:"issuedAt,omitempty"`
Outlooks []WeatherOutlook `json:"outlooks"` Outlooks []WeatherOutlook `json:"outlooks"`
Discussions []WeatherOutlookDiscussion `json:"discussions"`
}
// WeatherOutlookDiscussion is run-level SPC outlook prose for one outlook day.
type WeatherOutlookDiscussion struct {
Day int `json:"day"`
Headline string `json:"headline,omitempty"`
Summary string `json:"summary,omitempty"`
Discussion string `json:"discussion,omitempty"`
UpdatedAt *time.Time `json:"updatedAt,omitempty"`
} }
// WeatherOutlook is a canonical representation of one outlook polygon. // WeatherOutlook is a canonical representation of one outlook polygon.
@@ -32,9 +42,6 @@ type WeatherOutlook struct {
IssuedAt time.Time `json:"issuedAt"` IssuedAt time.Time `json:"issuedAt"`
ExpiresAt time.Time `json:"expiresAt"` ExpiresAt time.Time `json:"expiresAt"`
Forecaster string `json:"forecaster,omitempty"` Forecaster string `json:"forecaster,omitempty"`
Headline string `json:"headline,omitempty"`
Summary string `json:"summary,omitempty"`
Discussion string `json:"discussion,omitempty"`
SourceURL string `json:"sourceUrl,omitempty"` SourceURL string `json:"sourceUrl,omitempty"`
ImageURL string `json:"imageUrl,omitempty"` ImageURL string `json:"imageUrl,omitempty"`
ContainsLocation bool `json:"containsLocation"` ContainsLocation bool `json:"containsLocation"`

60
model/outlook_test.go Normal file
View File

@@ -0,0 +1,60 @@
package model
import (
"encoding/json"
"strings"
"testing"
"time"
)
func TestWeatherOutlookJSONShape(t *testing.T) {
updatedAt := time.Date(2026, 6, 11, 16, 30, 0, 0, time.UTC)
run := WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []WeatherOutlook{{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.UTC),
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 12, 34, 56, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: true,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
}},
Discussions: []WeatherOutlookDiscussion{{
Day: 1,
Headline: "Day 1 Convective Outlook",
Summary: "Severe thunderstorms are possible.",
Discussion: "Full discussion text.",
UpdatedAt: &updatedAt,
}},
}
raw, err := json.Marshal(run)
if err != nil {
t.Fatalf("Marshal(WeatherOutlookRun) error = %v", err)
}
got := string(raw)
for _, want := range []string{`"outlooks"`, `"discussions"`, `"headline"`, `"summary"`, `"discussion"`, `"updatedAt"`} {
if !strings.Contains(got, want) {
t.Fatalf("WeatherOutlookRun JSON missing %s: %s", want, got)
}
}
outlookStart := strings.Index(got, `"outlooks"`)
discussionStart := strings.Index(got, `"discussions"`)
if outlookStart == -1 || discussionStart == -1 || discussionStart <= outlookStart {
t.Fatalf("WeatherOutlookRun JSON has unexpected outlook/discussion order: %s", got)
}
outlookJSON := got[outlookStart:discussionStart]
for _, unwanted := range []string{`"headline"`, `"summary"`, `"discussion"`} {
if strings.Contains(outlookJSON, unwanted) {
t.Fatalf("WeatherOutlook JSON contains polygon-level prose key %s: %s", unwanted, got)
}
}
}

View File

@@ -97,5 +97,10 @@ func stringConstantsFromFile(t *testing.T, path string, prefix string, skip func
} }
func schemaConstantNotInCurrentContract(name string) bool { func schemaConstantNotInCurrentContract(name string) bool {
return name == "SchemaRawOpenWeatherHourlyForecastV1" switch name {
case "SchemaRawOpenWeatherHourlyForecastV1":
return true
default:
return false
}
} }

View File

@@ -33,4 +33,5 @@ const (
SchemaWeatherStoryV1 = "weather.weather_story.v1" SchemaWeatherStoryV1 = "weather.weather_story.v1"
SchemaWeatherAlertV1 = "weather.alert.v1" SchemaWeatherAlertV1 = "weather.alert.v1"
SchemaWeatherOutlookV1 = "weather.outlook.v1" SchemaWeatherOutlookV1 = "weather.outlook.v1"
SchemaWeatherOutlookV2 = "weather.outlook.v2"
) )