From 6a0b30b7c7e3304293f8ec07d2b4077620df167b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 11 Jun 2026 02:15:52 +0000 Subject: [PATCH] Centralize Postgres event envelope mapping --- internal/sinks/postgres/map.go | 125 ++++++++++++++------------------- 1 file changed, 52 insertions(+), 73 deletions(-) diff --git a/internal/sinks/postgres/map.go b/internal/sinks/postgres/map.go index e6adebd..2a2613e 100644 --- a/internal/sinks/postgres/map.go +++ b/internal/sinks/postgres/map.go @@ -48,13 +48,7 @@ func mapObservationEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { writes = append(writes, fksinks.PostgresWrite{ Table: tableObservations, - 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), + Values: parentEventValues(e, map[string]any{ "station_id": nullableString(obs.StationID), "station_name": nullableString(obs.StationName), "observed_at": observedAt, @@ -70,7 +64,7 @@ func mapObservationEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { "visibility_meters": nullableFloat64(obs.VisibilityMeters), "relative_humidity_percent": nullableFloat64(obs.RelativeHumidityPercent), "apparent_temperature_c": nullableFloat64(obs.ApparentTemperatureC), - }, + }), }) for i, pw := range obs.PresentWeather { @@ -109,23 +103,17 @@ func mapForecastEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { writes = append(writes, fksinks.PostgresWrite{ Table: tableForecasts, - 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), - "issued_at": issuedAt, - "updated_at": nullableTime(run.UpdatedAt), - "product": string(run.Product), - "latitude": nullableFloat64(run.Latitude), - "longitude": nullableFloat64(run.Longitude), - "elevation_meters": nullableFloat64(run.ElevationMeters), - "period_count": len(run.Periods), - }, + Values: parentEventValues(e, map[string]any{ + "location_id": nullableString(run.LocationID), + "location_name": nullableString(run.LocationName), + "issued_at": issuedAt, + "updated_at": nullableTime(run.UpdatedAt), + "product": string(run.Product), + "latitude": nullableFloat64(run.Latitude), + "longitude": nullableFloat64(run.Longitude), + "elevation_meters": nullableFloat64(run.ElevationMeters), + "period_count": len(run.Periods), + }), }) for i, p := range run.Periods { @@ -186,13 +174,7 @@ func mapForecastDiscussionEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error writes := make([]fksinks.PostgresWrite, 0, 1+len(run.KeyMessages)) writes = append(writes, fksinks.PostgresWrite{ Table: tableForecastDiscussions, - 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), + Values: parentEventValues(e, map[string]any{ "office_id": nullableString(run.OfficeID), "office_name": nullableString(run.OfficeName), "issued_at": issuedAt, @@ -205,7 +187,7 @@ func mapForecastDiscussionEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error "long_term_issued_at": longTermIssuedAt, "long_term_text": longTermText, "key_message_count": len(run.KeyMessages), - }, + }), }) for i, msg := range run.KeyMessages { @@ -236,17 +218,11 @@ func mapWeatherStoryEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Stories)) writes = append(writes, fksinks.PostgresWrite{ Table: tableWeatherStoryRuns, - 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), - "office_id": nullableString(run.OfficeID), - "as_of": asOf, - "story_count": len(run.Stories), - }, + Values: parentEventValues(e, map[string]any{ + "office_id": nullableString(run.OfficeID), + "as_of": asOf, + "story_count": len(run.Stories), + }), }) for i, story := range run.Stories { @@ -290,20 +266,14 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { writes = append(writes, fksinks.PostgresWrite{ Table: tableAlertRuns, - 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), - "as_of": asOf, - "latitude": nullableFloat64(run.Latitude), - "longitude": nullableFloat64(run.Longitude), - "alert_count": len(run.Alerts), - }, + Values: parentEventValues(e, map[string]any{ + "location_id": nullableString(run.LocationID), + "location_name": nullableString(run.LocationName), + "as_of": asOf, + "latitude": nullableFloat64(run.Latitude), + "longitude": nullableFloat64(run.Longitude), + "alert_count": len(run.Alerts), + }), }) for i, a := range run.Alerts { @@ -372,21 +342,15 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { 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), - }, + 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), + }), }) for i, outlook := range run.Outlooks { @@ -488,6 +452,21 @@ func decodePayload[T any](payload any) (T, error) { return out, nil } +func parentEventValues(e fkevent.Event, values map[string]any) map[string]any { + out := 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), + } + for k, v := range values { + out[k] = v + } + return out +} + func nullableDiscussionSection(section *model.WeatherForecastDiscussionSection) (any, any, any) { if section == nil { return nil, nil, nil