ejr/weatherfeeder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved HTTP polling helpers upstream into feedkit, and updated to feedkit v0.8.0
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
Details

Browse Source
This commit is contained in:
Eric Rakestraw
2026-03-28 10:02:50 -05:00
parent de5add59fd
commit eb27486466
12 changed files with 171 additions and 239 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
internal/sources/nws/alerts.go
View File

@@ -9,6 +9,7 @@ import (
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/common"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
@@ -22,14 +23,14 @@ import (
// Output schema:
// - standards.SchemaRawNWSAlertsV1
type AlertsSource struct {
http *common.HTTPSource
http *fksources.HTTPSource
}
func NewAlertsSource(cfg config.SourceConfig) (*AlertsSource, error) {
const driver = "nws_alerts"
// NWS alerts responses are GeoJSON-ish; allow fallback to plain JSON as well.
hs, err := common.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
if err != nil {
return nil, err
}
@@ -43,10 +44,13 @@ func (s *AlertsSource) Name() string { return s.http.Name }
func (s *AlertsSource) Kind() event.Kind { return event.Kind("alert") }
func (s *AlertsSource) Poll(ctx context.Context) ([]event.Event, error) {
raw, meta, err := s.fetchRaw(ctx)
raw, meta, changed, err := s.fetchRaw(ctx)
if err != nil {
return nil, err
}
if !changed {
return nil, nil
}
// EffectiveAt policy for alerts:
// Prefer the collection-level "updated" timestamp (best dedupe signal).
@@ -97,16 +101,19 @@ type alertsMeta struct {
ParsedLatestFeatureTime time.Time `json:"-"`
}
func (s *AlertsSource) fetchRaw(ctx context.Context) (json.RawMessage, alertsMeta, error) {
raw, err := s.http.FetchJSON(ctx)
func (s *AlertsSource) fetchRaw(ctx context.Context) (json.RawMessage, alertsMeta, bool, error) {
raw, changed, err := s.http.FetchJSONIfChanged(ctx)
if err != nil {
return nil, alertsMeta{}, err
return nil, alertsMeta{}, false, err
}
if !changed {
return nil, alertsMeta{}, false, nil
}
var meta alertsMeta
if err := json.Unmarshal(raw, &meta); err != nil {
// If metadata decode fails, still return raw; Poll will fall back to Source:EmittedAt.
return raw, alertsMeta{}, nil
return raw, alertsMeta{}, true, nil
}
// Top-level updated (preferred).
@@ -143,5 +150,5 @@ func (s *AlertsSource) fetchRaw(ctx context.Context) (json.RawMessage, alertsMet
}
meta.ParsedLatestFeatureTime = latest
return raw, meta, nil
return raw, meta, true, nil
}

23
internal/sources/nws/forecast_hourly.go
View File

@@ -9,6 +9,7 @@ import (
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/common"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
@@ -22,14 +23,14 @@ import (
// Output schema (current implementation):
// - standards.SchemaRawNWSHourlyForecastV1
type HourlyForecastSource struct {
http *common.HTTPSource
http *fksources.HTTPSource
}
func NewHourlyForecastSource(cfg config.SourceConfig) (*HourlyForecastSource, error) {
const driver = "nws_forecast_hourly"
// NWS forecast endpoints are GeoJSON (and sometimes also advertise json-ld/json).
hs, err := common.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
if err != nil {
return nil, err
}
@@ -43,10 +44,13 @@ func (s *HourlyForecastSource) Name() string { return s.http.Name }
func (s *HourlyForecastSource) Kind() event.Kind { return event.Kind("forecast") }
func (s *HourlyForecastSource) Poll(ctx context.Context) ([]event.Event, error) {
raw, meta, err := s.fetchRaw(ctx)
raw, meta, changed, err := s.fetchRaw(ctx)
if err != nil {
return nil, err
}
if !changed {
return nil, nil
}
// EffectiveAt is optional; for forecasts it’s most naturally the run “issued” time.
// NWS gridpoint forecasts expose generatedAt (preferred) and updateTime/updated.
@@ -94,16 +98,19 @@ type hourlyForecastMeta struct {
ParsedUpdateTime time.Time `json:"-"`
}
func (s *HourlyForecastSource) fetchRaw(ctx context.Context) (json.RawMessage, hourlyForecastMeta, error) {
raw, err := s.http.FetchJSON(ctx)
func (s *HourlyForecastSource) fetchRaw(ctx context.Context) (json.RawMessage, hourlyForecastMeta, bool, error) {
raw, changed, err := s.http.FetchJSONIfChanged(ctx)
if err != nil {
return nil, hourlyForecastMeta{}, err
return nil, hourlyForecastMeta{}, false, err
}
if !changed {
return nil, hourlyForecastMeta{}, false, nil
}
var meta hourlyForecastMeta
if err := json.Unmarshal(raw, &meta); err != nil {
// If metadata decode fails, still return raw; Poll will fall back to Source:EmittedAt.
return raw, hourlyForecastMeta{}, nil
return raw, hourlyForecastMeta{}, true, nil
}
// generatedAt (preferred)
@@ -125,5 +132,5 @@ func (s *HourlyForecastSource) fetchRaw(ctx context.Context) (json.RawMessage, h
}
}
return raw, meta, nil
return raw, meta, true, nil
}

23
internal/sources/nws/forecast_narrative.go
View File

@@ -9,6 +9,7 @@ import (
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/common"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
@@ -22,14 +23,14 @@ import (
// Output schema:
// - standards.SchemaRawNWSNarrativeForecastV1
type NarrativeForecastSource struct {
http *common.HTTPSource
http *fksources.HTTPSource
}
func NewNarrativeForecastSource(cfg config.SourceConfig) (*NarrativeForecastSource, error) {
const driver = "nws_forecast_narrative"
// NWS forecast endpoints are GeoJSON (and sometimes also advertise json-ld/json).
hs, err := common.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
if err != nil {
return nil, err
}
@@ -43,10 +44,13 @@ func (s *NarrativeForecastSource) Name() string { return s.http.Name }
func (s *NarrativeForecastSource) Kind() event.Kind { return event.Kind("forecast") }
func (s *NarrativeForecastSource) Poll(ctx context.Context) ([]event.Event, error) {
raw, meta, err := s.fetchRaw(ctx)
raw, meta, changed, err := s.fetchRaw(ctx)
if err != nil {
return nil, err
}
if !changed {
return nil, nil
}
// EffectiveAt is optional; for forecasts it’s most naturally the run “issued” time.
// NWS gridpoint forecasts expose generatedAt (preferred) and updateTime/updated.
@@ -94,16 +98,19 @@ type narrativeForecastMeta struct {
ParsedUpdateTime time.Time `json:"-"`
}
func (s *NarrativeForecastSource) fetchRaw(ctx context.Context) (json.RawMessage, narrativeForecastMeta, error) {
raw, err := s.http.FetchJSON(ctx)
func (s *NarrativeForecastSource) fetchRaw(ctx context.Context) (json.RawMessage, narrativeForecastMeta, bool, error) {
raw, changed, err := s.http.FetchJSONIfChanged(ctx)
if err != nil {
return nil, narrativeForecastMeta{}, err
return nil, narrativeForecastMeta{}, false, err
}
if !changed {
return nil, narrativeForecastMeta{}, false, nil
}
var meta narrativeForecastMeta
if err := json.Unmarshal(raw, &meta); err != nil {
// If metadata decode fails, still return raw; Poll will fall back to Source:EmittedAt.
return raw, narrativeForecastMeta{}, nil
return raw, narrativeForecastMeta{}, true, nil
}
// generatedAt (preferred)
@@ -125,5 +132,5 @@ func (s *NarrativeForecastSource) fetchRaw(ctx context.Context) (json.RawMessage
}
}
return raw, meta, nil
return raw, meta, true, nil
}

23
internal/sources/nws/observation.go
View File

@@ -9,6 +9,7 @@ import (
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/common"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
@@ -16,13 +17,13 @@ import (
// ObservationSource polls an NWS station observation endpoint and emits a RAW observation Event.
type ObservationSource struct {
http *common.HTTPSource
http *fksources.HTTPSource
}
func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
const driver = "nws_observation"
hs, err := common.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
if err != nil {
return nil, err
}
@@ -35,10 +36,13 @@ func (s *ObservationSource) Name() string { return s.http.Name }
func (s *ObservationSource) Kind() event.Kind { return event.Kind("observation") }
func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
raw, meta, err := s.fetchRaw(ctx)
raw, meta, changed, err := s.fetchRaw(ctx)
if err != nil {
return nil, err
}
if !changed {
return nil, nil
}
// EffectiveAt is optional; for observations it’s naturally the observation timestamp.
var effectiveAt *time.Time
@@ -72,16 +76,19 @@ type observationMeta struct {
ParsedTimestamp time.Time `json:"-"`
}
func (s *ObservationSource) fetchRaw(ctx context.Context) (json.RawMessage, observationMeta, error) {
raw, err := s.http.FetchJSON(ctx)
func (s *ObservationSource) fetchRaw(ctx context.Context) (json.RawMessage, observationMeta, bool, error) {
raw, changed, err := s.http.FetchJSONIfChanged(ctx)
if err != nil {
return nil, observationMeta{}, err
return nil, observationMeta{}, false, err
}
if !changed {
return nil, observationMeta{}, false, nil
}
var meta observationMeta
if err := json.Unmarshal(raw, &meta); err != nil {
// If metadata decode fails, still return raw; envelope will fall back to Source:EffectiveAt.
return raw, observationMeta{}, nil
return raw, observationMeta{}, true, nil
}
tsStr := strings.TrimSpace(meta.Properties.Timestamp)
@@ -91,5 +98,5 @@ func (s *ObservationSource) fetchRaw(ctx context.Context) (json.RawMessage, obse
}
}
return raw, meta, nil
return raw, meta, true, nil
}

63
internal/sources/nws/observation_test.go Normal file
View File

@@ -0,0 +1,63 @@
package nws
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
)
func TestObservationSourcePollReturnsNoEventsOn304(t *testing.T) {
var call int
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
call++
switch call {
case 1:
w.Header().Set("ETag", `"obs-v1"`)
_, _ = w.Write([]byte(`{"id":"obs-1","properties":{"timestamp":"2026-03-28T12:00:00Z"}}`))
case 2:
if got := r.Header.Get("If-None-Match"); got != `"obs-v1"` {
t.Fatalf("second request If-None-Match = %q", got)
}
w.WriteHeader(http.StatusNotModified)
default:
t.Fatalf("unexpected call count %d", call)
}
}))
defer srv.Close()
src, err := NewObservationSource(config.SourceConfig{
Name: "NWSObservationTest",
Driver: "nws_observation",
Mode: config.SourceModePoll,
Params: map[string]any{
"url": srv.URL,
"user_agent": "test-agent",
},
})
if err != nil {
t.Fatalf("NewObservationSource() error = %v", err)
}
first, err := src.Poll(context.Background())
if err != nil {
t.Fatalf("first Poll() error = %v", err)
}
if len(first) != 1 {
t.Fatalf("first Poll() len = %d, want 1", len(first))
}
if first[0].Kind != event.Kind("observation") {
t.Fatalf("first Poll() kind = %q", first[0].Kind)
}
second, err := src.Poll(context.Background())
if err != nil {
t.Fatalf("second Poll() error = %v", err)
}
if len(second) != 0 {
t.Fatalf("second Poll() len = %d, want 0", len(second))
}
}