diff --git a/internal/sinks/postgres/map.go b/internal/sinks/postgres/map.go index 303c809..1a68830 100644 --- a/internal/sinks/postgres/map.go +++ b/internal/sinks/postgres/map.go @@ -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(""), - "summary": nullableString(""), - "discussion": nullableString(""), "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 } diff --git a/internal/sinks/postgres/map_test.go b/internal/sinks/postgres/map_test.go index ae04c92..62fda91 100644 --- a/internal/sinks/postgres/map_test.go +++ b/internal/sinks/postgres/map_test.go @@ -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", @@ -281,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) @@ -300,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()) } @@ -321,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") } @@ -340,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 { @@ -378,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) } @@ -393,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") } @@ -415,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") } @@ -436,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 { @@ -505,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 { @@ -644,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 diff --git a/internal/sinks/postgres/schema.go b/internal/sinks/postgres/schema.go index 73923f0..f95e397 100644 --- a/internal/sinks/postgres/schema.go +++ b/internal/sinks/postgres/schema.go @@ -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, } diff --git a/internal/sinks/postgres/schema_test.go b/internal/sinks/postgres/schema_test.go index 5d03f58..c090f34 100644 --- a/internal/sinks/postgres/schema_test.go +++ b/internal/sinks/postgres/schema_test.go @@ -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 {