package postgres import ( "context" "database/sql" "encoding/json" "fmt" "time" "gitea.maximumdirect.net/ejr/weatherapi/internal/core/ports" "github.com/jackc/pgx/v5/pgxpool" ) type ObservationRepository struct { pool *pgxpool.Pool } func NewObservationRepository(pool *pgxpool.Pool) *ObservationRepository { return &ObservationRepository{pool: pool} } func (r *ObservationRepository) GetCurrentConditionsSummary(ctx context.Context, window time.Duration) (ports.ObservationCurrentConditionsMetric, error) { var temperature sql.NullFloat64 var apparent sql.NullFloat64 var dewpoint sql.NullFloat64 var relativeHumidity sql.NullFloat64 var windSpeed sql.NullFloat64 var windDirection sql.NullFloat64 var conditionCode sql.NullInt64 var isDay sql.NullBool if err := r.pool.QueryRow(ctx, queryCurrentConditionsSummary, windowMinutes(window)).Scan( &temperature, &apparent, &dewpoint, &relativeHumidity, &windSpeed, &windDirection, &conditionCode, &isDay, ); err != nil { return ports.ObservationCurrentConditionsMetric{}, fmt.Errorf("query current conditions summary: %w", err) } return ports.ObservationCurrentConditionsMetric{ TemperatureC: ptrFloat64(temperature), ApparentTemperatureC: ptrFloat64(apparent), DewpointC: ptrFloat64(dewpoint), RelativeHumidity: ptrFloat64(relativeHumidity), WindSpeedKmh: ptrFloat64(windSpeed), WindDirectionDegrees: ptrFloat64(windDirection), ConditionCode: ptrInt(conditionCode), IsDay: ptrBool(isDay), }, nil } func (r *ObservationRepository) ListRecentObservations(ctx context.Context, count int) ([]ports.ObservationRecordMetric, error) { rows, err := r.pool.Query(ctx, queryRecentObservations, count) if err != nil { return nil, fmt.Errorf("query recent observations: %w", err) } defer rows.Close() out := make([]ports.ObservationRecordMetric, 0, count) eventIDs := make([]string, 0, count) for rows.Next() { var eventID string var stationID sql.NullString var stationName sql.NullString var timestamp time.Time var conditionCode int var isDay sql.NullBool var textDescription sql.NullString var temperature sql.NullFloat64 var dewpoint sql.NullFloat64 var windDirection sql.NullFloat64 var windSpeed sql.NullFloat64 var windGust sql.NullFloat64 var pressure sql.NullFloat64 var visibility sql.NullFloat64 var relativeHumidity sql.NullFloat64 var apparent sql.NullFloat64 if err := rows.Scan( &eventID, &stationID, &stationName, ×tamp, &conditionCode, &isDay, &textDescription, &temperature, &dewpoint, &windDirection, &windSpeed, &windGust, &pressure, &visibility, &relativeHumidity, &apparent, ); err != nil { return nil, fmt.Errorf("scan recent observation row: %w", err) } eventIDs = append(eventIDs, eventID) out = append(out, ports.ObservationRecordMetric{ EventID: eventID, StationID: ptrString(stationID), StationName: ptrString(stationName), Timestamp: timestamp, ConditionCode: conditionCode, IsDay: ptrBool(isDay), TextDescription: ptrString(textDescription), TemperatureC: ptrFloat64(temperature), DewpointC: ptrFloat64(dewpoint), WindDirectionDegrees: ptrFloat64(windDirection), WindSpeedKmh: ptrFloat64(windSpeed), WindGustKmh: ptrFloat64(windGust), BarometricPressurePa: ptrFloat64(pressure), VisibilityMeters: ptrFloat64(visibility), RelativeHumidityPercent: ptrFloat64(relativeHumidity), ApparentTemperatureC: ptrFloat64(apparent), }) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate recent observations rows: %w", err) } if len(out) == 0 { return out, nil } presentWeatherByEventID, err := r.listObservationPresentWeather(ctx, eventIDs) if err != nil { return nil, err } for i := range out { out[i].PresentWeather = presentWeatherByEventID[out[i].EventID] } return out, nil } func (r *ObservationRepository) listObservationPresentWeather(ctx context.Context, eventIDs []string) (map[string][]ports.ObservationPresentWeatherMetric, error) { rows, err := r.pool.Query(ctx, queryObservationPresentWeather, eventIDs) if err != nil { return nil, fmt.Errorf("query observation present weather: %w", err) } defer rows.Close() out := make(map[string][]ports.ObservationPresentWeatherMetric, len(eventIDs)) for rows.Next() { var eventID string var weatherIndex int var rawText sql.NullString if err := rows.Scan(&eventID, &weatherIndex, &rawText); err != nil { return nil, fmt.Errorf("scan observation present weather row: %w", err) } var raw map[string]any if rawText.Valid { if err := json.Unmarshal([]byte(rawText.String), &raw); err != nil { return nil, fmt.Errorf("decode observation present weather raw payload: %w", err) } } out[eventID] = append(out[eventID], ports.ObservationPresentWeatherMetric{ Raw: raw, }) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate observation present weather rows: %w", err) } return out, nil } func windowMinutes(window time.Duration) int { return int(window / time.Minute) }