Refactored the application structure to better separate concerns between files and packages
Some checks failed
ci/woodpecker/push/build-image Pipeline failed
Some checks failed
ci/woodpecker/push/build-image Pipeline failed
This commit is contained in:
70
internal/adapters/outbound/postgres/alerts_mapper.go
Normal file
70
internal/adapters/outbound/postgres/alerts_mapper.go
Normal file
@@ -0,0 +1,70 @@
|
||||
// alerts_mapper.go maps alert rows into weather model payloads.
|
||||
// Layer: adapters/outbound/postgres alerts feature.
|
||||
package postgres
|
||||
|
||||
import "gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
|
||||
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),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
53
internal/adapters/outbound/postgres/alerts_queries.go
Normal file
53
internal/adapters/outbound/postgres/alerts_queries.go
Normal file
@@ -0,0 +1,53 @@
|
||||
// alerts_queries.go contains SQL text for alert-run reads.
|
||||
// Layer: adapters/outbound/postgres alerts feature.
|
||||
package postgres
|
||||
|
||||
const (
|
||||
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`
|
||||
)
|
||||
110
internal/adapters/outbound/postgres/alerts_read.go
Normal file
110
internal/adapters/outbound/postgres/alerts_read.go
Normal file
@@ -0,0 +1,110 @@
|
||||
// alerts_read.go executes alert-run, alerts, and reference queries.
|
||||
// Layer: adapters/outbound/postgres alerts feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
func (r *Repository) LatestAlertRun(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) 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
|
||||
}
|
||||
59
internal/adapters/outbound/postgres/alerts_rows.go
Normal file
59
internal/adapters/outbound/postgres/alerts_rows.go
Normal file
@@ -0,0 +1,59 @@
|
||||
// alerts_rows.go defines row DTOs for alert-run, alert, and reference reads.
|
||||
// Layer: adapters/outbound/postgres alerts feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
type alertRunParentRow struct {
|
||||
EventID string
|
||||
LocationID sql.NullString
|
||||
LocationName sql.NullString
|
||||
AsOf time.Time
|
||||
Latitude sql.NullFloat64
|
||||
Longitude sql.NullFloat64
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
30
internal/adapters/outbound/postgres/conditions_mapper.go
Normal file
30
internal/adapters/outbound/postgres/conditions_mapper.go
Normal file
@@ -0,0 +1,30 @@
|
||||
// conditions_mapper.go maps current-conditions DB rows into app models.
|
||||
// Layer: adapters/outbound/postgres conditions feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"gitea.maximumdirect.net/ejr/weatherapi/internal/app"
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
func mapCurrentConditionsRow(row currentConditionsRow) *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),
|
||||
DewpointC: float64Ptr(row.DewpointC),
|
||||
RelativeHumidityPercent: float64Ptr(row.RelativeHumidityPercent),
|
||||
WindSpeedKmh: float64Ptr(row.WindSpeedKmh),
|
||||
WindDirectionDegrees: float64Ptr(row.WindDirectionDegrees),
|
||||
ConditionCode: conditionCode,
|
||||
IsDay: boolPtr(row.IsDay),
|
||||
}
|
||||
}
|
||||
50
internal/adapters/outbound/postgres/conditions_queries.go
Normal file
50
internal/adapters/outbound/postgres/conditions_queries.go
Normal file
@@ -0,0 +1,50 @@
|
||||
// conditions_queries.go contains SQL text for current-conditions reads.
|
||||
// Layer: adapters/outbound/postgres conditions feature.
|
||||
package postgres
|
||||
|
||||
const (
|
||||
queryCurrentConditions = `
|
||||
WITH windowed AS (
|
||||
SELECT
|
||||
temperature_c,
|
||||
apparent_temperature_c,
|
||||
dewpoint_c,
|
||||
relative_humidity_percent,
|
||||
wind_speed_kmh,
|
||||
wind_direction_degrees,
|
||||
condition_code,
|
||||
is_day,
|
||||
observed_at
|
||||
FROM observations
|
||||
WHERE observed_at > CURRENT_TIMESTAMP - make_interval(mins => $1)
|
||||
)
|
||||
SELECT
|
||||
COUNT(*) AS sample_count,
|
||||
AVG(temperature_c) AS temperature_c,
|
||||
AVG(apparent_temperature_c) AS apparent_temperature_c,
|
||||
AVG(dewpoint_c) AS dewpoint_c,
|
||||
AVG(relative_humidity_percent) AS relative_humidity_percent,
|
||||
AVG(wind_speed_kmh) AS wind_speed_kmh,
|
||||
CASE
|
||||
WHEN atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
) < 0
|
||||
THEN atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
) + 360.0
|
||||
ELSE atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
)
|
||||
END AS wind_direction_degrees,
|
||||
MAX(condition_code) AS condition_code,
|
||||
(
|
||||
SELECT is_day
|
||||
FROM windowed
|
||||
ORDER BY observed_at DESC
|
||||
LIMIT 1
|
||||
) AS is_day
|
||||
FROM windowed`
|
||||
)
|
||||
39
internal/adapters/outbound/postgres/conditions_read.go
Normal file
39
internal/adapters/outbound/postgres/conditions_read.go
Normal file
@@ -0,0 +1,39 @@
|
||||
// conditions_read.go executes current-conditions queries.
|
||||
// Layer: adapters/outbound/postgres conditions feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherapi/internal/app"
|
||||
)
|
||||
|
||||
func (r *Repository) CurrentConditions(ctx context.Context, observationWindowMinutes int) (*app.CurrentConditions, error) {
|
||||
if r == nil || r.db == nil {
|
||||
return nil, fmt.Errorf("postgres repository is not configured")
|
||||
}
|
||||
|
||||
var row currentConditionsRow
|
||||
err := r.db.QueryRowContext(ctx, queryCurrentConditions, observationWindowMinutes).Scan(
|
||||
&row.SampleCount,
|
||||
&row.TemperatureC,
|
||||
&row.ApparentTemperatureC,
|
||||
&row.DewpointC,
|
||||
&row.RelativeHumidityPercent,
|
||||
&row.WindSpeedKmh,
|
||||
&row.WindDirectionDegrees,
|
||||
&row.ConditionCode,
|
||||
&row.IsDay,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query current conditions: %w", err)
|
||||
}
|
||||
|
||||
return mapCurrentConditionsRow(row), nil
|
||||
}
|
||||
17
internal/adapters/outbound/postgres/conditions_rows.go
Normal file
17
internal/adapters/outbound/postgres/conditions_rows.go
Normal file
@@ -0,0 +1,17 @@
|
||||
// conditions_rows.go defines row DTOs for current-conditions reads.
|
||||
// Layer: adapters/outbound/postgres conditions feature.
|
||||
package postgres
|
||||
|
||||
import "database/sql"
|
||||
|
||||
type currentConditionsRow struct {
|
||||
SampleCount int64
|
||||
TemperatureC sql.NullFloat64
|
||||
ApparentTemperatureC sql.NullFloat64
|
||||
DewpointC sql.NullFloat64
|
||||
RelativeHumidityPercent sql.NullFloat64
|
||||
WindSpeedKmh sql.NullFloat64
|
||||
WindDirectionDegrees sql.NullFloat64
|
||||
ConditionCode sql.NullInt64
|
||||
IsDay sql.NullBool
|
||||
}
|
||||
49
internal/adapters/outbound/postgres/forecast_mapper.go
Normal file
49
internal/adapters/outbound/postgres/forecast_mapper.go
Normal file
@@ -0,0 +1,49 @@
|
||||
// forecast_mapper.go maps forecast rows into weather model payloads.
|
||||
// Layer: adapters/outbound/postgres forecast feature.
|
||||
package postgres
|
||||
|
||||
import "gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
54
internal/adapters/outbound/postgres/forecast_queries.go
Normal file
54
internal/adapters/outbound/postgres/forecast_queries.go
Normal file
@@ -0,0 +1,54 @@
|
||||
// forecast_queries.go contains SQL text for hourly forecast reads.
|
||||
// Layer: adapters/outbound/postgres forecast feature.
|
||||
package postgres
|
||||
|
||||
const (
|
||||
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`
|
||||
)
|
||||
96
internal/adapters/outbound/postgres/forecast_read.go
Normal file
96
internal/adapters/outbound/postgres/forecast_read.go
Normal file
@@ -0,0 +1,96 @@
|
||||
// forecast_read.go executes hourly forecast and period queries.
|
||||
// Layer: adapters/outbound/postgres forecast feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
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) 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
|
||||
}
|
||||
50
internal/adapters/outbound/postgres/forecast_rows.go
Normal file
50
internal/adapters/outbound/postgres/forecast_rows.go
Normal file
@@ -0,0 +1,50 @@
|
||||
// forecast_rows.go defines row DTOs for forecast reads.
|
||||
// Layer: adapters/outbound/postgres forecast feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"time"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
42
internal/adapters/outbound/postgres/observations_mapper.go
Normal file
42
internal/adapters/outbound/postgres/observations_mapper.go
Normal file
@@ -0,0 +1,42 @@
|
||||
// observations_mapper.go maps observation rows into weather model payloads.
|
||||
// Layer: adapters/outbound/postgres observations feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
33
internal/adapters/outbound/postgres/observations_queries.go
Normal file
33
internal/adapters/outbound/postgres/observations_queries.go
Normal file
@@ -0,0 +1,33 @@
|
||||
// observations_queries.go contains SQL text for observation reads.
|
||||
// Layer: adapters/outbound/postgres observations feature.
|
||||
package postgres
|
||||
|
||||
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`
|
||||
)
|
||||
79
internal/adapters/outbound/postgres/observations_read.go
Normal file
79
internal/adapters/outbound/postgres/observations_read.go
Normal file
@@ -0,0 +1,79 @@
|
||||
// observations_read.go executes observation and present-weather queries.
|
||||
// Layer: adapters/outbound/postgres observations feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
|
||||
)
|
||||
|
||||
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) 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
|
||||
}
|
||||
32
internal/adapters/outbound/postgres/observations_rows.go
Normal file
32
internal/adapters/outbound/postgres/observations_rows.go
Normal file
@@ -0,0 +1,32 @@
|
||||
// observations_rows.go defines row DTOs for observation reads.
|
||||
// Layer: adapters/outbound/postgres observations feature.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"time"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
type observationPresentWeatherRow struct {
|
||||
WeatherIndex int
|
||||
RawText sql.NullString
|
||||
}
|
||||
@@ -1,188 +1,11 @@
|
||||
// repository.go defines the Postgres repository shell and constructor.
|
||||
// Layer: adapters/outbound/postgres repository root.
|
||||
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`
|
||||
|
||||
queryCurrentConditions = `
|
||||
WITH windowed AS (
|
||||
SELECT
|
||||
temperature_c,
|
||||
apparent_temperature_c,
|
||||
dewpoint_c,
|
||||
relative_humidity_percent,
|
||||
wind_speed_kmh,
|
||||
wind_direction_degrees,
|
||||
condition_code,
|
||||
is_day,
|
||||
observed_at
|
||||
FROM observations
|
||||
WHERE observed_at > CURRENT_TIMESTAMP - make_interval(mins => $1)
|
||||
)
|
||||
SELECT
|
||||
COUNT(*) AS sample_count,
|
||||
AVG(temperature_c) AS temperature_c,
|
||||
AVG(apparent_temperature_c) AS apparent_temperature_c,
|
||||
AVG(dewpoint_c) AS dewpoint_c,
|
||||
AVG(relative_humidity_percent) AS relative_humidity_percent,
|
||||
AVG(wind_speed_kmh) AS wind_speed_kmh,
|
||||
CASE
|
||||
WHEN atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
) < 0
|
||||
THEN atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
) + 360.0
|
||||
ELSE atan2d(
|
||||
AVG(sind(wind_direction_degrees)),
|
||||
AVG(cosd(wind_direction_degrees))
|
||||
)
|
||||
END AS wind_direction_degrees,
|
||||
MAX(condition_code) AS condition_code,
|
||||
(
|
||||
SELECT is_day
|
||||
FROM windowed
|
||||
ORDER BY observed_at DESC
|
||||
LIMIT 1
|
||||
) AS is_day
|
||||
FROM windowed`
|
||||
|
||||
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`
|
||||
"gitea.maximumdirect.net/ejr/weatherapi/internal/app"
|
||||
)
|
||||
|
||||
// Repository is a Postgres-backed implementation of weatherapi read ports.
|
||||
@@ -190,605 +13,8 @@ type Repository struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
var _ core.Repository = (*Repository)(nil)
|
||||
var _ app.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) CurrentConditions(ctx context.Context, observationWindowMinutes int) (*core.CurrentConditions, error) {
|
||||
if r == nil || r.db == nil {
|
||||
return nil, fmt.Errorf("postgres repository is not configured")
|
||||
}
|
||||
|
||||
var row currentConditionsRow
|
||||
err := r.db.QueryRowContext(ctx, queryCurrentConditions, observationWindowMinutes).Scan(
|
||||
&row.SampleCount,
|
||||
&row.TemperatureC,
|
||||
&row.ApparentTemperatureC,
|
||||
&row.DewpointC,
|
||||
&row.RelativeHumidityPercent,
|
||||
&row.WindSpeedKmh,
|
||||
&row.WindDirectionDegrees,
|
||||
&row.ConditionCode,
|
||||
&row.IsDay,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query current conditions: %w", err)
|
||||
}
|
||||
|
||||
return mapCurrentConditionsRow(row), 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
|
||||
}
|
||||
|
||||
type currentConditionsRow struct {
|
||||
SampleCount int64
|
||||
TemperatureC sql.NullFloat64
|
||||
ApparentTemperatureC sql.NullFloat64
|
||||
DewpointC sql.NullFloat64
|
||||
RelativeHumidityPercent sql.NullFloat64
|
||||
WindSpeedKmh sql.NullFloat64
|
||||
WindDirectionDegrees sql.NullFloat64
|
||||
ConditionCode sql.NullInt64
|
||||
IsDay sql.NullBool
|
||||
}
|
||||
|
||||
func mapCurrentConditionsRow(row currentConditionsRow) *core.CurrentConditions {
|
||||
if row.SampleCount == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
conditionCode := model.WMOUnknown
|
||||
if row.ConditionCode.Valid {
|
||||
conditionCode = model.WMOCode(row.ConditionCode.Int64)
|
||||
}
|
||||
|
||||
return &core.CurrentConditions{
|
||||
TemperatureC: float64Ptr(row.TemperatureC),
|
||||
ApparentTemperatureC: float64Ptr(row.ApparentTemperatureC),
|
||||
DewpointC: float64Ptr(row.DewpointC),
|
||||
RelativeHumidityPercent: float64Ptr(row.RelativeHumidityPercent),
|
||||
WindSpeedKmh: float64Ptr(row.WindSpeedKmh),
|
||||
WindDirectionDegrees: float64Ptr(row.WindDirectionDegrees),
|
||||
ConditionCode: conditionCode,
|
||||
IsDay: boolPtr(row.IsDay),
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
// repository_test.go validates Postgres row mapping and attachment helpers.
|
||||
// Layer: adapters/outbound/postgres mapper regression tests.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
|
||||
39
internal/adapters/outbound/postgres/scan_helpers.go
Normal file
39
internal/adapters/outbound/postgres/scan_helpers.go
Normal file
@@ -0,0 +1,39 @@
|
||||
// scan_helpers.go provides shared sql.Null* conversion helpers.
|
||||
// Layer: adapters/outbound/postgres helper utilities.
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"time"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user