diff --git a/internal/sinks/postgres/doc.go b/internal/sinks/postgres/doc.go index a71b804..c0a5998 100644 --- a/internal/sinks/postgres/doc.go +++ b/internal/sinks/postgres/doc.go @@ -11,6 +11,7 @@ // - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion // - weather.weather_story.v1 -> model.WeatherStoryRun // - weather.alert.v1 -> model.WeatherAlertRun +// - weather.outlook.v1 -> model.WeatherOutlookRun // // Parent/child relationships: // - observations.event_id -> observation_present_weather.event_id @@ -19,9 +20,10 @@ // - weather_story_runs.event_id -> weather_stories.run_event_id // - 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 // // Dedupe and retention behavior: -// - Parent primary keys (event_id): observations, forecasts, alert_runs. +// - Parent primary keys (event_id): observations, forecasts, alert_runs, outlook_runs. // - Child primary keys use positional indexes to preserve payload order. // - Prune columns: // - observations.observed_at @@ -35,11 +37,13 @@ // - alert_runs.as_of // - alerts.as_of // - alert_references.as_of +// - outlook_runs.as_of +// - outlooks.as_of // // Envelope field mapping (shared parent columns) // // These columns exist on parent tables such as observations, forecasts, -// forecast_discussions, weather_story_runs, and alert_runs: +// forecast_discussions, weather_story_runs, alert_runs, and outlook_runs: // - event_id TEXT -> event.id // - event_kind TEXT -> event.kind // - event_source TEXT -> event.source @@ -208,6 +212,46 @@ // - sender TEXT NULL -> payload.alerts[i].references[j].sender // - sent TIMESTAMPTZ NULL -> payload.alerts[i].references[j].sent // +// 10. outlook_runs (PK: event_id) +// +// - event_id TEXT -> event.id +// - event_kind TEXT -> event.kind +// - event_source TEXT -> event.source +// - event_schema TEXT -> event.schema +// - event_emitted_at TIMESTAMPTZ -> event.emitted_at +// - event_effective_at TIMESTAMPTZ NULL -> event.effective_at +// - location_id TEXT NULL -> payload.locationId +// - location_name TEXT NULL -> payload.locationName +// - latitude DOUBLE PRECISION NULL -> payload.latitude +// - longitude DOUBLE PRECISION NULL -> payload.longitude +// - as_of TIMESTAMPTZ -> payload.asOf +// - issued_at TIMESTAMPTZ NULL -> payload.issuedAt +// - outlook_count INTEGER -> len(payload.outlooks) +// +// 11. outlooks (PK: run_event_id, outlook_index) +// +// - run_event_id TEXT -> outlook_runs.event_id / payload.outlooks[i] +// - outlook_index INTEGER -> i (array position in payload.outlooks) +// - as_of TIMESTAMPTZ -> payload.asOf (copied from parent) +// - product TEXT -> payload.outlooks[i].product +// - day INTEGER -> payload.outlooks[i].day +// - outlook_type TEXT -> payload.outlooks[i].outlookType +// - label TEXT -> payload.outlooks[i].label +// - label_text TEXT NULL -> payload.outlooks[i].labelText +// - severity_rank INTEGER NULL -> payload.outlooks[i].severityRank +// - valid_from TIMESTAMPTZ -> payload.outlooks[i].validFrom +// - valid_to TIMESTAMPTZ -> payload.outlooks[i].validTo +// - 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 +// // Reconstructing canonical JSON payloads // // - WeatherObservation: @@ -226,4 +270,8 @@ // read one row from alert_runs, join alerts by run_event_id ordered by // alert_index, then join alert_references by (run_event_id, alert_index) // 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. package postgres diff --git a/internal/sinks/postgres/map.go b/internal/sinks/postgres/map.go index 9f6fcfd..02afddd 100644 --- a/internal/sinks/postgres/map.go +++ b/internal/sinks/postgres/map.go @@ -1,6 +1,7 @@ package postgres import ( + "bytes" "context" "encoding/json" "fmt" @@ -26,6 +27,8 @@ func mapPostgresEvent(_ context.Context, e fkevent.Event) ([]fksinks.PostgresWri return mapWeatherStoryEvent(e) case standards.SchemaWeatherAlertV1: return mapAlertEvent(e) + case standards.SchemaWeatherOutlookV1: + return mapOutlookEvent(e) default: return nil, nil } @@ -356,6 +359,107 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { return writes, nil } +func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { + run, err := decodePayload[model.WeatherOutlookRun](e.Payload) + if err != nil { + return nil, fmt.Errorf("decode outlook payload: %w", err) + } + if run.AsOf.IsZero() { + return nil, fmt.Errorf("decode outlook payload: asOf is required") + } + + asOf := run.AsOf.UTC() + writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks)) + writes = append(writes, fksinks.PostgresWrite{ + Table: tableOutlookRuns, + Values: map[string]any{ + "event_id": e.ID, + "event_kind": string(e.Kind), + "event_source": e.Source, + "event_schema": e.Schema, + "event_emitted_at": e.EmittedAt.UTC(), + "event_effective_at": nullableTime(e.EffectiveAt), + "location_id": nullableString(run.LocationID), + "location_name": nullableString(run.LocationName), + "latitude": nullableFloat64(run.Latitude), + "longitude": nullableFloat64(run.Longitude), + "as_of": asOf, + "issued_at": nullableTime(run.IssuedAt), + "outlook_count": len(run.Outlooks), + }, + }) + + for i, outlook := range run.Outlooks { + if err := validateOutlook(outlook, i); err != nil { + return nil, err + } + geometryJSON, err := requiredCompactJSONText(outlook.Geometry) + if err != nil { + return nil, fmt.Errorf("decode outlook payload: outlooks[%d].geometry: %w", i, err) + } + + writes = append(writes, fksinks.PostgresWrite{ + Table: tableOutlooks, + Values: map[string]any{ + "run_event_id": e.ID, + "outlook_index": i, + "as_of": asOf, + "product": outlook.Product, + "day": outlook.Day, + "outlook_type": outlook.OutlookType, + "label": outlook.Label, + "label_text": nullableString(outlook.LabelText), + "severity_rank": nullableInt(outlook.SeverityRank), + "valid_from": outlook.ValidFrom.UTC(), + "valid_to": outlook.ValidTo.UTC(), + "issued_at": outlook.IssuedAt.UTC(), + "expires_at": outlook.ExpiresAt.UTC(), + "forecaster": nullableString(outlook.Forecaster), + "headline": nullableString(outlook.Headline), + "summary": nullableString(outlook.Summary), + "discussion": nullableString(outlook.Discussion), + "source_url": nullableString(outlook.SourceURL), + "image_url": nullableString(outlook.ImageURL), + "contains_location": outlook.ContainsLocation, + "geometry_json": geometryJSON, + }, + }) + } + + return writes, nil +} + +func validateOutlook(outlook model.WeatherOutlook, index int) error { + if strings.TrimSpace(outlook.ID) == "" { + return fmt.Errorf("decode outlook payload: outlooks[%d].id is required", index) + } + if strings.TrimSpace(outlook.Provider) == "" { + return fmt.Errorf("decode outlook payload: outlooks[%d].provider is required", index) + } + if strings.TrimSpace(outlook.Product) == "" { + return fmt.Errorf("decode outlook payload: outlooks[%d].product is required", index) + } + if outlook.Day == 0 { + return fmt.Errorf("decode outlook payload: outlooks[%d].day is required", index) + } + if strings.TrimSpace(outlook.OutlookType) == "" { + return fmt.Errorf("decode outlook payload: outlooks[%d].outlookType is required", index) + } + if strings.TrimSpace(outlook.Label) == "" { + return fmt.Errorf("decode outlook payload: outlooks[%d].label is required", index) + } + if outlook.ValidFrom.IsZero() || outlook.ValidTo.IsZero() { + return fmt.Errorf("decode outlook payload: outlooks[%d] validFrom/validTo are required", index) + } + if outlook.IssuedAt.IsZero() || outlook.ExpiresAt.IsZero() { + return fmt.Errorf("decode outlook payload: outlooks[%d] issuedAt/expiresAt are required", index) + } + if len(outlook.Geometry) == 0 { + return fmt.Errorf("decode outlook payload: outlooks[%d].geometry is required", index) + } + return nil +} + func decodePayload[T any](payload any) (T, error) { var out T if payload == nil { @@ -418,6 +522,13 @@ func nullableBool(v *bool) any { return *v } +func nullableInt(v *int) any { + if v == nil { + return nil + } + return *v +} + func nullableTime(v *time.Time) any { if v == nil || v.IsZero() { return nil @@ -445,3 +556,19 @@ func compactJSONText(v any) (any, error) { } return string(b), nil } + +func requiredCompactJSONText(v any) (string, error) { + compact, err := compactJSONText(v) + if err != nil { + return "", err + } + s, ok := compact.(string) + if !ok || strings.TrimSpace(s) == "" || strings.TrimSpace(s) == "null" { + return "", fmt.Errorf("is required") + } + var buf bytes.Buffer + if err := json.Compact(&buf, []byte(s)); err != nil { + return "", err + } + return buf.String(), nil +} diff --git a/internal/sinks/postgres/map_test.go b/internal/sinks/postgres/map_test.go index a4cd7ee..15b550a 100644 --- a/internal/sinks/postgres/map_test.go +++ b/internal/sinks/postgres/map_test.go @@ -239,6 +239,149 @@ func TestMapPostgresEventWeatherStoryStructPayload(t *testing.T) { assertAllWritesIncludeAllColumns(t, writes) } +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)) + severity := 3 + run := model.WeatherOutlookRun{ + LocationID: "stl", + LocationName: "St. Louis, MO", + Latitude: &lat, + Longitude: &lon, + AsOf: time.Date(2026, 6, 12, 0, 45, 0, 0, time.UTC), + IssuedAt: &issuedAt, + Outlooks: []model.WeatherOutlook{ + { + ID: "outlook-1", + Provider: "spc", + Product: "convective", + Day: 1, + OutlookType: "categorical", + Label: "SLGT", + LabelText: "Slight Risk", + SeverityRank: &severity, + ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.FixedZone("UTC-5", -5*60*60)), + ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), + 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 ] ] ] }`), + }, + { + ID: "outlook-2", + Provider: "spc", + Product: "convective", + Day: 1, + OutlookType: "wind", + Label: "15", + 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: false, + Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-100,35],[-98,35],[-98,37],[-100,37],[-100,35]]]}`), + }, + }, + } + + writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", 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 writes[0].Table != tableOutlookRuns { + t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns) + } + if got := writes[0].Values["outlook_count"]; got != 2 { + t.Fatalf("outlook_runs outlook_count = %#v, want 2", got) + } + if got := writes[0].Values["issued_at"]; got != issuedAt.UTC() { + t.Fatalf("outlook_runs issued_at = %#v, want UTC %s", got, issuedAt.UTC()) + } + if writes[1].Table != tableOutlooks || writes[2].Table != tableOutlooks { + t.Fatalf("outlook writes not in expected order") + } + if got := writes[1].Values["outlook_index"]; got != 0 { + t.Fatalf("first outlook index = %#v, want 0", got) + } + if got := writes[1].Values["valid_from"]; got != run.Outlooks[0].ValidFrom.UTC() { + t.Fatalf("first valid_from = %#v, want UTC %s", got, run.Outlooks[0].ValidFrom.UTC()) + } + 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) + } + + assertAllWritesIncludeAllColumns(t, writes) +} + +func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) { + _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", model.WeatherOutlookRun{})) + if err == nil { + t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error") + } + if !strings.Contains(err.Error(), "asOf is required") { + t.Fatalf("error = %q, want asOf context", err) + } +} + +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]]]}`), + }}, + } + _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run)) + if err == nil { + t.Fatalf("mapPostgresEvent() error = nil, want missing time error") + } + if !strings.Contains(err.Error(), "outlooks[0] validFrom/validTo are required") { + t.Fatalf("error = %q, want outlook time context", err) + } +} + +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), + }}, + } + _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run)) + if err == nil { + t.Fatalf("mapPostgresEvent() error = nil, want geometry error") + } + if !strings.Contains(err.Error(), "outlooks[0].geometry is required") { + t.Fatalf("error = %q, want geometry context", err) + } +} + func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) { _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", model.WeatherStoryRun{})) if err == nil { diff --git a/internal/sinks/postgres/schema.go b/internal/sinks/postgres/schema.go index 1d98a87..d6c5560 100644 --- a/internal/sinks/postgres/schema.go +++ b/internal/sinks/postgres/schema.go @@ -16,6 +16,8 @@ const ( tableAlertRuns = "alert_runs" tableAlerts = "alerts" tableAlertReferences = "alert_references" + tableOutlookRuns = "outlook_runs" + tableOutlooks = "outlooks" ) // PostgresSchema returns weatherfeeder's Postgres schema definition. @@ -297,6 +299,63 @@ func PostgresSchema() fksinks.PostgresSchema { {Name: "idx_wf_alert_refs_sent", Columns: []string{"sent"}}, }, }, + { + Name: tableOutlookRuns, + Columns: []fksinks.PostgresColumn{ + {Name: "event_id", Type: "TEXT", Nullable: false}, + {Name: "event_kind", Type: "TEXT", Nullable: false}, + {Name: "event_source", Type: "TEXT", Nullable: false}, + {Name: "event_schema", Type: "TEXT", Nullable: false}, + {Name: "event_emitted_at", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "event_effective_at", Type: "TIMESTAMPTZ", Nullable: true}, + {Name: "location_id", Type: "TEXT", Nullable: true}, + {Name: "location_name", Type: "TEXT", Nullable: true}, + {Name: "latitude", Type: "DOUBLE PRECISION", Nullable: true}, + {Name: "longitude", Type: "DOUBLE PRECISION", Nullable: true}, + {Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true}, + {Name: "outlook_count", Type: "INTEGER", Nullable: false}, + }, + PrimaryKey: []string{"event_id"}, + PruneColumn: "as_of", + Indexes: []fksinks.PostgresIndex{ + {Name: "idx_wf_outlook_run_location_as_of", Columns: []string{"location_id", "as_of"}}, + {Name: "idx_wf_outlook_run_as_of", Columns: []string{"as_of"}}, + }, + }, + { + Name: tableOutlooks, + Columns: []fksinks.PostgresColumn{ + {Name: "run_event_id", Type: "TEXT REFERENCES outlook_runs(event_id) ON DELETE CASCADE", Nullable: false}, + {Name: "outlook_index", Type: "INTEGER", Nullable: false}, + {Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "product", Type: "TEXT", Nullable: false}, + {Name: "day", Type: "INTEGER", Nullable: false}, + {Name: "outlook_type", Type: "TEXT", Nullable: false}, + {Name: "label", Type: "TEXT", Nullable: false}, + {Name: "label_text", Type: "TEXT", Nullable: true}, + {Name: "severity_rank", Type: "INTEGER", Nullable: true}, + {Name: "valid_from", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "valid_to", Type: "TIMESTAMPTZ", Nullable: false}, + {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}, + {Name: "geometry_json", Type: "TEXT", Nullable: false}, + }, + PrimaryKey: []string{"run_event_id", "outlook_index"}, + PruneColumn: "as_of", + Indexes: []fksinks.PostgresIndex{ + {Name: "idx_wf_outlooks_contains_valid", Columns: []string{"contains_location", "valid_from", "valid_to"}}, + {Name: "idx_wf_outlooks_day_type_label", Columns: []string{"day", "outlook_type", "label"}}, + {Name: "idx_wf_outlooks_valid", Columns: []string{"valid_from", "valid_to"}}, + }, + }, }, MapEvent: mapPostgresEvent, } diff --git a/internal/sinks/postgres/schema_test.go b/internal/sinks/postgres/schema_test.go index 6c92e29..32f775d 100644 --- a/internal/sinks/postgres/schema_test.go +++ b/internal/sinks/postgres/schema_test.go @@ -1,6 +1,9 @@ package postgres -import "testing" +import ( + "strings" + "testing" +) func TestWeatherPostgresSchemaShape(t *testing.T) { s := PostgresSchema() @@ -20,6 +23,8 @@ func TestWeatherPostgresSchemaShape(t *testing.T) { tableAlertRuns: true, tableAlerts: true, tableAlertReferences: true, + tableOutlookRuns: true, + tableOutlooks: true, } if len(s.Tables) != len(wantTables) { @@ -43,6 +48,29 @@ 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"} { + if !runColumns[col] { + t.Fatalf("%s missing %s column", tableOutlookRuns, col) + } + } + assertTablePrimaryKey(t, tableOutlookRuns, []string{"event_id"}) + assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_location_as_of", []string{"location_id", "as_of"}) + 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", "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"} { + if !outlookColumns[col] { + t.Fatalf("%s missing %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"}) +} + func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) { runColumns := columnsForTable(t, tableWeatherStoryRuns) if !runColumns["as_of"] { @@ -60,6 +88,39 @@ func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(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 + } + 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 + } + } + t.Fatalf("%s missing index %s", table, name) + } + t.Fatalf("missing table %q", table) +} + func columnsForTable(t *testing.T, table string) map[string]bool { t.Helper()