Implemented NWS weather stories support
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
This commit is contained in:
118
internal/sources/nws/weatherstories.go
Normal file
118
internal/sources/nws/weatherstories.go
Normal file
@@ -0,0 +1,118 @@
|
||||
package nws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"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/standards"
|
||||
)
|
||||
|
||||
// WeatherStoriesSource polls an NWS weatherstories endpoint and emits a RAW weather story Event.
|
||||
//
|
||||
// Output schema:
|
||||
// - standards.SchemaRawNWSWeatherStoriesV1
|
||||
type WeatherStoriesSource struct {
|
||||
http *fksources.HTTPSource
|
||||
}
|
||||
|
||||
func NewWeatherStoriesSource(cfg config.SourceConfig) (*WeatherStoriesSource, error) {
|
||||
const driver = "nws_weatherstories"
|
||||
|
||||
hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &WeatherStoriesSource{http: hs}, nil
|
||||
}
|
||||
|
||||
func (s *WeatherStoriesSource) Name() string { return s.http.Name }
|
||||
|
||||
func (s *WeatherStoriesSource) Kinds() []event.Kind {
|
||||
return []event.Kind{event.Kind("weather_story")}
|
||||
}
|
||||
|
||||
func (s *WeatherStoriesSource) Poll(ctx context.Context) ([]event.Event, error) {
|
||||
raw, meta, changed, err := s.fetchRaw(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !changed {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var effectiveAt *time.Time
|
||||
switch {
|
||||
case !meta.ParsedLatestUpdateTime.IsZero():
|
||||
t := meta.ParsedLatestUpdateTime.UTC()
|
||||
effectiveAt = &t
|
||||
case !meta.ParsedLatestStartTime.IsZero():
|
||||
t := meta.ParsedLatestStartTime.UTC()
|
||||
effectiveAt = &t
|
||||
}
|
||||
|
||||
emittedAt := time.Now().UTC()
|
||||
eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
|
||||
|
||||
return fksources.SingleEvent(
|
||||
event.Kind("weather_story"),
|
||||
s.http.Name,
|
||||
standards.SchemaRawNWSWeatherStoriesV1,
|
||||
eventID,
|
||||
emittedAt,
|
||||
effectiveAt,
|
||||
raw,
|
||||
)
|
||||
}
|
||||
|
||||
type weatherStoriesMeta struct {
|
||||
Stories []struct {
|
||||
StartTime string `json:"startTime"`
|
||||
UpdateTime string `json:"updateTime"`
|
||||
} `json:"stories"`
|
||||
|
||||
ParsedLatestUpdateTime time.Time `json:"-"`
|
||||
ParsedLatestStartTime time.Time `json:"-"`
|
||||
}
|
||||
|
||||
func (s *WeatherStoriesSource) fetchRaw(ctx context.Context) (json.RawMessage, weatherStoriesMeta, bool, error) {
|
||||
raw, changed, err := s.http.FetchJSONIfChanged(ctx)
|
||||
if err != nil {
|
||||
return nil, weatherStoriesMeta{}, false, err
|
||||
}
|
||||
if !changed {
|
||||
return nil, weatherStoriesMeta{}, false, nil
|
||||
}
|
||||
|
||||
var meta weatherStoriesMeta
|
||||
if err := json.Unmarshal(raw, &meta); err != nil {
|
||||
return raw, weatherStoriesMeta{}, true, nil
|
||||
}
|
||||
|
||||
for _, story := range meta.Stories {
|
||||
if ts := strings.TrimSpace(story.UpdateTime); ts != "" {
|
||||
if t, err := nwscommon.ParseTime(ts); err == nil {
|
||||
t = t.UTC()
|
||||
if meta.ParsedLatestUpdateTime.IsZero() || t.After(meta.ParsedLatestUpdateTime) {
|
||||
meta.ParsedLatestUpdateTime = t
|
||||
}
|
||||
}
|
||||
}
|
||||
if ts := strings.TrimSpace(story.StartTime); ts != "" {
|
||||
if t, err := nwscommon.ParseTime(ts); err == nil {
|
||||
t = t.UTC()
|
||||
if meta.ParsedLatestStartTime.IsZero() || t.After(meta.ParsedLatestStartTime) {
|
||||
meta.ParsedLatestStartTime = t
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return raw, meta, true, nil
|
||||
}
|
||||
158
internal/sources/nws/weatherstories_test.go
Normal file
158
internal/sources/nws/weatherstories_test.go
Normal file
@@ -0,0 +1,158 @@
|
||||
package nws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/ejr/feedkit/config"
|
||||
"gitea.maximumdirect.net/ejr/feedkit/event"
|
||||
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
|
||||
)
|
||||
|
||||
func TestWeatherStoriesSourcePollEmitsExpectedEventAndPrefersLatestUpdateTime(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{
|
||||
"stories": [
|
||||
{"startTime":"2026-05-30T08:46:00+00:00","updateTime":"2026-05-30T09:00:34+00:00"},
|
||||
{"startTime":"2026-05-30T10:00:00+00:00","updateTime":"2026-05-30T11:00:34+00:00"}
|
||||
]
|
||||
}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL))
|
||||
if err != nil {
|
||||
t.Fatalf("NewWeatherStoriesSource() error = %v", err)
|
||||
}
|
||||
if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("weather_story") {
|
||||
t.Fatalf("Kinds() = %#v, want [weather_story]", got)
|
||||
}
|
||||
|
||||
events, err := src.Poll(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Poll() error = %v", err)
|
||||
}
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("Poll() len = %d, want 1", len(events))
|
||||
}
|
||||
|
||||
got := events[0]
|
||||
if got.Kind != event.Kind("weather_story") {
|
||||
t.Fatalf("Kind = %q, want weather_story", got.Kind)
|
||||
}
|
||||
if got.Schema != standards.SchemaRawNWSWeatherStoriesV1 {
|
||||
t.Fatalf("Schema = %q, want %q", got.Schema, standards.SchemaRawNWSWeatherStoriesV1)
|
||||
}
|
||||
wantEffectiveAt := time.Date(2026, 5, 30, 11, 0, 34, 0, time.UTC)
|
||||
if got.EffectiveAt == nil || !got.EffectiveAt.Equal(wantEffectiveAt) {
|
||||
t.Fatalf("EffectiveAt = %v, want %s", got.EffectiveAt, wantEffectiveAt.Format(time.RFC3339))
|
||||
}
|
||||
if _, ok := got.Payload.(json.RawMessage); !ok {
|
||||
t.Fatalf("Payload type = %T, want json.RawMessage", got.Payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeatherStoriesSourcePollEffectiveAtFallsBackToLatestStartTime(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{
|
||||
"stories": [
|
||||
{"startTime":"2026-05-30T08:46:00+00:00","updateTime":"bad"},
|
||||
{"startTime":"2026-05-31T11:00:00+00:00","updateTime":""}
|
||||
]
|
||||
}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL))
|
||||
if err != nil {
|
||||
t.Fatalf("NewWeatherStoriesSource() error = %v", err)
|
||||
}
|
||||
|
||||
events, err := src.Poll(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Poll() error = %v", err)
|
||||
}
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("Poll() len = %d, want 1", len(events))
|
||||
}
|
||||
wantEffectiveAt := time.Date(2026, 5, 31, 11, 0, 0, 0, time.UTC)
|
||||
if events[0].EffectiveAt == nil || !events[0].EffectiveAt.Equal(wantEffectiveAt) {
|
||||
t.Fatalf("EffectiveAt = %v, want %s", events[0].EffectiveAt, wantEffectiveAt.Format(time.RFC3339))
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeatherStoriesSourcePollReturnsNoEventsWhenUnchanged(t *testing.T) {
|
||||
const etag = `"stories-v1"`
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("If-None-Match") == etag {
|
||||
w.WriteHeader(http.StatusNotModified)
|
||||
return
|
||||
}
|
||||
w.Header().Set("ETag", etag)
|
||||
_, _ = w.Write([]byte(`{"stories":[]}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL))
|
||||
if err != nil {
|
||||
t.Fatalf("NewWeatherStoriesSource() 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))
|
||||
}
|
||||
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeatherStoriesSourcePollMetadataDecodeFailureStillEmitsRawEvent(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`not-json`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL))
|
||||
if err != nil {
|
||||
t.Fatalf("NewWeatherStoriesSource() error = %v", err)
|
||||
}
|
||||
|
||||
events, err := src.Poll(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Poll() error = %v", err)
|
||||
}
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("Poll() len = %d, want 1", len(events))
|
||||
}
|
||||
if events[0].EffectiveAt != nil {
|
||||
t.Fatalf("EffectiveAt = %v, want nil", events[0].EffectiveAt)
|
||||
}
|
||||
if events[0].Schema != standards.SchemaRawNWSWeatherStoriesV1 {
|
||||
t.Fatalf("Schema = %q, want %q", events[0].Schema, standards.SchemaRawNWSWeatherStoriesV1)
|
||||
}
|
||||
}
|
||||
|
||||
func weatherStoriesSourceConfig(url string) config.SourceConfig {
|
||||
return config.SourceConfig{
|
||||
Name: "test-weatherstories-source",
|
||||
Driver: "nws_weatherstories",
|
||||
Mode: config.SourceModePoll,
|
||||
Params: map[string]any{
|
||||
"url": url,
|
||||
"user_agent": "test-agent",
|
||||
},
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user