ejr/weatherfeeder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update Postgres outlook storage for v2

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-12 04:29:00 +00:00
parent 21a35a5205
commit 4d2cddf801
4 changed files with 325 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
internal/sinks/postgres/schema_test.go
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 {