package postgres

import (
	"context"
	"encoding/json"
	"strings"
	"testing"
	"time"

	fkevent "gitea.maximumdirect.net/ejr/feedkit/event"
	fksinks "gitea.maximumdirect.net/ejr/feedkit/sinks"
	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
)

func TestMapPostgresEventObservationStructPayload(t *testing.T) {
	isDay := true
	temp := 21.5
	obs := model.WeatherObservation{
		StationID:       "KSTL",
		StationName:     "St. Louis",
		Timestamp:       time.Date(2026, 3, 16, 19, 0, 0, 0, time.UTC),
		ConditionCode:   model.WMOCode(1),
		IsDay:           &isDay,
		TextDescription: "few clouds",
		TemperatureC:    &temp,
		PresentWeather:  []model.PresentWeather{{Raw: map[string]any{"a": 1, "b": "x"}}},
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherObservationV1, standards.KindObservation, obs))
	if err != nil {
		t.Fatalf("mapPostgresEvent() error = %v", err)
	}
	if len(writes) != 2 {
		t.Fatalf("mapPostgresEvent() writes len = %d, want 2", len(writes))
	}
	if writes[0].Table != tableObservations {
		t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableObservations)
	}
	if got := writes[0].Values["station_id"]; got != "KSTL" {
		t.Fatalf("observations station_id = %#v, want KSTL", got)
	}
	if writes[1].Table != tableObservationPresentWeather {
		t.Fatalf("writes[1].Table = %q, want %q", writes[1].Table, tableObservationPresentWeather)
	}
	if got := writes[1].Values["raw_text"]; got != `{"a":1,"b":"x"}` {
		t.Fatalf("present_weather raw_text = %#v, want compact JSON", got)
	}

	assertAllWritesIncludeAllColumns(t, writes)
}

func TestMapPostgresEventForecastStructPayload(t *testing.T) {
	isDay := true
	temp := 10.5
	run := model.WeatherForecastRun{
		LocationID:   "LOC-1",
		LocationName: "St. Louis",
		IssuedAt:     time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
		Product:      model.ForecastProductHourly,
		Periods: []model.WeatherForecastPeriod{
			{
				StartTime:     time.Date(2026, 3, 16, 19, 0, 0, 0, time.UTC),
				EndTime:       time.Date(2026, 3, 16, 20, 0, 0, 0, time.UTC),
				IsDay:         &isDay,
				ConditionCode: wmoCodePtr(model.WMOCode(2)),
				TemperatureC:  &temp,
			},
			{
				StartTime:     time.Date(2026, 3, 16, 20, 0, 0, 0, time.UTC),
				EndTime:       time.Date(2026, 3, 16, 21, 0, 0, 0, time.UTC),
				ConditionCode: nil,
			},
		},
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, 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 != tableForecasts {
		t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableForecasts)
	}
	if got := writes[0].Values["period_count"]; got != 2 {
		t.Fatalf("forecasts period_count = %#v, want 2", got)
	}
	if writes[1].Table != tableForecastPeriods || writes[2].Table != tableForecastPeriods {
		t.Fatalf("forecast period writes not in expected order")
	}
	if got := writes[1].Values["period_index"]; got != 0 {
		t.Fatalf("first period index = %#v, want 0", got)
	}
	if got := writes[2].Values["condition_code"]; got != nil {
		t.Fatalf("second period condition_code = %#v, want nil", got)
	}

	assertAllWritesIncludeAllColumns(t, writes)
}

func TestMapPostgresEventAlertStructPayload(t *testing.T) {
	sent := time.Date(2026, 3, 16, 17, 0, 0, 0, time.UTC)
	run := model.WeatherAlertRun{
		AsOf: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
		Alerts: []model.WeatherAlert{
			{
				ID:       "urn:alert:1",
				Headline: "Winter Weather Advisory",
				Severity: "Moderate",
				References: []model.AlertReference{
					{ID: "urn:ref:1", Sent: &sent},
					{Identifier: "ref-two"},
				},
			},
			{
				ID:       "urn:alert:2",
				Headline: "Second alert",
			},
		},
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherAlertV1, standards.KindAlert, run))
	if err != nil {
		t.Fatalf("mapPostgresEvent() error = %v", err)
	}
	if len(writes) != 5 {
		t.Fatalf("mapPostgresEvent() writes len = %d, want 5", len(writes))
	}

	counts := map[string]int{}
	for _, w := range writes {
		counts[w.Table]++
	}
	if counts[tableAlertRuns] != 1 || counts[tableAlerts] != 2 || counts[tableAlertReferences] != 2 {
		t.Fatalf("unexpected table write counts: %#v", counts)
	}

	firstAlert, ok := firstWriteForTable(writes, tableAlerts)
	if !ok {
		t.Fatalf("missing alerts write")
	}
	if got := firstAlert.Values["reference_count"]; got != 2 {
		t.Fatalf("alerts reference_count = %#v, want 2", got)
	}

	assertAllWritesIncludeAllColumns(t, writes)
}

func TestMapPostgresEventForecastDiscussionStructPayload(t *testing.T) {
	updatedAt := time.Date(2026, 3, 28, 20, 29, 47, 0, time.UTC)
	shortIssuedAt := time.Date(2026, 3, 28, 19, 19, 0, 0, time.UTC)
	run := model.WeatherForecastDiscussion{
		OfficeID:    "LSX",
		OfficeName:  "National Weather Service Saint Louis MO",
		Product:     model.ForecastDiscussionProductAFD,
		IssuedAt:    time.Date(2026, 3, 28, 19, 24, 0, 0, time.UTC),
		UpdatedAt:   &updatedAt,
		KeyMessages: []string{"msg one", "msg two"},
		ShortTerm:   &model.WeatherForecastDiscussionSection{Qualifier: "(Tonight)", IssuedAt: &shortIssuedAt, Text: "Short term text"},
		LongTerm:    &model.WeatherForecastDiscussionSection{Text: "Long term text"},
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, standards.KindForecastDiscussion, 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 != tableForecastDiscussions {
		t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableForecastDiscussions)
	}
	if got := writes[0].Values["key_message_count"]; got != 2 {
		t.Fatalf("forecast_discussions key_message_count = %#v, want 2", got)
	}
	if got := writes[0].Values["short_term_qualifier"]; got != "(Tonight)" {
		t.Fatalf("forecast_discussions short_term_qualifier = %#v, want (Tonight)", got)
	}
	if got := writes[0].Values["long_term_issued_at"]; got != nil {
		t.Fatalf("forecast_discussions long_term_issued_at = %#v, want nil", got)
	}
	if writes[1].Table != tableForecastDiscussionKeyMessages || writes[2].Table != tableForecastDiscussionKeyMessages {
		t.Fatalf("forecast discussion key message writes not in expected order")
	}
	if got := writes[2].Values["message_index"]; got != 1 {
		t.Fatalf("second key message index = %#v, want 1", got)
	}

	assertAllWritesIncludeAllColumns(t, writes)
}

func TestMapPostgresEventWeatherStoryStructPayload(t *testing.T) {
	run := model.WeatherStoryRun{
		OfficeID: "LSX",
		AsOf:     time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC),
		Stories: []model.WeatherStory{
			{
				OfficeID:    "LSX",
				StartTime:   time.Date(2026, 5, 30, 8, 46, 0, 0, time.UTC),
				EndTime:     time.Date(2026, 5, 31, 11, 0, 0, 0, time.UTC),
				UpdatedAt:   time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC),
				Title:       "Several Chances for Rain Through Monday",
				Description: "Scattered showers and thunderstorms.",
				AltText:     "This slide shows the forecast.",
				Priority:    true,
				Order:       1,
				DownloadURL: "https://api.weather.gov/offices/LSX/weatherstories/download/story-1",
			},
		},
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, run))
	if err != nil {
		t.Fatalf("mapPostgresEvent() error = %v", err)
	}
	if len(writes) != 2 {
		t.Fatalf("mapPostgresEvent() writes len = %d, want 2", len(writes))
	}
	if writes[0].Table != tableWeatherStoryRuns {
		t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableWeatherStoryRuns)
	}
	if got := writes[0].Values["story_count"]; got != 1 {
		t.Fatalf("weather_story_runs story_count = %#v, want 1", got)
	}
	if writes[1].Table != tableWeatherStories {
		t.Fatalf("writes[1].Table = %q, want %q", writes[1].Table, tableWeatherStories)
	}
	if got := writes[1].Values["download_url"]; got != "https://api.weather.gov/offices/LSX/weatherstories/download/story-1" {
		t.Fatalf("weather_stories download_url = %#v", got)
	}
	if got := writes[1].Values["story_order"]; got != 1 {
		t.Fatalf("weather_stories story_order = %#v, want 1", got)
	}

	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",
				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, 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 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["outlook_id"]; got != "outlook-1" {
		t.Fatalf("first outlook_id = %#v, want outlook-1", got)
	}
	if got := writes[1].Values["provider"]; got != "spc" {
		t.Fatalf("first provider = %#v, want spc", 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, standards.KindOutlook, 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 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]]]}`),
	}

	tests := []struct {
		name    string
		mutate  func(*model.WeatherOutlook)
		wantErr string
	}{
		{
			name:    "missing id",
			mutate:  func(outlook *model.WeatherOutlook) { outlook.ID = "" },
			wantErr: "outlooks[0].id is required",
		},
		{
			name:    "missing provider",
			mutate:  func(outlook *model.WeatherOutlook) { outlook.Provider = "" },
			wantErr: "outlooks[0].provider is required",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			outlook := base
			tt.mutate(&outlook)
			run := model.WeatherOutlookRun{
				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))
			if err == nil {
				t.Fatalf("mapPostgresEvent() error = nil, want %q", tt.wantErr)
			}
			if !strings.Contains(err.Error(), tt.wantErr) {
				t.Fatalf("error = %q, want %q", err, tt.wantErr)
			}
		})
	}
}

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, standards.KindOutlook, 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, standards.KindOutlook, 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, standards.KindWeatherStory, model.WeatherStoryRun{}))
	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 TestMapPostgresEventWeatherStoryRejectsMissingStoryTimes(t *testing.T) {
	run := model.WeatherStoryRun{
		AsOf:    time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC),
		Stories: []model.WeatherStory{{Title: "missing times"}},
	}
	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, run))
	if err == nil {
		t.Fatalf("mapPostgresEvent() error = nil, want missing story times error")
	}
	if !strings.Contains(err.Error(), "stories[0] startTime/endTime/updatedAt are required") {
		t.Fatalf("error = %q, want story time context", err)
	}
}

func TestMapPostgresEventMapPayload(t *testing.T) {
	run := model.WeatherForecastRun{
		IssuedAt: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
		Product:  model.ForecastProductHourly,
		Periods: []model.WeatherForecastPeriod{
			{
				StartTime:     time.Date(2026, 3, 16, 19, 0, 0, 0, time.UTC),
				EndTime:       time.Date(2026, 3, 16, 20, 0, 0, 0, time.UTC),
				ConditionCode: wmoCodePtr(model.WMOCode(2)),
			},
		},
	}
	b, err := json.Marshal(run)
	if err != nil {
		t.Fatalf("json.Marshal() error = %v", err)
	}
	var payload map[string]any
	if err := json.Unmarshal(b, &payload); err != nil {
		t.Fatalf("json.Unmarshal() error = %v", err)
	}

	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, payload))
	if err != nil {
		t.Fatalf("mapPostgresEvent() error = %v", err)
	}
	if len(writes) != 2 {
		t.Fatalf("mapPostgresEvent() writes len = %d, want 2", len(writes))
	}
	if writes[0].Table != tableForecasts {
		t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableForecasts)
	}

	assertAllWritesIncludeAllColumns(t, writes)
}

func TestMapPostgresEventUnknownSchemaNoOp(t *testing.T) {
	writes, err := mapPostgresEvent(context.Background(), testEvent("weather.unknown.v1", standards.KindObservation, map[string]any{"x": 1}))
	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 {
		t.Fatalf("mapPostgresEvent() expected error for malformed payload")
	}
	if !strings.Contains(err.Error(), "decode forecast payload") {
		t.Fatalf("error = %q, want decode forecast payload context", err)
	}
}

func TestMapPostgresEventForecastDiscussionMalformedPayload(t *testing.T) {
	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, standards.KindForecastDiscussion, "bad"))
	if err == nil {
		t.Fatalf("mapPostgresEvent() expected error for malformed payload")
	}
	if !strings.Contains(err.Error(), "decode forecast discussion payload") {
		t.Fatalf("error = %q, want decode forecast discussion payload context", err)
	}
}

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{
		ID:          "evt-1",
		Kind:        kind,
		Source:      "test-source",
		Schema:      schema,
		EmittedAt:   time.Date(2026, 3, 16, 18, 31, 0, 0, time.UTC),
		EffectiveAt: &effectiveAt,
		Payload:     payload,
	}
}

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 {
			return w, true
		}
	}
	return fksinks.PostgresWrite{}, false
}

func assertAllWritesIncludeAllColumns(t *testing.T, writes []fksinks.PostgresWrite) {
	t.Helper()
	colCounts := tableColumnCounts()
	for i, w := range writes {
		expectedCount, ok := colCounts[w.Table]
		if !ok {
			t.Fatalf("writes[%d] references unknown table %q", i, w.Table)
		}
		if len(w.Values) != expectedCount {
			t.Fatalf("writes[%d] table=%q has %d values, want %d", i, w.Table, len(w.Values), expectedCount)
		}
	}
}

func tableColumnCounts() map[string]int {
	s := PostgresSchema()
	m := make(map[string]int, len(s.Tables))
	for _, tbl := range s.Tables {
		m[tbl.Name] = len(tbl.Columns)
	}
	return m
}

func wmoCodePtr(v model.WMOCode) *model.WMOCode {
	out := v
	return &out
}