265 lines
12 KiB
Go
265 lines
12 KiB
Go
package postgres
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
|
|
"gitea.maximumdirect.net/ejr/feedkit/config"
|
|
fksinks "gitea.maximumdirect.net/ejr/feedkit/sinks"
|
|
)
|
|
|
|
const (
|
|
tableObservations = "observations"
|
|
tableObservationCloudLayers = "observation_cloud_layers"
|
|
tableObservationPresentWeather = "observation_present_weather"
|
|
tableForecasts = "forecasts"
|
|
tableForecastPeriods = "forecast_periods"
|
|
tableAlertRuns = "alert_runs"
|
|
tableAlerts = "alerts"
|
|
tableAlertReferences = "alert_references"
|
|
)
|
|
|
|
// RegisterPostgresSchemas registers weatherfeeder's Postgres schema for each
|
|
// configured sink using driver=postgres.
|
|
func RegisterPostgresSchemas(cfg *config.Config) error {
|
|
if cfg == nil {
|
|
return fmt.Errorf("register postgres schemas: config is nil")
|
|
}
|
|
|
|
schema := weatherPostgresSchema()
|
|
for i, sk := range cfg.Sinks {
|
|
if !isPostgresDriver(sk.Driver) {
|
|
continue
|
|
}
|
|
if err := fksinks.RegisterPostgresSchema(sk.Name, schema); err != nil {
|
|
return fmt.Errorf("register postgres schema for sinks[%d] name=%q: %w", i, sk.Name, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func isPostgresDriver(driver string) bool {
|
|
return strings.EqualFold(strings.TrimSpace(driver), "postgres")
|
|
}
|
|
|
|
func weatherPostgresSchema() fksinks.PostgresSchema {
|
|
return fksinks.PostgresSchema{
|
|
Tables: []fksinks.PostgresTable{
|
|
{
|
|
Name: tableObservations,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "event_id", Type: "TEXT", Nullable: false},
|
|
{Name: "event_kind", Type: "TEXT", Nullable: false},
|
|
{Name: "event_source", Type: "TEXT", Nullable: false},
|
|
{Name: "event_schema", Type: "TEXT", Nullable: false},
|
|
{Name: "event_emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "event_effective_at", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "station_id", Type: "TEXT", Nullable: true},
|
|
{Name: "station_name", Type: "TEXT", Nullable: true},
|
|
{Name: "observed_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "condition_code", Type: "INTEGER", Nullable: false},
|
|
{Name: "condition_text", Type: "TEXT", Nullable: true},
|
|
{Name: "is_day", Type: "BOOLEAN", Nullable: true},
|
|
{Name: "provider_raw_description", Type: "TEXT", Nullable: true},
|
|
{Name: "text_description", Type: "TEXT", Nullable: true},
|
|
{Name: "icon_url", Type: "TEXT", Nullable: true},
|
|
{Name: "temperature_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "dewpoint_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_direction_degrees", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_speed_kmh", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_gust_kmh", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "barometric_pressure_pa", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "sea_level_pressure_pa", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "visibility_meters", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "relative_humidity_percent", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "apparent_temperature_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "elevation_meters", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "raw_message", Type: "TEXT", Nullable: true},
|
|
},
|
|
PrimaryKey: []string{"event_id"},
|
|
PruneColumn: "observed_at",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_obs_station_observed_at", Columns: []string{"station_id", "observed_at"}},
|
|
{Name: "idx_wf_obs_observed_at", Columns: []string{"observed_at"}},
|
|
{Name: "idx_wf_obs_condition_code", Columns: []string{"condition_code"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableObservationCloudLayers,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "event_id", Type: "TEXT REFERENCES observations(event_id) ON DELETE CASCADE", Nullable: false},
|
|
{Name: "layer_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "observed_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "base_meters", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "amount", Type: "TEXT", Nullable: true},
|
|
},
|
|
PrimaryKey: []string{"event_id", "layer_index"},
|
|
PruneColumn: "observed_at",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_obs_cloud_observed_at", Columns: []string{"observed_at"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableObservationPresentWeather,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "event_id", Type: "TEXT REFERENCES observations(event_id) ON DELETE CASCADE", Nullable: false},
|
|
{Name: "weather_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "observed_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "raw_text", Type: "TEXT", Nullable: true},
|
|
},
|
|
PrimaryKey: []string{"event_id", "weather_index"},
|
|
PruneColumn: "observed_at",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_obs_present_observed_at", Columns: []string{"observed_at"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableForecasts,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "event_id", Type: "TEXT", Nullable: false},
|
|
{Name: "event_kind", Type: "TEXT", Nullable: false},
|
|
{Name: "event_source", Type: "TEXT", Nullable: false},
|
|
{Name: "event_schema", Type: "TEXT", Nullable: false},
|
|
{Name: "event_emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "event_effective_at", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "location_id", Type: "TEXT", Nullable: true},
|
|
{Name: "location_name", Type: "TEXT", Nullable: true},
|
|
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "updated_at", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "product", Type: "TEXT", Nullable: false},
|
|
{Name: "latitude", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "longitude", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "elevation_meters", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "period_count", Type: "INTEGER", Nullable: false},
|
|
},
|
|
PrimaryKey: []string{"event_id"},
|
|
PruneColumn: "issued_at",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_fc_location_product_issued_at", Columns: []string{"location_id", "product", "issued_at"}},
|
|
{Name: "idx_wf_fc_issued_at", Columns: []string{"issued_at"}},
|
|
{Name: "idx_wf_fc_product_issued_at", Columns: []string{"product", "issued_at"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableForecastPeriods,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "run_event_id", Type: "TEXT REFERENCES forecasts(event_id) ON DELETE CASCADE", Nullable: false},
|
|
{Name: "period_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "start_time", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "end_time", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "name", Type: "TEXT", Nullable: true},
|
|
{Name: "is_day", Type: "BOOLEAN", Nullable: true},
|
|
{Name: "condition_code", Type: "INTEGER", Nullable: false},
|
|
{Name: "condition_text", Type: "TEXT", Nullable: true},
|
|
{Name: "provider_raw_description", Type: "TEXT", Nullable: true},
|
|
{Name: "text_description", Type: "TEXT", Nullable: true},
|
|
{Name: "detailed_text", Type: "TEXT", Nullable: true},
|
|
{Name: "icon_url", Type: "TEXT", Nullable: true},
|
|
{Name: "temperature_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "temperature_c_min", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "temperature_c_max", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "dewpoint_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "relative_humidity_percent", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_direction_degrees", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_speed_kmh", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "wind_gust_kmh", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "barometric_pressure_pa", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "visibility_meters", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "apparent_temperature_c", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "cloud_cover_percent", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "probability_of_precipitation_percent", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "precipitation_amount_mm", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "snowfall_depth_mm", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "uv_index", Type: "DOUBLE PRECISION", Nullable: true},
|
|
},
|
|
PrimaryKey: []string{"run_event_id", "period_index"},
|
|
PruneColumn: "issued_at",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_fc_period_start_time", Columns: []string{"start_time"}},
|
|
{Name: "idx_wf_fc_period_end_time", Columns: []string{"end_time"}},
|
|
{Name: "idx_wf_fc_period_run_start", Columns: []string{"run_event_id", "start_time"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableAlertRuns,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "event_id", Type: "TEXT", Nullable: false},
|
|
{Name: "event_kind", Type: "TEXT", Nullable: false},
|
|
{Name: "event_source", Type: "TEXT", Nullable: false},
|
|
{Name: "event_schema", Type: "TEXT", Nullable: false},
|
|
{Name: "event_emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "event_effective_at", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "location_id", Type: "TEXT", Nullable: true},
|
|
{Name: "location_name", Type: "TEXT", Nullable: true},
|
|
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "latitude", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "longitude", Type: "DOUBLE PRECISION", Nullable: true},
|
|
{Name: "alert_count", Type: "INTEGER", Nullable: false},
|
|
},
|
|
PrimaryKey: []string{"event_id"},
|
|
PruneColumn: "as_of",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_alert_run_location_as_of", Columns: []string{"location_id", "as_of"}},
|
|
{Name: "idx_wf_alert_run_as_of", Columns: []string{"as_of"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableAlerts,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "run_event_id", Type: "TEXT REFERENCES alert_runs(event_id) ON DELETE CASCADE", Nullable: false},
|
|
{Name: "alert_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "alert_id", Type: "TEXT", Nullable: false},
|
|
{Name: "event", Type: "TEXT", Nullable: true},
|
|
{Name: "headline", Type: "TEXT", Nullable: true},
|
|
{Name: "severity", Type: "TEXT", Nullable: true},
|
|
{Name: "urgency", Type: "TEXT", Nullable: true},
|
|
{Name: "certainty", Type: "TEXT", Nullable: true},
|
|
{Name: "status", Type: "TEXT", Nullable: true},
|
|
{Name: "message_type", Type: "TEXT", Nullable: true},
|
|
{Name: "category", Type: "TEXT", Nullable: true},
|
|
{Name: "response", Type: "TEXT", Nullable: true},
|
|
{Name: "description", Type: "TEXT", Nullable: true},
|
|
{Name: "instruction", Type: "TEXT", Nullable: true},
|
|
{Name: "sent", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "effective", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "onset", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "expires", Type: "TIMESTAMPTZ", Nullable: true},
|
|
{Name: "area_description", Type: "TEXT", Nullable: true},
|
|
{Name: "sender_name", Type: "TEXT", Nullable: true},
|
|
{Name: "reference_count", Type: "INTEGER", Nullable: false},
|
|
},
|
|
PrimaryKey: []string{"run_event_id", "alert_index"},
|
|
PruneColumn: "as_of",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_alerts_alert_id", Columns: []string{"alert_id"}},
|
|
{Name: "idx_wf_alerts_severity_expires", Columns: []string{"severity", "expires"}},
|
|
{Name: "idx_wf_alerts_as_of", Columns: []string{"as_of"}},
|
|
},
|
|
},
|
|
{
|
|
Name: tableAlertReferences,
|
|
Columns: []fksinks.PostgresColumn{
|
|
{Name: "run_event_id", Type: "TEXT REFERENCES alert_runs(event_id) ON DELETE CASCADE", Nullable: false},
|
|
{Name: "alert_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "reference_index", Type: "INTEGER", Nullable: false},
|
|
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
|
|
{Name: "id", Type: "TEXT", Nullable: true},
|
|
{Name: "identifier", Type: "TEXT", Nullable: true},
|
|
{Name: "sender", Type: "TEXT", Nullable: true},
|
|
{Name: "sent", Type: "TIMESTAMPTZ", Nullable: true},
|
|
},
|
|
PrimaryKey: []string{"run_event_id", "alert_index", "reference_index"},
|
|
PruneColumn: "as_of",
|
|
Indexes: []fksinks.PostgresIndex{
|
|
{Name: "idx_wf_alert_refs_as_of", Columns: []string{"as_of"}},
|
|
{Name: "idx_wf_alert_refs_sent", Columns: []string{"sent"}},
|
|
},
|
|
},
|
|
},
|
|
MapEvent: mapPostgresEvent,
|
|
}
|
|
}
|