7 Commits

Author SHA1 Message Date
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
24 changed files with 877 additions and 181 deletions

View File

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

View File

@@ -9,8 +9,8 @@ import "gitea.maximumdirect.net/ejr/weatherfeeder/model"
## Purpose
Package `model` defines `weatherfeeder`'s canonical weather payload structs.
These structs are emitted as the `payload` of canonical `weather.*.v1` events
and are also the domain types consumed by downstream applications such as
These structs are emitted as the `payload` of canonical `weather.*` events and
are also the domain types consumed by downstream applications such as
`weatherapi`.
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.weather_story.v1` | `WeatherStoryRun` |
| `weather.alert.v1` | `WeatherAlertRun` |
| `weather.outlook.v1` | `WeatherOutlookRun` |
| `weather.outlook.v2` | `WeatherOutlookRun` |
Related child types include:
@@ -38,8 +38,12 @@ Related child types include:
- `WeatherAlert`
- `WeatherAlertReference`
- `WeatherOutlook`
- `WeatherOutlookDiscussion`
- `WMOCode`
`WeatherOutlookRun` includes `WeatherOutlookDiscussion` entries as run-level
SPC day discussions.
## Wire And Compatibility Rules
- JSON tags define canonical payload field names.

View File

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

View File

@@ -37,7 +37,7 @@ Canonical schemas emitted after normalization:
| `forecast_discussion` | `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` |
| `weather_story` | `weather.weather_story.v1` | `WeatherStoryRun` |
| `alert` | `weather.alert.v1` | `WeatherAlertRun` |
| `outlook` | `weather.outlook.v1` | `WeatherOutlookRun` |
| `outlook` | `weather.outlook.v2` | `WeatherOutlookRun` |
Raw upstream schemas emitted by current sources:
@@ -220,13 +220,15 @@ Payload type: `WeatherAlertRun`.
`references[]` entries contain optional `id`, `identifier`, `sender`, and
`sent` fields.
## `weather.outlook.v1`
## `weather.outlook.v2`
Payload type: `WeatherOutlookRun`.
The current producer is the SPC convective outlook normalizer. It emits Day 1-3
convective outlook polygons for categorical, tornado, hail, and wind products.
All timestamps are UTC.
convective outlook snapshots for categorical, tornado, hail, and wind products
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 |
|---|---|:---:|---|
@@ -236,7 +238,8 @@ All timestamps are UTC.
| `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. |
| `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:
@@ -255,19 +258,53 @@ All timestamps are UTC.
| `issuedAt` | timestamp | yes | Feature issue time. |
| `expiresAt` | timestamp | yes | Expiration time; currently equal to `validTo`. |
| `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. |
| `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` preserves the SPC feature geometry as compact GeoJSON using
`[longitude, latitude]` coordinate order. `containsLocation` is computed with
that geometry and the configured source `latitude`/`longitude`; boundary points
count as contained. All outlook polygons are emitted, including polygons that do
not contain the configured point.
count as contained. Polygons that do not contain the configured point are not
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

View File

@@ -23,7 +23,7 @@ Events are mapped only for canonical weather schemas:
- `weather.forecast_discussion.v1`
- `weather.weather_story.v1`
- `weather.alert.v1`
- `weather.outlook.v1`
- `weather.outlook.v2`
Unsupported schemas produce no writes for this sink. Mapped events are inserted
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` |
| `outlook_runs` | `event_id` | `as_of` |
| `outlooks` | `run_event_id`, `outlook_index` | `as_of` |
| `outlook_discussions` | `run_event_id`, `discussion_index` | `as_of` |
## Table Contract
@@ -408,6 +409,7 @@ Indexes:
| `as_of` | `TIMESTAMPTZ` | no | `payload.asOf` |
| `issued_at` | `TIMESTAMPTZ` | yes | `payload.issuedAt` |
| `outlook_count` | `INTEGER` | no | `len(payload.outlooks)` |
| `discussion_count` | `INTEGER` | no | `len(payload.discussions)` |
### `outlooks`
@@ -442,14 +444,36 @@ Indexes:
| `issued_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].issuedAt` |
| `expires_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].expiresAt` |
| `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` |
| `image_url` | `TEXT` | yes | `payload.outlooks[].imageUrl` |
| `contains_location` | `BOOLEAN` | no | `payload.outlooks[].containsLocation` |
| `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
When sink param `prune` is set, every successful write transaction deletes rows
@@ -472,5 +496,6 @@ binary does not provide CLI commands for them.
- `WeatherAlertRun`: read `alert_runs`, join `alerts` by `run_event_id` ordered
by `alert_index`, then join `alert_references` by `run_event_id` and
`alert_index` ordered by `reference_index`.
- `WeatherOutlookRun`: read `outlook_runs`, then join `outlooks` by
`run_event_id` ordered by `outlook_index`.
- `WeatherOutlookRun`: read `outlook_runs`, join `outlooks` by `run_event_id`
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 |
| --- | --- | --- | --- |
| `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
@@ -52,8 +52,8 @@ It also fetches three required print pages:
- Day 3 convective outlook print page
GeoJSON products are authoritative for outlook polygons, valid windows, issue
times, labels, and severity rank. Print pages are authoritative for headline,
summary, and discussion text.
times, labels, and severity rank. Print pages are authoritative for run-level
day discussion headline, summary, and discussion text.
## Accept Headers
@@ -99,23 +99,33 @@ Raw source `effective_at` prefers:
4. fetch time.
The normalizer sets canonical `asOf` and normalized event `effective_at` from
the latest valid outlook feature `issuedAt`, with fallback to print-page update
time and then the incoming event envelope.
the latest valid GeoJSON issue time across the complete raw bundle, including
empty no-risk placeholders, with fallback to print-page update time and then the
incoming event envelope.
## Mapping Notes
Each GeoJSON feature becomes one canonical outlook. Products are ordered by day,
then categorical, tornado, hail, and wind. Feature order is preserved within
each product.
The raw source fetches and envelopes the complete SPC bundle. The normalizer
decodes every configured GeoJSON product, skips empty no-risk
`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
longitude against compact GeoJSON `Polygon` or `MultiPolygon` geometry.
Coordinates use GeoJSON order, `[longitude, latitude]`, and boundary points
count as contained.
All outlook polygons are preserved, including polygons that do not contain the
configured point. Matching print-page headline, summary, and discussion text is
attached to every outlook for the same day.
Canonical outlook runs are emitted even when no polygons apply locally. In that
case the payload contains empty `outlooks` and `discussions` arrays.
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

View File

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

View File

@@ -19,11 +19,16 @@ Inputs are canonical feed events. The mapper currently handles these schemas:
- `weather.forecast_discussion.v1`
- `weather.weather_story.v1`
- `weather.alert.v1`
- `weather.outlook.v1`
- `weather.outlook.v2`
Outputs are feedkit `PostgresWrite` values for weatherfeeder-owned tables.
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
- Weatherfeeder owns table definitions in `schema.go`.
@@ -82,6 +87,7 @@ Child rows use positional indexes to preserve canonical array order:
- `alert_index`
- `reference_index`
- `outlook_index`
- `discussion_index`
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`;
- outlook runs require `asOf`, and each outlook requires `id`, `provider`,
`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.
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.
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
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.
- 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.
- 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.

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

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

View File

@@ -27,8 +27,8 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
if out.Schema != standards.SchemaWeatherOutlookV1 {
t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV1)
if out.Schema != standards.SchemaWeatherOutlookV2 {
t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV2)
}
if out.Kind != event.Kind(standards.KindOutlook) {
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 {
t.Fatalf("coordinates = %v,%v", run.Latitude, run.Longitude)
}
if len(run.Outlooks) != 9 {
t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks))
if len(run.Outlooks) != 4 {
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]
@@ -88,18 +103,6 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
if !got.ContainsLocation {
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" {
t.Fatalf("ID = %q", got.ID)
}
@@ -115,13 +118,39 @@ func TestConvectiveOutlookNormalizerAcceptsTypedSourcePayload(t *testing.T) {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if len(run.Outlooks) != 9 {
t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks))
if len(run.Outlooks) != 4 {
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) {
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 {
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) {
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 {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
assertAllOutlooksContainLocation(t, run.Outlooks)
for _, outlookType := range []string{
spcprovider.OutlookTypeTornado,
spcprovider.OutlookTypeHail,
@@ -168,7 +198,7 @@ func TestConvectiveOutlookNormalizerMapsProbabilisticOutlookTypes(t *testing.T)
}
func TestConvectiveOutlookNormalizerSkipsEmptyGeometryCollectionPlaceholder(t *testing.T) {
bundle := 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 {
@@ -185,9 +215,11 @@ func TestConvectiveOutlookNormalizerSkipsEmptyGeometryCollectionPlaceholder(t *t
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if len(run.Outlooks) != 8 {
t.Fatalf("Outlooks length = %d, want 8", len(run.Outlooks))
if len(run.Outlooks) != 3 {
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)
}
@@ -200,19 +232,40 @@ func TestConvectiveOutlookNormalizerSkipsEmptyGeometryCollectionPlaceholder(t *t
}
}
func TestConvectiveOutlookNormalizerContainsLocationFalseOutsidePolygon(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0)))
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 run.Outlooks[0].ContainsLocation {
t.Fatalf("ContainsLocation = true, want false")
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) {
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 {
t.Fatalf("Normalize() error = %v", err)
}
@@ -221,11 +274,12 @@ func TestConvectiveOutlookNormalizerPreservesCorrectionMarker(t *testing.T) {
if got == nil {
t.Fatalf("missing day 2 tornado outlook")
}
if !strings.Contains(got.Headline, "CORR 1") {
t.Fatalf("Headline = %q, want correction marker", got.Headline)
assertDiscussionDays(t, run.Discussions, 2)
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") {
t.Fatalf("Discussion = %q, want correction marker", got.Discussion)
if !strings.Contains(run.Discussions[0].Discussion, "CORR 1") {
t.Fatalf("day 2 discussion = %q, want correction marker", run.Discussions[0].Discussion)
}
}
@@ -291,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) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil {
@@ -301,12 +368,23 @@ func TestConvectiveOutlookNormalizerOutputJSONShape(t *testing.T) {
t.Fatalf("Marshal(payload) error = %v", err)
}
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) {
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) {
t.Fatalf("payload JSON exposed raw key %s: %s", unwanted, got)
}
@@ -408,6 +486,37 @@ func findOutlook(outlooks []model.WeatherOutlook, day int, outlookType string) *
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) {
t.Helper()
want := time.Date(year, month, day, hour, minute, second, 0, time.UTC)

View File

@@ -11,7 +11,7 @@
// - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion
// - weather.weather_story.v1 -> model.WeatherStoryRun
// - weather.alert.v1 -> model.WeatherAlertRun
// - weather.outlook.v1 -> model.WeatherOutlookRun
// - weather.outlook.v2 -> model.WeatherOutlookRun
//
// Parent/child relationships:
// - observations.event_id -> observation_present_weather.event_id
@@ -21,6 +21,7 @@
// - alert_runs.event_id -> alerts.run_event_id
// - 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 -> outlook_discussions.run_event_id
//
// Dedupe and retention behavior:
// - Parent primary keys (event_id): observations, forecasts, alert_runs, outlook_runs.
@@ -39,6 +40,7 @@
// - alert_references.as_of
// - outlook_runs.as_of
// - outlooks.as_of
// - outlook_discussions.as_of
//
// Envelope field mapping (shared parent columns)
//
@@ -227,6 +229,7 @@
// - as_of TIMESTAMPTZ -> payload.asOf
// - issued_at TIMESTAMPTZ NULL -> payload.issuedAt
// - outlook_count INTEGER -> len(payload.outlooks)
// - discussion_count INTEGER -> len(payload.discussions)
//
// 11. outlooks (PK: run_event_id, outlook_index)
//
@@ -246,14 +249,22 @@
// - issued_at TIMESTAMPTZ -> payload.outlooks[i].issuedAt
// - expires_at TIMESTAMPTZ -> payload.outlooks[i].expiresAt
// - 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
// - image_url TEXT NULL -> payload.outlooks[i].imageUrl
// - contains_location BOOLEAN -> payload.outlooks[i].containsLocation
// - 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
//
// - WeatherObservation:
@@ -274,6 +285,7 @@
// ordered by reference_index to rebuild references per alert.
//
// - WeatherOutlookRun:
// read one row from outlook_runs, then join outlooks by run_event_id ordered
// by outlook_index to rebuild outlooks.
// read one row from outlook_runs, join outlooks by run_event_id ordered by
// outlook_index to rebuild outlooks, then join outlook_discussions by
// run_event_id ordered by discussion_index to rebuild discussions.
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)
case standards.SchemaWeatherAlertV1:
return mapAlertEvent(e)
case standards.SchemaWeatherOutlookV1:
case standards.SchemaWeatherOutlookV2:
return mapOutlookEvent(e)
default:
return nil, nil
@@ -339,17 +339,22 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
}
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{
Table: tableOutlookRuns,
Values: parentEventValues(e, map[string]any{
"location_id": nullableString(run.LocationID),
"location_name": nullableString(run.LocationName),
"latitude": nullableFloat64(run.Latitude),
"longitude": nullableFloat64(run.Longitude),
"as_of": asOf,
"issued_at": nullableTime(run.IssuedAt),
"outlook_count": len(run.Outlooks),
"location_id": nullableString(run.LocationID),
"location_name": nullableString(run.LocationName),
"latitude": nullableFloat64(run.Latitude),
"longitude": nullableFloat64(run.Longitude),
"as_of": asOf,
"issued_at": nullableTime(run.IssuedAt),
"outlook_count": len(run.Outlooks),
"discussion_count": len(run.Discussions),
}),
})
@@ -381,9 +386,6 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
"issued_at": outlook.IssuedAt.UTC(),
"expires_at": outlook.ExpiresAt.UTC(),
"forecaster": nullableString(outlook.Forecaster),
"headline": nullableString(outlook.Headline),
"summary": nullableString(outlook.Summary),
"discussion": nullableString(outlook.Discussion),
"source_url": nullableString(outlook.SourceURL),
"image_url": nullableString(outlook.ImageURL),
"contains_location": outlook.ContainsLocation,
@@ -392,6 +394,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
}
@@ -423,6 +441,28 @@ func validateOutlook(outlook model.WeatherOutlook, index int) error {
if len(outlook.Geometry) == 0 {
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
}

View File

@@ -243,6 +243,7 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
lat := 38.6239
lon := -90.3571
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
run := model.WeatherOutlookRun{
LocationID: "stl",
@@ -266,9 +267,6 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
IssuedAt: issuedAt,
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
Forecaster: "SMITH",
Headline: "Day 1 Convective Outlook",
Summary: "Severe thunderstorms are possible.",
Discussion: "Full discussion text.",
SourceURL: "https://example.invalid/day1.geojson",
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 ] ] ] }`),
@@ -284,18 +282,27 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
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: false,
ContainsLocation: true,
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 {
t.Fatalf("mapPostgresEvent() error = %v", err)
}
if len(writes) != 3 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 3", len(writes))
if len(writes) != 4 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 4", len(writes))
}
if writes[0].Table != tableOutlookRuns {
t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns)
@@ -303,6 +310,9 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
if got := writes[0].Values["outlook_count"]; got != 2 {
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() {
t.Fatalf("outlook_runs issued_at = %#v, want UTC %s", got, issuedAt.UTC())
}
@@ -324,15 +334,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]]]}` {
t.Fatalf("first geometry_json = %#v", got)
}
if got := writes[2].Values["contains_location"]; got != false {
t.Fatalf("second contains_location = %#v, want false", got)
if got := writes[2].Values["contains_location"]; got != true {
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)
}
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 {
t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error")
}
@@ -343,17 +402,18 @@ func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) {
func TestMapPostgresEventOutlookRejectsMissingIDAndProvider(t *testing.T) {
base := 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),
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
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]]]}`),
}
tests := []struct {
@@ -381,7 +441,7 @@ func TestMapPostgresEventOutlookRejectsMissingIDAndProvider(t *testing.T) {
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
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 {
t.Fatalf("mapPostgresEvent() error = nil, want %q", tt.wantErr)
}
@@ -396,16 +456,17 @@ func TestMapPostgresEventOutlookRejectsMissingRequiredTimes(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []model.WeatherOutlook{{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
ContainsLocation: true,
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 {
t.Fatalf("mapPostgresEvent() error = nil, want missing time error")
}
@@ -418,19 +479,20 @@ func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []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),
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,
}},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV2, standards.KindOutlook, run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want geometry error")
}
@@ -439,6 +501,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) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, model.WeatherStoryRun{}))
if err == nil {
@@ -508,6 +631,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) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, "bad"))
if err == nil {
@@ -647,6 +781,23 @@ func tableColumnCounts() map[string]int {
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 {
out := v
return &out

View File

@@ -18,6 +18,7 @@ const (
tableAlertReferences = "alert_references"
tableOutlookRuns = "outlook_runs"
tableOutlooks = "outlooks"
tableOutlookDiscussions = "outlook_discussions"
)
// PostgresSchema returns weatherfeeder's Postgres schema definition.
@@ -279,6 +280,7 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "outlook_count", Type: "INTEGER", Nullable: false},
{Name: "discussion_count", Type: "INTEGER", Nullable: false},
}...),
PrimaryKey: []string{"event_id"},
PruneColumn: "as_of",
@@ -306,9 +308,6 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "expires_at", Type: "TIMESTAMPTZ", Nullable: false},
{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: "image_url", Type: "TEXT", Nullable: true},
{Name: "contains_location", Type: "BOOLEAN", Nullable: false},
@@ -322,6 +321,25 @@ func PostgresSchema() fksinks.PostgresSchema {
{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,
}

View File

@@ -28,6 +28,7 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
tableAlertReferences: true,
tableOutlookRuns: true,
tableOutlooks: true,
tableOutlookDiscussions: true,
}
if len(s.Tables) != len(wantTables) {
@@ -53,7 +54,7 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
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] {
t.Fatalf("%s missing %s column", tableOutlookRuns, col)
}
@@ -63,15 +64,31 @@ func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_as_of", []string{"as_of"})
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] {
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"})
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_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 TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
@@ -115,35 +132,56 @@ func TestWeatherPostgresSchemaParentTablesStartWithEnvelopeColumns(t *testing.T)
func assertTablePrimaryKey(t *testing.T, table string, want []string) {
t.Helper()
for _, tbl := range PostgresSchema().Tables {
if tbl.Name != table {
continue
}
if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") {
t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want)
}
return
tbl := tableByName(t, table)
if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") {
t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want)
}
}
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) {
t.Helper()
for _, tbl := range PostgresSchema().Tables {
if tbl.Name != table {
continue
}
for _, idx := range tbl.Indexes {
if idx.Name == name {
if strings.Join(idx.Columns, ",") != strings.Join(want, ",") {
t.Fatalf("%s index %s columns = %#v, want %#v", table, name, idx.Columns, want)
}
return
assertTableIndexWithUnique(t, table, name, want, false)
}
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 {
if idx.Name == name {
if strings.Join(idx.Columns, ",") != strings.Join(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
}
}
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("%s missing index %s", table, name)
}
t.Fatalf("missing table %q", table)
return fksinks.PostgresTable{}
}
func orderedColumnsForTable(t *testing.T, table string) []fksinks.PostgresColumn {

View File

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

View File

@@ -8,13 +8,23 @@ import (
// WeatherOutlookRun is a snapshot of convective outlook polygons for a
// configured location as-of a provider issue time.
type WeatherOutlookRun struct {
LocationID string `json:"locationId,omitempty"`
LocationName string `json:"locationName,omitempty"`
Latitude *float64 `json:"latitude,omitempty"`
Longitude *float64 `json:"longitude,omitempty"`
AsOf time.Time `json:"asOf"`
IssuedAt *time.Time `json:"issuedAt,omitempty"`
Outlooks []WeatherOutlook `json:"outlooks"`
LocationID string `json:"locationId,omitempty"`
LocationName string `json:"locationName,omitempty"`
Latitude *float64 `json:"latitude,omitempty"`
Longitude *float64 `json:"longitude,omitempty"`
AsOf time.Time `json:"asOf"`
IssuedAt *time.Time `json:"issuedAt,omitempty"`
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.
@@ -32,9 +42,6 @@ type WeatherOutlook struct {
IssuedAt time.Time `json:"issuedAt"`
ExpiresAt time.Time `json:"expiresAt"`
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"`
ImageURL string `json:"imageUrl,omitempty"`
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 {
return name == "SchemaRawOpenWeatherHourlyForecastV1"
switch name {
case "SchemaRawOpenWeatherHourlyForecastV1":
return true
default:
return false
}
}

View File

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