Split current conditions condition code querying

internal/adapters/outbound/postgres/conditions_mapper.go

@@ -7,16 +7,11 @@ import (
 	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
 )
 
-func mapCurrentConditionsRow(row currentConditionsRow) *app.CurrentConditions {
+func mapCurrentConditionsRow(row currentConditionsRow, conditionCode model.WMOCode) *app.CurrentConditions {
 	if row.SampleCount == 0 {
 		return nil
 	}
 
-	conditionCode := model.WMOUnknown
-	if row.ConditionCode.Valid {
-		conditionCode = model.WMOCode(row.ConditionCode.Int64)
-	}
-
 	return &app.CurrentConditions{
 		TemperatureC:            float64Ptr(row.TemperatureC),
 		ApparentTemperatureC:    float64Ptr(row.ApparentTemperatureC),

internal/adapters/outbound/postgres/conditions_queries.go

@@ -12,7 +12,6 @@ WITH windowed AS (
     relative_humidity_percent,
     wind_speed_kmh,
     wind_direction_degrees,
-    condition_code,
     is_day,
     observed_at
   FROM observations
@@ -39,7 +38,6 @@ SELECT
       AVG(cosd(wind_direction_degrees))
     )
   END AS wind_direction_degrees,
-  MAX(condition_code) AS condition_code,
   (
     SELECT is_day
     FROM windowed
@@ -47,4 +45,23 @@ SELECT
     LIMIT 1
   ) AS is_day
 FROM windowed`
+
+	queryCurrentConditionsConditionCodeCandidates = `
+WITH ranked AS (
+  SELECT
+    event_source,
+    condition_code,
+    ROW_NUMBER() OVER (
+      PARTITION BY event_source
+      ORDER BY observed_at DESC, event_emitted_at DESC
+    ) AS source_rank
+  FROM observations
+  WHERE observed_at > CURRENT_TIMESTAMP - make_interval(mins => $1)
+)
+SELECT
+  event_source,
+  condition_code
+FROM ranked
+WHERE source_rank = 1
+ORDER BY event_source`
 )

internal/adapters/outbound/postgres/conditions_read.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 
 	"gitea.maximumdirect.net/ejr/weatherapi/internal/app"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
 )
 
 func (r *Repository) CurrentConditions(ctx context.Context, observationWindowMinutes int) (*app.CurrentConditions, error) {
@@ -25,7 +26,6 @@ func (r *Repository) CurrentConditions(ctx context.Context, observationWindowMin
 		&row.RelativeHumidityPercent,
 		&row.WindSpeedKmh,
 		&row.WindDirectionDegrees,
-		&row.ConditionCode,
 		&row.IsDay,
 	)
 	if errors.Is(err, sql.ErrNoRows) {
@@ -35,5 +35,42 @@ func (r *Repository) CurrentConditions(ctx context.Context, observationWindowMin
 		return nil, fmt.Errorf("query current conditions: %w", err)
 	}
 
-	return mapCurrentConditionsRow(row), nil
+	if row.SampleCount == 0 {
+		return nil, nil
+	}
+
+	candidates, err := r.currentConditionsConditionCodeCandidates(ctx, observationWindowMinutes)
+	if err != nil {
+		return nil, err
+	}
+
+	return mapCurrentConditionsRow(row, selectCurrentConditionsConditionCode(candidates)), nil
+}
+
+func (r *Repository) currentConditionsConditionCodeCandidates(ctx context.Context, observationWindowMinutes int) ([]currentConditionsConditionCodeCandidate, error) {
+	rows, err := r.db.QueryContext(ctx, queryCurrentConditionsConditionCodeCandidates, observationWindowMinutes)
+	if err != nil {
+		return nil, fmt.Errorf("query current conditions condition code candidates: %w", err)
+	}
+	defer rows.Close()
+
+	var candidates []currentConditionsConditionCodeCandidate
+	for rows.Next() {
+		var (
+			eventSource   string
+			conditionCode int64
+		)
+		if err := rows.Scan(&eventSource, &conditionCode); err != nil {
+			return nil, fmt.Errorf("scan current conditions condition code candidate: %w", err)
+		}
+		candidates = append(candidates, currentConditionsConditionCodeCandidate{
+			EventSource:   eventSource,
+			ConditionCode: model.WMOCode(conditionCode),
+		})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate current conditions condition code candidates: %w", err)
+	}
+
+	return candidates, nil
 }

internal/adapters/outbound/postgres/conditions_read_test.go

@@ -0,0 +1,280 @@
+// conditions_read_test.go validates current-conditions repository read flow.
+// Layer: adapters/outbound/postgres conditions read tests.
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+)
+
+const currentConditionsTestDriverName = "weatherapi_current_conditions_read_test"
+
+func init() {
+	sql.Register(currentConditionsTestDriverName, currentConditionsTestDriver{})
+}
+
+func TestCurrentConditionsUsesConsensusConditionCode(t *testing.T) {
+	repo, closeDB := openCurrentConditionsTestRepository(t,
+		currentConditionsAggregateQuery(currentConditionsAggregateRow(3), nil),
+		currentConditionsConditionCodeCandidatesQuery([][]driver.Value{
+			{"source-a", int64(1)},
+			{"source-b", int64(2)},
+			{"source-c", int64(95)},
+		}, nil),
+	)
+	defer closeDB()
+
+	conditions, err := repo.CurrentConditions(context.Background(), 15)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if conditions == nil {
+		t.Fatal("expected current conditions")
+	}
+	if conditions.ConditionCode != 1 {
+		t.Fatalf("expected consensus condition code 1, got %d", conditions.ConditionCode)
+	}
+	if conditions.TemperatureC == nil || *conditions.TemperatureC != 15.5 {
+		t.Fatalf("expected temperature 15.5, got %v", conditions.TemperatureC)
+	}
+	if conditions.WindDirectionDegrees == nil || *conditions.WindDirectionDegrees != 182.5 {
+		t.Fatalf("expected wind direction 182.5, got %v", conditions.WindDirectionDegrees)
+	}
+	if conditions.IsDay == nil || !*conditions.IsDay {
+		t.Fatalf("expected isDay true, got %v", conditions.IsDay)
+	}
+	assertCurrentConditionsTestQueriesConsumed(t)
+}
+
+func TestCurrentConditionsNoSamplesSkipsConditionCodeCandidates(t *testing.T) {
+	repo, closeDB := openCurrentConditionsTestRepository(t,
+		currentConditionsAggregateQuery([]driver.Value{int64(0), nil, nil, nil, nil, nil, nil, nil}, nil),
+	)
+	defer closeDB()
+
+	conditions, err := repo.CurrentConditions(context.Background(), 15)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if conditions != nil {
+		t.Fatalf("expected nil current conditions, got %+v", conditions)
+	}
+	assertCurrentConditionsTestQueriesConsumed(t)
+}
+
+func TestCurrentConditionsConditionCodeCandidateQueryUsesLatestPerSourceOrdering(t *testing.T) {
+	query := compactSQL(queryCurrentConditionsConditionCodeCandidates)
+	want := "PARTITION BY event_source ORDER BY observed_at DESC, event_emitted_at DESC"
+	if !strings.Contains(query, want) {
+		t.Fatalf("expected condition code candidate query to contain %q, got %q", want, query)
+	}
+	if !strings.Contains(query, "SELECT event_source, condition_code") {
+		t.Fatalf("expected condition code candidate query to select event_source and condition_code, got %q", query)
+	}
+}
+
+func TestCurrentConditionsAggregateQueryDoesNotSelectConditionCode(t *testing.T) {
+	query := compactSQL(queryCurrentConditions)
+	if strings.Contains(query, "condition_code") {
+		t.Fatalf("expected aggregate query not to select condition_code, got %q", query)
+	}
+}
+
+func TestCurrentConditionsConditionCodeCandidateQueryWrapsErrors(t *testing.T) {
+	repo, closeDB := openCurrentConditionsTestRepository(t,
+		currentConditionsAggregateQuery(currentConditionsAggregateRow(1), nil),
+		scriptedCurrentConditionsQuery{
+			name:  "condition code candidates",
+			query: queryCurrentConditionsConditionCodeCandidates,
+			args:  []driver.Value{int64(15)},
+			err:   errors.New("candidate query unavailable"),
+		},
+	)
+	defer closeDB()
+
+	_, err := repo.CurrentConditions(context.Background(), 15)
+	if err == nil {
+		t.Fatal("expected error")
+	}
+	if !strings.Contains(err.Error(), "query current conditions condition code candidates") {
+		t.Fatalf("expected candidate query context, got %v", err)
+	}
+}
+
+func openCurrentConditionsTestRepository(t *testing.T, queries ...scriptedCurrentConditionsQuery) (*Repository, func()) {
+	t.Helper()
+	currentConditionsTestScript.set(queries)
+
+	db, err := sql.Open(currentConditionsTestDriverName, "")
+	if err != nil {
+		t.Fatalf("open test db: %v", err)
+	}
+	db.SetMaxOpenConns(1)
+
+	return NewRepository(db), func() {
+		_ = db.Close()
+		currentConditionsTestScript.set(nil)
+	}
+}
+
+func assertCurrentConditionsTestQueriesConsumed(t *testing.T) {
+	t.Helper()
+	if remaining := currentConditionsTestScript.remaining(); remaining != 0 {
+		t.Fatalf("expected all scripted queries consumed, got %d remaining", remaining)
+	}
+}
+
+func currentConditionsAggregateQuery(row []driver.Value, nextErr error) scriptedCurrentConditionsQuery {
+	return scriptedCurrentConditionsQuery{
+		name:    "aggregate",
+		query:   queryCurrentConditions,
+		args:    []driver.Value{int64(15)},
+		columns: []string{"sample_count", "temperature_c", "apparent_temperature_c", "dewpoint_c", "relative_humidity_percent", "wind_speed_kmh", "wind_direction_degrees", "is_day"},
+		rows:    [][]driver.Value{row},
+		nextErr: nextErr,
+	}
+}
+
+func currentConditionsConditionCodeCandidatesQuery(rows [][]driver.Value, nextErr error) scriptedCurrentConditionsQuery {
+	return scriptedCurrentConditionsQuery{
+		name:    "condition code candidates",
+		query:   queryCurrentConditionsConditionCodeCandidates,
+		args:    []driver.Value{int64(15)},
+		columns: []string{"event_source", "condition_code"},
+		rows:    rows,
+		nextErr: nextErr,
+	}
+}
+
+func currentConditionsAggregateRow(sampleCount int64) []driver.Value {
+	return []driver.Value{
+		sampleCount,
+		float64(15.5),
+		float64(14.2),
+		float64(10.1),
+		float64(72),
+		float64(24.8),
+		float64(182.5),
+		true,
+	}
+}
+
+type currentConditionsTestDriver struct{}
+
+func (currentConditionsTestDriver) Open(string) (driver.Conn, error) {
+	return currentConditionsTestConn{}, nil
+}
+
+type currentConditionsTestConn struct{}
+
+func (currentConditionsTestConn) Prepare(string) (driver.Stmt, error) {
+	return nil, errors.New("prepare is not supported")
+}
+
+func (currentConditionsTestConn) Close() error {
+	return nil
+}
+
+func (currentConditionsTestConn) Begin() (driver.Tx, error) {
+	return nil, errors.New("transactions are not supported")
+}
+
+func (currentConditionsTestConn) QueryContext(_ context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
+	return currentConditionsTestScript.next(query, args)
+}
+
+type scriptedCurrentConditionsQuery struct {
+	name    string
+	query   string
+	args    []driver.Value
+	columns []string
+	rows    [][]driver.Value
+	err     error
+	nextErr error
+}
+
+type currentConditionsTestScriptState struct {
+	mu      sync.Mutex
+	queries []scriptedCurrentConditionsQuery
+}
+
+var currentConditionsTestScript currentConditionsTestScriptState
+
+func (s *currentConditionsTestScriptState) set(queries []scriptedCurrentConditionsQuery) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.queries = append([]scriptedCurrentConditionsQuery(nil), queries...)
+}
+
+func (s *currentConditionsTestScriptState) remaining() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.queries)
+}
+
+func (s *currentConditionsTestScriptState) next(query string, args []driver.NamedValue) (driver.Rows, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if len(s.queries) == 0 {
+		return nil, fmt.Errorf("unexpected query: %s", compactSQL(query))
+	}
+	next := s.queries[0]
+	s.queries = s.queries[1:]
+
+	if compactSQL(query) != compactSQL(next.query) {
+		return nil, fmt.Errorf("expected %s query %q, got %q", next.name, compactSQL(next.query), compactSQL(query))
+	}
+	if len(args) != len(next.args) {
+		return nil, fmt.Errorf("expected %s args %v, got %v", next.name, next.args, namedValues(args))
+	}
+	for i, arg := range args {
+		if arg.Value != next.args[i] {
+			return nil, fmt.Errorf("expected %s arg %d to be %v, got %v", next.name, i, next.args[i], arg.Value)
+		}
+	}
+	if next.err != nil {
+		return nil, next.err
+	}
+	return &currentConditionsTestRows{
+		columns: append([]string(nil), next.columns...),
+		rows:    append([][]driver.Value(nil), next.rows...),
+		nextErr: next.nextErr,
+	}, nil
+}
+
+type currentConditionsTestRows struct {
+	columns []string
+	rows    [][]driver.Value
+	index   int
+	nextErr error
+}
+
+func (r *currentConditionsTestRows) Columns() []string {
+	return r.columns
+}
+
+func (r *currentConditionsTestRows) Close() error {
+	return nil
+}
+
+func (r *currentConditionsTestRows) Next(dest []driver.Value) error {
+	if r.index >= len(r.rows) {
+		if r.nextErr != nil {
+			err := r.nextErr
+			r.nextErr = nil
+			return err
+		}
+		return io.EOF
+	}
+	copy(dest, r.rows[r.index])
+	r.index++
+	return nil
+}

internal/adapters/outbound/postgres/conditions_rows.go

@@ -12,6 +12,5 @@ type currentConditionsRow struct {
 	RelativeHumidityPercent sql.NullFloat64
 	WindSpeedKmh            sql.NullFloat64
 	WindDirectionDegrees    sql.NullFloat64
-	ConditionCode           sql.NullInt64
 	IsDay                   sql.NullBool
 }

internal/adapters/outbound/postgres/repository_test.go

@@ -279,7 +279,7 @@ func TestMapAlertRowNullableEnds(t *testing.T) {
 func TestMapCurrentConditionsRowNoSamplesReturnsNil(t *testing.T) {
 	got := mapCurrentConditionsRow(currentConditionsRow{
 		SampleCount: 0,
-	})
+	}, model.WMOUnknown)
 	if got != nil {
 		t.Fatalf("expected nil for empty sample window, got %+v", got)
 	}
@@ -295,9 +295,8 @@ func TestMapCurrentConditionsRowMapsFields(t *testing.T) {
 		RelativeHumidityPercent: sql.NullFloat64{Float64: 72, Valid: true},
 		WindSpeedKmh:            sql.NullFloat64{Float64: 24.8, Valid: true},
 		WindDirectionDegrees:    sql.NullFloat64{Float64: 182.5, Valid: true},
-		ConditionCode:           sql.NullInt64{Int64: 65, Valid: true},
 		IsDay:                   sql.NullBool{Bool: isDay, Valid: true},
-	})
+	}, 65)
 	if got == nil {
 		t.Fatalf("expected mapped current conditions")
 	}