package postgres import ( "context" "database/sql" "encoding/json" "errors" "fmt" "strings" "time" "gitea.maximumdirect.net/ejr/weatherapi/internal/core" "gitea.maximumdirect.net/ejr/weatherfeeder/model" ) const ( queryLatestObservation = ` SELECT event_id, station_id, station_name, observed_at, condition_code, is_day, text_description, temperature_c, dewpoint_c, wind_direction_degrees, wind_speed_kmh, wind_gust_kmh, barometric_pressure_pa, visibility_meters, relative_humidity_percent, apparent_temperature_c FROM observations ORDER BY observed_at DESC, event_emitted_at DESC LIMIT 1` queryObservationPresentWeather = ` SELECT weather_index, raw_text FROM observation_present_weather WHERE event_id = $1 ORDER BY weather_index ASC` queryLatestHourlyForecast = ` SELECT event_id, location_id, location_name, issued_at, updated_at, product, latitude, longitude, elevation_meters FROM forecasts WHERE product = 'hourly' ORDER BY issued_at DESC, event_emitted_at DESC LIMIT 1` queryForecastPeriods = ` SELECT period_index, start_time, end_time, name, is_day, condition_code, condition_text, provider_raw_description, text_description, detailed_text, icon_url, temperature_c, temperature_c_min, temperature_c_max, dewpoint_c, relative_humidity_percent, wind_direction_degrees, wind_speed_kmh, wind_gust_kmh, barometric_pressure_pa, visibility_meters, apparent_temperature_c, cloud_cover_percent, probability_of_precipitation_percent, precipitation_amount_mm, snowfall_depth_mm, uv_index FROM forecast_periods WHERE run_event_id = $1 ORDER BY period_index ASC` queryLatestAlertRun = ` SELECT event_id, location_id, location_name, as_of, latitude, longitude FROM alert_runs ORDER BY as_of DESC, event_emitted_at DESC LIMIT 1` queryAlerts = ` SELECT alert_index, alert_id, event, headline, severity, urgency, certainty, status, message_type, category, response, description, instruction, sent, effective, onset, expires, area_description, sender_name FROM alerts WHERE run_event_id = $1 ORDER BY alert_index ASC` queryAlertReferences = ` SELECT alert_index, id, identifier, sender, sent FROM alert_references WHERE run_event_id = $1 ORDER BY alert_index ASC, reference_index ASC` ) // Repository is a Postgres-backed implementation of weatherapi read ports. type Repository struct { db *sql.DB } var _ core.Repository = (*Repository)(nil) func NewRepository(db *sql.DB) *Repository { return &Repository{db: db} } func (r *Repository) LatestObservation(ctx context.Context) (*model.WeatherObservation, error) { if r == nil || r.db == nil { return nil, fmt.Errorf("postgres repository is not configured") } var row observationParentRow err := r.db.QueryRowContext(ctx, queryLatestObservation).Scan( &row.EventID, &row.StationID, &row.StationName, &row.ObservedAt, &row.ConditionCode, &row.IsDay, &row.TextDescription, &row.TemperatureC, &row.DewpointC, &row.WindDirectionDegrees, &row.WindSpeedKmh, &row.WindGustKmh, &row.BarometricPressurePa, &row.VisibilityMeters, &row.RelativeHumidityPercent, &row.ApparentTemperatureC, ) if errors.Is(err, sql.ErrNoRows) { return nil, nil } if err != nil { return nil, fmt.Errorf("query latest observation: %w", err) } obs := mapObservationParentRow(row) presentWeather, err := r.loadObservationPresentWeather(ctx, row.EventID) if err != nil { return nil, err } obs.PresentWeather = presentWeather return &obs, nil } func (r *Repository) LatestHourlyForecast(ctx context.Context) (*model.WeatherForecastRun, error) { if r == nil || r.db == nil { return nil, fmt.Errorf("postgres repository is not configured") } var row forecastParentRow err := r.db.QueryRowContext(ctx, queryLatestHourlyForecast).Scan( &row.EventID, &row.LocationID, &row.LocationName, &row.IssuedAt, &row.UpdatedAt, &row.Product, &row.Latitude, &row.Longitude, &row.ElevationMeters, ) if errors.Is(err, sql.ErrNoRows) { return nil, nil } if err != nil { return nil, fmt.Errorf("query latest hourly forecast: %w", err) } run := mapForecastParentRow(row) periods, err := r.loadForecastPeriods(ctx, row.EventID) if err != nil { return nil, err } run.Periods = periods return &run, nil } func (r *Repository) LatestActiveAlerts(ctx context.Context) (*model.WeatherAlertRun, error) { if r == nil || r.db == nil { return nil, fmt.Errorf("postgres repository is not configured") } var row alertRunParentRow err := r.db.QueryRowContext(ctx, queryLatestAlertRun).Scan( &row.EventID, &row.LocationID, &row.LocationName, &row.AsOf, &row.Latitude, &row.Longitude, ) if errors.Is(err, sql.ErrNoRows) { return nil, nil } if err != nil { return nil, fmt.Errorf("query latest alert run: %w", err) } run := mapAlertRunParentRow(row) alerts, err := r.loadAlerts(ctx, row.EventID) if err != nil { return nil, err } run.Alerts = alerts return &run, nil } func (r *Repository) loadObservationPresentWeather(ctx context.Context, eventID string) ([]model.PresentWeather, error) { rows, err := r.db.QueryContext(ctx, queryObservationPresentWeather, eventID) if err != nil { return nil, fmt.Errorf("query observation present weather: %w", err) } defer rows.Close() out := make([]model.PresentWeather, 0) for rows.Next() { var row observationPresentWeatherRow if err := rows.Scan(&row.WeatherIndex, &row.RawText); err != nil { return nil, fmt.Errorf("scan observation present weather row: %w", err) } pw, err := mapObservationPresentWeatherRow(row) if err != nil { return nil, fmt.Errorf("decode observation present weather row (index=%d): %w", row.WeatherIndex, err) } out = append(out, pw) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate observation present weather rows: %w", err) } return out, nil } func (r *Repository) loadForecastPeriods(ctx context.Context, eventID string) ([]model.WeatherForecastPeriod, error) { rows, err := r.db.QueryContext(ctx, queryForecastPeriods, eventID) if err != nil { return nil, fmt.Errorf("query forecast periods: %w", err) } defer rows.Close() out := make([]model.WeatherForecastPeriod, 0) for rows.Next() { var row forecastPeriodRow if err := rows.Scan( &row.PeriodIndex, &row.StartTime, &row.EndTime, &row.Name, &row.IsDay, &row.ConditionCode, &row.ConditionText, &row.ProviderRawDescription, &row.TextDescription, &row.DetailedText, &row.IconURL, &row.TemperatureC, &row.TemperatureCMin, &row.TemperatureCMax, &row.DewpointC, &row.RelativeHumidityPercent, &row.WindDirectionDegrees, &row.WindSpeedKmh, &row.WindGustKmh, &row.BarometricPressurePa, &row.VisibilityMeters, &row.ApparentTemperatureC, &row.CloudCoverPercent, &row.ProbabilityOfPrecipitationPercent, &row.PrecipitationAmountMM, &row.SnowfallDepthMM, &row.UVIndex, ); err != nil { return nil, fmt.Errorf("scan forecast period row: %w", err) } out = append(out, mapForecastPeriodRow(row)) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate forecast period rows: %w", err) } return out, nil } func (r *Repository) loadAlerts(ctx context.Context, eventID string) ([]model.WeatherAlert, error) { alertsRows, err := r.db.QueryContext(ctx, queryAlerts, eventID) if err != nil { return nil, fmt.Errorf("query alerts: %w", err) } defer alertsRows.Close() indexedAlerts := make([]indexedAlert, 0) for alertsRows.Next() { var row alertRow if err := alertsRows.Scan( &row.AlertIndex, &row.AlertID, &row.Event, &row.Headline, &row.Severity, &row.Urgency, &row.Certainty, &row.Status, &row.MessageType, &row.Category, &row.Response, &row.Description, &row.Instruction, &row.Sent, &row.Effective, &row.Onset, &row.Expires, &row.AreaDescription, &row.SenderName, ); err != nil { return nil, fmt.Errorf("scan alerts row: %w", err) } indexedAlerts = append(indexedAlerts, mapAlertRow(row)) } if err := alertsRows.Err(); err != nil { return nil, fmt.Errorf("iterate alerts rows: %w", err) } referenceRows, err := r.db.QueryContext(ctx, queryAlertReferences, eventID) if err != nil { return nil, fmt.Errorf("query alert references: %w", err) } defer referenceRows.Close() indexedReferences := make([]indexedAlertReference, 0) for referenceRows.Next() { var row alertReferenceRow if err := referenceRows.Scan( &row.AlertIndex, &row.ID, &row.Identifier, &row.Sender, &row.Sent, ); err != nil { return nil, fmt.Errorf("scan alert reference row: %w", err) } indexedReferences = append(indexedReferences, mapAlertReferenceRow(row)) } if err := referenceRows.Err(); err != nil { return nil, fmt.Errorf("iterate alert reference rows: %w", err) } return attachAlertReferences(indexedAlerts, indexedReferences), nil } type observationParentRow struct { EventID string StationID sql.NullString StationName sql.NullString ObservedAt time.Time ConditionCode int IsDay sql.NullBool TextDescription sql.NullString TemperatureC sql.NullFloat64 DewpointC sql.NullFloat64 WindDirectionDegrees sql.NullFloat64 WindSpeedKmh sql.NullFloat64 WindGustKmh sql.NullFloat64 BarometricPressurePa sql.NullFloat64 VisibilityMeters sql.NullFloat64 RelativeHumidityPercent sql.NullFloat64 ApparentTemperatureC sql.NullFloat64 } func mapObservationParentRow(row observationParentRow) model.WeatherObservation { return model.WeatherObservation{ StationID: stringValue(row.StationID), StationName: stringValue(row.StationName), Timestamp: row.ObservedAt.UTC(), ConditionCode: model.WMOCode(row.ConditionCode), IsDay: boolPtr(row.IsDay), TextDescription: stringValue(row.TextDescription), TemperatureC: float64Ptr(row.TemperatureC), DewpointC: float64Ptr(row.DewpointC), WindDirectionDegrees: float64Ptr(row.WindDirectionDegrees), WindSpeedKmh: float64Ptr(row.WindSpeedKmh), WindGustKmh: float64Ptr(row.WindGustKmh), BarometricPressurePa: float64Ptr(row.BarometricPressurePa), VisibilityMeters: float64Ptr(row.VisibilityMeters), RelativeHumidityPercent: float64Ptr(row.RelativeHumidityPercent), ApparentTemperatureC: float64Ptr(row.ApparentTemperatureC), } } type observationPresentWeatherRow struct { WeatherIndex int RawText sql.NullString } func mapObservationPresentWeatherRow(row observationPresentWeatherRow) (model.PresentWeather, error) { if !row.RawText.Valid || strings.TrimSpace(row.RawText.String) == "" { return model.PresentWeather{}, nil } var raw map[string]any if err := json.Unmarshal([]byte(row.RawText.String), &raw); err != nil { return model.PresentWeather{}, err } return model.PresentWeather{Raw: raw}, nil } type forecastParentRow struct { EventID string LocationID sql.NullString LocationName sql.NullString IssuedAt time.Time UpdatedAt sql.NullTime Product string Latitude sql.NullFloat64 Longitude sql.NullFloat64 ElevationMeters sql.NullFloat64 } func mapForecastParentRow(row forecastParentRow) model.WeatherForecastRun { return model.WeatherForecastRun{ LocationID: stringValue(row.LocationID), LocationName: stringValue(row.LocationName), IssuedAt: row.IssuedAt.UTC(), UpdatedAt: timePtr(row.UpdatedAt), Product: model.ForecastProduct(row.Product), Latitude: float64Ptr(row.Latitude), Longitude: float64Ptr(row.Longitude), ElevationMeters: float64Ptr(row.ElevationMeters), } } type forecastPeriodRow struct { PeriodIndex int StartTime time.Time EndTime time.Time Name sql.NullString IsDay sql.NullBool ConditionCode int ConditionText sql.NullString ProviderRawDescription sql.NullString TextDescription sql.NullString DetailedText sql.NullString IconURL sql.NullString TemperatureC sql.NullFloat64 TemperatureCMin sql.NullFloat64 TemperatureCMax sql.NullFloat64 DewpointC sql.NullFloat64 RelativeHumidityPercent sql.NullFloat64 WindDirectionDegrees sql.NullFloat64 WindSpeedKmh sql.NullFloat64 WindGustKmh sql.NullFloat64 BarometricPressurePa sql.NullFloat64 VisibilityMeters sql.NullFloat64 ApparentTemperatureC sql.NullFloat64 CloudCoverPercent sql.NullFloat64 ProbabilityOfPrecipitationPercent sql.NullFloat64 PrecipitationAmountMM sql.NullFloat64 SnowfallDepthMM sql.NullFloat64 UVIndex sql.NullFloat64 } func mapForecastPeriodRow(row forecastPeriodRow) model.WeatherForecastPeriod { return model.WeatherForecastPeriod{ StartTime: row.StartTime.UTC(), EndTime: row.EndTime.UTC(), Name: stringValue(row.Name), IsDay: boolPtr(row.IsDay), ConditionCode: model.WMOCode(row.ConditionCode), ConditionText: stringValue(row.ConditionText), ProviderRawDescription: stringValue(row.ProviderRawDescription), TextDescription: stringValue(row.TextDescription), DetailedText: stringValue(row.DetailedText), IconURL: stringValue(row.IconURL), TemperatureC: float64Ptr(row.TemperatureC), TemperatureCMin: float64Ptr(row.TemperatureCMin), TemperatureCMax: float64Ptr(row.TemperatureCMax), DewpointC: float64Ptr(row.DewpointC), RelativeHumidityPercent: float64Ptr(row.RelativeHumidityPercent), WindDirectionDegrees: float64Ptr(row.WindDirectionDegrees), WindSpeedKmh: float64Ptr(row.WindSpeedKmh), WindGustKmh: float64Ptr(row.WindGustKmh), BarometricPressurePa: float64Ptr(row.BarometricPressurePa), VisibilityMeters: float64Ptr(row.VisibilityMeters), ApparentTemperatureC: float64Ptr(row.ApparentTemperatureC), CloudCoverPercent: float64Ptr(row.CloudCoverPercent), ProbabilityOfPrecipitationPercent: float64Ptr(row.ProbabilityOfPrecipitationPercent), PrecipitationAmountMm: float64Ptr(row.PrecipitationAmountMM), SnowfallDepthMM: float64Ptr(row.SnowfallDepthMM), UVIndex: float64Ptr(row.UVIndex), } } type alertRunParentRow struct { EventID string LocationID sql.NullString LocationName sql.NullString AsOf time.Time Latitude sql.NullFloat64 Longitude sql.NullFloat64 } func mapAlertRunParentRow(row alertRunParentRow) model.WeatherAlertRun { return model.WeatherAlertRun{ LocationID: stringValue(row.LocationID), LocationName: stringValue(row.LocationName), AsOf: row.AsOf.UTC(), Latitude: float64Ptr(row.Latitude), Longitude: float64Ptr(row.Longitude), } } type alertRow struct { AlertIndex int AlertID string Event sql.NullString Headline sql.NullString Severity sql.NullString Urgency sql.NullString Certainty sql.NullString Status sql.NullString MessageType sql.NullString Category sql.NullString Response sql.NullString Description sql.NullString Instruction sql.NullString Sent sql.NullTime Effective sql.NullTime Onset sql.NullTime Expires sql.NullTime AreaDescription sql.NullString SenderName sql.NullString } type indexedAlert struct { Index int Alert model.WeatherAlert } func mapAlertRow(row alertRow) indexedAlert { return indexedAlert{ Index: row.AlertIndex, Alert: model.WeatherAlert{ ID: row.AlertID, Event: stringValue(row.Event), Headline: stringValue(row.Headline), Severity: stringValue(row.Severity), Urgency: stringValue(row.Urgency), Certainty: stringValue(row.Certainty), Status: stringValue(row.Status), MessageType: stringValue(row.MessageType), Category: stringValue(row.Category), Response: stringValue(row.Response), Description: stringValue(row.Description), Instruction: stringValue(row.Instruction), Sent: timePtr(row.Sent), Effective: timePtr(row.Effective), Onset: timePtr(row.Onset), Expires: timePtr(row.Expires), AreaDescription: stringValue(row.AreaDescription), SenderName: stringValue(row.SenderName), }, } } type alertReferenceRow struct { AlertIndex int ID sql.NullString Identifier sql.NullString Sender sql.NullString Sent sql.NullTime } type indexedAlertReference struct { AlertIndex int Reference model.AlertReference } func mapAlertReferenceRow(row alertReferenceRow) indexedAlertReference { return indexedAlertReference{ AlertIndex: row.AlertIndex, Reference: model.AlertReference{ ID: stringValue(row.ID), Identifier: stringValue(row.Identifier), Sender: stringValue(row.Sender), Sent: timePtr(row.Sent), }, } } func attachAlertReferences(alerts []indexedAlert, references []indexedAlertReference) []model.WeatherAlert { refsByAlertIndex := make(map[int][]model.AlertReference, len(alerts)) for _, ref := range references { refsByAlertIndex[ref.AlertIndex] = append(refsByAlertIndex[ref.AlertIndex], ref.Reference) } out := make([]model.WeatherAlert, 0, len(alerts)) for _, alert := range alerts { mapped := alert.Alert if refs := refsByAlertIndex[alert.Index]; len(refs) > 0 { mapped.References = refs } out = append(out, mapped) } return out } func stringValue(v sql.NullString) string { if !v.Valid { return "" } return v.String } func boolPtr(v sql.NullBool) *bool { if !v.Valid { return nil } b := v.Bool return &b } func float64Ptr(v sql.NullFloat64) *float64 { if !v.Valid { return nil } f := v.Float64 return &f } func timePtr(v sql.NullTime) *time.Time { if !v.Valid { return nil } t := v.Time.UTC() return &t }