Implement remaining cleanup items prior to the next release

internal/sinks/postgres/map_test.go

@@ -528,6 +528,56 @@ func TestMapPostgresEventForecastDiscussionMalformedPayload(t *testing.T) {
 	}
 }
 
+func TestParentEventValuesAddsEnvelopeAndPreservesProductValues(t *testing.T) {
+	emittedAt := time.Date(2026, 3, 16, 13, 31, 0, 0, time.FixedZone("CDT", -5*60*60))
+	effectiveAt := time.Date(2026, 3, 16, 13, 30, 0, 0, time.FixedZone("CDT", -5*60*60))
+	event := fkevent.Event{
+		ID:          "evt-envelope",
+		Kind:        fkevent.Kind(standards.KindForecast),
+		Source:      "test-source",
+		Schema:      standards.SchemaWeatherForecastV1,
+		EmittedAt:   emittedAt,
+		EffectiveAt: &effectiveAt,
+	}
+
+	got := parentEventValues(event, map[string]any{"product_col": "product-value"})
+
+	assertParentEnvelopeValues(t, got, event)
+	if got["product_col"] != "product-value" {
+		t.Fatalf("product_col = %#v, want product-value", got["product_col"])
+	}
+}
+
+func TestParentEventValuesNullEffectiveAt(t *testing.T) {
+	base := fkevent.Event{
+		ID:        "evt-envelope",
+		Kind:      fkevent.Kind(standards.KindObservation),
+		Source:    "test-source",
+		Schema:    standards.SchemaWeatherObservationV1,
+		EmittedAt: time.Date(2026, 3, 16, 18, 31, 0, 0, time.UTC),
+	}
+
+	for _, tt := range []struct {
+		name string
+		mut  func(*fkevent.Event)
+	}{
+		{name: "nil", mut: func(*fkevent.Event) {}},
+		{name: "zero", mut: func(event *fkevent.Event) {
+			zero := time.Time{}
+			event.EffectiveAt = &zero
+		}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			event := base
+			tt.mut(&event)
+			got := parentEventValues(event, nil)
+			if got["event_effective_at"] != nil {
+				t.Fatalf("event_effective_at = %#v, want nil", got["event_effective_at"])
+			}
+		})
+	}
+}
+
 func testEvent(schema string, kind fkevent.Kind, payload any) fkevent.Event {
 	effectiveAt := time.Date(2026, 3, 16, 18, 30, 0, 0, time.UTC)
 	return fkevent.Event{
@@ -541,6 +591,30 @@ func testEvent(schema string, kind fkevent.Kind, payload any) fkevent.Event {
 	}
 }
 
+func assertParentEnvelopeValues(t *testing.T, values map[string]any, event fkevent.Event) {
+	t.Helper()
+
+	if got := values["event_id"]; got != event.ID {
+		t.Fatalf("event_id = %#v, want %q", got, event.ID)
+	}
+	if got := values["event_kind"]; got != string(event.Kind) {
+		t.Fatalf("event_kind = %#v, want %q", got, event.Kind)
+	}
+	if got := values["event_source"]; got != event.Source {
+		t.Fatalf("event_source = %#v, want %q", got, event.Source)
+	}
+	if got := values["event_schema"]; got != event.Schema {
+		t.Fatalf("event_schema = %#v, want %q", got, event.Schema)
+	}
+	if got := values["event_emitted_at"]; got != event.EmittedAt.UTC() {
+		t.Fatalf("event_emitted_at = %#v, want %s", got, event.EmittedAt.UTC())
+	}
+	wantEffective := nullableTime(event.EffectiveAt)
+	if got := values["event_effective_at"]; got != wantEffective {
+		t.Fatalf("event_effective_at = %#v, want %#v", got, wantEffective)
+	}
+}
+
 func firstWriteForTable(writes []fksinks.PostgresWrite, table string) (fksinks.PostgresWrite, bool) {
 	for _, w := range writes {
 		if w.Table == table {