weatherapi/internal/adapters/postgres/observations_repo.go

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,
			&timestamp,
			&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)
}